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 a98baacf5980..dd71c41ffb97 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.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 MergeFunctionFactory wrapped() { @Override public MergeFunction 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 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 2087eb66566c..1e743b0046d5 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 @@ 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 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 @@ public void testLevelNegative() { 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 @@ private LookupMergeFunction spillingFunction(boolean withIoManager) { 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 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 140f549d50e8..f6bc8fcc423c 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.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 void testBatchProjection() throws Exception { .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 splits = readBuilder.newScan().plan().splits(); + assertThat(splits).hasSize(1); + assertThat(((DataSplit) splits.get(0)).dataFiles()).hasSize(2); + + List result = new ArrayList<>(); + try (RecordReader 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 {