Skip to content

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

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
weiqingy wants to merge 5 commits into
base: master
Choose a base branch
from
Draft

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

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
12b1961
[FLINK-38071][table] Add table.exec.udf-metric-enabled and table.exec…
weiqingy Jul 8, 2026
68b5e90
[FLINK-38071][table-runtime] Add UdfMetrics helper for UDF metrics
weiqingy Jul 8, 2026
e8977cb
[FLINK-38071][table-planner] Instrument sync scalar and table UDF cal…
weiqingy Jul 8, 2026
f0d286a
[FLINK-38071][table-planner] Instrument async scalar and table UDF ca…
weiqingy Jul 8, 2026
154a637
[FLINK-38071][docs] Document UDF metrics
weiqingy Jul 8, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
44 changes: 44 additions & 0 deletions docs/content.zh/docs/ops/metrics.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down Expand Up @@ -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" >}}).
Expand Down
44 changes: 44 additions & 0 deletions docs/content/docs/ops/metrics.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down Expand Up @@ -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.

@davidradl davidradl Jul 15, 2026 •
edited
Loading

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

is this advice the same for all metrics - or is this metric particularly performance adverse.

Do we have an idea of what the performance overhead is when switching this on? I would expect metrics overhead would/should be minimal

@weiqingy weiqingy Jul 16, 2026 •
edited
Loading

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks for taking a look, David. Good questions on both.

On the wording: the "…may impact performance / recommended only for debugging" note mirrors the existing State Access Latency and state-size metric sections a bit higher on this same page. It's the standard caution for opt-in, per-record operator metrics, not a signal that UDF metrics are unusually costly.

On the actual overhead — it should be minimal, and that's the design intent, via two layers:

  • Off by default: when table.exec.udf-metric-enabled is false, nothing is registered and there is zero overhead.
  • When enabled, it uses the same counter-based sampling as state latency tracking (FLINK-21736): only every Nth invocation is timed (default sample-interval = 100) and the non-sampled fast path is a single integer increment. Exceptions are counted every invocation, but that's just a counter increment.

So on the common path the added per-record cost is essentially a branch plus an int increment, with the nanoTime()/histogram work amortized across the sampling interval. In local micro-benchmarking the non-sampled path was sub-nanosecond; a rigorous JMH benchmark belongs in flink-benchmarks and I can add one.

Happy to make the sampling context clearer in this note (e.g. that overhead is minimal at the default interval) when I make the PR review-ready.

One heads-up: this PR is intentionally still a draft. I'm keeping it in draft until the FLIP-485 vote concludes, then I'll mark it review-ready. Appreciate the early look.


## REST API integration

Metrics can be queried through the [Monitoring REST API]({{< ref "docs/ops/rest_api" >}}).
Expand Down
12 changes: 12 additions & 0 deletions docs/layouts/shortcodes/generated/execution_config_configuration.html
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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>
Expand Down
22 changes: 22 additions & 0 deletions ...able-api-java/src/main/java/org/apache/flink/table/api/config/ExecutionConfigOptions.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
// ------------------------------------------------------------------------
Expand Down
9 changes: 8 additions & 1 deletion ...able-planner/src/main/java/org/apache/flink/table/planner/codegen/AsyncCodeGenerator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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);
Expand All @@ -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
Expand Down
47 changes: 47 additions & 0 deletions ...-planner/src/main/scala/org/apache/flink/table/planner/codegen/CodeGeneratorContext.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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] =
Expand Down Expand Up @@ -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.
*
Expand Down
Loading