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