branch-3.1: [feature](index)Support light index add for inverted index without parser and ngram bf index

fe/fe-core/src/main/cup/sql_parser.cup

@@ -900,7 +900,6 @@ nonterminal Map<String, String> key_value_map, opt_key_value_map, opt_key_value_
             opt_ext_properties, opt_enable_feature_properties, properties;
 nonterminal ColumnDef column_definition;
 nonterminal IndexDef index_definition;
-nonterminal IndexDef build_index_definition;
 nonterminal ArrayList<ColumnDef> column_definition_list;
 nonterminal ArrayList<IndexDef> index_definition_list;
 nonterminal AggregateType opt_agg_type;
@@ -2194,7 +2193,7 @@ create_stmt ::=
     :}
     | KW_BUILD KW_INDEX ident:indexName KW_ON table_name:tableName opt_partition_names:partitionNames
     {:
-        RESULT = new AlterTableStmt(tableName, Lists.newArrayList(new BuildIndexClause(tableName, new IndexDef(indexName, partitionNames, true), false)));
+        RESULT = new AlterTableStmt(tableName, Lists.newArrayList(new BuildIndexClause(tableName, indexName, partitionNames, false)));
     :}
     /* stage */
     | KW_CREATE KW_STAGE opt_if_not_exists:ifNotExists ident:stageName KW_PROPERTIES opt_key_value_map:properties
@@ -4121,13 +4120,6 @@ index_definition ::=
     :}
     ;
 
-build_index_definition ::=
-    KW_INDEX ident:indexName opt_partition_names:partitionNames
-    {:
-        RESULT = new IndexDef(indexName, partitionNames, true);
-    :}
-    ;
-
 opt_nullable_type ::=
     {:
         RESULT = ColumnNullableType.DEFAULT;

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

@@ -1330,7 +1330,7 @@ private void createJob(String rawSql, long dbId, OlapTable olapTable, Map<Long,
             throw new DdlException(e.getMessage());
         }
 
-        // check bloom filter has change
+        // check bloom filter has been changed
         boolean hasBfChange = false;
         Set<String> oriBfColumns = olapTable.getCopiedBfColumns();
         double oriBfFpp = olapTable.getBfFpp();
@@ -1991,7 +1991,7 @@ public int getAsInt() {
 
             List<Index> newIndexes = olapTable.getCopiedIndexes();
             List<Index> alterIndexes = new ArrayList<>();
-            Map<Long, Set<String>> invertedIndexOnPartitions = new HashMap<>();
+            Map<Long, Set<String>> indexOnPartitions = new HashMap<>();
             boolean isDropIndex = false;
             Map<String, String> propertyMap = new HashMap<>();
             for (AlterClause alterClause : alterClauses) {
@@ -2145,10 +2145,23 @@ public int getAsInt() {
                     }
                     lightSchemaChange = false;
 
-                    if (index.isLightIndexChangeSupported() && !Config.isCloudMode()) {
+                    // Check if the index supports light index change and session variable is enabled
+                    boolean enableAddIndexForNewData = true;
+                    try {
+                        ConnectContext context = ConnectContext.get();
+                        if (context != null && context.getSessionVariable() != null) {
+                            enableAddIndexForNewData = context.getSessionVariable().isEnableAddIndexForNewData();
+                        }
+                    } catch (Exception e) {
+                        LOG.warn("Failed to get session variable enable_add_index_for_new_data, "
+                                + "using default value: false", e);
+                    }
+
+                    // ngram_bf index can do light_schema_change in both local and cloud mode
+                    // inverted index can only do light_schema_change in local mode
+                    if (index.isLightAddIndexSupported(enableAddIndexForNewData)) {
                         alterIndexes.add(index);
                         isDropIndex = false;
-                        // now only support light index change for inverted index
                         lightIndexChange = true;
                     }
                 } else if (alterClause instanceof BuildIndexClause) {
@@ -2159,54 +2172,14 @@ public int getAsInt() {
                         throw new DdlException("BUILD INDEX operation failed: No need to do it in cloud mode.");
                     }
 
-                    if (!olapTable.isPartitionedTable()) {
-                        List<String> specifiedPartitions = indexDef.getPartitionNames();
-                        if (!specifiedPartitions.isEmpty()) {
-                            throw new DdlException("table " + olapTable.getName()
-                                    + " is not partitioned, cannot build index with partitions.");
-                        }
-                    }
-                    List<Index> existedIndexes = olapTable.getIndexes();
-                    boolean found = false;
-                    for (Index existedIdx : existedIndexes) {
-                        if (existedIdx.getIndexName().equalsIgnoreCase(indexDef.getIndexName())) {
-                            found = true;
-                            if (!existedIdx.isLightIndexChangeSupported()) {
-                                throw new DdlException("BUILD INDEX operation failed: The index "
-                                        + existedIdx.getIndexName() + " of type " + existedIdx.getIndexType()
-                                        + " does not support lightweight index changes.");
-                            }
-                            for (Column column : olapTable.getBaseSchema()) {
-                                if (!column.getType().isVariantType()) {
-                                    continue;
-                                }
-                                // variant type column can not support for building index
-                                for (String indexColumn : existedIdx.getColumns()) {
-                                    if (column.getName().equalsIgnoreCase(indexColumn)) {
-                                        throw new DdlException("BUILD INDEX operation failed: The "
-                                                + indexDef.getIndexName() + " index can not be built on the "
-                                                + indexColumn + " column, because it is a variant type column.");
-                                    }
-                                }
-                            }
-                            index = existedIdx.clone();
-                            if (indexDef.getPartitionNames().isEmpty()) {
-                                invertedIndexOnPartitions.put(index.getIndexId(), olapTable.getPartitionNames());
-                            } else {
-                                invertedIndexOnPartitions.put(
-                                        index.getIndexId(), new HashSet<>(indexDef.getPartitionNames()));
-                            }
-                            break;
-                        }
-                    }
-                    if (!found) {
-                        throw new DdlException("index " + indexDef.getIndexName()
-                                + " not exist, cannot build it with defferred.");
+                    if (indexDef.getPartitionNames().isEmpty()) {
+                        indexOnPartitions.put(index.getIndexId(), olapTable.getPartitionNames());
+                    } else {
+                        indexOnPartitions.put(
+                                index.getIndexId(), new HashSet<>(indexDef.getPartitionNames()));
                     }
 
-                    if (indexDef.isInvertedIndex()) {
-                        alterIndexes.add(index);
-                    }
+                    alterIndexes.add(index);
                     buildIndexChange = true;
                     lightSchemaChange = false;
                 } else if (alterClause instanceof DropIndexClause) {
@@ -2224,7 +2197,9 @@ public int getAsInt() {
                             break;
                         }
                     }
-                    if (found.isLightIndexChangeSupported() && !Config.isCloudMode()) {
+                    // only inverted index with local mode can do light drop index change
+                    if (found != null && found.getIndexType() == IndexDef.IndexType.INVERTED
+                            && Config.isNotCloudMode()) {
                         alterIndexes.add(found);
                         isDropIndex = true;
                         lightIndexChange = true;
@@ -2248,13 +2223,16 @@ public int getAsInt() {
                                              null, isDropIndex, jobId, false);
             } else if (Config.enable_light_index_change && lightIndexChange) {
                 long jobId = Env.getCurrentEnv().getNextId();
-                //for schema change add/drop inverted index optimize, direct modify table meta firstly.
+                //for schema change add/drop inverted index and ngram_bf optimize, direct modify table meta firstly.
                 modifyTableLightSchemaChange(rawSql, db, olapTable, indexSchemaMap, newIndexes,
                                              alterIndexes, isDropIndex, jobId, false);
             } else if (buildIndexChange) {
+                if (alterIndexes.isEmpty()) {
+                    throw new DdlException("Altered index is empty. please check your alter stmt.");
+                }
                 if (Config.enable_light_index_change) {
                     buildOrDeleteTableInvertedIndices(db, olapTable, indexSchemaMap,
-                                                      alterIndexes, invertedIndexOnPartitions, false);
+                            alterIndexes, indexOnPartitions, false);
                 }
             } else {
                 createJob(rawSql, db.getId(), olapTable, indexSchemaMap, propertyMap, newIndexes);
@@ -2747,6 +2725,8 @@ private void cancelIndexJob(CancelAlterTableStmt cancelAlterTableStmt) throws Dd
             olapTable.writeUnlock();
         }
 
+        // if this table has ngram_bf index, we must run cancel for schema change job
+        boolean hasNGramBFIndex = ((OlapTable) olapTable).hasIndexOfType(IndexDef.IndexType.NGRAM_BF);
         // alter job v2's cancel must be called outside the table lock
         if (jobList.size() > 0) {
             for (IndexChangeJob job : jobList) {
@@ -2761,6 +2741,8 @@ private void cancelIndexJob(CancelAlterTableStmt cancelAlterTableStmt) throws Dd
                     LOG.info("cancel build index job {} on table {} success", jobId, tableName);
                 }
             }
+        } else if (hasNGramBFIndex) {
+            cancelColumnJob(cancelAlterTableStmt);
         } else {
             throw new DdlException("No job to cancel for Table[" + tableName + "]");
         }
@@ -2800,7 +2782,7 @@ private boolean processAddIndex(CreateIndexClause alterClause, OlapTable olapTab
             Column column = olapTable.getColumn(col);
             if (column != null) {
                 indexDef.checkColumn(column, olapTable.getKeysType(),
-                        olapTable.getTableProperty().getEnableUniqueKeyMergeOnWrite(),
+                        olapTable.getEnableUniqueKeyMergeOnWrite(),
                         olapTable.getInvertedIndexFileStorageFormat());
                 if (!InvertedIndexUtil.getInvertedIndexFieldPattern(indexDef.getProperties()).isEmpty()) {
                     throw new DdlException("Can not create index with field pattern");
@@ -3040,7 +3022,7 @@ public void modifyTableLightSchemaChange(String rawSql, Database db, OlapTable o
                     }
                     try {
                         buildOrDeleteTableInvertedIndices(db, olapTable, indexSchemaMap,
-                                              alterIndexes, invertedIndexOnPartitions, true);
+                                alterIndexes, invertedIndexOnPartitions, true);
                     } catch (Exception e) {
                         throw new DdlException(e.getMessage());
                     }

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

@@ -151,7 +151,7 @@ public class SchemaChangeJobV2 extends AlterJobV2 {
     private boolean hasEnableUniqueKeySkipBitmapChanged = false;
 
     // save all schema change tasks
-    private AgentBatchTask schemaChangeBatchTask = new AgentBatchTask();
+    AgentBatchTask schemaChangeBatchTask = new AgentBatchTask();
 
     protected SchemaChangeJobV2() {
         super(JobType.SCHEMA_CHANGE);
@@ -678,14 +678,13 @@ protected void runRunningJob() throws AlterCancelException {
                                 healthyReplicaNum++;
                             }
                         }
-                        if (!FeConstants.runningUnitTest) {
-                            if (healthyReplicaNum < expectReplicationNum / 2 + 1) {
-                                LOG.warn("shadow tablet {} has few healthy replicas: {}, schema change job: {}"
-                                        + " healthyReplicaNum {} expectReplicationNum {}",
-                                        shadowTablet.getId(), replicas, jobId, healthyReplicaNum, expectReplicationNum);
-                                throw new AlterCancelException(
+
+                        if ((healthyReplicaNum < expectReplicationNum / 2 + 1) && !FeConstants.runningUnitTest) {
+                            LOG.warn("shadow tablet {} has few healthy replicas: {}, schema change job: {}"
+                                    + " healthyReplicaNum {} expectReplicationNum {}",
+                                    shadowTablet.getId(), replicas, jobId, healthyReplicaNum, expectReplicationNum);
+                            throw new AlterCancelException(
                                     "shadow tablet " + shadowTablet.getId() + " has few healthy replicas");
-                            }
                         }
                     } // end for tablets
                 }

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

@@ -18,12 +18,17 @@
 package org.apache.doris.analysis;
 
 import org.apache.doris.alter.AlterOpType;
+import org.apache.doris.catalog.DatabaseIf;
 import org.apache.doris.catalog.Env;
 import org.apache.doris.catalog.Index;
+import org.apache.doris.catalog.OlapTable;
+import org.apache.doris.catalog.Table;
+import org.apache.doris.catalog.TableIf;
 import org.apache.doris.common.AnalysisException;
 
 import com.google.common.collect.Maps;
 
+import java.util.List;
 import java.util.Map;
 
 public class BuildIndexClause extends AlterTableClause {
@@ -36,11 +41,14 @@ public class BuildIndexClause extends AlterTableClause {
     private boolean alter;
     // index internal class
     private Index index;
+    private String indexName;
+    private PartitionNames partitionNames;
 
-    public BuildIndexClause(TableName tableName, IndexDef indexDef, boolean alter) {
+    public BuildIndexClause(TableName tableName, String indexName, PartitionNames partitionNames, boolean alter) {
         super(AlterOpType.SCHEMA_CHANGE);
         this.tableName = tableName;
-        this.indexDef = indexDef;
+        this.indexName = indexName;
+        this.partitionNames = partitionNames;
         this.alter = alter;
     }
 
@@ -67,17 +75,52 @@ public TableName getTableName() {
 
     @Override
     public void analyze(Analyzer analyzer) throws AnalysisException {
-        if (indexDef == null) {
-            throw new AnalysisException("index definition expected.");
+        tableName.analyze(analyzer);
+        DatabaseIf<Table> db = Env.getCurrentEnv().getCatalogMgr().getInternalCatalog()
+                .getDb(tableName.getDb()).orElse(null);
+        if (db == null) {
+            throw new AnalysisException("Database[" + tableName.getDb() + "] is not exist");
         }
-        if (indexDef.getIndexType() == IndexDef.IndexType.NGRAM_BF
-                || indexDef.getIndexType() == IndexDef.IndexType.BLOOMFILTER) {
+
+        TableIf table = db.getTable(tableName.getTbl()).orElse(null);
+        if (table == null) {
+            throw new AnalysisException("Table[" + tableName.getTbl() + "] is not exist");
+        }
+        if (!(table instanceof OlapTable)) {
+            throw new AnalysisException("Only olap table support build index");
+        }
+
+        Index existedIdx = null;
+        for (Index index : table.getTableIndexes().getIndexes()) {
+            if (index.getIndexName().equalsIgnoreCase(indexName)) {
+                existedIdx = index;
+                if (!existedIdx.isLightIndexChangeSupported()) {
+                    throw new AnalysisException("BUILD INDEX operation failed: The index "
+                        + existedIdx.getIndexName() + " of type " + existedIdx.getIndexType()
+                        + " does not support lightweight index changes.");
+                }
+                break;
+            }
+        }
+        if (existedIdx == null) {
+            throw new AnalysisException("Index[" + indexName + "] is not exist in table[" + tableName.getTbl() + "]");
+        }
+
+        IndexDef.IndexType indexType = existedIdx.getIndexType();
+        if (indexType == IndexDef.IndexType.NGRAM_BF
+                || indexType == IndexDef.IndexType.BLOOMFILTER) {
             throw new AnalysisException("ngram bloomfilter or bloomfilter index is not needed to build.");
         }
+        indexDef = new IndexDef(indexName, partitionNames, indexType, true);
+        if (!table.isPartitionedTable()) {
+            List<String> specifiedPartitions = indexDef.getPartitionNames();
+            if (!specifiedPartitions.isEmpty()) {
+                throw new AnalysisException("table " + table.getName()
+                    + " is not partitioned, cannot build index with partitions.");
+            }
+        }
         indexDef.analyze();
-        this.index = new Index(Env.getCurrentEnv().getNextId(), indexDef.getIndexName(),
-                indexDef.getColumns(), indexDef.getIndexType(),
-                indexDef.getProperties(), indexDef.getComment());
+        this.index = existedIdx.clone();
     }
 
     @Override

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

@@ -83,9 +83,9 @@ public IndexDef(String indexName, boolean ifNotExists, List<String> columns, Ind
         }
     }
 
-    public IndexDef(String indexName, PartitionNames partitionNames, boolean isBuildDeferred) {
+    public IndexDef(String indexName, PartitionNames partitionNames, IndexType indexType, boolean isBuildDeferred) {
         this.indexName = indexName;
-        this.indexType = IndexType.INVERTED;
+        this.indexType = indexType;
         this.partitionNames = partitionNames;
         this.isBuildDeferred = isBuildDeferred;
     }

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

@@ -20,6 +20,7 @@
 import org.apache.doris.analysis.IndexDef;
 import org.apache.doris.analysis.InvertedIndexUtil;
 import org.apache.doris.common.AnalysisException;
+import org.apache.doris.common.Config;
 import org.apache.doris.common.io.Text;
 import org.apache.doris.common.io.Writable;
 import org.apache.doris.common.util.PrintableMap;
@@ -152,6 +153,10 @@ public String getInvertedIndexParser() {
         return InvertedIndexUtil.getInvertedIndexParser(properties);
     }
 
+    public boolean isInvertedIndexParserNone() {
+        return InvertedIndexUtil.INVERTED_INDEX_PARSER_NONE.equals(getInvertedIndexParser());
+    }
+
     public String getInvertedIndexParserMode() {
         return InvertedIndexUtil.getInvertedIndexParserMode(properties);
     }
@@ -168,10 +173,28 @@ public String getInvertedIndexParserStopwords() {
         return InvertedIndexUtil.getInvertedIndexParserStopwords(properties);
     }
 
+    // Whether the index can be changed in light mode
     public boolean isLightIndexChangeSupported() {
         return indexType == IndexDef.IndexType.INVERTED;
     }
 
+    // Whether the index can be added in light mode
+    // cloud mode supports light add for ngram_bf index and non-tokenized inverted index (parser="none")
+    // local mode supports light add for both inverted index and ngram_bf index
+    // the rest of the index types do not support light add
+    public boolean isLightAddIndexSupported(boolean enableAddIndexForNewData) {
+        if (Config.isCloudMode()) {
+            if (indexType == IndexDef.IndexType.INVERTED) {
+                return isInvertedIndexParserNone() && enableAddIndexForNewData;
+            } else if (indexType == IndexDef.IndexType.NGRAM_BF) {
+                return enableAddIndexForNewData;
+            }
+            return false;
+        }
+        return (indexType == IndexDef.IndexType.NGRAM_BF && enableAddIndexForNewData)
+                || (indexType == IndexDef.IndexType.INVERTED);
+    }
+
     public String getInvertedIndexCustomAnalyzer() {
         return InvertedIndexUtil.getInvertedIndexCustomAnalyzer(properties);
     }