Restore compatibility constructors and fix compression metadata rewrites

Author: xiangfu0

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

@@ -472,8 +472,8 @@ public void testAddIndex(TestCase testCase) {
           throw new IllegalArgumentException("Unexpected index type " + indexType);
       }
 
-      config =
-          new FieldConfig(field.getName(), encoding, null, indexTypes, null, tstmpConfig, indexes, properties, null);
+      config = new FieldConfig.Builder(field.getName()).withEncodingType(encoding).withIndexTypes(indexTypes)
+          .withTimestampConfig(tstmpConfig).withIndexes(indexes).withProperties(properties).build();
 
       tableConfig.getFieldConfigList().add(config);
 

pinot-core/src/test/java/org/apache/pinot/core/operator/transform/function/BaseTransformFunctionTest.java

@@ -341,8 +341,8 @@ protected TableConfig getTableConfig() {
     ObjectNode indexNode = JsonNodeFactory.instance.objectNode();
     indexNode.put("json", jsonIndexProps);
     FieldConfig jsonFieldConfig =
-        new FieldConfig(JSON_STRING_SV_COLUMN, FieldConfig.EncodingType.DICTIONARY, null, null, null, null, indexNode,
-            null, null);
+        new FieldConfig.Builder(JSON_STRING_SV_COLUMN).withEncodingType(FieldConfig.EncodingType.DICTIONARY)
+            .withIndexes(indexNode).build();
     fieldConfigList.add(jsonFieldConfig);
     TableConfig tableConfig =
         new TableConfigBuilder(TableType.OFFLINE).setTableName("test").setTimeColumnName(TIME_COLUMN)

pinot-core/src/test/java/org/apache/pinot/queries/BaseFSTBasedRegexpLikeQueriesTest.java

@@ -91,8 +91,9 @@ private static List<FieldConfig> getFieldConfigs(String indexType) {
 
       // Create FieldConfig with the index configuration
       return List.of(
-          new FieldConfig(DOMAIN_NAMES_COL, EncodingType.DICTIONARY, null, null, null, null, indexes, null, null),
-          new FieldConfig(URL_COL, EncodingType.DICTIONARY, null, null, null, null, indexes, null, null));
+          new FieldConfig.Builder(DOMAIN_NAMES_COL).withEncodingType(EncodingType.DICTIONARY).withIndexes(indexes)
+              .build(),
+          new FieldConfig.Builder(URL_COL).withEncodingType(EncodingType.DICTIONARY).withIndexes(indexes).build());
     } catch (Exception e) {
       throw new RuntimeException("Failed to create field configs", e);
     }

pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/custom/TextIndicesTest.java

@@ -93,13 +93,12 @@ protected List<FieldConfig> getFieldConfigs() {
     }
 
     FieldConfig nullableTextConfig =
-        new FieldConfig(TEXT_COLUMN_NULL_NAME, FieldConfig.EncodingType.RAW, null, null, null, null, textColumnIndexes,
-            null,
-            null);
+        new FieldConfig.Builder(TEXT_COLUMN_NULL_NAME).withEncodingType(FieldConfig.EncodingType.RAW)
+            .withIndexes(textColumnIndexes).build();
 
     FieldConfig textColumnFieldConfig =
-        new FieldConfig(TEXT_COLUMN_NAME, FieldConfig.EncodingType.RAW, null, null, null, null, textColumnIndexes, null,
-            null);
+        new FieldConfig.Builder(TEXT_COLUMN_NAME).withEncodingType(FieldConfig.EncodingType.RAW)
+            .withIndexes(textColumnIndexes).build();
 
     ObjectNode textColumnCaseSensitiveIndexes;
     try {
@@ -115,8 +114,8 @@ protected List<FieldConfig> getFieldConfigs() {
       throw new RuntimeException(e);
     }
     FieldConfig textColumnCaseSensitiveFieldConfig =
-        new FieldConfig(TEXT_COLUMN_NAME_CASE_SENSITIVE, FieldConfig.EncodingType.RAW, null, null, null, null,
-            textColumnCaseSensitiveIndexes, null, null);
+        new FieldConfig.Builder(TEXT_COLUMN_NAME_CASE_SENSITIVE).withEncodingType(FieldConfig.EncodingType.RAW)
+            .withIndexes(textColumnCaseSensitiveIndexes).build();
     return Arrays.asList(nullableTextConfig, textColumnFieldConfig, textColumnCaseSensitiveFieldConfig);
   }
 

pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/loader/ForwardIndexHandler.java

@@ -33,7 +33,6 @@
 import org.apache.commons.configuration2.PropertiesConfiguration;
 import org.apache.commons.io.FileUtils;
 import org.apache.pinot.segment.local.io.util.PinotDataBitSet;
-import org.apache.pinot.segment.local.segment.creator.impl.BaseSegmentCreator;
 import org.apache.pinot.segment.local.segment.creator.impl.SegmentDictionaryCreator;
 import org.apache.pinot.segment.local.segment.creator.impl.fwd.MultiValueVarByteRawIndexCreator;
 import org.apache.pinot.segment.local.segment.creator.impl.stats.AbstractColumnStatisticsCollector;
@@ -397,16 +396,23 @@ private boolean shouldChangeRawCompressionType(String column, SegmentDirectory.R
     // Get the new compression type.
     ForwardIndexConfig forwardIndexConfig = _fieldIndexConfigs.get(column).getConfig(StandardIndexes.forward());
     ChunkCompressionType newCompressionType = forwardIndexConfig.getChunkCompressionType();
-    if (forwardIndexConfig.getCompressionCodecSpec() != null) {
-      String newCompressionCodecSpec = forwardIndexConfig.getCompressionCodecSpec().toConfigString();
-      String existingCompressionCodecSpec =
-          metadataProperties.getString(getKeyFor(column, FORWARD_INDEX_COMPRESSION_CODEC), null);
-      if (existingCompressionCodecSpec != null) {
-        return !existingCompressionCodecSpec.equals(newCompressionCodecSpec);
+    String existingCompressionCodecSpec =
+        metadataProperties.getString(getKeyFor(column, FORWARD_INDEX_COMPRESSION_CODEC), null);
+    if (forwardIndexConfig.getCompressionCodecSpec() != null
+        && forwardIndexConfig.getCompressionCodecSpec().hasLevel()) {
+      // Segments written before explicit codec metadata was introduced have an unknown level. Rewrite them when the
+      // table config now specifies an explicit level so the persisted forward index matches the configured codec spec.
+      if (existingCompressionCodecSpec == null) {
+        return true;
       }
-      if (forwardIndexConfig.getCompressionCodecSpec().hasLevel()) {
+      String newCompressionCodecSpec = forwardIndexConfig.getCompressionCodecSpec().toConfigString();
+      if (!newCompressionCodecSpec.equals(existingCompressionCodecSpec)) {
         return true;
       }
+    } else if (existingCompressionCodecSpec != null) {
+      // Persisted codec specs are only used for explicit levels. If an existing segment has one and the new
+      // configuration does not, rewrite the forward index to restore the plain codec behavior.
+      return true;
     }
 
     // Note that default compression type (PASS_THROUGH for metric and LZ4 for dimension) is not considered if the
@@ -476,12 +482,8 @@ private void rewriteForwardIndexForCompressionChange(String column, SegmentDirec
     // called during segmentWriter.close().
     segmentWriter.removeIndex(column, StandardIndexes.forward());
     LoaderUtils.writeIndexToV3Format(segmentWriter, column, fwdIndexFile, StandardIndexes.forward());
-    Map<String, String> metadataProperties = new HashMap<>();
-    BaseSegmentCreator.addForwardIndexCompressionCodecInfo(metadataProperties, column,
+    updateForwardIndexCompressionCodecMetadata(column,
         _fieldIndexConfigs.get(column).getConfig(StandardIndexes.forward()), hasDictionary);
-    if (!metadataProperties.isEmpty()) {
-      SegmentMetadataUtils.updateMetadataProperties(_segmentDirectory, metadataProperties);
-    }
 
     // Delete the marker file.
     FileUtils.deleteQuietly(inProgress);
@@ -919,6 +921,7 @@ private void createDictBasedForwardIndex(String column, SegmentDirectory.Writer
     metadataProperties.put(getKeyFor(column, CARDINALITY), String.valueOf(cardinality));
     metadataProperties.put(getKeyFor(column, BITS_PER_ELEMENT),
         String.valueOf(PinotDataBitSet.getNumBitsPerValue(cardinality - 1)));
+    metadataProperties.put(getKeyFor(column, FORWARD_INDEX_COMPRESSION_CODEC), null);
     SegmentMetadataUtils.updateMetadataProperties(_segmentDirectory, metadataProperties);
 
     // We remove indexes that have to be rewritten when a dictEnabled is toggled. Note that the respective index
@@ -971,14 +974,17 @@ private void disableDictionaryAndCreateRawForwardIndex(String column, SegmentDir
 
     LOGGER.info("Created raw forwardIndex. Updating metadata properties for segment={} and column={}", segmentName,
         column);
-    Map<String, String> metadataProperties = new HashMap<>();
-    metadataProperties.put(getKeyFor(column, HAS_DICTIONARY), String.valueOf(false));
-    metadataProperties.put(getKeyFor(column, DICTIONARY_ELEMENT_SIZE), String.valueOf(0));
-    BaseSegmentCreator.addForwardIndexCompressionCodecInfo(metadataProperties, column,
+    PropertiesConfiguration metadataProperties =
+        SegmentMetadataUtils.getPropertiesConfiguration(_segmentDirectory.getSegmentMetadata());
+    metadataProperties.setProperty(getKeyFor(column, HAS_DICTIONARY), String.valueOf(false));
+    metadataProperties.setProperty(getKeyFor(column, DICTIONARY_ELEMENT_SIZE), String.valueOf(0));
+    updateForwardIndexCompressionCodecMetadata(metadataProperties, column,
         _fieldIndexConfigs.get(column).getConfig(StandardIndexes.forward()), false);
     // TODO: See https://github.com/apache/pinot/pull/16921 for details
-    // metadataProperties.put(getKeyFor(column, BITS_PER_ELEMENT), String.valueOf(-1));
-    SegmentMetadataUtils.updateMetadataProperties(_segmentDirectory, metadataProperties);
+    // metadataProperties.setProperty(getKeyFor(column, BITS_PER_ELEMENT), String.valueOf(-1));
+    SegmentMetadataUtils.savePropertiesConfiguration(metadataProperties, _segmentDirectory.getSegmentMetadata()
+        .getIndexDir());
+    _segmentDirectory.reloadMetadata();
 
     // Remove range index, inverted index and FST index.
     removeDictRelatedIndexes(column, segmentWriter);
@@ -989,6 +995,28 @@ private void disableDictionaryAndCreateRawForwardIndex(String column, SegmentDir
     LOGGER.info("Created raw based forward index for segment: {}, column: {}", segmentName, column);
   }
 
+  private void updateForwardIndexCompressionCodecMetadata(String column,
+      @Nullable ForwardIndexConfig forwardIndexConfig, boolean hasDictionary)
+      throws Exception {
+    PropertiesConfiguration metadataProperties =
+        SegmentMetadataUtils.getPropertiesConfiguration(_segmentDirectory.getSegmentMetadata());
+    updateForwardIndexCompressionCodecMetadata(metadataProperties, column, forwardIndexConfig, hasDictionary);
+    SegmentMetadataUtils.savePropertiesConfiguration(metadataProperties, _segmentDirectory.getSegmentMetadata()
+        .getIndexDir());
+    _segmentDirectory.reloadMetadata();
+  }
+
+  private static void updateForwardIndexCompressionCodecMetadata(PropertiesConfiguration metadataProperties,
+      String column, @Nullable ForwardIndexConfig forwardIndexConfig, boolean hasDictionary) {
+    String metadataKey = getKeyFor(column, FORWARD_INDEX_COMPRESSION_CODEC);
+    if (!hasDictionary && forwardIndexConfig != null && forwardIndexConfig.getCompressionCodecSpec() != null
+        && forwardIndexConfig.getCompressionCodecSpec().hasLevel()) {
+      metadataProperties.setProperty(metadataKey, forwardIndexConfig.getCompressionCodecSpec().toConfigString());
+    } else {
+      metadataProperties.clearProperty(metadataKey);
+    }
+  }
+
   private void rewriteDictToRawForwardIndex(ColumnMetadata columnMetadata, SegmentDirectory.Writer segmentWriter,
       File indexDir)
       throws Exception {

pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/loader/InvertedIndexAndDictionaryBasedForwardIndexCreator.java

@@ -280,10 +280,11 @@ private Map<String, String> createForwardIndexForSVColumn()
       writeToForwardIndex(dictionary, context);
 
       // Setup and return the metadata properties to update
+      Map<String, String> metadataProperties = new HashMap<>();
       if (_dictionaryEnabled) {
-        return Map.of();
+        metadataProperties.put(getKeyFor(_columnName, FORWARD_INDEX_COMPRESSION_CODEC), null);
+        return metadataProperties;
       } else {
-        Map<String, String> metadataProperties = new HashMap<>();
         metadataProperties.put(getKeyFor(_columnName, HAS_DICTIONARY), String.valueOf(false));
         metadataProperties.put(getKeyFor(_columnName, DICTIONARY_ELEMENT_SIZE), String.valueOf(0));
         BaseSegmentCreator.addForwardIndexCompressionCodecInfo(metadataProperties, _columnName, _forwardIndexConfig,
@@ -378,7 +379,9 @@ private Map<String, String> createForwardIndexForMVColumn()
       metadataProperties.put(getKeyFor(_columnName, MAX_MULTI_VALUE_ELEMENTS),
           String.valueOf(maxNumberOfMultiValues[0]));
       metadataProperties.put(getKeyFor(_columnName, TOTAL_NUMBER_OF_ENTRIES), String.valueOf(_nextValueId));
-      if (!_dictionaryEnabled) {
+      if (_dictionaryEnabled) {
+        metadataProperties.put(getKeyFor(_columnName, FORWARD_INDEX_COMPRESSION_CODEC), null);
+      } else {
         metadataProperties.put(getKeyFor(_columnName, HAS_DICTIONARY), String.valueOf(false));
         metadataProperties.put(getKeyFor(_columnName, DICTIONARY_ELEMENT_SIZE), String.valueOf(0));
         BaseSegmentCreator.addForwardIndexCompressionCodecInfo(metadataProperties, _columnName, _forwardIndexConfig,

pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/creator/impl/BaseSegmentCreatorTest.java

@@ -0,0 +1,80 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.segment.local.segment.creator.impl;
+
+import java.util.HashMap;
+import java.util.Map;
+import org.apache.commons.configuration2.PropertiesConfiguration;
+import org.apache.pinot.segment.spi.V1Constants;
+import org.apache.pinot.segment.spi.index.ForwardIndexConfig;
+import org.apache.pinot.spi.config.table.CompressionCodecSpec;
+import org.testng.annotations.Test;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertFalse;
+
+
+/**
+ * Unit tests for forward-index compression metadata helpers in {@link BaseSegmentCreator}.
+ */
+public class BaseSegmentCreatorTest {
+  private static final String COLUMN = "rawCol";
+  private static final String METADATA_KEY = V1Constants.MetadataKeys.Column.getKeyFor(COLUMN,
+      V1Constants.MetadataKeys.Column.FORWARD_INDEX_COMPRESSION_CODEC);
+
+  @Test
+  public void testAddsExplicitLeveledCompressionCodecMetadata() {
+    ForwardIndexConfig forwardIndexConfig = new ForwardIndexConfig.Builder()
+        .withCompressionCodecSpec(CompressionCodecSpec.fromString("zstd(3)"))
+        .build();
+
+    PropertiesConfiguration properties = new PropertiesConfiguration();
+    BaseSegmentCreator.addForwardIndexCompressionCodecInfo(properties, COLUMN, forwardIndexConfig, false);
+    assertEquals(properties.getString(METADATA_KEY), "ZSTANDARD(3)");
+
+    Map<String, String> metadataProperties = new HashMap<>();
+    BaseSegmentCreator.addForwardIndexCompressionCodecInfo(metadataProperties, COLUMN, forwardIndexConfig, false);
+    assertEquals(metadataProperties.get(METADATA_KEY), "ZSTANDARD(3)");
+  }
+
+  @Test
+  public void testSkipsImplicitOrDictionaryCompressionCodecMetadata() {
+    ForwardIndexConfig plainCodecConfig = new ForwardIndexConfig.Builder()
+        .withCompressionCodecSpec(CompressionCodecSpec.fromString("LZ4"))
+        .build();
+    PropertiesConfiguration properties = new PropertiesConfiguration();
+    BaseSegmentCreator.addForwardIndexCompressionCodecInfo(properties, COLUMN, plainCodecConfig, false);
+    assertFalse(properties.containsKey(METADATA_KEY));
+
+    Map<String, String> metadataProperties = new HashMap<>();
+    BaseSegmentCreator.addForwardIndexCompressionCodecInfo(metadataProperties, COLUMN, plainCodecConfig, false);
+    assertFalse(metadataProperties.containsKey(METADATA_KEY));
+
+    ForwardIndexConfig leveledCodecConfig = new ForwardIndexConfig.Builder()
+        .withCompressionCodecSpec(CompressionCodecSpec.fromString("GZIP(6)"))
+        .build();
+    PropertiesConfiguration dictionaryProperties = new PropertiesConfiguration();
+    BaseSegmentCreator.addForwardIndexCompressionCodecInfo(dictionaryProperties, COLUMN, leveledCodecConfig, true);
+    assertFalse(dictionaryProperties.containsKey(METADATA_KEY));
+
+    Map<String, String> nullConfigProperties = new HashMap<>();
+    BaseSegmentCreator.addForwardIndexCompressionCodecInfo(nullConfigProperties, COLUMN, null, false);
+    assertFalse(nullConfigProperties.containsKey(METADATA_KEY));
+  }
+}