Skip to content

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

Draft
weiqingy wants to merge 2 commits into
apache:masterfrom
weiqingy:flink-38071-pr2-sync
Draft

[FLINK-40293][table] Add UDF metric config options and instrument sync scalar and table UDF calls#28879
weiqingy wants to merge 2 commits into
apache:masterfrom
weiqingy:flink-38071-pr2-sync

Conversation

@weiqingy

@weiqingy weiqingy commented Aug 2, 2026

Copy link
Copy Markdown
Contributor

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

flinkbot commented Aug 2, 2026

Copy link
Copy Markdown
Collaborator

CI report:

Bot commands The @flinkbot bot supports the following commands:
  • @flinkbot run azure re-run the last Azure build

Add a reusable UdfMetrics helper that registers udfProcessingTime (a
DescriptiveStatisticsHistogram of per-invocation nanoseconds) and
udfExceptionCount (a ThreadSafeSimpleCounter) under udf.<udfName> on the
executing operator's metric group, and owns the sampling decision, timing,
and exception counting shared by the sync and async instrumentation paths.

Sampling follows state latency tracking (FLINK-21736), including the
interval == 1 case that measures every invocation. The histogram is safe to
update from an async callback thread; the sampling counter is only advanced
on the task thread at dispatch.

No call site is added here; the first caller arrives with the sync
instrumentation.
…c scalar and table UDF calls

Introduce two opt-in configuration options for FLIP-485 UDF metrics:
table.exec.udf-metric-enabled (default false) and
table.exec.udf-metric.sample-interval (default 100).

Wrap the generated eval call site for sync scalar and table user-defined
functions (via the BridgingSqlFunction stack) with sampled udfProcessingTime
timing and udfExceptionCount counting, using the UdfMetrics helper registered
on the operator metric group under udf.<udfName>. Instrumentation is emitted
only when table.exec.udf-metric-enabled is true; the generated operator is
byte-identical when disabled. One shared handle is registered per (operator,
udfName). Lookup-join, ML-predict, vector-search, legacy CallGens, and
PROCESS_TABLE functions are not metered.
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

2 participants