From 12b1961d17ffb4dd16b25fbbdc96c9b374316fac Mon Sep 17 00:00:00 2001 From: weiqingy Date: Tue, 7 Jul 2026 17:41:58 -0700 Subject: [PATCH 1/5] [FLINK-38071][table] Add table.exec.udf-metric-enabled and table.exec.udf-metric.sample-interval options 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). No behavior is wired yet; subsequent commits register and populate the metrics when enabled. --- .../execution_config_configuration.html | 12 ++++++++++ .../api/config/ExecutionConfigOptions.java | 22 +++++++++++++++++++ 2 files changed, 34 insertions(+) diff --git a/docs/layouts/shortcodes/generated/execution_config_configuration.html b/docs/layouts/shortcodes/generated/execution_config_configuration.html index e71bf3d7673a9..fb2dfae186a63 100644 --- a/docs/layouts/shortcodes/generated/execution_config_configuration.html +++ b/docs/layouts/shortcodes/generated/execution_config_configuration.html @@ -362,6 +362,18 @@ Duration Specifies a minimum time interval for how long idle state (i.e. state which was not updated), will be retained. State will never be cleared until it was idle for less than the minimum time, and will be cleared at some time after it was idle. Default is never clean-up the state. NOTE: Cleaning up state requires additional overhead for bookkeeping. Default value is 0, which means that it will never clean up state. + +
table.exec.udf-metric-enabled

Streaming + false + Boolean + When enabled, per-operator UDF metrics (udfProcessingTime, udfExceptionCount) are registered for SQL/Table user-defined functions. Disabled by default; nothing is registered and there is zero overhead when off. + + +
table.exec.udf-metric.sample-interval

Streaming + 100 + Integer + When UDF metrics are enabled, udfProcessingTime is measured every N invocations (default 100). The non-sampled fast path is a single integer increment. +
table.exec.uid.format

Streaming "<id>_<transformation>" diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/config/ExecutionConfigOptions.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/config/ExecutionConfigOptions.java index a1ed3ec97ff58..81d7acbe38c2b 100644 --- a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/config/ExecutionConfigOptions.java +++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/config/ExecutionConfigOptions.java @@ -699,6 +699,28 @@ public class ExecutionConfigOptions { + TABLE_EXEC_MINIBATCH_ENABLED.key() + " is set true, its value must be positive."); + // ------------------------------------------------------------------------ + // UDF Options + // ------------------------------------------------------------------------ + @Documentation.TableOption(execMode = Documentation.ExecMode.STREAMING) + public static final ConfigOption TABLE_EXEC_UDF_METRIC_ENABLED = + key("table.exec.udf-metric-enabled") + .booleanType() + .defaultValue(false) + .withDescription( + "When enabled, per-operator UDF metrics (udfProcessingTime, " + + "udfExceptionCount) are registered for SQL/Table user-defined functions. " + + "Disabled by default; nothing is registered and there is zero overhead when off."); + + @Documentation.TableOption(execMode = Documentation.ExecMode.STREAMING) + public static final ConfigOption TABLE_EXEC_UDF_METRIC_SAMPLE_INTERVAL = + key("table.exec.udf-metric.sample-interval") + .intType() + .defaultValue(100) + .withDescription( + "When UDF metrics are enabled, udfProcessingTime is measured every N " + + "invocations (default 100). The non-sampled fast path is a single integer increment."); + // ------------------------------------------------------------------------ // Other Exec Options // ------------------------------------------------------------------------ From 68b5e90112f7e521f8aa2ffefdf83917bf12f159 Mon Sep 17 00:00:00 2001 From: weiqingy Date: Tue, 7 Jul 2026 19:35:08 -0700 Subject: [PATCH 2/5] [FLINK-38071][table-runtime] Add UdfMetrics helper for UDF metrics Reusable per-(operator, UDF) helper that registers udfProcessingTime (DescriptiveStatisticsHistogram) and udfExceptionCount (ThreadSafeSimpleCounter) under an addGroup("udf", udfName) scope, and implements FLINK-21736-style counter-based sampling. Not yet wired into code generation. --- .../runtime/operators/metrics/UdfMetrics.java | 98 +++++++++++++++ .../operators/metrics/UdfMetricsTest.java | 118 ++++++++++++++++++ 2 files changed, 216 insertions(+) create mode 100644 flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/metrics/UdfMetrics.java create mode 100644 flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/metrics/UdfMetricsTest.java diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/metrics/UdfMetrics.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/metrics/UdfMetrics.java new file mode 100644 index 0000000000000..3b04315d956e5 --- /dev/null +++ b/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 + * .udf.}: {@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 0000000000000..eb6608da90fc2 --- /dev/null +++ b/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 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); + } + + private UdfMetrics register(String udfName, int sampleInterval) { + return UdfMetrics.register(metricListener.getMetricGroup(), udfName, sampleInterval); + } +} From e8977cbee00e653346565a440aefd5172102d59f Mon Sep 17 00:00:00 2001 From: weiqingy Date: Tue, 7 Jul 2026 23:10:01 -0700 Subject: [PATCH 3/5] [FLINK-38071][table-planner] Instrument sync scalar and table UDF calls with metrics 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.. 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. --- .../codegen/CodeGeneratorContext.scala | 32 ++ .../calls/BridgingFunctionGenUtil.scala | 102 +++++- .../calls/BridgingSqlFunctionCallGen.scala | 6 +- .../runtime/stream/sql/UdfMetricsITCase.java | 296 ++++++++++++++++++ 4 files changed, 419 insertions(+), 17 deletions(-) create mode 100644 flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/stream/sql/UdfMetricsITCase.java diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/CodeGeneratorContext.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/CodeGeneratorContext.scala index fade7279606a1..2edf0dbac83ee 100644 --- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/CodeGeneratorContext.scala +++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/CodeGeneratorContext.scala @@ -161,6 +161,11 @@ class CodeGeneratorContext( // set of function instance term that will be added only once private val reusableFunctionTerms: mutable.HashSet[String] = mutable.HashSet[String]() + // UDF-metrics handle field terms keyed on UDF name. All call sites of the same function in one + // operator share a single UdfMetrics registration (one histogram + one counter), because + // registering the same metric name twice on a group is silently dropped. + private val reusableUdfMetricsTerms: mutable.Map[String, String] = mutable.Map[String, String]() + // map of type serializer that will be added only once // LogicalType -> reused_term private val reusableTypeSerializers: mutable.Map[LogicalType, String] = @@ -899,6 +904,33 @@ class CodeGeneratorContext( fieldTerm } + /** + * Returns the member field term of the [[UdfMetrics]] handle for the given UDF name, registering + * it in the generated `open()` on first use. Subsequent calls for the same name return the cached + * term, so every call site of one function in an operator feeds a single histogram and counter. + * + * @param udfName + * registered UDF identifier, used as the metric-group name under `.udf.` + * @param sampleInterval + * measure one invocation out of every `sampleInterval` + */ + def addReusableUdfMetrics(udfName: String, sampleInterval: Int): String = { + reusableUdfMetricsTerms.getOrElseUpdate( + udfName, { + val fieldTerm = newName(this, "udfMetrics") + addReusableMember( + s"private transient org.apache.flink.table.runtime.operators.metrics.UdfMetrics $fieldTerm;") + val escapedName = EncodingUtils.escapeJava(udfName) + addReusableOpenStatement( + s""" + |$fieldTerm = org.apache.flink.table.runtime.operators.metrics.UdfMetrics.register( + | getRuntimeContext().getMetricGroup(), "$escapedName", $sampleInterval); + |""".stripMargin) + fieldTerm + } + ) + } + /** * Adds a reusable [[DataStructureConverter]] to the member area of the generated class. * diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/BridgingFunctionGenUtil.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/BridgingFunctionGenUtil.scala index f557cb0e57049..770934d4ed95f 100644 --- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/BridgingFunctionGenUtil.scala +++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/BridgingFunctionGenUtil.scala @@ -21,6 +21,7 @@ import org.apache.flink.api.common.functions.{AbstractRichFunction, OpenContext, import org.apache.flink.configuration.ReadableConfig import org.apache.flink.table.api.{DataTypes, TableException} import org.apache.flink.table.api.Expressions.callSql +import org.apache.flink.table.api.config.ExecutionConfigOptions import org.apache.flink.table.data.{GenericRowData, RawValueData, StringData} import org.apache.flink.table.data.binary.{BinaryRawValueData, BinaryStringData} import org.apache.flink.table.expressions.ApiExpressionUtils.{typeLiteral, unresolvedCall, unresolvedRef} @@ -79,7 +80,8 @@ object BridgingFunctionGenUtil { callContext: CallContext, udf: UserDefinedFunction, functionName: String, - skipIfArgsNull: Boolean): GeneratedExpression = { + skipIfArgsNull: Boolean, + udfMetricName: Option[String] = None): GeneratedExpression = { val (call, _) = generateFunctionAwareCallWithDataType( ctx, @@ -89,7 +91,8 @@ object BridgingFunctionGenUtil { callContext, udf, functionName, - skipIfArgsNull) + skipIfArgsNull, + udfMetricName) call } @@ -102,7 +105,8 @@ object BridgingFunctionGenUtil { callContext: CallContext, udf: UserDefinedFunction, functionName: String, - skipIfArgsNull: Boolean): (GeneratedExpression, DataType) = { + skipIfArgsNull: Boolean, + udfMetricName: Option[String] = None): (GeneratedExpression, DataType) = { val result = generateFunctionAwareCallWithDataTypeAndTimeout( ctx, operands, @@ -111,7 +115,8 @@ object BridgingFunctionGenUtil { callContext, udf, functionName, - skipIfArgsNull) + skipIfArgsNull, + udfMetricName) (result._1, result._3) } @@ -165,7 +170,11 @@ object BridgingFunctionGenUtil { callContext, udf, function.toString, - skipIfArgsNull) + skipIfArgsNull, + // This is the async table function correlate entry; opt it into metrics under the UDF name, + // mirroring BridgingSqlFunctionCallGen for the other UDF kinds. + udfMetricName = Some(function.getName) + ) } /** @@ -189,7 +198,9 @@ object BridgingFunctionGenUtil { callContext: CallContext, udf: UserDefinedFunction, functionName: String, - skipIfArgsNull: Boolean): (GeneratedExpression, Option[GeneratedExpression], DataType) = { + skipIfArgsNull: Boolean, + udfMetricName: Option[String] = None) + : (GeneratedExpression, Option[GeneratedExpression], DataType) = { // enrich argument types with conversion class val castCallContext = TypeInferenceUtil.castArguments(inference, callContext, null) @@ -219,7 +230,8 @@ object BridgingFunctionGenUtil { returnType, udf, skipIfArgsNull, - None) + None, + udfMetricName) val timeoutCall = if (udf.getKind == FunctionKind.ASYNC_TABLE) { generateAsyncTableFunctionTimeoutCall( @@ -245,8 +257,15 @@ object BridgingFunctionGenUtil { returnType: LogicalType, udf: UserDefinedFunction, skipIfArgsNull: Boolean, - contextTerm: Option[String]): GeneratedExpression = { + contextTerm: Option[String], + udfMetricName: Option[String]): GeneratedExpression = { if (udf.getKind == FunctionKind.TABLE || udf.getKind == FunctionKind.PROCESS_TABLE) { + // Only plain TABLE functions are metered here. A PROCESS_TABLE function's runtime is + // generated by a dedicated stateful operator with its own eval and onTimer entry points, so + // its timing must be measured there rather than at this bridging call site; it always opts + // out of metrics. + val tableUdfMetricName = + if (udf.getKind == FunctionKind.PROCESS_TABLE) None else udfMetricName generateTableFunctionCall( ctx, functionTerm, @@ -254,7 +273,8 @@ object BridgingFunctionGenUtil { outputDataType, returnType, skipIfArgsNull, - contextTerm + contextTerm, + tableUdfMetricName ) } else if (udf.getKind == FunctionKind.ASYNC_TABLE) { generateAsyncTableFunctionCall( @@ -271,7 +291,7 @@ object BridgingFunctionGenUtil { returnType, outputDataType) } else { - generateScalarFunctionCall(ctx, functionTerm, externalOperands, outputDataType) + generateScalarFunctionCall(ctx, functionTerm, externalOperands, outputDataType, udfMetricName) } } @@ -351,7 +371,8 @@ object BridgingFunctionGenUtil { functionOutputDataType: DataType, outputType: LogicalType, skipIfArgsNull: Boolean, - contextTerm: Option[String] = None): GeneratedExpression = { + contextTerm: Option[String] = None, + udfMetricName: Option[String] = None): GeneratedExpression = { val resultCollectorTerm = generateResultCollector(ctx, functionOutputDataType, outputType) val setCollectorCode = s""" @@ -361,19 +382,25 @@ object BridgingFunctionGenUtil { val contextOperand = contextTerm.map(c => c + ", ").getOrElse("") + // The table eval is void (output flows through the collector); wrap the statement without + // capturing a result. + val evalStatement = + s"$functionTerm.eval($contextOperand${externalOperands.map(_.resultTerm).mkString(", ")});" + val instrumentedEval = instrumentUdfEval(ctx, udfMetricName, evalStatement) + val functionCallCode = if (skipIfArgsNull) { s""" |${externalOperands.map(_.code).mkString("\n")} |if (${externalOperands.map(_.nullTerm).mkString(" || ")}) { | // skip |} else { - | $functionTerm.eval($contextOperand${externalOperands.map(_.resultTerm).mkString(", ")}); + | $instrumentedEval |} |""".stripMargin } else { s""" |${externalOperands.map(_.code).mkString("\n")} - |$functionTerm.eval($contextOperand${externalOperands.map(_.resultTerm).mkString(", ")}); + |$instrumentedEval |""".stripMargin } @@ -497,11 +524,49 @@ object BridgingFunctionGenUtil { resultCollectorTerm } + /** + * Brackets a single UDF `eval` statement with sampled timing and exception counting when UDF + * metrics are enabled for this call. Returns the statement unchanged when the caller opted out + * ([[None]]) or the feature is disabled, so the generated code is then byte-identical to the + * un-instrumented path. Only the eval statement itself is wrapped — never the operand preparation + * — so nested UDFs time disjointly. + */ + private def instrumentUdfEval( + ctx: CodeGeneratorContext, + udfMetricName: Option[String], + evalStatement: String): String = { + udfMetricName match { + case Some(name) + if ctx.tableConfig.get(ExecutionConfigOptions.TABLE_EXEC_UDF_METRIC_ENABLED) => + val sampleInterval = + ctx.tableConfig + .get(ExecutionConfigOptions.TABLE_EXEC_UDF_METRIC_SAMPLE_INTERVAL) + .intValue() + val metricsTerm = ctx.addReusableUdfMetrics(name, sampleInterval) + val sampleTerm = ctx.addReusableLocalVariable("boolean", "udfSample") + val startNanosTerm = ctx.addReusableLocalVariable("long", "udfStartNanos") + val exceptionTerm = newName(ctx, "udfException") + s"""$sampleTerm = $metricsTerm.shouldSample(); + |$startNanosTerm = $sampleTerm ? System.nanoTime() : 0L; + |try { + | $evalStatement + |} catch (Throwable $exceptionTerm) { + | $metricsTerm.markException(); + | throw $exceptionTerm; + |} + |if ($sampleTerm) { + | $metricsTerm.update(System.nanoTime() - $startNanosTerm); + |}""".stripMargin + case _ => evalStatement + } + } + private def generateScalarFunctionCall( ctx: CodeGeneratorContext, functionTerm: String, externalOperands: Seq[GeneratedExpression], - outputDataType: DataType): GeneratedExpression = { + outputDataType: DataType, + udfMetricName: Option[String] = None): GeneratedExpression = { // result conversion val externalResultClass = outputDataType.getConversionClass @@ -516,11 +581,16 @@ object BridgingFunctionGenUtil { s"($externalResultTypeTerm) (${typeTerm(externalResultClassBoxed)})" } val externalResultTerm = ctx.addReusableLocalVariable(externalResultTypeTerm, "externalResult") + // Bracket only the eval assignment (not the operand preparation), so nested UDFs f(g(a)) time + // disjointly. + val evalStatement = + s"""$externalResultTerm = $externalResultCasting $functionTerm + | .$SCALAR_EVAL(${externalOperands.map(_.resultTerm).mkString(", ")});""".stripMargin + val instrumentedEval = instrumentUdfEval(ctx, udfMetricName, evalStatement) val externalCode = s""" |${externalOperands.map(_.code).mkString("\n")} - |$externalResultTerm = $externalResultCasting $functionTerm - | .$SCALAR_EVAL(${externalOperands.map(_.resultTerm).mkString(", ")}); + |$instrumentedEval |""".stripMargin val internalExpr = genToInternalConverterAll(ctx, outputDataType, externalResultTerm) diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/BridgingSqlFunctionCallGen.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/BridgingSqlFunctionCallGen.scala index 83223e63200d9..b78091f0bb19e 100644 --- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/BridgingSqlFunctionCallGen.scala +++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/BridgingSqlFunctionCallGen.scala @@ -76,6 +76,10 @@ class BridgingSqlFunctionCallGen(call: RexCall, rexProgram: RexProgram) extends callContext, udf, function.toString, - skipIfArgsNull = false) + skipIfArgsNull = false, + // Only the true user-UDF entry opts into metrics; lookup/ML/vector reuse the same util with + // the default None and stay un-instrumented. + udfMetricName = Some(function.getName) + ) } } diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/stream/sql/UdfMetricsITCase.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/stream/sql/UdfMetricsITCase.java new file mode 100644 index 0000000000000..5802ae8d0c5d1 --- /dev/null +++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/stream/sql/UdfMetricsITCase.java @@ -0,0 +1,296 @@ +/* + * 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.planner.runtime.stream.sql; + +import org.apache.flink.api.common.JobID; +import org.apache.flink.configuration.Configuration; +import org.apache.flink.configuration.RestartStrategyOptions; +import org.apache.flink.metrics.Counter; +import org.apache.flink.metrics.Histogram; +import org.apache.flink.metrics.Metric; +import org.apache.flink.runtime.testutils.InMemoryReporter; +import org.apache.flink.runtime.testutils.MiniClusterResourceConfiguration; +import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; +import org.apache.flink.table.api.TableResult; +import org.apache.flink.table.api.bridge.java.StreamTableEnvironment; +import org.apache.flink.table.api.config.ExecutionConfigOptions; +import org.apache.flink.table.functions.ScalarFunction; +import org.apache.flink.table.functions.TableFunction; +import org.apache.flink.table.planner.factories.TestValuesTableFactory; +import org.apache.flink.test.junit5.MiniClusterExtension; +import org.apache.flink.types.Row; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.RegisterExtension; + +import java.util.Arrays; +import java.util.List; +import java.util.Map; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** + * End-to-end tests for the opt-in per-operator UDF metrics (FLIP-485) registered under {@code + * .udf.} for sync scalar and table user-defined functions. + */ +class UdfMetricsITCase { + + private static final InMemoryReporter reporter = InMemoryReporter.createWithRetainedMetrics(); + + @RegisterExtension + private static final MiniClusterExtension MINI_CLUSTER = + new MiniClusterExtension( + new MiniClusterResourceConfiguration.Builder() + .setNumberTaskManagers(1) + .setNumberSlotsPerTaskManager(1) + .setConfiguration(reporter.addToConfiguration(new Configuration())) + .build()); + + private static final List SOURCE_ROWS = Arrays.asList(Row.of(1), Row.of(2), Row.of(3)); + + // Matches the processing-time metric of any UDF, for asserting that none is registered. + private static final String ANY_PROCESSING_TIME_PATTERN = "\\.udf\\..*\\.udfProcessingTime"; + private static final String ANY_EXCEPTION_COUNT_PATTERN = "\\.udf\\..*\\.udfExceptionCount"; + + /** The metric identifier group is {@code .udf.}. */ + private static String processingTimePattern(String udfName) { + return "\\.udf\\." + udfName + "\\.udfProcessingTime"; + } + + private static String exceptionCountPattern(String udfName) { + return "\\.udf\\." + udfName + "\\.udfExceptionCount"; + } + + /** Doubles an int; the metered sync scalar path. */ + public static class IntDoubler extends ScalarFunction { + public Integer eval(Integer i) { + return i == null ? null : i * 2; + } + } + + /** Negates an int; a second distinct scalar function. */ + public static class IntNegator extends ScalarFunction { + public Integer eval(Integer i) { + return i == null ? null : -i; + } + } + + /** Always throws; used to exercise the exception counter. */ + public static class AlwaysThrows extends ScalarFunction { + public Integer eval(Integer i) { + throw new RuntimeException("boom"); + } + } + + /** Emits each input twice; the metered sync table path. */ + public static class DuplicateRows extends TableFunction { + public void eval(Integer i) { + collect(i); + collect(i); + } + } + + @Test + void testSyncScalarMetricsRecorded() throws Exception { + StreamTableEnvironment tEnv = createTableEnv(true); + tEnv.createTemporarySystemFunction("scalarudf", IntDoubler.class); + createSource(tEnv, "src", "id INT"); + createBlackHoleSink(tEnv, "sink", "v INT"); + + JobID jobId = execute(tEnv, "INSERT INTO sink SELECT scalarudf(id) FROM src"); + + Histogram processingTime = histogram(jobId, processingTimePattern("scalarudf")); + // Sample interval 1 measures every invocation: one per input row. + assertThat(processingTime.getCount()).isEqualTo(SOURCE_ROWS.size()); + assertThat(counter(jobId, exceptionCountPattern("scalarudf")).getCount()).isZero(); + } + + @Test + void testSyncTableMetricsRecorded() throws Exception { + StreamTableEnvironment tEnv = createTableEnv(true); + tEnv.createTemporarySystemFunction("tableudf", DuplicateRows.class); + createSource(tEnv, "src", "id INT"); + createBlackHoleSink(tEnv, "sink", "v INT"); + + JobID jobId = + execute( + tEnv, + "INSERT INTO sink SELECT x FROM src, LATERAL TABLE(tableudf(id)) AS T(x)"); + + // eval is called once per input row (it emits two rows internally); one sample each. + assertThat(histogram(jobId, processingTimePattern("tableudf")).getCount()) + .isEqualTo(SOURCE_ROWS.size()); + } + + @Test + void testExceptionCounted() throws Exception { + StreamTableEnvironment tEnv = createTableEnv(true); + tEnv.createTemporarySystemFunction("boomudf", AlwaysThrows.class); + createSource(tEnv, "src", "id INT"); + createBlackHoleSink(tEnv, "sink", "v INT"); + + TableResult result = tEnv.executeSql("INSERT INTO sink SELECT boomudf(id) FROM src"); + JobID jobId = result.getJobClient().get().getJobID(); + assertThatThrownBy(result::await).isInstanceOf(Exception.class); + + assertThat(counter(jobId, exceptionCountPattern("boomudf")).getCount()) + .isGreaterThanOrEqualTo(1); + } + + @Test + void testDisabledRegistersNoUdfMetrics() throws Exception { + StreamTableEnvironment tEnv = createTableEnv(false); + tEnv.createTemporarySystemFunction("offudf", IntDoubler.class); + createSource(tEnv, "src", "id INT"); + createBlackHoleSink(tEnv, "sink", "v INT"); + + JobID jobId = execute(tEnv, "INSERT INTO sink SELECT offudf(id) FROM src"); + + assertThat(reporter.findMetrics(jobId, ANY_PROCESSING_TIME_PATTERN)).isEmpty(); + assertThat(reporter.findMetrics(jobId, ANY_EXCEPTION_COUNT_PATTERN)).isEmpty(); + } + + @Test + void testLookupJoinNotMetered() throws Exception { + StreamTableEnvironment tEnv = createTableEnv(true); + String probeId = TestValuesTableFactory.registerData(SOURCE_ROWS); + tEnv.executeSql( + "CREATE TABLE probe (id INT, proctime AS PROCTIME()) WITH (" + + "'connector' = 'values', 'bounded' = 'true', 'data-id' = '" + + probeId + + "')"); + String dimId = + TestValuesTableFactory.registerData( + Arrays.asList(Row.of(1, "a"), Row.of(2, "b"), Row.of(3, "c"))); + tEnv.executeSql( + "CREATE TABLE dim (id INT, name STRING) WITH (" + + "'connector' = 'values', 'data-id' = '" + + dimId + + "')"); + createBlackHoleSink(tEnv, "sink", "v STRING"); + + JobID jobId = + execute( + tEnv, + "INSERT INTO sink SELECT dim.name FROM probe " + + "JOIN dim FOR SYSTEM_TIME AS OF probe.proctime " + + "ON probe.id = dim.id"); + + // The lookup function reuses the same codegen util but opts out of metrics (SD1). + assertThat(reporter.findMetrics(jobId, ANY_PROCESSING_TIME_PATTERN)).isEmpty(); + } + + @Test + void testRepeatedFunctionSharesOneHandle() throws Exception { + StreamTableEnvironment tEnv = createTableEnv(true); + tEnv.createTemporarySystemFunction("sharedudf", IntDoubler.class); + String dataId = + TestValuesTableFactory.registerData( + Arrays.asList(Row.of(1, 10), Row.of(2, 20), Row.of(3, 30))); + tEnv.executeSql( + "CREATE TABLE src2 (a INT, b INT) WITH (" + + "'connector' = 'values', 'bounded' = 'true', 'data-id' = '" + + dataId + + "')"); + createBlackHoleSink(tEnv, "sink", "v INT"); + + JobID jobId = + execute(tEnv, "INSERT INTO sink SELECT sharedudf(a) + sharedudf(b) FROM src2"); + + // Two call sites of one function in one operator share a single handle, so both feed the + // same histogram: 2 evals per row. Without sharing, the second registration is dropped and + // only one call site's timings survive (one per row). + Map processingTimes = + reporter.findMetrics(jobId, processingTimePattern("sharedudf")); + assertThat(processingTimes).hasSize(1); + assertThat(((Histogram) processingTimes.values().iterator().next()).getCount()) + .isEqualTo(2L * SOURCE_ROWS.size()); + } + + @Test + void testDistinctFunctionsGetSeparateHandles() throws Exception { + StreamTableEnvironment tEnv = createTableEnv(true); + tEnv.createTemporarySystemFunction("doubleudf", IntDoubler.class); + tEnv.createTemporarySystemFunction("negateudf", IntNegator.class); + createSource(tEnv, "src", "id INT"); + createBlackHoleSink(tEnv, "sink", "d INT, n INT"); + + JobID jobId = + execute(tEnv, "INSERT INTO sink SELECT doubleudf(id), negateudf(id) FROM src"); + + // Two distinct functions register two separate udf. handles, each with its own + // histogram counting one sample per input row. + assertThat(histogram(jobId, processingTimePattern("doubleudf")).getCount()) + .isEqualTo(SOURCE_ROWS.size()); + assertThat(histogram(jobId, processingTimePattern("negateudf")).getCount()) + .isEqualTo(SOURCE_ROWS.size()); + } + + private static StreamTableEnvironment createTableEnv(boolean udfMetricEnabled) { + Configuration conf = new Configuration(); + conf.set(RestartStrategyOptions.RESTART_STRATEGY, "disable"); + StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf); + env.setParallelism(1); + StreamTableEnvironment tEnv = StreamTableEnvironment.create(env); + tEnv.getConfig() + .set(ExecutionConfigOptions.TABLE_EXEC_UDF_METRIC_ENABLED, udfMetricEnabled); + tEnv.getConfig().set(ExecutionConfigOptions.TABLE_EXEC_UDF_METRIC_SAMPLE_INTERVAL, 1); + return tEnv; + } + + private static void createSource(StreamTableEnvironment tEnv, String name, String schema) { + String dataId = TestValuesTableFactory.registerData(SOURCE_ROWS); + tEnv.executeSql( + "CREATE TABLE " + + name + + " (" + + schema + + ") WITH ('connector' = 'values', 'bounded' = 'true', 'data-id' = '" + + dataId + + "')"); + } + + private static void createBlackHoleSink( + StreamTableEnvironment tEnv, String name, String schema) { + tEnv.executeSql( + "CREATE TABLE " + name + " (" + schema + ") WITH ('connector' = 'blackhole')"); + } + + private static JobID execute(StreamTableEnvironment tEnv, String insert) throws Exception { + TableResult result = tEnv.executeSql(insert); + JobID jobId = result.getJobClient().get().getJobID(); + result.await(); + return jobId; + } + + private static Histogram histogram(JobID jobId, String pattern) { + return (Histogram) singleMetric(jobId, pattern); + } + + private static Counter counter(JobID jobId, String pattern) { + return (Counter) singleMetric(jobId, pattern); + } + + private static Metric singleMetric(JobID jobId, String pattern) { + Map found = reporter.findMetrics(jobId, pattern); + assertThat(found).hasSize(1); + return found.values().iterator().next(); + } +} From f0d286a53a62b89dcdcd76ea1405e3b476320cd3 Mon Sep 17 00:00:00 2001 From: weiqingy Date: Wed, 8 Jul 2026 10:24:19 -0700 Subject: [PATCH 4/5] [FLINK-38071][table-planner] Instrument async scalar and table UDF calls with metrics Extend UDF metrics to asynchronous scalar and table user-defined functions. The UdfMetrics handle is built once in the generated fetcher's open() and carried on the per-invocation delegating future. The sampling decision and start time are captured at dispatch on the task thread; udfProcessingTime (full dispatch-to-completion span) and udfExceptionCount are recorded at completion on the callback thread, touching only the synchronized histogram and thread-safe counter. Instrumentation is emitted only when table.exec.udf-metric-enabled is true; the generated code is byte-identical when disabled. --- .../planner/codegen/AsyncCodeGenerator.java | 9 +- .../codegen/CodeGeneratorContext.scala | 15 ++ .../calls/BridgingFunctionGenUtil.scala | 98 ++++++++--- .../runtime/stream/sql/UdfMetricsITCase.java | 155 ++++++++++++++++++ .../async/DelegatingAsyncResultFuture.java | 33 ++++ .../DelegatingAsyncTableResultFuture.java | 36 ++++ 6 files changed, 323 insertions(+), 23 deletions(-) diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/codegen/AsyncCodeGenerator.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/codegen/AsyncCodeGenerator.java index 5a4bbc1118d55..d027e14ca74ed 100644 --- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/codegen/AsyncCodeGenerator.java +++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/codegen/AsyncCodeGenerator.java @@ -152,9 +152,16 @@ private static String generateProcessCode( index++; } + // The async scalar call generated above registers the shared UdfMetrics handle when metrics + // are enabled; pass it into the per-invocation delegating future, which does the timing and + // exception counting. Null (feature off) yields the original two-argument construction. + String udfMetricsTerm = ctx.getSingleUdfMetricsTerm(); + String metricsCtorArg = udfMetricsTerm == null ? "" : ", " + udfMetricsTerm; + Map values = new HashMap<>(); values.put("delegatingFutureTerm", delegatingFutureTerm); values.put("delegatingFutureType", DelegatingAsyncResultFuture.class.getCanonicalName()); + values.put("metricsCtorArg", metricsCtorArg); values.put("collectorTerm", collectorTerm); values.put("typeTerm", GenericRowData.class.getCanonicalName()); values.put("recordTerm", recordTerm); @@ -169,7 +176,7 @@ private static String generateProcessCode( "\n", new String[] { "final ${delegatingFutureType} ${delegatingFutureTerm} ", - " = new ${delegatingFutureType}(${collectorTerm}, ${fieldCount});", + " = new ${delegatingFutureType}(${collectorTerm}, ${fieldCount}${metricsCtorArg});", "final org.apache.flink.types.RowKind rowKind = ${inputTerm}.getRowKind();\n", "try {", // Ensure that metadata setup come first so that we know that they're diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/CodeGeneratorContext.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/CodeGeneratorContext.scala index 2edf0dbac83ee..372713e8828f7 100644 --- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/CodeGeneratorContext.scala +++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/CodeGeneratorContext.scala @@ -33,6 +33,7 @@ import org.apache.flink.table.types.logical._ import org.apache.flink.table.types.logical.LogicalTypeRoot._ import org.apache.flink.table.utils.{DateTimeUtils, EncodingUtils} import org.apache.flink.util.InstantiationUtil +import org.apache.flink.util.Preconditions import java.time.ZoneId import java.util.TimeZone @@ -931,6 +932,20 @@ class CodeGeneratorContext( ) } + /** + * Returns the sole [[UdfMetrics]] handle term registered in this context, or `null` if none is. + * An async fetcher hosts exactly one async UDF, so at most one handle is ever registered; this + * lets the async scalar generator pass the handle into the per-invocation delegating future + * without re-deriving the UDF name. + */ + def getSingleUdfMetricsTerm: String = { + val errorMessage: Any = + s"An async fetcher hosts exactly one async UDF, but ${reusableUdfMetricsTerms.size} UDF " + + "metrics handles were registered in one context." + Preconditions.checkState(reusableUdfMetricsTerms.size <= 1, errorMessage) + if (reusableUdfMetricsTerms.size == 1) reusableUdfMetricsTerms.values.head else null + } + /** * Adds a reusable [[DataStructureConverter]] to the member area of the generated class. * diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/BridgingFunctionGenUtil.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/BridgingFunctionGenUtil.scala index 770934d4ed95f..defbe8ad37c02 100644 --- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/BridgingFunctionGenUtil.scala +++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/BridgingFunctionGenUtil.scala @@ -278,18 +278,21 @@ object BridgingFunctionGenUtil { ) } else if (udf.getKind == FunctionKind.ASYNC_TABLE) { generateAsyncTableFunctionCall( + ctx, functionTerm, externalOperands, returnType, outputDataType, - skipIfArgsNull) + skipIfArgsNull, + udfMetricName) } else if (udf.getKind == FunctionKind.ASYNC_SCALAR) { generateAsyncScalarFunctionCall( ctx, functionTerm, externalOperands, returnType, - outputDataType) + outputDataType, + udfMetricName) } else { generateScalarFunctionCall(ctx, functionTerm, externalOperands, outputDataType, udfMetricName) } @@ -409,11 +412,13 @@ object BridgingFunctionGenUtil { } private def generateAsyncTableFunctionCall( + ctx: CodeGeneratorContext, functionTerm: String, externalOperands: Seq[GeneratedExpression], returnType: LogicalType, outputDataType: DataType, - skipIfArgsNull: Boolean): GeneratedExpression = { + skipIfArgsNull: Boolean, + udfMetricName: Option[String]): GeneratedExpression = { val DELEGATE_ASYNC_TABLE = className[DelegatingAsyncTableResultFuture] val outputType = outputDataType.getLogicalType @@ -428,6 +433,17 @@ object BridgingFunctionGenUtil { ) ++ externalOperands.map(_.resultTerm) val anyNull = externalOperands.map(_.nullTerm) ++ Seq("false") + // When metrics are enabled the handle is passed into the delegating future, which takes the + // sample decision at construction (dispatch, task thread) and records the completion span in + // its callback. The extra ctor argument is omitted when off, keeping the code byte-identical. + val metricsTerm = udfMetricsTermIfEnabled(ctx, udfMetricName) + val metricsCtorArg = metricsTerm.map(t => s", $t").getOrElse("") + val constructDelegate = + s"""$DELEGATE_ASYNC_TABLE delegates = new $DELEGATE_ASYNC_TABLE($DEFAULT_COLLECTOR_TERM, + | $needsWrapping, $isInternal$metricsCtorArg);""".stripMargin + val instrumentedEval = + instrumentAsyncDispatch(ctx, metricsTerm, s"$functionTerm.eval(${arguments.mkString(", ")});") + val functionCallCode = { if (skipIfArgsNull) { s""" @@ -435,17 +451,15 @@ object BridgingFunctionGenUtil { |if (${anyNull.mkString(" || ")}) { | $DEFAULT_COLLECTOR_TERM.complete(java.util.Collections.emptyList()); |} else { - | $DELEGATE_ASYNC_TABLE delegates = new $DELEGATE_ASYNC_TABLE($DEFAULT_COLLECTOR_TERM, - | $needsWrapping, $isInternal); - | $functionTerm.eval(${arguments.mkString(", ")}); + | $constructDelegate + | $instrumentedEval |} |""".stripMargin } else { s""" |${externalOperands.map(_.code).mkString("\n")} - |$DELEGATE_ASYNC_TABLE delegates = new $DELEGATE_ASYNC_TABLE($DEFAULT_COLLECTOR_TERM, - | $needsWrapping, $isInternal); - | $functionTerm.eval(${arguments.mkString(", ")}); + |$constructDelegate + | $instrumentedEval |""".stripMargin } } @@ -459,17 +473,25 @@ object BridgingFunctionGenUtil { functionTerm: String, externalOperands: Seq[GeneratedExpression], outputType: LogicalType, - outputDataType: DataType): GeneratedExpression = { + outputDataType: DataType, + udfMetricName: Option[String]): GeneratedExpression = { val converterTerm = ctx.addReusableConverter(outputDataType) + // Registering the handle here lets the async scalar fetcher pass it into the delegating future + // (see AsyncCodeGenerator); the future then takes the sample decision in createAsyncFuture on + // the task thread and records the completion span in its callback. + val metricsTerm = udfMetricsTermIfEnabled(ctx, udfMetricName) + val evalStatement = + s"""$functionTerm.eval( + | $DEFAULT_DELEGATING_FUTURE_TERM.createAsyncFuture($converterTerm), + | ${externalOperands.map(_.resultTerm).mkString(", ")});""".stripMargin + val instrumentedEval = instrumentAsyncDispatch(ctx, metricsTerm, evalStatement) val functionCallCode = s""" |${externalOperands.map(_.code).mkString("\n")} |if (${externalOperands.map(_.nullTerm).mkString(" || ")}) { | $DEFAULT_DELEGATING_FUTURE_TERM.createAsyncFuture($converterTerm).complete(null); |} else { - | $functionTerm.eval( - | $DEFAULT_DELEGATING_FUTURE_TERM.createAsyncFuture($converterTerm), - | ${externalOperands.map(_.resultTerm).mkString(", ")}); + | $instrumentedEval |} |""".stripMargin @@ -535,14 +557,8 @@ object BridgingFunctionGenUtil { ctx: CodeGeneratorContext, udfMetricName: Option[String], evalStatement: String): String = { - udfMetricName match { - case Some(name) - if ctx.tableConfig.get(ExecutionConfigOptions.TABLE_EXEC_UDF_METRIC_ENABLED) => - val sampleInterval = - ctx.tableConfig - .get(ExecutionConfigOptions.TABLE_EXEC_UDF_METRIC_SAMPLE_INTERVAL) - .intValue() - val metricsTerm = ctx.addReusableUdfMetrics(name, sampleInterval) + udfMetricsTermIfEnabled(ctx, udfMetricName) match { + case Some(metricsTerm) => val sampleTerm = ctx.addReusableLocalVariable("boolean", "udfSample") val startNanosTerm = ctx.addReusableLocalVariable("long", "udfStartNanos") val exceptionTerm = newName(ctx, "udfException") @@ -557,10 +573,48 @@ object BridgingFunctionGenUtil { |if ($sampleTerm) { | $metricsTerm.update(System.nanoTime() - $startNanosTerm); |}""".stripMargin - case _ => evalStatement + case None => evalStatement } } + /** + * Returns the shared [[UdfMetrics]] handle term when metrics are enabled for this call, else + * [[None]]. Acquiring the term registers the handle member and its `open()` registration exactly + * once per UDF name in the current context. + */ + private def udfMetricsTermIfEnabled( + ctx: CodeGeneratorContext, + udfMetricName: Option[String]): Option[String] = udfMetricName match { + case Some(name) if ctx.tableConfig.get(ExecutionConfigOptions.TABLE_EXEC_UDF_METRIC_ENABLED) => + val sampleInterval = + ctx.tableConfig + .get(ExecutionConfigOptions.TABLE_EXEC_UDF_METRIC_SAMPLE_INTERVAL) + .intValue() + Some(ctx.addReusableUdfMetrics(name, sampleInterval)) + case _ => None + } + + /** + * Brackets an async UDF dispatch `eval` so a synchronous throw (before the future is handed to + * the framework) still increments the exception counter, mirroring the sync path. Exceptional + * completions are counted separately in the delegating future's completion callback. Returns the + * statement unchanged when metrics are disabled, keeping the generated code byte-identical. + */ + private def instrumentAsyncDispatch( + ctx: CodeGeneratorContext, + metricsTerm: Option[String], + evalStatement: String): String = metricsTerm match { + case Some(term) => + val exceptionTerm = newName(ctx, "udfException") + s"""try { + | $evalStatement + |} catch (Throwable $exceptionTerm) { + | $term.markException(); + | throw $exceptionTerm; + |}""".stripMargin + case None => evalStatement + } + private def generateScalarFunctionCall( ctx: CodeGeneratorContext, functionTerm: String, diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/stream/sql/UdfMetricsITCase.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/stream/sql/UdfMetricsITCase.java index 5802ae8d0c5d1..c98669e3b05e4 100644 --- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/stream/sql/UdfMetricsITCase.java +++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/stream/sql/UdfMetricsITCase.java @@ -30,6 +30,9 @@ import org.apache.flink.table.api.TableResult; import org.apache.flink.table.api.bridge.java.StreamTableEnvironment; import org.apache.flink.table.api.config.ExecutionConfigOptions; +import org.apache.flink.table.functions.AsyncScalarFunction; +import org.apache.flink.table.functions.AsyncTableFunction; +import org.apache.flink.table.functions.FunctionContext; import org.apache.flink.table.functions.ScalarFunction; import org.apache.flink.table.functions.TableFunction; import org.apache.flink.table.planner.factories.TestValuesTableFactory; @@ -40,8 +43,14 @@ import org.junit.jupiter.api.extension.RegisterExtension; import java.util.Arrays; +import java.util.Collection; import java.util.List; import java.util.Map; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatThrownBy; @@ -107,6 +116,88 @@ public void eval(Integer i) { } } + /** Doubles an int off the task thread; the metered async scalar path. */ + public static class AsyncIntDoubler extends AsyncScalarFunction { + private transient ScheduledExecutorService executor; + + @Override + public void open(FunctionContext context) { + executor = Executors.newSingleThreadScheduledExecutor(); + } + + @Override + public void close() { + if (executor != null) { + executor.shutdownNow(); + } + } + + public void eval(CompletableFuture future, Integer i) { + executor.schedule( + () -> future.complete(i == null ? null : i * 2), 5, TimeUnit.MILLISECONDS); + } + } + + /** + * Completes exceptionally the first {@code numFailures} invocations, then succeeds. Exercises + * the async completion-exception counter while the async operator's retry keeps the job alive. + */ + public static class AsyncFlaky extends AsyncScalarFunction { + private final int numFailures; + private final AtomicInteger failures = new AtomicInteger(); + private transient ScheduledExecutorService executor; + + public AsyncFlaky(int numFailures) { + this.numFailures = numFailures; + } + + @Override + public void open(FunctionContext context) { + executor = Executors.newSingleThreadScheduledExecutor(); + } + + @Override + public void close() { + if (executor != null) { + executor.shutdownNow(); + } + } + + public void eval(CompletableFuture future, Integer i) { + executor.schedule( + () -> { + if (failures.getAndIncrement() < numFailures) { + future.completeExceptionally(new RuntimeException("boom")); + } else { + future.complete(i); + } + }, + 5, + TimeUnit.MILLISECONDS); + } + } + + /** Emits each input twice off the task thread; the metered async table path. */ + public static class AsyncDuplicateRows extends AsyncTableFunction { + private transient ScheduledExecutorService executor; + + @Override + public void open(FunctionContext context) { + executor = Executors.newSingleThreadScheduledExecutor(); + } + + @Override + public void close() { + if (executor != null) { + executor.shutdownNow(); + } + } + + public void eval(CompletableFuture> future, Integer i) { + executor.schedule(() -> future.complete(Arrays.asList(i, i)), 5, TimeUnit.MILLISECONDS); + } + } + @Test void testSyncScalarMetricsRecorded() throws Exception { StreamTableEnvironment tEnv = createTableEnv(true); @@ -243,6 +334,70 @@ void testDistinctFunctionsGetSeparateHandles() throws Exception { .isEqualTo(SOURCE_ROWS.size()); } + @Test + void testAsyncScalarMetricsRecorded() throws Exception { + StreamTableEnvironment tEnv = createTableEnv(true); + tEnv.createTemporarySystemFunction("asyncudf", AsyncIntDoubler.class); + createSource(tEnv, "src", "id INT"); + createBlackHoleSink(tEnv, "sink", "v INT"); + + JobID jobId = execute(tEnv, "INSERT INTO sink SELECT asyncudf(id) FROM src"); + + // The processing time spans dispatch to off-thread completion; one sample per input row. + assertThat(histogram(jobId, processingTimePattern("asyncudf")).getCount()) + .isEqualTo(SOURCE_ROWS.size()); + assertThat(counter(jobId, exceptionCountPattern("asyncudf")).getCount()).isZero(); + } + + @Test + void testAsyncCompletionExceptionSurvivesJob() throws Exception { + StreamTableEnvironment tEnv = createTableEnv(true); + // Serialize invocations so the shared failure counter drives a deterministic retry. + tEnv.getConfig() + .set(ExecutionConfigOptions.TABLE_EXEC_ASYNC_SCALAR_MAX_CONCURRENT_OPERATIONS, 1); + tEnv.createTemporarySystemFunction("flakyudf", new AsyncFlaky(2)); + createSource(tEnv, "src", "id INT"); + createBlackHoleSink(tEnv, "sink", "v INT"); + + // Two exceptional completions are counted, then the async retry succeeds and the job + // finishes normally: an exceptional completion is a soft error, not a job failure. + JobID jobId = execute(tEnv, "INSERT INTO sink SELECT flakyudf(id) FROM src"); + + assertThat(counter(jobId, exceptionCountPattern("flakyudf")).getCount()) + .isGreaterThanOrEqualTo(1); + } + + @Test + void testAsyncTableMetricsRecorded() throws Exception { + StreamTableEnvironment tEnv = createTableEnv(true); + tEnv.createTemporarySystemFunction("asynctableudf", AsyncDuplicateRows.class); + createSource(tEnv, "src", "id INT"); + createBlackHoleSink(tEnv, "sink", "v INT"); + + JobID jobId = + execute( + tEnv, + "INSERT INTO sink SELECT x FROM src, " + + "LATERAL TABLE(asynctableudf(id)) AS T(x)"); + + // eval is called once per input row (it completes two rows); one sample each. + assertThat(histogram(jobId, processingTimePattern("asynctableudf")).getCount()) + .isEqualTo(SOURCE_ROWS.size()); + } + + @Test + void testAsyncDisabledRegistersNoUdfMetrics() throws Exception { + StreamTableEnvironment tEnv = createTableEnv(false); + tEnv.createTemporarySystemFunction("asyncoffudf", AsyncIntDoubler.class); + createSource(tEnv, "src", "id INT"); + createBlackHoleSink(tEnv, "sink", "v INT"); + + JobID jobId = execute(tEnv, "INSERT INTO sink SELECT asyncoffudf(id) FROM src"); + + assertThat(reporter.findMetrics(jobId, ANY_PROCESSING_TIME_PATTERN)).isEmpty(); + assertThat(reporter.findMetrics(jobId, ANY_EXCEPTION_COUNT_PATTERN)).isEmpty(); + } + private static StreamTableEnvironment createTableEnv(boolean udfMetricEnabled) { Configuration conf = new Configuration(); conf.set(RestartStrategyOptions.RESTART_STRATEGY, "disable"); diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/calc/async/DelegatingAsyncResultFuture.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/calc/async/DelegatingAsyncResultFuture.java index 1421be97ce9f9..b9396f34953fd 100644 --- a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/calc/async/DelegatingAsyncResultFuture.java +++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/calc/async/DelegatingAsyncResultFuture.java @@ -22,9 +22,12 @@ import org.apache.flink.table.data.GenericRowData; import org.apache.flink.table.data.RowData; import org.apache.flink.table.data.conversion.DataStructureConverter; +import org.apache.flink.table.runtime.operators.metrics.UdfMetrics; import org.apache.flink.types.RowKind; import org.apache.flink.util.Preconditions; +import javax.annotation.Nullable; + import java.util.Collections; import java.util.HashMap; import java.util.Map; @@ -47,10 +50,26 @@ public class DelegatingAsyncResultFuture implements BiConsumer delegatedResultFuture, int totalResultSize) { + this(delegatedResultFuture, totalResultSize, null); + } + + public DelegatingAsyncResultFuture( + ResultFuture delegatedResultFuture, + int totalResultSize, + @Nullable UdfMetrics udfMetrics) { this.delegatedResultFuture = delegatedResultFuture; this.totalResultSize = totalResultSize; + this.udfMetrics = udfMetrics; } public synchronized void setRowKind(RowKind rowKind) { @@ -73,12 +92,26 @@ public CompletableFuture createAsyncFuture( Preconditions.checkState(this.asyncIndex >= 0); future = new CompletableFuture<>(); this.converter = converter; + // Sample decision taken on the task thread; the sampler counter is never touched + // off-thread. + if (udfMetrics != null) { + sample = udfMetrics.shouldSample(); + startNanos = sample ? System.nanoTime() : 0L; + } future.whenComplete(this); return future; } @Override public void accept(Object o, Throwable throwable) { + if (udfMetrics != null) { + if (throwable != null) { + udfMetrics.markException(); + } + if (sample) { + udfMetrics.update(System.nanoTime() - startNanos); + } + } if (throwable != null) { delegatedResultFuture.completeExceptionally(throwable); } else { diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/correlate/async/DelegatingAsyncTableResultFuture.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/correlate/async/DelegatingAsyncTableResultFuture.java index e745eca01a80c..0350933619e4c 100644 --- a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/correlate/async/DelegatingAsyncTableResultFuture.java +++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/correlate/async/DelegatingAsyncTableResultFuture.java @@ -20,8 +20,11 @@ import org.apache.flink.streaming.api.functions.async.ResultFuture; import org.apache.flink.table.data.GenericRowData; +import org.apache.flink.table.runtime.operators.metrics.UdfMetrics; import org.apache.flink.types.Row; +import javax.annotation.Nullable; + import java.util.ArrayList; import java.util.Collection; import java.util.List; @@ -41,21 +44,54 @@ public class DelegatingAsyncTableResultFuture implements BiConsumer> completableFuture; + // Null unless UDF metrics are enabled. The sample decision and start-time are taken on the task + // thread in the constructor (invoked at dispatch, before eval); the histogram/counter are + // updated at completion (accept, callback thread). The two updated metrics are internally + // thread-safe; sample/startNanos are published to the callback via the future's completion. + @Nullable private final UdfMetrics udfMetrics; + private boolean sample; + private long startNanos; + public DelegatingAsyncTableResultFuture( ResultFuture delegatedResultFuture, boolean needsWrapping, boolean isInternalResultType) { + this(delegatedResultFuture, needsWrapping, isInternalResultType, null); + } + + public DelegatingAsyncTableResultFuture( + ResultFuture delegatedResultFuture, + boolean needsWrapping, + boolean isInternalResultType, + @Nullable UdfMetrics udfMetrics) { this.delegatedResultFuture = delegatedResultFuture; this.wrapFunction = needsWrapping ? (isInternalResultType ? this::wrapInternal : this::wrapExternal) : outs -> outs; this.completableFuture = new CompletableFuture<>(); + this.udfMetrics = udfMetrics; + // Sample decision taken on the task thread; the sampler counter is never touched + // off-thread. These writes must precede whenComplete below: the callback registration + // performs the volatile completion-stack push that establishes the happens-before edge + // carrying sample/startNanos to the completing thread's accept(). + if (udfMetrics != null) { + sample = udfMetrics.shouldSample(); + startNanos = sample ? System.nanoTime() : 0L; + } this.completableFuture.whenComplete(this); } @Override public void accept(Collection outs, Throwable throwable) { + if (udfMetrics != null) { + if (throwable != null) { + udfMetrics.markException(); + } + if (sample) { + udfMetrics.update(System.nanoTime() - startNanos); + } + } if (throwable != null) { delegatedResultFuture.completeExceptionally(throwable); return; From 154a63700c09ea2872c211f1e5738bff04aa17e2 Mon Sep 17 00:00:00 2001 From: weiqingy Date: Wed, 8 Jul 2026 14:18:32 -0700 Subject: [PATCH 5/5] [FLINK-38071][docs] Document UDF metrics Add a UDF Metrics section to the metrics documentation describing udfProcessingTime and udfExceptionCount, their operator-scoped naming (.udf..), the table.exec.udf-metric-enabled and table.exec.udf-metric.sample-interval options, and the covered function kinds. Also add an integration test asserting udfProcessingTime reflects a real UDF invocation delay. --- docs/content.zh/docs/ops/metrics.md | 44 +++++++++++++++++++ docs/content/docs/ops/metrics.md | 44 +++++++++++++++++++ .../runtime/stream/sql/UdfMetricsITCase.java | 37 +++++++++++++++- 3 files changed, 124 insertions(+), 1 deletion(-) diff --git a/docs/content.zh/docs/ops/metrics.md b/docs/content.zh/docs/ops/metrics.md index 1a7562484ae2a..ce35cec95db57 100644 --- a/docs/content.zh/docs/ops/metrics.md +++ b/docs/content.zh/docs/ops/metrics.md @@ -2382,6 +2382,34 @@ logged by `SystemResourcesMetricsInitializer` during the startup. +### UDF + +The following metrics are registered per SQL/Table user-defined function when UDF metrics are enabled (see [UDF metrics](#udf-metrics) below). They are attached to the operator that runs the function and scoped under `udf.`, so the full metric identifier is `.udf..`. + + + + + + + + + + + + + + + + + + + + + + + +
ScopeMetricsDescriptionType
Task/Operatorudf.<udf_name>.udfProcessingTimeThe time (in nanoseconds) taken by a single invocation of the user-defined function. For asynchronous functions this spans from dispatch to completion. Sampled: only one invocation out of every table.exec.udf-metric.sample-interval is measured.Histogram
udf.<udf_name>.udfExceptionCountThe number of exceptions escaping the user-defined function. For synchronous functions this counts exceptions propagating out of eval; for asynchronous functions it counts exceptional completions. Counted on every invocation, not sampled.Counter
+ ## End-to-End latency tracking Flink allows to track the latency of records travelling through the system. This feature is disabled by default. @@ -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.`: 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. + +Warning 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. + ## REST API integration Metrics can be queried through the [Monitoring REST API]({{< ref "docs/ops/rest_api" >}}). diff --git a/docs/content/docs/ops/metrics.md b/docs/content/docs/ops/metrics.md index 63fdc05152344..2e6ef60cc6fff 100644 --- a/docs/content/docs/ops/metrics.md +++ b/docs/content/docs/ops/metrics.md @@ -2382,6 +2382,34 @@ Metrics below can be used to measure the effectiveness of speculative execution. +### UDF + +The following metrics are registered per SQL/Table user-defined function when UDF metrics are enabled (see [UDF metrics](#udf-metrics) below). They are attached to the operator that runs the function and scoped under `udf.`, so the full metric identifier is `.udf..`. + + + + + + + + + + + + + + + + + + + + + + + +
ScopeMetricsDescriptionType
Task/Operatorudf.<udf_name>.udfProcessingTimeThe time (in nanoseconds) taken by a single invocation of the user-defined function. For asynchronous functions this spans from dispatch to completion. Sampled: only one invocation out of every table.exec.udf-metric.sample-interval is measured.Histogram
udf.<udf_name>.udfExceptionCountThe number of exceptions escaping the user-defined function. For synchronous functions this counts exceptions propagating out of eval; for asynchronous functions it counts exceptional completions. Counted on every invocation, not sampled.Counter
+ ## End-to-End latency tracking Flink allows to track the latency of records travelling through the system. This feature is disabled by default. @@ -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.`: 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. + +Warning 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. + ## REST API integration Metrics can be queried through the [Monitoring REST API]({{< ref "docs/ops/rest_api" >}}). diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/stream/sql/UdfMetricsITCase.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/stream/sql/UdfMetricsITCase.java index c98669e3b05e4..dadcfcddc6041 100644 --- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/stream/sql/UdfMetricsITCase.java +++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/stream/sql/UdfMetricsITCase.java @@ -23,6 +23,7 @@ import org.apache.flink.configuration.RestartStrategyOptions; import org.apache.flink.metrics.Counter; import org.apache.flink.metrics.Histogram; +import org.apache.flink.metrics.HistogramStatistics; import org.apache.flink.metrics.Metric; import org.apache.flink.runtime.testutils.InMemoryReporter; import org.apache.flink.runtime.testutils.MiniClusterResourceConfiguration; @@ -57,7 +58,8 @@ /** * End-to-end tests for the opt-in per-operator UDF metrics (FLIP-485) registered under {@code - * .udf.} for sync scalar and table user-defined functions. + * .udf.} for synchronous and asynchronous scalar and table user-defined + * functions. */ class UdfMetricsITCase { @@ -74,6 +76,8 @@ class UdfMetricsITCase { private static final List SOURCE_ROWS = Arrays.asList(Row.of(1), Row.of(2), Row.of(3)); + private static final long SLEEP_MILLIS = 20; + // Matches the processing-time metric of any UDF, for asserting that none is registered. private static final String ANY_PROCESSING_TIME_PATTERN = "\\.udf\\..*\\.udfProcessingTime"; private static final String ANY_EXCEPTION_COUNT_PATTERN = "\\.udf\\..*\\.udfExceptionCount"; @@ -108,6 +112,18 @@ public Integer eval(Integer i) { } } + /** Doubles an int after sleeping a fixed duration, so timing is measurably non-zero. */ + public static class SleepyDoubler extends ScalarFunction { + public Integer eval(Integer i) { + try { + Thread.sleep(SLEEP_MILLIS); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + return i == null ? null : i * 2; + } + } + /** Emits each input twice; the metered sync table path. */ public static class DuplicateRows extends TableFunction { public void eval(Integer i) { @@ -213,6 +229,25 @@ void testSyncScalarMetricsRecorded() throws Exception { assertThat(counter(jobId, exceptionCountPattern("scalarudf")).getCount()).isZero(); } + @Test + void testProcessingTimeReflectsDelay() throws Exception { + StreamTableEnvironment tEnv = createTableEnv(true); + tEnv.createTemporarySystemFunction("sleepyudf", SleepyDoubler.class); + createSource(tEnv, "src", "id INT"); + createBlackHoleSink(tEnv, "sink", "v INT"); + + JobID jobId = execute(tEnv, "INSERT INTO sink SELECT sleepyudf(id) FROM src"); + + HistogramStatistics stats = + histogram(jobId, processingTimePattern("sleepyudf")).getStatistics(); + // Every invocation is measured (interval 1), so the histogram must reflect the induced + // delay and expose percentile statistics rather than just a sample count. + long minExpectedNanos = TimeUnit.MILLISECONDS.toNanos(SLEEP_MILLIS); + assertThat(stats.getMax()).isGreaterThanOrEqualTo(minExpectedNanos); + assertThat(stats.getMean()).isGreaterThanOrEqualTo(minExpectedNanos); + assertThat(stats.getQuantile(0.95)).isPositive(); + } + @Test void testSyncTableMetricsRecorded() throws Exception { StreamTableEnvironment tEnv = createTableEnv(true);