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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
