[FLINK-40293][table] Add UDF metric config options and instrument sync scalar and table UDF calls

[weiqingy]

This is the second PR of the FLIP-485 implementation, split into a stack of small, independently reviewable PRs under the umbrella issue FLINK-38071. Landing order:

Step	Sub-task	Scope
PR-1	FLINK-40292	UdfMetrics helper: registration, sampling, timing, exception counting
PR-2 (this PR)	FLINK-40293	Config options + sync scalar/table instrumentation
PR-3	FLINK-40294	Async scalar/table instrumentation
PR-4	FLINK-40295	User-facing documentation

Stacked on #28878 and kept as a draft until that merges, so the diff here currently also shows the UdfMetrics commit.

What is the purpose of the change

Adds the two configuration options and the first end-to-end slice of FLIP-485: synchronous scalar and table UDF calls instrumented at code generation, using the UdfMetrics helper from PR-1.

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.
• udfExceptionCount, a Counter of exceptions that escape user code.

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 the option is on, so the generated operator is byte-identical to today when it is off. When on, only every Nth invocation is timed (table.exec.udf-metric.sample-interval, default 100), while exceptions are counted on every invocation.

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 the regenerated config docs.
• Instrument the generated eval call site for synchronous scalar and table UDFs in BridgingFunctionGenUtil, bracketing it with sampled timing and exception counting.
• Cache one UdfMetrics handle per (operator, udf name) in CodeGeneratorContext, so repeated call sites of the same function in one operator share a handle instead of the second registration being dropped.
• Meter only user functions on the modern BridgingSqlFunction stack, via an opt-in name passed from BridgingSqlFunctionCallGen. Lookup-join, ML-predict, vector-search, legacy CallGens, and PROCESS_TABLE functions are not metered.

Verifying this change

This change added tests and can be verified as follows:

• UdfMetricsITCase (integration, using InMemoryReporter) covers synchronous scalar and table functions, metric naming and scope, exception counting, the enabled/disabled gate, that a lookup join is not metered, that repeated call sites of one function share a single handle while distinct functions get separate ones, and that udfProcessingTime reflects a real induced delay rather than only a sample count.
• ConfigOptionsDocsCompletenessITCase and the regenerated config docs verify the two new options are documented.

On overhead: the disabled path is byte-identical to today, so it is zero by construction. With the feature on, a local micro-benchmark put the non-sampled fast path in the sub-nanosecond range, since it is a single integer increment, with the nanoTime() and histogram work amortized across the sampling interval. A rigorous JMH benchmark belongs in flink-benchmarks rather than here, and I am happy to follow up with one.

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, so the generated code is byte-identical when the feature is disabled, and counter-sampled when enabled, so the non-sampled path costs 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? JavaDocs and the generated config option docs here; the metrics.md section lands in PR-4.

---

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

• Yes (please specify the tool below)

Generated-by: Claude Code (Anthropic Claude Opus 4.8 and Claude Opus 5)

[flinkbot]

CI report:

• 720f7b0 Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

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