feat(core): support primary-key managed BLOBs and field-group compaction - #227
feat(core): support primary-key managed BLOBs and field-group compaction#227SteNicholas wants to merge 1 commit into
Conversation
21c13d9 to
14aa495
Compare
Adds two independent capabilities to the write and compaction paths. - Managed BLOB storage for primary-key tables. Payloads are externalized into rolling `.managed.blob` packs before they reach the merge-tree write buffer, so neither the buffer nor its spill ever carries payload bytes. A data file keeps a `.blobref` sidecar in its extra files listing the packs its rows reference, reads resolve the stored descriptors back to payload bytes, and schema validation enforces the merge-engine, key, ordering and option rules the storage relies on. A pack is owned by the commit message whose writer created it, so a failed commit and a `PrepareCommit` that gives up partway both roll their packs back through the same cleaner. Only top-level scalar BLOB columns are managed. - Compaction across the evolved field groups of a data-evolution table. Files covering the same rows form one field group; the planner packs the groups of a contiguous row-id run into bins, and each task rewrites one bin into a single file, preserving row ids and the merged sequence-number range. The rewrite keeps every input row, so the deletion vectors of the replaced groups are re-keyed onto the rewritten file and committed in the same snapshot, with `ConflictDetection` verifying that migration; both bitmap32 and bitmap64 vectors are supported. A separate entry point applies the deletions physically instead, which reassigns the row ids of the surviving rows and drops the global indexes over them. Either way a large table is compacted and committed in bounded rounds, so only one round's file metadata is live at a time. The supported surface and the known limitations are documented in docs/source/user_guide/primary_key_table.rst and docs/source/user_guide/compaction.rst.
14aa495 to
99e98f9
Compare
| if (result.empty()) { | ||
| return; | ||
| } | ||
| Status status = UncommittedFileCleaner::Delete( |
There was a problem hiding this comment.
result may contain a metadata-only upgrade, where CompactBefore and CompactAfter reuse the same file_name. If a later bucket fails during prepare, this guard invokes the cleaner, which treats that file as an uncommitted output and deletes it even though the previous snapshot still references it, including its .blobref. Before deleting CompactAfter, exclude files also present in CompactBefore, as MergeTreeWriter::DoClose already does.
| data_file_path_factory, | ||
| path_factory_->CreateDataFilePathFactory(entry.Partition(), entry.Bucket())); | ||
| } | ||
| std::vector<std::string> delete_file_paths = |
There was a problem hiding this comment.
Snapshot expiration deletes only the data file and its .blobref here, while the orphan cleaner always skips .managed.blob and does not support primary-key tables. After an update, first-row merge, or compaction drops the last descriptor reference, the pack remains forever, so managed BLOB storage grows without bound under normal workloads. Build a live set from sidecars reachable from retained snapshots, tags, and branches, then reclaim unreferenced packs.
| return std::make_unique<RecordBatch>(batch->GetPartition(), batch->GetBucket(), row_kinds, | ||
| &c_array); | ||
| }(); | ||
| if (!result.ok()) { |
There was a problem hiding this comment.
Abort() deletes every successfully written entry in uncommitted_packs_, even though the owning writer may already hold buffered or pending data that references those packs. Neither writer is poisoned after the error, so a later PrepareCommit can commit dangling descriptors; the same issue occurs after a seal failure because new_files_ has already been flushed. The failure path must either roll back the pending data and sidecars and make the writer unusable, or delete only packs not referenced by pending files.
| // re-plans whatever is left. A deletion-vector index file the round may have | ||
| // written is left to orphan cleaning, which is what collects unreferenced index | ||
| // files. | ||
| CleanupCompactOutputs(messages, path_factory, core_options); |
There was a problem hiding this comment.
FileStoreCommitImpl::TryCommitOnce deliberately treats an atomic snapshot-write error as an uncertain result and keeps outputs for recovery via FilterAndCommit. This branch instead deletes every CompactAfter on any commit error; if the snapshot was committed but the client timed out, the latest snapshot now references missing data files. Give each round a recoverable unique identifier and delete outputs only after confirming the commit did not land; otherwise keep them.
| return Status::OK(); | ||
| } | ||
|
|
||
| std::vector<const ManifestEntry*> existing_data_files; |
There was a problem hiding this comment.
This path does not first verify that a delta DELETE still matches the current file with the same Identifier by first_row_id and row_count. Because those fields are not part of Identifier, a stale DELETE can cancel a concurrently reassigned ADD; the materialization output has no row ID and therefore skips the later range check, potentially replacing current data with a stale rewrite. Compare the base ADD and delta DELETE metadata first and reject the commit on any mismatch.
| TriggerTask(single_group_bin.Drain(), partition, compact_min_file_num, tasks)); | ||
| continue; | ||
| } | ||
| PAIMON_RETURN_NOT_OK(bin.Add(std::move(field_group), weight)); |
There was a problem hiding this comment.
Blob files are dropped before packing (line 100-105), and a bin then packs several adjacent field groups, so the rewritten file spans a wider row range than the dedicated .blob files it left behind. The read side needs those ranges to match exactly, not just to be covered: MergeRangesAndSort groups files by overlapping closed ranges, and CreateUnionReader then requires every bunch of a group to report the same row count and the same first row id.
Concretely, on a data-evolution table with a blob column: commit 1 writes {id} for rows [0,2] (data file D1, no blob file); commit 2 writes {id, payload} for rows [3,5] (data file D2 + blob file B2 [3,5]). Both read fine today. A compaction with compaction.min.file-num=2 packs the two field groups into one bin and rewrites them into a single file O [0,5]. The live set is then {O[0,5], B2[3,5]}: they overlap, land in one group, and CreateUnionReader fails with "All files in a field merge split should have the same row count." Every scan of that range fails from then on, and the commit is already durable.
TestBlobFilesExcluded only covers one field group whose blob file shares its exact range, which is why this shape is not caught.
Blob coverage needs to be part of the packing decision: pass the partition's blob files into PlanPartition, compute for each field group the set of blob field ids whose files cover its exact range, and drain the bin whenever the next group's coverage set differs — so a rewritten file never spans a change in dedicated-blob coverage.
| continue; | ||
| } | ||
| const CompactIncrement& compact_increment = message_impl->GetCompactIncrement(); | ||
| if (DataEvolutionUtils::IsMaterializedCompaction(compact_increment.CompactBefore(), |
There was a problem hiding this comment.
IsMaterializedCompaction only asks whether every output lacks a first_row_id, and KeyValueDataFileWriter::CreateResult always builds its DataFileMeta with /*first_row_id=*/std::nullopt (key_value_data_file_writer.cpp:152). So every ordinary primary-key compaction — which replaces normal files and produces a merged file without a row id — satisfies the predicate and lands in materialized_buckets.
The commit is then routed through MaterializedIndexChangesProvider, which re-scans the index manifest and emits FileKind::Delete() for every global-index entry of that (partition, bucket). A plain compaction of a primary-key table carrying a pk-sorted index therefore drops an index it never touched. Tables without a global index still pay an unfiltered index-manifest read on every compaction attempt.
The other two IsMaterializedCompaction call sites are already inside the data-evolution path; this one is on the generic commit path and needs its own gate — e.g. only collecting when options_.DataEvolutionEnabled(), or requiring the replaced CompactBefore files to actually carry a first_row_id.
| data_file_path_factory.status().ToString().c_str()); | ||
| continue; | ||
| } | ||
| for (const auto& file : message_impl->GetCompactIncrement().CompactAfter()) { |
There was a problem hiding this comment.
This loop cleans up CompactAfter() data files only, so the deletion-vector index files DataEvolutionCompactDeletionVectorRewriter already wrote (CompactIncrement::NewIndexFiles()) stay on disk when the round's commit fails.
The comment on the failure path says they are "left to orphan cleaning, which is what collects unreferenced index files", but OrphanFilesCleanerImpl::SupportToClean accepts only manifest-*, manifest-list-*, *.tmp, and data-* with a data-format or .blobref suffix. An index file is named index-<uuid> (IndexPathFactory::INDEX_PREFIX), so it matches nothing and is never reclaimed.
Losing the optimistic-commit race is a normal outcome here — CheckDeletionVectorMigrationIsComplete explicitly tells the caller to "give up committing and plan again" — so every retry on a contended table leaves more index files behind for good. RunMaterializeRound has the same gap and does not even pass index_messages to the cleanup when DropGlobalIndexes fails after the rewrite succeeded.
UncommittedFileCleaner::Delete in this same PR already deletes data files, sidecars and index files for a message; calling it here (and passing the index messages on the materialize path) would cover this without new code.
2 participants
Purpose
Linked issue: close #204
Adds two independent capabilities to the write and compaction paths.
1. Managed BLOB storage for primary-key tables
A BLOB column of a primary-key table can now keep its payload out of the data file
entirely. Payloads are externalized into rolling
.managed.blobpacks before theyreach the merge-tree write buffer, so neither the buffer nor its spill ever carries
payload bytes — only the fixed-size descriptor that replaces them.
PrimaryKeyBlobExternalizersits in front of the merge-tree writer, seals a pack whenit reaches
blob.target-file-size, and hands the packs it opened to the commit messageits writer produces. A retract row drops its payload rather than storing one, and a
batch of nothing but retracts opens no pack at all. An input that already holds a
descriptor is re-materialized so the value is stored under this table's own packs.
.blobrefsidecar (ManagedBlobReferenceFile, a versionedbinary format with a golden-bytes test against the Java writer) listing the packs its
rows reference.
ManagedBlobReferenceCollectorwrites it on close; it travels in thedata file's
extra_files, so snapshot expiration, orphan cleaning and abort all treatit as part of the data file.
ManagedBlobResolvingBatchReader, which resolves each storeddescriptor back to payload bytes with one ranged read per value.
a compaction that merely inherits a pack does not own it.
UncommittedFileCleanerrollsback exactly the packs a failed commit — or a
PrepareCommitthat gives up partway —created, and leaves the inherited ones alone.
SchemaValidationenforces the merge-engine, primary-key, sequence-group ordering andoption rules the layout relies on, so an unsupported combination (
pk-clustering-override,blob-descriptor.source-table, a managed BLOB in a key or ordering field) is rejected atschema time rather than silently changing semantics.
Only top-level scalar BLOB columns are managed; a BLOB nested in a ROW/MAP/ARRAY and the
existing inline descriptor/view fields are untouched.
2. Compaction across the evolved field groups of a data-evolution table
AppendCompactCoordinatorpreviously refused a data-evolution table. It now plans such atable with
DataEvolutionCompactPlanner: files covering the exact same rows form oneevolved field group, the groups of a contiguous row-id run are bin-packed, and each task
rewrites one bin into a single normal file holding every non-dedicated column — preserving
row ids and the merged file-level sequence-number range.
re-keyed onto the rewritten file and committed in the same snapshot rather than
dropped.
DataEvolutionCompactDeletionVectorRewriterperforms the migration andConflictDetectionverifies it: a vector that went missing, shrank, or was left behindfails the commit instead of resurrecting deleted rows. Both bitmap32 and bitmap64 vectors
are supported.
MaterializeDeletionVectorsis the heavy alternative entry point: it applies thedeletions physically, which reassigns the row ids of the surviving rows and therefore
drops the global indexes over the touched partitions in the same commit. It is never done
automatically, and a range covered by a dedicated blob or vector-store file is rejected
rather than corrupted. Its commit uses the new
RowIdCheckConflictForMaterializeDeletionVectorsrange rule, since a rewrite of wholerow ranges cannot rely on the narrower column-overlap check.
RunAndCommitandMaterializeDeletionVectorssplit the row id space into boundedrounds and commit each round on its own, so only one round's file metadata is live at a
time. The split is a soft target: a cut can only land where one data manifest's row id
coverage ends before the next one's begins, and a snapshot without usable row id
statistics falls back to a single round. Rounds are independent — a failed round leaves
the committed ones committed and a later call re-plans the rest.
evolution, deletion vectors, bucket, blob layout) are now rejected when an override would
contradict the persisted schema, and the legacy
data-evolution.compaction.rewrite-row-ids=truemode fails instead of being ignored.Tests
Managed BLOB — unit
PrimaryKeyBlobExternalizerTest(13)PrepareCommithand-over, seal failure still closing the pack stream, inline descriptor fields left alone, re-materializing descriptor input, IO failureManagedBlobReferenceFileTest(14)ManagedBlobReferenceCollectorTest(8)ManagedBlobResolvingBatchReaderTest(7)UncommittedFileCleanerTest(4)SchemaValidationTest.TestPrimaryKeyManagedBlob{,SequenceGroups}SingleFileWriterTest.TestAbortExecutorRemovesCompanionFilesManaged BLOB — integration (
PkBlobTableInteTest, 10 cases)TestWriteAndReadManagedBlob,TestReadWithPrefetchAndReadAheadCache,TestCompactionRebuildsExactBlobReferences,TestFirstRowManagedBlobKeepsFirstValue,TestPartialUpdateManagedBlob,TestDeleteDropsRow,TestSnapshotExpirationRemovesSidecar,TestAbortDeletesTheManagedBlobPacksItRollsBack,TestAbortOfACompactionKeepsTheHistoricalPacks,TestAbortOfAWriteAndCompactionKeepsOnlyTheHistoricalPacks.Plus
CleanInteTest.TestOrphanFilesCleanKeepsCompanionFilesOfLiveDataFiles.Compaction — unit
DataEvolutionCompactPlannerTest(19)min-file-num, and the round windowing (cut only at coverage gaps, file budget, fallback without row id statistics, delete-only manifests ignored)DataEvolutionNormalCompactTaskTest(3),DataEvolutionMaterializeDeletionCompactTaskTest(5)DataEvolutionCompactDeletionVectorRewriterTest(6)DataEvolutionCompactGlobalIndexDropperTest(2)DataEvolutionConflictDetectionTest(10)ConflictDetectionTest(+2),AppendCompactCoordinatorTest(+2)RunAndCommit/MaterializeDeletionVectorson a plain append tableRowIdRangeConflictCheckerTest(2),MaterializedIndexChangesProviderTest(3)Compaction — integration (
DataEvolutionTableTest, 33 new cases)Field-group compaction (
TestCompactAcrossEvolvedFieldGroups, with partitions, with apartition filter, across schema evolution, after a dropped column), deletion-vector
migration (
TestCompactKeepsDeletionVectorsOfOneRowRangeGroup, bitmap64, shifts acrossgroups, only the touched index file rewritten, sibling vectors carried, out-of-group
deletion rejected), materialization (
TestMaterializeDeletionVectorsReassignsRowIds,no-op without deletions, untouched ranges left alone, global indexes dropped, other
partitions untouched, fully deleted range, option guards), bounded rounds
(
TestRunAndCommitSplitsRowIdSpaceIntoRounds, with a partition filter, moving deletionvectors every round for both bitmap kinds), and concurrency
(
TestStaleCompactMessagePreservesConcurrentPartialUpdate,TestSmallFileCompactConflictsWithConcurrentPartialUpdate,TestCompactKeepsConcurrentAppendForNextSmallFileMerge).Plus
PkCompactionInteTest.DeduplicateWith{,Bitmap64}DeletionVectors.API and Format
Yes — additive public API, plus one new on-disk file kind.
include/paimon/append/append_compact_coordinator.h: two new static entry points,RunAndCommitandMaterializeDeletionVectors, andkDefaultCandidateFilesPerRound.Runkeeps its signature; its contract is documented more tightly — on a data-evolutiontable with deletion vectors the returned vector also holds index-only messages and the
caller must commit the whole vector in one commit.
include/paimon/file_store_commit.h: new pure virtualRowIdCheckConflictForMaterializeDeletionVectors. This is a source-breaking change foran out-of-tree implementer of the interface; there is none in this repository.
include/paimon/format/format_writer.h: newLastPayloadRange()with astd::nulloptdefault, so the externalizer can point a descriptor at the bytes it just wrote without
downcasting the writer the format factory handed it.
include/paimon/defs.h: addsBLOB_COPY_BUFFER_SIZE,BLOB_DESCRIPTOR_SOURCE_TABLE,PK_CLUSTERING_OVERRIDEandDATA_EVOLUTION_COMPACTION_REWRITE_ROW_IDS(the last threedocumenting rejected configurations), and updates the
DELETION_VECTORS_ENABLEDandDELETION_VECTOR_BITMAP64notes now that a data-evolution table is compactable andbitmap64 is supported.
.managed.blobpack is the existing blob file format. New is the.blobrefsidecar, a versioned binary file with magic, version and a length-prefixedreference list;
ManagedBlobReferenceFileTest.TestJavaGoldenBytespins it against theJava bytes. The sidecar is referenced from
DataFileMeta::extra_files, an existing field,so the manifest format is unchanged. Deletion vector index files, snapshots and manifests
are unchanged.
Documentation
Yes.
docs/source/user_guide/primary_key_table.rstgains Managed BLOB Storage: how thepayload is externalized and resolved, the
.blobrefsidecar, pack ownership versusreference, and the supported surface (top-level scalar BLOB columns only) with the option
and merge-engine rules.
docs/source/user_guide/compaction.rstgains Data-Evolution Table Compaction, coveringevolved field groups, the deletion-vector migration and the commit check that verifies it,
Materializing Deletions and why it is a separate opt-in, and Bounded Rounds and
Committing. The previous note that such a table is never compacted is removed.
docs/source/user_guide/read.rstis updated where it stated the same limitation.Generative AI tooling
Generated-by: Claude Opus 5 (Claude Code)
🤖 Generated with Claude Code