[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