Fokko commented on code in PR #3816:
URL: https://github.com/apache/iceberg-python/pull/3816#discussion_r3907489595
##########
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:
Nice, this simplifies it quite a bit
--
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]
