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/10/27 03:30:32 UTC
[incubator-paimon] branch master updated: [core] Refactor code to introduce BinaryExternalSortBuffer.create (#2181)
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 93a38dcdd [core] Refactor code to introduce BinaryExternalSortBuffer.create (#2181)
93a38dcdd is described below
commit 93a38dcdd196d3efcbe02320e7b5d67950a3b9c7
Author: Jingsong Lee <ji...@gmail.com>
AuthorDate: Fri Oct 27 11:30:28 2023 +0800
[core] Refactor code to introduce BinaryExternalSortBuffer.create (#2181)
---
.../org/apache/paimon/mergetree/MergeSorter.java | 35 +++++-----------
.../paimon/sort/BinaryExternalSortBuffer.java | 46 ++++++++++++++++++++++
.../flink/sink/index/GlobalIndexAssigner.java | 31 +++------------
.../apache/paimon/flink/sorter/SortOperator.java | 31 +--------------
4 files changed, 63 insertions(+), 80 deletions(-)
diff --git a/paimon-core/src/main/java/org/apache/paimon/mergetree/MergeSorter.java b/paimon-core/src/main/java/org/apache/paimon/mergetree/MergeSorter.java
index a63c1d21c..15f6cfb9d 100644
--- a/paimon-core/src/main/java/org/apache/paimon/mergetree/MergeSorter.java
+++ b/paimon-core/src/main/java/org/apache/paimon/mergetree/MergeSorter.java
@@ -21,16 +21,10 @@ package org.apache.paimon.mergetree;
import org.apache.paimon.CoreOptions;
import org.apache.paimon.CoreOptions.SortEngine;
import org.apache.paimon.KeyValue;
-import org.apache.paimon.codegen.CodeGenUtils;
-import org.apache.paimon.codegen.NormalizedKeyComputer;
-import org.apache.paimon.codegen.RecordComparator;
import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.data.GenericRow;
import org.apache.paimon.data.InternalRow;
import org.apache.paimon.data.JoinedRow;
-import org.apache.paimon.data.serializer.BinaryRowSerializer;
-import org.apache.paimon.data.serializer.InternalRowSerializer;
-import org.apache.paimon.data.serializer.InternalSerializers;
import org.apache.paimon.disk.IOManager;
import org.apache.paimon.memory.CachelessSegmentPool;
import org.apache.paimon.memory.MemorySegmentPool;
@@ -40,11 +34,11 @@ import org.apache.paimon.mergetree.compact.MergeFunctionWrapper;
import org.apache.paimon.mergetree.compact.SortMergeReader;
import org.apache.paimon.reader.RecordReader;
import org.apache.paimon.sort.BinaryExternalSortBuffer;
-import org.apache.paimon.sort.BinaryInMemorySortBuffer;
import org.apache.paimon.sort.SortBuffer;
import org.apache.paimon.types.BigIntType;
import org.apache.paimon.types.DataField;
import org.apache.paimon.types.DataType;
+import org.apache.paimon.types.DataTypes;
import org.apache.paimon.types.IntType;
import org.apache.paimon.types.RowKind;
import org.apache.paimon.types.RowType;
@@ -181,37 +175,28 @@ public class MergeSorter {
private final SortBuffer buffer;
public ExternalSorterWithLevel() {
- // user key + sequenceNumber
- List<DataType> sortKeyTypes = new ArrayList<>(keyType.getFieldTypes());
- sortKeyTypes.add(new BigIntType(false));
-
- // for sort binary buffer
- NormalizedKeyComputer normalizedKeyComputer =
- CodeGenUtils.newNormalizedKeyComputer(sortKeyTypes, "MemTableKeyComputer");
- RecordComparator keyComparator =
- CodeGenUtils.newRecordComparator(sortKeyTypes, "MemTableComparator");
-
if (memoryPool.freePages() < 3) {
throw new IllegalArgumentException(
"Write buffer requires a minimum of 3 page memory, please increase write buffer memory size.");
}
+ // user key + sequenceNumber
+ List<DataType> sortKeyTypes = new ArrayList<>(keyType.getFieldTypes());
+ sortKeyTypes.add(new BigIntType(false));
+
+ // row type
List<DataField> fields = new ArrayList<>(keyType.getFields());
fields.add(new DataField(0, SEQUENCE_NUMBER, new BigIntType(false)));
fields.add(new DataField(1, VALUE_KIND, new TinyIntType(false)));
fields.add(new DataField(2, "_LEVEL", new IntType(false)));
fields.addAll(valueType.getFields());
- RowType schemaWithLevel = new RowType(fields);
- InternalRowSerializer serializer = InternalSerializers.create(schemaWithLevel);
this.buffer =
- new BinaryExternalSortBuffer(
- new BinaryRowSerializer(serializer.getArity()),
- keyComparator,
- memoryPool.pageSize(),
- BinaryInMemorySortBuffer.createBuffer(
- normalizedKeyComputer, serializer, keyComparator, memoryPool),
+ BinaryExternalSortBuffer.create(
ioManager,
+ DataTypes.ROW(sortKeyTypes.toArray(new DataType[0])),
+ new RowType(fields),
+ memoryPool,
spillSortMaxNumFiles);
}
diff --git a/paimon-core/src/main/java/org/apache/paimon/sort/BinaryExternalSortBuffer.java b/paimon-core/src/main/java/org/apache/paimon/sort/BinaryExternalSortBuffer.java
index a7593930f..4bfbbdb44 100644
--- a/paimon-core/src/main/java/org/apache/paimon/sort/BinaryExternalSortBuffer.java
+++ b/paimon-core/src/main/java/org/apache/paimon/sort/BinaryExternalSortBuffer.java
@@ -25,18 +25,25 @@ import org.apache.paimon.compression.Lz4BlockCompressionFactory;
import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.data.InternalRow;
import org.apache.paimon.data.serializer.BinaryRowSerializer;
+import org.apache.paimon.data.serializer.InternalRowSerializer;
import org.apache.paimon.disk.ChannelWithMeta;
import org.apache.paimon.disk.ChannelWriterOutputView;
import org.apache.paimon.disk.FileChannelUtil;
import org.apache.paimon.disk.FileIOChannel;
import org.apache.paimon.disk.IOManager;
+import org.apache.paimon.memory.HeapMemorySegmentPool;
+import org.apache.paimon.memory.MemorySegmentPool;
import org.apache.paimon.options.MemorySize;
+import org.apache.paimon.types.RowType;
import org.apache.paimon.utils.MutableObjectIterator;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
+import static org.apache.paimon.codegen.CodeGenUtils.newNormalizedKeyComputer;
+import static org.apache.paimon.codegen.CodeGenUtils.newRecordComparator;
+
/** A spillable {@link SortBuffer}. */
public class BinaryExternalSortBuffer implements SortBuffer {
@@ -82,6 +89,45 @@ public class BinaryExternalSortBuffer implements SortBuffer {
this.spillChannelIDs = new ArrayList<>();
}
+ public static BinaryExternalSortBuffer create(
+ IOManager ioManager,
+ RowType keyType,
+ RowType rowType,
+ long bufferSize,
+ int pageSize,
+ int maxNumFileHandles) {
+ return create(
+ ioManager,
+ keyType,
+ rowType,
+ new HeapMemorySegmentPool(bufferSize, pageSize),
+ maxNumFileHandles);
+ }
+
+ public static BinaryExternalSortBuffer create(
+ IOManager ioManager,
+ RowType keyType,
+ RowType rowType,
+ MemorySegmentPool pool,
+ int maxNumFileHandles) {
+ RecordComparator comparator =
+ newRecordComparator(keyType.getFieldTypes(), "ExternalSort_comparator");
+ BinaryInMemorySortBuffer sortBuffer =
+ BinaryInMemorySortBuffer.createBuffer(
+ newNormalizedKeyComputer(
+ keyType.getFieldTypes(), "ExternalSort_normalized_key"),
+ new InternalRowSerializer(rowType),
+ comparator,
+ pool);
+ return new BinaryExternalSortBuffer(
+ new BinaryRowSerializer(rowType.getFieldCount()),
+ comparator,
+ pool.pageSize(),
+ sortBuffer,
+ ioManager,
+ maxNumFileHandles);
+ }
+
@Override
public int size() {
return numRecords;
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/index/GlobalIndexAssigner.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/index/GlobalIndexAssigner.java
index 545b0cbd3..6618ddc57 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/index/GlobalIndexAssigner.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/index/GlobalIndexAssigner.java
@@ -20,22 +20,17 @@ package org.apache.paimon.flink.sink.index;
import org.apache.paimon.CoreOptions;
import org.apache.paimon.CoreOptions.MergeEngine;
-import org.apache.paimon.codegen.RecordComparator;
import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.data.GenericRow;
import org.apache.paimon.data.InternalRow;
-import org.apache.paimon.data.serializer.BinaryRowSerializer;
-import org.apache.paimon.data.serializer.InternalRowSerializer;
import org.apache.paimon.data.serializer.RowCompactedSerializer;
import org.apache.paimon.disk.IOManager;
import org.apache.paimon.flink.RocksDBOptions;
import org.apache.paimon.flink.lookup.RocksDBStateFactory;
import org.apache.paimon.flink.lookup.RocksDBValueState;
-import org.apache.paimon.memory.HeapMemorySegmentPool;
import org.apache.paimon.options.Options;
import org.apache.paimon.schema.TableSchema;
import org.apache.paimon.sort.BinaryExternalSortBuffer;
-import org.apache.paimon.sort.BinaryInMemorySortBuffer;
import org.apache.paimon.table.AbstractFileStoreTable;
import org.apache.paimon.table.Table;
import org.apache.paimon.table.sink.PartitionKeyExtractor;
@@ -59,15 +54,12 @@ import java.io.File;
import java.io.IOException;
import java.io.Serializable;
import java.io.UncheckedIOException;
-import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.TreeMap;
import java.util.UUID;
import java.util.function.BiConsumer;
-import static org.apache.paimon.codegen.CodeGenUtils.newNormalizedKeyComputer;
-import static org.apache.paimon.codegen.CodeGenUtils.newRecordComparator;
import static org.apache.paimon.utils.Preconditions.checkArgument;
/** Assign UPDATE_BEFORE and bucket for the input record, output record with bucket. */
@@ -156,26 +148,13 @@ public class GlobalIndexAssigner<T> implements Serializable, Closeable {
// create bootstrap sort buffer
this.bootstrap = true;
- int pageSize = coreOptions.pageSize();
- long bufferSize = coreOptions.writeBufferSize() / 2;
- RecordComparator comparator =
- newRecordComparator(
- Collections.singletonList(DataTypes.BYTES()), "binary_comparator");
- BinaryInMemorySortBuffer sortBuffer =
- BinaryInMemorySortBuffer.createBuffer(
- newNormalizedKeyComputer(
- Collections.singletonList(DataTypes.BYTES()),
- "binary_normalized_key"),
- new InternalRowSerializer(DataTypes.BYTES(), DataTypes.BYTES()),
- comparator,
- new HeapMemorySegmentPool(bufferSize, pageSize));
this.bootstrapBuffer =
- new BinaryExternalSortBuffer(
- new BinaryRowSerializer(2),
- comparator,
- pageSize,
- sortBuffer,
+ BinaryExternalSortBuffer.create(
ioManager,
+ RowType.of(DataTypes.BYTES()),
+ RowType.of(DataTypes.BYTES(), DataTypes.BYTES()),
+ coreOptions.writeBufferSize() / 2,
+ coreOptions.pageSize(),
coreOptions.localSortMaxNumFileHandles());
}
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sorter/SortOperator.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sorter/SortOperator.java
index 5d16c445a..a35a9f161 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sorter/SortOperator.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sorter/SortOperator.java
@@ -19,19 +19,10 @@
package org.apache.paimon.flink.sorter;
import org.apache.paimon.annotation.VisibleForTesting;
-import org.apache.paimon.codegen.CodeGenUtils;
-import org.apache.paimon.codegen.NormalizedKeyComputer;
-import org.apache.paimon.codegen.RecordComparator;
import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.data.InternalRow;
-import org.apache.paimon.data.serializer.BinaryRowSerializer;
-import org.apache.paimon.data.serializer.InternalRowSerializer;
-import org.apache.paimon.data.serializer.InternalSerializers;
import org.apache.paimon.disk.IOManager;
-import org.apache.paimon.memory.HeapMemorySegmentPool;
-import org.apache.paimon.memory.MemorySegmentPool;
import org.apache.paimon.sort.BinaryExternalSortBuffer;
-import org.apache.paimon.sort.BinaryInMemorySortBuffer;
import org.apache.paimon.types.RowType;
import org.apache.paimon.utils.MutableObjectIterator;
@@ -76,19 +67,6 @@ public class SortOperator extends TableStreamOperator<InternalRow>
@VisibleForTesting
void initBuffer() {
- InternalRowSerializer serializer = InternalSerializers.create(rowType);
- NormalizedKeyComputer normalizedKeyComputer =
- CodeGenUtils.newNormalizedKeyComputer(
- keyType.getFieldTypes(), "MemTableKeyComputer");
- RecordComparator keyComparator =
- CodeGenUtils.newRecordComparator(keyType.getFieldTypes(), "MemTableComparator");
-
- MemorySegmentPool memoryPool = new HeapMemorySegmentPool(maxMemory, pageSize);
-
- BinaryInMemorySortBuffer inMemorySortBuffer =
- BinaryInMemorySortBuffer.createBuffer(
- normalizedKeyComputer, serializer, keyComparator, memoryPool);
-
this.ioManager =
IOManager.create(
getContainingTask()
@@ -96,13 +74,8 @@ public class SortOperator extends TableStreamOperator<InternalRow>
.getIOManager()
.getSpillingDirectoriesPaths());
buffer =
- new BinaryExternalSortBuffer(
- new BinaryRowSerializer(serializer.getArity()),
- keyComparator,
- memoryPool.pageSize(),
- inMemorySortBuffer,
- ioManager,
- spillSortMaxNumFiles);
+ BinaryExternalSortBuffer.create(
+ ioManager, keyType, rowType, maxMemory, pageSize, spillSortMaxNumFiles);
}
@Override