You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@paimon.apache.org by lz...@apache.org on 2024/01/10 14:05:13 UTC
(incubator-paimon) 04/06: [core] Fix keyRearrange in PrimaryKeyLookupTable
This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-paimon.git
commit af25cd2c67d8aa8020971269055b972019f61ce4
Author: Jingsong <ji...@gmail.com>
AuthorDate: Tue Jan 9 20:17:28 2024 +0800
[core] Fix keyRearrange in PrimaryKeyLookupTable
---
.../flink/lookup/FixedBucketFromPkExtractor.java | 5 +----
.../paimon/flink/lookup/FullCacheLookupTable.java | 2 +-
.../paimon/flink/lookup/PrimaryKeyLookupTable.java | 26 ++++++++++++++++++++--
.../flink/lookup/PrimaryKeyPartialLookupTable.java | 23 ++++++++++++-------
.../flink/lookup/SecondaryIndexLookupTable.java | 2 +-
.../org/apache/paimon/flink/LookupJoinITCase.java | 8 ++++---
6 files changed, 47 insertions(+), 19 deletions(-)
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/FixedBucketFromPkExtractor.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/FixedBucketFromPkExtractor.java
index fa5acf427..61617988a 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/FixedBucketFromPkExtractor.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/FixedBucketFromPkExtractor.java
@@ -33,8 +33,6 @@ public class FixedBucketFromPkExtractor implements KeyAndBucketExtractor<Interna
private transient InternalRow primaryKey;
- private BinaryRow bucketKey;
-
private final boolean sameBucketKeyAndTrimmedPrimaryKey;
private final int numBuckets;
@@ -90,8 +88,7 @@ public class FixedBucketFromPkExtractor implements KeyAndBucketExtractor<Interna
return trimmedPrimaryKey();
}
- bucketKey = bucketKeyProjection.apply(primaryKey);
- return bucketKey;
+ return bucketKeyProjection.apply(primaryKey);
}
@Override
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 ed4451053..3b6a226a9 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
@@ -171,7 +171,7 @@ public abstract class FullCacheLookupTable implements LookupTable {
return new NoPrimaryKeyLookupTable(context, lruCacheSize);
} else {
if (new HashSet<>(primaryKeys).equals(new HashSet<>(context.joinKey))) {
- return new PrimaryKeyLookupTable(context, lruCacheSize);
+ return new PrimaryKeyLookupTable(context, lruCacheSize, context.joinKey);
} else {
return new SecondaryIndexLookupTable(context, lruCacheSize);
}
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/PrimaryKeyLookupTable.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/PrimaryKeyLookupTable.java
index 9ef35bea8..8757effa3 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/PrimaryKeyLookupTable.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/PrimaryKeyLookupTable.java
@@ -22,10 +22,14 @@ import org.apache.paimon.data.InternalRow;
import org.apache.paimon.data.serializer.InternalSerializers;
import org.apache.paimon.lookup.BulkLoader;
import org.apache.paimon.lookup.RocksDBValueState;
+import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.types.RowKind;
import org.apache.paimon.utils.KeyProjectedRow;
+import org.apache.paimon.utils.ProjectedRow;
import org.apache.paimon.utils.TypeUtils;
+import javax.annotation.Nullable;
+
import java.io.IOException;
import java.util.Collections;
import java.util.Iterator;
@@ -40,11 +44,15 @@ public class PrimaryKeyLookupTable extends FullCacheLookupTable {
protected final KeyProjectedRow primaryKeyRow;
- public PrimaryKeyLookupTable(Context context, long lruCacheSize) throws IOException {
+ @Nullable private final ProjectedRow keyRearrange;
+
+ public PrimaryKeyLookupTable(Context context, long lruCacheSize, List<String> joinKey)
+ throws IOException {
super(context);
List<String> fieldNames = projectedType.getFieldNames();
+ FileStoreTable table = context.table;
this.primaryKeyMapping =
- context.table.primaryKeys().stream().mapToInt(fieldNames::indexOf).toArray();
+ table.primaryKeys().stream().mapToInt(fieldNames::indexOf).toArray();
this.primaryKeyRow = new KeyProjectedRow(primaryKeyMapping);
this.tableState =
@@ -54,10 +62,24 @@ public class PrimaryKeyLookupTable extends FullCacheLookupTable {
TypeUtils.project(projectedType, primaryKeyMapping)),
InternalSerializers.create(projectedType),
lruCacheSize);
+
+ ProjectedRow keyRearrange = null;
+ if (!table.primaryKeys().equals(joinKey)) {
+ keyRearrange =
+ ProjectedRow.from(
+ table.primaryKeys().stream()
+ .map(joinKey::indexOf)
+ .mapToInt(value -> value)
+ .toArray());
+ }
+ this.keyRearrange = keyRearrange;
}
@Override
public List<InternalRow> innerGet(InternalRow key) throws IOException {
+ if (keyRearrange != null) {
+ key = keyRearrange.replaceRow(key);
+ }
InternalRow value = tableState.get(key);
return value == null ? Collections.emptyList() : Collections.singletonList(value);
}
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/PrimaryKeyPartialLookupTable.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/PrimaryKeyPartialLookupTable.java
index fc950948f..90511bb24 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/PrimaryKeyPartialLookupTable.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/PrimaryKeyPartialLookupTable.java
@@ -47,7 +47,7 @@ public class PrimaryKeyPartialLookupTable implements LookupTable {
private final TableFileMonitor fileMonitor;
- private final ProjectedRow projectedKey;
+ @Nullable private final ProjectedRow keyRearrange;
public PrimaryKeyPartialLookupTable(
FileStoreTable table,
@@ -71,12 +71,17 @@ public class PrimaryKeyPartialLookupTable implements LookupTable {
.withIOManager(new IOManagerImpl(tempPath.toString()));
this.extractor = new FixedBucketFromPkExtractor(table.schema());
this.fileMonitor = new TableFileMonitor(table, predicate);
- this.projectedKey =
- ProjectedRow.from(
- table.primaryKeys().stream()
- .map(joinKey::indexOf)
- .mapToInt(value -> value)
- .toArray());
+
+ ProjectedRow keyRearrange = null;
+ if (!table.primaryKeys().equals(joinKey)) {
+ keyRearrange =
+ ProjectedRow.from(
+ table.primaryKeys().stream()
+ .map(joinKey::indexOf)
+ .mapToInt(value -> value)
+ .toArray());
+ }
+ this.keyRearrange = keyRearrange;
}
@Override
@@ -86,7 +91,9 @@ public class PrimaryKeyPartialLookupTable implements LookupTable {
@Override
public List<InternalRow> get(InternalRow key) throws IOException {
- key = projectedKey.replaceRow(key);
+ if (keyRearrange != null) {
+ key = keyRearrange.replaceRow(key);
+ }
extractor.setRecord(key);
int bucket = extractor.bucket();
BinaryRow partition = extractor.partition();
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/SecondaryIndexLookupTable.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/SecondaryIndexLookupTable.java
index 793b99c5b..2f973bfd5 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/SecondaryIndexLookupTable.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/SecondaryIndexLookupTable.java
@@ -38,7 +38,7 @@ public class SecondaryIndexLookupTable extends PrimaryKeyLookupTable {
private final KeyProjectedRow secKeyRow;
public SecondaryIndexLookupTable(Context context, long lruCacheSize) throws IOException {
- super(context, lruCacheSize / 2);
+ super(context, lruCacheSize / 2, context.table.primaryKeys());
List<String> fieldNames = projectedType.getFieldNames();
int[] secKeyMapping = context.joinKey.stream().mapToInt(fieldNames::indexOf).toArray();
this.secKeyRow = new KeyProjectedRow(secKeyMapping);
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 30e34dcc3..3dff2e2a5 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
@@ -746,11 +746,13 @@ public class LookupJoinITCase extends CatalogITCaseBase {
iterator.close();
}
- @Test
- public void testPartialCacheBucketKeyOrder() throws Exception {
+ @ParameterizedTest
+ @EnumSource(LookupCacheMode.class)
+ public void testPartialCacheBucketKeyOrder(LookupCacheMode mode) throws Exception {
sql(
"CREATE TABLE DIM (k2 INT, k1 INT, j INT , i INT, PRIMARY KEY(i, j) NOT ENFORCED) WITH"
- + " ('continuous.discovery-interval'='1 ms', 'lookup.cache'='auto', 'bucket' = '2', 'bucket-key' = 'j')");
+ + " ('continuous.discovery-interval'='1 ms', 'lookup.cache'='%s', 'bucket' = '2', 'bucket-key' = 'j')",
+ mode);
sql("CREATE TABLE T2 (j INT, i INT, `proctime` AS PROCTIME())");