You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by lz...@apache.org on 2022/10/08 09:19:05 UTC
[flink-table-store] branch master updated: [FLINK-29292] Make result of MergeFunction generic
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/flink-table-store.git
The following commit(s) were added to refs/heads/master by this push:
new e4c19b66 [FLINK-29292] Make result of MergeFunction generic
e4c19b66 is described below
commit e4c19b66a8af319dd9f412aed250683d54969b34
Author: tsreaper <ts...@gmail.com>
AuthorDate: Sat Oct 8 17:19:00 2022 +0800
[FLINK-29292] Make result of MergeFunction generic
This closes #308
---
.../apache/flink/table/store/file/KeyValue.java | 5 --
.../flink/table/store/file/KeyValueFileStore.java | 4 +-
.../flink/table/store/file/mergetree/MemTable.java | 2 +-
.../store/file/mergetree/MergeTreeReader.java | 4 +-
.../store/file/mergetree/MergeTreeWriter.java | 4 +-
.../store/file/mergetree/SortBufferMemTable.java | 14 ++---
.../compact/DeduplicateMergeFunction.java | 15 +++---
.../file/mergetree/compact/MergeFunction.java | 7 ++-
.../mergetree/compact/MergeFunctionHelper.java | 9 ++--
.../compact/PartialUpdateMergeFunction.java | 20 +++++--
.../file/mergetree/compact/SortMergeReader.java | 10 ++--
.../mergetree/compact/ValueCountMergeFunction.java | 25 +++++++--
.../compact/aggregate/AggregateMergeFunction.java | 21 ++++++--
.../file/operation/KeyValueFileStoreRead.java | 6 +--
.../file/operation/KeyValueFileStoreWrite.java | 4 +-
.../table/ChangelogValueCountFileStoreTable.java | 2 +-
.../table/ChangelogWithKeyFileStoreTable.java | 4 +-
.../flink/table/store/file/TestFileStore.java | 4 +-
.../file/mergetree/SortBufferMemTableTestBase.java | 6 +--
.../compact/MergeFunctionHelperTestBase.java | 62 +++++++++++++++-------
.../mergetree/compact/SortMergeReaderTestBase.java | 6 +--
.../file/operation/KeyValueFileStoreReadTest.java | 5 +-
22 files changed, 148 insertions(+), 91 deletions(-)
diff --git a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/KeyValue.java b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/KeyValue.java
index 426b2696..5f124fe4 100644
--- a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/KeyValue.java
+++ b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/KeyValue.java
@@ -45,11 +45,6 @@ public class KeyValue {
private RowKind valueKind;
private RowData value;
- public KeyValue setValue(RowData value) {
- this.value = value;
- return this;
- }
-
public KeyValue replace(RowData key, RowKind valueKind, RowData value) {
return replace(key, UNKNOWN_SEQUENCE, valueKind, value);
}
diff --git a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/KeyValueFileStore.java b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/KeyValueFileStore.java
index 8d294aa0..20c1bb13 100644
--- a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/KeyValueFileStore.java
+++ b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/KeyValueFileStore.java
@@ -40,7 +40,7 @@ public class KeyValueFileStore extends AbstractFileStore<KeyValue> {
private final RowType keyType;
private final RowType valueType;
private final Supplier<Comparator<RowData>> keyComparatorSupplier;
- private final MergeFunction mergeFunction;
+ private final MergeFunction<KeyValue> mergeFunction;
public KeyValueFileStore(
SchemaManager schemaManager,
@@ -50,7 +50,7 @@ public class KeyValueFileStore extends AbstractFileStore<KeyValue> {
RowType bucketKeyType,
RowType keyType,
RowType valueType,
- MergeFunction mergeFunction) {
+ MergeFunction<KeyValue> mergeFunction) {
super(schemaManager, schemaId, options, partitionType);
this.bucketKeyType = bucketKeyType;
this.keyType = keyType;
diff --git a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/mergetree/MemTable.java b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/mergetree/MemTable.java
index 175166b4..25965454 100644
--- a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/mergetree/MemTable.java
+++ b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/mergetree/MemTable.java
@@ -56,7 +56,7 @@ public interface MemTable {
* MergeFunction}.
*/
Iterator<KeyValue> mergeIterator(
- Comparator<RowData> keyComparator, MergeFunction mergeFunction);
+ Comparator<RowData> keyComparator, MergeFunction<KeyValue> mergeFunction);
/** Removes all records from this table. The table will be empty after this call returns. */
void clear();
diff --git a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/mergetree/MergeTreeReader.java b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/mergetree/MergeTreeReader.java
index 11184b3c..67e65b06 100644
--- a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/mergetree/MergeTreeReader.java
+++ b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/mergetree/MergeTreeReader.java
@@ -48,7 +48,7 @@ public class MergeTreeReader implements RecordReader<KeyValue> {
boolean dropDelete,
KeyValueFileReaderFactory readerFactory,
Comparator<RowData> userKeyComparator,
- MergeFunction mergeFunction)
+ MergeFunction<KeyValue> mergeFunction)
throws IOException {
this.dropDelete = dropDelete;
@@ -113,7 +113,7 @@ public class MergeTreeReader implements RecordReader<KeyValue> {
List<SortedRun> section,
KeyValueFileReaderFactory readerFactory,
Comparator<RowData> userKeyComparator,
- MergeFunction mergeFunction)
+ MergeFunction<KeyValue> mergeFunction)
throws IOException {
List<RecordReader<KeyValue>> readers = new ArrayList<>();
for (SortedRun run : section) {
diff --git a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/mergetree/MergeTreeWriter.java b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/mergetree/MergeTreeWriter.java
index ecf89b70..1e168590 100644
--- a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/mergetree/MergeTreeWriter.java
+++ b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/mergetree/MergeTreeWriter.java
@@ -55,7 +55,7 @@ public class MergeTreeWriter implements RecordWriter<KeyValue>, MemoryOwner {
private final Comparator<RowData> keyComparator;
- private final MergeFunction mergeFunction;
+ private final MergeFunction<KeyValue> mergeFunction;
private final KeyValueFileWriterFactory writerFactory;
@@ -77,7 +77,7 @@ public class MergeTreeWriter implements RecordWriter<KeyValue>, MemoryOwner {
CompactManager compactManager,
long maxSequenceNumber,
Comparator<RowData> keyComparator,
- MergeFunction mergeFunction,
+ MergeFunction<KeyValue> mergeFunction,
KeyValueFileWriterFactory writerFactory,
boolean commitForceCompact,
ChangelogProducer changelogProducer) {
diff --git a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/mergetree/SortBufferMemTable.java b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/mergetree/SortBufferMemTable.java
index a6825565..1f472d00 100644
--- a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/mergetree/SortBufferMemTable.java
+++ b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/mergetree/SortBufferMemTable.java
@@ -106,7 +106,7 @@ public class SortBufferMemTable implements MemTable {
@Override
public Iterator<KeyValue> mergeIterator(
- Comparator<RowData> keyComparator, MergeFunction mergeFunction) {
+ Comparator<RowData> keyComparator, MergeFunction<KeyValue> mergeFunction) {
new QuickSort().sort(buffer);
return new MergeIterator(buffer.getIterator(), keyComparator, mergeFunction);
}
@@ -126,18 +126,20 @@ public class SortBufferMemTable implements MemTable {
private final Comparator<RowData> keyComparator;
private final MergeFunctionHelper mergeFunctionHelper;
- // holds the merged value
+ // previously read kv
private KeyValueSerializer previous;
private BinaryRowData previousRow;
// reads the next kv
private KeyValueSerializer current;
private BinaryRowData currentRow;
+
+ private KeyValue result;
private boolean advanced;
private MergeIterator(
MutableObjectIterator<BinaryRowData> kvIter,
Comparator<RowData> keyComparator,
- MergeFunction mergeFunction) {
+ MergeFunction<KeyValue> mergeFunction) {
this.kvIter = kvIter;
this.keyComparator = keyComparator;
this.mergeFunctionHelper = new MergeFunctionHelper(mergeFunction);
@@ -164,7 +166,7 @@ public class SortBufferMemTable implements MemTable {
return null;
}
advanced = false;
- return previous.getReusedKv();
+ return result;
}
private void advanceIfNeeded() {
@@ -173,7 +175,6 @@ public class SortBufferMemTable implements MemTable {
}
advanced = true;
- RowData result;
do {
swapSerializers();
if (previousRow == null) {
@@ -191,9 +192,8 @@ public class SortBufferMemTable implements MemTable {
mergeFunctionHelper.add(current.getReusedKv());
swapSerializers();
}
- result = mergeFunctionHelper.getValue();
+ result = mergeFunctionHelper.getResult();
} while (result == null);
- previous.getReusedKv().setValue(result);
}
private boolean readOnce() {
diff --git a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/mergetree/compact/DeduplicateMergeFunction.java b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/mergetree/compact/DeduplicateMergeFunction.java
index 12f3e050..bc22a190 100644
--- a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/mergetree/compact/DeduplicateMergeFunction.java
+++ b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/mergetree/compact/DeduplicateMergeFunction.java
@@ -18,7 +18,6 @@
package org.apache.flink.table.store.file.mergetree.compact;
-import org.apache.flink.table.data.RowData;
import org.apache.flink.table.store.file.KeyValue;
import javax.annotation.Nullable;
@@ -27,30 +26,30 @@ import javax.annotation.Nullable;
* A {@link MergeFunction} where key is primary key (unique) and value is the full record, only keep
* the latest one.
*/
-public class DeduplicateMergeFunction implements MergeFunction {
+public class DeduplicateMergeFunction implements MergeFunction<KeyValue> {
private static final long serialVersionUID = 1L;
- private RowData latestValue;
+ private transient KeyValue latestKv;
@Override
public void reset() {
- latestValue = null;
+ latestKv = null;
}
@Override
public void add(KeyValue kv) {
- latestValue = kv.value();
+ latestKv = kv;
}
@Override
@Nullable
- public RowData getValue() {
- return latestValue;
+ public KeyValue getResult() {
+ return latestKv;
}
@Override
- public MergeFunction copy() {
+ public MergeFunction<KeyValue> copy() {
return new DeduplicateMergeFunction();
}
}
diff --git a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/mergetree/compact/MergeFunction.java b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/mergetree/compact/MergeFunction.java
index 78d44716..125d2768 100644
--- a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/mergetree/compact/MergeFunction.java
+++ b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/mergetree/compact/MergeFunction.java
@@ -18,7 +18,6 @@
package org.apache.flink.table.store.file.mergetree.compact;
-import org.apache.flink.table.data.RowData;
import org.apache.flink.table.store.file.KeyValue;
import javax.annotation.Nullable;
@@ -26,7 +25,7 @@ import javax.annotation.Nullable;
import java.io.Serializable;
/** Merge function to merge multiple {@link KeyValue}s. */
-public interface MergeFunction extends Serializable {
+public interface MergeFunction<T> extends Serializable {
/** Reset the merge function to its default state. */
void reset();
@@ -36,8 +35,8 @@ public interface MergeFunction extends Serializable {
/** Get current merged value. Return null if this merged result should be skipped. */
@Nullable
- RowData getValue();
+ T getResult();
/** Create a new merge function object with the same functionality as this one. */
- MergeFunction copy();
+ MergeFunction<T> copy();
}
diff --git a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/mergetree/compact/MergeFunctionHelper.java b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/mergetree/compact/MergeFunctionHelper.java
index 0631351d..f82b8b2f 100644
--- a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/mergetree/compact/MergeFunctionHelper.java
+++ b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/mergetree/compact/MergeFunctionHelper.java
@@ -18,18 +18,17 @@
package org.apache.flink.table.store.file.mergetree.compact;
-import org.apache.flink.table.data.RowData;
import org.apache.flink.table.store.file.KeyValue;
/** Helper functions for the interaction with {@link MergeFunction}. */
public class MergeFunctionHelper {
- private final MergeFunction mergeFunction;
+ private final MergeFunction<KeyValue> mergeFunction;
private KeyValue initialKV;
private boolean isInitialized;
- public MergeFunctionHelper(MergeFunction mergeFunction) {
+ public MergeFunctionHelper(MergeFunction<KeyValue> mergeFunction) {
this.mergeFunction = mergeFunction;
}
@@ -58,7 +57,7 @@ public class MergeFunctionHelper {
}
/** Get current value of the {@link MergeFunction} helper. */
- public RowData getValue() {
- return isInitialized ? mergeFunction.getValue() : initialKV.value();
+ public KeyValue getResult() {
+ return isInitialized ? mergeFunction.getResult() : initialKV;
}
}
diff --git a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/mergetree/compact/PartialUpdateMergeFunction.java b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/mergetree/compact/PartialUpdateMergeFunction.java
index f93455cd..ed3a2bcc 100644
--- a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/mergetree/compact/PartialUpdateMergeFunction.java
+++ b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/mergetree/compact/PartialUpdateMergeFunction.java
@@ -26,18 +26,21 @@ import org.apache.flink.types.RowKind;
import javax.annotation.Nullable;
import static org.apache.flink.util.Preconditions.checkArgument;
+import static org.apache.flink.util.Preconditions.checkNotNull;
/**
* A {@link MergeFunction} where key is primary key (unique) and value is the partial record, update
* non-null fields on merge.
*/
-public class PartialUpdateMergeFunction implements MergeFunction {
+public class PartialUpdateMergeFunction implements MergeFunction<KeyValue> {
private static final long serialVersionUID = 1L;
private final RowData.FieldGetter[] getters;
+ private transient KeyValue latestKv;
private transient GenericRowData row;
+ private transient KeyValue reused;
public PartialUpdateMergeFunction(RowData.FieldGetter[] getters) {
this.getters = getters;
@@ -45,6 +48,7 @@ public class PartialUpdateMergeFunction implements MergeFunction {
@Override
public void reset() {
+ this.latestKv = null;
this.row = new GenericRowData(getters.length);
}
@@ -53,6 +57,7 @@ public class PartialUpdateMergeFunction implements MergeFunction {
checkArgument(
kv.valueKind() == RowKind.INSERT || kv.valueKind() == RowKind.UPDATE_AFTER,
"Partial update can not accept delete records. Partial delete is not supported!");
+ latestKv = kv;
for (int i = 0; i < getters.length; i++) {
Object field = getters[i].getFieldOrNull(kv.value());
if (field != null) {
@@ -63,12 +68,19 @@ public class PartialUpdateMergeFunction implements MergeFunction {
@Override
@Nullable
- public RowData getValue() {
- return row;
+ public KeyValue getResult() {
+ checkNotNull(
+ latestKv,
+ "Trying to get result from merge function without any input. This is unexpected.");
+
+ if (reused == null) {
+ reused = new KeyValue();
+ }
+ return reused.replace(latestKv.key(), latestKv.sequenceNumber(), RowKind.INSERT, row);
}
@Override
- public MergeFunction copy() {
+ public MergeFunction<KeyValue> copy() {
// RowData.FieldGetter is thread safe
return new PartialUpdateMergeFunction(getters);
}
diff --git a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/mergetree/compact/SortMergeReader.java b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/mergetree/compact/SortMergeReader.java
index 022974ad..d981e3c5 100644
--- a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/mergetree/compact/SortMergeReader.java
+++ b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/mergetree/compact/SortMergeReader.java
@@ -50,7 +50,7 @@ public class SortMergeReader implements RecordReader<KeyValue> {
protected SortMergeReader(
List<RecordReader<KeyValue>> readers,
Comparator<RowData> userKeyComparator,
- MergeFunction mergeFunction) {
+ MergeFunction<KeyValue> mergeFunction) {
this.nextBatchReaders = new ArrayList<>(readers);
this.userKeyComparator = userKeyComparator;
this.mergeFunctionHelper = new MergeFunctionHelper(mergeFunction);
@@ -70,7 +70,7 @@ public class SortMergeReader implements RecordReader<KeyValue> {
public static RecordReader<KeyValue> create(
List<RecordReader<KeyValue>> readers,
Comparator<RowData> userKeyComparator,
- MergeFunction mergeFunction) {
+ MergeFunction<KeyValue> mergeFunction) {
return readers.size() == 1
? readers.get(0)
: new SortMergeReader(readers, userKeyComparator, mergeFunction);
@@ -130,9 +130,9 @@ public class SortMergeReader implements RecordReader<KeyValue> {
if (!hasMore) {
return null;
}
- RowData mergedValue = mergeFunctionHelper.getValue();
- if (mergedValue != null) {
- return polled.get(polled.size() - 1).kv.setValue(mergedValue);
+ KeyValue result = mergeFunctionHelper.getResult();
+ if (result != null) {
+ return result;
}
}
}
diff --git a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/mergetree/compact/ValueCountMergeFunction.java b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/mergetree/compact/ValueCountMergeFunction.java
index b678a5e7..e4f6fb08 100644
--- a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/mergetree/compact/ValueCountMergeFunction.java
+++ b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/mergetree/compact/ValueCountMergeFunction.java
@@ -31,14 +31,17 @@ import static org.apache.flink.util.Preconditions.checkArgument;
* A {@link MergeFunction} where key is the full record and value is a count which represents number
* of records of the exact same fields.
*/
-public class ValueCountMergeFunction implements MergeFunction {
+public class ValueCountMergeFunction implements MergeFunction<KeyValue> {
private static final long serialVersionUID = 1L;
- private long total;
+ private transient KeyValue latestKv;
+ private transient long total;
+ private transient KeyValue reused;
@Override
public void reset() {
+ latestKv = null;
total = 0;
}
@@ -47,17 +50,29 @@ public class ValueCountMergeFunction implements MergeFunction {
checkArgument(
kv.valueKind() == RowKind.INSERT,
"In value count mode, only insert records come. This is a bug. Please file an issue.");
+ latestKv = kv;
total += count(kv.value());
}
@Override
@Nullable
- public RowData getValue() {
- return total == 0 ? null : GenericRowData.of(total);
+ public KeyValue getResult() {
+ if (total == 0) {
+ return null;
+ }
+
+ if (reused == null) {
+ reused = new KeyValue();
+ }
+ return reused.replace(
+ latestKv.key(),
+ latestKv.sequenceNumber(),
+ RowKind.INSERT,
+ GenericRowData.of(total));
}
@Override
- public MergeFunction copy() {
+ public MergeFunction<KeyValue> copy() {
return new ValueCountMergeFunction();
}
diff --git a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/mergetree/compact/aggregate/AggregateMergeFunction.java b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/mergetree/compact/aggregate/AggregateMergeFunction.java
index 3fbb4bbd..24729988 100644
--- a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/mergetree/compact/aggregate/AggregateMergeFunction.java
+++ b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/mergetree/compact/aggregate/AggregateMergeFunction.java
@@ -33,20 +33,22 @@ import java.util.List;
import static org.apache.flink.configuration.ConfigOptions.key;
import static org.apache.flink.util.Preconditions.checkArgument;
+import static org.apache.flink.util.Preconditions.checkNotNull;
/**
* A {@link MergeFunction} where key is primary key (unique) and value is the partial record,
* pre-aggregate non-null fields on merge.
*/
-public class AggregateMergeFunction implements MergeFunction {
+public class AggregateMergeFunction implements MergeFunction<KeyValue> {
private static final long serialVersionUID = 1L;
private final RowData.FieldGetter[] getters;
-
private final RowAggregator rowAggregator;
+ private transient KeyValue latestKv;
private transient GenericRowData row;
+ private transient KeyValue reused;
public AggregateMergeFunction(RowData.FieldGetter[] getters, RowAggregator rowAggregator) {
this.getters = getters;
@@ -55,6 +57,7 @@ public class AggregateMergeFunction implements MergeFunction {
@Override
public void reset() {
+ this.latestKv = null;
this.row = new GenericRowData(getters.length);
}
@@ -63,6 +66,7 @@ public class AggregateMergeFunction implements MergeFunction {
checkArgument(
kv.valueKind() == RowKind.INSERT || kv.valueKind() == RowKind.UPDATE_AFTER,
"Pre-aggregate can not accept delete records!");
+ latestKv = kv;
for (int i = 0; i < getters.length; i++) {
FieldAggregator fieldAggregator = rowAggregator.getFieldAggregatorAtPos(i);
Object accumulator = getters[i].getFieldOrNull(row);
@@ -74,12 +78,19 @@ public class AggregateMergeFunction implements MergeFunction {
@Nullable
@Override
- public RowData getValue() {
- return row;
+ public KeyValue getResult() {
+ checkNotNull(
+ latestKv,
+ "Trying to get result from merge function without any input. This is unexpected.");
+
+ if (reused == null) {
+ reused = new KeyValue();
+ }
+ return reused.replace(latestKv.key(), latestKv.sequenceNumber(), RowKind.INSERT, row);
}
@Override
- public MergeFunction copy() {
+ public MergeFunction<KeyValue> copy() {
// RowData.FieldGetter is thread safe
return new AggregateMergeFunction(getters, rowAggregator);
}
diff --git a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/operation/KeyValueFileStoreRead.java b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/operation/KeyValueFileStoreRead.java
index 4e4ca183..d07d69c2 100644
--- a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/operation/KeyValueFileStoreRead.java
+++ b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/operation/KeyValueFileStoreRead.java
@@ -60,7 +60,7 @@ public class KeyValueFileStoreRead implements FileStoreRead<KeyValue> {
private final TableSchema tableSchema;
private final KeyValueFileReaderFactory.Builder readerFactoryBuilder;
private final Comparator<RowData> keyComparator;
- private final MergeFunction mergeFunction;
+ private final MergeFunction<KeyValue> mergeFunction;
private final boolean valueCountMode;
@Nullable private int[][] keyProjectedFields;
@@ -75,7 +75,7 @@ public class KeyValueFileStoreRead implements FileStoreRead<KeyValue> {
RowType keyType,
RowType valueType,
Comparator<RowData> keyComparator,
- MergeFunction mergeFunction,
+ MergeFunction<KeyValue> mergeFunction,
FileFormat fileFormat,
FileStorePathFactory pathFactory) {
this.tableSchema = schemaManager.schema(schemaId);
@@ -153,7 +153,7 @@ public class KeyValueFileStoreRead implements FileStoreRead<KeyValue> {
private List<ConcatRecordReader.ReaderSupplier<KeyValue>> createSectionReaders(Split split) {
List<ConcatRecordReader.ReaderSupplier<KeyValue>> sectionReaders = new ArrayList<>();
- MergeFunction mergeFunc = mergeFunction.copy();
+ MergeFunction<KeyValue> mergeFunc = mergeFunction.copy();
for (List<SortedRun> section :
new IntervalPartition(split.files(), keyComparator).partition()) {
KeyValueFileReaderFactory readerFactory =
diff --git a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/operation/KeyValueFileStoreWrite.java b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/operation/KeyValueFileStoreWrite.java
index f18e3e5b..657a47b9 100644
--- a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/operation/KeyValueFileStoreWrite.java
+++ b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/operation/KeyValueFileStoreWrite.java
@@ -63,7 +63,7 @@ public class KeyValueFileStoreWrite extends AbstractFileStoreWrite<KeyValue> {
private final KeyValueFileReaderFactory.Builder readerFactoryBuilder;
private final KeyValueFileWriterFactory.Builder writerFactoryBuilder;
private final Supplier<Comparator<RowData>> keyComparatorSupplier;
- private final MergeFunction mergeFunction;
+ private final MergeFunction<KeyValue> mergeFunction;
private final CoreOptions options;
public KeyValueFileStoreWrite(
@@ -72,7 +72,7 @@ public class KeyValueFileStoreWrite extends AbstractFileStoreWrite<KeyValue> {
RowType keyType,
RowType valueType,
Supplier<Comparator<RowData>> keyComparatorSupplier,
- MergeFunction mergeFunction,
+ MergeFunction<KeyValue> mergeFunction,
FileStorePathFactory pathFactory,
SnapshotManager snapshotManager,
FileStoreScan scan,
diff --git a/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/ChangelogValueCountFileStoreTable.java b/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/ChangelogValueCountFileStoreTable.java
index e39216ed..e2a4e8c9 100644
--- a/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/ChangelogValueCountFileStoreTable.java
+++ b/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/ChangelogValueCountFileStoreTable.java
@@ -62,7 +62,7 @@ public class ChangelogValueCountFileStoreTable extends AbstractFileStoreTable {
RowType countType =
RowType.of(
new LogicalType[] {new BigIntType(false)}, new String[] {"_VALUE_COUNT"});
- MergeFunction mergeFunction = new ValueCountMergeFunction();
+ MergeFunction<KeyValue> mergeFunction = new ValueCountMergeFunction();
this.store =
new KeyValueFileStore(
schemaManager,
diff --git a/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/ChangelogWithKeyFileStoreTable.java b/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/ChangelogWithKeyFileStoreTable.java
index 02e457cd..18f44f75 100644
--- a/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/ChangelogWithKeyFileStoreTable.java
+++ b/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/ChangelogWithKeyFileStoreTable.java
@@ -70,6 +70,7 @@ public class ChangelogWithKeyFileStoreTable extends AbstractFileStoreTable {
ChangelogWithKeyFileStoreTable(
Path path, SchemaManager schemaManager, TableSchema tableSchema) {
super(path, tableSchema);
+
RowType rowType = tableSchema.logicalRowType();
Configuration conf = Configuration.fromMap(tableSchema.options());
CoreOptions.MergeEngine mergeEngine = conf.get(CoreOptions.MERGE_ENGINE);
@@ -78,7 +79,8 @@ public class ChangelogWithKeyFileStoreTable extends AbstractFileStoreTable {
for (int i = 0; i < fieldTypes.size(); i++) {
fieldGetters[i] = RowDataUtils.createNullCheckingFieldGetter(fieldTypes.get(i), i);
}
- MergeFunction mergeFunction;
+
+ MergeFunction<KeyValue> mergeFunction;
switch (mergeEngine) {
case DEDUPLICATE:
mergeFunction = new DeduplicateMergeFunction();
diff --git a/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/TestFileStore.java b/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/TestFileStore.java
index 157e16d2..d1f1a30a 100644
--- a/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/TestFileStore.java
+++ b/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/TestFileStore.java
@@ -92,7 +92,7 @@ public class TestFileStore extends KeyValueFileStore {
RowType partitionType,
RowType keyType,
RowType valueType,
- MergeFunction mergeFunction) {
+ MergeFunction<KeyValue> mergeFunction) {
Configuration conf = new Configuration();
conf.set(CoreOptions.WRITE_BUFFER_SIZE, WRITE_BUFFER_SIZE);
@@ -118,7 +118,7 @@ public class TestFileStore extends KeyValueFileStore {
RowType partitionType,
RowType keyType,
RowType valueType,
- MergeFunction mergeFunction) {
+ MergeFunction<KeyValue> mergeFunction) {
super(
new SchemaManager(options.path()),
0L,
diff --git a/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/mergetree/SortBufferMemTableTestBase.java b/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/mergetree/SortBufferMemTableTestBase.java
index 4469a8e2..8269bbbd 100644
--- a/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/mergetree/SortBufferMemTableTestBase.java
+++ b/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/mergetree/SortBufferMemTableTestBase.java
@@ -61,7 +61,7 @@ public abstract class SortBufferMemTableTestBase {
protected abstract List<ReusingTestData> getExpected(List<ReusingTestData> input);
- protected abstract MergeFunction createMergeFunction();
+ protected abstract MergeFunction<KeyValue> createMergeFunction();
@Test
public void testAndClear() throws IOException {
@@ -147,7 +147,7 @@ public abstract class SortBufferMemTableTestBase {
}
@Override
- protected MergeFunction createMergeFunction() {
+ protected MergeFunction<KeyValue> createMergeFunction() {
return new DeduplicateMergeFunction();
}
}
@@ -166,7 +166,7 @@ public abstract class SortBufferMemTableTestBase {
}
@Override
- protected MergeFunction createMergeFunction() {
+ protected MergeFunction<KeyValue> createMergeFunction() {
return new ValueCountMergeFunction();
}
diff --git a/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/mergetree/compact/MergeFunctionHelperTestBase.java b/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/mergetree/compact/MergeFunctionHelperTestBase.java
index cb689e10..9a2bdb81 100644
--- a/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/mergetree/compact/MergeFunctionHelperTestBase.java
+++ b/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/mergetree/compact/MergeFunctionHelperTestBase.java
@@ -19,7 +19,6 @@
package org.apache.flink.table.store.file.mergetree.compact;
import org.apache.flink.table.data.GenericRowData;
-import org.apache.flink.table.data.RowData;
import org.apache.flink.table.store.file.KeyValue;
import org.apache.flink.types.RowKind;
@@ -43,40 +42,53 @@ public abstract class MergeFunctionHelperTestBase {
protected MergeFunctionHelper mergeFunctionHelper;
- protected abstract MergeFunction createMergeFunction();
+ protected abstract MergeFunction<KeyValue> createMergeFunction();
- protected abstract RowData getExpected(List<RowData> rows);
+ protected abstract KeyValue getExpected(List<KeyValue> kvs);
@BeforeEach
void setUp() {
mergeFunctionHelper = new MergeFunctionHelper(createMergeFunction());
}
- @MethodSource("provideMergedRowData")
+ @MethodSource("provideMergedKeyValues")
@ParameterizedTest
- public void testMergeFunctionHelper(List<RowData> rows) {
- rows.forEach(r -> mergeFunctionHelper.add(new KeyValue().replace(null, RowKind.INSERT, r)));
- assertEquals(getExpected(rows), mergeFunctionHelper.getValue());
+ public void testMergeFunctionHelper(List<KeyValue> kvs) {
+ KeyValue expectedKv = getExpected(kvs);
+ kvs.forEach(kv -> mergeFunctionHelper.add(kv));
+ KeyValue mergedKv = mergeFunctionHelper.getResult();
+ assertEquals(expectedKv.key(), mergedKv.key());
+ assertEquals(expectedKv.sequenceNumber(), mergedKv.sequenceNumber());
+ assertEquals(expectedKv.valueKind(), mergedKv.valueKind());
+ assertEquals(expectedKv.value(), mergedKv.value());
}
- public static Stream<Arguments> provideMergedRowData() {
+ public static Stream<Arguments> provideMergedKeyValues() {
return Stream.of(
- Arguments.of(Collections.singletonList(row(1))),
- Arguments.of(Arrays.asList(row(-1), row(1))),
- Arguments.of(Arrays.asList(row(1), row(2))));
+ Arguments.of(
+ Collections.singletonList(
+ new KeyValue().replace(row(1), 1, RowKind.INSERT, row(1)))),
+ Arguments.of(
+ Arrays.asList(
+ new KeyValue().replace(row(2), 2, RowKind.INSERT, row(-1)),
+ new KeyValue().replace(row(2), 3, RowKind.INSERT, row(1)))),
+ Arguments.of(
+ Arrays.asList(
+ new KeyValue().replace(row(3), 4, RowKind.INSERT, row(1)),
+ new KeyValue().replace(row(3), 5, RowKind.INSERT, row(2)))));
}
/** Tests for {@link MergeFunctionHelper} with {@link DeduplicateMergeFunction}. */
public static class WithDeduplicateMergeFunctionTest extends MergeFunctionHelperTestBase {
@Override
- protected MergeFunction createMergeFunction() {
+ protected MergeFunction<KeyValue> createMergeFunction() {
return new DeduplicateMergeFunction();
}
@Override
- protected RowData getExpected(List<RowData> rows) {
- return rows.get(rows.size() - 1);
+ protected KeyValue getExpected(List<KeyValue> kvs) {
+ return kvs.get(kvs.size() - 1);
}
}
@@ -84,17 +96,27 @@ public abstract class MergeFunctionHelperTestBase {
public static class WithValueRecordMergeFunctionTest extends MergeFunctionHelperTestBase {
@Override
- protected MergeFunction createMergeFunction() {
+ protected MergeFunction<KeyValue> createMergeFunction() {
return new ValueCountMergeFunction();
}
@Override
- protected RowData getExpected(List<RowData> rows) {
- if (rows.size() == 1) {
- return rows.get(0);
+ protected KeyValue getExpected(List<KeyValue> kvs) {
+ if (kvs.size() == 1) {
+ return kvs.get(0);
} else {
- long total = rows.stream().mapToLong(r -> r.getLong(0)).sum();
- return total == 0 ? null : GenericRowData.of(total);
+ long total = kvs.stream().mapToLong(kv -> kv.value().getLong(0)).sum();
+ if (total == 0) {
+ return null;
+ } else {
+ KeyValue result = kvs.get(kvs.size() - 1);
+ return new KeyValue()
+ .replace(
+ result.key(),
+ result.sequenceNumber(),
+ RowKind.INSERT,
+ GenericRowData.of(total));
+ }
}
}
diff --git a/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/mergetree/compact/SortMergeReaderTestBase.java b/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/mergetree/compact/SortMergeReaderTestBase.java
index 19895bcf..800d927e 100644
--- a/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/mergetree/compact/SortMergeReaderTestBase.java
+++ b/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/mergetree/compact/SortMergeReaderTestBase.java
@@ -32,7 +32,7 @@ import java.util.List;
/** Tests for {@link SortMergeReader}. */
public abstract class SortMergeReaderTestBase extends CombiningRecordReaderTestBase {
- protected abstract MergeFunction createMergeFunction();
+ protected abstract MergeFunction<KeyValue> createMergeFunction();
@Override
protected RecordReader<KeyValue> createRecordReader(List<TestReusingRecordReader> readers) {
@@ -85,7 +85,7 @@ public abstract class SortMergeReaderTestBase extends CombiningRecordReaderTestB
}
@Override
- protected MergeFunction createMergeFunction() {
+ protected MergeFunction<KeyValue> createMergeFunction() {
return new DeduplicateMergeFunction();
}
}
@@ -104,7 +104,7 @@ public abstract class SortMergeReaderTestBase extends CombiningRecordReaderTestB
}
@Override
- protected MergeFunction createMergeFunction() {
+ protected MergeFunction<KeyValue> createMergeFunction() {
return new ValueCountMergeFunction();
}
diff --git a/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/operation/KeyValueFileStoreReadTest.java b/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/operation/KeyValueFileStoreReadTest.java
index a48bbdc4..f110e2e2 100644
--- a/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/operation/KeyValueFileStoreReadTest.java
+++ b/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/operation/KeyValueFileStoreReadTest.java
@@ -227,7 +227,10 @@ public class KeyValueFileStoreReadTest {
}
private TestFileStore createStore(
- RowType partitionType, RowType keyType, RowType valueType, MergeFunction mergeFunction)
+ RowType partitionType,
+ RowType keyType,
+ RowType valueType,
+ MergeFunction<KeyValue> mergeFunction)
throws Exception {
SchemaManager schemaManager = new SchemaManager(new Path(tempDir.toUri()));
boolean valueCountMode = mergeFunction instanceof ValueCountMergeFunction;