You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@paimon.apache.org by "Stephen0421 (via GitHub)" <gi...@apache.org> on 2026/10/14 12:38:06 UTC
[PR] [python] Externalize primary-key BLOBs and keep shared packs reachable. [paimon]
Stephen0421 opened a new pull request, #10570:
URL: https://github.com/apache/paimon/pull/10570
### Purpose
Primary-key Python writes externalize managed BLOB payloads into `.managed.blob` packs and store table-relative descriptors in the data file. Each data file's `extra_files` lists the `.blobref` sidecar and the pack names that file references. A pack may be listed on more than one data file when those files share it.
When `blob-as-descriptor` is false, primary-key reads resolve those descriptors against the table directory and return payload bytes. Append tables still use dedicated `.blob` files and are unchanged.
`commit.abort` deletes the aborted data files and their `.blobref` sidecars. A `.managed.blob` pack is deleted only when no `.blobref` outside that abort still names it, so aborting one file does not destroy a pack that a committed file still uses. Aborting every file that references a pack still removes the pack. Writer abort deletes packs that have not been handed to a data file. Packs already given to a `BlobConsumer` stay in place.
### Tests
- Primary-key write registers `.blobref` and `.managed.blob` on `extra_files`, and `new_read_builder()` returns payload bytes.
- Partial `commit.abort` keeps a pack that committed files still reference, and those rows still read back.
- Full abort deletes packs that no surviving `.blobref` references.
- Append dedicated-blob reads still return payload bytes.
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [python] Externalize primary-key BLOBs and keep shared packs reachable. [paimon]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi commented on code in PR #10570:
URL: https://github.com/apache/paimon/pull/10570#discussion_r4284919389
##########
paimon-python/pypaimon/write/writer/key_value_data_writer.py:
##########
@@ -116,6 +162,33 @@ def close(self):
finally:
self._buffer.reset()
+ def _write_data_to_file(self, data: pa.Table):
+ if self._blob_externalizer is not None:
+ data = self._blob_externalizer.externalize_table(data)
+ try:
+ super()._write_data_to_file(data)
+ except Exception:
+ if self._blob_externalizer is not None:
+ self._blob_externalizer.discard_pending_hand_off()
+ raise
+ if self._blob_externalizer is not None:
+ self._blob_externalizer.hand_off_pending()
+
+ def _additional_extra_files(self, file_path: str, data: pa.Table) -> List[str]:
+ if not self._managed_blob_fields:
+ return []
+ collector = ManagedBlobReferenceCollector(
+ self.file_io, file_path, self.table.fields, self._managed_blob_fields)
+ try:
+ collector.collect_table(data)
+ names = [collector.close()]
+ if self._blob_externalizer is not None:
+ names[0:0] = self._blob_externalizer.take_referenced_pack_names()
Review Comment:
[P1] Keep shared packs out of exclusively owned extra_files
Registering the same pack on several DataFileMeta.extra_files makes it subject to Java cleanup, which still treats every extra file as exclusively owned. On this exact head I wrote three rolled Python PK data files sharing three managed packs, committed the first two, serialized only the third (uncommitted) CommitMessage with v14, and called the public Java table.newCommit(...).abort([third]). FileStoreCommitImpl.abort -> DataFileMeta.collectFiles deleted all three shared packs. Snapshot 1 and its two committed files remained, but a subsequent public Python read failed with missing-pack. The Python-only abort guard cannot protect this path. Please either avoid publishing shared packs in generic extra_files (retain them through .blobref ownership), or make every consuming engine's cleanup honor shared references before introducing this metadata.
##########
paimon-python/pypaimon/read/table_read.py:
##########
@@ -886,6 +889,8 @@ 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)
+ batch = self._filter_resolved_managed_blob_batch(batch)
Review Comment:
[P2] Apply the row limit after resolved-payload filtering
The split reader receives remaining as its limit and the batch is also sliced before this newly added post-filter. With a real committed PK table containing (1, b"visible"), (2, b"other"), native write/read/scan disabled, equal("payload", b"other").with_limit(1) returns zero rows through both to_arrow and to_iterator; the expected row is id=2. The first nonmatching row exhausts the inner limit before its descriptor is resolved and filtered. Please disable the pre-filter limit when managed payload filtering is required, and count only matching rows towards the requested limit in all read paths.
##########
paimon-python/pypaimon/read/table_read.py:
##########
@@ -1631,6 +1640,95 @@ def _build_split_read(self, split: Split, read_type=None,
limit=effective_limit,
)
+ def _managed_blob_specs(self):
+ cached = getattr(self, '_managed_blob_specs_cache', None)
+ if cached is not None:
+ return cached
+ specs = managed_blob_output_specs(
+ self.table, self._adapter_read_type, self._blob_as_descriptor)
+ self._managed_blob_specs_cache = specs
+ return specs
+
+ def _materialize_managed_blob_batch(self, batch: pyarrow.RecordBatch) -> pyarrow.RecordBatch:
+ specs = self._managed_blob_specs()
+ if not specs:
+ return batch
+ file_io = self.table.file_io
+ table_path = self.table.table_path
+ for _index, name, kind in specs:
+ if name not in batch.schema.names:
+ continue
+ values = [
+ materialize_managed_blob(value, kind, file_io, table_path)
+ for value in batch.column(name).to_pylist()
+ ]
+ column_idx = batch.schema.names.index(name)
+ field = batch.schema.field(column_idx)
+ batch = batch.set_column(
+ column_idx, field, pyarrow.array(values, type=field.type))
+ return batch
+
+ def _materialize_managed_blob_values(self, values):
+ specs = self._managed_blob_specs()
+ if not specs:
+ return values
+ converted = list(values)
+ file_io = self.table.file_io
+ table_path = self.table.table_path
+ for index, _name, kind in specs:
+ if index >= len(converted):
+ continue
+ converted[index] = materialize_managed_blob(
+ converted[index], kind, file_io, table_path)
+ return tuple(converted)
+
+ def _managed_blob_post_filter_fields(self):
+ if self.predicate is None:
+ return set()
+ names = {name for _index, name, _kind in self._managed_blob_specs()}
+ if not names:
+ return set()
+ return predicate_field_names(self.predicate) & names
+
+ def _filter_resolved_managed_blob_batch(self, batch):
+ if batch is None or batch.num_rows == 0:
+ return batch
+ if not self._managed_blob_post_filter_fields():
+ return batch
+ expr = self.predicate.to_arrow(batch.schema)
+ if expr is None:
+ return batch
+ import pyarrow.dataset as ds
+ filtered = ds.InMemoryDataset(
+ pyarrow.Table.from_batches([batch])).scanner(filter=expr).to_table()
+ if filtered.num_rows == 0:
+ return batch.slice(0, 0)
+ return filtered.to_batches()[0]
Review Comment:
[P2] Preserve every batch produced by the payload filter
The dataset scanner can split one input batch into several output batches, so selecting [0] silently discards matching rows. I reproduced this via public APIs on a real native-disabled PK BLOB table: write 140,000 rows all carrying b"x", set read.batch-size=200000, then read with equal("payload", b"x"). The unfiltered read returns 140,000 rows, but the filtered read returns only 131,072 (ids 0..131071), losing 8,928 matches. Please return/iterate the entire filtered result or combine its chunks into one RecordBatch rather than taking the first.
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [python] Externalize primary-key BLOBs and keep shared packs reachable. [paimon]
Posted by "Stephen0421 (via GitHub)" <gi...@apache.org>.
Stephen0421 commented on code in PR #10570:
URL: https://github.com/apache/paimon/pull/10570#discussion_r4283309039
##########
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:
Fixed. A pack is handed off only after `_write_data_to_file` returns and the data file is in `committed_files`. A later externalize failure aborts only packs that have not been handed off, so the completed file and its pack stay together. Covered by `test_later_flush_failure_keeps_completed_pack`: `target-file-size=1 B`, `blob.target-file-size=1 GB`, the first file is flushed, and an `OSError` during the next payload externalize in `prepare_commit` leaves that pack in place.
##########
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:
Fixed. Predicates on primary-key managed BLOB columns are removed from `predicate_for_reader`, so they are not compared with descriptor bytes. The descriptors are resolved first, then the original predicate runs, before output projection. `blob-as-descriptor` still filters the stored descriptor. Covered by `test_payload_predicate_matches_resolved_bytes`: `equal('payload', b'visible')` returns `(1, b'visible')`.
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [python] Externalize primary-key BLOBs and keep shared packs reachable. [paimon]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi commented on code in PR #10570:
URL: https://github.com/apache/paimon/pull/10570#discussion_r4295074108
##########
paimon-python/pypaimon/blob/managed_blob_payload.py:
##########
@@ -0,0 +1,127 @@
+# 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.
+
+"""Resolve primary-key managed BLOB descriptors to payload bytes."""
+
+from pypaimon.blob.managed_blob_reference_collector import _map_items
+from pypaimon.schema.data_types import (
+ is_array_blob_type,
+ is_blob_type,
+ is_map_blob_type,
+ managed_blob_field_names,
+)
+from pypaimon.table.row.blob import Blob, BlobDescriptor
+from pypaimon.utils.path import resolve_path
+
+
+def resolve_table_blob_uri(table_path, uri):
+ """Join a table-relative pack URI onto the table directory.
+
+ Absolute paths and URIs that already carry a scheme are left unchanged so
+ descriptors written with a filesystem path still open.
+ """
+ if uri is None or table_path is None:
+ return uri
+ normalized = uri.replace('\\', '/')
+ if (normalized.startswith('/')
+ or normalized.startswith('file:')
+ or '://' in normalized):
+ return uri
+ return resolve_path(str(table_path), normalized)
+
+
+def managed_blob_output_specs(table, fields, blob_as_descriptor):
+ """``(index, name, kind)`` for primary-key managed BLOB columns.
+
+ Append tables store these columns in dedicated ``.blob`` files and the
+ reader has already turned them into payload bytes. Managed packs exist
+ only on primary-key tables.
+ """
+ if blob_as_descriptor or not table.is_primary_key_table:
+ return []
+ inline = (table.options.blob_descriptor_fields()
+ | table.options.blob_view_fields())
+ managed = managed_blob_field_names(table.fields, inline)
+ specs = []
+ for index, field in enumerate(fields):
+ if field.name not in managed:
Review Comment:
[P2] Resolve managed BLOB leaves selected from a MAP
A supported projection such as `with_projection(["mapping['k']"])` adapts the MAP<BLOB> value to a scalar BLOB field named `mapping_k`. That name is absent from the top-level managed set, so this check produces no materialization spec and the read returns serialized `BlobDescriptor` bytes instead of the payload. With a real committed table, full MAP reads returned `b'map-one'`, while both the selected-key projection and its named-alias form returned descriptor bytes; this reproduces for single-file raw reads and overlapping-file merge reads. Please carry the source field/projection identity into spec construction and materialize the adapted scalar leaf, with readback tests for these projections.
##########
paimon-python/pypaimon/blob/primary_key_blob_externalizer.py:
##########
@@ -0,0 +1,299 @@
+# 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._handed_off = set()
+ self._pending_hand_off = []
+ 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, self._handed_off)
+
+ 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()
+ # Successful data files already took ownership of their packs.
+ # A failed write followed by a retry can leave closed, unused packs.
+ self._delete_uncommitted_packs()
+ except Exception:
+ self.abort()
+ raise
+
+ def take_referenced_pack_names(self):
+ """Pack file names referenced by the data file just externalized.
+
+ Ownership stays with this writer until ``hand_off_pending``. A failure
+ while finishing the data file must still be able to delete these packs.
+ """
+ names = self._referenced_pack_names
+ self._referenced_pack_names = []
+ self._pending_hand_off = list(names)
+ return names
+
+ def hand_off_pending(self):
+ """The data file that references these packs is now recorded."""
+ names = self._pending_hand_off
+ self._pending_hand_off = []
+ self._hand_off_pack_names(names)
+
+ def discard_pending_hand_off(self):
+ """The data file was not recorded. Keep the packs deletable."""
+ self._pending_hand_off = []
+
+ def _hand_off_pack_names(self, names):
+ handed = set(names)
+ if not handed:
+ return
+ self._handed_off.update(handed)
+ self._uncommitted[:] = [
+ path for path in self._uncommitted
+ if _pack_basename(path) not in handed
+ ]
+
+ def abort(self):
+ for writer in self._writers.values():
+ writer.abort_current()
+ self._delete_uncommitted_packs()
+
+ def _delete_uncommitted_packs(self):
+ for path in list(self._uncommitted):
+ if _pack_basename(path) in self._handed_off:
+ continue
+ self._file_io.delete_quietly(path)
+ self._uncommitted[:] = [
+ path for path in self._uncommitted
+ if _pack_basename(path) in self._handed_off
+ ]
+
+ 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):
+ return normalized[len(prefix):]
Review Comment:
[P1] Preserve a pack URI identity that Java readers and GC understand
Stripping the table prefix publishes descriptors and blobref roots such as `bucket-0/<pack>.managed.blob`, but Java `BlobRef` opens that path directly and `ManagedBlobOrphanFilesClean.packIdentityForCleanup` resolves a relative reference against the process directory, while candidate packs have absolute table paths. I committed a real Python PK table, confirmed its payload was readable in Python, aged only the pack to 72 hours, then ran the public Java cleaner with a 24-hour cutoff. It deleted the live pack while snapshot 1, its data file and blobref remained; Python readback then failed. Java payload reads already failed before cleanup. Please publish qualified descriptor/reference URIs consistent with the Java contract, or implement table-relative resolution in all Java consumers before emitting this format, and cover Python write -> Java read/GC -> Python read in an integration test.
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [python] Externalize primary-key BLOBs and keep shared packs reachable. [paimon]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [python] Externalize primary-key BLOBs and keep shared packs reachable. [paimon]
Posted by "Stephen0421 (via GitHub)" <gi...@apache.org>.
Stephen0421 commented on code in PR #10570:
URL: https://github.com/apache/paimon/pull/10570#discussion_r4290774724
##########
paimon-python/pypaimon/write/writer/key_value_data_writer.py:
##########
@@ -116,6 +162,33 @@ def close(self):
finally:
self._buffer.reset()
+ def _write_data_to_file(self, data: pa.Table):
+ if self._blob_externalizer is not None:
+ data = self._blob_externalizer.externalize_table(data)
+ try:
+ super()._write_data_to_file(data)
+ except Exception:
+ if self._blob_externalizer is not None:
+ self._blob_externalizer.discard_pending_hand_off()
+ raise
+ if self._blob_externalizer is not None:
+ self._blob_externalizer.hand_off_pending()
+
+ def _additional_extra_files(self, file_path: str, data: pa.Table) -> List[str]:
+ if not self._managed_blob_fields:
+ return []
+ collector = ManagedBlobReferenceCollector(
+ self.file_io, file_path, self.table.fields, self._managed_blob_fields)
+ try:
+ collector.collect_table(data)
+ names = [collector.close()]
+ if self._blob_externalizer is not None:
+ names[0:0] = self._blob_externalizer.take_referenced_pack_names()
Review Comment:
Fixed. Shared managed packs are no longer registered in `extra_files`; only the exclusively owned `.blobref` sidecar is registered. Python abort discovers pack candidates through the sidecars and checks surviving references before deletion.
I also verified the cross-engine path using a Python v14 CommitMessage and the public Java abort API: aborting the uncommitted file preserves the shared packs, and the committed rows remain readable.
##########
paimon-python/pypaimon/read/table_read.py:
##########
@@ -886,6 +889,8 @@ 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)
+ batch = self._filter_resolved_managed_blob_batch(batch)
Review Comment:
Fixed. When managed-payload filtering is required, the limit is no longer pushed into the split reader. Filtering now happens before outer limit accounting, including the parallel read quota.
I reproduced the reported case: `equal("payload", b"other").with_limit(1)` now returns id=2 through both Arrow and iterator APIs. Added coverage for parallel reads and projected-out payload columns as well.
##########
paimon-python/pypaimon/read/table_read.py:
##########
@@ -1631,6 +1640,95 @@ def _build_split_read(self, split: Split, read_type=None,
limit=effective_limit,
)
+ def _managed_blob_specs(self):
+ cached = getattr(self, '_managed_blob_specs_cache', None)
+ if cached is not None:
+ return cached
+ specs = managed_blob_output_specs(
+ self.table, self._adapter_read_type, self._blob_as_descriptor)
+ self._managed_blob_specs_cache = specs
+ return specs
+
+ def _materialize_managed_blob_batch(self, batch: pyarrow.RecordBatch) -> pyarrow.RecordBatch:
+ specs = self._managed_blob_specs()
+ if not specs:
+ return batch
+ file_io = self.table.file_io
+ table_path = self.table.table_path
+ for _index, name, kind in specs:
+ if name not in batch.schema.names:
+ continue
+ values = [
+ materialize_managed_blob(value, kind, file_io, table_path)
+ for value in batch.column(name).to_pylist()
+ ]
+ column_idx = batch.schema.names.index(name)
+ field = batch.schema.field(column_idx)
+ batch = batch.set_column(
+ column_idx, field, pyarrow.array(values, type=field.type))
+ return batch
+
+ def _materialize_managed_blob_values(self, values):
+ specs = self._managed_blob_specs()
+ if not specs:
+ return values
+ converted = list(values)
+ file_io = self.table.file_io
+ table_path = self.table.table_path
+ for index, _name, kind in specs:
+ if index >= len(converted):
+ continue
+ converted[index] = materialize_managed_blob(
+ converted[index], kind, file_io, table_path)
+ return tuple(converted)
+
+ def _managed_blob_post_filter_fields(self):
+ if self.predicate is None:
+ return set()
+ names = {name for _index, name, _kind in self._managed_blob_specs()}
+ if not names:
+ return set()
+ return predicate_field_names(self.predicate) & names
+
+ def _filter_resolved_managed_blob_batch(self, batch):
+ if batch is None or batch.num_rows == 0:
+ return batch
+ if not self._managed_blob_post_filter_fields():
+ return batch
+ expr = self.predicate.to_arrow(batch.schema)
+ if expr is None:
+ return batch
+ import pyarrow.dataset as ds
+ filtered = ds.InMemoryDataset(
+ pyarrow.Table.from_batches([batch])).scanner(filter=expr).to_table()
+ if filtered.num_rows == 0:
+ return batch.slice(0, 0)
+ return filtered.to_batches()[0]
Review Comment:
Fixed by reusing `FilterRecordBatchReader`, which preserves all batches produced by the filter instead of taking only the first. The compatibility fallback also combines chunks correctly when `pyarrow.concat_batches` is unavailable.
The regression test uses the reported 140,000 matching rows with `read.batch-size=200000` and verifies that all 140,000 rows and IDs are returned.
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [python] Externalize primary-key BLOBs and keep shared packs reachable. [paimon]
Posted by "Stephen0421 (via GitHub)" <gi...@apache.org>.
Stephen0421 commented on code in PR #10570:
URL: https://github.com/apache/paimon/pull/10570#discussion_r4321317649
##########
paimon-python/pypaimon/blob/managed_blob_cleanup.py:
##########
@@ -0,0 +1,124 @@
+# 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.
+
+"""Reference-aware cleanup of shared primary-key BLOB packs."""
+
+import logging
+
+from pypaimon.blob.managed_blob_payload import resolve_table_blob_uri
+from pypaimon.blob.managed_blob_reference_file import ManagedBlobReferenceFile
+
+logger = logging.getLogger(__name__)
+
+
+def managed_blob_deletion_paths(table, paths):
+ """Keep shared packs while deleting exclusively owned data files and sidecars."""
+ deleting = list(paths)
+ deleting.extend(_managed_pack_paths(table, deleting))
+ protected = _protected_managed_packs(table, deleting)
+ return [path for path in deleting if str(path) not in protected]
+
+
+def _managed_pack_paths(table, paths):
+ """Collect colocated pack candidates before deleting aborted sidecars."""
+ packs = set()
+ for path in paths:
+ path = str(path)
+ if not path.endswith(ManagedBlobReferenceFile.REFERENCE_FILE_SUFFIX):
+ continue
+ try:
+ references = ManagedBlobReferenceFile.read(table.file_io, path)
+ parent = path.rpartition('/')[0]
+ for reference in references:
+ pack = resolve_table_blob_uri(
+ table.table_path,
+ reference.storage_root_id + '/' + reference.relative_path)
+ if (pack.rpartition('/')[0] == parent
Review Comment:
Fixed in a56428242.
Local path identities are now normalized consistently for both pack colocation and surviving-reference checks. Cleanup also accounts for Java's literal `file:` paths and Python's encoded file URIs, while preserving remote URI handling.
Regression tests cover both `writer.abort()` and `commit.abort()` with relative warehouses, including special characters. Partial abort preserves shared packs, and aborting the remaining references cleans them up. A real Java-generated sidecar probe also confirms that surviving packs remain readable.
##########
paimon-python/pypaimon/read/reader/managed_blob_record_batch_reader.py:
##########
@@ -0,0 +1,68 @@
+# 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.
+
+import pyarrow as pa
+
+from pypaimon.blob.managed_blob_payload import materialize_managed_blob
+from pypaimon.read.reader.iface.record_batch_reader import RecordBatchReader
+
+
+def _replace_batch_column(batch, index, field, array):
+ """Replace one column without ``RecordBatch.set_column``.
+
+ That method arrived in PyArrow 13. Python 3.6 and 3.7 stay on older releases.
+ """
+ columns = list(batch.columns)
+ fields = [batch.schema.field(i) for i in range(batch.num_columns)]
+ columns[index] = array
+ fields[index] = field
+ return pa.RecordBatch.from_arrays(
+ columns, schema=pa.schema(fields, metadata=batch.schema.metadata))
+
+
+class ManagedBlobRecordBatchReader(RecordBatchReader):
+ """Resolve managed payloads before filtering, limiting or projecting rows."""
+
+ def __init__(self, reader, specs, file_io, table_path):
+ self.reader = reader
+ self.specs = specs
+ self.table_path = table_path
+ self._adopt_metadata(reader)
+ self.file_io = file_io
+ # Resolved scalar BLOBs must be exposed as payloads by get_blob().
+ resolved = {index for index, _name, _kind in specs}
+ self.descriptor_field_indices = (
Review Comment:
Fixed in a56428242.
Resolved-payload metadata now propagates through projections and batch/row adapters. `OffsetRow.get_blob()` constructs `BlobData` directly for these fields, so payload bytes are not interpreted again as BlobView or BlobDescriptor references.
Regression tests cover payloads containing both reference headers, raw and merge reads, reordered and MAP-key projections, and Arrow/iterator consistency. Inline descriptor/view materialization preserves the same payload provenance.
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [python] Externalize primary-key BLOBs and keep shared packs reachable. [paimon]
Posted by "Stephen0421 (via GitHub)" <gi...@apache.org>.
Stephen0421 commented on code in PR #10570:
URL: https://github.com/apache/paimon/pull/10570#discussion_r4283306137
##########
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:
Fixed. `commit.abort` now keeps every candidate pack in a directory when reference discovery does not finish: `list_status` is missing, listing raises, or a listed entry has no usable name. A pack is deleted only after the listing completes and no surviving `.blobref` names it. Covered by `test_commit_abort_keeps_packs_when_listing_fails`: three files share a pack, `list_status` raises `OSError`, aborting the third file deletes the data file and `.blobref` and leaves the pack.
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [python] Externalize primary-key BLOBs and keep shared packs reachable. [paimon]
Posted by "Stephen0421 (via GitHub)" <gi...@apache.org>.
Stephen0421 commented on code in PR #10570:
URL: https://github.com/apache/paimon/pull/10570#discussion_r4290760004
##########
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:
Fixed by moving managed-payload resolution into the reader chain. The internal read schema retains predicate-only columns, and the original predicate is evaluated on resolved payloads before output projection. This also avoids reapplying payload predicates to descriptor bytes in the projection fallback.
Regression coverage includes projected-out payload columns and reordered projections through both Arrow and iterator APIs, for raw and merge reads.
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [python] Externalize primary-key BLOBs and keep shared packs reachable. [paimon]
Posted by "Stephen0421 (via GitHub)" <gi...@apache.org>.
Stephen0421 commented on code in PR #10570:
URL: https://github.com/apache/paimon/pull/10570#discussion_r4307306429
##########
paimon-python/pypaimon/blob/managed_blob_payload.py:
##########
@@ -0,0 +1,129 @@
+# 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.
+
+"""Resolve primary-key managed BLOB descriptors to payload bytes."""
+
+from pypaimon.blob.managed_blob_reference_collector import _map_items
+from pypaimon.schema.data_types import (
+ is_array_blob_type,
+ is_blob_type,
+ is_map_blob_type,
+ managed_blob_field_names,
+)
+from pypaimon.table.row.blob import Blob, BlobDescriptor
+from pypaimon.utils.path import resolve_path
+
+
+def resolve_table_blob_uri(table_path, uri):
+ """Join a table-relative pack URI onto the table directory.
+
+ Absolute paths and URIs that already carry a scheme are left unchanged so
+ descriptors written with a filesystem path still open.
+ """
+ if uri is None or table_path is None:
+ return uri
+ normalized = uri.replace('\\', '/')
+ if (normalized.startswith('/')
+ or normalized.startswith('file:')
+ or '://' in normalized):
+ return uri
+ return resolve_path(str(table_path), normalized)
Review Comment:
Fixed in 556033789.
A relative local warehouse path is now qualified with `os.path.abspath` before it is published in the descriptor and `.blobref`. Absolute paths, `file:` URIs, scheme URIs, and Windows drive paths are left unchanged, so `resolve_table_blob_uri` no longer prepends `table_path` a second time.
Covered by `test_relative_warehouse_publishes_absolute_pack_uri`: warehouse `./wh`, write `(1, b'visible')`, then read. The payload comes back, and the descriptor URI is absolute, starts with the absolute warehouse path, contains `default.db` once, and the pack exists.
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [python] Externalize primary-key BLOBs and keep shared packs reachable. [paimon]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi commented on code in PR #10570:
URL: https://github.com/apache/paimon/pull/10570#discussion_r4284923208
##########
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:
Verified fixed at c7f4414 in the Python abort path: a real three-file shared-pack fixture passed 9 cases / 144 assertions, including missing/failed/partial listings and unreadable sidecars; partial abort preserves committed payloads and full abort still cleans up. Resolving this original issue. The separate Java extra_files cleanup gap is reported in the new review.
##########
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:
Verified fixed at c7f4414 for the original failure/retry path: both one shared open pack and rolling 32-byte packs survived a later partial externalization failure, retry -> commit, and scalar/array/map payload readback. Explicit abort before prepare also cleaned owned packs. Resolving this original issue.
##########
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:
Still reproducible at c7f4414 when projection is involved. With committed (id=1,payload=b"visible"),(id=2,payload=b"other") and native write/read/scan disabled, equal("payload",b"visible") works without projection, but with_projection(["id"]) returns [1,2] instead of [1], and with_projection(["payload","id"]) or ["id","payload"] returns no rows instead of id=1. Both to_arrow and to_iterator behave this way (likewise for b"other"). _managed_blob_specs uses the output adapter schema, so projected-out predicate columns are never materialized/filtered; the outer-projection fallback can also reapply the original predicate to descriptor bytes before materialization. Please retain predicate dependencies in the internal read schema and ensure filtering occurs once, on resolved payload values, before output projection.
##########
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:
Verified fixed at c7f4414: real managed PK writes now support row-iterator get_blob(...).to_data() against table-relative descriptors, and the relevant URI/Blob/reader regressions pass. Resolving this original issue.
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [python] Externalize primary-key BLOBs and keep shared packs reachable. [paimon]
Posted by "Stephen0421 (via GitHub)" <gi...@apache.org>.
Stephen0421 commented on code in PR #10570:
URL: https://github.com/apache/paimon/pull/10570#discussion_r4321318904
##########
paimon-python/pypaimon/schema/schema_manager.py:
##########
@@ -349,6 +350,57 @@ def _assert_not_renaming_blob_column(
)
+def _effective_bucket_keys(options, primary_keys, partition_keys):
+ configured = (options or {}).get(CoreOptions.BUCKET_KEY.key())
+ if configured and str(configured).strip():
+ return [key.strip() for key in str(configured).split(',') if key.strip()]
+ return [key for key in primary_keys if key not in partition_keys]
+
+
+def _validate_primary_key_managed_blobs(fields, options, primary_keys, partition_keys):
+ """Match Java SchemaValidation for primary-key managed BLOB fields."""
+ from pypaimon.common.options.core_options import ChangelogProducer, MergeEngine
+
+ core_options = CoreOptions(Options(options or {}))
+ inline = core_options.blob_descriptor_fields() | core_options.blob_view_fields()
+ managed = managed_blob_field_names(fields, inline)
+ if not managed:
+ return
+
+ engine = core_options.merge_engine()
Review Comment:
Fixed in a56428242.
Validation now rejects BLOB sequence-group ordering fields, including descriptor/view fields, and enforces the managed-BLOB aggregate/retract restriction. It handles explicit and default aggregate functions, the ignore-delete exemption, and Java-compatible `ignore-retract` boolean values.
These checks run before schema persistence for both table creation and option changes. Python/Java accept/reject parity was verified for the reported cases and valid alternatives.
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [python] Externalize primary-key BLOBs and keep shared packs reachable. [paimon]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi commented on code in PR #10570:
URL: https://github.com/apache/paimon/pull/10570#discussion_r4306336421
##########
paimon-python/pypaimon/blob/managed_blob_payload.py:
##########
@@ -0,0 +1,129 @@
+# 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.
+
+"""Resolve primary-key managed BLOB descriptors to payload bytes."""
+
+from pypaimon.blob.managed_blob_reference_collector import _map_items
+from pypaimon.schema.data_types import (
+ is_array_blob_type,
+ is_blob_type,
+ is_map_blob_type,
+ managed_blob_field_names,
+)
+from pypaimon.table.row.blob import Blob, BlobDescriptor
+from pypaimon.utils.path import resolve_path
+
+
+def resolve_table_blob_uri(table_path, uri):
+ """Join a table-relative pack URI onto the table directory.
+
+ Absolute paths and URIs that already carry a scheme are left unchanged so
+ descriptors written with a filesystem path still open.
+ """
+ if uri is None or table_path is None:
+ return uri
+ normalized = uri.replace('\\', '/')
+ if (normalized.startswith('/')
+ or normalized.startswith('file:')
+ or '://' in normalized):
+ return uri
+ return resolve_path(str(table_path), normalized)
Review Comment:
[P2] Qualify pack URIs before publishing them from relative warehouses
A relative local warehouse is retained by FileSystemCatalog. The externalizer now publishes the pack writer's full path, so with warehouse `./wh` the descriptor already contains `./wh/default.db/t/bucket-0/<pack>.managed.blob`. This branch treats it as table-relative and prepends `table_path` again. I reproduced this through public Python APIs on this head with native write/read/scan disabled: create a PK table `(id INT, payload BLOB)`, write `(1, b'visible')`, commit, then `new_read().to_arrow(...)`. Commit succeeds, but read fails trying to open `wh/default.db/t/wh/default.db/t/bucket-0/<pack>.managed.blob`. Please publish a qualified/absolute pack URI for local warehouses (also matching Java's direct-URI contract), or reliably distinguish these complete relative paths from table-relative descriptors. Add the relative-warehouse write → commit → read case.
##########
paimon-python/pypaimon/casting/schema_evolution.py:
##########
@@ -187,4 +187,47 @@ def cast_array_for_schema_evolution(array, target_type, source_paimon_type=None,
if (pa.types.is_signed_integer(target_type)
and (pa.types.is_floating(array.type) or pa.types.is_decimal(array.type))):
return _numeric_to_integer(array, target_type)
- return array.cast(target_type, safe=False)
+ try:
+ return array.cast(target_type, safe=False)
+ except pa.ArrowNotImplementedError:
+ # PyArrow 6 cannot cast a map or list as a whole when only the child
+ # binary type changes. Child arrays themselves still cast.
+ rebuilt = _cast_nested_container(array, target_type)
+ if rebuilt is None:
+ raise
+ return rebuilt
+
+
+def _cast_child(array, target_type):
+ if array.type == target_type:
+ return array
+ try:
+ return array.cast(target_type, safe=False)
+ except pa.ArrowNotImplementedError:
+ return _cast_nested_container(array, target_type)
+
+
+def _cast_nested_container(array, target_type):
+ """Rebuild a list or map after casting its children."""
+ if isinstance(array, pa.ChunkedArray):
+ chunks = [_cast_nested_container(chunk, target_type) for chunk in array.chunks]
+ if any(chunk is None for chunk in chunks):
+ return None
+ return pa.chunked_array(chunks, type=target_type)
+ if pa.types.is_list(array.type) and pa.types.is_list(target_type):
+ values = _cast_child(array.values, target_type.value_type)
+ if values is None:
+ return None
+ return pa.ListArray.from_arrays(array.offsets, values, mask=array.is_null())
+ if pa.types.is_large_list(array.type) and pa.types.is_large_list(target_type):
+ values = _cast_child(array.values, target_type.value_type)
+ if values is None:
+ return None
+ return pa.LargeListArray.from_arrays(array.offsets, values, mask=array.is_null())
+ if pa.types.is_map(array.type) and pa.types.is_map(target_type):
+ keys = _cast_child(array.keys, target_type.key_type)
+ items = _cast_child(array.items, target_type.item_type)
+ if keys is None or items is None:
+ return None
+ return pa.MapArray.from_arrays(array.offsets, keys, items, mask=array.is_null())
Review Comment:
[P2] Use a MAP reconstruction path supported by PyArrow 6
The fallback is specifically needed on older Arrow, but PyArrow 6.0.1's MapArray has no `offsets` attribute. The current Python 3.6 CI job reproduces this in both newly added mixed tests: `test_py_write_managed_blob` and `test_read_python_managed_blob_after_java_gc`. Reading `MAP<STRING,BLOB>` first hits the unsupported `map<string,binary>` → `map<string,large_binary>` cast, then raises `AttributeError` here, so normal managed MAP BLOB reads fail on this supported runtime. See the failed job: https://github.com/apache/paimon/actions/runs/39092744792/job/117572794832 . Please rebuild the nested array using APIs available in the minimum supported PyArrow version, preserving sliced offsets and null maps, and run this mixed regression on that version.
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [python] Externalize primary-key BLOBs and keep shared packs reachable. [paimon]
Posted by "Stephen0421 (via GitHub)" <gi...@apache.org>.
Stephen0421 commented on code in PR #10570:
URL: https://github.com/apache/paimon/pull/10570#discussion_r4283311414
##########
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:
Fixed. `to_iterator` now carries the table path into `OffsetRow`, and `Blob.from_bytes` / `from_descriptor_bytes` resolve a table-relative URI against that path before opening the pack. Covered by `test_row_iterator_opens_table_relative_blob`: `to_iterator` then `row.get_blob(1).to_data()` returns `b'visible'`.
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [python] Externalize primary-key BLOBs and keep shared packs reachable. [paimon]
Posted by "Stephen0421 (via GitHub)" <gi...@apache.org>.
Stephen0421 commented on code in PR #10570:
URL: https://github.com/apache/paimon/pull/10570#discussion_r4296712021
##########
paimon-python/pypaimon/blob/primary_key_blob_externalizer.py:
##########
@@ -0,0 +1,299 @@
+# 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._handed_off = set()
+ self._pending_hand_off = []
+ 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, self._handed_off)
+
+ 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()
+ # Successful data files already took ownership of their packs.
+ # A failed write followed by a retry can leave closed, unused packs.
+ self._delete_uncommitted_packs()
+ except Exception:
+ self.abort()
+ raise
+
+ def take_referenced_pack_names(self):
+ """Pack file names referenced by the data file just externalized.
+
+ Ownership stays with this writer until ``hand_off_pending``. A failure
+ while finishing the data file must still be able to delete these packs.
+ """
+ names = self._referenced_pack_names
+ self._referenced_pack_names = []
+ self._pending_hand_off = list(names)
+ return names
+
+ def hand_off_pending(self):
+ """The data file that references these packs is now recorded."""
+ names = self._pending_hand_off
+ self._pending_hand_off = []
+ self._hand_off_pack_names(names)
+
+ def discard_pending_hand_off(self):
+ """The data file was not recorded. Keep the packs deletable."""
+ self._pending_hand_off = []
+
+ def _hand_off_pack_names(self, names):
+ handed = set(names)
+ if not handed:
+ return
+ self._handed_off.update(handed)
+ self._uncommitted[:] = [
+ path for path in self._uncommitted
+ if _pack_basename(path) not in handed
+ ]
+
+ def abort(self):
+ for writer in self._writers.values():
+ writer.abort_current()
+ self._delete_uncommitted_packs()
+
+ def _delete_uncommitted_packs(self):
+ for path in list(self._uncommitted):
+ if _pack_basename(path) in self._handed_off:
+ continue
+ self._file_io.delete_quietly(path)
+ self._uncommitted[:] = [
+ path for path in self._uncommitted
+ if _pack_basename(path) in self._handed_off
+ ]
+
+ 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):
+ return normalized[len(prefix):]
Review Comment:
Fixed in 27bffdba3.
Python now preserves the pack writer's full URI in both descriptors and `.blobref` references, matching Java's path resolution contract.
Added a cross-language regression covering Python write → Java table read → Java managed-pack GC → Python readback, including scalar, ARRAY and MAP BLOBs. With live packs aged 72 hours and a 24-hour GC cutoff, all three live packs remain readable while an equally old orphan pack is deleted.
##########
paimon-python/pypaimon/blob/managed_blob_payload.py:
##########
@@ -0,0 +1,127 @@
+# 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.
+
+"""Resolve primary-key managed BLOB descriptors to payload bytes."""
+
+from pypaimon.blob.managed_blob_reference_collector import _map_items
+from pypaimon.schema.data_types import (
+ is_array_blob_type,
+ is_blob_type,
+ is_map_blob_type,
+ managed_blob_field_names,
+)
+from pypaimon.table.row.blob import Blob, BlobDescriptor
+from pypaimon.utils.path import resolve_path
+
+
+def resolve_table_blob_uri(table_path, uri):
+ """Join a table-relative pack URI onto the table directory.
+
+ Absolute paths and URIs that already carry a scheme are left unchanged so
+ descriptors written with a filesystem path still open.
+ """
+ if uri is None or table_path is None:
+ return uri
+ normalized = uri.replace('\\', '/')
+ if (normalized.startswith('/')
+ or normalized.startswith('file:')
+ or '://' in normalized):
+ return uri
+ return resolve_path(str(table_path), normalized)
+
+
+def managed_blob_output_specs(table, fields, blob_as_descriptor):
+ """``(index, name, kind)`` for primary-key managed BLOB columns.
+
+ Append tables store these columns in dedicated ``.blob`` files and the
+ reader has already turned them into payload bytes. Managed packs exist
+ only on primary-key tables.
+ """
+ if blob_as_descriptor or not table.is_primary_key_table:
+ return []
+ inline = (table.options.blob_descriptor_fields()
+ | table.options.blob_view_fields())
+ managed = managed_blob_field_names(table.fields, inline)
+ specs = []
+ for index, field in enumerate(fields):
+ if field.name not in managed:
Review Comment:
Fixed in 27bffdba3.
Managed BLOB materialization now identifies projected fields by their preserved source field IDs rather than output names. Selected MAP values are therefore resolved as scalar BLOBs, including named aliases.
Added readback coverage for single-file and overlapping-file merge reads, selected-key projections, aliases, multiple keys, name collisions, null/missing values, and mixed projections. Also verified predicate/limit behavior and that `blob-as-descriptor=true` still returns 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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [python] Externalize primary-key BLOBs and keep shared packs reachable. [paimon]
Posted by "Stephen0421 (via GitHub)" <gi...@apache.org>.
Stephen0421 commented on code in PR #10570:
URL: https://github.com/apache/paimon/pull/10570#discussion_r4307307089
##########
paimon-python/pypaimon/casting/schema_evolution.py:
##########
@@ -187,4 +187,47 @@ def cast_array_for_schema_evolution(array, target_type, source_paimon_type=None,
if (pa.types.is_signed_integer(target_type)
and (pa.types.is_floating(array.type) or pa.types.is_decimal(array.type))):
return _numeric_to_integer(array, target_type)
- return array.cast(target_type, safe=False)
+ try:
+ return array.cast(target_type, safe=False)
+ except pa.ArrowNotImplementedError:
+ # PyArrow 6 cannot cast a map or list as a whole when only the child
+ # binary type changes. Child arrays themselves still cast.
+ rebuilt = _cast_nested_container(array, target_type)
+ if rebuilt is None:
+ raise
+ return rebuilt
+
+
+def _cast_child(array, target_type):
+ if array.type == target_type:
+ return array
+ try:
+ return array.cast(target_type, safe=False)
+ except pa.ArrowNotImplementedError:
+ return _cast_nested_container(array, target_type)
+
+
+def _cast_nested_container(array, target_type):
+ """Rebuild a list or map after casting its children."""
+ if isinstance(array, pa.ChunkedArray):
+ chunks = [_cast_nested_container(chunk, target_type) for chunk in array.chunks]
+ if any(chunk is None for chunk in chunks):
+ return None
+ return pa.chunked_array(chunks, type=target_type)
+ if pa.types.is_list(array.type) and pa.types.is_list(target_type):
+ values = _cast_child(array.values, target_type.value_type)
+ if values is None:
+ return None
+ return pa.ListArray.from_arrays(array.offsets, values, mask=array.is_null())
+ if pa.types.is_large_list(array.type) and pa.types.is_large_list(target_type):
+ values = _cast_child(array.values, target_type.value_type)
+ if values is None:
+ return None
+ return pa.LargeListArray.from_arrays(array.offsets, values, mask=array.is_null())
+ if pa.types.is_map(array.type) and pa.types.is_map(target_type):
+ keys = _cast_child(array.keys, target_type.key_type)
+ items = _cast_child(array.items, target_type.item_type)
+ if keys is None or items is None:
+ return None
+ return pa.MapArray.from_arrays(array.offsets, keys, items, mask=array.is_null())
Review Comment:
Fixed in 556033789.
The nested fallback no longer uses `MapArray.offsets` or `MapArray.from_arrays(..., mask=)`. Offsets are copied from `buffers()[1]` at `array.offset`, including slices, and a null map is a null entry in that offset array. The numeric offsets stay in place so neighboring rows still index the child arrays.
`test_map_binary_values_rebuild_when_container_cast_is_missing` covers a null map, a list rebuild, and a sliced map whose offsets are `[1, 1, 3]`. Python 3.6.15 CI now passes both mixed tests: `test_py_write_managed_blob` and `test_read_python_managed_blob_after_java_gc`.
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [python] Externalize primary-key BLOBs and keep shared packs reachable. [paimon]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi commented on code in PR #10570:
URL: https://github.com/apache/paimon/pull/10570#discussion_r4313742812
##########
paimon-python/pypaimon/blob/managed_blob_cleanup.py:
##########
@@ -0,0 +1,124 @@
+# 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.
+
+"""Reference-aware cleanup of shared primary-key BLOB packs."""
+
+import logging
+
+from pypaimon.blob.managed_blob_payload import resolve_table_blob_uri
+from pypaimon.blob.managed_blob_reference_file import ManagedBlobReferenceFile
+
+logger = logging.getLogger(__name__)
+
+
+def managed_blob_deletion_paths(table, paths):
+ """Keep shared packs while deleting exclusively owned data files and sidecars."""
+ deleting = list(paths)
+ deleting.extend(_managed_pack_paths(table, deleting))
+ protected = _protected_managed_packs(table, deleting)
+ return [path for path in deleting if str(path) not in protected]
+
+
+def _managed_pack_paths(table, paths):
+ """Collect colocated pack candidates before deleting aborted sidecars."""
+ packs = set()
+ for path in paths:
+ path = str(path)
+ if not path.endswith(ManagedBlobReferenceFile.REFERENCE_FILE_SUFFIX):
+ continue
+ try:
+ references = ManagedBlobReferenceFile.read(table.file_io, path)
+ parent = path.rpartition('/')[0]
+ for reference in references:
+ pack = resolve_table_blob_uri(
+ table.table_path,
+ reference.storage_root_id + '/' + reference.relative_path)
+ if (pack.rpartition('/')[0] == parent
Review Comment:
[P2] Normalize pack and sidecar paths before checking colocation
With warehouse './wh', the writer now publishes an absolute pack URI, but the data file and .blobref paths remain relative. Comparing their parent strings here therefore discards every pack candidate. I reproduced both public abort paths with real native-disabled PK BLOB tables: with an absolute warehouse, commit.abort(messages) and writer.abort() remove data, sidecar and pack; with a relative warehouse, both remove the data and sidecar but leave the entire .managed.blob payload behind. Abandoned/failed writes thus leak payload storage. Please normalize local path identities consistently before this colocation check and the surviving-reference checks, while preserving remote URI handling, and cover both abort APIs with relative warehouses.
##########
paimon-python/pypaimon/read/reader/managed_blob_record_batch_reader.py:
##########
@@ -0,0 +1,68 @@
+# 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.
+
+import pyarrow as pa
+
+from pypaimon.blob.managed_blob_payload import materialize_managed_blob
+from pypaimon.read.reader.iface.record_batch_reader import RecordBatchReader
+
+
+def _replace_batch_column(batch, index, field, array):
+ """Replace one column without ``RecordBatch.set_column``.
+
+ That method arrived in PyArrow 13. Python 3.6 and 3.7 stay on older releases.
+ """
+ columns = list(batch.columns)
+ fields = [batch.schema.field(i) for i in range(batch.num_columns)]
+ columns[index] = array
+ fields[index] = field
+ return pa.RecordBatch.from_arrays(
+ columns, schema=pa.schema(fields, metadata=batch.schema.metadata))
+
+
+class ManagedBlobRecordBatchReader(RecordBatchReader):
+ """Resolve managed payloads before filtering, limiting or projecting rows."""
+
+ def __init__(self, reader, specs, file_io, table_path):
+ self.reader = reader
+ self.specs = specs
+ self.table_path = table_path
+ self._adopt_metadata(reader)
+ self.file_io = file_io
+ # Resolved scalar BLOBs must be exposed as payloads by get_blob().
+ resolved = {index for index, _name, _kind in specs}
+ self.descriptor_field_indices = (
Review Comment:
[P2] Preserve raw-payload provenance after materializing managed BLOBs
Clearing descriptor_field_indices does not tell OffsetRow that these bytes are an already-resolved opaque payload: get_blob() still interprets magic headers as BlobView/BlobDescriptor. I wrote ordinary Arrow BLOB bytes equal to BlobViewStruct('db.user_payload', 7, 42).serialize(), then committed the real table. On this head, Arrow returns the original payload, but to_iterator(...).get_blob(1) creates a BlobView and .to_data() raises 'BlobView is not resolved'. Reading the exact same committed table with the base implementation returns a BlobRef whose .to_data() matches the original bytes. Please carry resolved-payload metadata through projection/adapters and construct BlobData directly for those fields, so valid binary payloads are not parsed again as storage references.
##########
paimon-python/pypaimon/schema/schema_manager.py:
##########
@@ -349,6 +350,57 @@ def _assert_not_renaming_blob_column(
)
+def _effective_bucket_keys(options, primary_keys, partition_keys):
+ configured = (options or {}).get(CoreOptions.BUCKET_KEY.key())
+ if configured and str(configured).strip():
+ return [key.strip() for key in str(configured).split(',') if key.strip()]
+ return [key for key in primary_keys if key not in partition_keys]
+
+
+def _validate_primary_key_managed_blobs(fields, options, primary_keys, partition_keys):
+ """Match Java SchemaValidation for primary-key managed BLOB fields."""
+ from pypaimon.common.options.core_options import ChangelogProducer, MergeEngine
+
+ core_options = CoreOptions(Options(options or {}))
+ inline = core_options.blob_descriptor_fields() | core_options.blob_view_fields()
+ managed = managed_blob_field_names(fields, inline)
+ if not managed:
+ return
+
+ engine = core_options.merge_engine()
Review Comment:
[P2] Match Java's partial-update BLOB schema restrictions
This newly permits partial-update managed BLOB schemas without all of Java's validation. For example, Python persists (id INT PK, payload BLOB, ts INT) with bucket=1, merge-engine=partial-update, fields.ts.sequence-group=payload, and fields.payload.aggregate-function=last_non_null_value; the public Java FileStoreTableFactory.create immediately rejects that schema because managed BLOB retract payloads are discarded and fields.payload.ignore-retract is missing. Likewise, using fields.payload.sequence-group=ts is accepted here but Java rejects a BLOB ordering field (also with blob-descriptor-field=payload). I persisted all three schemas through this head's Python SchemaManager and loaded them with this head's compiled Java classes. Please port the sequence-group ordering and managed-BLOB aggregate/retract restrictions before publishing schemas that other engines cannot open; this same validation gap also affects #10655.
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org