You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "weiqingy (via GitHub)" <gi...@apache.org> on 2026/07/09 00:41:08 UTC
[PR] [FLINK-38071][table] Add opt-in UDF metrics (FLIP-485) [flink]
weiqingy opened a new pull request, #28692:
URL: https://github.com/apache/flink/pull/28692
## What is the purpose of the change
This pull request is a **reference implementation of [FLIP-485: Add UDF Metrics](https://cwiki.apache.org/confluence/spaces/FLINK/pages/373885706/FLIP-485+Add+UDF+Metrics)** ([FLINK-38071](https://issues.apache.org/jira/browse/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` `ConfigOption`s 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?
- [X] Yes (Claude Code, Anthropic Claude Opus 4.8)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38071][table] Add opt-in UDF metrics (FLIP-485) [flink]
Posted by "davidradl (via GitHub)" <gi...@apache.org>.
davidradl commented on code in PR #28692:
URL: https://github.com/apache/flink/pull/28692#discussion_r3585909784
##########
docs/content/docs/ops/metrics.md:
##########
@@ -2441,6 +2469,22 @@ It is recommended to only use them for debugging purposes.
If state.ttl is enabled, the size of the value will include the size of the TTL-related timestamp.
The value size of AggregatingState is not accounted for because AggregatingState returns a result processed by a user-defined AggregateFunction, whereas currently, only the actual stored data size in the state can be tracked.
+## UDF metrics
+
+Flink can track the processing time and exceptions of SQL/Table user-defined functions on a per-operator basis. This feature is disabled by default.
+To enable it you must set `table.exec.udf-metric-enabled` to `true`. When disabled, nothing is registered and there is no overhead.
+
+Once enabled, each user-defined function gets two metrics on the operator that runs it, scoped under `udf.<udf_name>`: a `udfProcessingTime` histogram and a `udfExceptionCount` counter.
+
+Flink samples the processing time every `N` invocations, in which `N` is defined by `table.exec.udf-metric.sample-interval`.
+This configuration has a default value of 100. A smaller value will get more accurate results but have a higher performance impact since it is sampled more frequently.
+Exceptions are counted on every invocation and are not sampled.
+
+The metrics cover synchronous and asynchronous scalar and table functions. Process table functions (`PROCESS_TABLE`), aggregate functions, and functions registered through the deprecated `registerFunction` API are not covered.
+
+<span class="label label-danger">Warning</span> Enabling UDF metrics may impact the performance, both from the sampled timing and from the metric reporter.
+It is recommended to only use them for debugging purposes.
Review Comment:
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
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38071][table] Add opt-in UDF metrics (FLIP-485) [flink]
Posted by "weiqingy (via GitHub)" <gi...@apache.org>.
weiqingy commented on PR #28692:
URL: https://github.com/apache/flink/pull/28692#issuecomment-5160767517
Closing this in favor of a smaller PR series.
FLIP-485 passed the vote on 2026-08-01 with 3 binding +1 and no -1: 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: https://github.com/apache/flink/pull/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 thanks again for the overhead question earlier. The note that the sampling keeps the enabled-path cost down goes into the documentation PR, number 4 in the list above.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38071][table] Add opt-in UDF metrics (FLIP-485) [flink]
Posted by "flinkbot (via GitHub)" <gi...@apache.org>.
flinkbot commented on PR #28692:
URL: https://github.com/apache/flink/pull/28692#issuecomment-4920510764
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "7787f6d5240f972579b9136939873f9cc514b3d2",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "7787f6d5240f972579b9136939873f9cc514b3d2",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 7787f6d5240f972579b9136939873f9cc514b3d2 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38071][table] Add opt-in UDF metrics (FLIP-485) [flink]
Posted by "weiqingy (via GitHub)" <gi...@apache.org>.
weiqingy commented on code in PR #28692:
URL: https://github.com/apache/flink/pull/28692#discussion_r3700641675
##########
flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/metrics/UdfMetricsTest.java:
##########
@@ -0,0 +1,118 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.runtime.operators.metrics;
+
+import org.apache.flink.metrics.Counter;
+import org.apache.flink.metrics.Histogram;
+import org.apache.flink.metrics.ThreadSafeSimpleCounter;
+import org.apache.flink.metrics.testutils.MetricListener;
+import org.apache.flink.runtime.metrics.DescriptiveStatisticsHistogram;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Tests for {@link UdfMetrics}. */
+class UdfMetricsTest {
+
+ private MetricListener metricListener;
+
+ @BeforeEach
+ void setUp() {
+ metricListener = new MetricListener();
+ }
+
+ @Test
+ void sampleEveryNth() {
+ UdfMetrics metrics = register("myUdf", 100);
+
+ List<Integer> sampledCalls = new ArrayList<>();
+ for (int call = 1; call <= 300; call++) {
+ if (metrics.shouldSample()) {
+ sampledCalls.add(call);
+ }
+ }
+
+ // With interval 100 exactly one call in every 100 is sampled, on the 1st, 101st, 201st.
+ assertThat(sampledCalls).containsExactly(1, 101, 201);
+ }
+
+ @Test
+ void sampleEveryCallWhenIntervalOne() {
+ UdfMetrics metrics = register("myUdf", 1);
+
+ for (int call = 0; call < 10; call++) {
+ assertThat(metrics.shouldSample()).isTrue();
+ }
+ }
+
+ @Test
+ void rejectsNonPositiveInterval() {
+ assertThatThrownBy(() -> register("myUdf", 0)).isInstanceOf(IllegalArgumentException.class);
+ assertThatThrownBy(() -> register("myUdf", -1))
+ .isInstanceOf(IllegalArgumentException.class);
+ }
+
+ @Test
+ void registrationNamesAndTypes() {
+ register("myUdf", 100);
+
+ assertThat(metricListener.getHistogram("udf", "myUdf", "udfProcessingTime"))
+ .get()
+ .isInstanceOf(DescriptiveStatisticsHistogram.class);
+ assertThat(metricListener.getCounter("udf", "myUdf", "udfExceptionCount"))
+ .get()
+ .isInstanceOf(ThreadSafeSimpleCounter.class);
+ }
+
+ @Test
+ void updateFeedsHistogram() {
+ UdfMetrics metrics = register("myUdf", 1);
+ Histogram histogram =
+ metricListener.getHistogram("udf", "myUdf", "udfProcessingTime").get();
+
+ metrics.update(10L);
+ metrics.update(20L);
+ metrics.update(30L);
+
+ assertThat(histogram.getCount()).isEqualTo(3L);
+ assertThat(histogram.getStatistics().getMin()).isEqualTo(10L);
+ assertThat(histogram.getStatistics().getMax()).isEqualTo(30L);
+ }
+
+ @Test
+ void markExceptionCounts() {
+ UdfMetrics metrics = register("myUdf", 1);
+ Counter counter = metricListener.getCounter("udf", "myUdf", "udfExceptionCount").get();
+
+ metrics.markException();
+ metrics.markException();
+
+ assertThat(counter.getCount()).isEqualTo(2L);
+ }
+
Review Comment:
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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38071][table] Add opt-in UDF metrics (FLIP-485) [flink]
Posted by "davidradl (via GitHub)" <gi...@apache.org>.
davidradl commented on code in PR #28692:
URL: https://github.com/apache/flink/pull/28692#discussion_r3585909784
##########
docs/content/docs/ops/metrics.md:
##########
@@ -2441,6 +2469,22 @@ It is recommended to only use them for debugging purposes.
If state.ttl is enabled, the size of the value will include the size of the TTL-related timestamp.
The value size of AggregatingState is not accounted for because AggregatingState returns a result processed by a user-defined AggregateFunction, whereas currently, only the actual stored data size in the state can be tracked.
+## UDF metrics
+
+Flink can track the processing time and exceptions of SQL/Table user-defined functions on a per-operator basis. This feature is disabled by default.
+To enable it you must set `table.exec.udf-metric-enabled` to `true`. When disabled, nothing is registered and there is no overhead.
+
+Once enabled, each user-defined function gets two metrics on the operator that runs it, scoped under `udf.<udf_name>`: a `udfProcessingTime` histogram and a `udfExceptionCount` counter.
+
+Flink samples the processing time every `N` invocations, in which `N` is defined by `table.exec.udf-metric.sample-interval`.
+This configuration has a default value of 100. A smaller value will get more accurate results but have a higher performance impact since it is sampled more frequently.
+Exceptions are counted on every invocation and are not sampled.
+
+The metrics cover synchronous and asynchronous scalar and table functions. Process table functions (`PROCESS_TABLE`), aggregate functions, and functions registered through the deprecated `registerFunction` API are not covered.
+
+<span class="label label-danger">Warning</span> Enabling UDF metrics may impact the performance, both from the sampled timing and from the metric reporter.
+It is recommended to only use them for debugging purposes.
Review Comment:
is this advice the same for all metrics - or is this metric particularly performance adverse.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38071][table] Add opt-in UDF metrics (FLIP-485) [flink]
Posted by "weiqingy (via GitHub)" <gi...@apache.org>.
weiqingy commented on code in PR #28692:
URL: https://github.com/apache/flink/pull/28692#discussion_r3592851558
##########
docs/content/docs/ops/metrics.md:
##########
@@ -2441,6 +2469,22 @@ It is recommended to only use them for debugging purposes.
If state.ttl is enabled, the size of the value will include the size of the TTL-related timestamp.
The value size of AggregatingState is not accounted for because AggregatingState returns a result processed by a user-defined AggregateFunction, whereas currently, only the actual stored data size in the state can be tracked.
+## UDF metrics
+
+Flink can track the processing time and exceptions of SQL/Table user-defined functions on a per-operator basis. This feature is disabled by default.
+To enable it you must set `table.exec.udf-metric-enabled` to `true`. When disabled, nothing is registered and there is no overhead.
+
+Once enabled, each user-defined function gets two metrics on the operator that runs it, scoped under `udf.<udf_name>`: a `udfProcessingTime` histogram and a `udfExceptionCount` counter.
+
+Flink samples the processing time every `N` invocations, in which `N` is defined by `table.exec.udf-metric.sample-interval`.
+This configuration has a default value of 100. A smaller value will get more accurate results but have a higher performance impact since it is sampled more frequently.
+Exceptions are counted on every invocation and are not sampled.
+
+The metrics cover synchronous and asynchronous scalar and table functions. Process table functions (`PROCESS_TABLE`), aggregate functions, and functions registered through the deprecated `registerFunction` API are not covered.
+
+<span class="label label-danger">Warning</span> Enabling UDF metrics may impact the performance, both from the sampled timing and from the metric reporter.
+It is recommended to only use them for debugging purposes.
Review Comment:
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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38071][table] Add opt-in UDF metrics (FLIP-485) [flink]
Posted by "wbrycki (via GitHub)" <gi...@apache.org>.
wbrycki commented on code in PR #28692:
URL: https://github.com/apache/flink/pull/28692#discussion_r3699999695
##########
flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/metrics/UdfMetricsTest.java:
##########
@@ -0,0 +1,118 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.runtime.operators.metrics;
+
+import org.apache.flink.metrics.Counter;
+import org.apache.flink.metrics.Histogram;
+import org.apache.flink.metrics.ThreadSafeSimpleCounter;
+import org.apache.flink.metrics.testutils.MetricListener;
+import org.apache.flink.runtime.metrics.DescriptiveStatisticsHistogram;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Tests for {@link UdfMetrics}. */
+class UdfMetricsTest {
+
+ private MetricListener metricListener;
+
+ @BeforeEach
+ void setUp() {
+ metricListener = new MetricListener();
+ }
+
+ @Test
+ void sampleEveryNth() {
+ UdfMetrics metrics = register("myUdf", 100);
+
+ List<Integer> sampledCalls = new ArrayList<>();
+ for (int call = 1; call <= 300; call++) {
+ if (metrics.shouldSample()) {
+ sampledCalls.add(call);
+ }
+ }
+
+ // With interval 100 exactly one call in every 100 is sampled, on the 1st, 101st, 201st.
+ assertThat(sampledCalls).containsExactly(1, 101, 201);
+ }
+
+ @Test
+ void sampleEveryCallWhenIntervalOne() {
+ UdfMetrics metrics = register("myUdf", 1);
+
+ for (int call = 0; call < 10; call++) {
+ assertThat(metrics.shouldSample()).isTrue();
+ }
+ }
+
+ @Test
+ void rejectsNonPositiveInterval() {
+ assertThatThrownBy(() -> register("myUdf", 0)).isInstanceOf(IllegalArgumentException.class);
+ assertThatThrownBy(() -> register("myUdf", -1))
+ .isInstanceOf(IllegalArgumentException.class);
+ }
+
+ @Test
+ void registrationNamesAndTypes() {
+ register("myUdf", 100);
+
+ assertThat(metricListener.getHistogram("udf", "myUdf", "udfProcessingTime"))
+ .get()
+ .isInstanceOf(DescriptiveStatisticsHistogram.class);
+ assertThat(metricListener.getCounter("udf", "myUdf", "udfExceptionCount"))
+ .get()
+ .isInstanceOf(ThreadSafeSimpleCounter.class);
+ }
+
+ @Test
+ void updateFeedsHistogram() {
+ UdfMetrics metrics = register("myUdf", 1);
+ Histogram histogram =
+ metricListener.getHistogram("udf", "myUdf", "udfProcessingTime").get();
+
+ metrics.update(10L);
+ metrics.update(20L);
+ metrics.update(30L);
+
+ assertThat(histogram.getCount()).isEqualTo(3L);
+ assertThat(histogram.getStatistics().getMin()).isEqualTo(10L);
+ assertThat(histogram.getStatistics().getMax()).isEqualTo(30L);
+ }
+
+ @Test
+ void markExceptionCounts() {
+ UdfMetrics metrics = register("myUdf", 1);
+ Counter counter = metricListener.getCounter("udf", "myUdf", "udfExceptionCount").get();
+
+ metrics.markException();
+ metrics.markException();
+
+ assertThat(counter.getCount()).isEqualTo(2L);
+ }
+
Review Comment:
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.
```suggestion
@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);
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38071][table] Add opt-in UDF metrics (FLIP-485) [flink]
Posted by "weiqingy (via GitHub)" <gi...@apache.org>.
weiqingy commented on code in PR #28692:
URL: https://github.com/apache/flink/pull/28692#discussion_r3592851558
##########
docs/content/docs/ops/metrics.md:
##########
@@ -2441,6 +2469,22 @@ It is recommended to only use them for debugging purposes.
If state.ttl is enabled, the size of the value will include the size of the TTL-related timestamp.
The value size of AggregatingState is not accounted for because AggregatingState returns a result processed by a user-defined AggregateFunction, whereas currently, only the actual stored data size in the state can be tracked.
+## UDF metrics
+
+Flink can track the processing time and exceptions of SQL/Table user-defined functions on a per-operator basis. This feature is disabled by default.
+To enable it you must set `table.exec.udf-metric-enabled` to `true`. When disabled, nothing is registered and there is no overhead.
+
+Once enabled, each user-defined function gets two metrics on the operator that runs it, scoped under `udf.<udf_name>`: a `udfProcessingTime` histogram and a `udfExceptionCount` counter.
+
+Flink samples the processing time every `N` invocations, in which `N` is defined by `table.exec.udf-metric.sample-interval`.
+This configuration has a default value of 100. A smaller value will get more accurate results but have a higher performance impact since it is sampled more frequently.
+Exceptions are counted on every invocation and are not sampled.
+
+The metrics cover synchronous and asynchronous scalar and table functions. Process table functions (`PROCESS_TABLE`), aggregate functions, and functions registered through the deprecated `registerFunction` API are not covered.
+
+<span class="label label-danger">Warning</span> Enabling UDF metrics may impact the performance, both from the sampled timing and from the metric reporter.
+It is recommended to only use them for debugging purposes.
Review Comment:
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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38071][table] Add opt-in UDF metrics (FLIP-485) [flink]
Posted by "weiqingy (via GitHub)" <gi...@apache.org>.
weiqingy closed pull request #28692: [FLINK-38071][table] Add opt-in UDF metrics (FLIP-485)
URL: https://github.com/apache/flink/pull/28692
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org