[FLINK-38071][table] Add opt-in UDF metrics (FLIP-485)

[weiqingy]

What is the purpose of the change

This pull request is a reference implementation of FLIP-485: Add UDF Metrics (FLINK-38071). It adds opt-in, per-operator observability for SQL/Table user-defined functions so operators can see inside UDF "black boxes" when debugging latency, throughput, or errors — and feed a reliable "the problem is in user code" signal to autoscaling.

Two metrics are registered on the executing operator's OperatorMetricGroup, scoped as <operator_name>.udf.<udf_name>.<metric>:

• udfProcessingTime — a Histogram of per-invocation UDF time (for async UDFs, the full dispatch-to-completion span).
• udfExceptionCount — a Counter of exceptions escaping the UDF (synchronous exceptions propagating out of eval, and asynchronous exceptional completions).

The feature is off by default (table.exec.udf-metric-enabled = false) with zero overhead when disabled — the instrumentation is emitted at code generation only when enabled, so the generated operator is byte-identical to today when off. When enabled, it uses the same counter-based sampling as state latency tracking (FLINK-21736): only every Nth invocation is timed (table.exec.udf-metric.sample-interval, default 100), while exceptions are counted on every invocation.

It is opened as a draft to validate the FLIP design against master and give the DISCUSS/VOTE thread a concrete reference; it is not intended to presume the vote outcome.

Brief change log

• Add two @PublicEvolving options to ExecutionConfigOptions: table.exec.udf-metric-enabled (default false) and table.exec.udf-metric.sample-interval (default 100), plus generated config docs.
• Add a reusable UdfMetrics helper (flink-table-runtime) that registers the two metrics under addGroup("udf", udfName), owns the FLINK-21736 sampling decision, brackets timing, and counts exceptions. The processing-time histogram is a DescriptiveStatisticsHistogram and the exception counter is a ThreadSafeSimpleCounter.
• Instrument synchronous scalar and table UDF calls at code generation (BridgingFunctionGenUtil), sharing one handle per (operator, udf name). Only user functions on the modern BridgingSqlFunction stack are metered.
• Instrument asynchronous scalar and table UDF calls: the sampling decision and start time are captured at dispatch on the task thread; udfProcessingTime and udfExceptionCount are recorded at completion on the callback thread, touching only the synchronized histogram and the thread-safe counter.
• Document the feature under docs/…/ops/metrics.md.

Not metered (out of scope of this FLIP): Process Table Functions (PROCESS_TABLE), aggregate functions, and functions registered via the deprecated registerFunction API.

Verifying this change

This change added tests and can be verified as follows:

• UdfMetricsTest (unit) — the sampling decision (including the sample-interval = 1 "every call" case and the reset boundary), timing, and exception counting.
• UdfMetricsITCase (integration, InMemoryReporter) — 12 cases covering sync scalar and table, async scalar and table, metric naming/scope, values (a delayed UDF whose udfProcessingTime reflects the delay), exception counting, the enabled/disabled gate, the one-handle-per-(operator, udf) sharing, and — for async — an exceptional completion that increments udfExceptionCount while the job still finishes.
• ConfigOptionsDocsCompletenessITCase and the regenerated config docs verify the two new options are documented.
• Overhead was sanity-checked with a local micro-benchmark (the disabled path is byte-identical, i.e. zero; the enabled non-sampled fast path is a single integer increment). A rigorous JMH benchmark belongs in flink-benchmarks.

Does this pull request potentially affect one of the following parts:

• Dependencies (does it add or upgrade a dependency): no
• The public API, i.e., is any changed class annotated with @Public(Evolving): yes — two new @PublicEvolving ConfigOptions in ExecutionConfigOptions.
• The serializers: no
• The runtime per-record code paths (performance sensitive): yes — the instrumentation wraps the generated UDF call site. It is gated at code generation (byte-identical when disabled) and counter-sampled when enabled, so the disabled path has zero overhead and the enabled fast path is a single integer increment.
• Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: no
• The S3 file system connector: no

Documentation

• Does this pull request introduce a new feature? yes
• If yes, how is the feature documented? docs (docs/content/docs/ops/metrics.md and the Chinese copy) and JavaDocs.

---

Was generative AI tooling used to co-author this PR?

• Yes (Claude Code, Anthropic Claude Opus 4.8)

[flinkbot]

CI report:

• 154a637 Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

• @flinkbot run azure re-run the last Azure build

[davidradl]

is this advice the same for all metrics - or is this metric particularly performance adverse.

Do we have an idea of what the performance overhead is when switching this on? I would expect metrics overhead would/should be minimal

[weiqingy]

Thanks for taking a look, David. Good questions on both.

On the wording: the "…may impact performance / recommended only for debugging" note mirrors the existing State Access Latency and state-size metric sections a bit higher on this same page. It's the standard caution for opt-in, per-record operator metrics, not a signal that UDF metrics are unusually costly.

On the actual overhead — it should be minimal, and that's the design intent, via two layers:

• Off by default: when table.exec.udf-metric-enabled is false, nothing is registered and there is zero overhead.
• When enabled, it uses the same counter-based sampling as state latency tracking (FLINK-21736): only every Nth invocation is timed (default sample-interval = 100) and the non-sampled fast path is a single integer increment. Exceptions are counted every invocation, but that's just a counter increment.

So on the common path the added per-record cost is essentially a branch plus an int increment, with the nanoTime()/histogram work amortized across the sampling interval. In local micro-benchmarking the non-sampled path was sub-nanosecond; a rigorous JMH benchmark belongs in flink-benchmarks and I can add one.

Happy to make the sampling context clearer in this note (e.g. that overhead is minimal at the default interval) when I make the PR review-ready.

One heads-up: this PR is intentionally still a draft. I'm keeping it in draft until the FLIP-485 vote concludes, then I'll mark it review-ready. Appreciate the early look.

[wbrycki]

Would it be worth adding a test case that verifies metrics are independent for different UDFs?
For example, the test could register two UDFs, invoke them a different number of times, and verify that each UDF has its own udfProcessingTime histogram and udfExceptionCount counter.

Suggested change

	@Test
	void metricsAreIndependentForDifferentUdfs() {
	UdfMetrics firstMetrics = register("firstUdf", 1);
	UdfMetrics secondMetrics = register("secondUdf", 1);
	firstMetrics.update(10L);
	firstMetrics.markException();
	secondMetrics.update(20L);
	secondMetrics.update(30L);
	secondMetrics.markException();
	secondMetrics.markException();
	Histogram firstHistogram =
	metricListener.getHistogram("udf", "firstUdf", "udfProcessingTime").orElseThrow();
	Histogram secondHistogram =
	metricListener.getHistogram("udf", "secondUdf", "udfProcessingTime").orElseThrow();
	Counter firstCounter =
	metricListener.getCounter("udf", "firstUdf", "udfExceptionCount").orElseThrow();
	Counter secondCounter =
	metricListener.getCounter("udf", "secondUdf", "udfExceptionCount").orElseThrow();
	assertThat(firstHistogram.getCount()).isEqualTo(1L);
	assertThat(firstHistogram.getStatistics().getMin()).isEqualTo(10L);
	assertThat(firstCounter.getCount()).isEqualTo(1L);
	assertThat(secondHistogram.getCount()).isEqualTo(2L);
	assertThat(secondHistogram.getStatistics().getMin()).isEqualTo(20L);
	assertThat(secondHistogram.getStatistics().getMax()).isEqualTo(30L);
	assertThat(secondCounter.getCount()).isEqualTo(2L);
	}

[weiqingy]

Thanks for taking a look.
This PR was a draft reference implementation for the FLIP-485 vote. The FLIP passed on 2026-08-01, so the code is going up as four smaller PRs and this one is superseded. The file you commented on is now in #28878, which is open for review.
Your point holds there. I added metricsAreIndependentPerUdf to UdfMetricsTest in #28878, which registers two UDFs, records on one, and asserts the other's histogram and counter stay at zero. I kept it to the separation check, since the min/max and counting assertions in your snippet are already covered by the existing updateFeedsHistogram and markExceptionCounts tests.

[weiqingy]

Closing this in favor of a smaller PR series.

FLIP-485 passed the vote: https://lists.apache.org/thread/symqpswsohl2s5wmtkcw0jjp1w5dot0n

This draft was opened as a single reference implementation so the DISCUSS and VOTE threads had something concrete to read against master. Now that the design is settled, the same code is going up as four smaller, independently reviewable PRs:

1. UdfMetrics helper: [FLINK-40292][table-runtime] Add UdfMetrics helper for UDF metrics #28878
2. Config options plus synchronous scalar and table instrumentation
3. Asynchronous scalar and table instrumentation
4. Documentation

The code is unchanged from what was reviewed here. It is only partitioned and rebased onto current master.

@davidradl @wbrycki thanks for reviewing this. Both comments are answered in the inline threads.