[core] Support sequence.field sort order for pk table (apache#4385)

docs/layouts/shortcodes/generated/core_configuration.html

@@ -713,6 +713,12 @@
             <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>

paimon-codegen/src/main/java/org/apache/paimon/codegen/CodeGeneratorImpl.java

@@ -39,17 +39,17 @@ public GeneratedClass<NormalizedKeyComputer> generateNormalizedKeyComputer(
             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 GeneratedClass<RecordEqualiser> generateRecordEqualiser(
                 .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();
     }

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 List<String> sequenceField() {
                 .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 InlineElement getDescription() {
         }
     }
 
+    /** 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(

paimon-common/src/main/java/org/apache/paimon/codegen/CodeGenerator.java

@@ -44,9 +44,10 @@ GeneratedClass<NormalizedKeyComputer> generateNormalizedKeyComputer(
      * @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(

paimon-core/src/main/java/org/apache/paimon/codegen/CodeGenUtils.java

@@ -70,16 +70,20 @@ public static NormalizedKeyComputer newNormalizedKeyComputer(
     }
 
     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) {

paimon-core/src/main/java/org/apache/paimon/crosspartition/GlobalIndexAssigner.java

@@ -306,7 +306,8 @@ private void bulkLoadBootstrapRecords() {
                         coreOptions.pageSize(),
                         coreOptions.localSortMaxNumFileHandles(),
                         coreOptions.spillCompressOptions(),
-                        coreOptions.writeBufferSpillDiskSize());
+                        coreOptions.writeBufferSpillDiskSize(),
+                        coreOptions.sequenceFieldSortOrderIsAscending());
 
         Function<SortOrder, RowIterator> iteratorFunction =
                 sortOrder -> {

paimon-core/src/main/java/org/apache/paimon/lookup/RocksDBState.java

@@ -113,7 +113,8 @@ public static BinaryExternalSortBuffer createBulkLoadSorter(
                 options.pageSize(),
                 options.localSortMaxNumFileHandles(),
                 options.spillCompressOptions(),
-                options.writeBufferSpillDiskSize());
+                options.writeBufferSpillDiskSize(),
+                options.sequenceFieldSortOrderIsAscending());
     }
 
     /** A class wraps byte[] to implement equals and hashCode. */

paimon-core/src/main/java/org/apache/paimon/mergetree/SortBufferWriteBuffer.java

@@ -101,7 +101,7 @@ public SortBufferWriteBuffer(
         NormalizedKeyComputer normalizedKeyComputer =
                 CodeGenUtils.newNormalizedKeyComputer(fieldTypes, sortFieldArray);
         RecordComparator keyComparator =
-                CodeGenUtils.newRecordComparator(fieldTypes, sortFieldArray);
+                CodeGenUtils.newRecordComparator(fieldTypes, sortFieldArray, true);
 
         if (memoryPool.freePages() < 3) {
             throw new IllegalArgumentException(

paimon-core/src/main/java/org/apache/paimon/mergetree/compact/PartialUpdateMergeFunction.java

@@ -301,7 +301,7 @@ private Factory(Options options, RowType rowType, List<String> primaryKeys) {
                                     .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 MergeFunction<KeyValue> create(@Nullable int[][] projection) {
                                 projectedSeqComparators.put(
                                         newField,
                                         UserDefinedSeqComparator.create(
-                                                newRowType, newSequenceFields));
+                                                newRowType, newSequenceFields, true));
                             }
                         });
                 for (int i = 0; i < projects.length; i++) {

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 MergeFileSplitRead(
                 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 @@ private RecordReader<KeyValue> projectOuter(RecordReader<KeyValue> reader) {
     @Nullable
     public UserDefinedSeqComparator createUdsComparator() {
         return UserDefinedSeqComparator.create(
-                readerFactoryBuilder.readValueType(), sequenceFields);
+                readerFactoryBuilder.readValueType(), sequenceFields, sequenceOrder);
     }
 }

paimon-core/src/main/java/org/apache/paimon/sort/BinaryExternalSortBuffer.java

@@ -101,15 +101,17 @@ public static BinaryExternalSortBuffer create(
             int pageSize,
             int maxNumFileHandles,
             CompressOptions compression,
-            MemorySize maxDiskSize) {
+            MemorySize maxDiskSize,
+            boolean sequenceOrder) {
         return create(
                 ioManager,
                 rowType,
                 keyFields,
                 new HeapMemorySegmentPool(bufferSize, pageSize),
                 maxNumFileHandles,
                 compression,
-                maxDiskSize);
+                maxDiskSize,
+                sequenceOrder);
     }
 
     public static BinaryExternalSortBuffer create(
@@ -119,8 +121,10 @@ public static BinaryExternalSortBuffer create(
             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),

paimon-core/src/main/java/org/apache/paimon/utils/KeyComparatorSupplier.java

@@ -45,6 +45,6 @@ public KeyComparatorSupplier(RowType keyType) {
 
     @Override
     public RecordComparator get() {
-        return newRecordComparator(inputTypes, sortFields);
+        return newRecordComparator(inputTypes, sortFields, true);
     }
 }

paimon-core/src/main/java/org/apache/paimon/utils/UserDefinedSeqComparator.java

@@ -51,29 +51,33 @@ public int compare(InternalRow o1, InternalRow o2) {
 
     @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;
         }
 
         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);
     }
 }

paimon-core/src/test/java/org/apache/paimon/codegen/CodeGenUtilsTest.java

@@ -71,14 +71,15 @@ public void testNormalizedKeyComputerCodegenCacheMiss() {
     @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 @@ public void testRecordEqualiserCodegenCacheMiss() {
     @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}));
     }
 

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 FullCacheLookupTable(Context context) {
             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;

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 SortOperator(
             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 SortOperator(
         this.spillCompression = spillCompression;
         this.sinkParallelism = sinkParallelism;
         this.maxDiskSize = maxDiskSize;
+        this.sequenceOrder = sequenceOrder;
     }
 
     @Override
@@ -100,7 +103,8 @@ void initBuffer() {
                         pageSize,
                         spillSortMaxNumFiles,
                         spillCompression,
-                        maxDiskSize);
+                        maxDiskSize,
+                        sequenceOrder);
     }
 
     @Override

paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sorter/SortUtils.java

@@ -163,7 +163,8 @@ public Tuple2<KEY, RowData> map(RowData value) {
                                     options.localSortMaxNumFileHandles(),
                                     options.spillCompressOptions(),
                                     sinkParallelism,
-                                    options.writeBufferSpillDiskSize()))
+                                    options.writeBufferSpillDiskSize(),
+                                    options.sequenceFieldSortOrderIsAscending()))
                     .setParallelism(sinkParallelism)
                     // remove the key column from every row
                     .map(