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 4c760fe85a [flink] Fix lookup output projection with internal key
fields (#9392)
4c760fe85a is described below
commit 4c760fe85a549a9217020f2b946085a71d7cc15b
Author: YeJunHao <[email protected]>
AuthorDate: Wed Aug 26 14:52:07 2026 +0800
[flink] Fix lookup output projection with internal key fields (#9392)
---
.../flink/lookup/FileStoreLookupFunction.java | 19 +++++++++++---
.../org/apache/paimon/flink/LookupJoinITCase.java | 30 ++++++++++++++++++++++
.../flink/lookup/FileStoreLookupFunctionTest.java | 13 +++++++---
3 files changed, 56 insertions(+), 6 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 aa647f167c..762b0dd3ee 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
@@ -27,6 +27,7 @@ import
org.apache.paimon.flink.FlinkConnectorOptions.LookupCacheMode;
import org.apache.paimon.flink.FlinkRowData;
import org.apache.paimon.flink.FlinkRowDataWithBlob;
import org.apache.paimon.flink.FlinkRowWrapper;
+import org.apache.paimon.flink.ProjectedRowData;
import org.apache.paimon.flink.lookup.partitioner.ShuffleStrategy;
import org.apache.paimon.flink.metrics.FlinkMetricRegistry;
import org.apache.paimon.flink.utils.RuntimeContextUtils;
@@ -94,6 +95,9 @@ public class FileStoreLookupFunction implements Serializable,
Closeable {
private final FileStoreTable table;
@Nullable private final PartitionLoader partitionLoader;
private final List<String> projectFields;
+ /** Projects rows with internal lookup fields back to the fields requested
by Flink. */
+ @Nullable private final int[] outputProjection;
+
private final List<String> joinKeys;
@Nullable private final Predicate predicate;
@Nullable private final RefreshBlacklist refreshBlacklist;
@@ -143,10 +147,11 @@ public class FileStoreLookupFunction implements
Serializable, Closeable {
.mapToObj(i ->
rowType.getFieldNames().get(projection[i]))
.collect(Collectors.toList());
- this.projectFields =
+ List<String> outputFields =
Arrays.stream(projection)
.mapToObj(i -> rowType.getFieldNames().get(i))
.collect(Collectors.toList());
+ this.projectFields = new ArrayList<>(outputFields);
// add primary keys
for (String field : table.primaryKeys()) {
@@ -158,6 +163,10 @@ public class FileStoreLookupFunction implements
Serializable, Closeable {
if (partitionLoader != null) {
partitionLoader.addPartitionKeysTo(joinKeys, projectFields);
}
+ this.outputProjection =
+ outputFields.equals(projectFields)
+ ? null
+ :
outputFields.stream().mapToInt(projectFields::indexOf).toArray();
RowType projectedType = rowType.project(projectFields);
this.projectFieldsGetters =
IntStream.range(0, projectedType.getFieldCount())
@@ -359,10 +368,14 @@ public class FileStoreLookupFunction implements
Serializable, Closeable {
List<RowData> rows = new ArrayList<>();
List<InternalRow> lookupResults = lookupTable.get(key);
for (InternalRow matchedRow : lookupResults) {
- rows.add(
+ RowData rowData =
blobFields.isEmpty()
? new FlinkRowData(matchedRow)
- : new FlinkRowDataWithBlob(matchedRow, blobFields,
blobAsDescriptor));
+ : new FlinkRowDataWithBlob(matchedRow, blobFields,
blobAsDescriptor);
+ rows.add(
+ outputProjection == null
+ ? rowData
+ :
ProjectedRowData.from(outputProjection).replaceRow(rowData));
}
if (LOG.isDebugEnabled()) {
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/LookupJoinITCase.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/LookupJoinITCase.java
index 385d1d1cf6..fe522d765c 100644
---
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/LookupJoinITCase.java
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/LookupJoinITCase.java
@@ -302,6 +302,36 @@ public class LookupJoinITCase extends CatalogITCaseBase {
iterator.close();
}
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void testAsyncLookupWithDynamicPartitionProjection(boolean
primaryKey) throws Exception {
+ sql(
+ "CREATE TABLE ASYNC_DIM (i INT, j INT, dt STRING%s) "
+ + "PARTITIONED BY (`dt`) WITH ("
+ + "'lookup.cache' = 'memory', "
+ + "'continuous.discovery-interval' = '1 ms')",
+ primaryKey ? ", PRIMARY KEY (dt, i, j) NOT ENFORCED" : "");
+ sql(
+ "INSERT INTO ASYNC_DIM VALUES "
+ + "(1, 11, '2026-08-25'), "
+ + "(1, 12, '2026-08-25'), "
+ + "(2, 21, '2026-08-25')");
+
+ String query =
+ "SELECT T.i, D.j FROM T LEFT JOIN ASYNC_DIM "
+ + "/*+ OPTIONS("
+ + "'scan.partitions' = 'max_pt()', "
+ + "'lookup.dynamic-partition.refresh-interval' = '1
ms', "
+ + "'lookup.async' = 'true') */ "
+ + "FOR SYSTEM_TIME AS OF T.proctime AS D ON T.i = D.i";
+ BlockingIterator<Row, Row> iterator =
BlockingIterator.of(sEnv.executeSql(query).collect());
+
+ sql("INSERT INTO T VALUES (1), (2)");
+ assertThat(iterator.collect(3))
+ .containsExactlyInAnyOrder(Row.of(1, 11), Row.of(1, 12),
Row.of(2, 21));
+ iterator.close();
+ }
+
@Test
public void testLookupMaxPtDynamicBucketTable() throws Exception {
sql(
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 2db4d2f3be..4b3979b743 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
@@ -404,7 +404,7 @@ public class FileStoreLookupFunctionTest {
}
@Test
- public void testDebugLogRowsWithAppendedPrimaryKey() throws Exception {
+ public void testLookupProjectionWithAppendedPrimaryKey() throws Exception {
table = createStringFileStoreTable();
lookupFunction =
new FileStoreLookupFunction(table, new int[] {2, 1}, new int[]
{1}, null, null);
@@ -419,8 +419,15 @@ public class FileStoreLookupFunctionTest {
Appender appender = addLookupFunctionAppender(Level.INFO);
try {
- lookupFunction.lookup(
- new
FlinkRowData(GenericRow.of(BinaryString.fromString("key-1"))));
+ List<RowData> rows =
+ new ArrayList<>(
+ lookupFunction.lookup(
+ new FlinkRowData(
+
GenericRow.of(BinaryString.fromString("key-1")))));
+ assertThat(rows).hasSize(1);
+ assertThat(rows.get(0).getArity()).isEqualTo(2);
+
assertThat(rows.get(0).getString(0).toString()).isEqualTo("value-1");
+ assertThat(rows.get(0).getString(1).toString()).isEqualTo("key-1");
assertThat(((CollectingAppender) appender).messages)
.noneMatch(message -> message.contains("matched rows in
lookup table"));
} finally {
