kevinjqliu commented on code in PR #3816:
URL: https://github.com/apache/iceberg-python/pull/3816#discussion_r3826205613
##########
pyiceberg/table/__init__.py:
##########
@@ -926,19 +932,17 @@ def upsert(
# get list of rows that exist so we don't have to load the entire
target table
matched_predicate = upsert_util.create_match_filter(df, join_cols)
- # We must use Transaction.table_metadata for the scan. This includes
all uncommitted - but relevant - changes.
+ matched_iceberg_file_scan = self._scan(row_filter=matched_predicate,
case_sensitive=case_sensitive, branch=branch)
- matched_iceberg_record_batches_scan = DataScan(
+ # The target branch determines which files to read; the transaction
schema determines how to project their rows.
+ # These can differ because schema updates do not create snapshots.
+ matched_iceberg_record_batches = ArrowScan(
table_metadata=self.table_metadata,
io=self._table.io,
+ projected_schema=self.table_metadata.schema(),
row_filter=matched_predicate,
case_sensitive=case_sensitive,
- )
-
- if branch in self.table_metadata.refs:
- matched_iceberg_record_batches_scan =
matched_iceberg_record_batches_scan.use_ref(branch)
Review Comment:
this is now part of the `_scan` logic, by passing `branch` in directly
##########
pyiceberg/table/__init__.py:
##########
@@ -314,11 +314,19 @@ def _apply(
return self
- def _scan(self, row_filter: str | BooleanExpression = ALWAYS_TRUE,
case_sensitive: bool = True) -> DataScan:
- """Minimal data scan of the table with the current state of the
transaction."""
- return DataScan(
+ def _scan(
+ self,
+ row_filter: str | BooleanExpression = ALWAYS_TRUE,
+ case_sensitive: bool = True,
+ branch: str | None = None,
+ ) -> DataScan:
Review Comment:
reviewer note:
the change here is only adding
```
branch: str | None = None,
```
##########
pyiceberg/table/__init__.py:
##########
@@ -778,9 +786,7 @@ def delete(
bound_delete_filter = bind(self.table_metadata.schema(),
delete_filter, case_sensitive)
preserve_row_filter =
_expression_to_complementary_pyarrow(bound_delete_filter,
self.table_metadata.schema())
- file_scan = self._scan(row_filter=delete_filter,
case_sensitive=case_sensitive)
- if branch is not None:
- file_scan = file_scan.use_ref(branch)
+ file_scan = self._scan(row_filter=delete_filter,
case_sensitive=case_sensitive, branch=branch)
Review Comment:
simplifying the logic now that we can pass `branch` into `_scan`
##########
pyiceberg/table/__init__.py:
##########
@@ -926,20 +932,16 @@ def upsert(
# get list of rows that exist so we don't have to load the entire
target table
matched_predicate = upsert_util.create_match_filter(df, join_cols)
- # We must use Transaction.table_metadata for the scan. This includes
all uncommitted - but relevant - changes.
+ matched_iceberg_file_scan = self._scan(row_filter=matched_predicate,
case_sensitive=case_sensitive, branch=branch)
- matched_iceberg_record_batches_scan = DataScan(
- table_metadata=self.table_metadata,
- io=self._table.io,
- row_filter=matched_predicate,
- case_sensitive=case_sensitive,
+ # Plan files from the target branch, but read them using the
transaction's current schema.
+ # A schema update does not create a snapshot, so the branch snapshot
may use an older schema.
+ matched_iceberg_record_batches =
_to_arrow_batch_reader_via_file_scan_tasks(
+ matched_iceberg_file_scan,
+ self.table_metadata.schema(),
Review Comment:
this is the main fix to ensure that we're using the same schema!
--
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
