[feature](index change)Support light index change for inverted index without parser

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

@@ -1258,12 +1258,7 @@ private void checkAssignedTargetIndexName(String baseIndexName, String targetInd
     }
 
     private void createJob(String rawSql, long dbId, OlapTable olapTable, Map<Long, LinkedList<Column>> indexSchemaMap,
-                           Map<String, String> propertyMap, List<Index> indexes,
-                           boolean isBuildIndex) throws UserException {
-        if (isBuildIndex) {
-            // remove the index which is not the base index, only base index can be built index
-            indexSchemaMap.entrySet().removeIf(entry -> !entry.getKey().equals(olapTable.getBaseIndexId()));
-        }
+                           Map<String, String> propertyMap, List<Index> indexes) throws UserException {
         checkReplicaCount(olapTable);
 
         // process properties first
@@ -1299,7 +1294,7 @@ private void createJob(String rawSql, long dbId, OlapTable olapTable, Map<Long,
         boolean hasIndexChange = false;
         Set<Index> newSet = new HashSet<>(indexes);
         Set<Index> oriSet = new HashSet<>(olapTable.getIndexes());
-        if (!newSet.equals(oriSet) || isBuildIndex) {
+        if (!newSet.equals(oriSet)) {
             hasIndexChange = true;
         }
 
@@ -2092,9 +2087,21 @@ public int getAsInt() {
                     }
                     lightSchemaChange = false;
 
+                    // 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.isLightIndexChangeSupported()) {
+                    if (index.isLightAddIndexSupported(enableAddIndexForNewData)) {
                         alterIndexes.add(index);
                         isDropIndex = false;
                         lightIndexChange = true;
@@ -2103,7 +2110,7 @@ public int getAsInt() {
                     BuildIndexClause buildIndexClause = (BuildIndexClause) alterClause;
                     IndexDef indexDef = buildIndexClause.getIndexDef();
                     Index index = buildIndexClause.getIndex();
-                    if (Config.isCloudMode() && index.getIndexType() == IndexDef.IndexType.INVERTED) {
+                    if (Config.isCloudMode()) {
                         throw new DdlException("BUILD INDEX operation failed: No need to do it in cloud mode.");
                     }
 
@@ -2165,17 +2172,12 @@ public int getAsInt() {
                 if (alterIndexes.isEmpty()) {
                     throw new DdlException("Altered index is empty. please check your alter stmt.");
                 }
-                IndexDef.IndexType indexType = alterIndexes.get(0).getIndexType();
                 if (Config.enable_light_index_change) {
-                    if (indexType == IndexDef.IndexType.INVERTED) {
-                        buildOrDeleteTableInvertedIndices(db, olapTable, indexSchemaMap,
-                                alterIndexes, indexOnPartitions, false);
-                    } else {
-                        createJob(rawSql, db.getId(), olapTable, indexSchemaMap, propertyMap, newIndexes, true);
-                    }
+                    buildOrDeleteTableInvertedIndices(db, olapTable, indexSchemaMap,
+                            alterIndexes, indexOnPartitions, false);
                 }
             } else {
-                createJob(rawSql, db.getId(), olapTable, indexSchemaMap, propertyMap, newIndexes, false);
+                createJob(rawSql, db.getId(), olapTable, indexSchemaMap, propertyMap, newIndexes);
             }
         } finally {
             olapTable.writeUnlock();

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

@@ -116,10 +116,10 @@ public void analyze(Analyzer analyzer) throws AnalysisException {
         }
 
         IndexDef.IndexType indexType = existedIdx.getIndexType();
-        if (!existedIdx.isLightIndexChangeSupported()) {
-            throw new AnalysisException(indexType.toString() + " index is not needed to build.");
+        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();

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

@@ -153,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);
     }
@@ -170,16 +174,25 @@ public String getInvertedIndexParserStopwords() {
     }
 
     // Whether the index can be changed in light mode
-    // cloud mode only supports light change for ngram_bf index
-    // local mode supports light change for both inverted index and ngram_bf index
-    // the rest of the index types do not support light change
     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()) {
-            return indexType == IndexDef.IndexType.NGRAM_BF;
-        } else {
-            return indexType == IndexDef.IndexType.INVERTED
-                || indexType == IndexDef.IndexType.NGRAM_BF;
+            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() {

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

@@ -344,6 +344,14 @@ public OlapFile.TabletMetaCloudPB.Builder createTabletMetaBuilder(long tableId,
                 schemaBuilder.setInvertedIndexStorageFormat(OlapFile.InvertedIndexStorageFormatPB.V2);
             } else if (invertedIndexFileStorageFormat == TInvertedIndexFileStorageFormat.V3) {
                 schemaBuilder.setInvertedIndexStorageFormat(OlapFile.InvertedIndexStorageFormatPB.V3);
+            } else if (invertedIndexFileStorageFormat == TInvertedIndexFileStorageFormat.DEFAULT) {
+                if (Config.inverted_index_storage_format.equalsIgnoreCase("V1")) {
+                    schemaBuilder.setInvertedIndexStorageFormat(OlapFile.InvertedIndexStorageFormatPB.V1);
+                } else if (Config.inverted_index_storage_format.equalsIgnoreCase("V3")) {
+                    schemaBuilder.setInvertedIndexStorageFormat(OlapFile.InvertedIndexStorageFormatPB.V3);
+                } else {
+                    schemaBuilder.setInvertedIndexStorageFormat(OlapFile.InvertedIndexStorageFormatPB.V2);
+                }
             } else {
                 throw new DdlException("invalid inverted index storage format");
             }

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

@@ -114,8 +114,9 @@ public void validate(ConnectContext ctx) throws UserException {
         }
 
         IndexDef.IndexType indexType = existedIdx.getIndexType();
-        if (!existedIdx.isLightIndexChangeSupported()) {
-            throw new AnalysisException(indexType.toString() + " index is not needed to build.");
+        if (indexType == IndexDef.IndexType.NGRAM_BF
+                || indexType == IndexDef.IndexType.BLOOMFILTER) {
+            throw new AnalysisException(indexType + " index is not needed to build.");
         }
 
         indexDef = new IndexDefinition(indexName, partitionNamesInfo, indexType);

fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java

@@ -746,6 +746,7 @@ public class SessionVariable implements Serializable, Writable {
     public static final String SQL_CONVERTOR_CONFIG = "sql_convertor_config";
 
     public static final String PREFER_UDF_OVER_BUILTIN = "prefer_udf_over_builtin";
+    public static final String ENABLE_ADD_INDEX_FOR_NEW_DATA = "enable_add_index_for_new_data";
 
     /**
      * If set false, user couldn't submit analyze SQL and FE won't allocate any related resources.
@@ -2663,6 +2664,14 @@ public boolean isEnableESParallelScroll() {
         return enableESParallelScroll;
     }
 
+    @VariableMgr.VarAttr(name = ENABLE_ADD_INDEX_FOR_NEW_DATA, fuzzy = true, description = {
+            "是否启用仅对新数据生效的索引添加模式，开启时新建索引只对后续写入的数据生效，关闭时对全部数据重建索引",
+            "Whether to enable add index mode that only affects new data, "
+                    + "when enabled new indexes only affect subsequently written data, "
+                    + "when disabled rebuild indexes for all data"
+    })
+    public boolean enableAddIndexForNewData = false;
+
     // If this fe is in fuzzy mode, then will use initFuzzyModeVariables to generate some variables,
     // not the default value set in the code.
     @SuppressWarnings("checkstyle:Indentation")
@@ -4925,4 +4934,11 @@ public boolean showSplitProfileInfo() {
         return enableProfile() && getProfileLevel() > 1;
     }
 
+    public boolean isEnableAddIndexForNewData() {
+        return enableAddIndexForNewData;
+    }
+
+    public void setEnableAddIndexForNewData(boolean enableAddIndexForNewData) {
+        this.enableAddIndexForNewData = enableAddIndexForNewData;
+    }
 }