[Feature] Support alter table syntax for sequence column

docs/en/sql-reference/sql-statements/Data Definition/ALTER TABLE.md

@@ -183,8 +183,16 @@ under the License.
         grammar:
             ENABLE FEATURE "BATCH_DELETE"
         note:
-            Only support unique tables
+            1) Only support unique tables
+            2) Batch deletion is supported for old tables, while new tables are already supported when they are created
 
+    8. Enable the ability to import in order by the value of the sequence column
+        grammer:
+            ENABLE FEATURE "SEQUENCE_LOAD" WITH PROPERTIES ("function_column.sequence_type" = "Date")
+        note:
+            1) Only support unique tables
+            2) The sequence_type is used to specify the type of the sequence column, which can be integral and time type
+            3) Only the orderliness of newly imported data is supported. Historical data cannot be changed
 
 
     Rename supports modification of the following names:
@@ -355,6 +363,12 @@ under the License.
     15. Modify the in_memory property of the table
 
         ALTER TABLE example_db.my_table set ("in_memory" = "true");
+    16. Enable batch delete support
+
+        ALTER TABLE example_db.my_table ENABLE FEATURE "BATCH_DELETE"
+    17. Enable the ability to import in order by the value of the Sequence column
+
+        ALTER TABLE example_db.my_table ENABLE FEATURE "SEQUENCE_LOAD" WITH PROPERTIES ("function_column.sequence_type" = "Date")
 
     [rename]
     1. Modify the table named table1 to table2

docs/zh-CN/sql-reference/sql-statements/Data Definition/ALTER TABLE.md

@@ -170,19 +170,27 @@ under the License.
         注意：
             1) index 中的所有列都要写出来
             2) value 列在 key 列之后
+
+    6. 修改table的属性，目前支持修改bloom filter列, colocate_with 属性和dynamic_partition属性，replication_num和default.replication_num属性
+        语法：
+            PROPERTIES ("key"="value")
+        注意：
+            也可以合并到上面的schema change操作中来修改，见下面例子
 
-    6. 启用批量删除支持
+    7. 启用批量删除支持
         语法：
             ENABLE FEATURE "BATCH_DELETE"
         注意：
             1） 只能用在unique 表
             2) 用于旧表支持批量删除功能，新表创建时已经支持
-            
-    6. 修改table的属性，目前支持修改bloom filter列, colocate_with 属性和dynamic_partition属性，replication_num和default.replication_num属性
-        语法：
-            PROPERTIES ("key"="value")
+
+    8. 启用按照sequence column的值来保证导入顺序的功能
+        语法:
+            ENABLE FEATURE "SEQUENCE_LOAD" WITH PROPERTIES ("function_column.sequence_type" = "Date")
         注意：
-            也可以合并到上面的schema change操作中来修改，见下面例子
+            1）只能用在unique表
+            2) sequence_type用来指定sequence列的类型，可以为整型和时间类型
+            3) 只支持新导入数据的有序性，历史数据无法更改
 
 
     rename 支持对以下名称进行修改：
@@ -352,7 +360,9 @@ under the License.
         ALTER TABLE example_db.my_table set ("in_memory" = "true");
     16. 启用 批量删除功能
         ALTER TABLE example_db.my_table ENABLE FEATURE "BATCH_DELETE"
-
+    17. 启用按照sequence column的值来保证导入顺序的功能
+
+        ALTER TABLE example_db.my_table ENABLE FEATURE "SEQUENCE_LOAD" WITH PROPERTIES ("function_column.sequence_type" = "Date")
 
     [rename]
     1. 将名为 table1 的表修改为 table2

fe/fe-core/src/main/cup/sql_parser.cup

@@ -400,7 +400,7 @@ nonterminal SetVar option_value, option_value_follow_option_type, option_value_n
 nonterminal List<SetVar> option_value_list, option_value_list_continued, start_option_value_list,
         start_option_value_list_following_option_type, user_property_list;
 
-nonterminal Map<String, String> key_value_map, opt_key_value_map, opt_properties, opt_ext_properties;
+nonterminal Map<String, String> key_value_map, opt_key_value_map, opt_properties, opt_ext_properties, opt_enable_feature_properties;
 nonterminal ColumnDef column_definition;
 nonterminal IndexDef index_definition;
 nonterminal ArrayList<ColumnDef> column_definition_list;
@@ -977,9 +977,19 @@ alter_table_clause ::=
     {:
         RESULT = new DropIndexClause(indexName, null, true);
     :}
-    | KW_ENABLE KW_FEATURE STRING_LITERAL:featureName
+    | KW_ENABLE KW_FEATURE STRING_LITERAL:featureName opt_enable_feature_properties:properties
     {:
-        RESULT = new EnableFeatureClause(featureName);
+        RESULT = new EnableFeatureClause(featureName, properties);
+    :}
+    ;
+
+opt_enable_feature_properties ::=
+    {:
+    RESULT = null;
+    :}
+    | KW_WITH KW_PROPERTIES LPAREN key_value_map:map RPAREN
+    {:
+    RESULT = map;
     :}
     ;
 

fe/fe-core/src/main/java/org/apache/doris/alter/MaterializedViewHandler.java

@@ -478,6 +478,9 @@ private List<Column> checkAndPrepareMaterializedView(CreateMaterializedViewStmt
         if (KeysType.UNIQUE_KEYS == olapTable.getKeysType() && olapTable.hasDeleteSign()) {
             newMVColumns.add(new Column(olapTable.getDeleteSignColumn()));
         }
+        if (KeysType.UNIQUE_KEYS == olapTable.getKeysType() && olapTable.hasSequenceCol()) {
+            newMVColumns.add(new Column(olapTable.getSequenceCol()));
+        }
         return newMVColumns;
     }
 
@@ -553,19 +556,6 @@ public List<Column> checkAndPrepareMaterializedView(AddRollupClause addRollupCla
                     }
                 }
             }
-            if (KeysType.UNIQUE_KEYS == keysType && olapTable.hasSequenceCol()) {
-                if (meetValue) {
-                    // check sequence column already exist in the rollup schema
-                    for (Column col : rollupSchema) {
-                        if (col.isSequenceColumn()) {
-                            throw new DdlException("sequence column already exist in the Rollup schema");
-                        }
-                    }
-                    // add the sequence column
-                    rollupSchema.add(new Column(Column.SEQUENCE_COL, olapTable.getSequenceType(),
-                            false, AggregateType.REPLACE, true, null, "", false));
-                }
-            }
         } else if (KeysType.DUP_KEYS == keysType) {
             // supplement the duplicate key
             if (addRollupClause.getDupKeys() == null || addRollupClause.getDupKeys().isEmpty()) {

fe/fe-core/src/main/java/org/apache/doris/analysis/AlterTableStmt.java

@@ -22,15 +22,18 @@
 import org.apache.doris.catalog.KeysType;
 import org.apache.doris.catalog.MaterializedIndex;
 import org.apache.doris.catalog.OlapTable;
+import org.apache.doris.catalog.Type;
 import org.apache.doris.common.AnalysisException;
 import org.apache.doris.common.ErrorCode;
 import org.apache.doris.common.ErrorReport;
 import org.apache.doris.common.UserException;
+import org.apache.doris.common.util.PropertyAnalyzer;
 import org.apache.doris.mysql.privilege.PrivPredicate;
 import org.apache.doris.qe.ConnectContext;
 
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Map;
 
 // Alter table statement.
 public class AlterTableStmt extends DdlStmt {
@@ -79,11 +82,31 @@ public void analyze(Analyzer analyzer) throws UserException {
     public void rewriteAlterClause(OlapTable table) throws UserException {
         List<AlterClause> clauses = new ArrayList<>();
         for (AlterClause alterClause : ops) {
-            if (alterClause instanceof EnableFeatureClause
-                    && ((EnableFeatureClause) alterClause).getFeature() == EnableFeatureClause.Features.BATCH_DELETE) {
-                if (table.getKeysType() != KeysType.UNIQUE_KEYS) {
+            if (alterClause instanceof EnableFeatureClause) {
+                EnableFeatureClause.Features alterFeature  = ((EnableFeatureClause) alterClause).getFeature();
+                if (alterFeature == null || alterFeature == EnableFeatureClause.Features.UNKNOWN) {
+                    throw new AnalysisException("unknown feature for alter clause");
+                }
+                if (table.getKeysType() != KeysType.UNIQUE_KEYS && alterFeature == EnableFeatureClause.Features.BATCH_DELETE) {
                     throw new AnalysisException("Batch delete only supported in unique tables.");
                 }
+                if (table.getKeysType() != KeysType.UNIQUE_KEYS && alterFeature == EnableFeatureClause.Features.SEQUENCE_LOAD) {
+                    throw new AnalysisException("Sequence load only supported in unique tables.");
+                }
+                // analyse sequence column
+                Type sequenceColType = null;
+                if (alterFeature == EnableFeatureClause.Features.SEQUENCE_LOAD) {
+                    Map<String, String> propertyMap = alterClause.getProperties();
+                    try {
+                        sequenceColType = PropertyAnalyzer.analyzeSequenceType(propertyMap, table.getKeysType());
+                        if (sequenceColType == null) {
+                            throw new AnalysisException("unknown sequence column type");
+                        }
+                    } catch (Exception e) {
+                        throw new AnalysisException(e.getMessage());
+                    }
+                }
+
                 // has rollup table
                 if (table.getVisibleIndex().size() > 1) {
                     for (MaterializedIndex idx : table.getVisibleIndex()) {
@@ -92,27 +115,50 @@ public void rewriteAlterClause(OlapTable table) throws UserException {
                         if (idx.getId() == table.getBaseIndexId()) {
                             continue;
                         }
-                        AddColumnClause addColumnClause = new AddColumnClause(ColumnDef.newDeleteSignColumnDef(), null,
-                                table.getIndexNameById(idx.getId()), null);
+                        AddColumnClause addColumnClause = null;
+                        if (alterFeature == EnableFeatureClause.Features.BATCH_DELETE) {
+                            addColumnClause = new AddColumnClause(ColumnDef.newDeleteSignColumnDef(), null,
+                                    table.getIndexNameById(idx.getId()), null);
+                        } else if (alterFeature == EnableFeatureClause.Features.SEQUENCE_LOAD) {
+                            addColumnClause = new AddColumnClause(ColumnDef.newSequenceColumnDef(sequenceColType), null,
+                                    table.getIndexNameById(idx.getId()), null);
+                        } else {
+                            throw new AnalysisException("unknown feature : " + alterFeature);
+                        }
                         addColumnClause.analyze(analyzer);
                         clauses.add(addColumnClause);
                     }
                 } else {
                     // no rollup tables
-                    AddColumnClause addColumnClause = new AddColumnClause(ColumnDef.newDeleteSignColumnDef(), null,
-                           null, null);
+                    AddColumnClause addColumnClause = null;
+                    if (alterFeature == EnableFeatureClause.Features.BATCH_DELETE) {
+                        addColumnClause = new AddColumnClause(ColumnDef.newDeleteSignColumnDef(), null,
+                                null, null);
+                    } else if (alterFeature == EnableFeatureClause.Features.SEQUENCE_LOAD) {
+                        addColumnClause = new AddColumnClause(ColumnDef.newSequenceColumnDef(sequenceColType), null,
+                                null, null);
+                    }
                     addColumnClause.analyze(analyzer);
                     clauses.add(addColumnClause);
                 }
             // add hidden column to rollup table
             } else if (alterClause instanceof AddRollupClause && table.getKeysType() == KeysType.UNIQUE_KEYS
-                    && table.getColumn(Column.DELETE_SIGN) != null) {
-                if (!((AddRollupClause) alterClause).getColumnNames()
-                        .stream()
-                        .anyMatch(x -> x.equalsIgnoreCase(Column.DELETE_SIGN))) {
-                    ((AddRollupClause) alterClause).getColumnNames().add(Column.DELETE_SIGN);
-                    alterClause.analyze(analyzer);
+                    && table.hasHiddenColumn()) {
+                if (table.getColumn(Column.DELETE_SIGN) != null) {
+                    if (!((AddRollupClause) alterClause).getColumnNames()
+                            .stream()
+                            .anyMatch(x -> x.equalsIgnoreCase(Column.DELETE_SIGN))) {
+                        ((AddRollupClause) alterClause).getColumnNames().add(Column.DELETE_SIGN);
+                    }
+                }
+                if (table.getColumn(Column.SEQUENCE_COL) != null) {
+                    if (!((AddRollupClause) alterClause).getColumnNames()
+                            .stream()
+                            .anyMatch(x -> x.equalsIgnoreCase(Column.SEQUENCE_COL))) {
+                        ((AddRollupClause) alterClause).getColumnNames().add(Column.SEQUENCE_COL);
+                    }
                 }
+                alterClause.analyze(analyzer);
                 clauses.add(alterClause);
             } else {
                 clauses.add(alterClause);

fe/fe-core/src/main/java/org/apache/doris/analysis/ColumnDef.java

@@ -118,6 +118,15 @@ public static ColumnDef newDeleteSignColumnDef(AggregateType aggregateType) {
                 new ColumnDef.DefaultValue(true, "0"), "doris delete flag hidden column", false);
     }
 
+    public static ColumnDef newSequenceColumnDef(Type type) {
+        return new ColumnDef(Column.SEQUENCE_COL, new TypeDef(type), false, null, true, DefaultValue.NULL_DEFAULT_VALUE,
+                "sequence column hidden column", false);
+    }
+
+    public static ColumnDef newSequenceColumnDef(Type type, AggregateType aggregateType) {
+        return new ColumnDef(Column.SEQUENCE_COL, new TypeDef(type), false, aggregateType, true, DefaultValue.NULL_DEFAULT_VALUE,
+                "sequence column hidden column", false);
+    }
 
     public boolean isAllowNull() { return isAllowNull; }
     public String getDefaultValue() { return defaultValue.value; }

fe/fe-core/src/main/java/org/apache/doris/analysis/EnableFeatureClause.java

@@ -21,25 +21,35 @@
 import org.apache.doris.common.AnalysisException;
 import org.apache.doris.common.UserException;
 
+import org.apache.doris.common.util.PrintableMap;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 
+import java.util.Map;
+
 public class EnableFeatureClause extends AlterTableClause {
     private static final Logger LOG = LogManager.getLogger(EnableFeatureClause.class);
 
     public enum Features {
         BATCH_DELETE,
+        SEQUENCE_LOAD,
         UNKNOWN
     }
 
     private String featureName;
     private boolean needSchemaChange;
     private Features feature;
+    private Map<String, String> properties;
 
     public EnableFeatureClause(String featureName) {
+        this(featureName, null);
+    }
+
+    public EnableFeatureClause(String featureName, Map<String, String> properties) {
         super(AlterOpType.ENABLE_FEATURE);
         this.featureName = featureName;
         this.needSchemaChange = false;
+        this.properties = properties;
     }
 
     public boolean needSchemaChange() {
@@ -50,13 +60,25 @@ public Features getFeature() {
         return feature;
     }
 
+    @Override
+    public Map<String, String> getProperties() {
+        return this.properties;
+    }
+
     @Override
     public void analyze(Analyzer analyzer) throws UserException {
         switch (featureName.toUpperCase()) {
             case "BATCH_DELETE":
                 this.needSchemaChange = true;
                 this.feature = Features.BATCH_DELETE;
                 break;
+            case  "SEQUENCE_LOAD":
+                this.needSchemaChange = true;
+                this.feature = Features.SEQUENCE_LOAD;
+                if (properties == null || properties.isEmpty()) {
+                    throw new AnalysisException("Properties is not set");
+                }
+                break;
             default:
                 throw new AnalysisException("unknown feature name: " + featureName);
         }
@@ -66,6 +88,11 @@ public void analyze(Analyzer analyzer) throws UserException {
     public String toSql() {
         StringBuilder sb = new StringBuilder();
         sb.append("ENABLE FEATURE \"").append(featureName).append("\"");
+        if (properties != null && !properties.isEmpty()) {
+            sb.append(" WITH PROPERTIES (");
+            sb.append(new PrintableMap<String, String>(properties, "=", true, false));
+            sb.append(")");
+        }
         return sb.toString();
     }
 }

fe/fe-core/src/main/java/org/apache/doris/catalog/OlapTable.java

@@ -19,6 +19,7 @@
 
 import org.apache.doris.alter.MaterializedViewHandler;
 import org.apache.doris.analysis.AggregateInfo;
+import org.apache.doris.analysis.ColumnDef;
 import org.apache.doris.analysis.CreateTableStmt;
 import org.apache.doris.analysis.Expr;
 import org.apache.doris.analysis.SlotDescriptor;
@@ -810,7 +811,7 @@ public void setSequenceInfo(Type type) {
         this.sequenceType = type;
 
         // sequence column is value column with REPLACE aggregate type
-        Column sequenceCol = new Column(Column.SEQUENCE_COL, type, false, AggregateType.REPLACE, true, null, "", false);
+        Column sequenceCol = ColumnDef.newSequenceColumnDef(type, AggregateType.REPLACE).toColumn();
         // add sequence column at last
         fullSchema.add(sequenceCol);
         nameToColumn.put(Column.SEQUENCE_COL, sequenceCol);
@@ -833,6 +834,10 @@ public Boolean hasSequenceCol() {
         return getSequenceCol() != null;
     }
 
+    public boolean hasHiddenColumn() {
+        return getBaseSchema().stream().anyMatch(column -> !column.isVisible());
+    }
+
     public Type getSequenceType() {
         if (getSequenceCol() == null) {
             return null;