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 2023/06/08 09:30:49 UTC
[incubator-paimon] branch master updated: [core] Extract RowKeyExtractor (#1340)
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
The following commit(s) were added to refs/heads/master by this push:
new 66816c1a3 [core] Extract RowKeyExtractor (#1340)
66816c1a3 is described below
commit 66816c1a364012793cc6eb2624e62083619e3413
Author: Jingsong Lee <ji...@gmail.com>
AuthorDate: Thu Jun 8 17:30:44 2023 +0800
[core] Extract RowKeyExtractor (#1340)
---
.../paimon/table/AbstractFileStoreTable.java | 13 ++++
.../paimon/table/AppendOnlyFileStoreTable.java | 3 +-
.../table/ChangelogValueCountFileStoreTable.java | 3 +-
.../table/ChangelogWithKeyFileStoreTable.java | 3 +-
.../table/sink/FixedBucketRowKeyExtractor.java | 78 ++++++++++++++++++++++
...ndBucketExtractor.java => RowKeyExtractor.java} | 42 +-----------
...st.java => FixedBucketRowKeyExtractorTest.java} | 25 +++++--
.../flink/sink/RowDataKeyAndBucketExtractor.java | 6 +-
8 files changed, 118 insertions(+), 55 deletions(-)
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/AbstractFileStoreTable.java b/paimon-core/src/main/java/org/apache/paimon/table/AbstractFileStoreTable.java
index 683257dc2..cbb81815b 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/AbstractFileStoreTable.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/AbstractFileStoreTable.java
@@ -29,6 +29,8 @@ import org.apache.paimon.predicate.Predicate;
import org.apache.paimon.schema.SchemaManager;
import org.apache.paimon.schema.SchemaValidation;
import org.apache.paimon.schema.TableSchema;
+import org.apache.paimon.table.sink.FixedBucketRowKeyExtractor;
+import org.apache.paimon.table.sink.RowKeyExtractor;
import org.apache.paimon.table.sink.TableCommitImpl;
import org.apache.paimon.table.source.InnerStreamTableScan;
import org.apache.paimon.table.source.InnerStreamTableScanImpl;
@@ -80,6 +82,17 @@ public abstract class AbstractFileStoreTable implements FileStoreTable {
return store().bucketMode();
}
+ public RowKeyExtractor createRowKeyExtractor() {
+ switch (bucketMode()) {
+ case FIXED:
+ return new FixedBucketRowKeyExtractor(schema());
+ case DYNAMIC:
+ case UNAWARE:
+ default:
+ throw new UnsupportedOperationException("Unsupported mode: " + bucketMode());
+ }
+ }
+
@Override
public SnapshotSplitReader newSnapshotSplitReader() {
return new SnapshotSplitReaderImpl(
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/AppendOnlyFileStoreTable.java b/paimon-core/src/main/java/org/apache/paimon/table/AppendOnlyFileStoreTable.java
index c9f3b6006..6c1110cae 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/AppendOnlyFileStoreTable.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/AppendOnlyFileStoreTable.java
@@ -32,7 +32,6 @@ import org.apache.paimon.operation.ReverseReader;
import org.apache.paimon.predicate.Predicate;
import org.apache.paimon.reader.RecordReader;
import org.apache.paimon.schema.TableSchema;
-import org.apache.paimon.table.sink.InternalRowKeyAndBucketExtractor;
import org.apache.paimon.table.sink.TableWriteImpl;
import org.apache.paimon.table.source.AppendOnlySplitGenerator;
import org.apache.paimon.table.source.DataSplit;
@@ -132,7 +131,7 @@ public class AppendOnlyFileStoreTable extends AbstractFileStoreTable {
String commitUser, ManifestCacheFilter manifestFilter) {
return new TableWriteImpl<>(
store().newWrite(commitUser, manifestFilter),
- new InternalRowKeyAndBucketExtractor(tableSchema),
+ createRowKeyExtractor(),
record -> {
Preconditions.checkState(
record.row().getRowKind() == RowKind.INSERT,
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/ChangelogValueCountFileStoreTable.java b/paimon-core/src/main/java/org/apache/paimon/table/ChangelogValueCountFileStoreTable.java
index b91397348..5f74ca7fa 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/ChangelogValueCountFileStoreTable.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/ChangelogValueCountFileStoreTable.java
@@ -37,7 +37,6 @@ import org.apache.paimon.reader.RecordReader;
import org.apache.paimon.schema.KeyValueFieldsExtractor;
import org.apache.paimon.schema.TableSchema;
import org.apache.paimon.stats.BinaryTableStats;
-import org.apache.paimon.table.sink.InternalRowKeyAndBucketExtractor;
import org.apache.paimon.table.sink.TableWriteImpl;
import org.apache.paimon.table.source.InnerTableRead;
import org.apache.paimon.table.source.KeyValueTableRead;
@@ -151,7 +150,7 @@ public class ChangelogValueCountFileStoreTable extends AbstractFileStoreTable {
final KeyValue kv = new KeyValue();
return new TableWriteImpl<>(
store().newWrite(commitUser, manifestFilter),
- new InternalRowKeyAndBucketExtractor(tableSchema),
+ createRowKeyExtractor(),
record -> {
switch (record.row().getRowKind()) {
case INSERT:
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/ChangelogWithKeyFileStoreTable.java b/paimon-core/src/main/java/org/apache/paimon/table/ChangelogWithKeyFileStoreTable.java
index 2ff5903f4..03a68e6f8 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/ChangelogWithKeyFileStoreTable.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/ChangelogWithKeyFileStoreTable.java
@@ -39,7 +39,6 @@ import org.apache.paimon.predicate.Predicate;
import org.apache.paimon.reader.RecordReader;
import org.apache.paimon.schema.KeyValueFieldsExtractor;
import org.apache.paimon.schema.TableSchema;
-import org.apache.paimon.table.sink.InternalRowKeyAndBucketExtractor;
import org.apache.paimon.table.sink.SequenceGenerator;
import org.apache.paimon.table.sink.TableWriteImpl;
import org.apache.paimon.table.source.InnerTableRead;
@@ -226,7 +225,7 @@ public class ChangelogWithKeyFileStoreTable extends AbstractFileStoreTable {
final KeyValue kv = new KeyValue();
return new TableWriteImpl<>(
store().newWrite(commitUser, manifestFilter),
- new InternalRowKeyAndBucketExtractor(tableSchema),
+ createRowKeyExtractor(),
record -> {
long sequenceNumber =
sequenceGenerator == null
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/sink/FixedBucketRowKeyExtractor.java b/paimon-core/src/main/java/org/apache/paimon/table/sink/FixedBucketRowKeyExtractor.java
new file mode 100644
index 000000000..62b5dc215
--- /dev/null
+++ b/paimon-core/src/main/java/org/apache/paimon/table/sink/FixedBucketRowKeyExtractor.java
@@ -0,0 +1,78 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.table.sink;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.codegen.CodeGenUtils;
+import org.apache.paimon.codegen.Projection;
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.schema.TableSchema;
+
+import static org.apache.paimon.utils.Preconditions.checkArgument;
+
+/** {@link KeyAndBucketExtractor} for {@link InternalRow}. */
+public class FixedBucketRowKeyExtractor extends RowKeyExtractor {
+
+ private final int numBuckets;
+ private final boolean sameBucketKeyAndTrimmedPrimaryKey;
+ private final Projection bucketKeyProjection;
+
+ private BinaryRow reuseBucketKey;
+ private Integer reuseBucket;
+
+ public FixedBucketRowKeyExtractor(TableSchema schema) {
+ super(schema);
+ numBuckets = new CoreOptions(schema.options()).bucket();
+ checkArgument(numBuckets > 0, "Num bucket is illegal: " + numBuckets);
+ sameBucketKeyAndTrimmedPrimaryKey = schema.bucketKeys().equals(schema.trimmedPrimaryKeys());
+ bucketKeyProjection =
+ CodeGenUtils.newProjection(
+ schema.logicalRowType(), schema.projection(schema.bucketKeys()));
+ }
+
+ @Override
+ public void setRecord(InternalRow record) {
+ super.setRecord(record);
+ this.reuseBucketKey = null;
+ this.reuseBucket = null;
+ }
+
+ private BinaryRow bucketKey() {
+ if (sameBucketKeyAndTrimmedPrimaryKey) {
+ return trimmedPrimaryKey();
+ }
+
+ if (reuseBucketKey == null) {
+ reuseBucketKey = bucketKeyProjection.apply(record);
+ }
+ return reuseBucketKey;
+ }
+
+ @Override
+ public int bucket() {
+ BinaryRow bucketKey = bucketKey();
+ if (reuseBucket == null) {
+ reuseBucket =
+ KeyAndBucketExtractor.bucket(
+ KeyAndBucketExtractor.bucketKeyHashCode(bucketKey), numBuckets);
+ }
+ return reuseBucket;
+ }
+}
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/sink/InternalRowKeyAndBucketExtractor.java b/paimon-core/src/main/java/org/apache/paimon/table/sink/RowKeyExtractor.java
similarity index 64%
rename from paimon-core/src/main/java/org/apache/paimon/table/sink/InternalRowKeyAndBucketExtractor.java
rename to paimon-core/src/main/java/org/apache/paimon/table/sink/RowKeyExtractor.java
index 1c0942540..d25026c9e 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/sink/InternalRowKeyAndBucketExtractor.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/sink/RowKeyExtractor.java
@@ -18,7 +18,6 @@
package org.apache.paimon.table.sink;
-import org.apache.paimon.CoreOptions;
import org.apache.paimon.codegen.CodeGenUtils;
import org.apache.paimon.codegen.Projection;
import org.apache.paimon.data.BinaryRow;
@@ -26,34 +25,22 @@ import org.apache.paimon.data.InternalRow;
import org.apache.paimon.schema.TableSchema;
/** {@link KeyAndBucketExtractor} for {@link InternalRow}. */
-public class InternalRowKeyAndBucketExtractor implements KeyAndBucketExtractor<InternalRow> {
-
- private final int numBuckets;
- private final boolean sameBucketKeyAndTrimmedPrimaryKey;
+public abstract class RowKeyExtractor implements KeyAndBucketExtractor<InternalRow> {
private final Projection partitionProjection;
- private final Projection bucketKeyProjection;
private final Projection trimmedPrimaryKeyProjection;
private final Projection logPrimaryKeyProjection;
- private InternalRow record;
+ protected InternalRow record;
private BinaryRow partition;
- private BinaryRow bucketKey;
- private Integer bucket;
private BinaryRow trimmedPrimaryKey;
private BinaryRow logPrimaryKey;
- public InternalRowKeyAndBucketExtractor(TableSchema schema) {
- numBuckets = new CoreOptions(schema.options()).bucket();
- sameBucketKeyAndTrimmedPrimaryKey = schema.bucketKeys().equals(schema.trimmedPrimaryKeys());
-
+ public RowKeyExtractor(TableSchema schema) {
partitionProjection =
CodeGenUtils.newProjection(
schema.logicalRowType(), schema.projection(schema.partitionKeys()));
- bucketKeyProjection =
- CodeGenUtils.newProjection(
- schema.logicalRowType(), schema.projection(schema.bucketKeys()));
trimmedPrimaryKeyProjection =
CodeGenUtils.newProjection(
schema.logicalRowType(), schema.projection(schema.trimmedPrimaryKeys()));
@@ -65,10 +52,7 @@ public class InternalRowKeyAndBucketExtractor implements KeyAndBucketExtractor<I
@Override
public void setRecord(InternalRow record) {
this.record = record;
-
this.partition = null;
- this.bucketKey = null;
- this.bucket = null;
this.trimmedPrimaryKey = null;
this.logPrimaryKey = null;
}
@@ -81,29 +65,10 @@ public class InternalRowKeyAndBucketExtractor implements KeyAndBucketExtractor<I
return partition;
}
- @Override
- public int bucket() {
- if (bucketKey == null) {
- bucketKey = bucketKeyProjection.apply(record);
- if (sameBucketKeyAndTrimmedPrimaryKey) {
- trimmedPrimaryKey = bucketKey;
- }
- }
- if (bucket == null) {
- bucket =
- KeyAndBucketExtractor.bucket(
- KeyAndBucketExtractor.bucketKeyHashCode(bucketKey), numBuckets);
- }
- return bucket;
- }
-
@Override
public BinaryRow trimmedPrimaryKey() {
if (trimmedPrimaryKey == null) {
trimmedPrimaryKey = trimmedPrimaryKeyProjection.apply(record);
- if (sameBucketKeyAndTrimmedPrimaryKey) {
- bucketKey = trimmedPrimaryKey;
- }
}
return trimmedPrimaryKey;
}
@@ -111,7 +76,6 @@ public class InternalRowKeyAndBucketExtractor implements KeyAndBucketExtractor<I
@Override
public BinaryRow logPrimaryKey() {
if (logPrimaryKey == null) {
- assert logPrimaryKeyProjection != null;
logPrimaryKey = logPrimaryKeyProjection.apply(record);
}
return logPrimaryKey;
diff --git a/paimon-core/src/test/java/org/apache/paimon/table/sink/InternalRowKeyAndBucketExtractorTest.java b/paimon-core/src/test/java/org/apache/paimon/table/sink/FixedBucketRowKeyExtractorTest.java
similarity index 82%
rename from paimon-core/src/test/java/org/apache/paimon/table/sink/InternalRowKeyAndBucketExtractorTest.java
rename to paimon-core/src/test/java/org/apache/paimon/table/sink/FixedBucketRowKeyExtractorTest.java
index a36ffb9bd..e0e40708b 100644
--- a/paimon-core/src/test/java/org/apache/paimon/table/sink/InternalRowKeyAndBucketExtractorTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/table/sink/FixedBucketRowKeyExtractorTest.java
@@ -38,8 +38,8 @@ import static org.apache.paimon.CoreOptions.BUCKET_KEY;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.assertj.core.api.AssertionsForClassTypes.assertThat;
-/** Test for {@link InternalRowKeyAndBucketExtractor}. */
-public class InternalRowKeyAndBucketExtractorTest {
+/** Test for {@link FixedBucketRowKeyExtractor}. */
+public class FixedBucketRowKeyExtractorTest {
@Test
public void testInvalidBucket() {
@@ -65,16 +65,27 @@ public class InternalRowKeyAndBucketExtractorTest {
assertThat(bucket(extractor("", "a,b,c"), row)).isEqualTo(40);
}
- private int bucket(InternalRowKeyAndBucketExtractor extractor, InternalRow row) {
+ @Test
+ public void testIllegalBucket() {
+ assertThatThrownBy(() -> extractor("", "", "a", -1))
+ .hasMessageContaining("Num bucket is illegal");
+ }
+
+ private int bucket(FixedBucketRowKeyExtractor extractor, InternalRow row) {
extractor.setRecord(row);
return extractor.bucket();
}
- private InternalRowKeyAndBucketExtractor extractor(String bk, String pk) {
+ private FixedBucketRowKeyExtractor extractor(String bk, String pk) {
return extractor("", bk, pk);
}
- private InternalRowKeyAndBucketExtractor extractor(String partK, String bk, String pk) {
+ private FixedBucketRowKeyExtractor extractor(String partK, String bk, String pk) {
+ return extractor(partK, bk, pk, 100);
+ }
+
+ private FixedBucketRowKeyExtractor extractor(
+ String partK, String bk, String pk, int numBucket) {
RowType rowType =
new RowType(
Arrays.asList(
@@ -84,7 +95,7 @@ public class InternalRowKeyAndBucketExtractorTest {
List<DataField> fields = TableSchema.newFields(rowType);
Map<String, String> options = new HashMap<>();
options.put(BUCKET_KEY.key(), bk);
- options.put(BUCKET.key(), "100");
+ options.put(BUCKET.key(), String.valueOf(numBucket));
TableSchema schema =
new TableSchema(
0,
@@ -96,6 +107,6 @@ public class InternalRowKeyAndBucketExtractorTest {
"".equals(pk) ? Collections.emptyList() : Arrays.asList(pk.split(",")),
options,
"");
- return new InternalRowKeyAndBucketExtractor(schema);
+ return new FixedBucketRowKeyExtractor(schema);
}
}
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/RowDataKeyAndBucketExtractor.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/RowDataKeyAndBucketExtractor.java
index 8eec1c6e8..ae4d6e9eb 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/RowDataKeyAndBucketExtractor.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/RowDataKeyAndBucketExtractor.java
@@ -21,7 +21,7 @@ package org.apache.paimon.flink.sink;
import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.flink.FlinkRowWrapper;
import org.apache.paimon.schema.TableSchema;
-import org.apache.paimon.table.sink.InternalRowKeyAndBucketExtractor;
+import org.apache.paimon.table.sink.FixedBucketRowKeyExtractor;
import org.apache.paimon.table.sink.KeyAndBucketExtractor;
import org.apache.flink.table.data.RowData;
@@ -29,10 +29,10 @@ import org.apache.flink.table.data.RowData;
/** {@link KeyAndBucketExtractor} for {@link RowData}. */
public class RowDataKeyAndBucketExtractor implements KeyAndBucketExtractor<RowData> {
- private final InternalRowKeyAndBucketExtractor wrapped;
+ private final FixedBucketRowKeyExtractor wrapped;
public RowDataKeyAndBucketExtractor(TableSchema schema) {
- wrapped = new InternalRowKeyAndBucketExtractor(schema);
+ wrapped = new FixedBucketRowKeyExtractor(schema);
}
@Override