XiaoHongbo-Hope commented on code in PR #9840:
URL: https://github.com/apache/paimon/pull/9840#discussion_r4023043864
##########
paimon-python/pypaimon/write/table_update_by_row_id.py:
##########
@@ -432,8 +437,52 @@ def _read_original_file_data(self, first_row_id: int,
column_names: List[str]) -
predicate=None,
read_type=read_fields + [SpecialFields.ROW_ID],
)
- original = table_read.to_arrow([origin_split])
- return original.select([field.name for field in read_fields])
+ return table_read, origin_split
+
+ def _merged_batches(self, first_row_id, data, column_names):
+ """Merge ordinary columns a batch at a time in physical row order."""
+ table_read, split = self._original_file_read(first_row_id,
column_names)
+ updates =
sorted(enumerate(data[SpecialFields.ROW_ID.name].to_pylist()),
+ key=lambda item: item[1])
+ update_index = 0
+ offset = first_row_id
+ with table_read._to_managed_arrow_batch_reader([split]) as reader:
+ for batch in reader:
+ end = offset + batch.num_rows
+ selected = []
+ while update_index < len(updates) and updates[update_index][1]
< end:
+ selected.append(updates[update_index][0])
+ update_index += 1
+ original = pa.Table.from_batches([batch]).select(column_names)
+ if selected:
+ merged, _ = self._merge_update_with_original(
+ original, data.take(selected), column_names, offset)
+ else:
+ merged = original
+ yield from merged.to_batches()
+ offset = end
+ del batch, original, merged
+ if update_index != len(updates):
+ raise ValueError('Update row IDs extend past the original file
group')
+
+ def _write_group_streaming(self, partition, first_row_id, data,
column_names):
+ writer = AppendOnlyDataWriter(
+ self.table, tuple(partition.values), 0, 0,
+ self.table.options, write_cols=column_names)
+ batches = self._merged_batches(first_row_id, data, column_names)
+ try:
+ files = writer._write_batches(batches)
Review Comment:
> `_write_batches` is a private name carrying a public contract: it names
the file, owns the output stream, aggregates stats, appends to
`committed_files` and cleans up on failure. In other words it re-implements a
good part of `DataWriter._write_data_to_file`, yet it is reached only from here
and never from `DataWriter.write` / `prepare_commit`.
>
> The risk is drift: whoever later adds a step to `_write_data_to_file` (a
new sidecar, an extra metadata field, a validation) will silently miss this
second file-producing path. Extracting `_create_data_file_meta` is a good
start; consider either dropping the underscore to signal this is a real entry
point, or moving it into a small dedicated writer so both paths share the
finalization.
>
> Minor, same area: this is the first call site that uses
`_to_managed_arrow_batch_reader` as a context manager (`multimodal/temporal.py`
and `multimodal/query.py` do not). `_ClosableArrowBatchReader` implements
`__enter__` / `__exit__` so it works, and `RecordBatchReader.from_stream` only
exists on new enough pyarrow, but it is worth saying so in that method's
docstring.
I prefer to move it into a small dedicated writer.
--
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]
