[server] Support sequence groups in default and agg merge engine

fluss-common/src/main/java/org/apache/fluss/metadata/Schema.java

@@ -142,6 +142,11 @@ public Optional<AggFunction> getAggFunction(String columnName) {
                 .flatMap(Column::getAggFunction);
     }
 
+    /** Returns true if at least one column of this schema is protected by a sequence group. */
+    public boolean hasSequenceGroup() {
+        return columns.stream().anyMatch(col -> col.getSequenceColumns().isPresent());
+    }
+
     /** Returns the primary key indexes, if any, otherwise returns an empty array. */
     public int[] getPrimaryKeyIndexes() {
         final List<String> columns = getColumnNames();
@@ -365,7 +370,8 @@ public Builder fromColumns(List<Column> inputColumns) {
                                     column.dataType,
                                     column.comment,
                                     newColumnId,
-                                    column.aggFunction));
+                                    column.aggFunction,
+                                    column.sequenceColumns));
                 }
             }
 
@@ -488,6 +494,30 @@ public Builder withComment(@Nullable String comment) {
             return this;
         }
 
+        /**
+         * Apply the sequence columns ordering the previous column, i.e. put the previous column
+         * into the sequence group ordered by the given columns.
+         *
+         * <p>Passing more than one column declares a composite sequence key, where the columns are
+         * compared in the given order and the first unequal one decides.
+         *
+         * @param sequenceColumns the columns ordering the previous column
+         */
+        public Builder withSequenceColumns(String... sequenceColumns) {
+            checkNotNull(sequenceColumns, "Sequence columns must not be null.");
+            checkArgument(sequenceColumns.length > 0, "Sequence columns must not be empty.");
+            if (columns.isEmpty()) {
+                throw new IllegalArgumentException(
+                        "Method 'withSequenceColumns(...)' must be called after a column definition, "
+                                + "but there is no preceding column defined.");
+            }
+            columns.set(
+                    columns.size() - 1,
+                    columns.get(columns.size() - 1)
+                            .withSequenceColumns(Arrays.asList(sequenceColumns)));
+            return this;
+        }
+
         /**
          * Declares a primary key constraint for a set of given columns. Primary key uniquely
          * identify a row in a table. Neither of columns in a primary can be nullable. Adding a
@@ -589,6 +619,7 @@ public static final class Column implements Serializable {
         private final DataType dataType;
         private final @Nullable String comment;
         private final @Nullable AggFunction aggFunction;
+        private final @Nullable List<String> sequenceColumns;
 
         public Column(String columnName, DataType dataType) {
             this(columnName, dataType, null, UNKNOWN_COLUMN_ID, null);
@@ -609,11 +640,25 @@ public Column(
                 @Nullable String comment,
                 int columnId,
                 @Nullable AggFunction aggFunction) {
+            this(columnName, dataType, comment, columnId, aggFunction, null);
+        }
+
+        public Column(
+                String columnName,
+                DataType dataType,
+                @Nullable String comment,
+                int columnId,
+                @Nullable AggFunction aggFunction,
+                @Nullable List<String> sequenceColumns) {
             this.columnName = columnName;
             this.dataType = dataType;
             this.comment = comment;
             this.columnId = columnId;
             this.aggFunction = aggFunction;
+            this.sequenceColumns =
+                    sequenceColumns == null
+                            ? null
+                            : Collections.unmodifiableList(new ArrayList<>(sequenceColumns));
         }
 
         public String getName() {
@@ -641,12 +686,30 @@ public Optional<AggFunction> getAggFunction() {
             return Optional.ofNullable(aggFunction);
         }
 
+        /**
+         * Gets the sequence columns ordering this column: it only takes an incoming value when they
+         * are not older than the stored ones. More than one means a composite key, compared in the
+         * listed order until one differs.
+         *
+         * @return the sequence columns, or empty if the column is merged without order arbitration
+         */
+        public Optional<List<String>> getSequenceColumns() {
+            return Optional.ofNullable(sequenceColumns);
+        }
+
         public Column withComment(String comment) {
-            return new Column(columnName, dataType, comment, columnId, aggFunction);
+            return new Column(
+                    columnName, dataType, comment, columnId, aggFunction, sequenceColumns);
         }
 
         public Column withAggFunction(@Nullable AggFunction aggFunction) {
-            return new Column(columnName, dataType, comment, columnId, aggFunction);
+            return new Column(
+                    columnName, dataType, comment, columnId, aggFunction, sequenceColumns);
+        }
+
+        public Column withSequenceColumns(@Nullable List<String> sequenceColumns) {
+            return new Column(
+                    columnName, dataType, comment, columnId, aggFunction, sequenceColumns);
         }
 
         @Override
@@ -676,12 +739,14 @@ public boolean equals(Object o) {
                     && Objects.equals(dataType, that.dataType)
                     && Objects.equals(comment, that.comment)
                     && Objects.equals(columnId, that.columnId)
-                    && Objects.equals(aggFunction, that.aggFunction);
+                    && Objects.equals(aggFunction, that.aggFunction)
+                    && Objects.equals(sequenceColumns, that.sequenceColumns);
         }
 
         @Override
         public int hashCode() {
-            return Objects.hash(columnName, dataType, comment, columnId, aggFunction);
+            return Objects.hash(
+                    columnName, dataType, comment, columnId, aggFunction, sequenceColumns);
         }
     }
 
@@ -820,7 +885,8 @@ private static List<Column> normalizeColumns(
                                 column.getDataType().copy(false),
                                 column.getComment().isPresent() ? column.getComment().get() : null,
                                 column.getColumnId(),
-                                column.getAggFunction().orElse(null)));
+                                column.getAggFunction().orElse(null),
+                                column.sequenceColumns));
             } else {
                 newColumns.add(column);
             }

fluss-common/src/main/java/org/apache/fluss/utils/json/ColumnJsonSerde.java

@@ -27,8 +27,10 @@
 import org.apache.fluss.types.DataType;
 
 import java.io.IOException;
+import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.Iterator;
+import java.util.List;
 import java.util.Map;
 
 import static org.apache.fluss.metadata.Schema.Column.UNKNOWN_COLUMN_ID;
@@ -46,6 +48,7 @@ public class ColumnJsonSerde
     static final String AGG_FUNCTION = "agg_function";
     static final String AGG_FUNCTION_TYPE = "type";
     static final String AGG_FUNCTION_PARAMS = "parameters";
+    static final String SEQUENCE_COLUMNS = "sequence_columns";
 
     @Override
     public void serialize(Schema.Column column, JsonGenerator generator) throws IOException {
@@ -71,6 +74,13 @@ public void serialize(Schema.Column column, JsonGenerator generator) throws IOEx
             }
             generator.writeEndObject();
         }
+        if (column.getSequenceColumns().isPresent()) {
+            generator.writeArrayFieldStart(SEQUENCE_COLUMNS);
+            for (String sequenceColumn : column.getSequenceColumns().get()) {
+                generator.writeString(sequenceColumn);
+            }
+            generator.writeEndArray();
+        }
         generator.writeNumberField(ID, column.getColumnId());
 
         generator.writeEndObject();
@@ -105,11 +115,20 @@ public Schema.Column deserialize(JsonNode node) {
             }
         }
 
+        List<String> sequenceColumns = null;
+        if (node.hasNonNull(SEQUENCE_COLUMNS)) {
+            sequenceColumns = new ArrayList<>();
+            for (JsonNode sequenceColumn : node.get(SEQUENCE_COLUMNS)) {
+                sequenceColumns.add(sequenceColumn.asText());
+            }
+        }
+
         return new Schema.Column(
                 columnName,
                 dataType,
                 node.hasNonNull(COMMENT) ? node.get(COMMENT).asText() : null,
                 node.has(ID) ? node.get(ID).asInt() : UNKNOWN_COLUMN_ID,
-                aggFunction);
+                aggFunction,
+                sequenceColumns);
     }
 }

fluss-common/src/test/java/org/apache/fluss/utils/json/ColumnJsonSerdeTest.java

@@ -27,6 +27,7 @@
 import java.io.IOException;
 import java.nio.charset.StandardCharsets;
 import java.util.Arrays;
+import java.util.Collections;
 
 import static org.assertj.core.api.Assertions.assertThat;
 
@@ -38,7 +39,7 @@ protected ColumnJsonSerdeTest() {
 
     @Override
     protected Schema.Column[] createObjects() {
-        Schema.Column[] columns = new Schema.Column[5];
+        Schema.Column[] columns = new Schema.Column[6];
         columns[0] = new Schema.Column("a", DataTypes.STRING());
         columns[1] = new Schema.Column("b", DataTypes.INT(), "hello b");
         columns[2] = new Schema.Column("c", new IntType(false), "hello c");
@@ -53,6 +54,9 @@ protected Schema.Column[] createObjects() {
                                         DataTypes.FIELD("g", DataTypes.STRING(), 1))),
                         "hello c",
                         (short) 2);
+        columns[5] =
+                new Schema.Column("h", DataTypes.STRING(), null, (short) 3)
+                        .withSequenceColumns(Collections.singletonList("ts"));
         return columns;
     }
 
@@ -63,7 +67,8 @@ protected String[] expectedJsons() {
             "{\"name\":\"b\",\"data_type\":{\"type\":\"INTEGER\"},\"comment\":\"hello b\",\"id\":-1}",
             "{\"name\":\"c\",\"data_type\":{\"type\":\"INTEGER\",\"nullable\":false},\"comment\":\"hello c\",\"id\":-1}",
             "{\"name\":\"d\",\"data_type\":{\"type\":\"INTEGER\",\"nullable\":false},\"comment\":\"hello c\",\"id\":2}",
-            "{\"name\":\"e\",\"data_type\":{\"type\":\"ROW\",\"fields\":[{\"name\":\"f\",\"field_type\":{\"type\":\"STRING\"},\"field_id\":-1},{\"name\":\"g\",\"field_type\":{\"type\":\"STRING\"},\"field_id\":1}]},\"comment\":\"hello c\",\"id\":2}"
+            "{\"name\":\"e\",\"data_type\":{\"type\":\"ROW\",\"fields\":[{\"name\":\"f\",\"field_type\":{\"type\":\"STRING\"},\"field_id\":-1},{\"name\":\"g\",\"field_type\":{\"type\":\"STRING\"},\"field_id\":1}]},\"comment\":\"hello c\",\"id\":2}",
+            "{\"name\":\"h\",\"data_type\":{\"type\":\"STRING\"},\"sequence_columns\":[\"ts\"],\"id\":3}"
         };
     }
 

fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/utils/FlinkConversions.java

@@ -95,6 +95,9 @@
 /** Utils for conversion between Flink and Fluss. */
 public class FlinkConversions {
 
+    private static final String SEQUENCE_GROUP_PREFIX = "fields.";
+    private static final String SEQUENCE_GROUP_SUFFIX = ".sequence-group";
+
     private FlinkConversions() {}
 
     /** Convert Fluss's type to Flink's type. */
@@ -217,13 +220,21 @@ public static TableDescriptor toFlussTable(ResolvedCatalogBaseTable<?> catalogBa
         // Check if aggregation merge engine is enabled to optimize parsing
         boolean isAggregationEngine = isAggregationMergeEngine(flinkTableConf);
 
+        // Sequence groups apply to the primary key table without merge engine, so they are parsed
+        // regardless of the merge engine and rejected server side when unsupported
+        Map<String, List<String>> sequenceColumnsOf = parseSequenceGroups(flinkTableConf);
+
         // Build schema with physical columns
         resolvedSchema.getColumns().stream()
                 .filter(Column::isPhysical)
                 .forEachOrdered(
                         column ->
                                 addColumnToSchema(
-                                        schemBuilder, column, flinkTableConf, isAggregationEngine));
+                                        schemBuilder,
+                                        column,
+                                        flinkTableConf,
+                                        isAggregationEngine,
+                                        sequenceColumnsOf));
 
         // Configure auto-increment columns based on the 'auto-increment.fields' option.
         if (flinkTableConf.containsKey(AUTO_INCREMENT_FIELDS.key())) {
@@ -744,18 +755,100 @@ private static boolean isAggregationMergeEngine(Configuration tableConf) {
     }
 
     /**
-     * Add a column to the schema builder with optional aggregation function.
+     * Parses the sequence groups declared in the table options, keyed by the sequence columns and
+     * listing the columns they protect:
+     *
+     * <pre>
+     * 'fields.g1.sequence-group' = 'a,b'
+     * 'fields.g1,g2.sequence-group' = 'c'
+     * </pre>
+     *
+     * <p>The returned mapping is inverted, giving the sequence columns of each protected column,
+     * which is how {@link Schema.Column} stores the relation.
+     */
+    private static Map<String, List<String>> parseSequenceGroups(Configuration tableConf) {
+        Map<String, List<String>> sequenceColumnsOf = new HashMap<>();
+        for (String key : tableConf.keySet()) {
+            if (!key.startsWith(SEQUENCE_GROUP_PREFIX) || !key.endsWith(SEQUENCE_GROUP_SUFFIX)) {
+                continue;
+            }
+            List<String> sequenceColumns =
+                    splitColumns(
+                            key.substring(
+                                    SEQUENCE_GROUP_PREFIX.length(),
+                                    key.length() - SEQUENCE_GROUP_SUFFIX.length()),
+                            key,
+                            "sequence columns");
+            // the key comes from the option keys, so a value is always present
+            List<String> protectedColumns =
+                    splitColumns(tableConf.getString(key, ""), key, "protected columns");
+
+            for (String protectedColumn : protectedColumns) {
+                if (sequenceColumns.contains(protectedColumn)) {
+                    throw new IllegalArgumentException(
+                            String.format(
+                                    "Invalid option '%s': column '%s' must not be protected by itself.",
+                                    key, protectedColumn));
+                }
+                List<String> previous = sequenceColumnsOf.put(protectedColumn, sequenceColumns);
+                if (previous != null) {
+                    throw new IllegalArgumentException(
+                            String.format(
+                                    "Column '%s' is declared repeatedly by sequence groups %s and %s.",
+                                    protectedColumn, previous, sequenceColumns));
+                }
+            }
+        }
+        return sequenceColumnsOf;
+    }
+
+    /**
+     * Splits a comma separated list of column names, rejecting an empty list as well as an empty or
+     * repeated name. A repeated name is always a typo, since naming a column twice adds nothing.
+     *
+     * @param description names the parsed list in the rejection message
+     */
+    private static List<String> splitColumns(String value, String key, String description) {
+        if (value.trim().isEmpty()) {
+            throw new IllegalArgumentException(
+                    String.format(
+                            "Invalid option '%s': the %s must not be empty.", key, description));
+        }
+        List<String> columns = new ArrayList<>();
+        for (String column : value.split(",")) {
+            String trimmed = column.trim();
+            if (trimmed.isEmpty()) {
+                throw new IllegalArgumentException(
+                        String.format(
+                                "Invalid option '%s': the %s must not be empty.",
+                                key, description));
+            }
+            if (columns.contains(trimmed)) {
+                throw new IllegalArgumentException(
+                        String.format(
+                                "Invalid option '%s': the %s name '%s' is declared more than once.",
+                                key, description, trimmed));
+            }
+            columns.add(trimmed);
+        }
+        return columns;
+    }
+
+    /**
+     * Add a column to the schema builder with optional aggregation function and sequence columns.
      *
      * @param schemaBuilder the schema builder
      * @param column the Flink column
      * @param tableConf the table configuration
      * @param parseAggFunction whether to parse aggregation function from config
+     * @param sequenceColumnsOf the sequence columns ordering each protected column
      */
     private static void addColumnToSchema(
             Schema.Builder schemaBuilder,
             Column column,
             Configuration tableConf,
-            boolean parseAggFunction) {
+            boolean parseAggFunction,
+            Map<String, List<String>> sequenceColumnsOf) {
         String columnName = column.getName();
         DataType flussDataType = toFlussType(column.getDataType());
 
@@ -774,6 +867,12 @@ private static void addColumnToSchema(
 
         // Add comment if present
         column.getComment().ifPresent(schemaBuilder::withComment);
+
+        // Put the column into the sequence group ordering it, if any
+        List<String> sequenceColumns = sequenceColumnsOf.get(columnName);
+        if (sequenceColumns != null) {
+            schemaBuilder.withSequenceColumns(sequenceColumns.toArray(new String[0]));
+        }
     }
 
     private static Map<String, String> extractCustomProperties(