cleanup spec, drop metrics, tidy up

Author: clintropolis

processing/src/main/java/org/apache/druid/data/input/impl/ClusteredValueGroupsBaseTableProjectionSpec.java

@@ -58,7 +58,6 @@
 import java.io.FileOutputStream;
 import java.io.IOException;
 import java.util.ArrayList;
-import java.util.Arrays;
 import java.util.Comparator;
 import java.util.HashMap;
 import java.util.LinkedHashSet;
@@ -145,7 +144,6 @@ protected File makeIndexFiles(
       return makeClusteredIndexFiles(
           adapters,
           clusterSchemas,
-          segmentMetadata,
           outDir,
           progress,
           mergedMetrics,
@@ -364,7 +362,6 @@ protected File makeIndexFiles(
   private File makeClusteredIndexFiles(
       final List<IndexableAdapter> adapters,
       final List<ClusteredValueGroupsBaseTableSchema> clusterSchemas,
-      final Metadata segmentMetadata,
       final File outDir,
       final ProgressIndicator progress,
       final List<String> mergedMetrics,
@@ -527,8 +524,7 @@ private File makeClusteredIndexFiles(
         }
       }
 
-      // Assemble the merged summary schema. Aggregators use the merged (combining) form from the segment metadata,
-      // consistent with how the non-clustered base table metadata is built.
+      // Assemble the merged summary schema
       final ClusteredValueGroupsBaseTableSchema firstSchema = clusterSchemas.get(0);
       final List<String> summaryColumns = new ArrayList<>();
       summaryColumns.addAll(firstSchema.getClusteringColumns().getColumnNames());
@@ -541,7 +537,6 @@ private File makeClusteredIndexFiles(
       final ClusteredValueGroupsBaseTableSchema mergedSchema = new ClusteredValueGroupsBaseTableSchema(
           firstSchema.getVirtualColumns(),
           summaryColumns,
-          segmentMetadata.getAggregators(),
           firstSchema.getOrdering(),
           firstSchema.getClusteringColumns(),
           null,
@@ -597,18 +592,6 @@ private static MergedClusterGroups mergeClusterSchemas(List<ClusteredValueGroups
           first.getVirtualColumns(),
           schema.getVirtualColumns()
       );
-      // Per-group metric columns are written for the segment-wide merged metric set; the per-group merge derives its
-      // metric formats from only that group's source adapters. All adapters reaching here descend from a single
-      // IncrementalIndexSchema (intermediary persists of one ingest/compaction job), so every cluster group carries
-      // every metric and the two sets coincide. Assert that invariant defensively: a future path that merges
-      // arbitrary clustered segments with differing aggregators would otherwise NPE when writing a group missing a
-      // segment-wide metric.
-      DruidException.conditionalDefensive(
-          Arrays.equals(first.getAggregators(), schema.getAggregators()),
-          "Cannot merge clustered segments with different aggregators: [%s] vs [%s]",
-          Arrays.toString(first.getAggregators()),
-          Arrays.toString(schema.getAggregators())
-      );
     }
 
     // Union per-type dictionaries, sorted with nulls first, plus per-input value -> merged id lookups.

processing/src/main/java/org/apache/druid/segment/IndexMergerV10.java

@@ -25,7 +25,6 @@
 import org.apache.druid.java.util.common.io.Closer;
 import org.apache.druid.query.Order;
 import org.apache.druid.query.OrderBy;
-import org.apache.druid.query.aggregation.AggregatorFactory;
 import org.apache.druid.query.monomorphicprocessing.RuntimeShapeInspector;
 import org.apache.druid.segment.column.BaseColumn;
 import org.apache.druid.segment.column.BaseColumnHolder;
@@ -53,12 +52,11 @@
 import javax.annotation.Nullable;
 import java.io.IOException;
 import java.util.ArrayList;
-import java.util.Arrays;
+import java.util.Collections;
 import java.util.HashSet;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Set;
-import java.util.stream.Collectors;
 
 /**
  *
@@ -155,9 +153,7 @@ public List<String> getMetricNames()
   {
     final ClusteredValueGroupsBaseTableSchema clusteredSummary = input.getClusteredBaseSummary();
     if (clusteredSummary != null) {
-      return Arrays.stream(clusteredSummary.getAggregators())
-                   .map(AggregatorFactory::getName)
-                   .collect(Collectors.toList());
+      return Collections.emptyList();
     }
     final Set<String> columns = Sets.newLinkedHashSet(input.getColumnNames());
     final HashSet<String> dimensions = Sets.newHashSet(availableDimensions);

processing/src/main/java/org/apache/druid/segment/QueryableIndexIndexableAdapter.java

@@ -286,7 +286,7 @@ public QueryableIndex getClusterGroupQueryableIndex(TableClusterGroupSpec groupS
     final Map<String, Supplier<BaseColumnHolder>> groupColumns = clusterGroupColumns.get(index);
     final Metadata groupMetadata = new Metadata(
         null,
-        clusteredBaseSummary.getAggregators(),
+        null,
         null,
         clusteredBaseSummary.getEffectiveGranularity(),
         false,

processing/src/main/java/org/apache/druid/segment/SimpleQueryableIndex.java

@@ -25,9 +25,6 @@
 import org.apache.druid.error.DruidException;
 import org.apache.druid.java.util.common.parsers.ParseException;
 import org.apache.druid.query.OrderBy;
-import org.apache.druid.query.aggregation.Aggregator;
-import org.apache.druid.query.aggregation.AggregatorAndSize;
-import org.apache.druid.query.aggregation.AggregatorFactory;
 import org.apache.druid.segment.ColumnSelectorFactory;
 import org.apache.druid.segment.DimensionIndexer;
 import org.apache.druid.segment.EncodedKeyComponent;
@@ -38,15 +35,13 @@
 import org.apache.druid.segment.column.ColumnFormat;
 import org.apache.druid.segment.column.ColumnHolder;
 import org.apache.druid.segment.column.ColumnType;
-import org.apache.druid.segment.column.ValueType;
 
 import javax.annotation.Nullable;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
-import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicLong;
 
@@ -61,8 +56,9 @@
  * what gives the persisted segment the "share nothing across groups" property — at persist time each group's
  * dictionaries are written under its per-group file-bundle prefix.
  * <p>
- * Aggregator state is also per-group: rollup (when enabled) deduplicates within a group's facts holder, never
- * across groups, since two rows with different clustering tuples land in different groups by construction.
+ * Clustered base tables are never rollup and have no metric columns. When rollup is enabled, the facts holder
+ * deduplicates rows within a group (never across groups, since two rows with different clustering tuples land in
+ * different groups by construction); with no aggregators a rollup hit is simply a no-op dedup.
  */
 public final class OnHeapClusterGroup implements IncrementalIndexRowSelector
 {
@@ -72,24 +68,19 @@ public final class OnHeapClusterGroup implements IncrementalIndexRowSelector
 
   private final List<IncrementalIndex.DimensionDesc> dimensions;
   private final Map<String, IncrementalIndex.DimensionDesc> dimensionsMap;
-  private final Map<String, IncrementalIndex.MetricDesc> aggregatorsMap;
   private final Map<String, ColumnFormat> columnFormats;
-  private final AggregatorFactory[] aggregatorFactories;
   private final FactsHolder factsHolder;
-  private final ConcurrentHashMap<Integer, Aggregator[]> aggregators = new ConcurrentHashMap<>();
   private final AtomicInteger rowCounter = new AtomicInteger(0);
   private final AtomicInteger numEntries = new AtomicInteger(0);
   private final int groupTimePosition;
 
   private final ColumnSelectorFactory virtualSelectorFactory;
-  private final Map<String, ColumnSelectorFactory> aggSelectors;
 
   OnHeapClusterGroup(
       OnHeapClusteredBaseTable parent,
       Object[] clusteringValues,
       List<Integer> clusteringValueIds,
       List<DimensionSchema> nonClusteringDimensions,
-      AggregatorFactory[] aggregatorFactories,
       VirtualColumns virtualColumns,
       IncrementalIndex.InputRowHolder inputRowHolder,
       boolean rollup,
@@ -99,7 +90,6 @@ public final class OnHeapClusterGroup implements IncrementalIndexRowSelector
     this.parent = parent;
     this.clusteringValues = clusteringValues;
     this.clusteringValueIds = Collections.unmodifiableList(new ArrayList<>(clusteringValueIds));
-    this.aggregatorFactories = aggregatorFactories;
 
     this.dimensions = new ArrayList<>(nonClusteringDimensions.size());
     this.dimensionsMap = new LinkedHashMap<>();
@@ -127,9 +117,6 @@ public final class OnHeapClusterGroup implements IncrementalIndexRowSelector
     this.virtualSelectorFactory = new OnheapIncrementalIndex.CachingColumnSelectorFactory(
         IncrementalIndex.makeColumnSelectorFactory(virtualColumns, inputRowHolder, null)
     );
-    this.aggregatorsMap = new LinkedHashMap<>();
-    this.aggSelectors = new LinkedHashMap<>();
-    initializeAggregators(inputRowHolder);
   }
 
   /**
@@ -179,7 +166,7 @@ public List<String> getDimensionNames(boolean includeTime)
   @Override
   public List<String> getMetricNames()
   {
-    return ImmutableList.copyOf(aggregatorsMap.keySet());
+    return List.of();
   }
 
   @Override
@@ -193,7 +180,7 @@ public IncrementalIndex.DimensionDesc getDimension(String columnName)
   @Nullable
   public IncrementalIndex.MetricDesc getMetric(String columnName)
   {
-    return aggregatorsMap.get(columnName);
+    return null;
   }
 
   @Override
@@ -255,32 +242,32 @@ public int getLastRowIndex()
   @Override
   public float getMetricFloatValue(int rowOffset, int aggOffset)
   {
-    return aggregators.get(rowOffset)[aggOffset].getFloat();
+    throw DruidException.defensive("clustered base table groups have no metrics");
   }
 
   @Override
   public long getMetricLongValue(int rowOffset, int aggOffset)
   {
-    return aggregators.get(rowOffset)[aggOffset].getLong();
+    throw DruidException.defensive("clustered base table groups have no metrics");
   }
 
   @Override
   public double getMetricDoubleValue(int rowOffset, int aggOffset)
   {
-    return aggregators.get(rowOffset)[aggOffset].getDouble();
+    throw DruidException.defensive("clustered base table groups have no metrics");
   }
 
   @Override
   @Nullable
   public Object getMetricObjectValue(int rowOffset, int aggOffset)
   {
-    return aggregators.get(rowOffset)[aggOffset].get();
+    throw DruidException.defensive("clustered base table groups have no metrics");
   }
 
   @Override
   public boolean isNull(int rowOffset, int aggOffset)
   {
-    return aggregators.get(rowOffset)[aggOffset].isNull();
+    throw DruidException.defensive("clustered base table groups have no metrics");
   }
 
   @Override
@@ -294,18 +281,9 @@ public ColumnCapabilities getColumnCapabilities(String column)
     if (dim != null) {
       return dim.getCapabilities();
     }
-    final IncrementalIndex.MetricDesc metric = aggregatorsMap.get(column);
-    if (metric != null) {
-      return metric.getCapabilities();
-    }
     return null;
   }
 
-  public AggregatorFactory[] getAggregatorFactories()
-  {
-    return aggregatorFactories;
-  }
-
   /**
    * Add a row's non-clustering content to this group's facts holder. Clustering values aren't re-processed here —
    * they've already been resolved + encoded by the parent {@link OnHeapClusteredBaseTable} as part of selecting
@@ -344,39 +322,18 @@ boolean addToFacts(
     );
 
     final int priorIndex = factsHolder.getPriorIndex(subKey);
-    final Aggregator[] aggs;
     if (IncrementalIndexRow.EMPTY_ROW_INDEX != priorIndex) {
-      aggs = aggregators.get(priorIndex);
-      final long aggSizeDelta = OnheapIncrementalIndex.doAggregate(
-          aggregatorFactories,
-          aggs,
-          parent.getInputRowHolder(),
-          parseExceptionMessages,
-          false
-      );
-      totalSizeInBytes.addAndGet(aggSizeDelta);
+      // with no metric columns there is nothing to aggregate
       return false;
     } else {
-      aggs = new Aggregator[aggregatorFactories.length];
-      long aggSizeForRow = factorizeAggs(aggs);
-      aggSizeForRow += OnheapIncrementalIndex.doAggregate(
-          aggregatorFactories,
-          aggs,
-          parent.getInputRowHolder(),
-          parseExceptionMessages,
-          false
-      );
       final int rowIndex = rowCounter.getAndIncrement();
-      aggregators.put(rowIndex, aggs);
       final int prev = factsHolder.putIfAbsent(subKey, rowIndex);
       if (IncrementalIndexRow.EMPTY_ROW_INDEX == prev) {
         numEntries.incrementAndGet();
       } else {
         throw DruidException.defensive("Encountered existing fact entry for new key in cluster group");
       }
-      final long rowSize = subKey.estimateBytesInMemory()
-                           + aggSizeForRow
-                           + OnheapIncrementalIndex.ROUGH_OVERHEAD_PER_MAP_ENTRY;
+      final long rowSize = subKey.estimateBytesInMemory() + OnheapIncrementalIndex.ROUGH_OVERHEAD_PER_MAP_ENTRY;
       totalSizeInBytes.addAndGet(rowSize);
       return true;
     }
@@ -400,39 +357,4 @@ private void initializeDimensions(List<DimensionSchema> nonClusteringDimensions)
       columnFormats.put(schema.getName(), desc.getIndexer().getFormat());
     }
   }
-
-  private void initializeAggregators(IncrementalIndex.InputRowHolder inputRowHolder)
-  {
-    for (AggregatorFactory agg : aggregatorFactories) {
-      final IncrementalIndex.MetricDesc metricDesc = new IncrementalIndex.MetricDesc(aggregatorsMap.size(), agg);
-      aggregatorsMap.put(metricDesc.getName(), metricDesc);
-      columnFormats.put(
-          metricDesc.getName(),
-          new CapabilitiesBasedFormat(metricDesc.getCapabilities())
-      );
-      final ColumnSelectorFactory factory;
-      if (agg.getIntermediateType().is(ValueType.COMPLEX)) {
-        factory = new OnheapIncrementalIndex.CachingColumnSelectorFactory(
-            IncrementalIndex.makeColumnSelectorFactory(VirtualColumns.EMPTY, inputRowHolder, agg)
-        );
-      } else {
-        factory = virtualSelectorFactory;
-      }
-      aggSelectors.put(agg.getName(), factory);
-    }
-  }
-
-  private long factorizeAggs(Aggregator[] aggs)
-  {
-    long totalInitialSizeBytes = 0L;
-    final long aggReferenceSize = Long.BYTES;
-    for (int i = 0; i < aggregatorFactories.length; i++) {
-      final AggregatorFactory agg = aggregatorFactories[i];
-      final AggregatorAndSize aggregatorAndSize = agg.factorizeWithSize(aggSelectors.get(agg.getName()));
-      aggs[i] = aggregatorAndSize.getAggregator();
-      totalInitialSizeBytes += aggregatorAndSize.getInitialSizeBytes();
-      totalInitialSizeBytes += aggReferenceSize;
-    }
-    return totalInitialSizeBytes;
-  }
 }