FrankChen021 commented on code in PR #20452:
URL: https://github.com/apache/druid/pull/20452#discussion_r4144657295
##########
sql/src/main/java/org/apache/druid/sql/calcite/planner/ProjectionSpecTranslator.java:
##########
@@ -288,6 +306,128 @@ private static void collectDependencies(
}
}
+ /**
+ * Lift a GROUP BY body into the rollup layout's metadata. The body plans
like an aggregate projection body, and its
+ * outputs must name every declared column in declared order: an output
backed by a grouping dimension must reference
+ * the declared column itself (or {@code TIME_FLOOR(__time, <period>) AS
__time}, which becomes the query
+ * granularity carrier) and an output backed by an aggregate becomes the
aggregator filling the declared metric
+ * column it is named for.
+ */
+ private static RollupTableBaseTableMetadata liftRollupBaseTable(
+ final String tableName,
+ final List<ColumnSpec> columns,
+ final DruidQuery druidQuery
+ )
+ {
+ final PlannedAggregation planned =
plannedAggregation(BASE_PROJECTION_NAME, tableName, druidQuery);
+ if (planned.virtualColumnsAndFilter.filter(BASE_PROJECTION_NAME) != null) {
+ throw invalid(BASE_PROJECTION_NAME, "its body filters. The base table
stores every ingested row");
+ }
+
+ final List<String> outputNames =
druidQuery.getOutputRowType().getFieldNames();
+ if (outputNames.size() != columns.size()) {
+ throw invalid(
+ BASE_PROJECTION_NAME,
+ StringUtils.format(
+ "it selects %d column(s) but the table declares %d. The body
lists the columns in the order segments"
+ + " store them, so it must name every declared column",
+ outputNames.size(),
+ columns.size()
+ )
+ );
+ }
+
+ final Map<String, DimensionSpec> dimensionsByInternalName = new
HashMap<>();
+ for (DimensionSpec dimension : planned.dimensions) {
+ dimensionsByInternalName.put(dimension.getOutputName(), dimension);
+ }
+ final Map<String, AggregatorFactory> aggregatorsByInternalName = new
HashMap<>();
+ for (AggregatorFactory aggregator : planned.aggregators) {
+ aggregatorsByInternalName.put(aggregator.getName(), aggregator);
+ }
+
+ final RowSignature sources = druidQuery.getOutputRowSignature();
+ final List<VirtualColumn> carrier = new ArrayList<>(1);
+ final List<AggregatorFactory> aggregators = new
ArrayList<>(planned.aggregators.size());
+ for (int i = 0; i < columns.size(); i++) {
+ final String declared = columns.get(i).name();
+ if (!declared.equals(outputNames.get(i))) {
+ throw invalid(
+ BASE_PROJECTION_NAME,
+ StringUtils.format(
+ "its column %d is [%s] but the table declares [%s] there. The
body lists the columns in the order"
+ + " segments store them",
+ i + 1,
+ outputNames.get(i),
+ declared
+ )
+ );
+ }
+ final String source = sources.getColumnName(i);
+ final AggregatorFactory aggregator =
aggregatorsByInternalName.get(source);
+ if (aggregator != null) {
+ aggregators.add(aggregator.withName(declared));
Review Comment:
[P1] Use combining aggregators for rollup reaggregation
**Finding:** This keeps the planner's raw-ingestion factory instead of the
combining factory. For an accepted body such as COUNT(*) AS rows, the new
rollup base-table spec is fed directly into MSQ compaction, whose GroupBy uses
dataSchema.getAggregators() on already-stored rows, so each collapsed row
contributes 1 rather than its stored count. The same issue affects sketch-build
aggregators and any aggregator whose input field differs from its output,
silently changing results on compaction or re-rollup.
**Suggestion:** Make the compaction/reaggregation schema use each
aggregator's combining form, or reject and normalize rollup definitions whose
factory is not self-combining; add a COUNT/sketch compaction round-trip test.
##########
processing/src/main/java/org/apache/druid/data/input/impl/RollupTableProjectionSpec.java:
##########
@@ -0,0 +1,333 @@
+/*
+ * 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.druid.data.input.impl;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonIgnore;
+import com.fasterxml.jackson.annotation.JsonInclude;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.annotation.JsonTypeName;
+import org.apache.druid.error.InvalidInput;
+import org.apache.druid.java.util.common.granularity.Granularities;
+import org.apache.druid.java.util.common.granularity.Granularity;
+import org.apache.druid.query.OrderBy;
+import org.apache.druid.query.aggregation.AggregatorFactory;
+import org.apache.druid.segment.VirtualColumn;
+import org.apache.druid.segment.VirtualColumns;
+import org.apache.druid.segment.column.ColumnHolder;
+
+import javax.annotation.Nullable;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Objects;
+import java.util.Set;
+
+/**
+ * {@link BaseTableProjectionSpec} for a rollup table: the plain {@link
TableProjectionSpec} shape plus
+ * {@link #aggregators}. The declared {@link #groupingColumns} are the
grouping columns, rows with identical grouping values
+ * (after query-granularity flooring of {@code __time}) are aggregated into
one at ingest time, with each aggregator
+ * producing a metric column after the grouping columns. As in the plain
layout, the time position is an explicit
+ * positional entry in {@link #groupingColumns} named {@code __time}, declared
column order is the segment storage and sort
+ * order, and query granularity rides as the {@link
Granularities#GRANULARITY_VIRTUAL_COLUMN_NAME} carrier virtual
+ * column (the only virtual column accepted).
+ * <p>
+ * Operator-facing counterpart of the segment metadata-side
+ * {@link org.apache.druid.segment.projections.RollupTableProjectionSchema}.
+ * <p>
+ * Aggregators may be empty: a rollup table without metrics collapses rows
with identical grouping values, mirroring
+ * the legacy {@code rollup=true} granularity spec without a {@code
metricsSpec}.
+ */
+@JsonTypeName(RollupTableProjectionSpec.TYPE_NAME)
+public final class RollupTableProjectionSpec implements BaseTableProjectionSpec
+{
+ public static final String TYPE_NAME = "rollupTable";
+
+ private final VirtualColumns virtualColumns;
+ private final List<DimensionSchema> groupingColumns;
+ private final AggregatorFactory[] aggregators;
+ private final DimensionsSpec dimensionsSpec;
+ private final List<OrderBy> ordering;
+
+ public static Builder builder()
+ {
+ return new Builder();
+ }
+
+ @JsonCreator
+ public RollupTableProjectionSpec(
+ @JsonProperty("virtualColumns") @Nullable VirtualColumns virtualColumns,
+ @JsonProperty("groupingColumns") List<DimensionSchema> groupingColumns,
+ @JsonProperty("aggregators") @Nullable AggregatorFactory[] aggregators
+ )
+ {
+ BaseTableProjectionSpec.validateDeclaredColumns(groupingColumns,
"groupingColumns", TYPE_NAME);
+ this.virtualColumns = virtualColumns == null ? VirtualColumns.EMPTY :
virtualColumns;
+
BaseTableProjectionSpec.validateGranularityOnlyVirtualColumns(this.virtualColumns,
TYPE_NAME);
+ this.groupingColumns = Collections.unmodifiableList(new
ArrayList<>(groupingColumns));
+ this.aggregators = aggregators == null ? new AggregatorFactory[0] :
aggregators;
+ validateAggregators(this.groupingColumns, this.aggregators);
+ this.dimensionsSpec = DimensionsSpec.builder()
+ .setDimensions(this.groupingColumns)
+ .setForceSegmentSortByTime(false)
+ .build();
+ this.ordering =
BaseTableProjectionSpec.declaredOrderAscending(this.groupingColumns);
+ }
+
+ @Override
+ @JsonProperty
+ @JsonInclude(JsonInclude.Include.NON_DEFAULT)
+ public VirtualColumns getVirtualColumns()
+ {
+ return virtualColumns;
+ }
+
+ /**
+ * The full, ordered list of grouping columns in segment order, with the
explicit {@code __time} (or
+ * query-granularity) marker at its declared position.
+ */
+ @JsonProperty("groupingColumns")
+ public List<DimensionSchema> getGroupingColumns()
+ {
+ return groupingColumns;
+ }
+
+ /**
+ * The aggregators computed over the rows collapsed into each grouping
tuple, each producing a metric column stored
+ * after the grouping columns.
+ */
+ @Override
+ @JsonProperty("aggregators")
+ @JsonInclude(JsonInclude.Include.NON_EMPTY)
+ public AggregatorFactory[] getMetrics()
+ {
+ return aggregators;
+ }
+
+ @Override
+ @JsonIgnore
+ public List<OrderBy> getOrdering()
+ {
+ return ordering;
+ }
+
+ @Override
+ @JsonIgnore
+ public DimensionsSpec getDimensionsSpec()
+ {
+ return dimensionsSpec;
+ }
+
+ @Override
+ public boolean isRollup()
+ {
+ return true;
+ }
+
+ /**
+ * Returns a copy of this spec with a new {@code queryGranularity},
expressed as a
+ * {@link Granularities#GRANULARITY_VIRTUAL_COLUMN_NAME} virtual column
added to {@link #getVirtualColumns()}. A
+ * {@code null} or {@code NONE} granularity is a no-op (no flooring), so
this returns {@code this} unchanged.
+ * <p>
+ * {@code ALL} is rejected: it has no granularity virtual column
representation, so accepting it would silently
+ * degrade to {@code NONE} when the spec is read back.
+ * <p>
+ * Idempotent: if the spec already declares a query-granularity virtual
column, that one is authoritative and this is
+ * a no-op. (The compaction path attaches the virtual column up front; the
MSQ generation path then calls this again
+ * with the query-derived granularity, which must not double-add.)
+ */
+ @Override
+ public RollupTableProjectionSpec withQueryGranularity(@Nullable Granularity
queryGranularity)
+ {
+ if (Granularities.ALL.equals(queryGranularity)) {
+ throw InvalidInput.exception(
+ "Query granularity[ALL] is not supported for [%s] base tables",
+ TYPE_NAME
+ );
+ }
+ if (queryGranularity == null
+ || Granularities.NONE.equals(queryGranularity)
+ ||
virtualColumns.getVirtualColumn(Granularities.GRANULARITY_VIRTUAL_COLUMN_NAME)
!= null) {
+ return this;
+ }
+ BaseTableProjectionSpec.validateQueryGranularity(queryGranularity,
TYPE_NAME);
+ final VirtualColumn granularityVirtualColumn =
+ Granularities.toVirtualColumn(queryGranularity,
Granularities.GRANULARITY_VIRTUAL_COLUMN_NAME);
+ final List<VirtualColumn> merged = new
ArrayList<>(Arrays.asList(virtualColumns.getVirtualColumns()));
+ merged.add(granularityVirtualColumn);
+ return new RollupTableProjectionSpec(VirtualColumns.create(merged),
groupingColumns, aggregators);
+ }
+
+ /**
+ * Compares rollup-spec state for compaction up-to-date checks: query
granularity is compared separately (via its
+ * carrier virtual column), so it is stripped from both sides; everything
else (the grouping columns, the
+ * aggregators, and any other virtual columns) must match. A spec of a
different type is never equivalent.
+ */
+ @Override
+ public boolean hasEqualCompactionState(BaseTableProjectionSpec other)
+ {
+ if (!(other instanceof RollupTableProjectionSpec)) {
+ return false;
+ }
+ return withoutQueryGranularity().equals(((RollupTableProjectionSpec)
other).withoutQueryGranularity());
+ }
+
+ @Override
+ public RollupTableProjectionSpec withAdditionalColumns(@Nullable
List<DimensionSchema> additionalColumns)
+ {
+ if (additionalColumns == null || additionalColumns.isEmpty()) {
+ return this;
+ }
+ final List<DimensionSchema> revised = new
ArrayList<>(groupingColumns.size() + additionalColumns.size());
+ revised.addAll(groupingColumns);
+ for (DimensionSchema additionalColumn : additionalColumns) {
+ if (ColumnHolder.TIME_COLUMN_NAME.equals(additionalColumn.getName())) {
+ throw InvalidInput.exception(
+ "Cannot append column [%s] to a [%s] base table; it must be
declared at its position in the grouping column list",
+ ColumnHolder.TIME_COLUMN_NAME,
+ TYPE_NAME
+ );
+ }
+ revised.add(additionalColumn);
+ }
+ // Duplicates of a declared column or aggregator, and a column named for
the query-granularity carrier, are
+ // rejected by the constructor's validation.
+ return new RollupTableProjectionSpec(virtualColumns, revised, aggregators);
+ }
+
+ /**
+ * Returns a copy of this spec with the {@link
Granularities#GRANULARITY_VIRTUAL_COLUMN_NAME} virtual column removed,
+ * the inverse of {@link #withQueryGranularity(Granularity)}. If no such
virtual column is present this returns
+ * {@code this} unchanged. Used to compare schema independently of query
granularity in {@link #hasEqualCompactionState}.
+ */
+ private RollupTableProjectionSpec withoutQueryGranularity()
+ {
+ if
(virtualColumns.getVirtualColumn(Granularities.GRANULARITY_VIRTUAL_COLUMN_NAME)
== null) {
+ return this;
+ }
+ final List<VirtualColumn> remaining = new ArrayList<>();
+ for (VirtualColumn vc : virtualColumns.getVirtualColumns()) {
+ if
(!Granularities.GRANULARITY_VIRTUAL_COLUMN_NAME.equals(vc.getOutputName())) {
+ remaining.add(vc);
+ }
+ }
+ return new RollupTableProjectionSpec(VirtualColumns.create(remaining),
groupingColumns, aggregators);
+ }
+
+ /**
+ * Aggregators produce the metric columns, so their names share a namespace
with the grouping columns: a collision
+ * means one column with two definitions.
+ */
+ private static void validateAggregators(List<DimensionSchema>
groupingColumns, AggregatorFactory[] aggregators)
+ {
+ final Set<String> names = new HashSet<>();
+ for (DimensionSchema column : groupingColumns) {
+ names.add(column.getName());
+ }
+ for (AggregatorFactory aggregator : aggregators) {
Review Comment:
[P2] Reject metric names that collide with the granularity carrier
**Finding:** The shared declared-column validation is applied only to
groupingColumns, while this loop accepts an aggregator named
__virtualGranularity. Once query granularity is present, the spec contains both
that metric and the carrier virtual column;
IncrementalIndexColumnSelectorFactory resolves virtual columns before metrics,
so a metric selector is shadowed by the time-floor expression, while catalog
INSERTs reject the target as computed. This lets an invalid and potentially
corrupting rollup schema pass validation.
**Suggestion:** Reject aggregator names equal to
Granularities.GRANULARITY_VIRTUAL_COLUMN_NAME when constructing the rollup
spec, with catalog and ingestion regression coverage.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
