[Feature](update) Support update on current_timestamp

docs/en/docs/sql-manual/sql-reference/Data-Definition-Statements/Create/CREATE-TABLE.md

@@ -58,7 +58,7 @@ Column definition list:
 
     Column definition:
 
-    `column_name column_type [KEY] [aggr_type] [NULL] [AUTO_INCREMENT] [default_value] [column_comment]`
+    `column_name column_type [KEY] [aggr_type] [NULL] [AUTO_INCREMENT] [default_value] [on update current_timestamp] [column_comment]`
 
     * `column_type`
 
@@ -142,6 +142,10 @@ Column definition list:
             dt DATETIME DEFAULT CURRENT_TIMESTAMP
         ```
 
+    * `on update current_timestamp`
+
+        To indicate that whether the value of this column should be updated to the current timestamp (`current_timestamp`) when there is an update on the row. The feature is only available on unique table with merge-on-write enabled. Columns with this feature enabled must declare a default value, and the default value must be `current_timestamp`. If the precision of the timestamp is declared here, the timestamp precision in the default value of the column must be the same as the precision declared here."
+
     Example:
 
         ```
@@ -152,6 +156,7 @@ Column definition list:
         v2 BITMAP BITMAP_UNION,
         v3 HLL HLL_UNION,
         v4 INT SUM NOT NULL DEFAULT "1" COMMENT "This is column v4"
+        dt datetime(6) default current_timestamp(6) on update current_timestamp(6)
         ```
 
 #### index_definition_list

docs/zh-CN/docs/sql-manual/sql-reference/Data-Definition-Statements/Create/CREATE-TABLE.md

@@ -56,7 +56,7 @@ distribution_desc
 * `column_definition`
     列定义：
 
-    `column_name column_type [KEY] [aggr_type] [NULL] [AUTO_INCREMENT] [default_value] [column_comment]`
+    `column_name column_type [KEY] [aggr_type] [NULL] [AUTO_INCREMENT] [default_value] [on update current_timestamp] [column_comment]`
     * `column_type`
         列类型，支持以下类型：
         ```
@@ -129,6 +129,10 @@ distribution_desc
             // 只用于DATETIME类型，导入数据缺失该值时系统将赋予当前时间
             dt DATETIME DEFAULT CURRENT_TIMESTAMP
         ```
+  * `on update current_timestamp`
+
+        是否在该行有列更新时将该列的值更新为当前时间(`current_timestamp`)。该特性只能在开启了merge-on-write的unique表上使用，开启了这个特性的列必须声明默认值，且默认值必须为`current_timestamp`。如果此处声明了时间戳的精度，则该列默认值中的时间戳精度必须与该处的时间戳精度相同。
+
 
   示例：
 
@@ -140,6 +144,7 @@ distribution_desc
   v2 BITMAP BITMAP_UNION,
   v3 HLL HLL_UNION,
   v4 INT SUM NOT NULL DEFAULT "1" COMMENT "This is column v4"
+  dt datetime(6) default current_timestamp(6) on update current_timestamp(6)
   ```
 
 #### index_definition_list

fe/fe-core/src/main/antlr4/org/apache/doris/nereids/DorisParser.g4

@@ -462,7 +462,8 @@ columnDef
     : colName=identifier type=dataType
         KEY? (aggType=aggTypeDef)? ((NOT NULL) | NULL)?
         (DEFAULT (nullValue=NULL | INTEGER_VALUE | stringValue=STRING_LITERAL
-            | CURRENT_TIMESTAMP (LEFT_PAREN precision=number RIGHT_PAREN)?))?
+            | CURRENT_TIMESTAMP (LEFT_PAREN defaultValuePrecision=number RIGHT_PAREN)?))?
+        (ON UPDATE CURRENT_TIMESTAMP (LEFT_PAREN onUpdateValuePrecision=number RIGHT_PAREN)?)?
         (COMMENT comment=STRING_LITERAL)?
     ;
 

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

@@ -130,6 +130,12 @@ public class Column implements Writable, GsonPostProcessable {
 
     private boolean isCompoundKey = false;
 
+    @SerializedName(value = "hasOnUpdateDefaultValue")
+    private boolean hasOnUpdateDefaultValue = false;
+
+    @SerializedName(value = "onUpdateDefaultValueExprDef")
+    private DefaultValueExprDef onUpdateDefaultValueExprDef;
+
     public Column() {
         this.name = "";
         this.type = Type.NULL;
@@ -170,24 +176,33 @@ public Column(String name, Type type, boolean isKey, AggregateType aggregateType
     public Column(String name, Type type, boolean isKey, AggregateType aggregateType, boolean isAllowNull,
             String defaultValue, String comment) {
         this(name, type, isKey, aggregateType, isAllowNull, false, defaultValue, comment, true, null,
-                COLUMN_UNIQUE_ID_INIT_VALUE, defaultValue);
+                COLUMN_UNIQUE_ID_INIT_VALUE, defaultValue, false, null);
     }
 
     public Column(String name, Type type, boolean isKey, AggregateType aggregateType, boolean isAllowNull,
             String comment, boolean visible, int colUniqueId) {
-        this(name, type, isKey, aggregateType, isAllowNull, false, null, comment, visible, null, colUniqueId, null);
+        this(name, type, isKey, aggregateType, isAllowNull, false, null, comment, visible, null, colUniqueId, null,
+                false, null);
     }
 
     public Column(String name, Type type, boolean isKey, AggregateType aggregateType, boolean isAllowNull,
             String defaultValue, String comment, boolean visible, DefaultValueExprDef defaultValueExprDef,
             int colUniqueId, String realDefaultValue) {
         this(name, type, isKey, aggregateType, isAllowNull, false, defaultValue, comment, visible, defaultValueExprDef,
-                colUniqueId, realDefaultValue);
+                colUniqueId, realDefaultValue, false, null);
     }
 
     public Column(String name, Type type, boolean isKey, AggregateType aggregateType, boolean isAllowNull,
             boolean isAutoInc, String defaultValue, String comment, boolean visible,
             DefaultValueExprDef defaultValueExprDef, int colUniqueId, String realDefaultValue) {
+        this(name, type, isKey, aggregateType, isAllowNull, isAutoInc, defaultValue, comment, visible,
+                defaultValueExprDef, colUniqueId, realDefaultValue, false, null);
+    }
+
+    public Column(String name, Type type, boolean isKey, AggregateType aggregateType, boolean isAllowNull,
+            boolean isAutoInc, String defaultValue, String comment, boolean visible,
+            DefaultValueExprDef defaultValueExprDef, int colUniqueId, String realDefaultValue,
+            boolean hasOnUpdateDefaultValue, DefaultValueExprDef onUpdateDefaultValueExprDef) {
         this.name = name;
         if (this.name == null) {
             this.name = "";
@@ -212,6 +227,8 @@ public Column(String name, Type type, boolean isKey, AggregateType aggregateType
         this.children = new ArrayList<>();
         createChildrenColumn(this.type, this);
         this.uniqueId = colUniqueId;
+        this.hasOnUpdateDefaultValue = hasOnUpdateDefaultValue;
+        this.onUpdateDefaultValueExprDef = onUpdateDefaultValueExprDef;
 
         if (type.isAggStateType()) {
             AggStateType aggState = (AggStateType) type;
@@ -244,6 +261,8 @@ public Column(Column column) {
         this.uniqueId = column.getUniqueId();
         this.defineExpr = column.getDefineExpr();
         this.defineName = column.getDefineName();
+        this.hasOnUpdateDefaultValue = column.hasOnUpdateDefaultValue;
+        this.onUpdateDefaultValueExprDef = column.onUpdateDefaultValueExprDef;
     }
 
     public void createChildrenColumn(Type type, Column column) {
@@ -489,6 +508,14 @@ public int getOlapColumnIndexSize() {
         }
     }
 
+    public boolean hasOnUpdateDefaultValue() {
+        return hasOnUpdateDefaultValue;
+    }
+
+    public Expr getOnUpdateDefaultValueExpr() {
+        return onUpdateDefaultValueExprDef.getExpr(type);
+    }
+
     public TColumn toThrift() {
         TColumn tColumn = new TColumn();
         tColumn.setColumnName(removeNamePrefix(this.name));
@@ -766,6 +793,9 @@ public String toSql(boolean isUniqueTable, boolean isCompatible) {
                 sb.append(" DEFAULT \"").append(defaultValue).append("\"");
             }
         }
+        if (hasOnUpdateDefaultValue) {
+            sb.append(" ON UPDATE ").append(defaultValue).append("");
+        }
         if (StringUtils.isNotBlank(comment)) {
             sb.append(" COMMENT '").append(getComment(true)).append("'");
         }

fe/fe-core/src/main/java/org/apache/doris/datasource/InternalCatalog.java

@@ -2166,6 +2166,16 @@ private void createOlapTable(Database db, CreateTableStmt stmt) throws UserExcep
         }
         olapTable.setEnableSingleReplicaCompaction(enableSingleReplicaCompaction);
 
+        // check `update on current_timestamp`
+        if (!enableUniqueKeyMergeOnWrite) {
+            for (Column column : baseSchema) {
+                if (column.hasOnUpdateDefaultValue()) {
+                    throw new DdlException("'ON UPDATE CURRENT_TIMESTAMP' is only supportted"
+                            + " in unique table with merge-on-write enabled.");
+                }
+            }
+        }
+
         // analyze bloom filter columns
         Set<String> bfColumns = null;
         double bfFpp = 0;

fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/PhysicalPlanTranslator.java

@@ -378,31 +378,9 @@ public PlanFragment visitPhysicalOlapTableSink(PhysicalOlapTableSink<? extends P
         HashSet<String> partialUpdateCols = new HashSet<>();
         boolean isPartialUpdate = olapTableSink.isPartialUpdate();
         if (isPartialUpdate) {
-            OlapTable olapTable = olapTableSink.getTargetTable();
-            if (!olapTable.getEnableUniqueKeyMergeOnWrite()) {
-                throw new AnalysisException("Partial update is only allowed in"
-                        + "unique table with merge-on-write enabled.");
-            }
-            for (Column col : olapTable.getFullSchema()) {
-                boolean exists = false;
-                for (Column insertCol : olapTableSink.getCols()) {
-                    if (insertCol.getName() != null && insertCol.getName().equals(col.getName())) {
-                        exists = true;
-                        break;
-                    }
-                }
-                if (col.isKey() && !exists) {
-                    throw new AnalysisException("Partial update should include all key columns, missing: "
-                            + col.getName());
-                }
-            }
             for (Column col : olapTableSink.getCols()) {
                 partialUpdateCols.add(col.getName());
             }
-            if (olapTable.hasSequenceCol() && olapTable.getSequenceMapCol() != null
-                        && partialUpdateCols.contains(olapTable.getSequenceMapCol())) {
-                partialUpdateCols.add(Column.SEQUENCE_COL);
-            }
         }
         TupleDescriptor olapTuple = context.generateTupleDesc();
         List<Column> targetTableColumns = olapTableSink.getTargetTable().getFullSchema();

fe/fe-core/src/main/java/org/apache/doris/nereids/parser/LogicalPlanBuilder.java

@@ -2148,6 +2148,7 @@ public ColumnDefinition visitColumnDef(ColumnDefContext ctx) {
         boolean isNotNull = ctx.NOT() != null;
         String aggTypeString = ctx.aggType != null ? ctx.aggType.getText() : null;
         Optional<DefaultValue> defaultValue = Optional.empty();
+        Optional<DefaultValue> onUpdateDefaultValue = Optional.empty();
         if (ctx.DEFAULT() != null) {
             if (ctx.INTEGER_VALUE() != null) {
                 defaultValue = Optional.of(new DefaultValue(ctx.INTEGER_VALUE().getText()));
@@ -2156,14 +2157,24 @@ public ColumnDefinition visitColumnDef(ColumnDefContext ctx) {
             } else if (ctx.nullValue != null) {
                 defaultValue = Optional.of(DefaultValue.NULL_DEFAULT_VALUE);
             } else if (ctx.CURRENT_TIMESTAMP() != null) {
-                if (ctx.precision == null) {
+                if (ctx.defaultValuePrecision == null) {
                     defaultValue = Optional.of(DefaultValue.CURRENT_TIMESTAMP_DEFAULT_VALUE);
                 } else {
                     defaultValue = Optional.of(DefaultValue
-                            .currentTimeStampDefaultValueWithPrecision(Long.valueOf(ctx.precision.getText())));
+                            .currentTimeStampDefaultValueWithPrecision(
+                                    Long.valueOf(ctx.defaultValuePrecision.getText())));
                 }
             }
         }
+        if (ctx.UPDATE() != null) {
+            if (ctx.onUpdateValuePrecision == null) {
+                onUpdateDefaultValue = Optional.of(DefaultValue.CURRENT_TIMESTAMP_DEFAULT_VALUE);
+            } else {
+                onUpdateDefaultValue = Optional.of(DefaultValue
+                        .currentTimeStampDefaultValueWithPrecision(
+                                Long.valueOf(ctx.onUpdateValuePrecision.getText())));
+            }
+        }
         AggregateType aggType = null;
         if (aggTypeString != null) {
             try {
@@ -2174,7 +2185,8 @@ public ColumnDefinition visitColumnDef(ColumnDefContext ctx) {
             }
         }
         String comment = ctx.comment != null ? ctx.comment.getText() : "";
-        return new ColumnDefinition(colName, colType, isKey, aggType, !isNotNull, defaultValue, comment);
+        return new ColumnDefinition(colName, colType, isKey, aggType, !isNotNull, defaultValue,
+                onUpdateDefaultValue, comment);
     }
 
     @Override