YannByron opened a new pull request, #3559: URL: https://github.com/apache/fluss/pull/3559
<!-- Generated-by: Qoder following the guidelines at https://github.com/apache/fluss/blob/main/AGENTS.md --> ### Purpose Linked issue: relates to #3213 The Fluss Spark connector currently ships four scan builders — `FlussAppendScanBuilder` / `FlussLakeAppendScanBuilder` / `FlussUpsertScanBuilder` / `FlussLakeUpsertScanBuilder` — split along two orthogonal axes (append vs upsert, native vs lake). The lake batches additionally carry a runtime fallback path (`isFallback`) that morphs a "lake" batch into a "non-lake" batch when the readable lake snapshot is absent. Two consequences follow: 1. Each `FlussLake*Batch` violates SRP by doing both lake planning and native fallback planning, and dynamically picks the reader factory at runtime. Reasoning about which physical read path a query will take requires reading both classes end-to-end. 2. `FlussUpsertScan` has no `pushedPredicate` (upsert never accepts data predicates) but `FlussLakeUpsertScan` does — an asymmetric API surface between siblings. 3. Streaming (`toMicroBatchStream`) is shared between the lake and non-lake variants, but batch (`toBatch`) is split — an inconsistent split axis in the class hierarchy. This PR consolidates the four scan builders into two (`FlussAppendScanBuilder`, `FlussUpsertScanBuilder`) and routes lake-union vs log-only reads through a single `SplitPlanner` that is probed at scan build time. The runtime fallback path is removed: presence/absence of the readable lake snapshot is decided deterministically at planner construction, and `plan()` picks a branch accordingly. ### Brief change log **Introduce `SplitPlanner` (`fluss-spark-common/.../read/SplitPlanner.scala`, new)** - `sealed trait SplitPlanner extends AutoCloseable` with marker sub-traits `AppendSplitPlanner` and `UpsertSplitPlanner` (so `FlussAppendBatch` / `FlussUpsertBatch` can constrain the planner type they accept). - `AbstractSplitPlanner` owns the shared Fluss client `Connection` / `Admin` lifecycle and centralizes the lake-snapshot probe: if `TableConfig.isDataLakeEnabled` is false, no probe; otherwise `admin.getReadableLakeSnapshot` and treat `LakeTableSnapshotNotExistException` (unwrapped via `ExceptionUtils.stripExecutionException`) as "no snapshot". - Two concrete planners — `AppendPlanner` and `UpsertPlanner` — probe the readable lake snapshot exactly once at construction, expose `hasLakeSnapshot: Boolean` and `logTailPredicate: Option[FlussPredicate]`, and dispatch inside `plan()` to either `planLakeUnion(snap)` (lake splits unioned with a Fluss log-tail / kv+log-tail) or `planLogOnly()` (pure Fluss read). **Remove lake-specific Batch classes** - `FlussLakeAppendBatch.scala`, `FlussLakeUpsertBatch.scala`, and `FlussLakeBatch.scala` are deleted. `FlussAppendBatch` / `FlussUpsertBatch` now accept an `AppendSplitPlanner` / `UpsertSplitPlanner` respectively and dispatch `createReaderFactory` based on `planner.hasLakeSnapshot`. The lake union path routes through the universal `FlussLakePartitionReaderFactory` (which already dispatches by input-partition type). The Batch is `AutoCloseable` and owns the planner's `close()`. **Remove startup-mode gating for batch reads** - Batch scans no longer consume `SCAN_START_UP_MODE`. Batch semantics are always "the full table". Rationale: 1. Letting a streaming-oriented config gate batch planning creates asymmetric semantics on append vs upsert tables (KV snapshot has no partial-read semantics). 2. `mode=latest` with no writes since planning time produces `start == stop == tail`, an empty offset range that trips the reader-side `Invalid offset range` guard. 3. Time-range batch reads should be expressed via predicate pushdown on the timestamp column, not startup mode. - `AppendPlanner` uses `OffsetsInitializer.full()` instead of `earliest()` so start offsets resolve to concrete numeric values (needed by the `SCAN_MAX_RECORDS_PER_PARTITION` range-split logic). Per the `OffsetsInitializer.full` javadoc the two are semantically equivalent for a log table, but `.full()` RPC-resolves each bucket to a concrete offset whereas `.earliest()` returns the `LogScanner.EARLIEST_OFFSET` (-2) sentinel. **Merge the two `V2Filters` push-down traits** - `FlussSupportsPushDownV2Filters` and `FlussLakeSupportsPushDownV2Filters` are collapsed into a single `FlussSupportsPushDownV2Filters` trait. Pushdown decisions here are schema-level (partition predicate extraction plus ARROW/non-PK data-predicate acceptance); any further pushdown to `LakeSource` happens inside the planner (in the lake-union branch) and is transparent to Spark. Lake and Fluss are expected to accept the same predicate set, so no lake-only pushdown regresses. **`SparkTable` is no longer lake-aware** - `newScanBuilder` constructs a plain `FlussAppendScanBuilder` / `FlussUpsertScanBuilder`; the planner probes the readable lake snapshot internally. Removes the `admin`-backed lake-snapshot lazy val on `SparkTable` and the associated `Catalog.loadTable` caching load-bearing assumption. **Test changes** - `SparkPrimaryKeyTableReadTest`: the "throws when SCAN_START_UP_MODE=latest" case is inverted to assert the batch scan returns the full table regardless of startup mode, cross-referenced with the log-table symmetric case. - `SparkLakeTableReadTestBase`: the helper that collects input partitions from `FlussScan` no longer discriminates by planner type — after consolidation, both native and lake paths flow through the same `FlussAppendBatch` / `FlussUpsertBatch`. **Terminology alignment** - Internal branch naming across `SplitPlanner`, `FlussBatch`, and `FlussScanBuilder` scaladocs uses "lake-union" (snapshot present) and "log-only" (snapshot absent) consistently. The former legacy term "Native" is not used to describe branches. **`logTailPredicate` signature cleanup** - `SplitPlanner.logTailPredicate` becomes a `val` on each planner instead of a method taking `Option[FlussPredicate]`. The planner already holds `pushedPredicate` from its constructor, so the previous plumbing (ScanBuilder → Scan → Batch → planner) that threaded the same value back into the planner is removed. ### Tests - Unit / IT suites executed locally on `Corretto 11` with `-Dspotless.check.skip=true`: - `SparkLogTableReadTest` — 25 / 25 pass. Notably includes: - `Spark Read: split partition by config` (validates the range-split logic that requires `.full()` to return concrete numeric offsets). - `Spark Read: ignores SCAN_START_UP_MODE (symmetric with upsert)` (asserts full-table batch semantics regardless of startup mode). - `SparkPrimaryKeyTableReadTest` — 13 / 13 pass. Includes the inverted `SCAN_START_UP_MODE=latest` case now asserting a full-table batch result. - `SparkLakePaimonLogTableReadTest` — full suite pass on the prior verification pass (partition pushdown / union read / filter pushdown / limit pushdown / falls back when no lake snapshot / non-FULL startup mode skips lake path / partition filter pushdown in fallback). - Iceberg lake IT (`SparkLakeIcebergLogTableReadTest`) locally intermittently fails on `Not all buckets synced to lake within PT2M` — a mini-cluster tiering timing issue that reproduces on `main` and is not a regression of this refactor. ### API and Format No public API change. No storage-format change. The only user-visible behavior change is that batch reads no longer honor `SCAN_START_UP_MODE`; this is intentional (batch = full table). Streaming continues to honor the config as before. ### Documentation No new user-facing feature to document. A follow-up docs change will note the batch-full-table convention. <!-- Generative AI disclosure: This PR was authored with the assistance of Qoder (https://qoder.com). All architectural decisions, terminology choices, and test surgery were reviewed and directed by the human author. --> - [x] Yes (Qoder) -- 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]
