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