[draft](ivm) for run pipeline test

be/test/storage/segment/row_binlog_source_data_writer_test.cpp

@@ -0,0 +1,118 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include <gtest/gtest.h>
+
+#include <memory>
+#include <string>
+#include <vector>
+
+#include "core/field.h"
+#include "core/value/decimalv2_value.h"
+#include "storage/binlog.h"
+#include "storage/iterator/olap_data_convertor.h"
+#include "storage/segment/row_binlog_segment_writer.h"
+#include "storage/tablet/tablet_schema.h"
+
+namespace doris::segment_v2 {
+
+namespace {
+
+TabletColumn create_column(int32_t unique_id, const std::string& name, FieldType type, bool is_key,
+                           bool visible) {
+    TabletColumn column(FieldAggregationMethod::OLAP_FIELD_AGGREGATION_NONE, type, !is_key);
+    column.set_unique_id(unique_id);
+    column.set_name(name);
+    column.set_is_key(is_key);
+    column.set_length(type == FieldType::OLAP_FIELD_TYPE_LARGEINT ? 16 : 4);
+    column.set_index_length(column.length());
+    column._visible = visible;
+    return column;
+}
+
+TabletSchemaSPtr create_ivm_source_schema() {
+    auto schema = std::make_shared<TabletSchema>();
+    schema->append_column(create_column(0, "k1", FieldType::OLAP_FIELD_TYPE_INT, true, true));
+    schema->append_column(create_column(1, "__DORIS_IVM_ROW_ID_COL__",
+                                        FieldType::OLAP_FIELD_TYPE_LARGEINT, true, false));
+    schema->append_column(create_column(2, "v1", FieldType::OLAP_FIELD_TYPE_INT, false, true));
+    schema->append_column(create_column(3, "__DORIS_IVM_AGG_STATE_COL__",
+                                        FieldType::OLAP_FIELD_TYPE_INT, false, false));
+    schema->_keys_type = UNIQUE_KEYS;
+    return schema;
+}
+
+Block create_source_block(const TabletSchemaSPtr& schema) {
+    Block block = schema->create_block();
+    auto columns_guard = block.mutate_columns_scoped();
+    auto& columns = columns_guard.mutable_columns();
+    for (int i = 0; i < 2; ++i) {
+        columns[0]->insert(Field::create_field<TYPE_INT>(10 + i));
+        columns[1]->insert(Field::create_field<TYPE_LARGEINT>(static_cast<int128_t>(1000 + i)));
+        columns[2]->insert(Field::create_field<TYPE_INT>(100 + i));
+        columns[3]->insert(Field::create_field<TYPE_INT>(10000 + i));
+    }
+    return block;
+}
+
+} // namespace
+
+TEST(RowBinlogSourceDataWriterTest, collectHiddenRowIdKeyInNormalPrefix) {
+    auto source_schema = create_ivm_source_schema();
+    SegmentWriteBinlogOptions options;
+    options.source.tablet_schema = source_schema;
+
+    RowBinlogSourceDataWriter writer(options);
+    ASSERT_TRUE(writer.init().ok());
+    EXPECT_EQ(3, writer.normal_column_count());
+
+    Block block = create_source_block(source_schema);
+    Block full_block = source_schema->create_block();
+    std::vector<uint32_t> partial_source_cids;
+    ASSERT_TRUE(
+            writer.prepare_by_source_block(&block, 0, 2, partial_source_cids, &full_block).ok());
+
+    const auto& key_columns = writer.source_key_columns();
+    ASSERT_EQ(2, key_columns.size());
+
+    ASSERT_NE(nullptr, key_columns[0]->get_data_at(1));
+    EXPECT_EQ(11, *reinterpret_cast<const int32_t*>(key_columns[0]->get_data_at(1)));
+
+    ASSERT_NE(nullptr, key_columns[1]->get_data_at(1));
+    EXPECT_EQ(static_cast<int128_t>(1001),
+              *reinterpret_cast<const int128_t*>(key_columns[1]->get_data_at(1)));
+}
+
+TEST(RowBinlogSourceDataWriterTest, collectVisibleColumnAfterHiddenNonKeyColumn) {
+    auto source_schema = std::make_shared<TabletSchema>();
+    source_schema->append_column(
+            create_column(0, "k1", FieldType::OLAP_FIELD_TYPE_INT, true, true));
+    source_schema->append_column(create_column(1, "__DORIS_IVM_AGG_STATE_COL__",
+                                               FieldType::OLAP_FIELD_TYPE_INT, false, false));
+    source_schema->append_column(
+            create_column(2, "v1", FieldType::OLAP_FIELD_TYPE_INT, false, true));
+    source_schema->_keys_type = UNIQUE_KEYS;
+
+    SegmentWriteBinlogOptions options;
+    options.source.tablet_schema = source_schema;
+
+    RowBinlogSourceDataWriter writer(options);
+    ASSERT_TRUE(writer.init().ok());
+    EXPECT_EQ(2, writer.normal_column_count());
+}
+
+} // namespace doris::segment_v2

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

@@ -51,11 +51,27 @@ public class Column implements GsonPostProcessable {
     public static final String HIDDEN_COLUMN_PREFIX = "__DORIS_";
     // all shadow indexes should have this prefix in name
     public static final String SHADOW_NAME_PREFIX = "__doris_shadow_";
+    public static final String IVM_HIDDEN_COLUMN_PREFIX = "__DORIS_IVM_";
     // NOTE: you should name hidden column start with '__DORIS_' !!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!
     public static final String DELETE_SIGN = "__DORIS_DELETE_SIGN__";
     public static final String WHERE_SIGN = "__DORIS_WHERE_SIGN__";
     public static final String SEQUENCE_COL = "__DORIS_SEQUENCE_COL__";
     public static final String GLOBAL_ROWID_COL = "__DORIS_GLOBAL_ROWID_COL__";
+    public static final String IVM_ROW_ID_COL = "__DORIS_IVM_ROW_ID_COL__";
+    public static final String IVM_AGG_COUNT_COL = "__DORIS_IVM_AGG_COUNT_COL__";
+    public static final String IVM_DML_FACTOR_COL = "__DORIS_IVM_DML_FACTOR_COL__";
+    public static final String IVM_BASE_OP_COL = "__DORIS_IVM_BASE_OP_COL__";
+    public static final String IVM_DELTA_GROUP_COUNT_COL = "__DORIS_IVM_DELTA_GROUP_COUNT_COL__";
+    // Prefix for sink-level IVM identity key hidden columns (__DORIS_IVM_KEY_).
+    public static final String IVM_KEY_COL_PREFIX = IVM_HIDDEN_COLUMN_PREFIX + "KEY_";
+    // Prefix for union arm-index columns (__DORIS_IVM_UNION_ARM_INDEX_).
+    public static final String IVM_UNION_ARM_INDEX_COL_PREFIX = IVM_HIDDEN_COLUMN_PREFIX + "UNION_ARM_INDEX_";
+    // Prefix for union positional key columns (__DORIS_IVM_UNION_KEY_).
+    public static final String IVM_UNION_KEY_COL_PREFIX = IVM_HIDDEN_COLUMN_PREFIX + "UNION_KEY_";
+    // Suffix for the renamed base-table row-id column (__DORIS_IVM_{n}_ROW_ID_COL__), using the
+    // standard IVM hidden-column prefix. Kept as an identity key when a cascading MV with only
+    // row-id keys is scanned under a join.
+    public static final String IVM_BASE_ROW_ID_COL_SUFFIX = "_ROW_ID_COL__";
     public static final String ROW_STORE_COL = "__DORIS_ROW_STORE_COL__";
     public static final String VERSION_COL = "__DORIS_VERSION_COL__";
     public static final String SKIP_BITMAP_COL = "__DORIS_SKIP_BITMAP_COL__";
@@ -285,6 +301,7 @@ public Column(String name, Type type, boolean isKey, AggregateType aggregateType
                 false, null, null,  Sets.newHashSet(), null);
     }
 
+
     public Column(String name, Type type, boolean isKey, AggregateType aggregateType, boolean isAllowNull,
                   String defaultValue, String comment, boolean visible, int colUniqueId) {
         this(name, type, isKey, aggregateType, isAllowNull, -1, defaultValue, comment, visible, null, colUniqueId, null,

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

@@ -35,6 +35,7 @@
 import org.apache.doris.catalog.Tablet;
 import org.apache.doris.catalog.View;
 import org.apache.doris.catalog.info.TableNameInfo;
+import org.apache.doris.catalog.stream.BaseTableStream;
 import org.apache.doris.cloud.alter.CloudSchemaChangeHandler;
 import org.apache.doris.common.AnalysisException;
 import org.apache.doris.common.Config;
@@ -49,9 +50,15 @@
 import org.apache.doris.datasource.ExternalTable;
 import org.apache.doris.info.TableNameInfoUtils;
 import org.apache.doris.mtmv.BaseTableInfo;
+import org.apache.doris.mtmv.MTMVPartitionUtil;
+import org.apache.doris.mtmv.MTMVPropertyUtil;
+import org.apache.doris.mtmv.MTMVRelation;
+import org.apache.doris.mtmv.MTMVUtil;
+import org.apache.doris.mtmv.ivm.IvmUtil;
 import org.apache.doris.nereids.trees.plans.commands.AlterSystemCommand;
 import org.apache.doris.nereids.trees.plans.commands.AlterTableCommand;
 import org.apache.doris.nereids.trees.plans.commands.AlterViewCommand;
+import org.apache.doris.nereids.trees.plans.commands.CreateMTMVCommand;
 import org.apache.doris.nereids.trees.plans.commands.CreateMaterializedViewCommand;
 import org.apache.doris.nereids.trees.plans.commands.DropMaterializedViewCommand;
 import org.apache.doris.nereids.trees.plans.commands.info.AddColumnOp;
@@ -359,7 +366,6 @@ private boolean processAlterOlapTableInternal(List<AlterOp> alterOps, OlapTable
             Env.getCurrentEnv().getMtmvService()
                 .alterTable(oldBaseTableInfo, newBaseTableInfo, currentAlterOps.hasReplaceTableOp());
         }
-
         olapTable.writeLock();
         try {
             NereidsSqlCacheManager sqlCacheManager = Env.getCurrentEnv().getSqlCacheManager();
@@ -678,6 +684,12 @@ public void processReplaceTable(Database db, OlapTable origTable, String newTblN
                 if (swapTable) {
                     origTable.checkAndSetName(newTblName, true);
                 }
+                // REPLACE TABLE affects dependencies of both logical names. Persist both
+                // barriers before changing the catalog or journaling the replacement.
+                Env.getCurrentEnv().getMtmvService().getRelationManager().markIvmBaselineRebuild(
+                        new BaseTableInfo(origTable), "Base table replace changed IVM baseline");
+                Env.getCurrentEnv().getMtmvService().getRelationManager().markIvmBaselineRebuild(
+                        new BaseTableInfo(olapNewTbl), "Base table replace changed IVM baseline");
                 replaceTableInternal(db, origTable, olapNewTbl, swapTable, false, isForce);
                 // write edit log
                 ReplaceTableOperationLog log = new ReplaceTableOperationLog(db.getId(),
@@ -743,6 +755,10 @@ private void replaceTableInternal(Database db, OlapTable origTable, OlapTable ne
             if (!isReplay) {
                 Env.getCurrentEnv().getConstraintManager().checkNoReferencingForeignKeys(origTableInfo);
             }
+            if (origTable instanceof MTMV) {
+                Env.getCurrentInternalCatalog().unprotectDropIvmStreams(
+                        db, (MTMV) origTable, isForce, isReplay);
+            }
             Env.getCurrentEnv().getConstraintManager().dropAndRenameConstraints(origTableInfo, newTableInfo);
         }
 
@@ -1267,7 +1283,6 @@ public AlterHandler getSystemHandler() {
     public void processAlterMTMV(AlterMTMV alterMTMV, boolean isReplay) {
         TableNameInfo tbl = alterMTMV.getMvName();
         MTMV mtmv = null;
-        boolean alterSuccess = true;
         try {
             Database db = Env.getCurrentInternalCatalog().getDbOrDdlException(tbl.getDb());
             mtmv = (MTMV) db.getTableOrMetaException(tbl.getTbl(), TableType.MATERIALIZED_VIEW);
@@ -1279,31 +1294,103 @@ public void processAlterMTMV(AlterMTMV alterMTMV, boolean isReplay) {
                     mtmv.alterStatus(alterMTMV.getStatus());
                     break;
                 case ALTER_PROPERTY:
-                    mtmv.alterMvProperties(alterMTMV.getMvProperties());
-                    break;
+                    if (mtmv.isIvm() && alterMTMV.getMvProperties().containsKey(
+                            PropertyAnalyzer.PROPERTIES_EXCLUDED_TRIGGER_TABLES)) {
+                        Set<TableNameInfo> oldExcludedTriggerTables = mtmv.getExcludedTriggerTables();
+                        Set<TableNameInfo> newExcludedTriggerTables = MTMVPropertyUtil.parseTableNameInfos(
+                                alterMTMV.getMvProperties().get(
+                                        PropertyAnalyzer.PROPERTIES_EXCLUDED_TRIGGER_TABLES));
+                        updateIvmStreamsForExcludedTables(db, mtmv, oldExcludedTriggerTables,
+                                newExcludedTriggerTables, isReplay);
+                    }
+                    mtmv.alterMvProperties(alterMTMV, isReplay);
+                    return;
                 case ADD_TASK:
-                    alterSuccess = mtmv.addTaskResult(alterMTMV.getTask(), alterMTMV.getRelation(),
-                            alterMTMV.getPartitionSnapshots(),
-                            isReplay);
+                    if (!mtmv.addTaskResult(alterMTMV, isReplay)) {
+                        return;
+                    }
                     // If it is not a replay thread, it means that the current service is already a new version
                     // and does not require compatibility
                     if (isReplay) {
                         mtmv.compatible(Env.getCurrentEnv().getCatalogMgr());
                     }
+                    return;
+                case ALTER_IVM_INFO:
+                    // Live IVM changes are journaled inside MTMV; this branch applies the journal snapshot.
+                    mtmv.alterIvmInfo(alterMTMV.getIvmInfo());
                     break;
                 default:
                     throw new RuntimeException("Unknown type value: " + alterMTMV.getOpType());
             }
             if (alterMTMV.isNeedRebuildJob()) {
                 Env.getCurrentEnv().getMtmvService().alterJob(mtmv, isReplay);
             }
-            // 4. log it and replay it in the follower
-            if (!isReplay && alterSuccess) {
+            if (!isReplay) {
                 Env.getCurrentEnv().getEditLog().logAlterMTMV(alterMTMV);
             }
         } catch (UserException e) {
             // if MTMV has been dropped, ignore this exception
             LOG.warn(e);
         }
     }
+
+    private void updateIvmStreamsForExcludedTables(Database db, MTMV mtmv,
+            Set<TableNameInfo> oldExcludedTriggerTables, Set<TableNameInfo> newExcludedTriggerTables,
+            boolean isReplay) throws UserException {
+        MTMVRelation relation = mtmv.getRelation();
+        if (relation == null || relation.getBaseTables() == null) {
+            return;
+        }
+        Set<BaseTableInfo> baseTables = relation.getBaseTables();
+        for (BaseTableInfo baseTableInfo : baseTables) {
+            TableNameInfo baseTableName = new TableNameInfo(baseTableInfo.getCtlName(),
+                    baseTableInfo.getDbName(), baseTableInfo.getTableName());
+            boolean wasExcluded = MTMVPartitionUtil.isTableExcluded(
+                    oldExcludedTriggerTables, baseTableName);
+            boolean isExcluded = MTMVPartitionUtil.isTableExcluded(
+                    newExcludedTriggerTables, baseTableName);
+            if (wasExcluded && !isExcluded) {
+                if (!isReplay) {
+                    TableIf baseTable = MTMVUtil.getTable(baseTableInfo);
+                    CreateMTMVCommand.createTableStream(ConnectContext.get(), db, mtmv, baseTable);
+                }
+            }
+        }
+        db.writeLockOrDdlException();
+        try {
+            for (BaseTableInfo baseTableInfo : baseTables) {
+                TableNameInfo baseTableName = new TableNameInfo(baseTableInfo.getCtlName(),
+                        baseTableInfo.getDbName(), baseTableInfo.getTableName());
+                if (!MTMVPartitionUtil.isTableExcluded(newExcludedTriggerTables, baseTableName)) {
+                    continue;
+                }
+                List<String> baseTableFullQualifiers = baseTableInfo.toList();
+                String streamName = IvmUtil.streamName(mtmv.getId(), baseTableFullQualifiers);
+                TableIf streamTable = db.getTableNullable(streamName);
+                if (streamTable == null) {
+                    continue;
+                }
+                if (!(streamTable instanceof BaseTableStream)) {
+                    LOG.warn("skip dropping IVM stream candidate {} because it is not a stream", streamName);
+                    continue;
+                }
+                if (!IvmUtil.isStreamOwnedBy((BaseTableStream) streamTable, baseTableFullQualifiers)) {
+                    LOG.warn("skip dropping IVM stream candidate {} because it belongs to another base table",
+                            streamName);
+                    continue;
+                }
+                Table table = (Table) streamTable;
+                table.writeLock();
+                try {
+                    Env.getCurrentEnv().unprotectDropTable(db, table, true, isReplay, 0L);
+                } finally {
+                    table.writeUnlock();
+                }
+                LOG.info("dropped IVM stream {} because its base table is excluded from MTMV",
+                        streamName);
+            }
+        } finally {
+            db.writeUnlock();
+        }
+    }
 }

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

@@ -29,6 +29,7 @@
 import org.apache.doris.common.util.DynamicPartitionUtil;
 import org.apache.doris.common.util.MasterDaemon;
 import org.apache.doris.common.util.TimeUtils;
+import org.apache.doris.mtmv.BaseTableInfo;
 import org.apache.doris.persist.RecoverInfo;
 import org.apache.doris.persist.gson.GsonUtils;
 import org.apache.doris.qe.GlobalVariable;
@@ -49,6 +50,7 @@
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Collection;
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.Iterator;
 import java.util.List;
@@ -1065,6 +1067,10 @@ public void recoverPartition(long dbId, OlapTable table, String partitionName,
                 recoverPartition.setName(newPartitionName);
             }
 
+            Env.getCurrentEnv().getMtmvService().getRelationManager().markIvmBaselineRebuildForPartitionChange(
+                    new BaseTableInfo(table, dbId),
+                    Collections.singletonMap(partitionName, recoverPartition.getId()),
+                    "Base table partition was recovered without row binlog");
             // recover partition
             table.addPartition(recoverPartition);