[flink] Fix LakeSource copy isolation

[Shawn-Hx]

Purpose

Linked issue: close #4056

Brief change log

Add an explicit LakeSource copy contract and state-preserving copies for Paimon, Iceberg, Hudi, and testing sources.

Make FlinkTableSource copies isolate mutable lake pushdown state while preserving projection, filter, aggregate, limit, and watermark settings. Add regression coverage for UNION scans with different projections and filters.

Tests

UT and IT are added.

API and Format

No.

Documentation

No.

[Copilot]

Pull request overview

This PR fixes a Flink connector correctness issue where DynamicTableSource.copy() could accidentally share a mutable LakeSource instance across planner alternatives (e.g., UNION ALL branches), causing pushdown state (projection/filters) to leak between scans. It introduces an explicit LakeSource.copy() contract and updates Flink/lake implementations and tests to ensure copy isolation while preserving previously applied pushdowns.

Changes:

• Add LakeSource.copy() SPI contract and implement state-preserving, independent copies for Paimon, Iceberg, Hudi, and test sources.
• Update FlinkTableSource.copy() to perform a state-preserving copy, including using LakeSource.copy() and defensive copying of mutable fields.
• Add regression coverage for UNION queries with differing projections/filters and for “empty filters” not triggering catalog access.

Reviewed changes

Copilot reviewed 14 out of 14 changed files in this pull request and generated no comments.

Show a summary per file

File	Description
fluss-common/src/main/java/org/apache/fluss/lake/source/LakeSource.java	Adds the copy() contract to ensure independent lake source instances with preserved pushdown state.
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/FlinkTableSource.java	Reworks copy() into a copy-constructor that deep-copies mutable state and uses lakeSource.copy() to prevent aliasing.
fluss-common/src/main/java/org/apache/fluss/utils/ArrayUtils.java	Adds deepCopy(int[][]) to defensively copy projection arrays.
fluss-common/src/test/java/org/apache/fluss/utils/ArrayUtilsTest.java	Adds unit coverage for ArrayUtils.deepCopy.
fluss-common/src/test/java/org/apache/fluss/lake/source/TestingLakeSource.java	Implements copy() for the test lake source to match the new SPI contract.
fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/source/PaimonLakeSource.java	Implements copy() and deep-copies projection/filter pushdown inputs; handles empty filter pushdown without catalog access.
fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/source/IcebergLakeSource.java	Implements copy() and deep-copies projection/filter pushdown inputs; handles empty filter pushdown without catalog access.
fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/source/HudiLakeSource.java	Implements copy(), deep-copies projection state and predicates list; exposes predicates for testing.
fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/source/PaimonRecordReaderTest.java	Verifies copy isolation and deep-copy behavior for projection state.
fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/source/PaimonLakeSourceTest.java	Adds tests for “empty filters” behavior and for filter isolation across copy().
fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/flink/FlinkUnionReadPrimaryKeyTableITCase.java	Adds IT regressions for UNION with differing pushdowns under both source-reuse enabled/disabled scenarios.
fluss-lake/fluss-lake-iceberg/src/test/java/org/apache/fluss/lake/iceberg/source/IcebergRecordReaderTest.java	Verifies copy isolation and deep-copy behavior for projection state.
fluss-lake/fluss-lake-iceberg/src/test/java/org/apache/fluss/lake/iceberg/source/IcebergLakeSourceTest.java	Adds tests for “empty filters” behavior and for filter isolation across copy().
fluss-lake/fluss-lake-hudi/src/test/java/org/apache/fluss/lake/hudi/source/HudiLakeSourceTest.java	Adds tests validating copied projections and filters remain independent from the original source.

---

💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.