[
https://issues.apache.org/jira/browse/FLINK-38871?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Vishal Kamlapure updated FLINK-38871:
-------------------------------------
Environment:
*Python :-* 3.11.14
*PyFlink :-* 1.19.1 & 2.2.0
*Java :-* 11
*Tested On:-* MacOs (Locally) & Yarn Flink Cluster EMR
was:
*Python :-* 3.11.14
*Flink :-* 1.19.1 & 2.2.0
*Java :-* 11
*Tested On:-* MacOs (Locally) & Yarn Flink Cluster (1.19.1 - EMR)
> PyFlink Planner incorrectly propagates constants into downstream UDF inputs
> when filtering on materialized ROW fields
> ---------------------------------------------------------------------------------------------------------------------
>
> Key: FLINK-38871
> URL: https://issues.apache.org/jira/browse/FLINK-38871
> Project: Flink
> Issue Type: Bug
> Components: API / Python, Table SQL / Planner
> Affects Versions: 1.19.1, 2.2.0
> Environment: *Python :-* 3.11.14
> *PyFlink :-* 1.19.1 & 2.2.0
> *Java :-* 11
> *Tested On:-* MacOs (Locally) & Yarn Flink Cluster EMR
> Reporter: Vishal Kamlapure
> Priority: Major
> Attachments: Screenshot 2026-01-08 at 2.50.40 AM.png
>
>
> I have identified a correctness/performance bug in the Flink Table Planner
> regarding the optimization of {{ROW}} fields returned by Python UDFs.
> When a Python UDF returns a {{ROW}} type and a field from that {{ROW}} is
> materialized into a top-level column (using {{{}add_or_replace_columns{}}})
> and subsequently filtered, the planner aggressively applies {*}Constant
> Propagation{*}.
> If this column is used as input for a _second_ downstream Python UDF, the
> planner replaces the actual column reference with the constant literal
> derived from the filter (e.g., rewriting {{amountId}} to {{{}'0'{}}}). This
> causes the downstream UDF to execute on rows that {*}do not satisfy the
> filter predicate{*}, receiving the constant value instead of the actual data.
> This behavior breaks filter semantics and corrupts intermediate data observed
> by UDFs. Notably, this issue *does not occur* when the upstream UDF returns a
> {{MAP}} type, suggesting the issue is specific to {{ROW}} field optimization
> rules.
> h3. Reproduction Script (PyFlink)
> The following script reproduces the issue. It generates a batch with
> {{amountId}} values {{"0"}} and {{{}"1"{}}}. We filter for {{{}"0"{}}}, but
> the downstream UDF prints {{"0"}} for *all* rows, proving that the value
> {{"1"}} was overwritten by the planner.
> {code:python}
> from pyflink.table import EnvironmentSettings, TableEnvironment, DataTypes
> from pyflink.table.udf import udf
> from pyflink.table.expressions import col
> import pandas as pd
> def reproducer():
> env_settings = EnvironmentSettings.in_batch_mode()
> t_env = TableEnvironment.create(env_settings)
> # 1. Source Data: Row 'B' has amountId="1" (Should be filtered out)
> t = t_env.from_elements(
> [
> ("A", "0"),
> ("B", "1"),
> ("C", "0"),
> ],
> DataTypes.ROW([
> DataTypes.FIELD("transactionText", DataTypes.STRING()),
> DataTypes.FIELD("amountId", DataTypes.STRING())
> ])
> )
> # 2. First UDF: Returns a ROW
> @udf(result_type=DataTypes.ROW([
> DataTypes.FIELD("out_text", DataTypes.STRING()),
> DataTypes.FIELD("out_amountId", DataTypes.STRING())
> ]), func_type="pandas")
> def validate_udf(text: pd.Series, amountId: pd.Series) -> pd.DataFrame:
> return pd.DataFrame({
> "out_text": text,
> "out_amountId": amountId
> })
> # 3. Second UDF: Inspects the value it receives
> @udf(result_type=DataTypes.ROW([
> DataTypes.FIELD("seen_amountId", DataTypes.STRING())
> ]), func_type="pandas")
> def second_udf(amountId: pd.Series) -> pd.DataFrame:
> # DEBUG: Print exactly what the UDF sees
> print("\n[second_udf] received batch:", amountId.tolist())
> return pd.DataFrame({
> "seen_amountId": amountId
> })
> # 4. Pipeline Construction
> validated = t.add_columns(validate_udf(t.transactionText,
> t.amountId).alias("v"))
> # Materialize ROW fields to top-level columns
> materialized = validated.add_or_replace_columns(
> col("v").out_text.alias("transactionText"),
> col("v").out_amountId.alias("amountId")
> )
> # Filter: We only want amountId == "0"
> filtered = materialized.filter(col("amountId") == "0")
> # Apply downstream UDF on the filtered data
> final = filtered.add_columns(
> second_udf(filtered.amountId).alias("s")
> ).select(
> col("transactionText"),
> col("amountId"),
> col("s").seen_amountId
> )
> print("=== PLAN ===")
> print(final.explain())
> print("=== EXECUTION ===")
> final.execute().print()
> if __name__ == '__main__':
> reproducer()
> {code}
> h3. Observed Behavior
> The logs from {{second_udf}} show that it receives the value {{'0'}} for the
> second row, even though that row actually contains {{{}'1'{}}}.
>
> [second_udf] received batch: ['0', '0', '0']
> _(Note: The batch size is 3, meaning the row with ID '1' was not dropped, but
> its value was overwritten to '0'.)_
> The execution plan reveals that the planner created a {{Calc}} node that
> forcefully casts the column to a constant before calling the Python UDF:
> Plaintext
>
> Calc(select=[..., CAST('0' AS VARCHAR) AS amountId, ...])
> h3. Expected Behavior
> # The downstream UDF should only process rows that satisfy the filter (or if
> execution pipelining allows processing, it must see the *original* values).
> # The planner should not rewrite input columns to constants based on
> downstream filters if those columns are inputs to Python UDFs.
> h3. Analysis & Workaround
> The issue appears to be an unsafe application of constant folding on {{ROW}}
> fields materialized from {{PythonCalc}} outputs.
> * *ROW Type (Buggy):* The planner views the {{ROW}} fields as transparent
> and applies constant propagation ({{{}amountId = '0'{}}}) _before_ the filter
> is physically enforced, and passes this constant to the next
> {{{}PythonCalc{}}}.
> * *MAP Type (Workaround):* Changing {{validate_udf}} to return {{MAP<STRING,
> STRING>}} fixes the issue. The planner treats the map access {{ITEM(map,
> 'key')}} as opaque, preventing constant propagation and forcing the correct
> execution order (UDF -> Filter -> UDF).
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
