[FLINK-40293][table] Add UDF metric config options and instrument sync scalar and table UDF calls

docs/layouts/shortcodes/generated/execution_config_configuration.html

@@ -362,6 +362,18 @@
             <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 every N invocations (default 100). 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;">"&lt;id&gt;_&lt;transformation&gt;"</td>

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<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 every N "
+                                    + "invocations (default 100). The non-sampled fast path is a single integer increment.");
+
     // ------------------------------------------------------------------------
     //  Other Exec Options
     // ------------------------------------------------------------------------

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.
    *

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,16 +257,24 @@ 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,
         externalOperands,
         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)

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