add ability to customize json (and auto) column physical storage details on per column basis

benchmarks/src/test/java/org/apache/druid/benchmark/FilterPartitionBenchmark.java

@@ -166,7 +166,7 @@ public void setup() throws IOException
     indexFile = INDEX_MERGER_V9.persist(
         incIndex,
         tmpDir,
-        IndexSpec.DEFAULT,
+        IndexSpec.getDefault(),
         null
     );
     qIndex = INDEX_IO.loadIndex(indexFile);

benchmarks/src/test/java/org/apache/druid/benchmark/FilteredAggregatorBenchmark.java

@@ -272,7 +272,7 @@ public void setup(FilteredAggregatorBenchmark global) throws IOException
       File indexFile = INDEX_MERGER_V9.persist(
           incIndex,
           qIndexesDir,
-          IndexSpec.DEFAULT,
+          IndexSpec.getDefault(),
           null
       );
       incIndex.close();

benchmarks/src/test/java/org/apache/druid/benchmark/GroupByTypeInterfaceBenchmark.java

@@ -310,7 +310,7 @@ public void setup() throws IOException
       final File file = INDEX_MERGER_V9.persist(
           index,
           new File(tmpDir, String.valueOf(i)),
-          IndexSpec.DEFAULT,
+          IndexSpec.getDefault(),
           null
       );
 

benchmarks/src/test/java/org/apache/druid/benchmark/TopNTypeInterfaceBenchmark.java

@@ -277,7 +277,7 @@ public void setup() throws IOException
       File indexFile = INDEX_MERGER_V9.persist(
           incIndexes.get(i),
           tmpFile,
-          IndexSpec.DEFAULT,
+          IndexSpec.getDefault(),
           null
       );
 

benchmarks/src/test/java/org/apache/druid/benchmark/indexing/IndexMergeBenchmark.java

@@ -134,7 +134,7 @@ public void setup() throws IOException
       File indexFile = indexMergerV9.persist(
           incIndex,
           tmpDir,
-          IndexSpec.DEFAULT,
+          IndexSpec.getDefault(),
           null
       );
 
@@ -159,7 +159,7 @@ public void mergeV9(Blackhole blackhole) throws Exception
           rollup,
           schemaInfo.getAggsArray(),
           tmpFile,
-          IndexSpec.DEFAULT,
+          IndexSpec.getDefault(),
           null,
           -1
       );

benchmarks/src/test/java/org/apache/druid/benchmark/indexing/IndexPersistBenchmark.java

@@ -185,7 +185,7 @@ public void persistV9(Blackhole blackhole) throws Exception
     File indexFile = INDEX_MERGER_V9.persist(
         incIndex,
         tmpDir,
-        IndexSpec.DEFAULT,
+        IndexSpec.getDefault(),
         null
     );
 

benchmarks/src/test/java/org/apache/druid/benchmark/lookup/SqlReverseLookupBenchmark.java

@@ -105,7 +105,7 @@ public void setup()
     final SegmentGenerator segmentGenerator = closer.register(new SegmentGenerator());
 
     final QueryableIndex index =
-        segmentGenerator.generate(dataSegment, schemaInfo, IndexSpec.DEFAULT, Granularities.NONE, 1);
+        segmentGenerator.generate(dataSegment, schemaInfo, IndexSpec.getDefault(), Granularities.NONE, 1);
 
     final Pair<PlannerFactory, SqlEngine> sqlSystem = SqlBaseBenchmark.createSqlSystem(
         ImmutableMap.of(dataSegment, index),

benchmarks/src/test/java/org/apache/druid/benchmark/query/GroupByBenchmark.java

@@ -578,7 +578,7 @@ public void setup(GroupByBenchmark global) throws IOException
         File indexFile = INDEX_MERGER_V9.persist(
             incIndex,
             new File(qIndexesDir, String.valueOf(i)),
-            IndexSpec.DEFAULT,
+            IndexSpec.getDefault(),
             null
         );
         incIndex.close();

benchmarks/src/test/java/org/apache/druid/benchmark/query/InPlanningBenchmark.java

@@ -162,14 +162,14 @@ public void setup() throws JsonProcessingException
     List<DimensionSchema> columnSchemas = schemaInfo.getDimensionsSpec()
                                                     .getDimensions()
                                                     .stream()
-                                                    .map(x -> new AutoTypeColumnSchema(x.getName(), null))
+                                                    .map(x -> AutoTypeColumnSchema.of(x.getName()))
                                                     .collect(Collectors.toList());
     index = segmentGenerator.generate(
         dataSegment,
         schemaInfo,
         DimensionsSpec.builder().setDimensions(columnSchemas).build(),
         TransformSpec.NONE,
-        IndexSpec.DEFAULT,
+        IndexSpec.getDefault(),
         Granularities.NONE,
         rowsPerSegment
     );

benchmarks/src/test/java/org/apache/druid/benchmark/query/ScanBenchmark.java

@@ -330,7 +330,7 @@ public void setup(ScanBenchmark global) throws IOException
         File indexFile = INDEX_MERGER_V9.persist(
             incIndex,
             new File(qIndexesDir, String.valueOf(i)),
-            IndexSpec.DEFAULT,
+            IndexSpec.getDefault(),
             null
         );
         incIndex.close();

benchmarks/src/test/java/org/apache/druid/benchmark/query/SearchBenchmark.java

@@ -397,7 +397,7 @@ public void setup(SearchBenchmark global) throws IOException
         File indexFile = INDEX_MERGER_V9.persist(
             incIndex,
             new File(qIndexesDir, String.valueOf(i)),
-            IndexSpec.DEFAULT,
+            IndexSpec.getDefault(),
             null
         );
         incIndex.close();

benchmarks/src/test/java/org/apache/druid/benchmark/query/SqlBaseBenchmark.java

@@ -226,7 +226,8 @@ protected IndexSpec getIndexSpec()
                     .withComplexMetricCompression(
                         CompressionStrategy.valueOf(StringUtils.toUpperCase(complexCompression))
                     )
-                    .build();
+                    .build()
+                    .getEffectiveSpec();
   }
 
   @Setup(Level.Trial)

benchmarks/src/test/java/org/apache/druid/benchmark/query/SqlBenchmarkDatasets.java

@@ -116,7 +116,7 @@ public class SqlBenchmarkDatasets
                 ImmutableList.copyOf(
                     Iterables.concat(
                         expressionsSchema.getDimensionsSpecExcludeAggs().getDimensions(),
-                        Collections.singletonList(new AutoTypeColumnSchema("nested", null))
+                        Collections.singletonList(AutoTypeColumnSchema.of("nested"))
                     )
                 )
             ).build(),
@@ -414,7 +414,7 @@ public BenchmarkSchema asAutoDimensions()
           dimensionsSpec.withDimensions(
                     dimensionsSpec.getDimensions()
                                   .stream()
-                                  .map(dim -> new AutoTypeColumnSchema(dim.getName(), null))
+                                  .map(dim -> AutoTypeColumnSchema.of(dim.getName()))
                                   .collect(Collectors.toList())
           ),
           aggregators,
@@ -425,7 +425,7 @@ public BenchmarkSchema asAutoDimensions()
                                                     .groupingColumns(
                                                         projection.getGroupingColumns()
                                                                   .stream()
-                                                                  .map(dim -> new AutoTypeColumnSchema(dim.getName(), null))
+                                                                  .map(dim -> AutoTypeColumnSchema.of(dim.getName()))
                                                                   .collect(Collectors.toList())
                                                     )
                                                     .build()

benchmarks/src/test/java/org/apache/druid/benchmark/query/TimeseriesBenchmark.java

@@ -325,7 +325,7 @@ public void setup(TimeseriesBenchmark global) throws IOException
         File indexFile = INDEX_MERGER_V9.persist(
             incIndex,
             new File(qIndexesDir, String.valueOf(i)),
-            IndexSpec.DEFAULT,
+            IndexSpec.getDefault(),
             null
         );
         incIndex.close();

benchmarks/src/test/java/org/apache/druid/benchmark/query/TopNBenchmark.java

@@ -301,7 +301,7 @@ public void setup(TopNBenchmark global) throws IOException
         File indexFile = INDEX_MERGER_V9.persist(
             incIndex,
             new File(qIndexesDir, String.valueOf(i)),
-            IndexSpec.DEFAULT,
+            IndexSpec.getDefault(),
             null
         );
         incIndex.close();

benchmarks/src/test/java/org/apache/druid/benchmark/query/timecompare/TimeCompareBenchmark.java

@@ -329,7 +329,7 @@ public void setup() throws IOException
       File indexFile = INDEX_MERGER_V9.persist(
           incIndexes.get(i),
           tmpDir,
-          IndexSpec.DEFAULT,
+          IndexSpec.getDefault(),
           null
       );
 

embedded-tests/src/test/java/org/apache/druid/testing/embedded/compact/AutoCompactionTest.java

@@ -62,7 +62,9 @@
 import org.apache.druid.query.aggregation.datasketches.theta.SketchModule;
 import org.apache.druid.query.filter.SelectorDimFilter;
 import org.apache.druid.segment.AutoTypeColumnSchema;
+import org.apache.druid.segment.IndexSpec;
 import org.apache.druid.segment.column.ColumnType;
+import org.apache.druid.segment.nested.NestedCommonFormatColumnFormatSpec;
 import org.apache.druid.segment.transform.CompactionTransformSpec;
 import org.apache.druid.server.compaction.FixedIntervalOrderPolicy;
 import org.apache.druid.server.coordinator.AutoCompactionSnapshot;
@@ -503,7 +505,7 @@ public void testAutoCompactionPreservesCreateBitmapIndexInDimensionSchema(Compac
 
       List<DimensionSchema> dimensionSchemas = ImmutableList.of(
           new StringDimensionSchema("language", DimensionSchema.MultiValueHandling.SORTED_ARRAY, false),
-          new AutoTypeColumnSchema("deleted", ColumnType.DOUBLE)
+          new AutoTypeColumnSchema("deleted", ColumnType.DOUBLE, null)
       );
 
       submitCompactionConfig(
@@ -519,7 +521,21 @@ public void testAutoCompactionPreservesCreateBitmapIndexInDimensionSchema(Compac
       // Compacted into 1 segment for the entire year.
       forceTriggerAutoCompaction(1);
       verifySegmentsCompacted(1, MAX_ROWS_PER_SEGMENT_COMPACTED);
-      verifySegmentsCompactedDimensionSchema(dimensionSchemas);
+      List<DimensionSchema> expectedDimensionSchemas = List.of(
+          dimensionSchemas.get(0),
+          new AutoTypeColumnSchema(
+              "deleted",
+              ColumnType.DOUBLE,
+              // json serialization doesn't store bitmap in compaction state, so need to set to null
+              NestedCommonFormatColumnFormatSpec.builder(
+                  NestedCommonFormatColumnFormatSpec.getEffectiveFormatSpec(
+                      null,
+                      IndexSpec.getDefault().getEffectiveSpec()
+                  )
+              ).setBitmapEncoding(null).build()
+          )
+      );
+      verifySegmentsCompactedDimensionSchema(expectedDimensionSchemas);
     }
   }
 

extensions-contrib/druid-deltalake-extensions/src/test/java/org/apache/druid/delta/input/ComplexTypesDeltaTable.java

@@ -124,11 +124,11 @@ public class ComplexTypesDeltaTable
       new TimestampSpec("na", "posix", DateTimes.of("2024-01-01")),
       new DimensionsSpec(
           ImmutableList.of(
-              new AutoTypeColumnSchema("id", null),
-              new AutoTypeColumnSchema("array_info", null),
-              new AutoTypeColumnSchema("struct_info", null),
-              new AutoTypeColumnSchema("nested_struct_info", null),
-              new AutoTypeColumnSchema("map_info", null)
+              AutoTypeColumnSchema.of("id"),
+              AutoTypeColumnSchema.of("array_info"),
+              AutoTypeColumnSchema.of("struct_info"),
+              AutoTypeColumnSchema.of("nested_struct_info"),
+              AutoTypeColumnSchema.of("map_info")
           )
       ),
       ColumnsFilter.all()

extensions-contrib/druid-deltalake-extensions/src/test/java/org/apache/druid/delta/input/SnapshotDeltaTable.java

@@ -120,8 +120,8 @@ public class SnapshotDeltaTable
       new TimestampSpec("na", "posix", DateTimes.of("2024-01-01")),
       new DimensionsSpec(
           ImmutableList.of(
-              new AutoTypeColumnSchema("id", null),
-              new AutoTypeColumnSchema("map_info", null)
+              AutoTypeColumnSchema.of("id"),
+              AutoTypeColumnSchema.of("map_info")
           )
       ),
       ColumnsFilter.all()

extensions-contrib/rabbit-stream-indexing-service/src/test/java/org/apache/druid/indexing/rabbitstream/RabbitStreamIndexTaskTuningConfigTest.java

@@ -65,7 +65,7 @@ public void testSerdeWithDefaults() throws Exception
     Assert.assertEquals(5_000_000, config.getMaxRowsPerSegment().intValue());
     Assert.assertEquals(new Period("PT10M"), config.getIntermediatePersistPeriod());
     Assert.assertEquals(0, config.getMaxPendingPersists());
-    // Assert.assertEquals(IndexSpec.DEFAULT, config.getIndexSpec());
+    // Assert.assertEquals(IndexSpec.getDefault(), config.getIndexSpec());
     Assert.assertEquals(false, config.isReportParseExceptions());
     Assert.assertEquals(Duration.ofMinutes(15).toMillis(), config.getHandoffConditionTimeout());
 
@@ -155,12 +155,13 @@ public void testtoString() throws Exception
                     "intermediatePersistPeriod=PT1H, " +
                     "maxPendingPersists=100, " +
                     "indexSpec=IndexSpec{" +
-                    "bitmapSerdeFactory=RoaringBitmapSerdeFactory{}, " +
-                    "dimensionCompression=lz4, " +
-                    "stringDictionaryEncoding=Utf8{}, " +
-                    "metricCompression=lz4, " +
-                    "longEncoding=longs, " +
+                    "bitmapSerdeFactory=null, " +
+                    "dimensionCompression=null, " +
+                    "stringDictionaryEncoding=null, " +
+                    "metricCompression=null, " +
+                    "longEncoding=null, " +
                     "complexMetricCompression=null, " +
+                    "autoColumnFormatSpec=null, " +
                     "jsonCompression=null, " +
                     "segmentLoader=null" +
                     "}, " +

extensions-contrib/rabbit-stream-indexing-service/src/test/java/org/apache/druid/indexing/rabbitstream/supervisor/RabbitStreamSupervisorTuningConfigTest.java

@@ -63,7 +63,7 @@ public void testSerdeWithDefaults() throws Exception
     Assert.assertEquals(5_000_000, config.getMaxRowsPerSegment().intValue());
     Assert.assertEquals(new Period("PT10M"), config.getIntermediatePersistPeriod());
     Assert.assertEquals(0, config.getMaxPendingPersists());
-    // Assert.assertEquals(IndexSpec.DEFAULT, config.getIndexSpec());
+    // Assert.assertEquals(IndexSpec.getDefault(), config.getIndexSpec());
     Assert.assertEquals(false, config.isReportParseExceptions());
     Assert.assertEquals(java.time.Duration.ofMinutes(15).toMillis(), config.getHandoffConditionTimeout());
     Assert.assertNull(config.getWorkerThreads());

extensions-core/avro-extensions/src/test/java/org/apache/druid/data/input/AvroStreamInputFormatTest.java

@@ -303,15 +303,15 @@ public void testParseTransformNested() throws SchemaValidationException, IOExcep
 
     DimensionsSpec dimensionsSpec = new DimensionsSpec(
         ImmutableList.of(
-            new AutoTypeColumnSchema("someIntValueMap", null),
-            new AutoTypeColumnSchema("someStringValueMap", null),
-            new AutoTypeColumnSchema("someRecord", null),
-            new AutoTypeColumnSchema("someRecordArray", null),
+            AutoTypeColumnSchema.of("someIntValueMap"),
+            AutoTypeColumnSchema.of("someStringValueMap"),
+            AutoTypeColumnSchema.of("someRecord"),
+            AutoTypeColumnSchema.of("someRecordArray"),
             new LongDimensionSchema("tSomeIntValueMap8"),
             new LongDimensionSchema("tSomeIntValueMap8_2"),
             new StringDimensionSchema("tSomeStringValueMap8"),
             new LongDimensionSchema("tSomeRecordSubLong"),
-            new AutoTypeColumnSchema("tSomeRecordArray0", null),
+            AutoTypeColumnSchema.of("tSomeRecordArray0"),
             new StringDimensionSchema("tSomeRecordArray0nestedString")
         )
     );

extensions-core/datasketches/src/test/java/org/apache/druid/segment/DatasketchesProjectionTest.java

@@ -129,7 +129,7 @@ public class DatasketchesProjectionTest extends InitializedNullHandlingTest
                                                 .groupingColumns(
                                                     projection.getGroupingColumns()
                                                               .stream()
-                                                              .map(x -> new AutoTypeColumnSchema(x.getName(), null))
+                                                              .map(x -> AutoTypeColumnSchema.of(x.getName()))
                                                               .collect(Collectors.toList())
                                                 )
                                                 .build()
@@ -167,7 +167,7 @@ public static Collection<?> constructorFeeder()
 
     List<DimensionSchema> autoDims = dimsOrdered.getDimensions()
                                                 .stream()
-                                                .map(x -> new AutoTypeColumnSchema(x.getName(), null))
+                                                .map(x -> AutoTypeColumnSchema.of(x.getName()))
                                                 .collect(Collectors.toList());
     for (boolean incremental : new boolean[]{true, false}) {
       for (boolean sortByDim : new boolean[]{true, false}) {

extensions-core/kafka-indexing-service/src/test/java/org/apache/druid/indexing/kafka/KafkaIndexTaskTuningConfigTest.java

@@ -69,8 +69,8 @@ public void testSerdeWithDefaults() throws Exception
     Assert.assertNull(config.getMaxTotalRows());
     Assert.assertEquals(new Period("PT10M"), config.getIntermediatePersistPeriod());
     Assert.assertEquals(0, config.getMaxPendingPersists());
-    Assert.assertEquals(IndexSpec.DEFAULT, config.getIndexSpec());
-    Assert.assertEquals(IndexSpec.DEFAULT, config.getIndexSpecForIntermediatePersists());
+    Assert.assertEquals(IndexSpec.getDefault(), config.getIndexSpec());
+    Assert.assertEquals(IndexSpec.getDefault(), config.getIndexSpecForIntermediatePersists());
     Assert.assertFalse(config.isReportParseExceptions());
     Assert.assertEquals(Duration.ofMinutes(15).toMillis(), config.getHandoffConditionTimeout());
     Assert.assertEquals(1, config.getNumPersistThreads());
@@ -139,8 +139,8 @@ public void testConvert()
         .withMaxRowsPerSegment(2)
         .withMaxTotalRows(10L)
         .withMaxPendingPersists(4)
-        .withIndexSpec(IndexSpec.DEFAULT)
-        .withIndexSpecForIntermediatePersists(IndexSpec.DEFAULT)
+        .withIndexSpec(IndexSpec.getDefault())
+        .withIndexSpecForIntermediatePersists(IndexSpec.getDefault())
         .withReportParseExceptions(true)
         .withMaxColumnsToMerge(5)
         .build();
@@ -154,7 +154,7 @@ public void testConvert()
     Assert.assertEquals(new Period("PT3S"), copy.getIntermediatePersistPeriod());
     Assert.assertNull(copy.getBasePersistDirectory());
     Assert.assertEquals(4, copy.getMaxPendingPersists());
-    Assert.assertEquals(IndexSpec.DEFAULT, copy.getIndexSpec());
+    Assert.assertEquals(IndexSpec.getDefault(), copy.getIndexSpec());
     Assert.assertTrue(copy.isReportParseExceptions());
     Assert.assertEquals(5L, copy.getHandoffConditionTimeout());
     Assert.assertEquals(2, copy.getNumPersistThreads());
@@ -174,8 +174,8 @@ public void testSerdeWithModifiedTuningConfigAddedField() throws IOException
         new Period("PT3S"),
         new File("/tmp/xxx"),
         4,
-        IndexSpec.DEFAULT,
-        IndexSpec.DEFAULT,
+        IndexSpec.getDefault(),
+        IndexSpec.getDefault(),
         true,
         5L,
         null,
@@ -227,8 +227,8 @@ public void testSerdeWithModifiedTuningConfigRemovedField() throws IOException
         10L,
         new Period("PT3S"),
         4,
-        IndexSpec.DEFAULT,
-        IndexSpec.DEFAULT,
+        IndexSpec.getDefault(),
+        IndexSpec.getDefault(),
         true,
         5L,
         null,
@@ -273,7 +273,7 @@ public void testEqualsAndHashCode()
     EqualsVerifier.forClass(KafkaIndexTaskTuningConfig.class)
                   .withPrefabValues(
                       IndexSpec.class,
-                      IndexSpec.DEFAULT,
+                      IndexSpec.getDefault(),
                       IndexSpec.builder().withDimensionCompression(CompressionStrategy.ZSTD).build()
                   )
                   .usingGetClass()

extensions-core/kafka-indexing-service/src/test/java/org/apache/druid/indexing/kafka/supervisor/KafkaSupervisorTuningConfigTest.java

@@ -62,8 +62,8 @@ public void testSerdeWithDefaults() throws Exception
     Assert.assertEquals(5_000_000, config.getMaxRowsPerSegment().intValue());
     Assert.assertEquals(new Period("PT10M"), config.getIntermediatePersistPeriod());
     Assert.assertEquals(0, config.getMaxPendingPersists());
-    Assert.assertEquals(IndexSpec.DEFAULT, config.getIndexSpec());
-    Assert.assertEquals(IndexSpec.DEFAULT, config.getIndexSpecForIntermediatePersists());
+    Assert.assertEquals(IndexSpec.getDefault(), config.getIndexSpec());
+    Assert.assertEquals(IndexSpec.getDefault(), config.getIndexSpecForIntermediatePersists());
     Assert.assertFalse(config.isReportParseExceptions());
     Assert.assertEquals(java.time.Duration.ofMinutes(15).toMillis(), config.getHandoffConditionTimeout());
     Assert.assertNull(config.getWorkerThreads());