diff --git a/fe/fe-core/src/main/java/org/apache/doris/alter/SchemaChangeHandler.java b/fe/fe-core/src/main/java/org/apache/doris/alter/SchemaChangeHandler.java index 0bc39904026..51cd7ae55bd 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/alter/SchemaChangeHandler.java +++ b/fe/fe-core/src/main/java/org/apache/doris/alter/SchemaChangeHandler.java @@ -2325,6 +2325,9 @@ public class SchemaChangeHandler extends AlterHandler { //alterClauses can or cannot light schema change boolean lightSchemaChange = true; boolean lightIndexChange = false; + // A clause needs a heavy schema change job (a column or property change that can not be light, or + // an index that has to be built for the existing data). The light index change path can not run it. + boolean requiresHeavyChange = false; boolean buildIndexChange = false; // index id -> index schema @@ -2508,10 +2511,12 @@ public class SchemaChangeHandler extends AlterHandler { indexSchemaMap, colUniqueIdSupplierMap); if (!clauseCanLightSchemaChange) { lightSchemaChange = false; + requiresHeavyChange = true; } // TODO sequence mapping do not support light schema yet if (olapTable.hasColumnSeqMapping() || !sequenceMapping.isEmpty()) { lightSchemaChange = false; + requiresHeavyChange = true; } } else if (alterOp instanceof AddColumnsOp) { // add columns @@ -2521,10 +2526,12 @@ public class SchemaChangeHandler extends AlterHandler { indexSchemaMap, false, colUniqueIdSupplierMap); if (!clauseCanLightSchemaChange) { lightSchemaChange = false; + requiresHeavyChange = true; } // TODO sequence mapping do not support light schema yet if (olapTable.hasColumnSeqMapping() || !sequenceMapping.isEmpty()) { lightSchemaChange = false; + requiresHeavyChange = true; } } else if (alterOp instanceof DropColumnOp) { // drop column and drop indexes on this column @@ -2533,10 +2540,12 @@ public class SchemaChangeHandler extends AlterHandler { indexSchemaMap, newIndexes); if (!clauseCanLightSchemaChange) { lightSchemaChange = false; + requiresHeavyChange = true; } // TODO sequence mapping do not support light schema yet if (olapTable.hasColumnSeqMapping() || !sequenceMapping.isEmpty()) { lightSchemaChange = false; + requiresHeavyChange = true; } } else if (alterOp instanceof ModifyColumnOp) { // modify column @@ -2544,15 +2553,18 @@ public class SchemaChangeHandler extends AlterHandler { olapTable, indexSchemaMap); if (!clauseCanLightSchemaChange) { lightSchemaChange = false; + requiresHeavyChange = true; } } else if (alterOp instanceof ReorderColumnsOp) { // reorder column processReorderColumn((ReorderColumnsOp) alterOp, olapTable, indexSchemaMap); lightSchemaChange = false; + requiresHeavyChange = true; } else if (alterOp instanceof ModifyTablePropertiesOp) { // modify table properties // do nothing, properties are already in propertyMap lightSchemaChange = false; + requiresHeavyChange = true; } else if (alterOp instanceof CreateIndexOp) { CreateIndexOp createIndexOp = (CreateIndexOp) alterOp; Index index = createIndexOp.getIndex(); @@ -2567,6 +2579,9 @@ public class SchemaChangeHandler extends AlterHandler { alterIndexes.add(index); isDropIndex = false; lightIndexChange = true; + } else { + // The index has to be built for the existing data as well. + requiresHeavyChange = true; } } else if (alterOp instanceof BuildIndexOp) { BuildIndexOp buildIndexOp = (BuildIndexOp) alterOp; @@ -2672,7 +2687,12 @@ public class SchemaChangeHandler extends AlterHandler { lightIndexChange, buildIndexChange, indexSchemaMap); } - if (olapTable.hasChangelog() && !(lightSchemaChange || lightIndexChange)) { + // The light index change path only publishes the new schema and submits index delete jobs. When a + // clause also needs a heavy schema change, the whole statement runs as a heavy schema change job, + // which applies the index changes as well. + boolean useLightIndexChange = Config.enable_light_index_change && lightIndexChange + && !requiresHeavyChange; + if (olapTable.hasChangelog() && !(lightSchemaChange || useLightIndexChange)) { throw new DdlException("only support light schema change operator when use table with changelog"); } @@ -2681,7 +2701,7 @@ public class SchemaChangeHandler extends AlterHandler { //for schema change add/drop value column optimize, direct modify table meta. modifyTableLightSchemaChange(rawSql, db, olapTable, indexSchemaMap, newIndexes, null, isDropIndex, jobId, false, propertyMap); - } else if (Config.enable_light_index_change && lightIndexChange) { + } else if (useLightIndexChange) { Index.checkConflict(newIndexes, olapTable.getCopiedBfColumns()); long jobId = Env.getCurrentEnv().getNextId(); // For light index changes, directly modify table metadata first. diff --git a/fe/fe-core/src/test/java/org/apache/doris/alter/SchemaChangeHandlerTest.java b/fe/fe-core/src/test/java/org/apache/doris/alter/SchemaChangeHandlerTest.java index eea5c4111c6..fccf8447720 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/alter/SchemaChangeHandlerTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/alter/SchemaChangeHandlerTest.java @@ -1485,6 +1485,68 @@ public class SchemaChangeHandlerTest extends TestWithFeService { tbl.getIndexes().stream().map(Index::getIndexName).collect(Collectors.toList())); } + @Test + public void testLightIndexChangeWithHeavyWorkRunsHeavySchemaChange() throws Exception { + String tableName = "sc_light_index_with_heavy_work"; + dropTable("test." + tableName, false); + createTable("CREATE TABLE test." + tableName + " (\n" + + "k1 INT,\n" + + "vi INT,\n" + + "v1 STRING,\n" + + "v2 STRING,\n" + + "v3 STRING,\n" + + "INDEX idx_v1 (v1) USING INVERTED,\n" + + "INDEX idx_v3 (v3) USING INVERTED\n" + + ") ENGINE=OLAP\n" + + "DUPLICATE KEY(k1)\n" + + "DISTRIBUTED BY HASH(k1) BUCKETS 1\n" + + "PROPERTIES ('replication_num' = '1', 'light_schema_change' = 'true');"); + + Database db = Env.getCurrentInternalCatalog().getDbOrMetaException("test"); + OlapTable tbl = (OlapTable) db.getTableOrMetaException(tableName, Table.TableType.OLAP); + SchemaChangeHandler handler = Env.getCurrentEnv().getSchemaChangeHandler(); + String alterPrefix = "ALTER TABLE test." + tableName + " "; + + // A type change needs a BE conversion job, so the light DROP INDEX must not publish the shadow schema. + assertRunsHeavySchemaChange(handler, tbl, alterPrefix + "MODIFY COLUMN vi BIGINT NULL, DROP INDEX idx_v1"); + Assertions.assertEquals(PrimitiveType.BIGINT, tbl.getColumn("vi").getType().getPrimitiveType()); + Assertions.assertTrue(tbl.getBaseSchema(true).stream() + .noneMatch(column -> column.getName().startsWith(SchemaChangeHandler.SHADOW_NAME_PREFIX))); + Assertions.assertTrue(tbl.getIndexes().stream().noneMatch(idx -> idx.getIndexName().equals("idx_v1"))); + + // Without new-data-only indexing, the BLOOMFILTER index has to be built for the existing data too. + boolean enableAddIndexForNewData = connectContext.getSessionVariable().isEnableAddIndexForNewData(); + connectContext.getSessionVariable().setEnableAddIndexForNewData(false); + try { + assertRunsHeavySchemaChange(handler, tbl, + alterPrefix + "ADD INDEX bf_v2 (v2) USING BLOOMFILTER, DROP INDEX idx_v3"); + } finally { + connectContext.getSessionVariable().setEnableAddIndexForNewData(enableAddIndexForNewData); + } + Assertions.assertEquals(Lists.newArrayList("bf_v2"), + tbl.getIndexes().stream().map(Index::getIndexName).collect(Collectors.toList())); + } + + // Runs the statement and checks it ran as a heavy schema change job, not as a light index change that + // publishes the schema at once and submits index delete jobs. + private void assertRunsHeavySchemaChange(SchemaChangeHandler handler, OlapTable tbl, String sql) + throws Exception { + Set alterJobIdsBefore = Sets.newHashSet(handler.getAlterJobsV2().keySet()); + long indexChangeJobsOfTable = handler.getIndexChangeJobs().values().stream() + .filter(job -> job.getTableId() == tbl.getId()).count(); + + alterTable(sql, connectContext); + jobSize++; + + List newJobs = handler.getAlterJobsV2().values().stream() + .filter(job -> job.getTableId() == tbl.getId() && !alterJobIdsBefore.contains(job.getJobId())) + .collect(Collectors.toList()); + Assertions.assertEquals(1, newJobs.size()); + Assertions.assertEquals(indexChangeJobsOfTable, handler.getIndexChangeJobs().values().stream() + .filter(job -> job.getTableId() == tbl.getId()).count()); + waitAlterJobDone(handler.getAlterJobsV2()); + } + @Test public void testAddInvertedIndexStoresCanonicalBuiltinAnalyzer() throws Exception { createAnalyzerAliasTable("sc_ik_alias");