[flink] Lake reads may fail when UNION branches use different pushdowns

[Shawn-Hx]

Search before asking

• I searched in the issues and found nothing similar.

Fluss version

main (development)

Please describe the bug 🐞

Description

FlinkTableSource.copy() creates a new table source but reuses the original mutable LakeSource instance:

source.lakeSource = lakeSource;

Flink's projection and filter pushdown rules copy a DynamicTableSource before applying abilities. Multiple planner alternatives or branches can be derived from the same canonical table scan. Each FlinkTableSource copy retains its own producedDataType and projected fields, while the current Paimon, Iceberg, and Hudi LakeSource implementations mutate their projection and filter fields in place.

Because the copies alias one LakeSource, the last pushdown can overwrite the state used by another scan. The lake reader then returns columns using one branch's projection while another branch deserializes them with its own schema. If the shifted physical types differ, the job fails; if they are compatible, it may silently read the wrong column.

Reproduction conditions

The issue is observable when all of the following are true:

1. The Fluss table has data lake enabled and the query actually reads tiered lake data, for example a stateless FULL startup.
2. Flink derives multiple copies of the same table source, such as two references to one table in UNION ALL.
3. The copies receive different lake pushdown state, most commonly different projections or converted filters.
4. The final planner path does not replace those scans with a fresh shared source using a safe superset projection.

For Flink 1.20, two deterministic cases are:

• table.optimizer.reuse-source-enabled=false with different projections.
• table.optimizer.reuse-source-enabled=true with different branch filters, which prevents the scans from being grouped for source reuse.

The default table.optimizer.reuse-sub-plan-enabled value is sufficient and does not need to be changed.

The problem may be masked when source reuse is enabled and both scans have otherwise identical ability specs: Flink 1.20 can rebuild a fresh source with the union of both projections. It is also masked when no rows have reached the lake snapshot, only one effective scan exists, or every copy ends with identical pushdown state. Compatible shifted column types may turn the exception into incorrect results instead of making the problem disappear.

Minimal reproduction

Configure Paimon lake storage, create a non-partitioned lake-enabled primary-key table, and write two rows:

CREATE TABLE t (
  id INT NOT NULL,
  n BIGINT,
  s STRING,
  PRIMARY KEY (id) NOT ENFORCED
) WITH (
  'table.datalake.enabled' = 'true',
  'table.datalake.freshness' = '500ms'
);

-- Write (1, 4, 'string') and (2, 40, 'another_string').
-- Wait until the Paimon snapshot has caught up with the Fluss log end offset.

Start a fresh streaming query without restored state:

SET 'table.optimizer.reuse-source-enabled' = 'false';

SELECT s
FROM t /*+ OPTIONS('scan.startup.mode' = 'full') */
UNION ALL
SELECT CAST(n AS STRING)
FROM t /*+ OPTIONS('scan.startup.mode' = 'full') */;

Expected rows:

string
another_string
4
40

Actual result on the current implementation: the query can emit only part of the result and then fail while deserializing a lake record:

Caused by: java.lang.ClassCastException:
class org.apache.paimon.data.columnar.heap.HeapLongVector
cannot be cast to class org.apache.paimon.data.columnar.BytesColumnVector
    at org.apache.paimon.data.columnar.VectorizedColumnBatch.getString(...)
    at org.apache.fluss.lake.paimon.utils.PaimonRowAsFlussRow.getString(...)
    at org.apache.fluss.flink.utils.FlussRowToFlinkRowConverter.toFlinkRowData(...)

A second reproducible form keeps source reuse enabled but gives the branches different filters:

SET 'table.optimizer.reuse-source-enabled' = 'true';

SELECT s
FROM t /*+ OPTIONS('scan.startup.mode' = 'full') */
WHERE s = 'string'
UNION ALL
SELECT CAST(n AS STRING)
FROM t /*+ OPTIONS('scan.startup.mode' = 'full') */
WHERE n = 40;

This was reproduced on Flink 1.20.3 against Fluss main with a Paimon streaming FULL read. The shallow assignment is in fluss-flink-common, so all Flink connector variants are potentially affected when their planner reaches this copy path. Paimon is integration-tested here; Iceberg and Hudi have the same mutable LakeSource state pattern and are structurally affected.

Expected behavior

Every FlinkTableSource copy should preserve the abilities already applied to the original while isolating all state that later pushdown calls can mutate. One scan's projection or filter must not change another scan's lake reader.

Actual behavior

Planner copies share one mutable LakeSource. Applying a projection or filter to one copy changes the lake reader state used by the other copies, causing deserialization failures, missing results, or potentially incorrect column values.

Solution

Introduce an explicit LakeSource.copy() contract and use it from FlinkTableSource.copy().

• The copy must retain all already-pushed projection/filter state and return an independent instance.
• Copying must be an in-memory operation without catalog access, schema resolution, or ability replay.
• Paimon, Iceberg, and Hudi should implement explicit copy constructors and defensively deep-copy the nested projection arrays.
• FlinkTableSource should use a private copy constructor and preserve all ability state, including projection, filters, aggregate/count state, limit, and watermark configuration.
• Add Paimon FULL regression tests covering source reuse disabled with different projections, and source reuse enabled with different filters.

This changes the public-evolving LakeSource SPI, so third-party implementations need to implement the copy contract and be rebuilt together with the updated connector.

Are you willing to submit a PR?

• I'm willing to submit a PR!