[FLINK-40292][table-runtime] Add UdfMetrics helper for UDF metrics - #28878
[FLINK-40292][table-runtime] Add UdfMetrics helper for UDF metrics#28878weiqingy wants to merge 2 commits into
Conversation
|
This is the first of four PRs landing the accepted design (FLIP-485). It replaces the single reference-implementation draft in #28692, which I'm closing. This PR adds only the The series:
Each PR builds and passes its own tests independently. I'll open the next one as the previous merges. Hi @xuyangzhong @RocMarshal @HuangZhenQiu, since you voted on the FLIP, I’d appreciate it if you could take a look when you have time. |
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.
6556131 to
15bcbfa
Compare
| * time should be measured. Must be called once per invocation, on the task thread only. | ||
| */ | ||
| public boolean shouldSample() { | ||
| if (sampleInterval == 1) { |
There was a problem hiding this comment.
If the sampling is determined 1 / operation counts, sampleInterval is probably not the best variable name.
There was a problem hiding this comment.
Fair point, the name doesn't tell you that on its own.
Would it be worth keeping for consistency though? Flink's state latency tracking does the same kind of sampling with the same word: state.latency-track.sample-interval is an int, defaults to 100, and means "track the latency every 100 access requests". Its MetricsTrackingStateConfig declares the same private final int sampleInterval with the same >= 1 check. FLIP-485 followed that pattern, and the matching option here, table.exec.udf-metric.sample-interval, lands in the next PR.
What I'd worry about with a rename is the field no longer matching the option users actually set. So I've made the field say it counts invocations, not time. Does that address it, or would you still prefer a different name?
There was a problem hiding this comment.
Make sense. Thanks for the context.
…ations State at the field declaration and in the register() @PARAM that sampleInterval is a number of invocations rather than a duration, so a reader does not have to infer the unit from the class javadoc.
| * The two registered metrics are safe for the async completion thread to touch: the histogram | ||
| * synchronizes internally and the exception counter is {@link ThreadSafeSimpleCounter}. | ||
| */ | ||
| public final class UdfMetrics { |
There was a problem hiding this comment.
| public final class UdfMetrics { | |
| public final class UDFMetrics { |
There was a problem hiding this comment.
Thanks for the review!
Flink seems to treat acronyms as words in class names: AbstractUdfStreamOperator and UdfStreamOperatorFactory in flink-runtime, and TableAbstractCoUdfStreamOperator right next door in flink-table-runtime. The wider pattern looks the same, Sql and Json and Rest outnumber the all-caps spellings in class names by a long way.
The all-caps ones I can find are the older RuntimeUDFContext family in flink-core and a couple of test fixtures.
Happy to rename if you'd still prefer it though. It's this class plus its two tests and a few codegen references.
There was a problem hiding this comment.
Thanks for the clarify.
Just keep original naming.
4 participants
This is the first PR of the FLIP-485 implementation, split into a stack of small, independently reviewable PRs under the umbrella issue FLINK-38071. Landing order:
UdfMetricshelper: registration, sampling, timing, exception countingFLIP-485 passed the vote on 2026-08-01. This series supersedes the single reference-implementation draft in #28692, which is closed.
What is the purpose of the change
FLIP-485 adds opt-in, per-operator observability for SQL/Table user-defined functions, so operators can see inside UDF "black boxes" when debugging latency or errors, and so autoscaling gets a reliable "the problem is in user code" signal.
This PR adds only the shared runtime helper,
UdfMetrics. It owns metric registration, the sampling decision, timing, and exception counting, and it is the piece both the synchronous and the asynchronous paths use. It has no caller yet; the first caller arrives in PR-2. Keeping it separate makes the sampling and registration logic reviewable on its own.Brief change log
UdfMetricsinflink-table-runtime(org.apache.flink.table.runtime.operators.metrics, besideSimpleGauge). It registersudfProcessingTimeandudfExceptionCountunderaddGroup("udf", udfName)on the operator metric group, so the full identifier is<operator_name>.udf.<udf_name>.<metric>.udfProcessingTimeis aDescriptiveStatisticsHistogramof per-invocation nanoseconds, sampled with the same counter-based scheme as state latency tracking (FLINK-21736), including theinterval == 1"measure every call" case.udfExceptionCountis aThreadSafeSimpleCounter, incremented on every exception and not sampled.Verifying this change
This change added tests and can be verified as follows:
UdfMetricsTestcovers metric registration and naming, the sampling decision across the interval boundary, thesample-interval = 1case, rejection of a non-positive interval, timing recorded into the histogram, exception counting, and that two UDFs registered on the same operator get independent metrics.Does this pull request potentially affect one of the following parts:
@Public(Evolving): noDocumentation
Was generative AI tooling used to co-author this PR?
Generated-by: Claude Code (Anthropic Claude Opus 4.8 and Claude Opus 5)