weiqingy commented on code in PR #28879:
URL: https://github.com/apache/flink/pull/28879#discussion_r3849924016
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/BridgingFunctionGenUtil.scala:
##########
@@ -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)
Review Comment:
Good catch, that over-claimed. Reworded: the entry supplies the UDF name
that instrumented call generators scope their metrics under. Instrumenting the
async-table path is FLINK-40294, the next PR in the series.
##########
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.");
Review Comment:
Fixed. The description now says the value must be at least 1, and that 1
measures every invocation, matching the check in `UdfMetrics.register`. Config
docs regenerated.
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/stream/sql/UdfMetricsITCase.java:
##########
@@ -0,0 +1,331 @@
+/*
+ * 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) {
+ Thread.currentThread().interrupt();
+ }
Review Comment:
Fixed, it now rethrows after restoring the interrupt flag.
Small correction: the original didn't swallow it.
`Thread.currentThread().interrupt()` restores the flag, which is the usual
idiom. Your underlying point was right though, skipping the sleep would have
failed on a timing assertion instead of saying it was interrupted.
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/stream/sql/UdfMetricsITCase.java:
##########
@@ -0,0 +1,331 @@
+/*
+ * 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) {
+ Thread.currentThread().interrupt();
+ }
+ 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);
Review Comment:
Would it be OK to keep the current threshold? `Thread.sleep` overshoots
rather than undershoots, since the OS rounds up to the next timer tick, so the
only way to skip the sleep is an interrupt, which now throws.
A lower threshold would also stop the test showing that the histogram
reflects a real delay, which is what it exists to prove.
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/stream/sql/UdfMetricsITCase.java:
##########
@@ -0,0 +1,331 @@
+/*
+ * 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";
+ }
Review Comment:
Every name reaching these helpers is a lowercase literal declared in this
file (`scalarudf`, `sleepyudf`, `tableudf`), so is there a case that needs
escaping today? The name-built patterns feed positive lookups that fail loudly
if they miss, and the "nothing is registered" checks use the fixed `ANY_*`
constants instead.
Happy to add `Pattern.quote` if you'd rather have the guard anyway.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
