This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 6cda45e61b [core] Allow first-row merge engine for primary-key managed
BLOB tables. (#9201)
6cda45e61b is described below
commit 6cda45e61be66ac4e0e34deacf797519b6f37429
Author: Wenchao Wu <[email protected]>
AuthorDate: Thu Aug 13 10:08:31 2026 +0800
[core] Allow first-row merge engine for primary-key managed BLOB tables.
(#9201)
---
docs/docs/primary-key-table/blob-storage.md | 33 +++++-
.../org/apache/paimon/schema/SchemaValidation.java | 7 +-
.../operation/PrimaryKeyManagedBlobStoreTest.java | 117 ++++++++++++++++++++-
.../apache/paimon/schema/SchemaValidationTest.java | 16 +++
4 files changed, 168 insertions(+), 5 deletions(-)
diff --git a/docs/docs/primary-key-table/blob-storage.md
b/docs/docs/primary-key-table/blob-storage.md
index 871822087e..44ccd638d3 100644
--- a/docs/docs/primary-key-table/blob-storage.md
+++ b/docs/docs/primary-key-table/blob-storage.md
@@ -115,7 +115,7 @@ Primary-key managed BLOB storage has the following
requirements:
| Item | Requirement |
|------|-------------|
| Managed BLOB declaration | `BLOB`, `ARRAY<BLOB>`, and `MAP<K, BLOB>` use
`blob-field`; `blob-descriptor-field` remains inline |
-| Merge engine | `deduplicate` or `partial-update` |
+| Merge engine | `deduplicate`, `partial-update`, or `first-row` |
| Changelog producer | `none` only |
| Key usage | A managed BLOB column cannot be a primary, partition, bucket, or
sequence key |
| External data paths | `data-file.external-paths` is not supported |
@@ -191,6 +191,37 @@ collector is available.
when `blob-view.resolve.enabled` is true (default). Append upstream tables
used by `sys.blob_view(...)` must enable
`row-tracking.enabled` and `data-evolution.enabled`.
+### First-row with BLOB fields
+
+Primary-key tables may use `merge-engine=first-row` with managed `blob-field`
(`BLOB`, `ARRAY<BLOB>`, or
+`MAP<K, BLOB>`):
+
+| Mode | First-row | Changelog | Notes |
+|------|-----------|-----------|-------|
+| `blob-field` (managed scalar, array, or map) | Supported |
`changelog-producer=none` only | Keeps the first non-retract row and its
managed payload |
+
+For managed BLOB columns, set `changelog-producer=none`. `first-row` normally
also supports the `lookup` changelog
+producer, but managed BLOB storage requires `none`.
+
+`first-row` keeps the earliest surviving row for the same primary key. Later
updates to the same key therefore do not
+replace an existing managed BLOB payload. This differs from `deduplicate`,
which keeps the latest row.
+
+By default, `first-row` does not accept `DELETE` or `UPDATE_BEFORE` records.
Configure `ignore-delete=true` if your
+pipeline may emit them.
+
+```sql
+CREATE TABLE training_chunks (
+ id BIGINT,
+ name STRING,
+ chunk BYTES COMMENT '__BLOB_FIELD; raw training block',
+ PRIMARY KEY (id) NOT ENFORCED
+) WITH (
+ 'merge-engine' = 'first-row',
+ 'changelog-producer' = 'none',
+ 'blob-field' = 'chunk'
+);
+```
+
## Managed BLOB Update, Delete, and Compaction
Each incoming non-null scalar BLOB, array element, or map value is written as
a new descriptor and payload before merge
diff --git
a/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java
b/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java
index 3bd841460f..1bd587d4ca 100644
--- a/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java
+++ b/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java
@@ -1432,9 +1432,10 @@ public class SchemaValidation {
checkArgument(
options.mergeEngine() == MergeEngine.DEDUPLICATE
- || options.mergeEngine() == MergeEngine.PARTIAL_UPDATE,
- "Primary-key managed BLOB tables only support the deduplicate
or "
- + "partial-update merge engine.");
+ || options.mergeEngine() == MergeEngine.PARTIAL_UPDATE
+ || options.mergeEngine() == MergeEngine.FIRST_ROW,
+ "Primary-key managed BLOB tables only support the deduplicate,
"
+ + "partial-update or first-row merge engine.");
checkArgument(
options.changelogProducer() == ChangelogProducer.NONE,
"Primary-key managed BLOB tables only support
changelog-producer 'none'.");
diff --git
a/paimon-core/src/test/java/org/apache/paimon/operation/PrimaryKeyManagedBlobStoreTest.java
b/paimon-core/src/test/java/org/apache/paimon/operation/PrimaryKeyManagedBlobStoreTest.java
index 01a0455e0f..bc83e80956 100644
---
a/paimon-core/src/test/java/org/apache/paimon/operation/PrimaryKeyManagedBlobStoreTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/operation/PrimaryKeyManagedBlobStoreTest.java
@@ -39,6 +39,9 @@ import org.apache.paimon.io.DataFilePathFactory;
import org.apache.paimon.manifest.ManifestCommittable;
import org.apache.paimon.manifest.ManifestEntry;
import org.apache.paimon.mergetree.compact.DeduplicateMergeFunction;
+import org.apache.paimon.mergetree.compact.FirstRowMergeFunction;
+import org.apache.paimon.mergetree.compact.MergeFunctionFactory;
+import org.apache.paimon.options.Options;
import org.apache.paimon.postpone.PostponeBucketWriter;
import org.apache.paimon.schema.KeyValueFieldsExtractor;
import org.apache.paimon.schema.Schema;
@@ -353,6 +356,59 @@ class PrimaryKeyManagedBlobStoreTest {
assertThat(result.valueArray().getBlob(0).toData()).isEqualTo(expected);
}
+ @Test
+ void testFirstRowManagedBlobKeepsFirstValue() throws Exception {
+ FileIO fileIO = LocalFileIO.create();
+ TestFileStore store = createFirstRowStore(fileIO);
+ byte[] first = "first".getBytes(StandardCharsets.UTF_8);
+ byte[] second = "second".getBytes(StandardCharsets.UTF_8);
+
+ try (IOManager ioManager =
IOManager.create(tempDir.resolve("io").toString())) {
+ commitFirstRowData(
+ store,
+ ioManager,
+ Collections.singletonList(keyValue(1, RowKind.INSERT,
first)),
+ 0L);
+ ManifestEntry firstFile = store.newScan().plan().files().get(0);
+ ManagedBlobReferenceFile.Reference firstReference =
+ references(fileIO, store, firstFile).get(0);
+
+ commitFirstRowData(
+ store,
+ ioManager,
+ Collections.singletonList(keyValue(1,
RowKind.UPDATE_AFTER, second)),
+ 1L);
+ List<ManifestEntry> filesBeforeCompaction =
store.newScan().plan().files();
+ assertThat(filesBeforeCompaction).hasSize(2);
+ ManagedBlobReferenceFile.Reference secondReference = null;
+ for (ManifestEntry file : filesBeforeCompaction) {
+ for (ManagedBlobReferenceFile.Reference reference :
+ references(fileIO, store, file)) {
+ if (!reference.equals(firstReference)) {
+ secondReference = reference;
+ }
+ }
+ }
+ assertThat(secondReference).isNotNull();
+ assertThat(secondReference).isNotEqualTo(firstReference);
+
+ KeyValue read =
+
store.readKvsFromSnapshot(store.snapshotManager().latestSnapshotId()).get(0);
+ assertThat(read.value().getBlob(1).toData()).isEqualTo(first);
+
+ forceFullCompaction(store, ioManager);
+
+ List<ManifestEntry> files = store.newScan().plan().files();
+ assertThat(files).hasSize(1);
+ List<ManagedBlobReferenceFile.Reference> compactedReferences =
+ references(fileIO, store, files.get(0));
+ assertThat(compactedReferences).containsExactly(firstReference);
+ assertThat(compactedReferences).doesNotContain(secondReference);
+ read =
store.readKvsFromSnapshot(store.snapshotManager().latestSnapshotId()).get(0);
+ assertThat(read.value().getBlob(1).toData()).isEqualTo(first);
+ }
+ }
+
@Test
void testCompactionRebuildsExactBlobReferences() throws Exception {
FileIO fileIO = LocalFileIO.create();
@@ -480,6 +536,22 @@ class PrimaryKeyManagedBlobStoreTest {
return createStore(fileIO, "payloads",
DataTypes.MAP(DataTypes.STRING(), DataTypes.BLOB()));
}
+ private TestFileStore createFirstRowStore(FileIO fileIO) throws Exception {
+ return createStore(
+ fileIO,
+ "payload",
+ DataTypes.BLOB(),
+ 1,
+ options -> {
+ options.put(
+ CoreOptions.MERGE_ENGINE.key(),
+ CoreOptions.MergeEngine.FIRST_ROW.toString());
+ options.put(
+ CoreOptions.CHANGELOG_PRODUCER.key(),
+ CoreOptions.ChangelogProducer.NONE.toString());
+ });
+ }
+
private TestFileStore createStore(FileIO fileIO, String payloadName,
DataType payloadType)
throws Exception {
return createStore(fileIO, payloadName, payloadType, 1);
@@ -487,6 +559,16 @@ class PrimaryKeyManagedBlobStoreTest {
private TestFileStore createStore(
FileIO fileIO, String payloadName, DataType payloadType, int
bucket) throws Exception {
+ return createStore(fileIO, payloadName, payloadType, bucket, ignored
-> {});
+ }
+
+ private TestFileStore createStore(
+ FileIO fileIO,
+ String payloadName,
+ DataType payloadType,
+ int bucket,
+ java.util.function.Consumer<Map<String, String>> optionsCustomizer)
+ throws Exception {
Path tablePath = new Path(tempDir.toUri());
List<DataField> valueFields =
Arrays.asList(
@@ -504,6 +586,13 @@ class PrimaryKeyManagedBlobStoreTest {
options.put(CoreOptions.BUCKET.key(), String.valueOf(bucket));
options.put(CoreOptions.BLOB_FIELD.key(), payloadName);
options.put(CoreOptions.BLOB_TARGET_FILE_SIZE.key(), "1 b");
+ optionsCustomizer.accept(options);
+ MergeFunctionFactory<KeyValue> mergeFunctionFactory =
+ CoreOptions.MergeEngine.FIRST_ROW
+ .toString()
+
.equals(options.get(CoreOptions.MERGE_ENGINE.key()))
+ ?
FirstRowMergeFunction.factory(Options.fromMap(options))
+ : DeduplicateMergeFunction.factory();
TableSchema schema =
new SchemaManager(fileIO, tablePath)
.createTable(
@@ -535,7 +624,7 @@ class PrimaryKeyManagedBlobStoreTest {
keyType,
valueType,
extractor,
- DeduplicateMergeFunction.factory(),
+ mergeFunctionFactory,
schema)
.build();
}
@@ -557,9 +646,35 @@ class PrimaryKeyManagedBlobStoreTest {
return ManagedBlobReferenceFile.read(fileIO, sidecar);
}
+ private void commitFirstRowData(
+ TestFileStore store, IOManager ioManager, List<KeyValue> kvs, long
identifier)
+ throws Exception {
+ AbstractFileStoreWrite<KeyValue> write = store.newWrite();
+ try {
+ write.withIOManager(ioManager);
+ for (KeyValue kv : kvs) {
+ write.write(BinaryRow.EMPTY_ROW, 0, kv);
+ }
+ List<CommitMessage> messages = write.prepareCommit(false,
identifier);
+ try (FileStoreCommit commit = store.newCommit()) {
+ commit.commit(new ManifestCommittable(identifier, null,
messages), false);
+ }
+ } finally {
+ write.close();
+ }
+ }
+
private void forceFullCompaction(TestFileStore store) throws Exception {
+ forceFullCompaction(store, null);
+ }
+
+ private void forceFullCompaction(
+ TestFileStore store, @javax.annotation.Nullable IOManager
ioManager) throws Exception {
AbstractFileStoreWrite<KeyValue> write = store.newWrite();
try {
+ if (ioManager != null) {
+ write.withIOManager(ioManager);
+ }
write.compact(BinaryRow.EMPTY_ROW, 0, true);
List<CommitMessage> messages = write.prepareCommit(true, 1000L);
try (FileStoreCommit commit = store.newCommit()) {
diff --git
a/paimon-core/src/test/java/org/apache/paimon/schema/SchemaValidationTest.java
b/paimon-core/src/test/java/org/apache/paimon/schema/SchemaValidationTest.java
index 417a00b93b..a1801ab9c0 100644
---
a/paimon-core/src/test/java/org/apache/paimon/schema/SchemaValidationTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/schema/SchemaValidationTest.java
@@ -631,6 +631,22 @@ class SchemaValidationTest {
assertThatCode(() ->
validateTableSchema(schema)).doesNotThrowAnyException();
}
+ @Test
+ public void testPrimaryKeyManagedBlobAllowsFirstRow() {
+ Map<String, String> options = new HashMap<>();
+ options.put(BUCKET.key(), "1");
+ options.put(CoreOptions.BLOB_FIELD.key(), "payload");
+ options.put(CoreOptions.MERGE_ENGINE.key(), "first-row");
+ options.put(CoreOptions.CHANGELOG_PRODUCER.key(), "none");
+
+ assertThatCode(
+ () ->
+ validateTableSchema(
+ primaryKeyBlobSchema(
+ options, singletonList("id"),
emptyList())))
+ .doesNotThrowAnyException();
+ }
+
@Test
public void testPrimaryKeyBlobRejectsUnsupportedSemantics() {
Map<String, String> options = new HashMap<>();
