JingsongLi commented on code in PR #10898: URL: https://github.com/apache/paimon/pull/10898#discussion_r4435381554
########## paimon-python/pypaimon/read/fallback_read_builder.py: ########## @@ -0,0 +1,142 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. + +"""Batch read builder honouring the ``scan.fallback-branch`` table option. + +Mirrors Java's ``org.apache.paimon.table.FallbackReadFileStoreTable``: +partitions that exist in the main branch are read from main; partitions +that exist only in the fallback branch are read from the fallback branch. +All branches of a table share the same physical data files (only snapshot, +tag and schema metadata are branch scoped), and the fallback branch must +have the same row type as main, so a split planned on either branch can be +read with a single :class:`~pypaimon.read.table_read.TableRead`. +""" + +from typing import Dict, List, TYPE_CHECKING, Union + +from pypaimon.read.plan import Plan +from pypaimon.read.read_builder import ReadBuilder + + +if TYPE_CHECKING: + from pypaimon.read.split import Split + from pypaimon.read.table_read import TableRead + from pypaimon.read.table_scan import TableScan + from pypaimon.table.file_store_table import FileStoreTable + + +def _partition_key(split: "Split") -> tuple: + """Return a hashable, branch-independent key for a split's partition.""" + partition = split.partition + values = getattr(partition, "values", None) + if values is None: + return (partition,) + return tuple(values) + + +class FallbackReadBuilder(ReadBuilder): + """A :class:`ReadBuilder` that merges main and fallback-branch partitions. + + Configuration (filter, projection, read type, limit) is applied to both + the main and the fallback builders so the fallback scan reads the same + columns and rows as main. + """ + + def __init__(self, table: "FileStoreTable", fallback_branch: str): + super().__init__(table) + self._fallback_table = table.copy({"branch": fallback_branch}) Review Comment: [P1] Load the fallback branch's own authorization options `table.copy({'branch': ...})` changes the identifier but retains the main schema/options. With query auth disabled on main and enabled on branch `b`, the copied fallback scan consequently also has auth disabled. I reproduced a stored `scan.fallback-branch=b` on a real REST-catalog append table: reading `b` directly under a NULL mask returns `{'k': None, 'p': 'b'}`, while reading main through this builder returns the original `{'k': 2, 'p': 'b'}`. Main and `b` had independent, compatible schema histories (main changed `metadata.stats-mode`, `b` enabled auth), so the file could be read without a missing-schema error. Please load the actual branch table/schema/options and authorize that branch before planning; the current validation is normally comparing main's fields with a copy of themselves. ########## paimon-python/pypaimon/read/fallback_read_builder.py: ########## @@ -0,0 +1,142 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. + +"""Batch read builder honouring the ``scan.fallback-branch`` table option. + +Mirrors Java's ``org.apache.paimon.table.FallbackReadFileStoreTable``: +partitions that exist in the main branch are read from main; partitions +that exist only in the fallback branch are read from the fallback branch. +All branches of a table share the same physical data files (only snapshot, +tag and schema metadata are branch scoped), and the fallback branch must +have the same row type as main, so a split planned on either branch can be +read with a single :class:`~pypaimon.read.table_read.TableRead`. +""" + +from typing import Dict, List, TYPE_CHECKING, Union + +from pypaimon.read.plan import Plan +from pypaimon.read.read_builder import ReadBuilder + + +if TYPE_CHECKING: + from pypaimon.read.split import Split + from pypaimon.read.table_read import TableRead + from pypaimon.read.table_scan import TableScan + from pypaimon.table.file_store_table import FileStoreTable + + +def _partition_key(split: "Split") -> tuple: + """Return a hashable, branch-independent key for a split's partition.""" + partition = split.partition + values = getattr(partition, "values", None) + if values is None: + return (partition,) + return tuple(values) + + +class FallbackReadBuilder(ReadBuilder): + """A :class:`ReadBuilder` that merges main and fallback-branch partitions. + + Configuration (filter, projection, read type, limit) is applied to both + the main and the fallback builders so the fallback scan reads the same + columns and rows as main. + """ + + def __init__(self, table: "FileStoreTable", fallback_branch: str): + super().__init__(table) + self._fallback_table = table.copy({"branch": fallback_branch}) + self._validate_same_row_type(table, self._fallback_table, fallback_branch) + self._fallback_builder = ReadBuilder(self._fallback_table) + + @staticmethod + def _validate_same_row_type(main_table, fallback_table, fallback_branch): + main_fields = main_table.table_schema.fields + fallback_fields = fallback_table.table_schema.fields + same = len(main_fields) == len(fallback_fields) and all( + m.name == f.name and str(m.type) == str(f.type) + for m, f in zip(main_fields, fallback_fields) + ) + if not same: + raise ValueError( + "The fallback branch must have the same schema as the main " + "branch. This validation is triggered because " + "'scan.fallback-branch' is configured to '{}'.".format( + fallback_branch)) + + # -- configuration is mirrored onto the fallback builder --------------- + + def with_filter(self, predicate) -> "FallbackReadBuilder": + super().with_filter(predicate) + self._fallback_builder.with_filter(predicate) + return self + + def with_partition_filter(self, partition_filter) -> "FallbackReadBuilder": + super().with_partition_filter(partition_filter) + self._fallback_builder.with_partition_filter(partition_filter) + return self + + def with_projection( + self, projection: Union[List[str], Dict[str, str]] + ) -> "FallbackReadBuilder": + super().with_projection(projection) + self._fallback_builder.with_projection(projection) + return self + + def with_read_type(self, read_type) -> "FallbackReadBuilder": + super().with_read_type(read_type) + self._fallback_builder.with_read_type(read_type) + return self + + def with_limit(self, limit: int) -> "FallbackReadBuilder": + super().with_limit(limit) + self._fallback_builder.with_limit(limit) + return self + + # -- planning merges the two branches ---------------------------------- + + def new_scan(self) -> "TableScan": + return _FallbackTableScan( + super().new_scan(), self._fallback_builder.new_scan()) + + def new_read(self) -> "TableRead": Review 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. ########## paimon-python/pypaimon/read/fallback_read_builder.py: ########## @@ -0,0 +1,142 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. + +"""Batch read builder honouring the ``scan.fallback-branch`` table option. + +Mirrors Java's ``org.apache.paimon.table.FallbackReadFileStoreTable``: +partitions that exist in the main branch are read from main; partitions +that exist only in the fallback branch are read from the fallback branch. +All branches of a table share the same physical data files (only snapshot, +tag and schema metadata are branch scoped), and the fallback branch must +have the same row type as main, so a split planned on either branch can be +read with a single :class:`~pypaimon.read.table_read.TableRead`. +""" + +from typing import Dict, List, TYPE_CHECKING, Union + +from pypaimon.read.plan import Plan +from pypaimon.read.read_builder import ReadBuilder + + +if TYPE_CHECKING: + from pypaimon.read.split import Split + from pypaimon.read.table_read import TableRead + from pypaimon.read.table_scan import TableScan + from pypaimon.table.file_store_table import FileStoreTable + + +def _partition_key(split: "Split") -> tuple: + """Return a hashable, branch-independent key for a split's partition.""" + partition = split.partition + values = getattr(partition, "values", None) + if values is None: + return (partition,) + return tuple(values) + + +class FallbackReadBuilder(ReadBuilder): + """A :class:`ReadBuilder` that merges main and fallback-branch partitions. + + Configuration (filter, projection, read type, limit) is applied to both + the main and the fallback builders so the fallback scan reads the same + columns and rows as main. + """ + + def __init__(self, table: "FileStoreTable", fallback_branch: str): + super().__init__(table) + self._fallback_table = table.copy({"branch": fallback_branch}) + self._validate_same_row_type(table, self._fallback_table, fallback_branch) + self._fallback_builder = ReadBuilder(self._fallback_table) + + @staticmethod + def _validate_same_row_type(main_table, fallback_table, fallback_branch): + main_fields = main_table.table_schema.fields + fallback_fields = fallback_table.table_schema.fields + same = len(main_fields) == len(fallback_fields) and all( + m.name == f.name and str(m.type) == str(f.type) + for m, f in zip(main_fields, fallback_fields) + ) + if not same: + raise ValueError( + "The fallback branch must have the same schema as the main " + "branch. This validation is triggered because " + "'scan.fallback-branch' is configured to '{}'.".format( + fallback_branch)) + + # -- configuration is mirrored onto the fallback builder --------------- + + def with_filter(self, predicate) -> "FallbackReadBuilder": + super().with_filter(predicate) + self._fallback_builder.with_filter(predicate) + return self + + def with_partition_filter(self, partition_filter) -> "FallbackReadBuilder": + super().with_partition_filter(partition_filter) + self._fallback_builder.with_partition_filter(partition_filter) + return self + + def with_projection( + self, projection: Union[List[str], Dict[str, str]] + ) -> "FallbackReadBuilder": + super().with_projection(projection) + self._fallback_builder.with_projection(projection) + return self + + def with_read_type(self, read_type) -> "FallbackReadBuilder": + super().with_read_type(read_type) + self._fallback_builder.with_read_type(read_type) + return self + + def with_limit(self, limit: int) -> "FallbackReadBuilder": + super().with_limit(limit) + self._fallback_builder.with_limit(limit) + return self + + # -- planning merges the two branches ---------------------------------- + + def new_scan(self) -> "TableScan": + return _FallbackTableScan( + super().new_scan(), self._fallback_builder.new_scan()) + + def new_read(self) -> "TableRead": + return super().new_read() + + +class _FallbackTableScan: + """Wraps the main and fallback scans and merges their plans by partition.""" + + def __init__(self, main_scan: "TableScan", fallback_scan: "TableScan"): + self._main_scan = main_scan + self._fallback_scan = fallback_scan + + 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} Review 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. ########## paimon-python/pypaimon/read/fallback_read_builder.py: ########## @@ -0,0 +1,142 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. + +"""Batch read builder honouring the ``scan.fallback-branch`` table option. + +Mirrors Java's ``org.apache.paimon.table.FallbackReadFileStoreTable``: +partitions that exist in the main branch are read from main; partitions +that exist only in the fallback branch are read from the fallback branch. +All branches of a table share the same physical data files (only snapshot, +tag and schema metadata are branch scoped), and the fallback branch must +have the same row type as main, so a split planned on either branch can be +read with a single :class:`~pypaimon.read.table_read.TableRead`. +""" + +from typing import Dict, List, TYPE_CHECKING, Union + +from pypaimon.read.plan import Plan +from pypaimon.read.read_builder import ReadBuilder + + +if TYPE_CHECKING: + from pypaimon.read.split import Split + from pypaimon.read.table_read import TableRead + from pypaimon.read.table_scan import TableScan + from pypaimon.table.file_store_table import FileStoreTable + + +def _partition_key(split: "Split") -> tuple: + """Return a hashable, branch-independent key for a split's partition.""" + partition = split.partition + values = getattr(partition, "values", None) + if values is None: + return (partition,) + return tuple(values) + + +class FallbackReadBuilder(ReadBuilder): + """A :class:`ReadBuilder` that merges main and fallback-branch partitions. + + Configuration (filter, projection, read type, limit) is applied to both + the main and the fallback builders so the fallback scan reads the same + columns and rows as main. + """ + + def __init__(self, table: "FileStoreTable", fallback_branch: str): + super().__init__(table) + self._fallback_table = table.copy({"branch": fallback_branch}) + self._validate_same_row_type(table, self._fallback_table, fallback_branch) + self._fallback_builder = ReadBuilder(self._fallback_table) + + @staticmethod + def _validate_same_row_type(main_table, fallback_table, fallback_branch): + main_fields = main_table.table_schema.fields + fallback_fields = fallback_table.table_schema.fields + same = len(main_fields) == len(fallback_fields) and all( + m.name == f.name and str(m.type) == str(f.type) + for m, f in zip(main_fields, fallback_fields) + ) + if not same: + raise ValueError( + "The fallback branch must have the same schema as the main " + "branch. This validation is triggered because " + "'scan.fallback-branch' is configured to '{}'.".format( + fallback_branch)) + + # -- configuration is mirrored onto the fallback builder --------------- + + def with_filter(self, predicate) -> "FallbackReadBuilder": + super().with_filter(predicate) + self._fallback_builder.with_filter(predicate) + return self + + def with_partition_filter(self, partition_filter) -> "FallbackReadBuilder": + super().with_partition_filter(partition_filter) + self._fallback_builder.with_partition_filter(partition_filter) + return self + + def with_projection( + self, projection: Union[List[str], Dict[str, str]] + ) -> "FallbackReadBuilder": + super().with_projection(projection) + self._fallback_builder.with_projection(projection) + return self + + def with_read_type(self, read_type) -> "FallbackReadBuilder": + super().with_read_type(read_type) + self._fallback_builder.with_read_type(read_type) + return self + + def with_limit(self, limit: int) -> "FallbackReadBuilder": + super().with_limit(limit) + self._fallback_builder.with_limit(limit) + return self + + # -- planning merges the two branches ---------------------------------- + + def new_scan(self) -> "TableScan": + return _FallbackTableScan( + super().new_scan(), self._fallback_builder.new_scan()) + + def new_read(self) -> "TableRead": + return super().new_read() + + +class _FallbackTableScan: + """Wraps the main and fallback scans and merges their plans by partition.""" + + def __init__(self, main_scan: "TableScan", fallback_scan: "TableScan"): + self._main_scan = main_scan + self._fallback_scan = fallback_scan + + 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} + + fallback_splits = [ + split + for split in self._fallback_scan.plan().splits() + if _partition_key(split) not in main_partitions + ] + + return Plan(_splits=main_splits + fallback_splits, + snapshot_id=main_plan.snapshot_id) + + def __getattr__(self, name): + # Forward any other scan API (shard/slice/stats/...) to the main scan. + return getattr(self._main_scan, name) Review 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). -- 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]
