Skip to content

[core] Fix lookup merge buffer serialization for projected reads #10549

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
JingsongLi merged 1 commit into from
Oct 16, 2026
Merged

[core] Fix lookup merge buffer serialization for projected reads #10549

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
24 changes: 23 additions & 1 deletion paimon-core/src/main/java/org/apache/paimon/mergetree/compact/LookupMergeFunction.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -185,8 +188,27 @@ public MergeFunctionFactory<KeyValue> wrapped() {

@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
Expand Down
103 changes: 101 additions & 2 deletions paimon-core/src/test/java/org/apache/paimon/mergetree/compact/LookupMergeFunctionTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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 =
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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<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();
}
}
}
}
46 changes: 46 additions & 0 deletions paimon-core/src/test/java/org/apache/paimon/table/PrimaryKeySimpleTableTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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<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 {
Expand Down
Loading