[FLINK-41448][table-planner] Name UDF metric groups after the normalized function name

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

@@ -2384,7 +2384,7 @@ 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 registered only on the operator that runs the function, under the infix `udf.<udf_name>`, so the identifier is the operator scope (`metrics.scope.operator`) followed by `udf.<udf_name>.<metric>`.
+The following metrics are registered per SQL/Table user-defined function when UDF metrics are enabled (see [UDF metrics](#udf-metrics) below). They are registered only on the operator that runs the function, under the infix `udf.<udf_name>`, so the identifier is the operator scope (`metrics.scope.operator`) followed by `udf.<udf_name>.<metric>`. For a function called by name, `<udf_name>` is its name in lower case.
 
 <table class="table table-bordered">
   <thead>
@@ -2477,6 +2477,8 @@ Flink can track the processing time and exceptions of SQL/Table user-defined fun
 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.
+Function names are case-insensitive, so for a function called by name, `<udf_name>` is its name in lower case: calls written as `MyUdf(x)` and `MYUDF(x)` both report under `udf.myudf`.
+A function called inline in the Table API, such as `call(new MyUdf(), $("x"))`, is not registered under a name, so it reports under a name generated from its class, such as `*com.example.MyUdf*`.
 
 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.

docs/content/docs/ops/metrics.md

@@ -2384,7 +2384,7 @@ 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 registered only on the operator that runs the function, under the infix `udf.<udf_name>`, so the identifier is the operator scope (`metrics.scope.operator`) followed by `udf.<udf_name>.<metric>`.
+The following metrics are registered per SQL/Table user-defined function when UDF metrics are enabled (see [UDF metrics](#udf-metrics) below). They are registered only on the operator that runs the function, under the infix `udf.<udf_name>`, so the identifier is the operator scope (`metrics.scope.operator`) followed by `udf.<udf_name>.<metric>`. For a function called by name, `<udf_name>` is its name in lower case.
 
 <table class="table table-bordered">
   <thead>
@@ -2477,6 +2477,8 @@ Flink can track the processing time and exceptions of SQL/Table user-defined fun
 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.
+Function names are case-insensitive, so for a function called by name, `<udf_name>` is its name in lower case: calls written as `MyUdf(x)` and `MYUDF(x)` both report under `udf.myudf`.
+A function called inline in the Table API, such as `call(new MyUdf(), $("x"))`, is not registered under a name, so it reports under a name generated from its class, such as `*com.example.MyUdf*`.
 
 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.

flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/BridgingFunctionGenUtil.scala

@@ -173,7 +173,7 @@ object BridgingFunctionGenUtil {
       skipIfArgsNull,
       // Async table function correlate entry; supplies the UDF name that instrumented call
       // generators scope their metrics under, mirroring BridgingSqlFunctionCallGen.
-      udfMetricName = Some(function.getName)
+      udfMetricName = Some(udfMetricNameOf(function))
     )
   }
 
@@ -582,6 +582,16 @@ object BridgingFunctionGenUtil {
     }
   }
 
+  /**
+   * Returns the metric group name of a UDF call. Function names resolve case-insensitively, so a
+   * named function reports under its normalized name and every spelling of it shares one group. An
+   * inline function is not resolved by name and keeps its generated, class-based name.
+   */
+  private[calls] def udfMetricNameOf(function: BridgingSqlFunction): String = {
+    val name = function.getName
+    if (function.getResolvedFunction.isAnonymous) name else FunctionIdentifier.normalizeName(name)
+  }
+
   /**
    * 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

flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/BridgingSqlFunctionCallGen.scala

@@ -19,7 +19,7 @@ package org.apache.flink.table.planner.codegen.calls
 
 import org.apache.flink.table.functions.{ScalarFunction, TableFunction, UserDefinedFunctionHelper}
 import org.apache.flink.table.planner.codegen._
-import org.apache.flink.table.planner.codegen.calls.BridgingFunctionGenUtil.{generateFunctionAwareCall, DefaultExpressionEvaluatorFactory}
+import org.apache.flink.table.planner.codegen.calls.BridgingFunctionGenUtil.{generateFunctionAwareCall, udfMetricNameOf, DefaultExpressionEvaluatorFactory}
 import org.apache.flink.table.planner.delegation.PlannerBase
 import org.apache.flink.table.planner.functions.bridging.BridgingSqlFunction
 import org.apache.flink.table.planner.functions.inference.OperatorBindingCallContext
@@ -79,7 +79,7 @@ class BridgingSqlFunctionCallGen(call: RexCall, rexProgram: RexProgram) extends
       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)
+      udfMetricName = Some(udfMetricNameOf(function))
     )
   }
 }

flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/stream/sql/UdfMetricsITCase.java

@@ -53,7 +53,10 @@
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
+import java.util.regex.Pattern;
 
+import static org.apache.flink.table.api.Expressions.$;
+import static org.apache.flink.table.api.Expressions.call;
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
@@ -83,17 +86,17 @@ class UdfMetricsITCase {
     // span is measurably non-zero.
     private static final long ASYNC_DELAY_MILLIS = 5;
 
-    // Matches the processing-time metric of any UDF, for asserting that none is registered.
+    // Matches the processing-time metric of any UDF.
     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";
+        return "\\.udf\\." + Pattern.quote(udfName) + "\\.udfProcessingTime";
     }
 
     private static String exceptionCountPattern(String udfName) {
-        return "\\.udf\\." + udfName + "\\.udfExceptionCount";
+        return "\\.udf\\." + Pattern.quote(udfName) + "\\.udfExceptionCount";
     }
 
     /** Doubles an int; the metered sync scalar path. */
@@ -363,6 +366,43 @@ void testRepeatedFunctionSharesOneHandle() throws Exception {
                 .isEqualTo(2L * SOURCE_ROWS.size());
     }
 
+    @Test
+    void testFunctionNameSpellingsShareOneLowerCaseGroup() throws Exception {
+        StreamTableEnvironment tEnv = createTableEnv(true);
+        tEnv.createTemporarySystemFunction("MixedCaseUdf", IntDoubler.class);
+        createSource(tEnv, "src", "id INT");
+        createBlackHoleSink(tEnv, "sink", "a INT, b INT");
+
+        JobID jobId =
+                execute(
+                        tEnv,
+                        "INSERT INTO sink SELECT mixedCaseUdf(id), MIXEDCASEUDF(id) FROM src");
+
+        // Function names are case-insensitive, so both spellings call one function and feed one
+        // group named after the lower-case name: 2 evals per row.
+        assertThat(reporter.findMetrics(jobId, ANY_PROCESSING_TIME_PATTERN)).hasSize(1);
+        assertThat(histogram(jobId, processingTimePattern("mixedcaseudf")).getCount())
+                .isEqualTo(2L * SOURCE_ROWS.size());
+    }
+
+    @Test
+    void testInlineFunctionKeepsClassBasedGroupName() throws Exception {
+        StreamTableEnvironment tEnv = createTableEnv(true);
+        createSource(tEnv, "src", "id INT");
+        createBlackHoleSink(tEnv, "sink", "v INT");
+
+        TableResult result =
+                tEnv.from("src").select(call(IntDoubler.class, $("id"))).executeInsert("sink");
+        JobID jobId = result.getJobClient().get().getJobID();
+        result.await();
+
+        // An inline function is not called by name, so its generated name keeps the class name's
+        // case instead of being lower-cased.
+        String groupName = "*" + IntDoubler.class.getName() + "*";
+        assertThat(histogram(jobId, processingTimePattern(groupName)).getCount())
+                .isEqualTo(SOURCE_ROWS.size());
+    }
+
     @Test
     void testDistinctFunctionsGetSeparateHandles() throws Exception {
         StreamTableEnvironment tEnv = createTableEnv(true);
@@ -525,6 +565,25 @@ void testAsyncTableMetricsRecorded() throws Exception {
                 .isEqualTo(SOURCE_ROWS.size());
     }
 
+    @Test
+    void testAsyncTableGroupNameIsLowerCase() 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)");
+
+        // An async table function call is generated apart from the scalar one, so its group name
+        // is lower-cased on that path as well.
+        assertThat(histogram(jobId, processingTimePattern("asynctableudf")).getCount())
+                .isEqualTo(SOURCE_ROWS.size());
+    }
+
     @Test
     void testAsyncDisabledRegistersNoUdfMetrics() throws Exception {
         StreamTableEnvironment tEnv = createTableEnv(false);