[FLINK-40292][table-runtime] Add UdfMetrics helper for UDF metrics

flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/metrics/UdfMetrics.java

@@ -0,0 +1,98 @@
+/*
+ * 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.MetricGroup;
+import org.apache.flink.metrics.ThreadSafeSimpleCounter;
+import org.apache.flink.runtime.metrics.DescriptiveStatisticsHistogram;
+import org.apache.flink.util.Preconditions;
+
+/**
+ * Per-operator metrics for a single user-defined function, registered under {@code
+ * <operator>.udf.<udfName>}: {@code udfProcessingTime} (a latency histogram) and {@code
+ * udfExceptionCount} (a counter of exceptions escaping the function).
+ *
+ * <p>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 <operatorMetricGroup>.udf.<udfName>}.
+     *
+     * @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();
+    }
+}

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<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);
+    }
+
+    @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);
+    }
+}