}: {@code udfProcessingTime} (a latency histogram) and {@code
+ * udfExceptionCount} (a counter of exceptions escaping the function).
+ *
+ * Timing is sampled: only one invocation out of every {@code sampleInterval} is measured, so a
+ * hot function does not pay {@code System.nanoTime()} on every record. The sample decision advances
+ * a plain {@code int} and is only ever taken on the task thread, so it needs no synchronization.
+ * 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 {
+
+ private static final int HISTORY_SIZE = 128;
+
+ private final int sampleInterval;
+ private final Histogram processingTime;
+ private final Counter exceptionCount;
+
+ // Sample counter; only touched on the task thread, hence a plain int with no synchronization.
+ private int invocationCount = 0;
+
+ private UdfMetrics(int sampleInterval, Histogram processingTime, Counter exceptionCount) {
+ this.sampleInterval = sampleInterval;
+ this.processingTime = processingTime;
+ this.exceptionCount = exceptionCount;
+ }
+
+ /**
+ * Registers the {@code udfProcessingTime} histogram and {@code udfExceptionCount} counter under
+ * {@code .udf.}.
+ *
+ * @param sampleInterval measure one invocation out of every {@code sampleInterval}; must be
+ * {@code >= 1} ({@code 1} measures every invocation)
+ */
+ public static UdfMetrics register(
+ MetricGroup operatorMetricGroup, String udfName, int sampleInterval) {
+ Preconditions.checkArgument(
+ sampleInterval >= 1,
+ "UDF metric sample interval must be >= 1, but was %s.",
+ sampleInterval);
+ MetricGroup group = operatorMetricGroup.addGroup("udf", udfName);
+ return new UdfMetrics(
+ sampleInterval,
+ group.histogram(
+ "udfProcessingTime", new DescriptiveStatisticsHistogram(HISTORY_SIZE)),
+ group.counter("udfExceptionCount", new ThreadSafeSimpleCounter()));
+ }
+
+ /**
+ * Returns {@code true} for the one invocation in every {@code sampleInterval} whose processing
+ * time should be measured. Must be called once per invocation, on the task thread only.
+ */
+ public boolean shouldSample() {
+ if (sampleInterval == 1) {
+ return true;
+ }
+ invocationCount = (invocationCount + 1 < sampleInterval) ? invocationCount + 1 : 0;
+ return invocationCount == 1;
+ }
+
+ /** Records an elapsed processing time (nanoseconds) for a sampled invocation. */
+ public void update(long elapsedNanos) {
+ processingTime.update(elapsedNanos);
+ }
+
+ /** Counts one exception escaping the user-defined function. */
+ public void markException() {
+ exceptionCount.inc();
+ }
+}
diff --git a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/metrics/UdfMetricsTest.java b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/metrics/UdfMetricsTest.java
new file mode 100644
index 00000000000000..49a5c787144795
--- /dev/null
+++ b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/metrics/UdfMetricsTest.java
@@ -0,0 +1,140 @@
+/*
+ * 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 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);
+ }
+
+ @Test
+ void metricsAreIndependentPerUdf() {
+ UdfMetrics first = register("firstUdf", 1);
+ register("secondUdf", 1);
+ Histogram firstHistogram =
+ metricListener.getHistogram("udf", "firstUdf", "udfProcessingTime").get();
+ Counter firstCounter =
+ metricListener.getCounter("udf", "firstUdf", "udfExceptionCount").get();
+ Histogram secondHistogram =
+ metricListener.getHistogram("udf", "secondUdf", "udfProcessingTime").get();
+ Counter secondCounter =
+ metricListener.getCounter("udf", "secondUdf", "udfExceptionCount").get();
+
+ first.update(10L);
+ first.markException();
+
+ assertThat(firstHistogram.getCount()).isEqualTo(1L);
+ assertThat(firstCounter.getCount()).isEqualTo(1L);
+ assertThat(secondHistogram.getCount()).isZero();
+ assertThat(secondCounter.getCount()).isZero();
+ }
+
+ private UdfMetrics register(String udfName, int sampleInterval) {
+ return UdfMetrics.register(metricListener.getMetricGroup(), udfName, sampleInterval);
+ }
+}