JingsongLi commented on code in PR #9840:
URL: https://github.com/apache/paimon/pull/9840#discussion_r4022250414
##########
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')
Review Comment:
This only catches update row ids past the *end* of the group. A row id below
`first_row_id` is picked up by the `< end` test in the very first batch and
then surfaces from `_merge_update_with_original` as `IndexError: Update
position -N is outside column range [0, M)`, which does not point at the real
cause.
`_calculate_first_row_id`'s `valid_row_id_ranges` check normally rejects
such input, so this is about diagnosability rather than correctness. Making the
selection two-sided would cover both directions:
```python
while update_index < len(updates) and updates[update_index][1] < end:
if updates[update_index][1] < offset:
raise ValueError('Update row IDs precede the original file group')
```
##########
paimon-python/pypaimon/write/writer/append_only_data_writer.py:
##########
@@ -16,15 +16,142 @@
# under the License.
import pyarrow as pa
+import pyarrow.parquet as pq
+import uuid
+from pypaimon.manifest.schema.simple_stats import SimpleStats
+from pypaimon.schema.data_types import PyarrowFieldParser
+from pypaimon.table.row.generic_row import GenericRow
from pypaimon.write.writer.data_writer import DataWriter
+from pypaimon.write.writer.write_buffer import WriteBuffer
class AppendOnlyDataWriter(DataWriter):
"""Data writer for append-only tables."""
+ _ROW_GROUP_MAX_ROWS = 1024 * 1024
+
def _process_data(self, data: pa.RecordBatch) -> pa.Table:
return pa.Table.from_batches([data])
def _merge_data(self, existing_data: pa.Table, new_data: pa.Table) ->
pa.Table:
return pa.concat_tables([existing_data, new_data])
+
+ @staticmethod
+ def _row_group_slice(batch, offset, count):
+ piece = batch.slice(offset, count)
+ # Arrow 6 nbytes counts full backing buffers even for slices. Compact
+ # the slice there so both accounting and retained buffers stay bounded.
+ if int(pa.__version__.split('.')[0]) < 7:
Review Comment:
`pa.__version__` is re-parsed on every slice, and `_row_group_slice` is
called O(log n) times per row group from the binary search. Worth hoisting to a
module-level constant, e.g. `_ARROW_MAJOR = int(pa.__version__.split('.')[0])`.
##########
paimon-python/pypaimon/write/writer/append_only_data_writer.py:
##########
@@ -16,15 +16,142 @@
# under the License.
import pyarrow as pa
+import pyarrow.parquet as pq
+import uuid
+from pypaimon.manifest.schema.simple_stats import SimpleStats
+from pypaimon.schema.data_types import PyarrowFieldParser
+from pypaimon.table.row.generic_row import GenericRow
from pypaimon.write.writer.data_writer import DataWriter
+from pypaimon.write.writer.write_buffer import WriteBuffer
class AppendOnlyDataWriter(DataWriter):
"""Data writer for append-only tables."""
+ _ROW_GROUP_MAX_ROWS = 1024 * 1024
+
def _process_data(self, data: pa.RecordBatch) -> pa.Table:
return pa.Table.from_batches([data])
def _merge_data(self, existing_data: pa.Table, new_data: pa.Table) ->
pa.Table:
return pa.concat_tables([existing_data, new_data])
+
+ @staticmethod
+ def _row_group_slice(batch, offset, count):
+ piece = batch.slice(offset, count)
+ # Arrow 6 nbytes counts full backing buffers even for slices. Compact
+ # the slice there so both accounting and retained buffers stay bounded.
+ if int(pa.__version__.split('.')[0]) < 7:
+ piece = pa.RecordBatch.from_arrays(
+ [pa.concat_arrays([column]) for column in piece.columns],
schema=piece.schema)
+ return piece
+
+ def _row_groups(self, batches):
+ """Bound row-group buffering independently of reader batch boundaries.
+
+ Arrow bytes are an estimate, not the encoded Parquet block size.
+ One oversized row and the current input batch can exceed the target.
+ """
+ configured = self.options.file_block_size()
+ target_bytes = configured.get_bytes() if configured is not None else
128 * 1024 * 1024
Review Comment:
Two small things here.
- `128 * 1024 * 1024` duplicates the Parquet block-size default that lives
on the Java side, and `file.block-size` is declared `no_default_value()`, so
this magic number is the effective default for Python writers. A named constant
with a short comment would make that coupling explicit.
- The budget is measured in Arrow in-memory bytes, not encoded Parquet
bytes, so the resulting row groups end up substantially smaller than
`file.block-size` suggests. The docstring already says the byte count is an
estimate; spelling out that it is an *Arrow-side* estimate would set
expectations better.
Also, the `file.block-size must be positive` check on the next line lives in
a generator body, so it only fires once the first item is pulled. By then
`_write_batches` has already created the output file (it is removed by
`delete_quietly`, so nothing leaks). Config validation like this fits better at
writer construction time.
##########
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:
Review Comment:
`_original_file_read` deliberately keeps `SpecialFields.ROW_ID` in
`read_type` ("Keep _ROW_ID as a row-count anchor"), but the actual values are
then dropped by `select(column_names)` and the row mapping is re-derived
arithmetically from `offset = first_row_id` and `end = offset + batch.num_rows`.
That silently assumes the reader hands back every row of the group,
contiguously, in physical order. It holds today, but any future row-level
filtering on the read path (predicate push-down, an authorization filter,
deletion vectors, a reordering or parallel reader) would make updates land on
the wrong rows with no error at all. Since the column is already being read,
one check turns that class of regression into a loud failure:
```python
row_ids = batch[SpecialFields.ROW_ID.name]
if row_ids[0].as_py() != offset:
raise ValueError(
f'Original file group is not contiguous at {offset}')
```
The previous implementation made the same assumption, but it materialized
the whole group in one shot, so the assumption was less load-bearing; streaming
relies on it per batch.
##########
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.
--
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]
