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