[FLINK-38071][table] Add opt-in UDF metrics (FLIP-485)

docs/content.zh/docs/ops/metrics.md

@@ -2382,6 +2382,34 @@ logged by `SystemResourcesMetricsInitializer` during the startup.
   </tbody>
 </table>
 
+### 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.<udf_name>`, so the full metric identifier is `<operator_name>.udf.<udf_name>.<metric>`.
+
+<table class="table table-bordered">
+  <thead>
+    <tr>
+      <th class="text-left" style="width: 18%">Scope</th>
+      <th class="text-left" style="width: 30%">Metrics</th>
+      <th class="text-left" style="width: 44%">Description</th>
+      <th class="text-left" style="width: 8%">Type</th>
+    </tr>
+  </thead>
+  <tbody>
+    <tr>
+      <th rowspan="2"><strong>Task/Operator</strong></th>
+      <td>udf.&lt;udf_name&gt;.udfProcessingTime</td>
+      <td>The 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 <samp>table.exec.udf-metric.sample-interval</samp> is measured.</td>
+      <td>Histogram</td>
+    </tr>
+    <tr>
+      <td>udf.&lt;udf_name&gt;.udfExceptionCount</td>
+      <td>The number of exceptions escaping the user-defined function. For synchronous functions this counts exceptions propagating out of <samp>eval</samp>; for asynchronous functions it counts exceptional completions. Counted on every invocation, not sampled.</td>
+      <td>Counter</td>
+    </tr>
+  </tbody>
+</table>
+
 ## 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.<udf_name>`: 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.
+
+<span class="label label-danger">Warning</span> 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" >}}).

docs/content/docs/ops/metrics.md

@@ -2382,6 +2382,34 @@ Metrics below can be used to measure the effectiveness of speculative execution.
   </tbody>
 </table>
 
+### 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.<udf_name>`, so the full metric identifier is `<operator_name>.udf.<udf_name>.<metric>`.
+
+<table class="table table-bordered">
+  <thead>
+    <tr>
+      <th class="text-left" style="width: 18%">Scope</th>
+      <th class="text-left" style="width: 30%">Metrics</th>
+      <th class="text-left" style="width: 44%">Description</th>
+      <th class="text-left" style="width: 8%">Type</th>
+    </tr>
+  </thead>
+  <tbody>
+    <tr>
+      <th rowspan="2"><strong>Task/Operator</strong></th>
+      <td>udf.&lt;udf_name&gt;.udfProcessingTime</td>
+      <td>The 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 <samp>table.exec.udf-metric.sample-interval</samp> is measured.</td>
+      <td>Histogram</td>
+    </tr>
+    <tr>
+      <td>udf.&lt;udf_name&gt;.udfExceptionCount</td>
+      <td>The number of exceptions escaping the user-defined function. For synchronous functions this counts exceptions propagating out of <samp>eval</samp>; for asynchronous functions it counts exceptional completions. Counted on every invocation, not sampled.</td>
+      <td>Counter</td>
+    </tr>
+  </tbody>
+</table>
+
 ## 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.<udf_name>`: 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.
+
+<span class="label label-danger">Warning</span> 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" >}}).

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/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<String, String> 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

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
@@ -161,6 +162,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 +905,47 @@ 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
+      }
+    )
+  }
+
+  /**
+   * 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.
    *