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 7d6d146bba [core] Fix lookup merge buffer serialization for projected
reads (#10549)
7d6d146bba is described below
commit 7d6d146bbaf33a349dc83c3caf71334bd551e3d5
Author: reswqa <[email protected]>
AuthorDate: Fri Oct 16 08:54:12 2026 +0800
[core] Fix lookup merge buffer serialization for projected reads (#10549)
---
.../mergetree/compact/LookupMergeFunction.java | 24 ++++-
.../mergetree/compact/LookupMergeFunctionTest.java | 103 ++++++++++++++++++++-
.../paimon/table/PrimaryKeySimpleTableTest.java | 46 +++++++++
3 files changed, 170 insertions(+), 3 deletions(-)
diff --git
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/LookupMergeFunction.java
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/LookupMergeFunction.java
index a98baacf59..dd71c41ffb 100644
---
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/LookupMergeFunction.java
+++
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/LookupMergeFunction.java
@@ -22,12 +22,15 @@ import org.apache.paimon.CoreOptions;
import org.apache.paimon.KeyValue;
import org.apache.paimon.data.InternalRow;
import org.apache.paimon.disk.IOManager;
+import org.apache.paimon.types.DataField;
import org.apache.paimon.types.RowType;
import org.apache.paimon.utils.CloseableIterator;
import javax.annotation.Nullable;
+import java.util.ArrayList;
import java.util.Comparator;
+import java.util.List;
/**
* A {@link MergeFunction} for lookup, this wrapper only considers the latest
high level record,
@@ -185,8 +188,27 @@ public class LookupMergeFunction implements
MergeFunction<KeyValue> {
@Override
public MergeFunction<KeyValue> create(@Nullable RowType readType) {
+ RowType bufferType = valueType;
+ if (readType != null) {
+ // PR #9978(https://github.com/apache/paimon/pull/9978) makes
some fields nullable
+ // in valueType so deletes can omit their values,
+ // while readType may still mark them NOT NULL. Keep the read
layout, but preserve
+ // this nullability so buffered deletes can be serialized
safely when spilling.
+ List<DataField> bufferFields = new ArrayList<>();
+ for (DataField readField : readType.getFields()) {
+ bufferFields.add(
+ valueType.containsField(readField.id())
+ && valueType
+ .getField(readField.id())
+ .type()
+ .isNullable()
+ ? readField.copy(true)
+ : readField);
+ }
+ bufferType = readType.copy(bufferFields);
+ }
return new LookupMergeFunction(
- wrapped.create(readType), options, keyType, valueType,
ioManager);
+ wrapped.create(readType), options, keyType, bufferType,
ioManager);
}
@Override
diff --git
a/paimon-core/src/test/java/org/apache/paimon/mergetree/compact/LookupMergeFunctionTest.java
b/paimon-core/src/test/java/org/apache/paimon/mergetree/compact/LookupMergeFunctionTest.java
index 2087eb6656..1e743b0046 100644
---
a/paimon-core/src/test/java/org/apache/paimon/mergetree/compact/LookupMergeFunctionTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/mergetree/compact/LookupMergeFunctionTest.java
@@ -20,9 +20,15 @@ package org.apache.paimon.mergetree.compact;
import org.apache.paimon.CoreOptions;
import org.apache.paimon.KeyValue;
+import org.apache.paimon.data.GenericArray;
+import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.data.serializer.InternalRowSerializer;
import org.apache.paimon.disk.IOManager;
+import org.apache.paimon.disk.IOManagerImpl;
import org.apache.paimon.options.Options;
import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.RowKind;
import org.apache.paimon.types.RowType;
import org.junit.jupiter.api.Test;
@@ -33,11 +39,17 @@ import javax.annotation.Nullable;
import java.nio.file.Path;
import static org.apache.paimon.io.DataFileTestUtils.row;
+import static org.apache.paimon.types.RowKind.DELETE;
import static org.apache.paimon.types.RowKind.INSERT;
import static org.assertj.core.api.Assertions.assertThat;
class LookupMergeFunctionTest {
+ private static final RowType KEY_TYPE =
+ DataTypes.ROW(DataTypes.FIELD(0, "_KEY_id", DataTypes.INT()));
+
+ @TempDir Path tempDir;
+
@Test
public void testKeepLowestHighLevel() {
LookupMergeFunction function =
@@ -68,8 +80,6 @@ class LookupMergeFunctionTest {
assertThat(kv.value().getInt(0)).isEqualTo(1);
}
- @TempDir Path tempDir;
-
/**
* Same scenario as {@link #testKeepLowestHighLevel()}, but with the
candidates spilled: every
* iteration over a spilled buffer deserializes fresh instances, so an
identity check against
@@ -123,4 +133,93 @@ class LookupMergeFunctionTest {
factory.withIOManager(ioManager);
return (LookupMergeFunction) factory.create();
}
+
+ @Test
+ public void testSpillWithProjectedReadType() throws Exception {
+ RowType valueType =
+ DataTypes.ROW(
+ DataTypes.FIELD(0, "id", DataTypes.INT()),
+ DataTypes.FIELD(1, "payload", DataTypes.INT()),
+ DataTypes.FIELD(2, "items",
DataTypes.ARRAY(DataTypes.INT())));
+ InternalRow result =
+ mergeWithSpill(
+ valueType,
+ valueType.project("items", "id"),
+ GenericRow.of(new GenericArray(new int[] {10, 20}), 1),
+ INSERT);
+
+ assertThat(result.getFieldCount()).isEqualTo(2);
+ assertThat(result.getArray(0).toIntArray()).containsExactly(10, 20);
+ assertThat(result.getInt(1)).isEqualTo(1);
+ }
+
+ @Test
+ public void testSpillWithNotNullReadType() throws Exception {
+ RowType valueType =
+ DataTypes.ROW(
+ DataTypes.FIELD(0, "id", DataTypes.INT().notNull()),
+ DataTypes.FIELD(1, "payload", DataTypes.INT()));
+ RowType readType = DataTypes.ROW(valueType.getField(0),
valueType.getField(1).copy(false));
+
+ InternalRow result = mergeWithSpill(valueType, readType,
GenericRow.of(1, null), DELETE);
+
+ assertThat(result.getFieldCount()).isEqualTo(2);
+ assertThat(result.getInt(0)).isEqualTo(1);
+ assertThat(result.isNullAt(1)).isTrue();
+ assertThat(readType.getTypeAt(1).isNullable()).isFalse();
+ }
+
+ @Test
+ public void testSpillWithNestedReadType() throws Exception {
+ RowType nestedType =
+ DataTypes.ROW(
+ DataTypes.FIELD(2, "items",
DataTypes.ARRAY(DataTypes.INT())),
+ DataTypes.FIELD(3, "payload",
DataTypes.INT().notNull()));
+ RowType valueType =
+ DataTypes.ROW(
+ DataTypes.FIELD(0, "id", DataTypes.INT()),
+ DataTypes.FIELD(1, "nested", nestedType));
+ RowType readType =
+ DataTypes.ROW(
+ valueType.getField(0),
+ DataTypes.FIELD(1, "nested",
nestedType.project("payload").notNull()));
+
+ InternalRow result =
+ mergeWithSpill(valueType, readType, GenericRow.of(1,
GenericRow.of(42)), INSERT);
+
+ assertThat(result.getFieldCount()).isEqualTo(2);
+ assertThat(result.getInt(0)).isEqualTo(1);
+ assertThat(result.getRow(1, 1).getInt(0)).isEqualTo(42);
+ }
+
+ private InternalRow mergeWithSpill(
+ RowType valueType, RowType readType, InternalRow value, RowKind
valueKind)
+ throws Exception {
+ Options options = new Options();
+ options.set(CoreOptions.LOOKUP_MERGE_RECORDS_THRESHOLD, 1);
+ LookupMergeFunction.Factory factory =
+ (LookupMergeFunction.Factory)
+ LookupMergeFunction.wrap(
+ DeduplicateMergeFunction.factory(),
+ new CoreOptions(options),
+ KEY_TYPE,
+ valueType);
+ try (IOManager ioManager = new IOManagerImpl(tempDir.toString())) {
+ factory.withIOManager(ioManager);
+ MergeFunction<KeyValue> function = factory.create(readType);
+ function.reset();
+ try {
+ function.add(new KeyValue().replace(row(1), 1, valueKind,
value).setLevel(0));
+ function.add(new KeyValue().replace(row(1), 2, valueKind,
value).setLevel(0));
+ KeyValue result = function.getResult();
+ assertThat(result).isNotNull();
+ assertThat(result.key().getInt(0)).isEqualTo(1);
+ assertThat(result.sequenceNumber()).isEqualTo(2);
+ assertThat(result.valueKind()).isEqualTo(valueKind);
+ return new
InternalRowSerializer(readType).copy(result.value());
+ } finally {
+ function.reset();
+ }
+ }
+ }
}
diff --git
a/paimon-core/src/test/java/org/apache/paimon/table/PrimaryKeySimpleTableTest.java
b/paimon-core/src/test/java/org/apache/paimon/table/PrimaryKeySimpleTableTest.java
index 140f549d50..f6bc8fcc42 100644
---
a/paimon-core/src/test/java/org/apache/paimon/table/PrimaryKeySimpleTableTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/table/PrimaryKeySimpleTableTest.java
@@ -24,6 +24,7 @@ import org.apache.paimon.KeyValue;
import org.apache.paimon.Snapshot;
import org.apache.paimon.catalog.TableQueryAuthResult;
import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.data.GenericArray;
import org.apache.paimon.data.GenericRow;
import org.apache.paimon.data.InternalRow;
import org.apache.paimon.disk.IOManager;
@@ -643,6 +644,51 @@ public class PrimaryKeySimpleTableTest extends
SimpleTableTestBase {
.isEqualTo(Arrays.asList("20001|21", "202|22"));
}
+ @Test
+ public void testBatchProjectionWithLookupSpill() throws Exception {
+ RowType rowType =
+ DataTypes.ROW(
+ DataTypes.FIELD(0, "pt", DataTypes.INT()),
+ DataTypes.FIELD(1, "a", DataTypes.INT()),
+ DataTypes.FIELD(2, "payload", DataTypes.INT()),
+ DataTypes.FIELD(3, "items",
DataTypes.ARRAY(DataTypes.INT())));
+ FileStoreTable table =
+ createFileStoreTable(
+ options -> {
+ options.set(FILE_FORMAT, "avro");
+ options.set(CHANGELOG_PRODUCER, LOOKUP);
+ options.set(CoreOptions.WRITE_ONLY, true);
+
options.set(CoreOptions.LOOKUP_MERGE_RECORDS_THRESHOLD, 1);
+ },
+ rowType);
+
+ try (IOManager ioManager = new IOManagerImpl(tempDir.toString())) {
+ try (StreamTableWrite write =
table.newWrite(commitUser).withIOManager(ioManager);
+ StreamTableCommit commit = table.newCommit(commitUser)) {
+ write.write(GenericRow.of(1, 10, 100, new GenericArray(new
int[] {1, 2})));
+ commit.commit(0, write.prepareCommit(true, 0));
+ write.write(GenericRow.of(1, 10, 200, new GenericArray(new
int[] {3, 4})));
+ commit.commit(1, write.prepareCommit(true, 1));
+ }
+
+ ReadBuilder readBuilder =
+ table.newReadBuilder().withReadType(rowType.project("pt",
"a", "payload"));
+ List<Split> splits = readBuilder.newScan().plan().splits();
+ assertThat(splits).hasSize(1);
+ assertThat(((DataSplit) splits.get(0)).dataFiles()).hasSize(2);
+
+ List<String> result = new ArrayList<>();
+ try (RecordReader<InternalRow> reader =
+
readBuilder.newRead().withIOManager(ioManager).createReader(splits.get(0))) {
+ reader.forEachRemaining(
+ row ->
+ result.add(
+ row.getInt(0) + "|" + row.getInt(1) +
"|" + row.getInt(2)));
+ }
+ assertThat(result).containsExactly("1|10|200");
+ }
+ }
+
@ParameterizedTest
@ValueSource(booleans = {true, false})
public void testBatchFilter(boolean statsDenseStore) throws Exception {
