[BugFix] Fix fail to do fast schema change on tables with sync mv in shared data (backport #55859)

fe/fe-core/src/main/java/com/starrocks/alter/LakeTableAsyncFastSchemaChangeJob.java

@@ -91,15 +91,25 @@ protected TabletMetadataUpdateAgentTask createTask(PhysicalPartition partition,
                                                        Set<Long> tablets) {
         String tag = String.format("%d_%d", partition.getId(), index.getId());
         TabletMetadataUpdateAgentTask task = null;
+        boolean needUpdateSchema = false;
         for (IndexSchemaInfo info : schemaInfos) {
             if (info.indexId == index.getId()) {
+                needUpdateSchema = true;
                 // `Set.add()` returns true means this set did not already contain the specified element
                 boolean createSchemaFile = partitionsWithSchemaFile.add(tag);
                 task = TabletMetadataUpdateAgentTaskFactory.createTabletSchemaUpdateTask(nodeId,
                         new ArrayList<>(tablets), info.schemaInfo.toTabletSchema(), createSchemaFile);
                 break;
             }
         }
+
+        // if the index is not in schemaInfos, it means the schema of index are not needed to be modified,
+        // but we still need to update the tablet meta to improve the meta version
+        if (!needUpdateSchema) {
+            task = TabletMetadataUpdateAgentTaskFactory.createTabletSchemaUpdateTask(nodeId,
+                    new ArrayList<>(tablets), null, false);
+        }
+
         return task;
     }
 

fe/fe-core/src/main/java/com/starrocks/task/TabletMetadataUpdateAgentTaskFactory.java

@@ -314,7 +314,9 @@ private UpdateTabletSchemaTask(long backendId, List<Long> tablets, TTabletSchema
                                        boolean createSchemaFile) {
             super(backendId, tablets.hashCode());
             this.tablets = new ArrayList<>(tablets);
-            this.tabletSchema = Objects.requireNonNull(tabletSchema, "tabletSchema is null");
+            // tabletSchema may be null when the table has multi materialized index
+            // and the schema of some materialized indexes are not needed to be updated
+            this.tabletSchema = tabletSchema;
             this.createSchemaFile = createSchemaFile;
         }
 
@@ -330,7 +332,11 @@ public List<TTabletMetaInfo> getTTabletMetaInfoList() {
             for (Long tabletId : tablets) {
                 TTabletMetaInfo metaInfo = new TTabletMetaInfo();
                 metaInfo.setTablet_id(tabletId);
-                metaInfo.setTablet_schema(tabletSchema);
+
+                if (tabletSchema != null) {
+                    metaInfo.setTablet_schema(tabletSchema);
+                }
+
                 metaInfos.add(metaInfo);
                 metaInfo.setCreate_schema_file(create);
                 create = false;

fe/fe-core/src/test/java/com/starrocks/alter/LakeSyncMaterializedViewTest.java

@@ -658,4 +658,26 @@ public void testCreateMVWithAggState() throws Exception {
         }
         starRocksAssert.dropTable("t1");
     }
+
+    @Test
+    public void testDropColumnWithMVByFastSchema() throws Exception {
+        starRocksAssert.useDatabase("test");
+        starRocksAssert.withTable("CREATE TABLE t1 (\n" +
+                "  k1  int,\n" +
+                "  k2  int,\n" +
+                "  k3  int,\n" +
+                "  k4  int)\n" +
+                "  DUPLICATE KEY(k1)\n" +
+                "  DISTRIBUTED BY HASH(k1) BUCKETS 3;");
+        {
+            starRocksAssert.withMaterializedView("CREATE MATERIALIZED VIEW mv1 " +
+                    "AS SELECT k1,sum(k2) AS sum_k2 FROM t1 WHERE k3>2 GROUP BY k1;");
+
+            starRocksAssert.alterTable("ALTER TABLE t1 DROP COLUMN k4;");
+            starRocksAssert.checkSchemaChangeJob();
+
+            starRocksAssert.dropTable("t1");
+            starRocksAssert.dropMaterializedView("mv1");
+        }
+    }
 }

fe/fe-core/src/test/java/com/starrocks/utframe/StarRocksAssert.java

@@ -1166,6 +1166,26 @@ private void checkAlterJob() throws InterruptedException {
         }
     }
 
+    public void checkSchemaChangeJob() throws Exception {
+        Map<Long, AlterJobV2> alterJobs = GlobalStateMgr.getCurrentState().
+                getSchemaChangeHandler().getAlterJobsV2();
+        for (AlterJobV2 alterJobV2 : alterJobs.values()) {
+            if (alterJobV2.getJobState().isFinalState()) {
+                continue;
+            }
+            Database database = GlobalStateMgr.getCurrentState().getLocalMetastore().getDb(alterJobV2.getDbId());
+            Table table =
+                    GlobalStateMgr.getCurrentState().getLocalMetastore().getTable(database.getId(), alterJobV2.getTableId());
+            Preconditions.checkState(table instanceof OlapTable);
+            OlapTable olapTable = (OlapTable) table;
+            int retry = 0;
+            while (olapTable.getState() != OlapTable.OlapTableState.NORMAL && retry++ < 6000) {
+                Thread.sleep(10);
+            }
+            Assert.assertEquals(AlterJobV2.JobState.FINISHED, alterJobV2.getJobState());
+        }
+    }
+
     public QueryAssert query(String sql) {
         return new QueryAssert(ctx, sql);
     }