[Fix](delete command) Mark delete sign when do delete command in MoW table #35917

fe/fe-common/src/main/java/org/apache/doris/common/Config.java

@@ -2395,6 +2395,15 @@ public class Config extends ConfigBase {
     @ConfField
     public static boolean ignore_bdbje_log_checksum_read = false;
 
+    @ConfField(mutable = true, masterOnly = true, description = {
+            "是否在unique表mow上开启delete语句写delete predicate。若开启，会提升delete语句的性能，"
+                    + "但delete后进行部分列更新可能会出现部分数据错误的情况。若关闭，会降低delete语句的性能来保证正确性。",
+            "Enable the 'delete predicate' for DELETE statements. If enabled, it will enhance the performance of "
+                    + "DELETE statements, but partial column updates after a DELETE may result in erroneous data. "
+                    + "If disabled, it will reduce the performance of DELETE statements to ensure accuracy."
+    })
+    public static boolean enable_mow_light_delete = false;
+
     @ConfField(description = {
             "是否开启 Proxy Protocol 支持",
             "Whether to enable proxy protocol"

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

@@ -513,6 +513,7 @@ public void processAlterTable(AlterTableStmt stmt) throws UserException {
                 // currently, only in memory and storage policy property could reach here
                 Preconditions.checkState(properties.containsKey(PropertyAnalyzer.PROPERTIES_INMEMORY)
                         || properties.containsKey(PropertyAnalyzer.PROPERTIES_STORAGE_POLICY)
+                        || properties.containsKey(PropertyAnalyzer.PROPERTIES_ENABLE_MOW_LIGHT_DELETE)
                         || properties.containsKey(PropertyAnalyzer.PROPERTIES_IS_BEING_SYNCED)
                         || properties.containsKey(PropertyAnalyzer.PROPERTIES_COMPACTION_POLICY)
                         || properties.containsKey(PropertyAnalyzer.PROPERTIES_TIME_SERIES_COMPACTION_GOAL_SIZE_MBYTES)

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

@@ -2215,6 +2215,7 @@ public void updateTableProperties(Database db, String tableName, Map<String, Str
 
         if (isInMemory < 0 && storagePolicyId < 0 && compactionPolicy == null && timeSeriesCompactionConfig.isEmpty()
                 && !properties.containsKey(PropertyAnalyzer.PROPERTIES_IS_BEING_SYNCED)
+                && !properties.containsKey(PropertyAnalyzer.PROPERTIES_ENABLE_MOW_LIGHT_DELETE)
                 && !properties.containsKey(PropertyAnalyzer.PROPERTIES_ENABLE_SINGLE_REPLICA_COMPACTION)
                 && !properties.containsKey(PropertyAnalyzer.PROPERTIES_SKIP_WRITE_INDEX_ON_LOAD)) {
             LOG.info("Properties already up-to-date");
@@ -2232,6 +2233,13 @@ public void updateTableProperties(Database db, String tableName, Map<String, Str
                     "enable_single_replica_compaction property is not supported for merge-on-write table");
         }
 
+        String enableMowLightDelete = properties.get(
+                PropertyAnalyzer.PROPERTIES_ENABLE_MOW_LIGHT_DELETE);
+        if (enableMowLightDelete != null && !enableUniqueKeyMergeOnWrite) {
+            throw new UserException(
+                    "enable_mow_light_delete property is only supported for unique merge-on-write table");
+        }
+
         String skipWriteIndexOnLoad = properties.get(PropertyAnalyzer.PROPERTIES_SKIP_WRITE_INDEX_ON_LOAD);
         int skip = -1; // < 0 means don't update
         if (skipWriteIndexOnLoad != null) {

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

@@ -123,7 +123,8 @@ public void analyze(Analyzer analyzer) throws UserException {
         }
 
         // analyze predicate
-        if (fromClause == null) {
+        if ((fromClause == null && !((OlapTable) targetTable).getEnableUniqueKeyMergeOnWrite())
+                || (fromClause == null && ((OlapTable) targetTable).getEnableMowLightDelete())) {
             if (wherePredicate == null) {
                 throw new AnalysisException("Where clause is not set");
             }

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

@@ -18,6 +18,8 @@
 package org.apache.doris.analysis;
 
 import org.apache.doris.alter.AlterOpType;
+import org.apache.doris.catalog.Env;
+import org.apache.doris.catalog.OlapTable;
 import org.apache.doris.catalog.ReplicaAllocation;
 import org.apache.doris.catalog.TableProperty;
 import org.apache.doris.common.AnalysisException;
@@ -254,6 +256,39 @@ public void analyze(Analyzer analyzer) throws AnalysisException {
             }
             this.needTableStable = false;
             this.opType = AlterOpType.MODIFY_TABLE_PROPERTY_SYNC;
+        } else if (properties.containsKey(PropertyAnalyzer.PROPERTIES_ENABLE_MOW_LIGHT_DELETE)) {
+            if (!properties.get(PropertyAnalyzer.PROPERTIES_ENABLE_MOW_LIGHT_DELETE)
+                    .equalsIgnoreCase("true")
+                    && !properties.get(PropertyAnalyzer
+                    .PROPERTIES_ENABLE_MOW_LIGHT_DELETE).equalsIgnoreCase("false")) {
+                throw new AnalysisException(
+                        "Property "
+                                + PropertyAnalyzer.PROPERTIES_ENABLE_MOW_LIGHT_DELETE
+                                + " should be set to true or false");
+            }
+            OlapTable table = null;
+            if (tableName != null) {
+                table = (OlapTable) (Env.getCurrentInternalCatalog().getDbOrAnalysisException(tableName.getDb())
+                        .getTableOrAnalysisException(tableName.getTbl()));
+            }
+            if (table == null || !table.getEnableUniqueKeyMergeOnWrite()) {
+                throw new AnalysisException(
+                        "enable_mow_light_delete property is "
+                                + "only supported for unique merge-on-write table");
+            }
+            this.needTableStable = false;
+            this.opType = AlterOpType.MODIFY_TABLE_PROPERTY_SYNC;
+        } else if (properties.containsKey(PropertyAnalyzer.PROPERTIES_DISABLE_AUTO_COMPACTION)) {
+            if (!properties.get(PropertyAnalyzer.PROPERTIES_DISABLE_AUTO_COMPACTION).equalsIgnoreCase("true")
+                    && !properties.get(PropertyAnalyzer
+                    .PROPERTIES_DISABLE_AUTO_COMPACTION).equalsIgnoreCase("false")) {
+                throw new AnalysisException(
+                        "Property "
+                                + PropertyAnalyzer.PROPERTIES_DISABLE_AUTO_COMPACTION
+                                + " should be set to true or false");
+            }
+            this.needTableStable = false;
+            this.opType = AlterOpType.MODIFY_TABLE_PROPERTY_SYNC;
         } else if (properties.containsKey(PropertyAnalyzer.PROPERTIES_ESTIMATE_PARTITION_SIZE)) {
             throw new AnalysisException("You can not modify estimate partition size");
         } else {

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

@@ -3304,6 +3304,10 @@ public static void getDdlStmt(DdlStmt ddlStmt, String dbName, TableIf table, Lis
                 sb.append(olapTable.isDuplicateWithoutKey()).append("\"");
             }
 
+            // enable mow light delete
+            sb.append(",\n\"").append(PropertyAnalyzer.PROPERTIES_ENABLE_MOW_LIGHT_DELETE).append("\" = \"");
+            sb.append(olapTable.getEnableMowLightDelete()).append("\"");
+
             sb.append("\n)");
         } else if (table.getType() == TableType.MYSQL) {
             MysqlTable mysqlTable = (MysqlTable) table;

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

@@ -1126,6 +1126,14 @@ public Column getRowStoreCol() {
         return null;
     }
 
+    public void setEnableMowLightDelete(boolean enable) {
+        getOrCreatTableProperty().setEnableMowLightDelete(enable);
+    }
+
+    public boolean getEnableMowLightDelete() {
+        return getOrCreatTableProperty().getEnableMowLightDelete();
+    }
+
     public Boolean hasSequenceCol() {
         return getSequenceCol() != null;
     }

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

@@ -513,6 +513,16 @@ public boolean getEnableUniqueKeyMergeOnWrite() {
                 PropertyAnalyzer.ENABLE_UNIQUE_KEY_MERGE_ON_WRITE, "false"));
     }
 
+    public void setEnableMowLightDelete(boolean enable) {
+        properties.put(PropertyAnalyzer.PROPERTIES_ENABLE_MOW_LIGHT_DELETE, Boolean.toString(enable));
+    }
+
+    public boolean getEnableMowLightDelete() {
+        return Boolean.parseBoolean(properties.getOrDefault(
+                PropertyAnalyzer.PROPERTIES_ENABLE_MOW_LIGHT_DELETE,
+                Boolean.toString(PropertyAnalyzer.PROPERTIES_ENABLE_MOW_LIGHT_DELETE_DEFAULT_VALUE)));
+    }
+
     public void setSequenceMapCol(String colName) {
         properties.put(PropertyAnalyzer.PROPERTIES_FUNCTION_COLUMN + "."
                 + PropertyAnalyzer.PROPERTIES_SEQUENCE_COL, colName);

fe/fe-core/src/main/java/org/apache/doris/common/util/PropertyAnalyzer.java

@@ -171,6 +171,11 @@ public class PropertyAnalyzer {
     private static final double MAX_FPP = 0.05;
     private static final double MIN_FPP = 0.0001;
 
+    public static final String PROPERTIES_ENABLE_MOW_LIGHT_DELETE =
+            "enable_mow_light_delete";
+    public static final boolean PROPERTIES_ENABLE_MOW_LIGHT_DELETE_DEFAULT_VALUE
+            = Config.enable_mow_light_delete;
+
     // compaction policy
     public static final String SIZE_BASED_COMPACTION_POLICY = "size_based";
     public static final String TIME_SERIES_COMPACTION_POLICY = "time_series";
@@ -1178,6 +1183,25 @@ public static boolean analyzeUniqueKeyMergeOnWrite(Map<String, String> propertie
         throw new AnalysisException(PropertyAnalyzer.ENABLE_UNIQUE_KEY_MERGE_ON_WRITE + " must be `true` or `false`");
     }
 
+    public static boolean analyzeEnableDeleteOnDeletePredicate(Map<String, String> properties)
+            throws AnalysisException {
+        if (properties == null || properties.isEmpty()) {
+            return false;
+        }
+        String value = properties.get(PropertyAnalyzer.PROPERTIES_ENABLE_MOW_LIGHT_DELETE);
+        if (value == null) {
+            return false;
+        }
+        properties.remove(PropertyAnalyzer.PROPERTIES_ENABLE_MOW_LIGHT_DELETE);
+        if (value.equals("true")) {
+            return true;
+        } else if (value.equals("false")) {
+            return false;
+        }
+        throw new AnalysisException(
+                PropertyAnalyzer.PROPERTIES_ENABLE_MOW_LIGHT_DELETE + " must be `true` or `false`");
+    }
+
     /**
      * Check the type property of the catalog props.
      */

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

@@ -2203,6 +2203,18 @@ private boolean createOlapTable(Database db, CreateTableStmt stmt) throws UserEx
         }
         olapTable.setEnableUniqueKeyMergeOnWrite(enableUniqueKeyMergeOnWrite);
 
+        boolean enableDeleteOnDeletePredicate = false;
+        try {
+            enableDeleteOnDeletePredicate = PropertyAnalyzer.analyzeEnableDeleteOnDeletePredicate(properties);
+        } catch (AnalysisException e) {
+            throw new DdlException(e.getMessage());
+        }
+        if (enableDeleteOnDeletePredicate && !enableUniqueKeyMergeOnWrite) {
+            throw new DdlException(PropertyAnalyzer.PROPERTIES_ENABLE_MOW_LIGHT_DELETE
+                    + " property is only supported for unique merge-on-write table");
+        }
+        olapTable.setEnableMowLightDelete(enableDeleteOnDeletePredicate);
+
         boolean enableSingleReplicaCompaction = false;
         try {
             enableSingleReplicaCompaction = PropertyAnalyzer.analyzeEnableSingleReplicaCompaction(properties);

fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/DeleteCommand.java

@@ -21,10 +21,10 @@
 import org.apache.doris.catalog.KeysType;
 import org.apache.doris.catalog.OlapTable;
 import org.apache.doris.catalog.TableIf;
+import org.apache.doris.nereids.analyzer.UnboundAlias;
 import org.apache.doris.nereids.analyzer.UnboundOlapTableSink;
 import org.apache.doris.nereids.analyzer.UnboundSlot;
 import org.apache.doris.nereids.exceptions.AnalysisException;
-import org.apache.doris.nereids.trees.expressions.Alias;
 import org.apache.doris.nereids.trees.expressions.NamedExpression;
 import org.apache.doris.nereids.trees.expressions.literal.TinyIntLiteral;
 import org.apache.doris.nereids.trees.plans.Explainable;
@@ -98,15 +98,16 @@ public LogicalPlan completeQueryPlan(ConnectContext ctx, LogicalPlan logicalQuer
         String tableName = tableAlias != null ? tableAlias : targetTable.getName();
         for (Column column : targetTable.getFullSchema()) {
             if (column.getName().equalsIgnoreCase(Column.DELETE_SIGN)) {
-                selectLists.add(new Alias(new TinyIntLiteral(((byte) 1)), Column.DELETE_SIGN));
-            } else if (column.getName().equalsIgnoreCase(Column.SEQUENCE_COL)) {
+                selectLists.add(new UnboundAlias(new TinyIntLiteral(((byte) 1)), Column.DELETE_SIGN));
+            } else if (column.getName().equalsIgnoreCase(Column.SEQUENCE_COL)
+                    && targetTable.getSequenceMapCol() != null) {
                 selectLists.add(new UnboundSlot(tableName, targetTable.getSequenceMapCol()));
             } else if (column.isKey()) {
                 selectLists.add(new UnboundSlot(tableName, column.getName()));
             } else if (!isMow && (!column.isVisible() || (!column.isAllowNull() && !column.hasDefaultValue()))) {
                 selectLists.add(new UnboundSlot(tableName, column.getName()));
             } else {
-                continue;
+                selectLists.add(new UnboundSlot(tableName, column.getName()));
             }
             cols.add(column.getName());
         }