JingsongLi commented on code in PR #10570:
URL: https://github.com/apache/paimon/pull/10570#discussion_r4282273559
##########
paimon-python/pypaimon/write/file_store_commit.py:
##########
@@ -86,6 +86,80 @@ def _reject_compact_increment(messages: List[CommitMessage]):
'Committing a compact increment requires a separate COMPACT
snapshot.')
+def _delete_abort_paths(table, paths):
+ for path_to_delete in paths:
+ try:
+ table.file_io.delete_quietly(str(path_to_delete))
+ except Exception as error:
+ logger.warning(
+ "Failed to clean up file %s during abort: %s",
+ path_to_delete,
+ error,
+ )
+
+
+def _protected_managed_packs(table, paths):
+ """Keep a pack when a blobref outside this abort still names it.
+
+ One pack can be listed on every data file that referenced it. Deleting it
+ with the first aborted file would make the other files unreadable.
+ """
+ from pypaimon.blob.managed_blob_reference_file import
ManagedBlobReferenceFile
+
+ pack_suffix = ManagedBlobReferenceFile.MANAGED_BLOB_SUFFIX
+ ref_suffix = ManagedBlobReferenceFile.REFERENCE_FILE_SUFFIX
+ deleting = {str(path) for path in paths}
+ packs_by_dir = {}
+ for path in deleting:
+ if not path.endswith(pack_suffix):
+ continue
+ parent, _, name = path.rpartition("/")
+ if parent and name:
+ packs_by_dir.setdefault(parent, {})[name] = path
+ if not packs_by_dir:
+ return set()
+
+ list_status = getattr(table.file_io, "list_status", None)
+ if list_status is None:
+ return set()
+
+ protected = set()
+ for parent, packs in packs_by_dir.items():
+ try:
+ statuses = list(list_status(parent))
+ except TypeError:
+ continue
+ except Exception as error:
Review Comment:
[P1] Preserve shared packs when reference discovery fails
A failed directory listing is treated as if no other file references these
packs. `abort()` then deletes them. I reproduced this using the new
partial-abort scenario: three data files share a pack, commit the first two,
and abort the third while `list_status` raises one `OSError`. The abort deletes
the shared pack and both previously committed rows become unreadable. The
read-error branch below correctly protects every candidate pack, but the
listing-error branch does not. Please preserve the directory's candidate packs
(or fail without deleting them) whenever reference discovery cannot complete; a
transient filesystem/RPC failure must not delete committed payloads.
##########
paimon-python/pypaimon/blob/primary_key_blob_externalizer.py:
##########
@@ -0,0 +1,258 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+"""Externalize primary-key BLOB values into managed packs before the data
file."""
+
+import pyarrow as pa
+
+from pypaimon.blob.managed_blob_reference_collector import _map_items
+from pypaimon.schema.data_types import (
+ AtomicType,
+ DataField,
+ is_array_blob_type,
+ is_blob_type,
+ is_map_blob_type,
+)
+from pypaimon.table.row.blob import BlobDescriptor
+from pypaimon.table.row.generic_row import GenericRow
+from pypaimon.table.row.row_kind import RowKind
+from pypaimon.write.blob_format_writer import BlobFormatWriter
+
+
+class PrimaryKeyBlobExternalizer:
+ """Write managed BLOB payloads to ``.managed.blob`` packs and store
descriptors.
+
+ Retract rows drop their payload, matching ``PrimaryKeyBlobExternalizer`` in
+ Java. Packs stay owned by this writer until ``prepare_commit``.
+ """
+
+ _RETRACT_KINDS = frozenset({
+ RowKind.UPDATE_BEFORE.value,
+ RowKind.DELETE.value,
+ })
+
+ def __init__(
+ self,
+ file_io,
+ new_pack_path,
+ value_fields,
+ managed_blob_fields,
+ target_file_size,
+ copy_buffer_size,
+ uri_reader_factory=None,
+ table_path=None):
+ if target_file_size <= 0:
+ raise ValueError(
+ "Managed BLOB target file size must be positive, but was %s."
+ % target_file_size)
+ self._file_io = file_io
+ self._uri_reader_factory = uri_reader_factory
+ self._table_path = None if table_path is None else str(table_path)
+ self._uncommitted = []
+ self._referenced_pack_names = []
+ self._writers = {}
+ self._field_specs = []
+ for field in value_fields:
+ if field.name not in managed_blob_fields:
+ continue
+ if is_blob_type(field.type):
+ kind = "scalar"
+ elif is_array_blob_type(field.type):
+ kind = "array"
+ elif is_map_blob_type(field.type):
+ kind = "map"
+ else:
+ continue
+ self._field_specs.append((field.name, kind))
+ self._writers[field.name] = _ManagedBlobPackWriter(
+ file_io, new_pack_path, target_file_size, copy_buffer_size,
+ self._uncommitted)
+
+ def externalize_table(self, data: pa.Table) -> pa.Table:
+ self._referenced_pack_names = []
+ if not self._field_specs:
+ return data
+ try:
+ if "_VALUE_KIND" in data.schema.names:
+ kinds = data.column("_VALUE_KIND").to_pylist()
+ else:
+ kinds = [RowKind.INSERT.value] * data.num_rows
+ for field_name, field_kind in self._field_specs:
+ if field_name not in data.schema.names:
+ continue
+ # TODO: to_pylist() materializes the whole BLOB column.
+ values = data.column(field_name).to_pylist()
+ replaced = []
+ for index, value in enumerate(values):
+ if kinds[index] in self._RETRACT_KINDS:
+ replaced.append(None)
+ else:
+ replaced.append(
+ self._externalize_value(value, field_kind,
field_name))
+ field = data.schema.field(field_name)
+ column_index = data.schema.get_field_index(field_name)
+ data = data.set_column(
+ column_index, field, pa.array(replaced, type=field.type))
+ return data
+ except Exception:
Review Comment:
[P1] Keep completed data files and their packs consistent after a flush
failure
`self.abort()` deletes all packs accumulated since the previous prepare,
including packs referenced by data files already flushed into
`KeyValueDataWriter.committed_files`. Those file metadata remain because
`_flush_all()` is outside `prepare_commit()`'s abort-on-error block. With a
stream writer (`target-file-size=1 KB`, `blob.target-file-size=1 GB`), write a
2 KB payload to flush file A, buffer a second small row, and inject one
`OSError` while externalizing the latter during `prepare_commit(1)`. A's pack
is deleted but its data file and metadata survive. Retrying `prepare_commit(1)`
and committing succeeds, publishing a snapshot whose first row fails with
missing-pack. Please coordinate externalizer failure with the data writer, or
roll back only the failed operation, so retry cannot publish metadata for
deleted packs.
##########
paimon-python/pypaimon/read/table_read.py:
##########
@@ -886,6 +887,7 @@ def _arrow_batch_generator(self, splits: List[Split],
schema: pyarrow.Schema,
batch = batch.slice(0, remaining)
batch = self._serialize_blob_views_as_descriptors(
batch, reader)
+ batch = self._materialize_managed_blob_batch(batch)
Review Comment:
[P2] Resolve managed payloads before evaluating payload predicates
This conversion runs after the split reader has already evaluated the
predicate against the stored descriptor bytes. A real Python-only write/commit
of `(id=1, payload=b"visible")` and `(id=2, payload=b"other")` reads both
payloads correctly without a filter, but
`new_predicate_builder().equal("payload", b"visible")` returns no rows. I
reproduced it with write/read/scan native execution disabled. The native
managed-BLOB tests explicitly require payload predicates to work, including
when payload is projected out. Please resolve managed BLOB predicate
dependencies before filtering and avoid pushing payload predicates into
descriptor-valued file columns; the current path silently returns incorrect
query results.
##########
paimon-python/pypaimon/blob/primary_key_blob_externalizer.py:
##########
@@ -0,0 +1,258 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+"""Externalize primary-key BLOB values into managed packs before the data
file."""
+
+import pyarrow as pa
+
+from pypaimon.blob.managed_blob_reference_collector import _map_items
+from pypaimon.schema.data_types import (
+ AtomicType,
+ DataField,
+ is_array_blob_type,
+ is_blob_type,
+ is_map_blob_type,
+)
+from pypaimon.table.row.blob import BlobDescriptor
+from pypaimon.table.row.generic_row import GenericRow
+from pypaimon.table.row.row_kind import RowKind
+from pypaimon.write.blob_format_writer import BlobFormatWriter
+
+
+class PrimaryKeyBlobExternalizer:
+ """Write managed BLOB payloads to ``.managed.blob`` packs and store
descriptors.
+
+ Retract rows drop their payload, matching ``PrimaryKeyBlobExternalizer`` in
+ Java. Packs stay owned by this writer until ``prepare_commit``.
+ """
+
+ _RETRACT_KINDS = frozenset({
+ RowKind.UPDATE_BEFORE.value,
+ RowKind.DELETE.value,
+ })
+
+ def __init__(
+ self,
+ file_io,
+ new_pack_path,
+ value_fields,
+ managed_blob_fields,
+ target_file_size,
+ copy_buffer_size,
+ uri_reader_factory=None,
+ table_path=None):
+ if target_file_size <= 0:
+ raise ValueError(
+ "Managed BLOB target file size must be positive, but was %s."
+ % target_file_size)
+ self._file_io = file_io
+ self._uri_reader_factory = uri_reader_factory
+ self._table_path = None if table_path is None else str(table_path)
+ self._uncommitted = []
+ self._referenced_pack_names = []
+ self._writers = {}
+ self._field_specs = []
+ for field in value_fields:
+ if field.name not in managed_blob_fields:
+ continue
+ if is_blob_type(field.type):
+ kind = "scalar"
+ elif is_array_blob_type(field.type):
+ kind = "array"
+ elif is_map_blob_type(field.type):
+ kind = "map"
+ else:
+ continue
+ self._field_specs.append((field.name, kind))
+ self._writers[field.name] = _ManagedBlobPackWriter(
+ file_io, new_pack_path, target_file_size, copy_buffer_size,
+ self._uncommitted)
+
+ def externalize_table(self, data: pa.Table) -> pa.Table:
+ self._referenced_pack_names = []
+ if not self._field_specs:
+ return data
+ try:
+ if "_VALUE_KIND" in data.schema.names:
+ kinds = data.column("_VALUE_KIND").to_pylist()
+ else:
+ kinds = [RowKind.INSERT.value] * data.num_rows
+ for field_name, field_kind in self._field_specs:
+ if field_name not in data.schema.names:
+ continue
+ # TODO: to_pylist() materializes the whole BLOB column.
+ values = data.column(field_name).to_pylist()
+ replaced = []
+ for index, value in enumerate(values):
+ if kinds[index] in self._RETRACT_KINDS:
+ replaced.append(None)
+ else:
+ replaced.append(
+ self._externalize_value(value, field_kind,
field_name))
+ field = data.schema.field(field_name)
+ column_index = data.schema.get_field_index(field_name)
+ data = data.set_column(
+ column_index, field, pa.array(replaced, type=field.type))
+ return data
+ except Exception:
+ self.abort()
+ raise
+
+ def finish_current(self):
+ for writer in self._writers.values():
+ writer.close_current()
+
+ def prepare_commit(self):
+ try:
+ self.finish_current()
+ # Clear in place. Each pack writer holds this same list.
+ self._uncommitted.clear()
+ except Exception:
+ self.abort()
+ raise
+
+ def take_referenced_pack_names(self):
+ """Pack file names referenced by the data file just externalized."""
+ names = self._referenced_pack_names
+ self._referenced_pack_names = []
+ return names
+
+ def abort(self):
+ for writer in self._writers.values():
+ writer.abort_current()
+ for path in list(self._uncommitted):
+ self._file_io.delete_quietly(path)
+ self._uncommitted.clear()
+
+ def _externalize_value(self, value, field_kind, field_name):
+ if hasattr(value, "as_py"):
+ value = value.as_py()
+ if value is None:
+ return None
+ if field_kind == "scalar":
+ return self._write_descriptor(field_name, value)
+ if field_kind == "array":
+ if isinstance(value, (bytes, bytearray, str)) or not
hasattr(value, "__iter__"):
+ raise ValueError(
+ "ARRAY<BLOB> field '%s' must be a list of blob values." %
field_name)
+ return [
+ None if element is None else
self._write_descriptor(field_name, element)
+ for element in value
+ ]
+ if field_kind == "map":
+ externalized = []
+ for key, map_value in _map_items(value):
+ if map_value is None:
+ externalized.append((key, None))
+ else:
+ externalized.append(
+ (key, self._write_descriptor(field_name, map_value)))
+ if isinstance(value, dict):
+ return dict(externalized)
+ return externalized
+ return value
+
+ def _write_descriptor(self, field_name, raw):
+ blob = BlobFormatWriter._to_blob(raw, self._uri_reader_factory)
+ descriptor = self._writers[field_name].write(blob)
+ uri = self._table_relative_uri(descriptor.uri)
+ name = uri.replace('\\', '/').rsplit('/', 1)[-1]
+ if name not in self._referenced_pack_names:
+ self._referenced_pack_names.append(name)
+ return BlobDescriptor(uri, descriptor.offset,
descriptor.length).serialize()
+
+ def _table_relative_uri(self, uri):
+ """Pack location relative to the table directory.
+
+ Append dedicated-blob descriptors still store the writer path.
+ """
+ root = self._table_path
+ if not root:
+ return uri
+ normalized = uri.replace('\\', '/')
+ prefix = root.replace('\\', '/').rstrip('/') + '/'
+ if normalized.startswith(prefix):
Review Comment:
[P2] Resolve table-relative descriptors in the row iterator too
New packs are stored with a table-relative URI, but only the Arrow output
paths call the new resolver. After a normal committed primary-key BLOB write,
`new_read().to_iterator(splits)` followed by `row.get_blob(1).to_data()` tries
to open `bucket-0/<pack>.managed.blob` relative to the process working
directory and raises `FileNotFoundError`; `to_arrow(splits)` on the same table
succeeds. The row iterator's Blob/URI reader needs the table base as well.
Please cover the existing row-iterator API, including its lazy Blob access,
before relying on these relative descriptors.
--
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]
