This is an automated email from the ASF dual-hosted git repository.
HuangZhenQiu pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 44db1fc214f [FLINK-40293][table] Add UDF metric config options and
instrument sync scalar and table UDF calls (#28879)
44db1fc214f is described below
commit 44db1fc214fe05324370217d7cb7b5ca046b7bbf
Author: Weiqing Yang <[email protected]>
AuthorDate: Fri Oct 2 21:09:46 2026 -0700
[FLINK-40293][table] Add UDF metric config options and instrument sync
scalar and table UDF calls (#28879)
* [FLINK-40293][table] Add UDF metric config options and instrument sync
scalar and table UDF calls
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).
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.<udfName>. 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.
* [FLINK-40293][table] Address review comments on UDF metric options and
tests
Document that table.exec.udf-metric.sample-interval must be at least 1 and
that
a value of 1 measures every invocation, matching the check in
UdfMetrics.register, and regenerate the config docs.
Reword the async table function correlate comment so it no longer reads as
if
that path were already metered; it supplies the UDF name that instrumented
call
generators scope their metrics under.
Fail fast when SleepyDoubler is interrupted instead of skipping the sleep,
so an
interruption surfaces as itself rather than as a confusing timing assertion
failure.
---
.../generated/execution_config_configuration.html | 12 +
.../table/api/config/ExecutionConfigOptions.java | 23 ++
.../planner/codegen/CodeGeneratorContext.scala | 32 ++
.../codegen/calls/BridgingFunctionGenUtil.scala | 102 ++++++-
.../codegen/calls/BridgingSqlFunctionCallGen.scala | 6 +-
.../runtime/stream/sql/UdfMetricsITCase.java | 334 +++++++++++++++++++++
6 files changed, 492 insertions(+), 17 deletions(-)
diff --git
a/docs/layouts/shortcodes/generated/execution_config_configuration.html
b/docs/layouts/shortcodes/generated/execution_config_configuration.html
index e71bf3d7673..f79ac2f0005 100644
--- a/docs/layouts/shortcodes/generated/execution_config_configuration.html
+++ b/docs/layouts/shortcodes/generated/execution_config_configuration.html
@@ -362,6 +362,18 @@ By default no operator is disabled.</td>
<td>Duration</td>
<td>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.</td>
</tr>
+ <tr>
+ <td><h5>table.exec.udf-metric-enabled</h5><br> <span class="label
label-primary">Streaming</span></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>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.</td>
+ </tr>
+ <tr>
+ <td><h5>table.exec.udf-metric.sample-interval</h5><br> <span
class="label label-primary">Streaming</span></td>
+ <td style="word-wrap: break-word;">100</td>
+ <td>Integer</td>
+ <td>When UDF metrics are enabled, udfProcessingTime is measured
once every N invocations (default 100). Must be at least 1; a value of 1
measures every invocation. The non-sampled fast path is a single integer
increment.</td>
+ </tr>
<tr>
<td><h5>table.exec.uid.format</h5><br> <span class="label
label-primary">Streaming</span></td>
<td style="word-wrap:
break-word;">"<id>_<transformation>"</td>
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 a1ed3ec97ff..60ccd18801f 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,29 @@ 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<Boolean> 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<Integer>
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 once every N "
+ + "invocations (default 100). Must be at
least 1; a value of 1 measures every "
+ + "invocation. The non-sampled fast path
is a single integer increment.");
+
// ------------------------------------------------------------------------
// Other Exec Options
// ------------------------------------------------------------------------
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 fade7279606..2edf0dbac83 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
`<operator>.udf.<udfName>`
+ * @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 f557cb0e570..96b71affcfe 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,
+ // Async table function correlate entry; supplies the UDF name that
instrumented call
+ // generators scope their metrics under, mirroring
BridgingSqlFunctionCallGen.
+ 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 83223e63200..b78091f0bb1 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 00000000000..135d8389fd2
--- /dev/null
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/stream/sql/UdfMetricsITCase.java
@@ -0,0 +1,334 @@
+/*
+ * 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.HistogramStatistics;
+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 java.util.concurrent.TimeUnit;
+
+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
+ * <operator>.udf.<udfName>} 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<Row> 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";
+
+ /** The metric identifier group is {@code <operator>.udf.<udfName>}. */
+ 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");
+ }
+ }
+
+ /** 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) {
+ // Without the sleep the timing assertions would fail on an
unrelated symptom, so
+ // surface the interruption instead.
+ Thread.currentThread().interrupt();
+ throw new RuntimeException("Interrupted while sleeping in
SleepyDoubler", e);
+ }
+ return i == null ? null : i * 2;
+ }
+ }
+
+ /** Emits each input twice; the metered sync table path. */
+ public static class DuplicateRows extends TableFunction<Integer> {
+ 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 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);
+ 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<String, Metric> 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.<name> 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<String, Metric> found = reporter.findMetrics(jobId, pattern);
+ assertThat(found).hasSize(1);
+ return found.values().iterator().next();
+ }
+}
