JingsongLi commented on code in PR #9375:
URL: https://github.com/apache/paimon/pull/9375#discussion_r3841867238
##########
paimon-python/pypaimon/write/writer/data_writer.py:
##########
@@ -122,9 +123,8 @@ def write(self, data: pa.RecordBatch):
raise e
def prepare_commit(self) -> List[DataFileMeta]:
- if self.pending_data is not None and self.pending_data.num_rows > 0:
- self._write_data_to_file(self.pending_data)
- self.pending_data = None
+ if self._buffer.num_rows > 0:
+ self._write_data_to_file(self._buffer.take())
Review Comment:
[P2] Preserve buffered rows until the prepare flush succeeds
`take()` resets the buffer before `_write_data_to_file` performs fallible
file I/O. For a reusable `StreamTableWrite`, a transient storage error followed
by retrying `prepare_commit` on the same writer silently omits these rows: with
a fail-once writer, this head leaves 0 rows buffered and the retry writes
nothing, while the base version retains all 3 rows and writes them on retry.
Please materialize without draining, perform the write, and reset only after it
succeeds (or explicitly poison/abort the writer and reject reuse).
##########
paimon-python/pypaimon/write/writer/data_vector_writer.py:
##########
@@ -188,27 +188,29 @@ def _split_data(self, data: pa.RecordBatch) ->
Tuple[pa.RecordBatch, pa.RecordBa
return normal_data, vector_data
def _should_roll_normal(self) -> bool:
- if self.pending_normal_data is None:
+ # Runs on every write, so it answers from the running counts only.
+ if self._normal_buffer.is_empty:
return False
- if self.pending_normal_data.num_rows >= self.target_file_row_num:
+ if self._normal_buffer.num_rows >= self.target_file_row_num:
return True
if self.record_count % self.CHECK_ROLLING_RECORD_CNT != 0:
return False
- return self.pending_normal_data.nbytes > self.target_file_size
+ return self._normal_buffer.nbytes > self.target_file_size
def _current_row_count(self) -> int:
- if self.pending_normal_data is not None:
- return self.pending_normal_data.num_rows
- if self.vector_writer is not None and self.vector_writer.pending_data
is not None:
- return self.vector_writer.pending_data.num_rows
+ if not self._normal_buffer.is_empty:
+ return self._normal_buffer.num_rows
+ if self.vector_writer is not None:
+ # Running count, not a folded buffer: this runs on every write.
+ return self.vector_writer.pending_row_count
return 0
def _close_current_writers(self):
- has_normal = self.pending_normal_data is not None and
self.pending_normal_data.num_rows > 0
+ normal_data = self._normal_buffer.take()
Review Comment:
[P2] Keep the normal and sidecar flush failure-atomic
`_normal_buffer.take()` discards the normal rows before the normal file is
written. If that write fails, the vector buffer remains pending; retrying
`prepare_commit` then has no `normal_meta`, flushes the vector files, and skips
the row-count consistency check, so it can return sidecar-only metadata and
lose the normal half. Please retain the normal table until the coordinated
flush succeeds, or abort/poison the composite writer on any phase failure. The
same ordering issue is present in
`DedicatedFormatWriter._close_current_writers` at its `_normal_buffer.take()`
call.
--
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]
