Repository navigation
[python] Read scan.fallback-branch partitions in batch reads - #10898
henriquejsza wants to merge 1 commit into
Conversation
Batch reads in pypaimon only rejected scan.fallback-branch as a catalog option and never read it from the table, so a read left out the partitions that exist only in the fallback branch. Spark returns those rows, as the issue reports. FileStoreTable.new_read_builder now returns a FallbackReadBuilder when the table option scan.fallback-branch is set and the current branch is not the fallback branch itself. Mirroring Java's FallbackReadFileStoreTable, partitions present in the main branch are read from main and partitions that exist only in the fallback branch are read from the fallback branch. Filter, projection, read type and limit are applied to both branch scans, and the fallback branch must have the same row type as main. All branches share the same physical data files, so the merged splits are read with a single TableRead. Closes apache#10884 Signed-off-by: henriquejsza <henriquejsza@gmail.com>
There was a problem hiding this comment.
Requirement fit: SUPPORTED. Implementation: FINDINGS — three confirmed issues.
The missing fallback-only partitions are a real end-to-end problem, but the confirmed failures below block production use: partition priority under value pruning, branch-specific schema history, and the fluent scan/shard API. The three new tests and 36 existing branch/schema-evolution tests pass locally, as does configured Flake8. Three additional actual-table fixtures fail with successful direct-read controls.
Correction: I withdrew the earlier authorization claim because its manual policy-response hook bypassed the repository's supported branch/policy validation. That is not counted as a finding. No production code was modified during review.
|
|
||
| def __init__(self, table: "FileStoreTable", fallback_branch: str): | ||
| super().__init__(table) | ||
| self._fallback_table = table.copy({"branch": fallback_branch}) |
There was a problem hiding this comment.
Withdrawn finding: unsupported branch-policy fixture
I am withdrawing this P1 authorization claim. The manual AuthTableQueryResponse hook in the reproduction bypassed the repository's policy-management validation: installing data policies/column grants on tables with branches is explicitly rejected, and branch creation rejects existing restricted policies. Consequently that fixture does not establish a supported production authorization regression. The separately reported branch-schema-history failure below is reproduced with normal filesystem catalog operations and remains valid.
| def plan(self) -> Plan: | ||
| main_plan = self._main_scan.plan() | ||
| main_splits = list(main_plan.splits()) | ||
| main_partitions = {_partition_key(split) for split in main_splits} |
There was a problem hiding this comment.
[P1] Determine partition ownership before data-file pruning
These are partitions remaining after the value filter/limit/shard has affected the main scan, not all partitions owned by main. With metadata.stats-mode=full, main has (k=1,p='a') and fallback has (k=99,p='a'); the unfiltered read correctly hides 99. Adding k=99 prunes the main file, empties this set, and the filtered read incorrectly returns 99 from the shadowed fallback partition instead of no rows. This is reproducible with actual writes/scans. Inventory main partitions independently of value filtering and split truncation, using only the applicable partition selection, then exclude those partitions from fallback; Java's newPartitionListingScan keeps these decisions separate.
| return _FallbackTableScan( | ||
| super().new_scan(), self._fallback_builder.new_scan()) | ||
|
|
||
| def new_read(self) -> "TableRead": |
There was a problem hiding this comment.
[P2] Keep branch schema history when reading fallback splits
Sharing physical data paths does not make schema IDs branch independent. I created branch b, added x, changed a branch option, wrote a file at branch schema 2, and dropped x; main and b again both expose exactly (k,p). After storing the fallback option on main (schema 1), a direct read of b succeeds, but the merged read fails in SplitRead._resolve_schema with AttributeError: 'NoneType' object has no attribute 'fields': this main-only TableRead looks for the fallback file's schema 2 in main's history. Preserve branch provenance in the splits and dispatch to the corresponding branch reader/schema manager, as the Java fallback reader does. Merely comparing current row types cannot validate historical schemas.
|
|
||
| def __getattr__(self, name): | ||
| # Forward any other scan API (shard/slice/stats/...) to the main scan. | ||
| return getattr(self._main_scan, name) |
There was a problem hiding this comment.
[P2] Preserve the fallback scan in fluent scan methods
The forwarded bound method returns the underlying main TableScan. Consequently even the identity operation builder.new_scan().with_shard(0, 1).plan() loses the wrapper: in the actual two-partition fixture it returns only main's a row and drops fallback-only b. Keeping the wrapper in a variable does not solve this either, because then sharding is applied only to main and fallback remains unsharded. Other fluent methods such as row-range/global-index selection have the same delegation issue. Implement the supported mutators explicitly, return this wrapper, and apply selection/distribution consistently to the branch or combined plan (or reject unsupported combinations).
Purpose
Closes #10884.
pypaimon only checked
scan.fallback-branchin the catalog options, whereit is rejected, and never read it from the table. A batch read therefore
left out the partitions that exist only in the fallback branch, while
Spark's
SELECT ... FROM paimon.db.ton the same table returns them.FileStoreTable.new_read_buildernow returns aFallbackReadBuilderwhenthe table option
scan.fallback-branchis set and the current branch is notthe fallback branch itself. Mirroring Java's
FallbackReadFileStoreTable,partitions that exist in the main branch are read from main, and partitions
that exist only in the fallback branch are read from the fallback branch.
Filter, projection, read type and limit are applied to both branch scans,
and the fallback branch must have the same row type as main (same check Java
performs). All branches of a table share the same physical data files (only
snapshot, tag and schema metadata are branch scoped), so the merged splits
are read with a single
TableRead.Tests
pypaimon/tests/batch_read_fallback_branch_test.py:test_batch_read_includes_fallback_only_partitionsreproduces the issue(a partition missing from main is read from the fallback branch).
test_main_partition_shadows_fallbackasserts a partition present in mainis read from main, not the fallback.
test_no_fallback_option_reads_main_onlyasserts the read stays on mainwhen the option is unset.
flake8 --config=./dev/cfg.inipasses on the changed files, and the licenseheader check passes.