Cleanup setters in IndexLoadingConfig and do not initialize IndexConfig from it

pinot-core/src/main/java/org/apache/pinot/core/data/manager/BaseTableDataManager.java

@@ -349,7 +349,6 @@ public Pair<TableConfig, Schema> fetchTableConfigAndSchema() {
   public IndexLoadingConfig getIndexLoadingConfig(TableConfig tableConfig, @Nullable Schema schema) {
     IndexLoadingConfig indexLoadingConfig = new IndexLoadingConfig(_instanceDataManagerConfig, tableConfig, schema);
     indexLoadingConfig.setTableDataDir(_tableDataDir);
-    indexLoadingConfig.setInstanceTierConfigs(_instanceDataManagerConfig.getTierConfigs());
     return indexLoadingConfig;
   }
 
@@ -620,7 +619,6 @@ public void reloadSegment(String segmentName, IndexLoadingConfig indexLoadingCon
     String segmentTier = getSegmentCurrentTier(segmentName);
     indexLoadingConfig.setSegmentTier(segmentTier);
     indexLoadingConfig.setTableDataDir(_tableDataDir);
-    indexLoadingConfig.setInstanceTierConfigs(_instanceDataManagerConfig.getTierConfigs());
     File indexDir = getSegmentDataDir(segmentName, segmentTier, indexLoadingConfig.getTableConfig());
     Lock segmentLock = getSegmentLock(segmentName);
     segmentLock.lock();

pinot-core/src/test/java/org/apache/pinot/core/data/manager/BaseTableDataManagerTest.java

@@ -720,7 +720,6 @@ private IndexLoadingConfig createTierIndexLoadingConfig(TableConfig tableConfig)
     when(instanceDataManagerConfig.getConfig()).thenReturn(new PinotConfiguration());
     IndexLoadingConfig indexLoadingConfig = new IndexLoadingConfig(instanceDataManagerConfig, tableConfig, null);
     indexLoadingConfig.setTableDataDir(TEMP_DIR.getAbsolutePath() + File.separator + tableConfig.getTableName());
-    indexLoadingConfig.setInstanceTierConfigs(Map.of());
     indexLoadingConfig.setSegmentTier(TIER_NAME);
     return indexLoadingConfig;
   }

pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/bloom/BloomIndexType.java

@@ -24,8 +24,6 @@
 import java.util.Map;
 import javax.annotation.Nullable;
 import org.apache.pinot.segment.local.segment.creator.impl.bloom.OnHeapGuavaBloomFilterCreator;
-import org.apache.pinot.segment.local.segment.index.loader.ConfigurableFromIndexLoadingConfig;
-import org.apache.pinot.segment.local.segment.index.loader.IndexLoadingConfig;
 import org.apache.pinot.segment.local.segment.index.loader.bloomfilter.BloomFilterHandler;
 import org.apache.pinot.segment.local.segment.index.readers.bloom.BloomFilterReaderFactory;
 import org.apache.pinot.segment.spi.ColumnMetadata;
@@ -49,9 +47,7 @@
 import org.apache.pinot.spi.data.Schema;
 
 
-public class BloomIndexType
-    extends AbstractIndexType<BloomFilterConfig, BloomFilterReader, BloomFilterCreator>
-    implements ConfigurableFromIndexLoadingConfig<BloomFilterConfig> {
+public class BloomIndexType extends AbstractIndexType<BloomFilterConfig, BloomFilterReader, BloomFilterCreator> {
   public static final String INDEX_DISPLAY_NAME = "bloom";
   private static final List<String> EXTENSIONS =
       Collections.singletonList(V1Constants.Indexes.BLOOM_FILTER_FILE_EXTENSION);
@@ -65,11 +61,6 @@ public Class<BloomFilterConfig> getIndexConfigClass() {
     return BloomFilterConfig.class;
   }
 
-  @Override
-  public Map<String, BloomFilterConfig> fromIndexLoadingConfig(IndexLoadingConfig indexLoadingConfig) {
-    return indexLoadingConfig.getBloomFilterConfigs();
-  }
-
   @Override
   public BloomFilterConfig getDefaultConfig() {
     return BloomFilterConfig.DISABLED;
@@ -82,16 +73,14 @@ public String getPrettyName() {
 
   @Override
   public ColumnConfigDeserializer<BloomFilterConfig> createDeserializer() {
-    return IndexConfigDeserializer.fromIndexes(getPrettyName(), getIndexConfigClass())
-        .withExclusiveAlternative(
-            IndexConfigDeserializer.ifIndexingConfig(// reads tableConfig.indexingConfig.bloomFilterConfigs
-                IndexConfigDeserializer.fromMap(tableConfig -> tableConfig.getIndexingConfig().getBloomFilterConfigs())
-                  .withFallbackAlternative(// reads tableConfig.indexingConfig.bloomFilterColumns
-                      IndexConfigDeserializer.fromCollection(
-                          tableConfig -> tableConfig.getIndexingConfig().getBloomFilterColumns(),
-                          (accum, column) -> accum.put(column, BloomFilterConfig.DEFAULT)))
-            )
-        );
+    ColumnConfigDeserializer<BloomFilterConfig> fromIndexes =
+        IndexConfigDeserializer.fromIndexes(getPrettyName(), getIndexConfigClass());
+    ColumnConfigDeserializer<BloomFilterConfig> fromBloomFilterConfigs =
+        IndexConfigDeserializer.fromMap(tableConfig -> tableConfig.getIndexingConfig().getBloomFilterConfigs());
+    ColumnConfigDeserializer<BloomFilterConfig> fromBloomFilterColumns =
+        IndexConfigDeserializer.fromCollection(tableConfig -> tableConfig.getIndexingConfig().getBloomFilterColumns(),
+            (accum, column) -> accum.put(column, BloomFilterConfig.DEFAULT));
+    return fromIndexes.withExclusiveAlternative(fromBloomFilterConfigs.withFallbackAlternative(fromBloomFilterColumns));
   }
 
   @Override

pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/dictionary/DictionaryIndexType.java

@@ -26,7 +26,6 @@
 import java.io.UncheckedIOException;
 import java.util.ArrayList;
 import java.util.Collections;
-import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
@@ -37,8 +36,6 @@
 import org.apache.pinot.segment.local.io.util.PinotDataBitSet;
 import org.apache.pinot.segment.local.realtime.impl.dictionary.MutableDictionaryFactory;
 import org.apache.pinot.segment.local.segment.creator.impl.SegmentDictionaryCreator;
-import org.apache.pinot.segment.local.segment.index.loader.ConfigurableFromIndexLoadingConfig;
-import org.apache.pinot.segment.local.segment.index.loader.IndexLoadingConfig;
 import org.apache.pinot.segment.local.segment.index.readers.BigDecimalDictionary;
 import org.apache.pinot.segment.local.segment.index.readers.BytesDictionary;
 import org.apache.pinot.segment.local.segment.index.readers.DoubleDictionary;
@@ -88,8 +85,7 @@
 
 
 public class DictionaryIndexType
-    extends AbstractIndexType<DictionaryIndexConfig, Dictionary, SegmentDictionaryCreator>
-    implements ConfigurableFromIndexLoadingConfig<DictionaryIndexConfig> {
+    extends AbstractIndexType<DictionaryIndexConfig, Dictionary, SegmentDictionaryCreator> {
   private static final Logger LOGGER = LoggerFactory.getLogger(DictionaryIndexType.class);
   private static final List<String> EXTENSIONS = Collections.singletonList(V1Constants.Dict.FILE_EXTENSION);
 
@@ -102,24 +98,6 @@ public Class<DictionaryIndexConfig> getIndexConfigClass() {
     return DictionaryIndexConfig.class;
   }
 
-  @Override
-  public Map<String, DictionaryIndexConfig> fromIndexLoadingConfig(
-      IndexLoadingConfig indexLoadingConfig) {
-    Map<String, DictionaryIndexConfig> result = new HashMap<>();
-    Set<String> noDictionaryCols = indexLoadingConfig.getNoDictionaryColumns();
-    Set<String> onHeapCols = indexLoadingConfig.getOnHeapDictionaryColumns();
-    Set<String> varLengthCols = indexLoadingConfig.getVarLengthDictionaryColumns();
-    for (String column : indexLoadingConfig.getAllKnownColumns()) {
-      if (noDictionaryCols.contains(column)) {
-        result.put(column, DictionaryIndexConfig.disabled());
-      } else {
-        // Intern configs can only be used if dictionary is enabled through FieldConfigLists.
-        result.put(column, new DictionaryIndexConfig(onHeapCols.contains(column), varLengthCols.contains(column)));
-      }
-    }
-    return result;
-  }
-
   @Override
   public DictionaryIndexConfig getDefaultConfig() {
     return DictionaryIndexConfig.DEFAULT;

pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/fst/FstIndexType.java

@@ -22,14 +22,10 @@
 import com.google.common.base.Preconditions;
 import com.google.common.collect.ImmutableList;
 import java.io.IOException;
-import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
-import java.util.Set;
 import javax.annotation.Nullable;
 import org.apache.pinot.segment.local.segment.creator.impl.inv.text.LuceneFSTIndexCreator;
-import org.apache.pinot.segment.local.segment.index.loader.ConfigurableFromIndexLoadingConfig;
-import org.apache.pinot.segment.local.segment.index.loader.IndexLoadingConfig;
 import org.apache.pinot.segment.local.segment.index.loader.invertedindex.FSTIndexHandler;
 import org.apache.pinot.segment.local.segment.index.readers.LuceneFSTIndexReader;
 import org.apache.pinot.segment.local.utils.nativefst.FSTHeader;
@@ -60,8 +56,7 @@
 import org.apache.pinot.spi.data.Schema;
 
 
-public class FstIndexType extends AbstractIndexType<FstIndexConfig, TextIndexReader, FSTIndexCreator>
-    implements ConfigurableFromIndexLoadingConfig<FstIndexConfig> {
+public class FstIndexType extends AbstractIndexType<FstIndexConfig, TextIndexReader, FSTIndexCreator> {
   public static final String INDEX_DISPLAY_NAME = "fst";
   private static final List<String> EXTENSIONS =
       ImmutableList.of(V1Constants.Indexes.LUCENE_FST_INDEX_FILE_EXTENSION,
@@ -77,51 +72,6 @@ public Class<FstIndexConfig> getIndexConfigClass() {
     return FstIndexConfig.class;
   }
 
-  @Override
-  public Map<String, FstIndexConfig> fromIndexLoadingConfig(IndexLoadingConfig indexLoadingConfig) {
-    Map<String, FstIndexConfig> result = new HashMap<>();
-    Set<String> fstIndexColumns = indexLoadingConfig.getFSTIndexColumns();
-    for (String column : indexLoadingConfig.getAllKnownColumns()) {
-      if (fstIndexColumns.contains(column)) {
-        FSTType fstType = getFstTypeFromIndexLoadingConfig(indexLoadingConfig, column);
-        FstIndexConfig conf = new FstIndexConfig(fstType);
-        result.put(column, conf);
-      } else {
-        result.put(column, FstIndexConfig.DISABLED);
-      }
-    }
-    return result;
-  }
-
-  private FSTType getFstTypeFromIndexLoadingConfig(IndexLoadingConfig indexLoadingConfig, String column) {
-
-    FSTType fstType = indexLoadingConfig.getFSTIndexType();
-
-    TableConfig tableConfig = indexLoadingConfig.getTableConfig();
-    if (tableConfig != null) {
-      List<FieldConfig> fieldConfigList = tableConfig.getFieldConfigList();
-      if (fieldConfigList != null) {
-        FieldConfig fieldConfig = fieldConfigList.stream()
-            .filter(fc -> fc.getName().equals(column))
-            .findAny()
-            .orElse(null);
-        if (fieldConfig != null) {
-          Map<String, String> textProperties = fieldConfig.getProperties();
-          if (textProperties != null) {
-            for (Map.Entry<String, String> entry : textProperties.entrySet()) {
-              if (entry.getKey().equalsIgnoreCase(FieldConfig.TEXT_FST_TYPE)) {
-                fstType = FSTType.NATIVE;
-              } else {
-                fstType = FSTType.LUCENE;
-              }
-            }
-          }
-        }
-      }
-    }
-    return fstType;
-  }
-
   @Override
   public FstIndexConfig getDefaultConfig() {
     return FstIndexConfig.DISABLED;

pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/inverted/InvertedIndexType.java

@@ -24,14 +24,10 @@
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
-import java.util.function.Function;
-import java.util.stream.Collectors;
 import javax.annotation.Nullable;
 import org.apache.pinot.segment.local.realtime.impl.invertedindex.RealtimeInvertedIndex;
 import org.apache.pinot.segment.local.segment.creator.impl.inv.OffHeapBitmapInvertedIndexCreator;
 import org.apache.pinot.segment.local.segment.creator.impl.inv.OnHeapBitmapInvertedIndexCreator;
-import org.apache.pinot.segment.local.segment.index.loader.ConfigurableFromIndexLoadingConfig;
-import org.apache.pinot.segment.local.segment.index.loader.IndexLoadingConfig;
 import org.apache.pinot.segment.local.segment.index.loader.invertedindex.InvertedIndexHandler;
 import org.apache.pinot.segment.local.segment.index.readers.BitmapInvertedIndexReader;
 import org.apache.pinot.segment.spi.ColumnMetadata;
@@ -60,8 +56,7 @@
 
 
 public class InvertedIndexType
-    extends AbstractIndexType<IndexConfig, InvertedIndexReader, DictionaryBasedInvertedIndexCreator>
-    implements ConfigurableFromIndexLoadingConfig<IndexConfig> {
+    extends AbstractIndexType<IndexConfig, InvertedIndexReader, DictionaryBasedInvertedIndexCreator> {
   public static final String INDEX_DISPLAY_NAME = "inverted";
   private static final List<String> EXTENSIONS =
       Collections.singletonList(V1Constants.Indexes.BITMAP_INVERTED_INDEX_FILE_EXTENSION);
@@ -75,12 +70,6 @@ public Class<IndexConfig> getIndexConfigClass() {
     return IndexConfig.class;
   }
 
-  @Override
-  public Map<String, IndexConfig> fromIndexLoadingConfig(IndexLoadingConfig indexLoadingConfig) {
-    return indexLoadingConfig.getInvertedIndexColumns().stream()
-        .collect(Collectors.toMap(Function.identity(), v -> IndexConfig.ENABLED));
-  }
-
   @Override
   public IndexConfig getDefaultConfig() {
     return IndexConfig.DISABLED;
@@ -93,11 +82,12 @@ public String getPrettyName() {
 
   @Override
   public ColumnConfigDeserializer<IndexConfig> createDeserializer() {
-    ColumnConfigDeserializer<IndexConfig> fromInvertedCols = IndexConfigDeserializer.fromCollection(
-        tableConfig -> tableConfig.getIndexingConfig().getInvertedIndexColumns(),
-        (acum, column) -> acum.put(column, IndexConfig.ENABLED));
-    return IndexConfigDeserializer.fromIndexes(getPrettyName(), getIndexConfigClass())
-        .withExclusiveAlternative(IndexConfigDeserializer.ifIndexingConfig(fromInvertedCols));
+    ColumnConfigDeserializer<IndexConfig> fromIndexes =
+        IndexConfigDeserializer.fromIndexes(getPrettyName(), getIndexConfigClass());
+    ColumnConfigDeserializer<IndexConfig> fromInvertedIndexColumns =
+        IndexConfigDeserializer.fromCollection(tableConfig -> tableConfig.getIndexingConfig().getInvertedIndexColumns(),
+            (acum, column) -> acum.put(column, IndexConfig.ENABLED));
+    return fromIndexes.withExclusiveAlternative(fromInvertedIndexColumns);
   }
 
   public DictionaryBasedInvertedIndexCreator createIndexCreator(IndexCreationContext context)
@@ -112,8 +102,7 @@ public DictionaryBasedInvertedIndexCreator createIndexCreator(IndexCreationConte
   }
 
   @Override
-  public DictionaryBasedInvertedIndexCreator createIndexCreator(IndexCreationContext context,
-      IndexConfig indexConfig)
+  public DictionaryBasedInvertedIndexCreator createIndexCreator(IndexCreationContext context, IndexConfig indexConfig)
       throws IOException {
     return createIndexCreator(context);
   }
@@ -170,8 +159,8 @@ public InvertedIndexReader createIndexReader(SegmentDirectory.Reader segmentRead
             + "index if it has no dictionary");
       }
       if (metadata.isSorted() && metadata.isSingleValue()) {
-        ForwardIndexReader fwdReader = StandardIndexes.forward().getReaderFactory()
-            .createIndexReader(segmentReader, fieldIndexConfigs, metadata);
+        ForwardIndexReader fwdReader =
+            StandardIndexes.forward().getReaderFactory().createIndexReader(segmentReader, fieldIndexConfigs, metadata);
         Preconditions.checkState(fwdReader instanceof SortedIndexReader);
         return (SortedIndexReader) fwdReader;
       } else {

pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/json/JsonIndexType.java

@@ -28,8 +28,6 @@
 import org.apache.pinot.segment.local.realtime.impl.json.MutableJsonIndexImpl;
 import org.apache.pinot.segment.local.segment.creator.impl.inv.json.OffHeapJsonIndexCreator;
 import org.apache.pinot.segment.local.segment.creator.impl.inv.json.OnHeapJsonIndexCreator;
-import org.apache.pinot.segment.local.segment.index.loader.ConfigurableFromIndexLoadingConfig;
-import org.apache.pinot.segment.local.segment.index.loader.IndexLoadingConfig;
 import org.apache.pinot.segment.local.segment.index.loader.invertedindex.JsonIndexHandler;
 import org.apache.pinot.segment.local.segment.index.readers.json.ImmutableJsonIndexReader;
 import org.apache.pinot.segment.spi.ColumnMetadata;
@@ -56,8 +54,7 @@
 import org.apache.pinot.spi.data.Schema;
 
 
-public class JsonIndexType extends AbstractIndexType<JsonIndexConfig, JsonIndexReader, JsonIndexCreator>
-    implements ConfigurableFromIndexLoadingConfig<JsonIndexConfig> {
+public class JsonIndexType extends AbstractIndexType<JsonIndexConfig, JsonIndexReader, JsonIndexCreator> {
   public static final String INDEX_DISPLAY_NAME = "json";
   private static final List<String> EXTENSIONS =
       Collections.singletonList(V1Constants.Indexes.JSON_INDEX_FILE_EXTENSION);
@@ -71,11 +68,6 @@ public Class<JsonIndexConfig> getIndexConfigClass() {
     return JsonIndexConfig.class;
   }
 
-  @Override
-  public Map<String, JsonIndexConfig> fromIndexLoadingConfig(IndexLoadingConfig indexLoadingConfig) {
-    return indexLoadingConfig.getJsonIndexConfigs();
-  }
-
   @Override
   public JsonIndexConfig getDefaultConfig() {
     return JsonIndexConfig.DISABLED;
@@ -88,17 +80,14 @@ public String getPrettyName() {
 
   @Override
   public ColumnConfigDeserializer<JsonIndexConfig> createDeserializer() {
-    // reads tableConfig.indexingConfig.jsonIndexConfigs
-    ColumnConfigDeserializer<JsonIndexConfig> fromJsonIndexConf =
+    ColumnConfigDeserializer<JsonIndexConfig> fromIndexes =
+        IndexConfigDeserializer.fromIndexes(getPrettyName(), getIndexConfigClass());
+    ColumnConfigDeserializer<JsonIndexConfig> fromJsonIndexConfigs =
         IndexConfigDeserializer.fromMap(tableConfig -> tableConfig.getIndexingConfig().getJsonIndexConfigs());
-    // reads tableConfig.indexingConfig.jsonIndexColumns
-    ColumnConfigDeserializer<JsonIndexConfig> fromJsonIndexCols =
-        IndexConfigDeserializer.fromCollection(
-            tableConfig -> tableConfig.getIndexingConfig().getJsonIndexColumns(),
-            (accum, column) -> accum.put(column, new JsonIndexConfig()));
-    return IndexConfigDeserializer.fromIndexes(getPrettyName(), getIndexConfigClass())
-        .withExclusiveAlternative(
-            IndexConfigDeserializer.ifIndexingConfig(fromJsonIndexCols.withExclusiveAlternative(fromJsonIndexConf)));
+    ColumnConfigDeserializer<JsonIndexConfig> fromJsonIndexColumns =
+        IndexConfigDeserializer.fromCollection(tableConfig -> tableConfig.getIndexingConfig().getJsonIndexColumns(),
+            (accum, column) -> accum.put(column, JsonIndexConfig.DEFAULT));
+    return fromIndexes.withExclusiveAlternative(fromJsonIndexConfigs.withFallbackAlternative(fromJsonIndexColumns));
   }
 
   @Override
@@ -108,8 +97,8 @@ public JsonIndexCreator createIndexCreator(IndexCreationContext context, JsonInd
         "Json index is currently only supported on single-value columns");
     Preconditions.checkState(context.getFieldSpec().getDataType().getStoredType() == FieldSpec.DataType.STRING,
         "Json index is currently only supported on STRING columns");
-    return context.isOnHeap()
-        ? new OnHeapJsonIndexCreator(context.getIndexDir(), context.getFieldSpec().getName(), indexConfig)
+    return context.isOnHeap() ? new OnHeapJsonIndexCreator(context.getIndexDir(), context.getFieldSpec().getName(),
+        indexConfig)
         : new OffHeapJsonIndexCreator(context.getIndexDir(), context.getFieldSpec().getName(), indexConfig);
   }