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/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 6a6d1243c [core] Support sequence.field sort order for pk table (#4385)
6a6d1243c is described below
commit 6a6d1243c26fbc41ce224a98ff5769395be2eb06
Author: xuzifu666 <[email protected]>
AuthorDate: Wed Oct 30 12:50:54 2024 +0800
[core] Support sequence.field sort order for pk table (#4385)
---
.../shortcodes/generated/core_configuration.html | 6 ++++
.../apache/paimon/codegen/CodeGeneratorImpl.java | 10 +++---
.../main/java/org/apache/paimon/CoreOptions.java | 36 ++++++++++++++++++++++
.../org/apache/paimon/codegen/CodeGenerator.java | 3 +-
.../org/apache/paimon/codegen/CodeGenUtils.java | 10 ++++--
.../paimon/crosspartition/GlobalIndexAssigner.java | 3 +-
.../org/apache/paimon/lookup/RocksDBState.java | 3 +-
.../paimon/mergetree/SortBufferWriteBuffer.java | 2 +-
.../compact/PartialUpdateMergeFunction.java | 4 +--
.../paimon/operation/MergeFileSplitRead.java | 4 ++-
.../paimon/sort/BinaryExternalSortBuffer.java | 12 +++++---
.../apache/paimon/utils/KeyComparatorSupplier.java | 2 +-
.../paimon/utils/UserDefinedSeqComparator.java | 14 ++++++---
.../apache/paimon/codegen/CodeGenUtilsTest.java | 9 +++---
.../paimon/flink/lookup/FullCacheLookupTable.java | 8 +++--
.../apache/paimon/flink/sorter/SortOperator.java | 8 +++--
.../org/apache/paimon/flink/sorter/SortUtils.java | 3 +-
.../apache/paimon/flink/SchemaChangeITCase.java | 30 ++++++++++++++++++
.../paimon/flink/sorter/SortOperatorTest.java | 6 ++--
19 files changed, 137 insertions(+), 36 deletions(-)
diff --git a/docs/layouts/shortcodes/generated/core_configuration.html
b/docs/layouts/shortcodes/generated/core_configuration.html
index 9bbed03bd..84ba86124 100644
--- a/docs/layouts/shortcodes/generated/core_configuration.html
+++ b/docs/layouts/shortcodes/generated/core_configuration.html
@@ -713,6 +713,12 @@ This config option does not affect the default filesystem
metastore.</td>
<td>String</td>
<td>The field that generates the sequence number for primary key
table, the sequence number determines which data is the most recent.</td>
</tr>
+ <tr>
+ <td><h5>sequence.field.sort-order</h5></td>
+ <td style="word-wrap: break-word;">ascending</td>
+ <td><p>Enum</p></td>
+ <td>Specify the order of sequence.field.<br /><br />Possible
values:<ul><li>"ascending": specifies sequence.field sort order is
ascending.</li><li>"descending": specifies sequence.field sort order is
descending.</li></ul></td>
+ </tr>
<tr>
<td><h5>sink.watermark-time-zone</h5></td>
<td style="word-wrap: break-word;">"UTC"</td>
diff --git
a/paimon-codegen/src/main/java/org/apache/paimon/codegen/CodeGeneratorImpl.java
b/paimon-codegen/src/main/java/org/apache/paimon/codegen/CodeGeneratorImpl.java
index 29096e96b..5a6a27738 100644
---
a/paimon-codegen/src/main/java/org/apache/paimon/codegen/CodeGeneratorImpl.java
+++
b/paimon-codegen/src/main/java/org/apache/paimon/codegen/CodeGeneratorImpl.java
@@ -39,17 +39,17 @@ public class CodeGeneratorImpl implements CodeGenerator {
List<DataType> inputTypes, int[] sortFields) {
return new SortCodeGenerator(
RowType.builder().fields(inputTypes).build(),
- getAscendingSortSpec(sortFields))
+ getAscendingSortSpec(sortFields, true))
.generateNormalizedKeyComputer("NormalizedKeyComputer");
}
@Override
public GeneratedClass<RecordComparator> generateRecordComparator(
- List<DataType> inputTypes, int[] sortFields) {
+ List<DataType> inputTypes, int[] sortFields, boolean
isAscendingOrder) {
return ComparatorCodeGenerator.gen(
"RecordComparator",
RowType.builder().fields(inputTypes).build(),
- getAscendingSortSpec(sortFields));
+ getAscendingSortSpec(sortFields, isAscendingOrder));
}
@Override
@@ -59,10 +59,10 @@ public class CodeGeneratorImpl implements CodeGenerator {
.generateRecordEqualiser("RecordEqualiser");
}
- private SortSpec getAscendingSortSpec(int[] sortFields) {
+ private SortSpec getAscendingSortSpec(int[] sortFields, boolean
isAscendingOrder) {
SortSpec.SortSpecBuilder builder = SortSpec.builder();
for (int sortField : sortFields) {
- builder.addField(sortField, true, false);
+ builder.addField(sortField, isAscendingOrder, false);
}
return builder.build();
}
diff --git a/paimon-common/src/main/java/org/apache/paimon/CoreOptions.java
b/paimon-common/src/main/java/org/apache/paimon/CoreOptions.java
index 18dae0d38..39eda336b 100644
--- a/paimon-common/src/main/java/org/apache/paimon/CoreOptions.java
+++ b/paimon-common/src/main/java/org/apache/paimon/CoreOptions.java
@@ -609,6 +609,13 @@ public class CoreOptions implements Serializable {
"The field that generates the sequence number for
primary key table,"
+ " the sequence number determines which
data is the most recent.");
+ @Immutable
+ public static final ConfigOption<SortOrder> SEQUENCE_FIELD_SORT_ORDER =
+ key("sequence.field.sort-order")
+ .enumType(SortOrder.class)
+ .defaultValue(SortOrder.ASCENDING)
+ .withDescription("Specify the order of sequence.field.");
+
@Immutable
public static final ConfigOption<Boolean>
PARTIAL_UPDATE_REMOVE_RECORD_ON_DELETE =
key("partial-update.remove-record-on-delete")
@@ -2043,6 +2050,10 @@ public class CoreOptions implements Serializable {
.orElse(Collections.emptyList());
}
+ public boolean sequenceFieldSortOrderIsAscending() {
+ return options.get(SEQUENCE_FIELD_SORT_ORDER) == SortOrder.ASCENDING;
+ }
+
public boolean partialUpdateRemoveRecordOnDelete() {
return options.get(PARTIAL_UPDATE_REMOVE_RECORD_ON_DELETE);
}
@@ -2382,6 +2393,31 @@ public class CoreOptions implements Serializable {
}
}
+ /** Specifies the sort order for field sequence id. */
+ public enum SortOrder implements DescribedEnum {
+ ASCENDING("ascending", "specifies sequence.field sort order is
ascending."),
+
+ DESCENDING("descending", "specifies sequence.field sort order is
descending.");
+
+ private final String value;
+ private final String description;
+
+ SortOrder(String value, String description) {
+ this.value = value;
+ this.description = description;
+ }
+
+ @Override
+ public String toString() {
+ return value;
+ }
+
+ @Override
+ public InlineElement getDescription() {
+ return text(description);
+ }
+ }
+
/** Specifies the log consistency mode for table. */
public enum LogConsistency implements DescribedEnum {
TRANSACTIONAL(
diff --git
a/paimon-common/src/main/java/org/apache/paimon/codegen/CodeGenerator.java
b/paimon-common/src/main/java/org/apache/paimon/codegen/CodeGenerator.java
index e13761914..324a8d726 100644
--- a/paimon-common/src/main/java/org/apache/paimon/codegen/CodeGenerator.java
+++ b/paimon-common/src/main/java/org/apache/paimon/codegen/CodeGenerator.java
@@ -44,9 +44,10 @@ public interface CodeGenerator {
* @param inputTypes input types.
* @param sortFields the sort key fields. Records are compared by the
first field, then the
* second field, then the third field and so on. All fields are
compared in ascending order.
+ * @param isAscendingOrder decide the sort key fields order whether is
ascending
*/
GeneratedClass<RecordComparator> generateRecordComparator(
- List<DataType> inputTypes, int[] sortFields);
+ List<DataType> inputTypes, int[] sortFields, boolean
isAscendingOrder);
/** Generate a {@link RecordEqualiser} with fields. */
GeneratedClass<RecordEqualiser> generateRecordEqualiser(
diff --git
a/paimon-core/src/main/java/org/apache/paimon/codegen/CodeGenUtils.java
b/paimon-core/src/main/java/org/apache/paimon/codegen/CodeGenUtils.java
index 76aeae547..d452df5fb 100644
--- a/paimon-core/src/main/java/org/apache/paimon/codegen/CodeGenUtils.java
+++ b/paimon-core/src/main/java/org/apache/paimon/codegen/CodeGenUtils.java
@@ -70,16 +70,20 @@ public class CodeGenUtils {
}
public static RecordComparator newRecordComparator(List<DataType>
inputTypes) {
- return newRecordComparator(inputTypes, IntStream.range(0,
inputTypes.size()).toArray());
+ return newRecordComparator(
+ inputTypes, IntStream.range(0, inputTypes.size()).toArray(),
true);
}
public static RecordComparator newRecordComparator(
- List<DataType> inputTypes, int[] sortFields) {
+ List<DataType> inputTypes, int[] sortFields, boolean
isAscendingOrder) {
return generate(
RecordComparator.class,
inputTypes,
sortFields,
- () -> getCodeGenerator().generateRecordComparator(inputTypes,
sortFields));
+ () ->
+ getCodeGenerator()
+ .generateRecordComparator(
+ inputTypes, sortFields,
isAscendingOrder));
}
public static RecordEqualiser newRecordEqualiser(List<DataType>
fieldTypes) {
diff --git
a/paimon-core/src/main/java/org/apache/paimon/crosspartition/GlobalIndexAssigner.java
b/paimon-core/src/main/java/org/apache/paimon/crosspartition/GlobalIndexAssigner.java
index 311b997a4..4700e7399 100644
---
a/paimon-core/src/main/java/org/apache/paimon/crosspartition/GlobalIndexAssigner.java
+++
b/paimon-core/src/main/java/org/apache/paimon/crosspartition/GlobalIndexAssigner.java
@@ -306,7 +306,8 @@ public class GlobalIndexAssigner implements Serializable,
Closeable {
coreOptions.pageSize(),
coreOptions.localSortMaxNumFileHandles(),
coreOptions.spillCompressOptions(),
- coreOptions.writeBufferSpillDiskSize());
+ coreOptions.writeBufferSpillDiskSize(),
+ coreOptions.sequenceFieldSortOrderIsAscending());
Function<SortOrder, RowIterator> iteratorFunction =
sortOrder -> {
diff --git
a/paimon-core/src/main/java/org/apache/paimon/lookup/RocksDBState.java
b/paimon-core/src/main/java/org/apache/paimon/lookup/RocksDBState.java
index 25e58984e..0181917a7 100644
--- a/paimon-core/src/main/java/org/apache/paimon/lookup/RocksDBState.java
+++ b/paimon-core/src/main/java/org/apache/paimon/lookup/RocksDBState.java
@@ -113,7 +113,8 @@ public abstract class RocksDBState<K, V, CacheV> {
options.pageSize(),
options.localSortMaxNumFileHandles(),
options.spillCompressOptions(),
- options.writeBufferSpillDiskSize());
+ options.writeBufferSpillDiskSize(),
+ options.sequenceFieldSortOrderIsAscending());
}
/** A class wraps byte[] to implement equals and hashCode. */
diff --git
a/paimon-core/src/main/java/org/apache/paimon/mergetree/SortBufferWriteBuffer.java
b/paimon-core/src/main/java/org/apache/paimon/mergetree/SortBufferWriteBuffer.java
index 76c84fd4c..433ff2158 100644
---
a/paimon-core/src/main/java/org/apache/paimon/mergetree/SortBufferWriteBuffer.java
+++
b/paimon-core/src/main/java/org/apache/paimon/mergetree/SortBufferWriteBuffer.java
@@ -101,7 +101,7 @@ public class SortBufferWriteBuffer implements WriteBuffer {
NormalizedKeyComputer normalizedKeyComputer =
CodeGenUtils.newNormalizedKeyComputer(fieldTypes,
sortFieldArray);
RecordComparator keyComparator =
- CodeGenUtils.newRecordComparator(fieldTypes, sortFieldArray);
+ CodeGenUtils.newRecordComparator(fieldTypes, sortFieldArray,
true);
if (memoryPool.freePages() < 3) {
throw new IllegalArgumentException(
diff --git
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/PartialUpdateMergeFunction.java
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/PartialUpdateMergeFunction.java
index 11661851c..b15d9388a 100644
---
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/PartialUpdateMergeFunction.java
+++
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/PartialUpdateMergeFunction.java
@@ -301,7 +301,7 @@ public class PartialUpdateMergeFunction implements
MergeFunction<KeyValue> {
.collect(Collectors.toList());
Supplier<FieldsComparator> userDefinedSeqComparator =
- () -> UserDefinedSeqComparator.create(rowType,
sequenceFields);
+ () -> UserDefinedSeqComparator.create(rowType,
sequenceFields, true);
Arrays.stream(v.split(FIELDS_SEPARATOR))
.map(
fieldName ->
@@ -390,7 +390,7 @@ public class PartialUpdateMergeFunction implements
MergeFunction<KeyValue> {
projectedSeqComparators.put(
newField,
UserDefinedSeqComparator.create(
- newRowType,
newSequenceFields));
+ newRowType, newSequenceFields,
true));
}
});
for (int i = 0; i < projects.length; i++) {
diff --git
a/paimon-core/src/main/java/org/apache/paimon/operation/MergeFileSplitRead.java
b/paimon-core/src/main/java/org/apache/paimon/operation/MergeFileSplitRead.java
index d1f332a03..c21c3683c 100644
---
a/paimon-core/src/main/java/org/apache/paimon/operation/MergeFileSplitRead.java
+++
b/paimon-core/src/main/java/org/apache/paimon/operation/MergeFileSplitRead.java
@@ -77,6 +77,7 @@ public class MergeFileSplitRead implements
SplitRead<KeyValue> {
private final MergeFunctionFactory<KeyValue> mfFactory;
private final MergeSorter mergeSorter;
private final List<String> sequenceFields;
+ private final boolean sequenceOrder;
@Nullable private RowType readKeyType;
@@ -106,6 +107,7 @@ public class MergeFileSplitRead implements
SplitRead<KeyValue> {
new MergeSorter(
CoreOptions.fromMap(tableSchema.options()), keyType,
valueType, null);
this.sequenceFields = options.sequenceField();
+ this.sequenceOrder = options.sequenceFieldSortOrderIsAscending();
}
public Comparator<InternalRow> keyComparator() {
@@ -338,6 +340,6 @@ public class MergeFileSplitRead implements
SplitRead<KeyValue> {
@Nullable
public UserDefinedSeqComparator createUdsComparator() {
return UserDefinedSeqComparator.create(
- readerFactoryBuilder.readValueType(), sequenceFields);
+ readerFactoryBuilder.readValueType(), sequenceFields,
sequenceOrder);
}
}
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 1ae453546..4bfbcd5ec 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
@@ -101,7 +101,8 @@ public class BinaryExternalSortBuffer implements SortBuffer
{
int pageSize,
int maxNumFileHandles,
CompressOptions compression,
- MemorySize maxDiskSize) {
+ MemorySize maxDiskSize,
+ boolean sequenceOrder) {
return create(
ioManager,
rowType,
@@ -109,7 +110,8 @@ public class BinaryExternalSortBuffer implements SortBuffer
{
new HeapMemorySegmentPool(bufferSize, pageSize),
maxNumFileHandles,
compression,
- maxDiskSize);
+ maxDiskSize,
+ sequenceOrder);
}
public static BinaryExternalSortBuffer create(
@@ -119,8 +121,10 @@ public class BinaryExternalSortBuffer implements
SortBuffer {
MemorySegmentPool pool,
int maxNumFileHandles,
CompressOptions compression,
- MemorySize maxDiskSize) {
- RecordComparator comparator =
newRecordComparator(rowType.getFieldTypes(), keyFields);
+ MemorySize maxDiskSize,
+ boolean sequenceOrder) {
+ RecordComparator comparator =
+ newRecordComparator(rowType.getFieldTypes(), keyFields,
sequenceOrder);
BinaryInMemorySortBuffer sortBuffer =
BinaryInMemorySortBuffer.createBuffer(
newNormalizedKeyComputer(rowType.getFieldTypes(),
keyFields),
diff --git
a/paimon-core/src/main/java/org/apache/paimon/utils/KeyComparatorSupplier.java
b/paimon-core/src/main/java/org/apache/paimon/utils/KeyComparatorSupplier.java
index 6f90cef01..25fd07b7c 100644
---
a/paimon-core/src/main/java/org/apache/paimon/utils/KeyComparatorSupplier.java
+++
b/paimon-core/src/main/java/org/apache/paimon/utils/KeyComparatorSupplier.java
@@ -45,6 +45,6 @@ public class KeyComparatorSupplier implements
SerializableSupplier<Comparator<In
@Override
public RecordComparator get() {
- return newRecordComparator(inputTypes, sortFields);
+ return newRecordComparator(inputTypes, sortFields, true);
}
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/utils/UserDefinedSeqComparator.java
b/paimon-core/src/main/java/org/apache/paimon/utils/UserDefinedSeqComparator.java
index ec7a00bcb..5e5eb293d 100644
---
a/paimon-core/src/main/java/org/apache/paimon/utils/UserDefinedSeqComparator.java
+++
b/paimon-core/src/main/java/org/apache/paimon/utils/UserDefinedSeqComparator.java
@@ -51,11 +51,13 @@ public class UserDefinedSeqComparator implements
FieldsComparator {
@Nullable
public static UserDefinedSeqComparator create(RowType rowType, CoreOptions
options) {
- return create(rowType, options.sequenceField());
+ return create(
+ rowType, options.sequenceField(),
options.sequenceFieldSortOrderIsAscending());
}
@Nullable
- public static UserDefinedSeqComparator create(RowType rowType,
List<String> sequenceFields) {
+ public static UserDefinedSeqComparator create(
+ RowType rowType, List<String> sequenceFields, boolean
isAscendingOrder) {
if (sequenceFields.isEmpty()) {
return null;
}
@@ -63,17 +65,19 @@ public class UserDefinedSeqComparator implements
FieldsComparator {
List<String> fieldNames = rowType.getFieldNames();
int[] fields =
sequenceFields.stream().mapToInt(fieldNames::indexOf).toArray();
- return create(rowType, fields);
+ return create(rowType, fields, isAscendingOrder);
}
@Nullable
- public static UserDefinedSeqComparator create(RowType rowType, int[]
sequenceFields) {
+ public static UserDefinedSeqComparator create(
+ RowType rowType, int[] sequenceFields, boolean isAscendingOrder) {
if (sequenceFields.length == 0) {
return null;
}
RecordComparator comparator =
- CodeGenUtils.newRecordComparator(rowType.getFieldTypes(),
sequenceFields);
+ CodeGenUtils.newRecordComparator(
+ rowType.getFieldTypes(), sequenceFields,
isAscendingOrder);
return new UserDefinedSeqComparator(sequenceFields, comparator);
}
}
diff --git
a/paimon-core/src/test/java/org/apache/paimon/codegen/CodeGenUtilsTest.java
b/paimon-core/src/test/java/org/apache/paimon/codegen/CodeGenUtilsTest.java
index b7e416af1..c0da56f54 100644
--- a/paimon-core/src/test/java/org/apache/paimon/codegen/CodeGenUtilsTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/codegen/CodeGenUtilsTest.java
@@ -71,14 +71,15 @@ class CodeGenUtilsTest {
@Test
public void testRecordComparatorCodegenCache() {
assertClassEquals(
- () -> newRecordComparator(Arrays.asList(STRING(), INT()), new
int[] {0, 1}));
+ () -> newRecordComparator(Arrays.asList(STRING(), INT()), new
int[] {0, 1}, true));
}
@Test
public void testRecordComparatorCodegenCacheMiss() {
assertClassNotEquals(
- newRecordComparator(Arrays.asList(STRING(), INT()), new int[]
{0, 1}),
- newRecordComparator(Arrays.asList(STRING(), INT(), DOUBLE()),
new int[] {0, 1, 2}));
+ newRecordComparator(Arrays.asList(STRING(), INT()), new int[]
{0, 1}, true),
+ newRecordComparator(
+ Arrays.asList(STRING(), INT(), DOUBLE()), new int[]
{0, 1, 2}, true));
}
@Test
@@ -96,7 +97,7 @@ class CodeGenUtilsTest {
@Test
public void testHybridNotEqual() {
assertClassNotEquals(
- newRecordComparator(Arrays.asList(STRING(), INT()), new int[]
{0, 1}),
+ newRecordComparator(Arrays.asList(STRING(), INT()), new int[]
{0, 1}, true),
newNormalizedKeyComputer(Arrays.asList(STRING(), INT()), new
int[] {0, 1}));
}
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 89134585b..de69c67a4 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
@@ -91,8 +91,9 @@ public abstract class FullCacheLookupTable implements
LookupTable {
public FullCacheLookupTable(Context context) {
this.table = context.table;
List<String> sequenceFields = new ArrayList<>();
+ CoreOptions coreOptions = new CoreOptions(table.options());
if (table.primaryKeys().size() > 0) {
- sequenceFields = new CoreOptions(table.options()).sequenceField();
+ sequenceFields = coreOptions.sequenceField();
}
RowType projectedType = TypeUtils.project(table.rowType(),
context.projection);
if (sequenceFields.size() > 0) {
@@ -111,7 +112,10 @@ public abstract class FullCacheLookupTable implements
LookupTable {
projectedType = builder.build();
context =
context.copy(table.rowType().getFieldIndices(projectedType.getFieldNames()));
this.userDefinedSeqComparator =
- UserDefinedSeqComparator.create(projectedType,
sequenceFields);
+ UserDefinedSeqComparator.create(
+ projectedType,
+ sequenceFields,
+ coreOptions.sequenceFieldSortOrderIsAscending());
this.appendUdsFieldNumber = appendUdsFieldNumber.get();
} else {
this.userDefinedSeqComparator = null;
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 52ad89689..d4d5dd741 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
@@ -48,6 +48,7 @@ public class SortOperator extends
TableStreamOperator<InternalRow>
private final CompressOptions spillCompression;
private final int sinkParallelism;
private final MemorySize maxDiskSize;
+ private final boolean sequenceOrder;
private transient BinaryExternalSortBuffer buffer;
private transient IOManager ioManager;
@@ -60,7 +61,8 @@ public class SortOperator extends
TableStreamOperator<InternalRow>
int spillSortMaxNumFiles,
CompressOptions spillCompression,
int sinkParallelism,
- MemorySize maxDiskSize) {
+ MemorySize maxDiskSize,
+ boolean sequenceOrder) {
this.keyType = keyType;
this.rowType = rowType;
this.maxMemory = maxMemory;
@@ -70,6 +72,7 @@ public class SortOperator extends
TableStreamOperator<InternalRow>
this.spillCompression = spillCompression;
this.sinkParallelism = sinkParallelism;
this.maxDiskSize = maxDiskSize;
+ this.sequenceOrder = sequenceOrder;
}
@Override
@@ -100,7 +103,8 @@ public class SortOperator extends
TableStreamOperator<InternalRow>
pageSize,
spillSortMaxNumFiles,
spillCompression,
- maxDiskSize);
+ maxDiskSize,
+ sequenceOrder);
}
@Override
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sorter/SortUtils.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sorter/SortUtils.java
index e163ac364..f590c2fb7 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sorter/SortUtils.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sorter/SortUtils.java
@@ -163,7 +163,8 @@ public class SortUtils {
options.localSortMaxNumFileHandles(),
options.spillCompressOptions(),
sinkParallelism,
- options.writeBufferSpillDiskSize()))
+ options.writeBufferSpillDiskSize(),
+
options.sequenceFieldSortOrderIsAscending()))
.setParallelism(sinkParallelism)
// remove the key column from every row
.map(
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/SchemaChangeITCase.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/SchemaChangeITCase.java
index e1655bcb6..08f79efcc 100644
---
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/SchemaChangeITCase.java
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/SchemaChangeITCase.java
@@ -1007,6 +1007,36 @@ public class SchemaChangeITCase extends
CatalogITCaseBase {
.doesNotContain("schema");
}
+ @Test
+ public void testSequenceFieldSortOrder() {
+ // test default condition which get the largest record
+ sql(
+ "CREATE TABLE T1 (a STRING PRIMARY KEY NOT ENFORCED, b STRING,
c STRING) WITH ('sequence.field'='c')");
+ sql("INSERT INTO T1 VALUES ('a', 'b', 'l')");
+ sql("INSERT INTO T1 VALUES ('a', 'd', 'n')");
+ sql("INSERT INTO T1 VALUES ('a', 'e', 'm')");
+ List<Row> sql = sql("select * from T1");
+ assertThat(sql("select * from T1").toString()).isEqualTo("[+I[a, d,
n]]");
+
+ // test for get small record
+ sql(
+ "CREATE TABLE T2 (a STRING PRIMARY KEY NOT ENFORCED, b STRING,
c BIGINT) WITH ('sequence.field'='c',
'sequence.field.sort-order'='descending')");
+ sql("INSERT INTO T2 VALUES ('a', 'b', 1)");
+ sql("INSERT INTO T2 VALUES ('a', 'd', 3)");
+ sql("INSERT INTO T2 VALUES ('a', 'e', 2)");
+ sql = sql("select * from T2");
+ assertThat(sql("select * from T2").toString()).isEqualTo("[+I[a, b,
1]]");
+
+ // test for get largest record
+ sql(
+ "CREATE TABLE T3 (a STRING PRIMARY KEY NOT ENFORCED, b STRING,
c DOUBLE) WITH ('sequence.field'='c',
'sequence.field.sort-order'='ascending')");
+ sql("INSERT INTO T3 VALUES ('a', 'b', 1.0)");
+ sql("INSERT INTO T3 VALUES ('a', 'd', 3.0)");
+ sql("INSERT INTO T3 VALUES ('a', 'e', 2.0)");
+ sql = sql("select * from T3");
+ assertThat(sql("select * from T3").toString()).isEqualTo("[+I[a, d,
3.0]]");
+ }
+
@Test
public void testAlterTableMetadataComment() {
sql("CREATE TABLE T (a INT, name VARCHAR METADATA COMMENT 'header1', b
INT)");
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sorter/SortOperatorTest.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sorter/SortOperatorTest.java
index 155e259e0..c74c1c2c1 100644
---
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sorter/SortOperatorTest.java
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sorter/SortOperatorTest.java
@@ -65,7 +65,8 @@ public class SortOperatorTest {
128,
CompressOptions.defaultOptions(),
1,
- MemorySize.MAX_VALUE) {};
+ MemorySize.MAX_VALUE,
+ true) {};
OneInputStreamOperatorTestHarness harness =
createTestHarness(sortOperator);
harness.open();
@@ -114,7 +115,8 @@ public class SortOperatorTest {
128,
CompressOptions.defaultOptions(),
1,
- MemorySize.MAX_VALUE) {};
+ MemorySize.MAX_VALUE,
+ true) {};
OneInputStreamOperatorTestHarness harness =
createTestHarness(sortOperator);
harness.open();
File[] files =
harness.getEnvironment().getIOManager().getSpillingDirectories();
