You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@paimon.apache.org by ju...@apache.org on 2024/06/17 07:23:02 UTC
(paimon) branch master updated: [flink] Fix lookup IndexOutOfBounds for sequence defined with projection (#3527)
This is an automated email from the ASF dual-hosted git repository.
junhao 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 61d194512 [flink] Fix lookup IndexOutOfBounds for sequence defined with projection (#3527)
61d194512 is described below
commit 61d1945121ca472133bff18d6a67b88201800c56
Author: Jingsong Lee <ji...@gmail.com>
AuthorDate: Mon Jun 17 15:22:57 2024 +0800
[flink] Fix lookup IndexOutOfBounds for sequence defined with projection (#3527)
---
.../main/java/org/apache/paimon/types/RowType.java | 9 +++++
.../paimon/flink/lookup/FullCacheLookupTable.java | 15 +++++++-
.../paimon/flink/lookup/LookupTableTest.java | 40 ++++++++++++++++++++++
3 files changed, 63 insertions(+), 1 deletion(-)
diff --git a/paimon-common/src/main/java/org/apache/paimon/types/RowType.java b/paimon-common/src/main/java/org/apache/paimon/types/RowType.java
index fe11976de..a691dbc0b 100644
--- a/paimon-common/src/main/java/org/apache/paimon/types/RowType.java
+++ b/paimon-common/src/main/java/org/apache/paimon/types/RowType.java
@@ -99,6 +99,15 @@ public final class RowType extends DataType {
return -1;
}
+ public int[] getFieldIndices(List<String> projectFields) {
+ List<String> fieldNames = getFieldNames();
+ int[] projection = new int[projectFields.size()];
+ for (int i = 0; i < projection.length; i++) {
+ projection[i] = fieldNames.indexOf(projectFields.get(i));
+ }
+ return projection;
+ }
+
public boolean containsField(String fieldName) {
for (DataField field : fields) {
if (field.name().equals(fieldName)) {
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/FullCacheLookupTable.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/FullCacheLookupTable.java
index bd0848e7a..ed5c4c13f 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/FullCacheLookupTable.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/FullCacheLookupTable.java
@@ -85,7 +85,6 @@ public abstract class FullCacheLookupTable implements LookupTable {
private Predicate specificPartition;
public FullCacheLookupTable(Context context) {
- this.context = context;
this.table = context.table;
List<String> sequenceFields = new ArrayList<>();
if (table.primaryKeys().size() > 0) {
@@ -106,6 +105,7 @@ public abstract class FullCacheLookupTable implements LookupTable {
builder.field(f.name(), f.type());
});
projectedType = builder.build();
+ context = context.copy(table.rowType().getFieldIndices(projectedType.getFieldNames()));
this.userDefinedSeqComparator =
UserDefinedSeqComparator.create(projectedType, sequenceFields);
this.appendUdsFieldNumber = appendUdsFieldNumber.get();
@@ -114,6 +114,8 @@ public abstract class FullCacheLookupTable implements LookupTable {
this.appendUdsFieldNumber = 0;
}
+ this.context = context;
+
Options options = Options.fromMap(context.table.options());
this.projectedType = projectedType;
this.refreshAsync = options.get(LOOKUP_REFRESH_ASYNC);
@@ -354,5 +356,16 @@ public abstract class FullCacheLookupTable implements LookupTable {
this.joinKey = joinKey;
this.requiredCachedBucketIds = requiredCachedBucketIds;
}
+
+ public Context copy(int[] newProjection) {
+ return new Context(
+ table,
+ newProjection,
+ tablePredicate,
+ projectedPredicate,
+ tempPath,
+ joinKey,
+ requiredCachedBucketIds);
+ }
}
}
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/lookup/LookupTableTest.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/lookup/LookupTableTest.java
index 7cca8e25d..14643542e 100644
--- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/lookup/LookupTableTest.java
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/lookup/LookupTableTest.java
@@ -222,6 +222,46 @@ public class LookupTableTest extends TableTestBase {
assertRow(result.get(0), 1, 22, 222);
}
+ @Test
+ public void testPkTableWithSequenceFieldProjection() throws Exception {
+ Options options = new Options();
+ options.set(CoreOptions.SEQUENCE_FIELD, "f2");
+ options.set(CoreOptions.BUCKET, 1);
+ FileStoreTable storeTable = createTable(singletonList("f0"), options);
+ FullCacheLookupTable.Context context =
+ new FullCacheLookupTable.Context(
+ storeTable,
+ new int[] {0, 1},
+ null,
+ null,
+ tempDir.toFile(),
+ singletonList("f0"),
+ null);
+ table = FullCacheLookupTable.create(context, ThreadLocalRandom.current().nextInt(2) * 10);
+ table.open();
+
+ // first write
+ write(storeTable, GenericRow.of(1, 11, 111));
+ table.refresh();
+ List<InternalRow> result = table.get(row(1));
+ assertThat(result).hasSize(1);
+ assertRow(result.get(0), 1, 11);
+
+ // second write
+ write(storeTable, GenericRow.of(1, 22, 222));
+ table.refresh();
+ result = table.get(row(1));
+ assertThat(result).hasSize(1);
+ assertRow(result.get(0), 1, 22);
+
+ // not update
+ write(storeTable, GenericRow.of(1, 33, 111));
+ table.refresh();
+ result = table.get(row(1));
+ assertThat(result).hasSize(1);
+ assertRow(result.get(0), 1, 22);
+ }
+
@Test
public void testPkTablePkFilter() throws Exception {
FileStoreTable storeTable = createTable(singletonList("f0"), new Options());