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 55433db933 [flink] IndexOutOfBoundsException when lookup join (#8655)
55433db933 is described below
commit 55433db933550775d7eca2e30edd133ade986800
Author: ouyangwulin <[email protected]>
AuthorDate: Thu Jul 16 13:26:54 2026 +0800
[flink] IndexOutOfBoundsException when lookup join (#8655)
---
.../flink/lookup/FileStoreLookupFunction.java | 28 +++---
.../flink/lookup/FileStoreLookupFunctionTest.java | 107 +++++++++++++++++++++
2 files changed, 121 insertions(+), 14 deletions(-)
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/FileStoreLookupFunction.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/FileStoreLookupFunction.java
index 42c92fbdd8..77d74526a6 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/FileStoreLookupFunction.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/FileStoreLookupFunction.java
@@ -143,14 +143,6 @@ public class FileStoreLookupFunction implements
Serializable, Closeable {
.mapToObj(i -> rowType.getFieldNames().get(i))
.collect(Collectors.toList());
- this.projectFieldsGetters =
- IntStream.range(0, projection.length)
- .mapToObj(
- i ->
- InternalRow.createFieldGetter(
-
rowType.getTypeAt(projection[i]), i))
- .collect(Collectors.toList());
-
// add primary keys
for (String field : table.primaryKeys()) {
if (!projectFields.contains(field)) {
@@ -162,6 +154,10 @@ public class FileStoreLookupFunction implements
Serializable, Closeable {
partitionLoader.addPartitionKeysTo(joinKeys, projectFields);
}
RowType projectedType = rowType.project(projectFields);
+ this.projectFieldsGetters =
+ IntStream.range(0, projectedType.getFieldCount())
+ .mapToObj(i ->
InternalRow.createFieldGetter(projectedType.getTypeAt(i), i))
+ .collect(Collectors.toList());
this.blobFields =
IntStream.range(0, projectedType.getFieldCount())
.filter(i ->
BlobType.isBlobFileField(projectedType.getTypeAt(i)))
@@ -340,12 +336,16 @@ public class FileStoreLookupFunction implements
Serializable, Closeable {
}
if (LOG.isDebugEnabled()) {
- LOG.debug(
- "matched rows in lookup table, size:{}, rows:{}",
- lookupResults.size(),
- lookupResults.stream()
- .map(row -> logRow(projectFieldsGetters, row))
- .collect(Collectors.toList()));
+ try {
+ LOG.debug(
+ "matched rows in lookup table, size:{}, rows:{}",
+ lookupResults.size(),
+ lookupResults.stream()
+ .map(row -> logRow(projectFieldsGetters, row))
+ .collect(Collectors.toList()));
+ } catch (Exception e) {
+ LOG.debug("Failed to log matched rows in lookup table.", e);
+ }
}
return rows;
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/lookup/FileStoreLookupFunctionTest.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/lookup/FileStoreLookupFunctionTest.java
index 7aa0200647..ad315237b9 100644
---
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/lookup/FileStoreLookupFunctionTest.java
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/lookup/FileStoreLookupFunctionTest.java
@@ -19,6 +19,7 @@
package org.apache.paimon.flink.lookup;
import org.apache.paimon.CoreOptions;
+import org.apache.paimon.data.BinaryString;
import org.apache.paimon.data.GenericRow;
import org.apache.paimon.data.InternalRow;
import org.apache.paimon.flink.FlinkConnectorOptions;
@@ -44,6 +45,16 @@ import org.apache.paimon.types.RowType;
import org.apache.paimon.utils.TraceableFileIO;
import org.apache.flink.table.data.RowData;
+import org.apache.logging.log4j.Level;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.core.Appender;
+import org.apache.logging.log4j.core.LogEvent;
+import org.apache.logging.log4j.core.LoggerContext;
+import org.apache.logging.log4j.core.appender.AbstractAppender;
+import org.apache.logging.log4j.core.config.Configuration;
+import org.apache.logging.log4j.core.config.LoggerConfig;
+import org.apache.logging.log4j.core.config.Property;
+import org.apache.logging.log4j.core.layout.PatternLayout;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
@@ -64,6 +75,7 @@ import java.util.Collections;
import java.util.List;
import java.util.Random;
import java.util.UUID;
+import java.util.concurrent.CopyOnWriteArrayList;
import static org.apache.paimon.data.BinaryRow.EMPTY_ROW;
import static
org.apache.paimon.flink.FlinkConnectorOptions.LOOKUP_REFRESH_TIME_PERIODS_BLACKLIST;
@@ -367,6 +379,44 @@ public class FileStoreLookupFunctionTest {
assertThat(resultRow.getInt(1)).isEqualTo(expectedRow.getInt(1));
}
+ @Test
+ public void testDebugLogRowsWithAppendedPrimaryKey() throws Exception {
+ table = createStringFileStoreTable();
+ lookupFunction =
+ new FileStoreLookupFunction(table, new int[] {2, 1}, new int[]
{1}, null, null);
+ lookupFunction.open(tempDir.toString());
+
+ StreamTableWrite writer = table.newStreamWriteBuilder().newWrite();
+ writer.write(
+ GenericRow.of(
+ -1, BinaryString.fromString("key-1"),
BinaryString.fromString("value-1")));
+ commit(writer.prepareCommit(true, 1));
+ writer.close();
+
+ Appender appender = addLookupFunctionAppender(Level.INFO);
+ try {
+ lookupFunction.lookup(
+ new
FlinkRowData(GenericRow.of(BinaryString.fromString("key-1"))));
+ assertThat(((CollectingAppender) appender).messages)
+ .noneMatch(message -> message.contains("matched rows in
lookup table"));
+ } finally {
+ removeLookupFunctionAppender(appender);
+ }
+
+ appender = addLookupFunctionAppender(Level.DEBUG);
+ try {
+ lookupFunction.lookup(
+ new
FlinkRowData(GenericRow.of(BinaryString.fromString("key-1"))));
+ assertThat(((CollectingAppender) appender).messages)
+ .anyMatch(
+ message ->
+ message.contains("matched rows in lookup
table")
+ && message.contains("[value-1,
key-1, -1]"));
+ } finally {
+ removeLookupFunctionAppender(appender);
+ }
+ }
+
private void commit(List<CommitMessage> messages) throws Exception {
TableCommitImpl commit = table.newCommit(commitUser);
commit.commit(messages);
@@ -386,4 +436,61 @@ public class FileStoreLookupFunctionTest {
private InternalRow randomRow() {
return GenericRow.of(RANDOM.nextInt(100), RANDOM.nextInt(100),
RANDOM.nextLong());
}
+
+ private FileStoreTable createStringFileStoreTable() throws Exception {
+ SchemaManager schemaManager = new SchemaManager(fileIO, tablePath);
+ Options conf = new Options();
+ conf.set(CoreOptions.BUCKET, 2);
+ conf.set(RocksDBOptions.LOOKUP_CONTINUOUS_DISCOVERY_INTERVAL,
Duration.ofSeconds(1));
+
+ RowType rowType =
+ RowType.of(
+ new DataType[] {DataTypes.INT(), DataTypes.STRING(),
DataTypes.STRING()},
+ new String[] {"pt", "k", "v"});
+ Schema schema =
+ new Schema(
+ rowType.getFields(),
+ Collections.emptyList(),
+ Arrays.asList("pt", "k"),
+ conf.toMap(),
+ "");
+ TableSchema tableSchema = schemaManager.createTable(schema);
+ return FileStoreTableFactory.create(fileIO, tablePath, tableSchema);
+ }
+
+ private Appender addLookupFunctionAppender(Level level) {
+ LoggerContext context = (LoggerContext) LogManager.getContext(false);
+ Configuration configuration = context.getConfiguration();
+ CollectingAppender appender = new
CollectingAppender("lookup-function-test-appender");
+ appender.start();
+
+ LoggerConfig loggerConfig =
+ new LoggerConfig(FileStoreLookupFunction.class.getName(),
level, false);
+ loggerConfig.addAppender(appender, Level.DEBUG, null);
+ configuration.addLogger(FileStoreLookupFunction.class.getName(),
loggerConfig);
+ context.updateLoggers();
+ return appender;
+ }
+
+ private void removeLookupFunctionAppender(Appender appender) {
+ LoggerContext context = (LoggerContext) LogManager.getContext(false);
+ Configuration configuration = context.getConfiguration();
+ configuration.removeLogger(FileStoreLookupFunction.class.getName());
+ appender.stop();
+ context.updateLoggers();
+ }
+
+ private static class CollectingAppender extends AbstractAppender {
+
+ private final List<String> messages = new CopyOnWriteArrayList<>();
+
+ private CollectingAppender(String name) {
+ super(name, null, PatternLayout.createDefaultLayout(), false,
Property.EMPTY_ARRAY);
+ }
+
+ @Override
+ public void append(LogEvent event) {
+ messages.add(event.getMessage().getFormattedMessage());
+ }
+ }
}
