Copilot commented on code in PR #28879: URL: https://github.com/apache/flink/pull/28879#discussion_r3844493972
########## 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: SleepyDoubler swallows InterruptedException and continues, which can make the timing assertions flaky (the sleep may be skipped) while also losing the failure signal. In tests, it’s better to fail fast when the thread is interrupted. ########## 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: The comment says this correlate entry “opts it into metrics”, but the async-table UDF path currently calls generateAsyncTableFunctionCall(...) which does not instrument eval. This makes the comment misleading for ASYNC_TABLE UDFs. ########## 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: Thread.sleep(...) is not guaranteed to sleep for at least the requested duration on all platforms; asserting mean/max >= SLEEP_MILLIS can be timing-flaky under scheduler jitter. Using a smaller, non-zero threshold still validates that the histogram reflects a real delay without requiring an exact minimum sleep time. ########## 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: TABLE_EXEC_UDF_METRIC_SAMPLE_INTERVAL is enforced to be >= 1 at runtime (UdfMetrics.register throws otherwise), but the option description doesn’t mention the valid range or that 1 means “measure every invocation”. This can lead to confusing failures when users set 0 or a negative value. ########## 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: processingTimePattern builds a regex by concatenating the UDF name without quoting. If a test UDF name ever contains regex metacharacters (e.g., '.' or '$'), reporter.findMetrics(...) can match the wrong metrics or fail to match at all. This issue also appears on line 81 of the same file. -- 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]
