Repository navigation
[python] Externalize primary-key BLOBs and keep shared packs reachable. - #10570
Stephen0421 wants to merge 2 commits into
Conversation
21f3739 to
7a70eea
Compare
JingsongLi
left a comment
JingsongLi
left a comment
There was a problem hiding this comment.
Managed primary-key BLOB writes and shared-pack retention have end-to-end value. 244 relevant tests passed locally (34 native-marked tests skipped), and changed-file flake8 passed. Additional real write/commit/read and injected-IO probes found two data-loss paths and two read-path gaps below.
| batch = batch.slice(0, remaining) | ||
| batch = self._serialize_blob_views_as_descriptors( | ||
| batch, reader) | ||
| batch = self._materialize_managed_blob_batch(batch) |
There was a problem hiding this 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.
There was a problem hiding this 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').
There was a problem hiding this 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.
There was a problem hiding this 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.
7a70eea to
c7f4414
Compare
JingsongLi
left a comment
JingsongLi
left a comment
There was a problem hiding this comment.
Re-reviewed the complete current head c7f4414, including real Python write/commit/read and exact-head Java abort interoperability. 461 relevant Python tests and 15 Java DataFileMeta tests passed; changed-file flake8 and Python 3.6 syntax checks passed. The original Python abort-listing and flush-retry P1s, and relative-URI iterator issue, are fixed in the reproduced paths. However, one cross-engine data-loss path and three incorrect-result cases remain: Java abort deletes shared packs, projected payload predicates still fail (details on the existing thread), limits run before payload filtering, and large filtered batches are truncated. The native CI failure IDs match the unrelated 64-failure baseline exactly; the findings below come from separate real-table reproductions.
| collector.collect_table(data) | ||
| names = [collector.close()] | ||
| if self._blob_externalizer is not None: | ||
| names[0:0] = self._blob_externalizer.take_referenced_pack_names() |
There was a problem hiding this 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.
There was a problem hiding this 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.
| batch = self._serialize_blob_views_as_descriptors( | ||
| batch, reader) | ||
| batch = self._materialize_managed_blob_batch(batch) | ||
| batch = self._filter_resolved_managed_blob_batch(batch) |
There was a problem hiding this 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.
There was a problem hiding this 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.
| 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] |
There was a problem hiding this 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.
There was a problem hiding this 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.
c7f4414 to
4c2cb8f
Compare
JingsongLi
left a comment
JingsongLi
left a comment
There was a problem hiding this comment.
Requirement fit: SUPPORTED. Python primary-key BLOB writes need the managed-pack lifecycle and payload reads already available in Java. Implementation: FINDINGS.
Validated this head with 326 Python tests (BLOB writes/lifecycle/references/ownership, schema validation, append reads and batch filtering) and the repository flake8 configuration. The earlier ordinary payload-predicate/projection/limit and 140k-row batch regressions pass. Additional committed-table probes found the cross-engine cleanup and MAP-key projection failures below. For the Java cleanup probe I compiled the relevant Java classes directly from this PR head and used the public table reader and LocalManagedBlobOrphanFilesClean; the retained snapshot was unchanged.
| normalized = uri.replace('\\', '/') | ||
| prefix = root.replace('\\', '/').rstrip('/') + '/' | ||
| if normalized.startswith(prefix): | ||
| return normalized[len(prefix):] |
There was a problem hiding this 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.
There was a problem hiding this comment.
Fixed in 27bffdb.
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.
| managed = managed_blob_field_names(table.fields, inline) | ||
| specs = [] | ||
| for index, field in enumerate(fields): | ||
| if field.name not in managed: |
There was a problem hiding this comment.
[P2] Resolve managed BLOB leaves selected from a MAP
A supported projection such as with_projection(["mapping['k']"]) adapts the MAP 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.
There was a problem hiding this comment.
Fixed in 27bffdb.
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.
4c2cb8f to
27bffdb
Compare
f7adeff to
6fda1f1
Compare
JingsongLi
left a comment
JingsongLi
left a comment
There was a problem hiding this comment.
Requirement fit: SUPPORTED. Implementation: FINDINGS. Re-reviewed the complete current head in an isolated worktree, including the previous ownership, projection, filtering, limit and cross-engine GC findings.
Local Python 3.11: 476 relevant tests passed; all changed Python files passed the repository flake8 configuration and Python 3.6 syntax checks. The actual Python write → Java table read/managed-pack GC → Python readback chain also passed (scalar, ARRAY and MAP BLOBs, aged live packs retained, orphan deleted). The previous reproduced P1 paths are repaired.
Two current compatibility/read failures remain, detailed inline: relative local warehouses produce duplicated pack paths, and the supported Python 3.6/PyArrow 6 runtime fails in the newly added MAP cast fallback. The latter is reproduced by this exact head's CI, rather than inferred from syntax validation.
| or normalized.startswith('file:') | ||
| or '://' in normalized): | ||
| return uri | ||
| return resolve_path(str(table_path), normalized) |
There was a problem hiding this 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.
There was a problem hiding this comment.
Fixed in 5560337.
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.
| 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()) |
There was a problem hiding this 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.
There was a problem hiding this comment.
Fixed in 5560337.
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.
6fda1f1 to
411b807
Compare
The .blobref file lists packs referenced by one data file. Explicit commit abort deletes newly written data, BLOB packs, sidecars, and .row files. Writer abort still leaves packs in place when a BlobConsumer has been given them. prepare_commit still keeps file ownership on the writer.
5d018ca to
5560337
Compare
JingsongLi
left a comment
JingsongLi
left a comment
There was a problem hiding this comment.
Requirement fit: SUPPORTED. Implementation: FINDINGS. Re-reviewed the complete current head. 479 relevant Python tests passed (2 optional skips, 185 subtests), changed Python files passed flake8 and Python 3.6 syntax parsing, and the actual Python write → Java table read/managed-pack GC → Python readback chain passed with regular Maven checks. The earlier relative-warehouse read case passes; current Python 3.6 CI also passes. Two sampled unrelated CI failures reproduce identically on the exact base. Three additional production-path gaps are detailed below.
| pack = resolve_table_blob_uri( | ||
| table.table_path, | ||
| reference.storage_root_id + '/' + reference.relative_path) | ||
| if (pack.rpartition('/')[0] == parent |
There was a problem hiding this 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.
There was a problem hiding this comment.
Fixed in a564282.
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.
| 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 = ( |
There was a problem hiding this 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.
There was a problem hiding this comment.
Fixed in a564282.
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.
| if not managed: | ||
| return | ||
|
|
||
| engine = core_options.merge_engine() |
There was a problem hiding this 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.
There was a problem hiding this comment.
Fixed in a564282.
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.
Primary-key writes store payloads in .managed.blob packs and preserve the pack writer URIs in descriptors and reference sidecars for Java reader and GC compatibility. Data files register only their exclusively owned .blobref sidecars in extra_files, keeping shared packs safe from cross-engine abort cleanup. Python abort discovers pack candidates through sidecars and preserves packs when surviving references exist or reference discovery cannot complete. Resolve managed payloads in the reader chain before payload predicates, limits, and output projection. Retain predicate dependencies and resolved BLOB metadata, identify projected MAP values by source field ID, and preserve every batch produced by filtering. Finalize prepare and close by cleaning up unreferenced packs left by failed writes without discarding retry state or deleting packs already handed to successful data files. Add regression coverage for shared-pack cleanup, projected predicates and MAP keys, parallel limits, large batches, mixed BLOB iteration, and failed-write retry, close and abort paths. Cover Python writes followed by Java table reads and pack GC, then Python readback, in the mixed-language test workflow.
5560337 to
a564282
Compare
2 participants
Purpose
Primary-key Python writes externalize managed BLOB payloads into
.managed.blobpacks and store table-relative descriptors in the data file. Each data file'sextra_fileslists the.blobrefsidecar 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-descriptoris false, primary-key reads resolve those descriptors against the table directory and return payload bytes. Append tables still use dedicated.blobfiles and are unchanged.commit.abortdeletes the aborted data files and their.blobrefsidecars. A.managed.blobpack is deleted only when no.blobrefoutside 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 aBlobConsumerstay in place.Tests
.blobrefand.managed.blobonextra_files, andnew_read_builder()returns payload bytes.commit.abortkeeps a pack that committed files still reference, and those rows still read back..blobrefreferences.