diff --git a/processing/src/main/java/org/apache/druid/data/input/impl/ClusteredValueGroupsBaseTableProjectionSpec.java b/processing/src/main/java/org/apache/druid/data/input/impl/ClusteredValueGroupsBaseTableProjectionSpec.java index 9685ab03c07f..401816421dc5 100644 --- a/processing/src/main/java/org/apache/druid/data/input/impl/ClusteredValueGroupsBaseTableProjectionSpec.java +++ b/processing/src/main/java/org/apache/druid/data/input/impl/ClusteredValueGroupsBaseTableProjectionSpec.java @@ -33,7 +33,10 @@ 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.ColumnCapabilities; import org.apache.druid.segment.column.ColumnHolder; +import org.apache.druid.segment.column.ColumnType; +import org.apache.druid.segment.column.RowSignature; import org.apache.druid.segment.projections.Projections; import org.apache.druid.utils.CollectionUtils; @@ -41,6 +44,7 @@ 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; @@ -48,9 +52,10 @@ /** * {@link BaseTableProjectionSpec} for the clustered-value-groups base table mode: rows are partitioned into per-tuple * "cluster groups" keyed by one or more typed clustering columns, optionally derived from {@link #virtualColumns}. - * Essentially, each group is stored as its own internal sub-segment.. + * Essentially, each group is stored as its own internal sub-segment. Logically, a clustered base table spec is a + * projection of an (imaginary/unstored) base table into the stored clustered table. *

- * The operator declares a single ordered {@link #columns} list — the full set of columns in segment order, plus a + * The operator declares a single ordered {@link #columns} list, the full set of columns in segment order, plus a * {@link #clusteringColumns} list of NAMES designating the leading prefix of {@link #columns} that rows are * clustered by. The time position is an explicit positional entry in {@link #columns} named {@code __time}; clustering * by the time column is not yet supported, so {@code __time} must be a non-clustering column. A clustered base table @@ -94,6 +99,7 @@ public ClusteredValueGroupsBaseTableProjectionSpec( { validate(columns, clusteringColumns); this.virtualColumns = virtualColumns == null ? VirtualColumns.EMPTY : virtualColumns; + validateVirtualColumns(this.virtualColumns, columns); this.columns = Collections.unmodifiableList(new ArrayList<>(columns)); this.clusteringColumns = Collections.unmodifiableList(new ArrayList<>(clusteringColumns)); this.clusteringColumnSchemas = this.columns.subList(0, this.clusteringColumns.size()); @@ -311,6 +317,98 @@ private static void validate(List columns, List cluster } } + /** + * Rules to keep virtual columns definitions reasonable: + *

+ * The query-granularity carrier ({@link Granularities#GRANULARITY_VIRTUAL_COLUMN_NAME}) is special handled to + * capture how __time is computed, so it is exempt from the output rule. + */ + private static void validateVirtualColumns(VirtualColumns virtualColumns, List columns) + { + final VirtualColumn[] all = virtualColumns.getVirtualColumns(); + if (all.length == 0) { + return; + } + // Declared column types, doubling as the ColumnInspector used to infer each materialized virtual column's output + // type (an expression's output type can depend on its input column types). + final RowSignature.Builder signatureBuilder = RowSignature.builder(); + for (DimensionSchema column : columns) { + signatureBuilder.add(column.getName(), column.getColumnType()); + } + final RowSignature rowSignature = signatureBuilder.build(); + // The output rule below lets a virtual column go unstored when it is exempt: an intermediary that feeds another + // virtual column (collected during the input pass), or the metadata-only query-granularity carrier (seeded here). + final Set outputExempt = new HashSet<>(); + outputExempt.add(Granularities.GRANULARITY_VIRTUAL_COLUMN_NAME); + // input rule: every input must be a stored column or another virtual column; an input that is a virtual column + // makes that virtual column an intermediary, exempt from having to be materialized itself. + for (VirtualColumn virtualColumn : all) { + if (virtualColumn.usesDotNotation()) { + throw InvalidInput.exception( + "virtual column [%s] uses dot notation, which is not supported for clusteredValueGroups base table specs", + virtualColumn.getOutputName() + ); + } + for (String input : virtualColumn.requiredColumns()) { + final boolean isStored = rowSignature.contains(input); + final boolean isVirtual = virtualColumns.exists(input); + if (!isStored && !isVirtual) { + throw InvalidInput.exception( + "virtual column [%s] reads column [%s], which is neither a stored column nor another virtual column;" + + " clustered base table virtual columns must be computable from stored columns (retain [%s] in" + + " 'columns', or use a transformSpec for in-place transforms)", + virtualColumn.getOutputName(), + input, + input + ); + } + if (isVirtual) { + outputExempt.add(input); + } + } + } + // output rule: every virtual column must be materialized (stored) or exempt (intermediary / granularity carrier); + // a materialized virtual column must additionally produce the declared type of the column it fills. + for (VirtualColumn virtualColumn : all) { + final String outputName = virtualColumn.getOutputName(); + final boolean stored = rowSignature.contains(outputName); + if (!stored && !outputExempt.contains(outputName)) { + throw InvalidInput.exception( + "virtual column [%s] is not stored (not declared in 'columns') and does not feed another virtual column;" + + " clustered base table virtual columns must materialize a stored column or be an input to one that does", + outputName + ); + } + if (stored) { + final ColumnCapabilities outputCapabilities = + virtualColumns.getColumnCapabilitiesWithFallback(rowSignature, outputName); + final ColumnType outputType = ColumnType.fromCapabilities(outputCapabilities); + final ColumnType declaredType = rowSignature.getColumnType(outputName).orElse(null); + if (outputType != null && !outputType.equals(declaredType)) { + throw InvalidInput.exception( + "virtual column [%s] produces type [%s] but the column it materializes is declared as [%s] in 'columns';" + + " a clustered base table virtual column must match the declared type of the column it fills", + outputName, + outputType, + declaredType + ); + } + } + } + } + private static DruidException clusteringPrefixException( List columns, List clusteringColumns diff --git a/processing/src/main/java/org/apache/druid/segment/ConcatenatingCursor.java b/processing/src/main/java/org/apache/druid/segment/ConcatenatingCursor.java index 1208c3df2021..d23069c3137a 100644 --- a/processing/src/main/java/org/apache/druid/segment/ConcatenatingCursor.java +++ b/processing/src/main/java/org/apache/druid/segment/ConcatenatingCursor.java @@ -25,6 +25,7 @@ import javax.annotation.Nullable; import java.util.List; +import java.util.Map; /** * {@link Cursor} that walks a sequence of per-group cursors back-to-back, presenting them to the caller as a single @@ -40,6 +41,7 @@ public final class ConcatenatingCursor implements Cursor private final List> holderSuppliers; private final List clusteringValuesByGroup; private final ClusteringColumnSelectorFactory wrapperFactory; + private final ColumnSelectorFactory exposedFactory; private int currentIdx; @Nullable @@ -49,7 +51,8 @@ public final class ConcatenatingCursor implements Cursor public ConcatenatingCursor( List> holderSuppliers, List clusteringValuesByGroup, - ClusteringColumnSelectorFactory wrapperFactory + ClusteringColumnSelectorFactory wrapperFactory, + Map virtualColumnRemap ) { if (holderSuppliers.size() != clusteringValuesByGroup.size()) { @@ -65,6 +68,9 @@ public ConcatenatingCursor( this.holderSuppliers = holderSuppliers; this.clusteringValuesByGroup = clusteringValuesByGroup; this.wrapperFactory = wrapperFactory; + this.exposedFactory = virtualColumnRemap.isEmpty() + ? wrapperFactory + : new RemapColumnSelectorFactory(wrapperFactory, virtualColumnRemap); this.currentIdx = -1; } @@ -101,7 +107,7 @@ private void advanceToNextNonEmptyGroup() public ColumnSelectorFactory getColumnSelectorFactory() { initializeIfNeeded(); - return wrapperFactory; + return exposedFactory; } @Override diff --git a/processing/src/main/java/org/apache/druid/segment/MergingClusterGroupCursor.java b/processing/src/main/java/org/apache/druid/segment/MergingClusterGroupCursor.java index a3a22fe55e3e..33404f95958c 100644 --- a/processing/src/main/java/org/apache/druid/segment/MergingClusterGroupCursor.java +++ b/processing/src/main/java/org/apache/druid/segment/MergingClusterGroupCursor.java @@ -26,6 +26,7 @@ import org.apache.druid.segment.projections.MergingColumnSelectorFactory; import java.util.List; +import java.util.Map; /** * {@link Cursor} that presents a set of individually {@code __time}-sorted per-cluster-group cursors as a single @@ -38,7 +39,11 @@ * all surviving group cursors up front (so unlike the concatenating path it does not benefit from early-exit * laziness) and, on each {@link #advance()}, emits the row with the smallest (or largest, when descending) * {@code __time} across the groups. Each per-group sub-index already exposes the clustering columns as constants, so - * the {@link MergingColumnSelectorFactory} simply dispatches every column to the winning group. + * the {@link MergingColumnSelectorFactory} simply dispatches every column to the winning group. When the plan carries a + * query-virtual-column-to-materialized-column remap, the merging factory is wrapped in a + * {@link RemapColumnSelectorFactory} so reads of a remapped query virtual column's output name resolve to its + * materialized column (mirroring {@link ConcatenatingCursor}); the per-group specs already dropped those virtual + * columns and added their materialized targets as physical columns. *

* An {@link IntHeapPriorityQueue} of group indices, keyed on each group's current {@code __time} via * {@link #compareGroups}, drives the merge: the queue head is the winning group, {@link IntHeapPriorityQueue#changed()} @@ -50,6 +55,7 @@ public final class MergingClusterGroupCursor implements Cursor { private final List> holderSuppliers; private final boolean descending; + private final Map virtualColumnRemap; private boolean initialized; // Indexed by group. groupCursors[i] is null if that group's holder produced no cursor. @@ -62,11 +68,14 @@ public final class MergingClusterGroupCursor implements Cursor private int currentGroup; // Monotonically increasing output-row id, one per emitted row; the merge emits exactly one row per advance. private long outputRowId; - private MergingColumnSelectorFactory factory; + // The factory exposed to callers: the MergingColumnSelectorFactory, wrapped in a RemapColumnSelectorFactory when + // virtualColumnRemap is non-empty. + private ColumnSelectorFactory exposedFactory; public MergingClusterGroupCursor( List> holderSuppliers, - boolean descending + boolean descending, + Map virtualColumnRemap ) { if (holderSuppliers.isEmpty()) { @@ -74,6 +83,7 @@ public MergingClusterGroupCursor( } this.holderSuppliers = holderSuppliers; this.descending = descending; + this.virtualColumnRemap = virtualColumnRemap; } private void initializeIfNeeded() @@ -106,14 +116,18 @@ private void initializeIfNeeded() } } currentGroup = heap.isEmpty() ? 0 : heap.firstInt(); - factory = new MergingColumnSelectorFactory(groupFactories, () -> currentGroup, () -> outputRowId); + final MergingColumnSelectorFactory mergingFactory = + new MergingColumnSelectorFactory(groupFactories, () -> currentGroup, () -> outputRowId); + exposedFactory = virtualColumnRemap.isEmpty() + ? mergingFactory + : new RemapColumnSelectorFactory(mergingFactory, virtualColumnRemap); } @Override public ColumnSelectorFactory getColumnSelectorFactory() { initializeIfNeeded(); - return factory; + return exposedFactory; } @Override diff --git a/processing/src/main/java/org/apache/druid/segment/QueryableIndexCursorFactory.java b/processing/src/main/java/org/apache/druid/segment/QueryableIndexCursorFactory.java index ac0e843260a2..1206000e6cc4 100644 --- a/processing/src/main/java/org/apache/druid/segment/QueryableIndexCursorFactory.java +++ b/processing/src/main/java/org/apache/druid/segment/QueryableIndexCursorFactory.java @@ -43,6 +43,7 @@ import org.apache.druid.segment.vector.ConcatenatingVectorCursor; import org.apache.druid.segment.vector.MultiValueDimensionVectorSelector; import org.apache.druid.segment.vector.ReadableVectorInspector; +import org.apache.druid.segment.vector.RemapVectorColumnSelectorFactory; import org.apache.druid.segment.vector.SingleValueDimensionVectorSelector; import org.apache.druid.segment.vector.VectorColumnSelectorFactory; import org.apache.druid.segment.vector.VectorCursor; @@ -57,6 +58,7 @@ import java.util.Arrays; import java.util.LinkedHashSet; import java.util.List; +import java.util.Map; public class QueryableIndexCursorFactory implements ResidentCursorFactory { @@ -202,13 +204,46 @@ private CursorHolder makeSingleGroupClusteredCursorHolder( final List ordering = useTimeOrderedCursors(spec, summary) ? summary.getGroupOrdering() : summary.getOrdering(); - // groupIndex exposes the group's clustering columns as constant columns, no selector wrapper is needed + if (plan.virtualColumnRemap().isEmpty()) { + return new QueryableIndexCursorHolder( + groupIndex, + plan.rebuildCursorBuildSpec(spec, valueGroup), + QueryableIndexTimeBoundaryInspector.create(groupIndex), + ordering + ); + } + return new QueryableIndexCursorHolder( groupIndex, plan.rebuildCursorBuildSpec(spec, valueGroup), QueryableIndexTimeBoundaryInspector.create(groupIndex), ordering - ); + ) + { + @Override + protected ColumnSelectorFactory makeColumnSelectorFactoryForOffset( + ColumnCache columnCache, + Offset baseOffset + ) + { + return new RemapColumnSelectorFactory( + super.makeColumnSelectorFactoryForOffset(columnCache, baseOffset), + plan.virtualColumnRemap() + ); + } + + @Override + protected VectorColumnSelectorFactory makeVectorColumnSelectorFactoryForOffset( + ColumnCache columnCache, + VectorOffset baseOffset + ) + { + return new RemapVectorColumnSelectorFactory( + super.makeVectorColumnSelectorFactoryForOffset(columnCache, baseOffset), + plan.virtualColumnRemap() + ); + } + }; } /** @@ -258,12 +293,14 @@ private CursorHolder makeMultiGroupClusteredCursorHolder( return makeTimeMergedClusteredCursorHolder( holderSuppliers, closer, - Cursors.getTimeOrdering(spec.getPreferredOrdering()) + Cursors.getTimeOrdering(spec.getPreferredOrdering()), + plan.virtualColumnRemap() ); } return makeConcatenatedClusteredCursorHolder( spec, + plan, this, clusteringColumns, clusteringValuesByGroup, @@ -369,6 +406,7 @@ public List getAggregatorsForPreAggregated() */ private static CursorHolder makeConcatenatedClusteredCursorHolder( CursorBuildSpec spec, + ClusterGroupQueryPlan plan, ColumnInspector inspector, RowSignature clusteringColumns, List clusteringValuesByGroup, @@ -383,24 +421,28 @@ private static CursorHolder makeConcatenatedClusteredCursorHolder( final ClusteringColumnSelectorFactory wrapperFactory = new ClusteringColumnSelectorFactory( ClusteringColumnSelectorFactory.UNINITIALIZED_DELEGATE, clusteringColumns, - clusteringValuesByGroup.get(0) + clusteringValuesByGroup.get(0), + spec.getVirtualColumns() ); final ClusteringVectorColumnSelectorFactory vectorWrapperFactory = new ClusteringVectorColumnSelectorFactory( UNINITIALIZED_VECTOR_DELEGATE, clusteringColumns, clusteringValuesByGroup.get(0), - vectorSize + vectorSize, + spec.getVirtualColumns() ); final ConcatenatingCursor cursor = new ConcatenatingCursor( holderSuppliers, clusteringValuesByGroup, - wrapperFactory + wrapperFactory, + plan.virtualColumnRemap() ); final ConcatenatingVectorCursor vectorCursor = new ConcatenatingVectorCursor( holderSuppliers, clusteringValuesByGroup, - vectorWrapperFactory + vectorWrapperFactory, + plan.virtualColumnRemap() ); // each group gets a different rewritten filter, so the conservative thing to do here is require the original query @@ -410,7 +452,7 @@ private static CursorHolder makeConcatenatedClusteredCursorHolder( final boolean filterCanVectorize = queryFilter == null || queryFilter.canVectorizeMatcher(spec.getVirtualColumns().wrapInspector(inspector)); // we still check that the first holder is vectorizable to make sure all the non-filter parts can be vectorized - final boolean canVectorize = filterCanVectorize && holderSuppliers.get(0).get().canVectorize(); + final boolean canVectorize = filterCanVectorize && holderSuppliers.getFirst().get().canVectorize(); return new CursorHolder() { @@ -453,16 +495,19 @@ public void close() /** * Builds a {@link CursorHolder} whose non-vectorized cursor is a globally {@code __time}-ordered {@link * MergingClusterGroupCursor} k-way-merging the per-group cursors. Only invoked when the query requested {@code - * __time} ordering and each group is individually {@code __time}-sorted (see caller). + * __time} ordering and each group is individually {@code __time}-sorted (see caller). The {@code virtualColumnRemap} + * (of query virtual columns equivalent to a materialized column) is applied on top of the merge, mirroring + * {@link #makeConcatenatedClusteredCursorHolder}. */ private static CursorHolder makeTimeMergedClusteredCursorHolder( List> holderSuppliers, Closer closer, - Order timeOrder + Order timeOrder, + Map virtualColumnRemap ) { final boolean descending = timeOrder == Order.DESCENDING; - final MergingClusterGroupCursor cursor = new MergingClusterGroupCursor(holderSuppliers, descending); + final MergingClusterGroupCursor cursor = new MergingClusterGroupCursor(holderSuppliers, descending, virtualColumnRemap); final List ordering = descending ? Cursors.descendingTimeOrder() : Cursors.ascendingTimeOrder(); return new CursorHolder() { diff --git a/processing/src/main/java/org/apache/druid/segment/incremental/ClusteringAwareIncrementalIndexColumnSelectorFactory.java b/processing/src/main/java/org/apache/druid/segment/incremental/ClusteringAwareIncrementalIndexColumnSelectorFactory.java new file mode 100644 index 000000000000..2703e720ea64 --- /dev/null +++ b/processing/src/main/java/org/apache/druid/segment/incremental/ClusteringAwareIncrementalIndexColumnSelectorFactory.java @@ -0,0 +1,112 @@ +/* + * 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.segment.incremental; + +import org.apache.druid.query.Order; +import org.apache.druid.query.dimension.DimensionSpec; +import org.apache.druid.segment.ColumnValueSelector; +import org.apache.druid.segment.CursorBuildSpec; +import org.apache.druid.segment.DimensionSelector; +import org.apache.druid.segment.VirtualColumns; +import org.apache.druid.segment.column.ColumnCapabilities; +import org.apache.druid.segment.column.RowSignature; +import org.apache.druid.segment.projections.ClusteringColumnSelectorFactory; + +import javax.annotation.Nullable; + +/** + * An {@link IncrementalIndexColumnSelectorFactory} that also serves a cluster group's clustering columns as per-group + * constants. A clustered base table stores each cluster group's rows WITHOUT the (constant) clustering columns, so the + * group's row selector cannot resolve them directly. + *

+ * The clustering-constant interception only applies when the requested name is NOT owned by a query virtual column. The + * superclass resolves virtual columns before physical columns, so a query VC whose output name shadows a clustering + * column must win: intercepting it here would make that VC — and any other retained VC that reads it — silently read + * the per-group constant instead of the computed value. + */ +final class ClusteringAwareIncrementalIndexColumnSelectorFactory extends IncrementalIndexColumnSelectorFactory +{ + private final RowSignature clusteringColumns; + private final VirtualColumns queryVirtualColumns; + /** + * Serves the clustering columns as per-group constants. Its delegate is intentionally the throwing placeholder: this + * factory is only ever asked for clustering columns (and only when no query VC shadows them), and + * {@link ClusteringColumnSelectorFactory} serves those from the group's constant tuple without touching its delegate. + */ + private final ClusteringColumnSelectorFactory clusteringConstants; + + ClusteringAwareIncrementalIndexColumnSelectorFactory( + IncrementalIndexRowSelector rowSelector, + IncrementalIndexRowHolder rowHolder, + CursorBuildSpec cursorBuildSpec, + Order timeOrder, + RowSignature clusteringColumns, + Object[] clusteringValues + ) + { + super(rowSelector, rowHolder, cursorBuildSpec, timeOrder); + this.clusteringColumns = clusteringColumns; + this.queryVirtualColumns = cursorBuildSpec.getVirtualColumns(); + this.clusteringConstants = new ClusteringColumnSelectorFactory( + ClusteringColumnSelectorFactory.UNINITIALIZED_DELEGATE, + clusteringColumns, + clusteringValues + ); + } + + /** + * Whether {@code name} should be served as this group's clustering constant: only when it is a clustering column AND + * no query virtual column owns the name. A query VC that shares a clustering column's name shadows it (the superclass + * resolves virtual columns before physical columns), so deferring to the superclass lets that VC — and anything that + * reads it through this factory — observe the computed value rather than the constant. + */ + private boolean servesClusteringConstant(String name) + { + return clusteringColumns.indexOf(name) >= 0 && !queryVirtualColumns.exists(name); + } + + @Override + public DimensionSelector makeDimensionSelector(DimensionSpec dimensionSpec) + { + if (servesClusteringConstant(dimensionSpec.getDimension())) { + return clusteringConstants.makeDimensionSelector(dimensionSpec); + } + return super.makeDimensionSelector(dimensionSpec); + } + + @Override + public ColumnValueSelector makeColumnValueSelector(String columnName) + { + if (servesClusteringConstant(columnName)) { + return clusteringConstants.makeColumnValueSelector(columnName); + } + return super.makeColumnValueSelector(columnName); + } + + @Nullable + @Override + public ColumnCapabilities getColumnCapabilities(String columnName) + { + if (servesClusteringConstant(columnName)) { + return clusteringConstants.getColumnCapabilities(columnName); + } + return super.getColumnCapabilities(columnName); + } +} diff --git a/processing/src/main/java/org/apache/druid/segment/incremental/IncrementalIndexCursorFactory.java b/processing/src/main/java/org/apache/druid/segment/incremental/IncrementalIndexCursorFactory.java index 07cebffa2c96..74f2ddaa0142 100644 --- a/processing/src/main/java/org/apache/druid/segment/incremental/IncrementalIndexCursorFactory.java +++ b/processing/src/main/java/org/apache/druid/segment/incremental/IncrementalIndexCursorFactory.java @@ -154,19 +154,37 @@ private CursorHolder makeClusteredCursorHolder(CursorBuildSpec spec, OnHeapClust final List> holderSuppliers = new ArrayList<>(surviving.size()); final Closer closer = Closer.create(); for (TableClusterGroupSpec valueGroup : surviving) { - final OnHeapClusterGroup group = clusteredBaseTable.getGroupForClusteringValues( - valueGroup.lookupClusteringValues() - ); + final Object[] groupClusteringValues = valueGroup.lookupClusteringValues(); + final OnHeapClusterGroup group = clusteredBaseTable.getGroupForClusteringValues(groupClusteringValues); if (group == null) { throw DruidException.defensive( "No cluster group for clustering values [%s]", - Arrays.toString(valueGroup.lookupClusteringValues()) + Arrays.toString(groupClusteringValues) ); } - clusteringValuesByGroup.add(valueGroup.lookupClusteringValues()); + clusteringValuesByGroup.add(groupClusteringValues); final CursorBuildSpec groupSpec = plan.rebuildCursorBuildSpec(spec, valueGroup); holderSuppliers.add( - Suppliers.memoize(() -> closer.register(new IncrementalIndexCursorHolder(group, groupSpec))) + Suppliers.memoize(() -> closer.register( + new IncrementalIndexCursorHolder(group, groupSpec) + { + @Override + public ColumnSelectorFactory makeSelectorFactory( + CursorBuildSpec buildSpec, + IncrementalIndexRowHolder currEntry + ) + { + return new ClusteringAwareIncrementalIndexColumnSelectorFactory( + group, + currEntry, + buildSpec, + getTimeOrder(), + clusteringColumns, + groupClusteringValues + ); + } + } + )) ); } @@ -175,9 +193,15 @@ private CursorHolder makeClusteredCursorHolder(CursorBuildSpec spec, OnHeapClust final ClusteringColumnSelectorFactory wrapperFactory = new ClusteringColumnSelectorFactory( ClusteringColumnSelectorFactory.UNINITIALIZED_DELEGATE, clusteringColumns, - clusteringValuesByGroup.get(0) + clusteringValuesByGroup.getFirst(), + spec.getVirtualColumns() + ); + final ConcatenatingCursor cursor = new ConcatenatingCursor( + holderSuppliers, + clusteringValuesByGroup, + wrapperFactory, + plan.virtualColumnRemap() ); - final ConcatenatingCursor cursor = new ConcatenatingCursor(holderSuppliers, clusteringValuesByGroup, wrapperFactory); return new CursorHolder() { diff --git a/processing/src/main/java/org/apache/druid/segment/projections/ClusterGroupQueryPlan.java b/processing/src/main/java/org/apache/druid/segment/projections/ClusterGroupQueryPlan.java index 6f87c1984df6..e145433f2659 100644 --- a/processing/src/main/java/org/apache/druid/segment/projections/ClusterGroupQueryPlan.java +++ b/processing/src/main/java/org/apache/druid/segment/projections/ClusterGroupQueryPlan.java @@ -21,32 +21,42 @@ import org.apache.druid.query.filter.Filter; import org.apache.druid.segment.CursorBuildSpec; +import org.apache.druid.segment.VirtualColumn; +import org.apache.druid.segment.VirtualColumns; import org.apache.druid.segment.filter.FalseFilter; import org.apache.druid.segment.filter.TrueFilter; import javax.annotation.Nullable; +import java.util.ArrayList; +import java.util.HashSet; import java.util.List; +import java.util.Map; +import java.util.Set; import java.util.function.Function; /** * Per-query plan produced by {@link Projections#planClusterGroupQuery} for a clustered base-table segment: the - * subset of cluster groups whose clustering values can't rule the query filter out (the surviving groups), plus a - * per-group rewriter that produces the per-group cursor's filter by folding clustering-column leaves against each - * group's constant clustering tuple. Production callers (the cursor factory) iterate {@link #survivingGroups()} - * and call {@link #rewriteFor} once per surviving group. + * subset of cluster groups whose clustering values can't rule the query filter out (the surviving groups), a per-group + * filter rewrite that folds clustering-column leaves against each group's constant clustering tuple (see + * {@link #rewriteFor}), and a query-level {@link #virtualColumnRemap()} of query virtual columns equivalent to a + * materialized column. Production callers (the cursor factories) iterate {@link #survivingGroups()} and call + * {@link #rebuildCursorBuildSpec} once per surviving group to produce that group's {@link CursorBuildSpec}. */ public final class ClusterGroupQueryPlan { private final List survivingGroups; private final Function rewriter; + private final Map virtualColumnRemap; ClusterGroupQueryPlan( List survivingGroups, - Function rewriter + Function rewriter, + Map virtualColumnRemap ) { this.survivingGroups = survivingGroups; this.rewriter = rewriter; + this.virtualColumnRemap = virtualColumnRemap; } /** @@ -59,10 +69,21 @@ public List survivingGroups() } /** - * Per-group rewrite of the query's filter against {@code group}'s constant clustering tuple: clustering-column - * leaves collapse to {@link TrueFilter} / {@link FalseFilter} and fold through AND / OR / NOT, so the rewritten - * filter no longer references columns the per-group {@link org.apache.druid.segment.QueryableIndex} doesn't - * physically carry. Non-clustering leaves and unrecognized filter types are left as-is. + * Query-level mapping of {@code queryVirtualColumnOutputName -> materializedColumnName} for query virtual columns + * that are equivalent to a clustering column or a non-clustering materialized column in the clustered base table. + * Empty when no query virtual column has a materialized equivalent (or there are no query virtual columns). + */ + public Map virtualColumnRemap() + { + return virtualColumnRemap; + } + + /** + * Per-group rewrite of the query's filter against {@code group}'s constant clustering tuple: recognized + * equality / in / null leaves on a clustering column collapse to {@link TrueFilter} / {@link FalseFilter} and fold + * through AND / OR / NOT. Non-clustering leaves and other filter shapes on a clustering column (e.g. range / like) + * are left as-is for the per-group cursor to evaluate per-row; the cursor exposes clustering columns as per-group + * constants, so a surviving clustering-column leaf still resolves rather than reading a missing column. *

* Returns {@code null} when the original query filter was {@code null} (no rewrite needed), or * {@link FalseFilter} for a group that didn't survive pruning (would have been excluded from @@ -76,21 +97,63 @@ public Filter rewriteFor(TableClusterGroupSpec group) /** * Rebuild {@code spec} for {@code group}'s per-group cursor by swapping in this plan's per-group filter rewrite - * (see {@link #rewriteFor}). Returns {@code spec} unchanged when there is no filter, or when the rewrite is - * identical to the original (no clustering leaves folded), so the common no-op case allocates nothing. Shared by - * the {@link org.apache.druid.segment.QueryableIndexCursorFactory} (historical) and - * {@link org.apache.druid.segment.incremental.IncrementalIndexCursorFactory} (realtime) clustered dispatch. + * (see {@link #rewriteFor}) and, when {@link #virtualColumnRemap()} is non-empty, substituting any query virtual + * columns that are equivalent to a materialized column. + *

+ * When there is a remap, the matched query virtual columns (the remap keys) are dropped from the spec's virtual + * columns, and the per-group filter's required columns are rewritten via the remap so any surviving leaf that + * referenced a dropped VC now points at its materialized column: a non-clustering VC maps to the stored physical + * column, a clustering-equivalent VC to the clustering column (the per-group cursor exposes it as a constant). + * Foldable clustering-VC leaves (equality / in / null) were already collapsed to TRUE / FALSE by the per-group + * rewrite walk and never reach the rewrite. Grouping / select / aggregator reads of a dropped VC are served by the + * {@link org.apache.druid.segment.RemapColumnSelectorFactory} the cursor factory wraps around the per-group selector + * factory (which exposes clustering columns as constants, whether from the fabricated per-group index or the + * {@link ClusteringColumnSelectorFactory}). */ public CursorBuildSpec rebuildCursorBuildSpec(CursorBuildSpec spec, TableClusterGroupSpec group) { - if (spec.getFilter() == null) { - return spec; + if (virtualColumnRemap.isEmpty()) { + if (spec.getFilter() == null) { + return spec; + } + final Filter rewritten = rewriteFor(group); + if (rewritten == spec.getFilter()) { + return spec; + } + return CursorBuildSpec.builder(spec).setFilter(rewritten).build(); } - final Filter rewritten = rewriteFor(group); - if (rewritten == spec.getFilter()) { - return spec; + + // Drop the remapped query virtual columns from the per-group spec; the materialized columns they map to are either + // the group's constant clustering column or a stored per-group physical column. The cursor factory wraps the + // per-group selector factory with a RemapColumnSelectorFactory so reads of the original query VC names resolve to + // those materialized columns. + final List prunedVcs = new ArrayList<>(); + for (VirtualColumn vc : spec.getVirtualColumns().getVirtualColumns()) { + if (!virtualColumnRemap.containsKey(vc.getOutputName())) { + prunedVcs.add(vc); + } } - return CursorBuildSpec.builder(spec).setFilter(rewritten).build(); + + // Per-group filter rewrite first (folds recognized clustering leaves), then remap any surviving leaf that still + // references a dropped VC to its materialized column. + Filter rewritten = spec.getFilter() == null ? null : rewriteFor(group); + if (rewritten != null) { + rewritten = Projections.rewriteFilterRequiredColumns(rewritten, virtualColumnRemap); + } + + final CursorBuildSpec.CursorBuildSpecBuilder builder = CursorBuildSpec.builder(spec) + .setFilter(rewritten) + .setVirtualColumns(VirtualColumns.create(prunedVcs)); + + // The dropped query VCs now read their materialized target columns, so add them + final Set physicalColumns = spec.getPhysicalColumns(); + if (physicalColumns != null) { + final Set withTargets = new HashSet<>(physicalColumns); + withTargets.addAll(virtualColumnRemap.values()); + builder.setPhysicalColumns(withTargets); + } + + return builder.build(); } } diff --git a/processing/src/main/java/org/apache/druid/segment/projections/ClusteringColumnSelectorFactory.java b/processing/src/main/java/org/apache/druid/segment/projections/ClusteringColumnSelectorFactory.java index ae719a96fb23..b224bd3b2e30 100644 --- a/processing/src/main/java/org/apache/druid/segment/projections/ClusteringColumnSelectorFactory.java +++ b/processing/src/main/java/org/apache/druid/segment/projections/ClusteringColumnSelectorFactory.java @@ -33,6 +33,7 @@ import org.apache.druid.segment.DimensionSelector; import org.apache.druid.segment.IdLookup; import org.apache.druid.segment.RowIdSupplier; +import org.apache.druid.segment.VirtualColumns; import org.apache.druid.segment.column.ColumnCapabilities; import org.apache.druid.segment.column.ColumnCapabilitiesImpl; import org.apache.druid.segment.column.ColumnType; @@ -48,6 +49,12 @@ * carrying the group's constant value, while delegating all other column lookups to a wrapped factory. This is the * mechanism by which a cluster group's clustering columns, which are NOT stored in the per-group column data since * they're constant across the group, are made visible to query engines as if they were ordinary columns. + *

+ * A clustering column whose name is also a query virtual column's output is NOT intercepted; it is left to the delegate + * (which resolves virtual columns before physical columns), so a shadowing VC — and anything reading it — observes the + * computed value rather than the constant. Names that were remapped away (a query VC equivalent to a materialized + * column) never reach this wrapper: the enclosing {@code RemapColumnSelectorFactory} rewrites them to their materialized + * target first. */ public class ClusteringColumnSelectorFactory implements ColumnSelectorFactory { @@ -80,6 +87,8 @@ public ColumnCapabilities getColumnCapabilities(String column) }; private final RowSignature clusteringColumns; + // Query virtual columns whose output names shadow a clustering column are deferred to the delegate; see class doc. + private final VirtualColumns queryVirtualColumns; private ColumnSelectorFactory delegate; private Object[] clusteringValues; // Bumped on every setDelegate(...) so per-call selector wrappers can detect group transitions and rebuild their @@ -92,11 +101,31 @@ public ClusteringColumnSelectorFactory( RowSignature clusteringColumns, Object[] clusteringValues ) + { + this(delegate, clusteringColumns, clusteringValues, VirtualColumns.EMPTY); + } + + public ClusteringColumnSelectorFactory( + ColumnSelectorFactory delegate, + RowSignature clusteringColumns, + Object[] clusteringValues, + VirtualColumns queryVirtualColumns + ) { this.clusteringColumns = clusteringColumns; + this.queryVirtualColumns = queryVirtualColumns; setDelegate(delegate, clusteringValues); } + /** + * Whether {@code name} should be served as this group's clustering constant: a clustering column not shadowed by a + * query virtual column of the same output name (see class doc). + */ + private boolean servesClusteringConstant(String name) + { + return clusteringColumns.indexOf(name) >= 0 && !queryVirtualColumns.exists(name); + } + /** * Update the underlying factory and the constant values for the current cluster group. Called by a multi-group * concatenating cursor on each group transition. Selectors previously returned by this factory will, on their next @@ -129,20 +158,20 @@ long getGeneration() @Override public DimensionSelector makeDimensionSelector(DimensionSpec dimensionSpec) { - final int idx = clusteringColumns.indexOf(dimensionSpec.getDimension()); - if (idx < 0) { + final String name = dimensionSpec.getDimension(); + if (!servesClusteringConstant(name)) { return new DelegatingDimensionSelector(this, dimensionSpec); } - return new ClusteringDimensionSelector(this, idx, dimensionSpec); + return new ClusteringDimensionSelector(this, clusteringColumns.indexOf(name), dimensionSpec); } @Override public ColumnValueSelector makeColumnValueSelector(String columnName) { - final int idx = clusteringColumns.indexOf(columnName); - if (idx < 0) { + if (!servesClusteringConstant(columnName)) { return new DelegatingColumnValueSelector(this, columnName); } + final int idx = clusteringColumns.indexOf(columnName); return new ClusteringColumnValueSelector(this, idx, clusteringColumns.getColumnType(idx).orElseThrow()); } @@ -150,9 +179,9 @@ public ColumnValueSelector makeColumnValueSelector(String columnName) @Override public ColumnCapabilities getColumnCapabilities(String column) { - final int idx = clusteringColumns.indexOf(column); - if (idx < 0) { - // Non-clustering columns are stored per cluster group, each with its own local dictionary. The + if (!servesClusteringConstant(column)) { + // Non-clustering columns (or a clustering column shadowed by a query VC, resolved by the delegate) are stored + // per cluster group, each with its own local dictionary. The // ConcatenatingCursor walks those groups behind a single cursor, so a column's dictionary IDs are NOT stable // across the whole cursor. We therefore must not advertise dictionary encoding here: otherwise the group-by // engine keys on the (per-group-local) IDs and conflates distinct values from different groups. Reporting the @@ -167,6 +196,7 @@ public ColumnCapabilities getColumnCapabilities(String column) .setDictionaryValuesUnique(false) .setHasBitmapIndexes(false); } + final int idx = clusteringColumns.indexOf(column); final ColumnType type = clusteringColumns.getColumnType(idx).orElseThrow(); if (type.is(ValueType.STRING)) { return ColumnCapabilitiesImpl.createSimpleSingleValueStringColumnCapabilities(); diff --git a/processing/src/main/java/org/apache/druid/segment/projections/ClusteringVectorColumnSelectorFactory.java b/processing/src/main/java/org/apache/druid/segment/projections/ClusteringVectorColumnSelectorFactory.java index e82c0a767221..4ff67e9d6176 100644 --- a/processing/src/main/java/org/apache/druid/segment/projections/ClusteringVectorColumnSelectorFactory.java +++ b/processing/src/main/java/org/apache/druid/segment/projections/ClusteringVectorColumnSelectorFactory.java @@ -21,6 +21,7 @@ import org.apache.druid.error.DruidException; import org.apache.druid.query.dimension.DimensionSpec; +import org.apache.druid.segment.VirtualColumns; import org.apache.druid.segment.column.ColumnCapabilities; import org.apache.druid.segment.column.ColumnCapabilitiesImpl; import org.apache.druid.segment.column.ColumnType; @@ -40,6 +41,8 @@ * Vectorized counterpart of {@link ClusteringColumnSelectorFactory}. Wraps a delegate * {@link VectorColumnSelectorFactory} and intercepts requests for clustering columns, returning constant-typed * vector selectors (via {@link ConstantVectorSelectors}). Other column requests pass through to delegating wrappers. + * As with the scalar wrapper, a clustering column whose name is also a query virtual column's output is NOT + * intercepted but left to the delegate (which resolves virtual columns before physical columns). * * The factory is mutable via {@link #setDelegate(VectorColumnSelectorFactory, Object[])}: a multi-group * {@code ConcatenatingVectorCursor} swaps the underlying delegate + clustering values on each group transition. @@ -52,6 +55,8 @@ public class ClusteringVectorColumnSelectorFactory implements VectorColumnSelectorFactory { private final RowSignature clusteringColumns; + // Query virtual columns whose output names shadow a clustering column are deferred to the delegate; see class doc. + private final VirtualColumns queryVirtualColumns; private final int maxVectorSize; private VectorColumnSelectorFactory delegate; private Object[] clusteringValues; @@ -79,12 +84,33 @@ public ClusteringVectorColumnSelectorFactory( Object[] clusteringValues, int maxVectorSize ) + { + this(delegate, clusteringColumns, clusteringValues, maxVectorSize, VirtualColumns.EMPTY); + } + + public ClusteringVectorColumnSelectorFactory( + VectorColumnSelectorFactory delegate, + RowSignature clusteringColumns, + Object[] clusteringValues, + int maxVectorSize, + VirtualColumns queryVirtualColumns + ) { this.clusteringColumns = clusteringColumns; + this.queryVirtualColumns = queryVirtualColumns; this.maxVectorSize = maxVectorSize; setDelegate(delegate, clusteringValues); } + /** + * Whether {@code column} should be served as this group's clustering constant: a clustering column not shadowed by a + * query virtual column of the same output name (see class doc). + */ + private boolean servesClusteringConstant(String column) + { + return clusteringColumns.indexOf(column) >= 0 && !queryVirtualColumns.exists(column); + } + /** * Update the underlying delegate and the constant clustering values for the current cluster group. Called by a * multi-group {@code ConcatenatingVectorCursor} on each group transition. @@ -156,30 +182,28 @@ public MultiValueDimensionVectorSelector makeMultiValueDimensionSelector(Dimensi @Override public VectorValueSelector makeValueSelector(String column) { - final int idx = clusteringColumns.indexOf(column); - if (idx < 0) { + if (!servesClusteringConstant(column)) { return new DelegatingVectorValueSelector(this, column); } - return new ClusteringVectorValueSelector(this, idx); + return new ClusteringVectorValueSelector(this, clusteringColumns.indexOf(column)); } @Override public VectorObjectSelector makeObjectSelector(String column) { - final int idx = clusteringColumns.indexOf(column); - if (idx < 0) { + if (!servesClusteringConstant(column)) { return new DelegatingVectorObjectSelector(this, column); } - return new ClusteringVectorObjectSelector(this, idx); + return new ClusteringVectorObjectSelector(this, clusteringColumns.indexOf(column)); } @Nullable @Override public ColumnCapabilities getColumnCapabilities(String column) { - final int idx = clusteringColumns.indexOf(column); - if (idx < 0) { - // Non-clustering columns are stored per cluster group with per-group local dictionaries that are NOT stable + if (!servesClusteringConstant(column)) { + // Non-clustering columns (or a clustering column shadowed by a query VC, resolved by the delegate) are stored + // per cluster group with per-group local dictionaries that are NOT stable // across the concatenating vector cursor. We must not advertise dictionary encoding: the vectorized group-by // keys on the selector's (per-group-local) IDs when the column reports as dictionary-encoded // (GroupByVectorColumnProcessorFactory#useDictionaryEncodedSelector), conflating distinct values across groups. @@ -195,6 +219,7 @@ public ColumnCapabilities getColumnCapabilities(String column) .setDictionaryValuesUnique(false) .setHasBitmapIndexes(false); } + final int idx = clusteringColumns.indexOf(column); final ColumnType type = clusteringColumns.getColumnType(idx).orElseThrow(); if (type.is(ValueType.STRING)) { return ColumnCapabilitiesImpl.createSimpleSingleValueStringColumnCapabilities(); diff --git a/processing/src/main/java/org/apache/druid/segment/projections/Projections.java b/processing/src/main/java/org/apache/druid/segment/projections/Projections.java index 54c214f4b37d..ef211ae6e241 100644 --- a/processing/src/main/java/org/apache/druid/segment/projections/Projections.java +++ b/processing/src/main/java/org/apache/druid/segment/projections/Projections.java @@ -56,8 +56,12 @@ import org.joda.time.Interval; import javax.annotation.Nullable; +import java.util.ArrayDeque; import java.util.ArrayList; +import java.util.Collections; +import java.util.Deque; import java.util.HashMap; +import java.util.HashSet; import java.util.IdentityHashMap; import java.util.List; import java.util.Map; @@ -258,6 +262,10 @@ public static ProjectionMatchBuilder matchFilter( if (projection.getFilter() != null) { final Filter queryFilter = queryCursorBuildSpec.getFilter(); if (queryFilter != null) { + if (!canRemapFilterToProjection(matchBuilder, queryFilter)) { + logTrace(queryCursorBuildSpec.getQueryContext(), "matchFilter: projection [%s] rejected — query filter cannot be rewritten to the projection column namespace", projection.getName()); + return null; + } // try to rewrite the query filter into a projection filter, if the rewrite is valid, we can proceed final Filter projectionFilter = projection.getFilter().toOptimizedFilter(false); final Filter remappedQueryFilter = remapFilterToProjection(matchBuilder, queryFilter); @@ -376,6 +384,10 @@ public static ProjectionMatchBuilder matchAggregators( if (filteredCombining != null) { FilteredAggregatorFactory filteredQueryAgg = (FilteredAggregatorFactory) queryAgg; final Filter aggFilter = filteredQueryAgg.getFilter().toFilter(); + if (!canRemapFilterToProjection(matchBuilder, aggFilter)) { + logTrace(queryCursorBuildSpec.getQueryContext(), "matchAggregators: projection [%s] rejected — filtered aggregator [%s] filter cannot be rewritten to the projection column namespace", projection.getName(), queryAgg.getName()); + return null; + } final Filter remappedAggFilter = remapFilterToProjection(matchBuilder, aggFilter); for (String column : aggFilter.getRequiredColumns()) { matchBuilder = matchRequiredColumn( @@ -652,9 +664,9 @@ public static ClusterGroupQueryPlan planClusterGroupQuery( { final Filter queryFilter = cursorBuildSpec.getFilter(); final VirtualColumns queryVcs = cursorBuildSpec.getVirtualColumns(); - if (groups.isEmpty() || queryFilter == null) { - // No filter (or no groups): every group survives, per-group rewrite is a no-op (null filter). - return new ClusterGroupQueryPlan(groups, group -> null); + if (groups.isEmpty()) { + // No groups: nothing to plan, nothing to remap. + return new ClusterGroupQueryPlan(groups, group -> null, Map.of()); } // Every spec in the list shares one summary by construction (set once in the schema constructor), so @@ -663,6 +675,16 @@ public static ClusterGroupQueryPlan planClusterGroupQuery( final RowSignature clusteringColumns = summary.getClusteringColumns(); final VirtualColumns groupVcs = summary.getVirtualColumns(); + final Set materializedColumns = new HashSet<>(summary.getColumns()); + final Map virtualColumnRemap = + buildClusterVirtualColumnRemap(queryVcs, groupVcs, materializedColumns, queryFilter); + + if (queryFilter == null) { + // No filter: every group survives, per-group filter rewrite is a no-op (null filter), but the VC remap (if any) + // still drives grouping / select substitution at cursor build time. + return new ClusterGroupQueryPlan(groups, group -> null, virtualColumnRemap); + } + // Single walk per group: produces the rewritten filter, and a top-level FalseFilter means the group prunes. // Cache the rewrite for every group (including pruned ones, where it's FalseFilter) so rewriteFor doesn't // re-walk for either the cursor factory or callers that want to inspect a pruned group's outcome directly. @@ -681,13 +703,100 @@ public static ClusterGroupQueryPlan planClusterGroupQuery( kept.add(group); } } - return new ClusterGroupQueryPlan(kept, rewriteCache::get); + return new ClusterGroupQueryPlan(kept, rewriteCache::get, virtualColumnRemap); + } + + /** + * Build a query-level remap of {@code queryVirtualColumnOutputName -> materializedColumnName} for each query virtual + * column that has an equivalent materialized column in the clustered base table (a clustering column produced by a + * group virtual column, or a non-clustering materialized virtual-column output). + */ + private static Map buildClusterVirtualColumnRemap( + VirtualColumns queryVcs, + VirtualColumns groupVcs, + Set materializedColumns, + @Nullable Filter queryFilter + ) + { + final VirtualColumn[] all = queryVcs.getVirtualColumns(); + if (all.length == 0) { + return Map.of(); + } + // A filter that can't rewrite its required columns (spatial, javascript, column-comparison, ...) can't have a + // remapped VC name swapped for the materialized column in its own required-column set, so any VC it references + // must stay in place. ClusteredValueGroupsBaseTableProjectionSpec guarantees every clustered VC is recomputable + // from stored columns, so leaving it unremapped simply lets the filter read the recomputed value rather than + // throwing. + final Set unrewritableFilterColumns = + queryFilter != null && !queryFilter.supportsRequiredColumnRewrite() + ? queryFilter.getRequiredColumns() + : Set.of(); + // Candidate substitutions: query VCs that have a differently-named equivalent materialized (stored) column. + final Map candidates = new HashMap<>(); + final Map byName = new HashMap<>(); + for (VirtualColumn vc : all) { + final String outputName = vc.getOutputName(); + byName.put(outputName, vc); + final VirtualColumns.Node queryNode = queryVcs.getNode(outputName); + if (queryNode == null) { + continue; + } + final VirtualColumn equivalent = groupVcs.findEquivalent(queryNode); + if (equivalent != null + && !outputName.equals(equivalent.getOutputName()) + && materializedColumns.contains(equivalent.getOutputName()) + // Don't remap to a materialized column whose name is shadowed by another query virtual column: the remapped + // read goes through the per-group delegate, which would resolve the target name to that (unrelated) query VC + // instead of the stored column. Leaving this VC makes it recompute from its own inputs instead. + && queryVcs.getVirtualColumn(equivalent.getOutputName()) == null + // Don't remap a VC an unrewritable filter (spatial, javascript, column-comparison, ...) references: the + // filter can't have the VC name swapped for the materialized column, so keep the VC in place and let the + // filter read its recomputed value. Clustered specs guarantee every VC's inputs are stored (see + // ClusteredValueGroupsBaseTableProjectionSpec), so recompute is always a correct fallback. + && !unrewritableFilterColumns.contains(outputName)) { + candidates.put(outputName, equivalent.getOutputName()); + } + } + if (candidates.isEmpty()) { + return Map.of(); + } + + // Transitive "must keep" closure: seed with the non-substituted query VCs (those recompute), then follow + // requiredColumns() to every query VC they depend on. + final Set keep = new HashSet<>(); + final Deque toVisit = new ArrayDeque<>(); + for (VirtualColumn vc : all) { + if (!candidates.containsKey(vc.getOutputName())) { + toVisit.add(vc.getOutputName()); + } + } + while (!toVisit.isEmpty()) { + final VirtualColumn vc = byName.get(toVisit.poll()); + if (vc == null) { + continue; + } + for (String dep : vc.requiredColumns()) { + // Only query VCs matter here; a dep that a kept VC recomputes from must itself be kept (and its deps) + if (byName.containsKey(dep) && keep.add(dep)) { + toVisit.add(dep); + } + } + } + + final Map remap = new HashMap<>(); + for (Map.Entry e : candidates.entrySet()) { + if (!keep.contains(e.getKey())) { + remap.put(e.getKey(), e.getValue()); + } + } + return remap; } /** * Walk the filter tree against {@code group}'s constant clustering tuple and return a rewritten filter where each - * leaf whose column resolves to a clustering column (physical or virtual) is folded to {@link TrueFilter} / - * {@link FalseFilter}, with those constants propagated through AND / OR / NOT. All other filters remain unchanged. + * recognized equality / in / null leaf whose column resolves to a clustering column (physical or virtual) is folded + * to {@link TrueFilter} / {@link FalseFilter}, with those constants propagated through AND / OR / NOT. All other + * filters (including other predicate shapes on a clustering column, e.g. range / like) remain unchanged. *

* Output shape encodes the truth value implicitly: a top-level {@link FalseFilter} means the filter is provably * FALSE against this group's clustering tuple (the planner uses this to decide which groups to prune); a top-level @@ -857,16 +966,47 @@ private static boolean isUnalignedInterval( private static Filter remapFilterToProjection(ProjectionMatchBuilder matchBuilder, Filter aggFilter) { + return rewriteFilterRequiredColumns(aggFilter, matchBuilder.getRemapColumns()); + } + + /** + * Whether {@code filter} can be remapped into the projection's column namespace via {@link #remapFilterToProjection}: + * either it references none of the remapped columns (nothing to rewrite) or it supports required-column rewrite. A + * filter that references a remapped column but can't rewrite can't be remapped, so callers must reject the + * projection match and fall back to the base table rather than letting {@link Filter#rewriteRequiredColumns} throw. + */ + private static boolean canRemapFilterToProjection(ProjectionMatchBuilder matchBuilder, Filter filter) + { + return filter.supportsRequiredColumnRewrite() + || Collections.disjoint(filter.getRequiredColumns(), matchBuilder.getRemapColumns().keySet()); + } + + /** + * Rewrite {@code filter}'s required columns through {@code remap}, tolerating columns the map doesn't mention. + * {@link Filter#rewriteRequiredColumns} throws on any required column missing from the rewrite map, so we seed an + * identity mapping over the filter's own required columns and overlay {@code remap} on top: residual leaves (a + * predicate on a column with no remap entry, e.g. {@code region = 'us-east-1'}) keep their own column, while matched + * columns are redirected to the materialized column. Shared by the aggregate-projection remap + * ({@link #remapFilterToProjection}) and the clustered base-table virtual-column remap + * ({@link ClusterGroupQueryPlan#rebuildCursorBuildSpec}). + *

+ * When the filter references none of the remapped columns, it is returned unchanged without calling + * {@link Filter#rewriteRequiredColumns} at all as there is nothing to rewrite. + */ + static Filter rewriteFilterRequiredColumns(Filter filter, Map remap) + { + final Set requiredColumns = filter.getRequiredColumns(); + if (Collections.disjoint(requiredColumns, remap.keySet())) { + return filter; + } final Map filterRewrites = new HashMap<>(); - // start with identity - for (String required : aggFilter.getRequiredColumns()) { + // start with identity so residual columns not mentioned in the remap are preserved rather than rejected + for (String required : requiredColumns) { filterRewrites.put(required, required); } - // overlay projection rewrites - filterRewrites.putAll(matchBuilder.getRemapColumns()); - - final Filter remappedAggFilter = aggFilter.rewriteRequiredColumns(filterRewrites); - return remappedAggFilter; + // overlay the remap so matched columns win over their identity entry + filterRewrites.putAll(remap); + return filter.rewriteRequiredColumns(filterRewrites); } /** diff --git a/processing/src/main/java/org/apache/druid/segment/vector/ConcatenatingVectorCursor.java b/processing/src/main/java/org/apache/druid/segment/vector/ConcatenatingVectorCursor.java index c2cfe44efd7d..73328d4d21b3 100644 --- a/processing/src/main/java/org/apache/druid/segment/vector/ConcatenatingVectorCursor.java +++ b/processing/src/main/java/org/apache/druid/segment/vector/ConcatenatingVectorCursor.java @@ -25,6 +25,7 @@ import org.apache.druid.segment.projections.ClusteringVectorColumnSelectorFactory; import java.util.List; +import java.util.Map; /** * Vector-cursor counterpart of {@link org.apache.druid.segment.ConcatenatingCursor}. Walks a sequence of per-group @@ -45,6 +46,7 @@ public final class ConcatenatingVectorCursor implements VectorCursor private final List> holderSuppliers; private final List clusteringValuesByGroup; private final ClusteringVectorColumnSelectorFactory wrapperFactory; + private final VectorColumnSelectorFactory exposedFactory; private int currentIdx; private VectorCursor currentCursor; @@ -53,7 +55,8 @@ public final class ConcatenatingVectorCursor implements VectorCursor public ConcatenatingVectorCursor( List> holderSuppliers, List clusteringValuesByGroup, - ClusteringVectorColumnSelectorFactory wrapperFactory + ClusteringVectorColumnSelectorFactory wrapperFactory, + Map virtualColumnRemap ) { if (holderSuppliers.size() != clusteringValuesByGroup.size()) { @@ -69,6 +72,9 @@ public ConcatenatingVectorCursor( this.holderSuppliers = holderSuppliers; this.clusteringValuesByGroup = clusteringValuesByGroup; this.wrapperFactory = wrapperFactory; + this.exposedFactory = virtualColumnRemap.isEmpty() + ? wrapperFactory + : new RemapVectorColumnSelectorFactory(wrapperFactory, virtualColumnRemap); this.currentIdx = -1; } @@ -106,7 +112,7 @@ private void advanceToNextNonEmptyGroup() public VectorColumnSelectorFactory getColumnSelectorFactory() { initializeIfNeeded(); - return wrapperFactory; + return exposedFactory; } @Override diff --git a/processing/src/test/java/org/apache/druid/data/input/impl/ClusteredValueGroupsBaseTableProjectionSpecTest.java b/processing/src/test/java/org/apache/druid/data/input/impl/ClusteredValueGroupsBaseTableProjectionSpecTest.java index f6edc088dfdf..603919e71128 100644 --- a/processing/src/test/java/org/apache/druid/data/input/impl/ClusteredValueGroupsBaseTableProjectionSpecTest.java +++ b/processing/src/test/java/org/apache/druid/data/input/impl/ClusteredValueGroupsBaseTableProjectionSpecTest.java @@ -21,11 +21,23 @@ import org.apache.druid.error.DruidException; import org.apache.druid.java.util.common.granularity.Granularities; +import org.apache.druid.query.dimension.DimensionSpec; +import org.apache.druid.query.expression.TestExprMacroTable; +import org.apache.druid.segment.ColumnSelectorFactory; +import org.apache.druid.segment.ColumnValueSelector; +import org.apache.druid.segment.DimensionSelector; import org.apache.druid.segment.VirtualColumn; +import org.apache.druid.segment.VirtualColumns; +import org.apache.druid.segment.column.ColumnCapabilities; +import org.apache.druid.segment.column.ColumnType; +import org.apache.druid.segment.virtual.ExpressionVirtualColumn; import org.apache.druid.testing.InitializedNullHandlingTest; import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.Test; +import java.util.Collections; +import java.util.List; + class ClusteredValueGroupsBaseTableProjectionSpecTest extends InitializedNullHandlingTest { private static ClusteredValueGroupsBaseTableProjectionSpec tenantSpec() @@ -132,4 +144,209 @@ void testWithQueryGranularityIsIdempotentWhenAlreadyPresent() ) ); } + + @Test + void testVirtualColumnMaterializedFromStoredInputIsValid() + { + // region_upper := upper(region): the input region is stored and the output region_upper is stored -> valid. + final ClusteredValueGroupsBaseTableProjectionSpec spec = ClusteredValueGroupsBaseTableProjectionSpec.builder() + .virtualColumns(VirtualColumns.create( + new ExpressionVirtualColumn("region_upper", "upper(region)", ColumnType.STRING, TestExprMacroTable.INSTANCE) + )) + .columns( + new StringDimensionSchema("tenant"), + new StringDimensionSchema("region"), + new StringDimensionSchema("region_upper"), + new LongDimensionSchema("__time") + ) + .clusteringColumns("tenant") + .build(); + Assertions.assertNotNull(spec.getVirtualColumns().getVirtualColumn("region_upper")); + } + + @Test + void testVirtualColumnChainWithUnstoredIntermediateIsValid() + { + // Chain: clustering column tenant_key := upper(v0) where v0 := lower(tenant). The physical leaf tenant is stored + // and the clustering output tenant_key is stored; the intermediary v0 is NOT stored but feeds tenant_key, so it is + // exempt from the output rule -> valid. + final ClusteredValueGroupsBaseTableProjectionSpec spec = ClusteredValueGroupsBaseTableProjectionSpec.builder() + .virtualColumns(VirtualColumns.create( + new ExpressionVirtualColumn("v0", "lower(tenant)", ColumnType.STRING, TestExprMacroTable.INSTANCE), + new ExpressionVirtualColumn("tenant_key", "upper(v0)", ColumnType.STRING, TestExprMacroTable.INSTANCE) + )) + .columns( + new StringDimensionSchema("tenant_key"), + new StringDimensionSchema("tenant"), + new LongDimensionSchema("__time") + ) + .clusteringColumns("tenant_key") + .build(); + Assertions.assertNotNull(spec.getVirtualColumns().getVirtualColumn("tenant_key")); + Assertions.assertNotNull(spec.getVirtualColumns().getVirtualColumn("v0")); + } + + @Test + void testVirtualColumnWithUnretainedInputIsRejected() + { + // tenant_lower := lower(tenant) is materialized (the clustering column), but the raw tenant input is NOT stored, so + // it can't be recomputed from stored columns -> rejected. + final DruidException e = Assertions.assertThrows( + DruidException.class, + () -> ClusteredValueGroupsBaseTableProjectionSpec.builder() + .virtualColumns(VirtualColumns.create( + new ExpressionVirtualColumn("tenant_lower", "lower(tenant)", ColumnType.STRING, TestExprMacroTable.INSTANCE) + )) + .columns( + new StringDimensionSchema("tenant_lower"), + new LongDimensionSchema("__time") + ) + .clusteringColumns("tenant_lower") + .build() + ); + Assertions.assertTrue(e.getMessage().contains("[tenant]")); + } + + @Test + void testDanglingVirtualColumnIsRejected() + { + // region_upper := upper(region) reads a stored column, but its output is neither stored nor an input to another + // virtual column -> dead metadata, rejected. + final DruidException e = Assertions.assertThrows( + DruidException.class, + () -> ClusteredValueGroupsBaseTableProjectionSpec.builder() + .virtualColumns(VirtualColumns.create( + new ExpressionVirtualColumn("region_upper", "upper(region)", ColumnType.STRING, TestExprMacroTable.INSTANCE) + )) + .columns( + new StringDimensionSchema("tenant"), + new StringDimensionSchema("region"), + new LongDimensionSchema("__time") + ) + .clusteringColumns("tenant") + .build() + ); + Assertions.assertTrue(e.getMessage().contains("[region_upper]")); + } + + @Test + void testVirtualColumnOutputTypeMismatchIsRejected() + { + // region_upper := upper(region) produces STRING, but the column it materializes is declared LONG -> rejected. + final DruidException e = Assertions.assertThrows( + DruidException.class, + () -> ClusteredValueGroupsBaseTableProjectionSpec.builder() + .virtualColumns(VirtualColumns.create( + new ExpressionVirtualColumn("region_upper", "upper(region)", ColumnType.STRING, TestExprMacroTable.INSTANCE) + )) + .columns( + new StringDimensionSchema("tenant"), + new StringDimensionSchema("region"), + new LongDimensionSchema("region_upper"), + new LongDimensionSchema("__time") + ) + .clusteringColumns("tenant") + .build() + ); + Assertions.assertTrue(e.getMessage().contains("[region_upper]")); + Assertions.assertTrue(e.getMessage().contains("STRING")); + Assertions.assertTrue(e.getMessage().contains("LONG")); + } + + @Test + void testNumericVirtualColumnMatchingDeclaredTypeIsValid() + { + // region_len := strlen(region) produces LONG and is declared LONG -> accepted (proves numeric output-type inference + // works, not just STRING). + final ClusteredValueGroupsBaseTableProjectionSpec spec = ClusteredValueGroupsBaseTableProjectionSpec.builder() + .virtualColumns(VirtualColumns.create( + new ExpressionVirtualColumn("region_len", "strlen(region)", ColumnType.LONG, TestExprMacroTable.INSTANCE) + )) + .columns( + new StringDimensionSchema("tenant"), + new StringDimensionSchema("region"), + new LongDimensionSchema("region_len"), + new LongDimensionSchema("__time") + ) + .clusteringColumns("tenant") + .build(); + Assertions.assertNotNull(spec.getVirtualColumns().getVirtualColumn("region_len")); + } + + @Test + void testDotNotationVirtualColumnIsRejected() + { + // Dot-notation virtual columns are referenced as "name.subfield" and have no fixed output identity to materialize + // or cluster on, so they are not supported in clustered base table specs. + final DruidException e = Assertions.assertThrows( + DruidException.class, + () -> ClusteredValueGroupsBaseTableProjectionSpec.builder() + .virtualColumns(VirtualColumns.create(new DotNotationVirtualColumn("dotty"))) + .columns( + new StringDimensionSchema("tenant"), + new StringDimensionSchema("region"), + new LongDimensionSchema("__time") + ) + .clusteringColumns("tenant") + .build() + ); + Assertions.assertTrue(e.getMessage().contains("dot notation")); + Assertions.assertTrue(e.getMessage().contains("[dotty]")); + } + + /** + * Minimal test-only virtual column whose only meaningful behavior is {@link #usesDotNotation()} returning true; the + * selector/capability methods are never reached by spec validation. (No core virtual column uses dot notation.) + */ + private static final class DotNotationVirtualColumn implements VirtualColumn + { + private final String name; + + private DotNotationVirtualColumn(String name) + { + this.name = name; + } + + @Override + public String getOutputName() + { + return name; + } + + @Override + public DimensionSelector makeDimensionSelector(DimensionSpec dimensionSpec, ColumnSelectorFactory factory) + { + throw new UnsupportedOperationException(); + } + + @Override + public ColumnValueSelector makeColumnValueSelector(String columnName, ColumnSelectorFactory factory) + { + throw new UnsupportedOperationException(); + } + + @Override + public ColumnCapabilities capabilities(String columnName) + { + throw new UnsupportedOperationException(); + } + + @Override + public List requiredColumns() + { + return Collections.emptyList(); + } + + @Override + public boolean usesDotNotation() + { + return true; + } + + @Override + public byte[] getCacheKey() + { + throw new UnsupportedOperationException(); + } + } } diff --git a/processing/src/test/java/org/apache/druid/segment/ClusteredSegmentTimeOrderedQueryTest.java b/processing/src/test/java/org/apache/druid/segment/ClusteredSegmentTimeOrderedQueryTest.java index e3eb0b0778de..b067b4ccc942 100644 --- a/processing/src/test/java/org/apache/druid/segment/ClusteredSegmentTimeOrderedQueryTest.java +++ b/processing/src/test/java/org/apache/druid/segment/ClusteredSegmentTimeOrderedQueryTest.java @@ -29,6 +29,7 @@ import org.apache.druid.java.util.common.DateTimes; import org.apache.druid.java.util.common.Intervals; import org.apache.druid.java.util.common.granularity.Granularities; +import org.apache.druid.math.expr.ExpressionProcessing; import org.apache.druid.query.DefaultGenericQueryMetricsFactory; import org.apache.druid.query.Druids; import org.apache.druid.query.Order; @@ -36,6 +37,7 @@ import org.apache.druid.query.QueryRunnerTestHelper; import org.apache.druid.query.Result; import org.apache.druid.query.aggregation.LongSumAggregatorFactory; +import org.apache.druid.query.expression.TestExprMacroTable; import org.apache.druid.query.scan.ScanQuery; import org.apache.druid.query.scan.ScanQueryConfig; import org.apache.druid.query.scan.ScanQueryEngine; @@ -49,7 +51,9 @@ import org.apache.druid.query.timeseries.TimeseriesQueryRunnerFactory; import org.apache.druid.query.timeseries.TimeseriesResultValue; import org.apache.druid.segment.column.ColumnHolder; +import org.apache.druid.segment.column.ColumnType; import org.apache.druid.segment.incremental.IncrementalIndexSchema; +import org.apache.druid.segment.virtual.ExpressionVirtualColumn; import org.apache.druid.segment.writeout.OffHeapMemorySegmentWriteOutMediumFactory; import org.apache.druid.testing.InitializedNullHandlingTest; import org.apache.druid.timeline.SegmentId; @@ -107,6 +111,7 @@ class ClusteredSegmentTimeOrderedQueryTest extends InitializedNullHandlingTest @BeforeEach void setUp() { + ExpressionProcessing.initializeForTests(); clusteredSegment = new QueryableIndexSegment(buildClustered("clustered", ROWS), SegmentId.dummy(DATA_SOURCE)); nonClusteredSegment = new QueryableIndexSegment(buildNonClustered("plain", ROWS), SegmentId.dummy(DATA_SOURCE)); } @@ -199,6 +204,27 @@ void testDescendingScanOverSingleGroupClusteredSegment() Assertions.assertEquals(runScanRows(plain, Order.DESCENDING), runScanRows(clustered, Order.DESCENDING)); } + @Test + void testAscendingScanWithQueryVcEquivalentToClusteringColumn() + { + // Clustered on tenant_lower := lower(tenant), with __time the first non-clustering column, so a time-ordered scan + // takes the k-way merge. A query VC v0 := lower(tenant) is equivalent to the materialized clustering column + // tenant_lower, so the merge's RemapColumnSelectorFactory must resolve reads of v0 to tenant_lower. Without that + // wrapper the per-group cursor would not find v0 (rebuildCursorBuildSpec drops the remapped VC from the per-group + // spec), so v0 would read as null and this test would fail. + final Segment clustered = new QueryableIndexSegment( + buildClusteredWithMaterializedVc("clustered-remap"), + SegmentId.dummy(DATA_SOURCE) + ); + final List> expected = List.of( + Arrays.asList(T0, "acme", 1L), + Arrays.asList(T0 + MINUTE, "globex", 2L), + Arrays.asList(T0 + 2 * MINUTE, "acme", 4L), + Arrays.asList(T0 + 3 * MINUTE, "globex", 8L) + ); + Assertions.assertEquals(expected, runScanWithLowerTenantVc(clustered, Order.ASCENDING)); + } + private static List> runTimeseries(Segment segment) { final TimeseriesQuery query = Druids.newTimeseriesQueryBuilder() @@ -254,6 +280,83 @@ private static List> runScanRows(Segment segment, Order order) return rows; } + /** + * Time-ordered scan projecting the query virtual column {@code v0 := lower(tenant)} (plus {@code __time} and + * {@code m}), extracting {@code [__time, v0, m]} rows. Used to exercise the query-VC -> materialized-column remap on + * the {@code __time} merge path. + */ + private static List> runScanWithLowerTenantVc(Segment segment, Order order) + { + final ScanQuery query = Druids.newScanQueryBuilder() + .dataSource(DATA_SOURCE) + .intervals(new MultipleIntervalSegmentSpec(List.of(INTERVAL))) + .virtualColumns(new ExpressionVirtualColumn( + "v0", + "lower(tenant)", + ColumnType.STRING, + TestExprMacroTable.INSTANCE + )) + .columns(ColumnHolder.TIME_COLUMN_NAME, "v0", "m") + .order(order) + .resultFormat(ScanQuery.ResultFormat.RESULT_FORMAT_LIST) + .build(); + final ScanQueryRunnerFactory factory = new ScanQueryRunnerFactory( + new ScanQueryQueryToolChest(DefaultGenericQueryMetricsFactory.instance()), + new ScanQueryEngine(), + new ScanQueryConfig() + ); + final List results = factory.createRunner(segment).run(QueryPlus.wrap(query)).toList(); + final List> rows = new ArrayList<>(); + for (ScanResultValue result : results) { + for (Object event : (List) result.getEvents()) { + @SuppressWarnings("unchecked") + final Map row = (Map) event; + rows.add(Arrays.asList( + DimensionHandlerUtils.convertObjectToLong(row.get(ColumnHolder.TIME_COLUMN_NAME)), + row.get("v0"), + DimensionHandlerUtils.convertObjectToLong(row.get("m")) + )); + } + } + return rows; + } + + private QueryableIndex buildClusteredWithMaterializedVc(String dirName) + { + // Cluster on the group VC tenant_lower := lower(tenant); declare __time immediately after the clustering column so + // it is the first non-clustering column (each group is __time-sorted => merge path). The raw tenant input is + // retained so the group VC's input is materialized (a spec-construction requirement). + final ClusteredValueGroupsBaseTableProjectionSpec clusterSpec = + ClusteredValueGroupsBaseTableProjectionSpec.builder() + .virtualColumns(VirtualColumns.create( + new ExpressionVirtualColumn("tenant_lower", "lower(tenant)", ColumnType.STRING, TestExprMacroTable.INSTANCE) + )) + .columns( + new StringDimensionSchema("tenant_lower"), + new LongDimensionSchema("__time"), + StringDimensionSchema.create("tenant"), + new LongDimensionSchema("m") + ) + .clusteringColumns("tenant_lower") + .build(); + final IncrementalIndexSchema schema = + IncrementalIndexSchema.builder() + .withMinTimestamp(T0) + .withTimestampSpec(TIMESTAMP_SPEC) + .withQueryGranularity(Granularities.NONE) + .withDimensionsSpec(clusterSpec.getDimensionsSpec()) + .withRollup(false) + .withClusterSpec(clusterSpec) + .build(); + return IndexBuilder.create() + .useV10() + .tmpDir(new File(tempDir, dirName)) + .segmentWriteOutMediumFactory(OffHeapMemorySegmentWriteOutMediumFactory.instance()) + .schema(schema) + .rows(ROWS) + .buildMMappedIndex(); + } + private QueryableIndex buildClustered(String dirName, List rows) { // columns [tenant, __time, m], clustering [tenant] => ordering [tenant, __time, m], so __time is the first diff --git a/processing/src/test/java/org/apache/druid/segment/ConcatenatingCursorTest.java b/processing/src/test/java/org/apache/druid/segment/ConcatenatingCursorTest.java index 36d2f8c547fc..29a6cc54c260 100644 --- a/processing/src/test/java/org/apache/druid/segment/ConcatenatingCursorTest.java +++ b/processing/src/test/java/org/apache/druid/segment/ConcatenatingCursorTest.java @@ -34,6 +34,7 @@ import javax.annotation.Nullable; import java.util.ArrayList; import java.util.List; +import java.util.Map; class ConcatenatingCursorTest { @@ -56,7 +57,8 @@ void testWalksTwoNonEmptyGroupsBackToBack() ConcatenatingCursor c = new ConcatenatingCursor( List.of(holderSupplier(a), holderSupplier(b)), List.of(new Object[]{"acme"}, new Object[]{"globex"}), - wrapper + wrapper, + Map.of() ); ColumnValueSelector tenant = c.getColumnSelectorFactory().makeColumnValueSelector("tenant"); @@ -99,7 +101,8 @@ void testSkipsLeadingEmptyGroup() ConcatenatingCursor c = new ConcatenatingCursor( List.of(holderSupplier(empty), holderSupplier(full)), List.of(new Object[]{"a"}, new Object[]{"b"}), - wrapper + wrapper, + Map.of() ); ColumnValueSelector tenant = c.getColumnSelectorFactory().makeColumnValueSelector("tenant"); @@ -127,7 +130,8 @@ void testSkipsTrailingEmptyGroup() ConcatenatingCursor c = new ConcatenatingCursor( List.of(holderSupplier(full), holderSupplier(empty)), List.of(new Object[]{"a"}, new Object[]{"b"}), - wrapper + wrapper, + Map.of() ); ColumnValueSelector tenant = c.getColumnSelectorFactory().makeColumnValueSelector("tenant"); @@ -152,7 +156,8 @@ void testAllEmptyGroups() ConcatenatingCursor c = new ConcatenatingCursor( List.of(holderSupplier(e1), holderSupplier(e2)), List.of(new Object[]{"a"}, new Object[]{"b"}), - wrapper + wrapper, + Map.of() ); Assertions.assertTrue(c.isDone()); @@ -172,7 +177,8 @@ void testSingleGroupDegenerateCase() ConcatenatingCursor c = new ConcatenatingCursor( List.of(holderSupplier(only)), List.of(new Object[]{"a"}), - wrapper + wrapper, + Map.of() ); ColumnValueSelector tenant = c.getColumnSelectorFactory().makeColumnValueSelector("tenant"); @@ -202,7 +208,8 @@ void testCloseClosesAllOpenedHolders() ConcatenatingCursor c = new ConcatenatingCursor( List.of(holderSupplier(a), holderSupplier(b)), List.of(new Object[]{"x"}, new Object[]{"y"}), - wrapper + wrapper, + Map.of() ); // Walk through, opens both holders. @@ -236,7 +243,8 @@ void testGroupsAreOpenedLazilyOnTransitionNotEagerly() ConcatenatingCursor c = new ConcatenatingCursor( suppliers, List.of(new Object[]{"a"}, new Object[]{"b"}), - wrapper + wrapper, + Map.of() ); // Init opens first group only. Second is still untouched. @@ -265,7 +273,8 @@ void testResetReIteratesWithoutClosingHolders() ConcatenatingCursor c = new ConcatenatingCursor( List.of(holderSupplier(a), holderSupplier(b)), List.of(new Object[]{"acme"}, new Object[]{"globex"}), - wrapper + wrapper, + Map.of() ); ColumnValueSelector tenant = c.getColumnSelectorFactory().makeColumnValueSelector("tenant"); diff --git a/processing/src/test/java/org/apache/druid/segment/MergingClusterGroupCursorTest.java b/processing/src/test/java/org/apache/druid/segment/MergingClusterGroupCursorTest.java index daf38d3a2cca..9c3f3b187f1b 100644 --- a/processing/src/test/java/org/apache/druid/segment/MergingClusterGroupCursorTest.java +++ b/processing/src/test/java/org/apache/druid/segment/MergingClusterGroupCursorTest.java @@ -29,6 +29,7 @@ import javax.annotation.Nullable; import java.util.ArrayList; import java.util.List; +import java.util.Map; class MergingClusterGroupCursorTest { @@ -153,10 +154,36 @@ void testRowIdMonotonicAndStableWithinRow() } } + @Test + void testVirtualColumnRemapRedirectsReads() + { + // With a remap {aliased -> v}, reads of the query virtual column's output name "aliased" through the exposed + // factory must resolve to the materialized "v" column of the winning group. Without the remap the per-group + // factories reject "aliased" (see ListCursor), so this proves the RemapColumnSelectorFactory is applied on top of + // the merge. + final MergingClusterGroupCursor cursor = new MergingClusterGroupCursor( + new ArrayList<>(List.of( + group(new long[]{1, 3}, new String[]{"g0@1", "g0@3"}), + group(new long[]{2, 4}, new String[]{"g1@2", "g1@4"}) + )), + false, + Map.of("aliased", "v") + ); + final ColumnSelectorFactory factory = cursor.getColumnSelectorFactory(); + final ColumnValueSelector timeSelector = factory.makeColumnValueSelector(ColumnHolder.TIME_COLUMN_NAME); + final ColumnValueSelector aliasSelector = factory.makeColumnValueSelector("aliased"); + final List rows = new ArrayList<>(); + while (!cursor.isDone()) { + rows.add(new Object[]{timeSelector.getLong(), aliasSelector.getObject()}); + cursor.advance(); + } + assertRows(rows, new long[]{1, 2, 3, 4}, new String[]{"g0@1", "g1@2", "g0@3", "g1@4"}); + } + @SafeVarargs private static MergingClusterGroupCursor cursor(boolean descending, Supplier... groups) { - return new MergingClusterGroupCursor(new ArrayList<>(List.of(groups)), descending); + return new MergingClusterGroupCursor(new ArrayList<>(List.of(groups)), descending, Map.of()); } private static List drain(MergingClusterGroupCursor cursor) @@ -237,6 +264,11 @@ public boolean isNull() } }; } + if (!"v".equals(columnName)) { + // Strict on purpose: a per-group cursor only serves its materialized columns. This makes the remap test + // meaningful (reading a query VC output name must be redirected to "v" before reaching this factory). + throw new UnsupportedOperationException("unknown column [" + columnName + "]"); + } return new TestObjectColumnSelector() { @Override diff --git a/processing/src/test/java/org/apache/druid/segment/QueryableIndexCursorFactoryClusteredTest.java b/processing/src/test/java/org/apache/druid/segment/QueryableIndexCursorFactoryClusteredTest.java index e89123ac9f9c..05cd220ec8d6 100644 --- a/processing/src/test/java/org/apache/druid/segment/QueryableIndexCursorFactoryClusteredTest.java +++ b/processing/src/test/java/org/apache/druid/segment/QueryableIndexCursorFactoryClusteredTest.java @@ -39,6 +39,7 @@ import org.apache.druid.query.Result; import org.apache.druid.query.aggregation.CountAggregatorFactory; import org.apache.druid.query.dimension.DefaultDimensionSpec; +import org.apache.druid.query.expression.TestExprMacroTable; import org.apache.druid.query.filter.ColumnIndexSelector; import org.apache.druid.query.filter.EqualityFilter; import org.apache.druid.query.filter.Filter; @@ -63,6 +64,10 @@ import org.apache.druid.segment.filter.OrFilter; import org.apache.druid.segment.incremental.IncrementalIndexSchema; import org.apache.druid.segment.index.BitmapColumnIndex; +import org.apache.druid.segment.vector.VectorCursor; +import org.apache.druid.segment.vector.VectorObjectSelector; +import org.apache.druid.segment.virtual.ExpressionVirtualColumn; +import org.apache.druid.testing.InitializedNullHandlingTest; import org.joda.time.Interval; import org.junit.jupiter.api.AfterAll; import org.junit.jupiter.api.AfterEach; @@ -85,7 +90,7 @@ * segments directly via {@link IndexBuilder} (ingesting flat rows that the writer partitions into cluster groups by * clustering value) and runs the full per-group cursor pipeline against the actual on-disk clustered segment data. */ -class QueryableIndexCursorFactoryClusteredTest +class QueryableIndexCursorFactoryClusteredTest extends InitializedNullHandlingTest { private static final Interval INTERVAL = Intervals.of("2025-01-01/2025-01-02"); @@ -139,48 +144,306 @@ static void tearDownEngines() throws Exception private QueryableIndex segmentIndex; @AfterEach - void tearDown() throws java.io.IOException + void tearDown() { if (segmentIndex != null) { segmentIndex.close(); } } - private static InputRow row(String tenant, String ts, String region) + /** + * Cluster spec for a segment clustered on {@code tenant_lower := lower(tenant)} (a clustering column produced by a + * group VC) plus a non-clustering materialized {@code region_upper := upper(region)} column. The raw inputs + * {@code tenant} and {@code region} are retained as stored columns, so the query-VC -> materialized-column remap is a + * pure optimization (the query VC could also be recomputed from the retained input). Columns: + * {@code [tenant_lower (clustering), tenant, region, region_upper, __time]}. + */ + private static final ClusteredValueGroupsBaseTableProjectionSpec VIRTUAL_CLUSTER_SPEC = + ClusteredValueGroupsBaseTableProjectionSpec.builder() + .virtualColumns(VirtualColumns.create( + new ExpressionVirtualColumn("tenant_lower", "lower(tenant)", ColumnType.STRING, TestExprMacroTable.INSTANCE), + new ExpressionVirtualColumn("region_upper", "upper(region)", ColumnType.STRING, TestExprMacroTable.INSTANCE) + )) + .columns( + new StringDimensionSchema("tenant_lower"), + StringDimensionSchema.create("tenant"), + StringDimensionSchema.create("region"), + StringDimensionSchema.create("region_upper"), + new LongDimensionSchema("__time") + ) + .clusteringColumns("tenant_lower") + .build(); + + @Test + void testQueryVcEquivalentToClusteringColumnReadsMaterializedColumnViaAsCursor() { - return new MapBasedInputRow( - DateTimes.of(ts), - List.of("tenant", "region"), - Map.of("tenant", tenant, "region", region) + // Query VC v0 := lower(tenant) is equivalent to the clustering column tenant_lower. The scalar selector remap + // substitutes the materialized tenant_lower clustering constant instead of recomputing lower(tenant) from the + // retained raw column. Read via asCursor() (non-vector) to exercise the scalar ClusteringColumnSelectorFactory path. + segmentIndex = buildVirtualClusteringSegment(); + final QueryableIndexCursorFactory factory = new QueryableIndexCursorFactory( + segmentIndex, + QueryableIndexTimeBoundaryInspector.create(segmentIndex) ); + final CursorBuildSpec buildSpec = CursorBuildSpec.builder() + .setVirtualColumns(VirtualColumns.create( + new ExpressionVirtualColumn("v0", "lower(tenant)", ColumnType.STRING, TestExprMacroTable.INSTANCE) + )) + .build(); + try (CursorHolder holder = factory.makeCursorHolder(buildSpec)) { + Assertions.assertEquals(List.of("acme", "acme", "globex"), collectDimension(holder.asCursor(), "v0")); + } } - private QueryableIndex buildSegment(List rows) + @Test + void testQueryVcEquivalentToNonClusteringMaterializedColumnReadsMaterializedColumnViaAsCursor() { - final IncrementalIndexSchema schema = - IncrementalIndexSchema.builder() - .withMinTimestamp(INTERVAL.getStartMillis()) - .withTimestampSpec(new TimestampSpec("__time", "auto", null)) - .withQueryGranularity(Granularities.NONE) - .withDimensionsSpec(CLUSTER_SPEC.getDimensionsSpec()) - .withRollup(false) - .withClusterSpec(CLUSTER_SPEC) - .build(); - return IndexBuilder.create() - .useV10() - .tmpDir(tmpDir) - .schema(schema) - .rows(rows) - .buildMMappedIndex(INTERVAL); + // Query VC v1 := upper(region) is equivalent to the non-clustering materialized column region_upper. The remap + // makes makeDimensionSelector("v1") read the per-group physical region_upper column. + segmentIndex = buildVirtualClusteringSegment(); + final QueryableIndexCursorFactory factory = new QueryableIndexCursorFactory( + segmentIndex, + QueryableIndexTimeBoundaryInspector.create(segmentIndex) + ); + final CursorBuildSpec buildSpec = CursorBuildSpec.builder() + .setVirtualColumns(VirtualColumns.create( + new ExpressionVirtualColumn("v1", "upper(region)", ColumnType.STRING, TestExprMacroTable.INSTANCE) + )) + .build(); + try (CursorHolder holder = factory.makeCursorHolder(buildSpec)) { + Assertions.assertEquals( + List.of("US-EAST-1", "US-WEST-2", "EU-WEST-1"), + collectDimension(holder.asCursor(), "v1") + ); + } } - private QueryableIndex standardTwoGroup() + @Test + void testQueryVcWithNoEquivalentStillRecomputesViaAsCursor() { - return buildSegment(List.of( - row("acme", "2025-01-01T00:00:00", "us-east-1"), - row("acme", "2025-01-01T01:00:00", "us-west-2"), - row("globex", "2025-01-01T00:30:00", "eu-west-1") - )); + // No-regression: query VC with no materialized equivalent (lower(region_upper)) is recomputed from the stored + // region_upper column, not remapped. + segmentIndex = buildVirtualClusteringSegment(); + final QueryableIndexCursorFactory factory = new QueryableIndexCursorFactory( + segmentIndex, + QueryableIndexTimeBoundaryInspector.create(segmentIndex) + ); + final CursorBuildSpec buildSpec = CursorBuildSpec.builder() + .setVirtualColumns(VirtualColumns.create( + new ExpressionVirtualColumn("v2", "lower(region_upper)", ColumnType.STRING, TestExprMacroTable.INSTANCE) + )) + .build(); + try (CursorHolder holder = factory.makeCursorHolder(buildSpec)) { + Assertions.assertEquals( + List.of("us-east-1", "us-west-2", "eu-west-1"), + collectDimension(holder.asCursor(), "v2") + ); + } + } + + @Test + void testQueryVcShadowingClusteringColumnWinsOverConstantViaAsCursor() + { + // A query VC named "tenant" shadows the clustering column of the same name (computed from region instead). The + // concat wrapper must NOT serve the per-group clustering constant for "tenant"; it must defer to the delegate, + // which resolves the query VC (virtual columns win over physical columns). Two groups + no time ordering => the + // concatenating path. Without the shadow guard, reads of "tenant" would return the clustering constant + // ("acme"/"globex") instead of upper(region). + segmentIndex = standardTwoGroup(); + final QueryableIndexCursorFactory factory = new QueryableIndexCursorFactory( + segmentIndex, + QueryableIndexTimeBoundaryInspector.create(segmentIndex) + ); + final CursorBuildSpec buildSpec = CursorBuildSpec.builder() + .setVirtualColumns(VirtualColumns.create( + new ExpressionVirtualColumn("tenant", "upper(region)", ColumnType.STRING, TestExprMacroTable.INSTANCE) + )) + .build(); + try (CursorHolder holder = factory.makeCursorHolder(buildSpec)) { + Assertions.assertEquals( + List.of("US-EAST-1", "US-WEST-2", "EU-WEST-1"), + collectDimension(holder.asCursor(), "tenant") + ); + } + } + + @Test + void testQueryVcEquivalentToClusteringColumnSingleGroupReadsMaterializedColumn() + { + // Single surviving group (filter on the equivalent VC prunes to one group), exercising the single-group cursor + // holder path's scalar remap wrap. The remap reads the materialized tenant_lower clustering constant instead of + // recomputing lower(tenant) from the retained raw column. + segmentIndex = buildVirtualClusteringSegment(); + final QueryableIndexCursorFactory factory = new QueryableIndexCursorFactory( + segmentIndex, + QueryableIndexTimeBoundaryInspector.create(segmentIndex) + ); + final CursorBuildSpec buildSpec = CursorBuildSpec.builder() + .setVirtualColumns(VirtualColumns.create( + new ExpressionVirtualColumn("v0", "lower(tenant)", ColumnType.STRING, TestExprMacroTable.INSTANCE) + )) + .setFilter(new EqualityFilter("v0", ColumnType.STRING, "globex", null)) + .build(); + try (CursorHolder holder = factory.makeCursorHolder(buildSpec)) { + Assertions.assertEquals(List.of("globex"), collectDimension(holder.asCursor(), "v0")); + } + } + + @Test + void testEquivalentVcFilterPrunesToSingleGroupNotConcatenated() + { + // A filter on the ORIGINAL expression used to build the clustering column (lower(tenant) = 'acme', planned as a + // query VC) must prune to the single matching cluster group and take the single-group cursor path -- NOT survive + // every group and post-filter through the ConcatenatingCursor. + segmentIndex = buildVirtualClusteringSegment(); + final QueryableIndexCursorFactory factory = new QueryableIndexCursorFactory( + segmentIndex, + QueryableIndexTimeBoundaryInspector.create(segmentIndex) + ); + final VirtualColumns queryVcs = VirtualColumns.create( + new ExpressionVirtualColumn("v0", "lower(tenant)", ColumnType.STRING, TestExprMacroTable.INSTANCE) + ); + + // Control: unfiltered -> both groups survive -> multi-group ConcatenatingCursor. + try (CursorHolder holder = + factory.makeCursorHolder(CursorBuildSpec.builder().setVirtualColumns(queryVcs).build())) { + Assertions.assertInstanceOf(ConcatenatingCursor.class, holder.asCursor()); + } + + // Filtered on the equivalent VC -> the clustering leaf resolves to tenant_lower and folds, pruning to the single + // acme group -> single-group cursor (not a ConcatenatingCursor), and no residual filter on the missing raw column. + final CursorBuildSpec buildSpec = CursorBuildSpec.builder() + .setVirtualColumns(queryVcs) + .setFilter(new EqualityFilter("v0", ColumnType.STRING, "acme", null)) + .build(); + try (CursorHolder holder = factory.makeCursorHolder(buildSpec)) { + final Cursor cursor = holder.asCursor(); + Assertions.assertFalse(cursor instanceof ConcatenatingCursor); + Assertions.assertEquals(List.of("us-east-1", "us-west-2"), collectDimension(cursor, "region")); + } + } + + @Test + void testResidualPhysicalFilterPreservedAlongsideRemappedClusteringVc() + { + // A remapped clustering-VC leaf (v0 := lower(tenant) folds against the group tuple) combined with a residual + // predicate on a non-remapped physical column (region). Once the clustering leaf folds to TRUE, the per-group + // filter is region = 'us-east-1', which references no remap key -- the remap rewrite must preserve it (identity + // fallback) rather than throw for a column missing from {v0 -> tenant_lower}. + segmentIndex = buildVirtualClusteringSegment(); + final QueryableIndexCursorFactory factory = new QueryableIndexCursorFactory( + segmentIndex, + QueryableIndexTimeBoundaryInspector.create(segmentIndex) + ); + final CursorBuildSpec buildSpec = CursorBuildSpec.builder() + .setVirtualColumns(VirtualColumns.create( + new ExpressionVirtualColumn("v0", "lower(tenant)", ColumnType.STRING, TestExprMacroTable.INSTANCE) + )) + .setFilter(new AndFilter(List.of( + new EqualityFilter("v0", ColumnType.STRING, "acme", null), + new EqualityFilter("region", ColumnType.STRING, "us-east-1", null) + ))) + .build(); + try (CursorHolder holder = factory.makeCursorHolder(buildSpec)) { + // Only the Acme / us-east-1 row survives; v0 still resolves to the materialized tenant_lower clustering constant. + Assertions.assertEquals(List.of("acme"), collectDimension(holder.asCursor(), "v0")); + } + try (CursorHolder holder = factory.makeCursorHolder(buildSpec)) { + Assertions.assertEquals(List.of("us-east-1"), collectDimension(holder.asCursor(), "region")); + } + } + + @Test + void testResidualPhysicalFilterPreservedAlongsideRemappedNonClusteringVc() + { + // Reviewer's case: a matched non-clustering VC filter (v1 := upper(region), equivalent to region_upper) AND a + // residual physical predicate (region = 'us-east-1'). No clustering leaf folds, so the per-group filter stays an + // AndFilter whose recursion hits the residual region leaf; the identity fallback must keep region as-is while + // remapping v1 -> region_upper. + segmentIndex = buildVirtualClusteringSegment(); + final QueryableIndexCursorFactory factory = new QueryableIndexCursorFactory( + segmentIndex, + QueryableIndexTimeBoundaryInspector.create(segmentIndex) + ); + final CursorBuildSpec buildSpec = CursorBuildSpec.builder() + .setVirtualColumns(VirtualColumns.create( + new ExpressionVirtualColumn("v1", "upper(region)", ColumnType.STRING, TestExprMacroTable.INSTANCE) + )) + .setFilter(new AndFilter(List.of( + new EqualityFilter("v1", ColumnType.STRING, "US-EAST-1", null), + new EqualityFilter("region", ColumnType.STRING, "us-east-1", null) + ))) + .build(); + try (CursorHolder holder = factory.makeCursorHolder(buildSpec)) { + Assertions.assertEquals(List.of("us-east-1"), collectDimension(holder.asCursor(), "region")); + } + } + + @Test + void testQueryVcEquivalentToClusteringColumnReadsMaterializedColumnViaVectorCursor() + { + // the query-VC -> materialized-column remap is applied on the vector factory too, so a + // vectorized read of v0 := lower(tenant) resolves to the materialized clustering column tenant_lower (instead of + // recomputing lower(tenant) from the retained raw column), and the remap no longer forces the scalar path. + segmentIndex = buildVirtualClusteringSegment(); + final QueryableIndexCursorFactory factory = new QueryableIndexCursorFactory( + segmentIndex, + QueryableIndexTimeBoundaryInspector.create(segmentIndex) + ); + final CursorBuildSpec buildSpec = CursorBuildSpec.builder() + .setVirtualColumns(VirtualColumns.create( + new ExpressionVirtualColumn("v0", "lower(tenant)", ColumnType.STRING, TestExprMacroTable.INSTANCE) + )) + .build(); + try (CursorHolder holder = factory.makeCursorHolder(buildSpec)) { + Assertions.assertTrue(holder.canVectorize()); + Assertions.assertEquals(List.of("acme", "acme", "globex"), collectObjectVector(holder.asVectorCursor(), "v0")); + } + } + + @Test + void testQueryVcEquivalentToNonClusteringMaterializedColumnReadsMaterializedColumnViaVectorCursor() + { + // non-clustering: v1 := upper(region) resolves to the materialized region_upper physical column. + segmentIndex = buildVirtualClusteringSegment(); + final QueryableIndexCursorFactory factory = new QueryableIndexCursorFactory( + segmentIndex, + QueryableIndexTimeBoundaryInspector.create(segmentIndex) + ); + final CursorBuildSpec buildSpec = CursorBuildSpec.builder() + .setVirtualColumns(VirtualColumns.create( + new ExpressionVirtualColumn("v1", "upper(region)", ColumnType.STRING, TestExprMacroTable.INSTANCE) + )) + .build(); + try (CursorHolder holder = factory.makeCursorHolder(buildSpec)) { + Assertions.assertTrue(holder.canVectorize()); + Assertions.assertEquals( + List.of("US-EAST-1", "US-WEST-2", "EU-WEST-1"), + collectObjectVector(holder.asVectorCursor(), "v1") + ); + } + } + + @Test + void testQueryVcEquivalentToClusteringColumnSingleGroupVectorCursor() + { + // single surviving group (filter on the equivalent VC prunes to one group): exercises the single-group holder's + // vector remap wrap. + segmentIndex = buildVirtualClusteringSegment(); + final QueryableIndexCursorFactory factory = new QueryableIndexCursorFactory( + segmentIndex, + QueryableIndexTimeBoundaryInspector.create(segmentIndex) + ); + final CursorBuildSpec buildSpec = CursorBuildSpec.builder() + .setVirtualColumns(VirtualColumns.create( + new ExpressionVirtualColumn("v0", "lower(tenant)", ColumnType.STRING, TestExprMacroTable.INSTANCE) + )) + .setFilter(new EqualityFilter("v0", ColumnType.STRING, "globex", null)) + .build(); + try (CursorHolder holder = factory.makeCursorHolder(buildSpec)) { + Assertions.assertTrue(holder.canVectorize()); + Assertions.assertEquals(List.of("globex"), collectObjectVector(holder.asVectorCursor(), "v0")); + } } @Test @@ -238,32 +501,6 @@ void testGetColumnCapabilitiesForUnknownColumnIsNull() Assertions.assertNull(factory.getColumnCapabilities("nope")); } - /** - * Iterate the (tenant, region) pairs out of {@code cursor} until done. Verifies that the clustering-column - * selector returns the right per-group constant and that the rewritten filter / bitmap-index path on the - * per-group QueryableIndex produces the rows it should. - */ - private static List> collectTenantRegionRows(Cursor cursor) - { - final DimensionSelector tenantSel = - cursor.getColumnSelectorFactory().makeDimensionSelector(DefaultDimensionSpec.of("tenant")); - final DimensionSelector regionSel = - cursor.getColumnSelectorFactory().makeDimensionSelector(DefaultDimensionSpec.of("region")); - final List> out = new ArrayList<>(); - while (!cursor.isDone()) { - final String tenant = tenantSel.getRow().size() == 0 ? null : tenantSel.lookupName(tenantSel.getRow().get(0)); - final String region = regionSel.getRow().size() == 0 ? null : regionSel.lookupName(regionSel.getRow().get(0)); - out.add(Arrays.asList(tenant, region)); - cursor.advance(); - } - return out; - } - - private static CursorBuildSpec specWith(Filter filter) - { - return CursorBuildSpec.builder().setFilter(filter).build(); - } - @Test void testUnfilteredScanWalksAllGroupsAndInjectsClusteringConstants() { @@ -497,38 +734,6 @@ void testMultiGroupCanVectorizeAccountsForEveryGroupRewrite() } } - /** - * Filter wrapper whose value matcher cannot vectorize and which exposes no bitmap index, forcing the matcher - * path. The cluster-group filter walker doesn't recognize the wrapper, so it passes through rewrites unchanged. - */ - private static final class NonVectorizableFilter implements Filter - { - private final Filter delegate; - - private NonVectorizableFilter(Filter delegate) - { - this.delegate = delegate; - } - - @Override - public BitmapColumnIndex getBitmapColumnIndex(ColumnIndexSelector selector) - { - return null; - } - - @Override - public ValueMatcher makeMatcher(ColumnSelectorFactory factory) - { - return delegate.makeMatcher(factory); - } - - @Override - public Set getRequiredColumns() - { - return delegate.getRequiredColumns(); - } - } - @Test void testAllGroupsPrunedTimeseriesReturnsEmptyResult() { @@ -764,6 +969,66 @@ void testGroupByClusteringColumnWithinSingleGroup() Assertions.assertEquals(2L, ((Number) results.get(0).get(1)).longValue()); } + private static InputRow row(String tenant, String ts, String region) + { + return new MapBasedInputRow( + DateTimes.of(ts), + List.of("tenant", "region"), + Map.of("tenant", tenant, "region", region) + ); + } + + private QueryableIndex buildSegment(List rows) + { + final IncrementalIndexSchema schema = + IncrementalIndexSchema.builder() + .withMinTimestamp(INTERVAL.getStartMillis()) + .withTimestampSpec(new TimestampSpec("__time", "auto", null)) + .withQueryGranularity(Granularities.NONE) + .withDimensionsSpec(CLUSTER_SPEC.getDimensionsSpec()) + .withRollup(false) + .withClusterSpec(CLUSTER_SPEC) + .build(); + return IndexBuilder.create() + .useV10() + .tmpDir(tmpDir) + .schema(schema) + .rows(rows) + .buildMMappedIndex(INTERVAL); + } + + private QueryableIndex buildVirtualClusteringSegment() + { + final IncrementalIndexSchema schema = + IncrementalIndexSchema.builder() + .withMinTimestamp(INTERVAL.getStartMillis()) + .withTimestampSpec(new TimestampSpec("__time", "auto", null)) + .withQueryGranularity(Granularities.NONE) + .withDimensionsSpec(VIRTUAL_CLUSTER_SPEC.getDimensionsSpec()) + .withRollup(false) + .withClusterSpec(VIRTUAL_CLUSTER_SPEC) + .build(); + return IndexBuilder.create() + .useV10() + .tmpDir(tmpDir) + .schema(schema) + .rows(List.of( + row("Acme", "2025-01-01T00:00:00", "us-east-1"), + row("Acme", "2025-01-01T01:00:00", "us-west-2"), + row("Globex", "2025-01-01T00:30:00", "eu-west-1") + )) + .buildMMappedIndex(INTERVAL); + } + + private QueryableIndex standardTwoGroup() + { + return buildSegment(List.of( + row("acme", "2025-01-01T00:00:00", "us-east-1"), + row("acme", "2025-01-01T01:00:00", "us-west-2"), + row("globex", "2025-01-01T00:30:00", "eu-west-1") + )); + } + private static Druids.TimeseriesQueryBuilder newTimeseries() { return Druids.newTimeseriesQueryBuilder() @@ -772,4 +1037,89 @@ private static Druids.TimeseriesQueryBuilder newTimeseries() .intervals(List.of(Intervals.ETERNITY)) .aggregators(new CountAggregatorFactory("count")); } + + private static List collectDimension(Cursor cursor, String column) + { + final DimensionSelector sel = + cursor.getColumnSelectorFactory().makeDimensionSelector(DefaultDimensionSpec.of(column)); + final List out = new ArrayList<>(); + while (!cursor.isDone()) { + out.add(sel.getRow().size() == 0 ? null : sel.lookupName(sel.getRow().get(0))); + cursor.advance(); + } + return out; + } + + private static List collectObjectVector(VectorCursor cursor, String column) + { + final VectorObjectSelector sel = cursor.getColumnSelectorFactory().makeObjectSelector(column); + final List out = new ArrayList<>(); + while (!cursor.isDone()) { + final Object[] vector = sel.getObjectVector(); + final int size = cursor.getCurrentVectorSize(); + for (int i = 0; i < size; i++) { + out.add((String) vector[i]); + } + cursor.advance(); + } + return out; + } + + /** + * Iterate the (tenant, region) pairs out of {@code cursor} until done. Verifies that the clustering-column + * selector returns the right per-group constant and that the rewritten filter / bitmap-index path on the + * per-group QueryableIndex produces the rows it should. + */ + private static List> collectTenantRegionRows(Cursor cursor) + { + final DimensionSelector tenantSel = + cursor.getColumnSelectorFactory().makeDimensionSelector(DefaultDimensionSpec.of("tenant")); + final DimensionSelector regionSel = + cursor.getColumnSelectorFactory().makeDimensionSelector(DefaultDimensionSpec.of("region")); + final List> out = new ArrayList<>(); + while (!cursor.isDone()) { + final String tenant = tenantSel.getRow().size() == 0 ? null : tenantSel.lookupName(tenantSel.getRow().get(0)); + final String region = regionSel.getRow().size() == 0 ? null : regionSel.lookupName(regionSel.getRow().get(0)); + out.add(Arrays.asList(tenant, region)); + cursor.advance(); + } + return out; + } + + private static CursorBuildSpec specWith(Filter filter) + { + return CursorBuildSpec.builder().setFilter(filter).build(); + } + + /** + * Filter wrapper whose value matcher cannot vectorize and which exposes no bitmap index, forcing the matcher + * path. The cluster-group filter walker doesn't recognize the wrapper, so it passes through rewrites unchanged. + */ + private static final class NonVectorizableFilter implements Filter + { + private final Filter delegate; + + private NonVectorizableFilter(Filter delegate) + { + this.delegate = delegate; + } + + @Override + public BitmapColumnIndex getBitmapColumnIndex(ColumnIndexSelector selector) + { + return null; + } + + @Override + public ValueMatcher makeMatcher(ColumnSelectorFactory factory) + { + return delegate.makeMatcher(factory); + } + + @Override + public Set getRequiredColumns() + { + return delegate.getRequiredColumns(); + } + } } diff --git a/processing/src/test/java/org/apache/druid/segment/incremental/IncrementalIndexCursorFactoryClusteredTest.java b/processing/src/test/java/org/apache/druid/segment/incremental/IncrementalIndexCursorFactoryClusteredTest.java index 36b663d45051..027813f7e64a 100644 --- a/processing/src/test/java/org/apache/druid/segment/incremental/IncrementalIndexCursorFactoryClusteredTest.java +++ b/processing/src/test/java/org/apache/druid/segment/incremental/IncrementalIndexCursorFactoryClusteredTest.java @@ -30,6 +30,7 @@ import org.apache.druid.query.expression.TestExprMacroTable; import org.apache.druid.query.filter.EqualityFilter; import org.apache.druid.query.filter.Filter; +import org.apache.druid.query.filter.RangeFilter; import org.apache.druid.query.filter.TypedInFilter; import org.apache.druid.segment.Cursor; import org.apache.druid.segment.CursorBuildSpec; @@ -176,6 +177,127 @@ void testNonClusteringVirtualColumnDimensionIsMaterialized() } } + /** + * Build an index clustered on {@code tenant_lower := lower(tenant)} (a clustering column produced by a group VC) with + * a non-clustering materialized {@code region_upper := upper(region)} column. The raw inputs {@code tenant} and + * {@code region} are retained as stored columns, so the query-VC -> materialized-column remap is a pure optimization + * can be tested. Columns are {@code [tenant_lower (clustering), tenant, region, region_upper, __time]}. + */ + private static OnheapIncrementalIndex virtualClusteringIndex() + { + final ClusteredValueGroupsBaseTableProjectionSpec spec = ClusteredValueGroupsBaseTableProjectionSpec.builder() + .virtualColumns(VirtualColumns.create( + new ExpressionVirtualColumn("tenant_lower", "lower(tenant)", ColumnType.STRING, TestExprMacroTable.INSTANCE), + new ExpressionVirtualColumn("region_upper", "upper(region)", ColumnType.STRING, TestExprMacroTable.INSTANCE) + )) + .columns( + new StringDimensionSchema("tenant_lower"), + new StringDimensionSchema("tenant"), + new StringDimensionSchema("region"), + new StringDimensionSchema("region_upper"), + new LongDimensionSchema("__time") + ) + .clusteringColumns("tenant_lower") + .build(); + final IncrementalIndexSchema schema = IncrementalIndexSchema.builder() + .withMinTimestamp(T0) + .withTimestampSpec(TIMESTAMP_SPEC) + .withQueryGranularity(Granularities.NONE) + .withDimensionsSpec(spec.getDimensionsSpec()) + .withRollup(false) + .withClusterSpec(spec) + .build(); + final OnheapIncrementalIndex index = (OnheapIncrementalIndex) new OnheapIncrementalIndex.Builder() + .setIndexSchema(schema) + .setMaxRowCount(10_000) + .build(); + index.add(row(T0, "Acme", "us-east-1")); + index.add(row(T0 + 1, "Acme", "us-west-2")); + index.add(row(T0 + 2, "Globex", "eu-west-1")); + return index; + } + + @Test + void testQueryVirtualColumnEquivalentToClusteringColumnReadsMaterializedColumn() + { + // Query VC v0 := lower(tenant) is equivalent to the clustering column tenant_lower (also lower(tenant)). The remap + // substitutes the materialized tenant_lower clustering constant (instead of recomputing lower(tenant) from the + // retained raw column) so makeDimensionSelector("v0") returns the per-group value. + try (OnheapIncrementalIndex index = virtualClusteringIndex()) { + final IncrementalIndexCursorFactory factory = new IncrementalIndexCursorFactory(index); + final CursorBuildSpec buildSpec = CursorBuildSpec.builder() + .setVirtualColumns(VirtualColumns.create( + new ExpressionVirtualColumn("v0", "lower(tenant)", ColumnType.STRING, TestExprMacroTable.INSTANCE) + )) + .build(); + try (CursorHolder holder = factory.makeCursorHolder(buildSpec)) { + final Cursor cursor = holder.asCursor(); + final DimensionSelector v0Sel = + cursor.getColumnSelectorFactory().makeDimensionSelector(DefaultDimensionSpec.of("v0")); + final List out = new ArrayList<>(); + while (!cursor.isDone()) { + out.add(v0Sel.getRow().size() == 0 ? null : v0Sel.lookupName(v0Sel.getRow().get(0))); + cursor.advance(); + } + // acme group (tenant_lower=acme) first, then globex; both materialized from the clustering constant. + Assertions.assertEquals(List.of("acme", "acme", "globex"), out); + } + } + } + + @Test + void testQueryVirtualColumnEquivalentToNonClusteringMaterializedColumnReadsMaterializedColumn() + { + // Query VC v1 := upper(region) is equivalent to the non-clustering materialized column region_upper. The remap + // makes makeDimensionSelector("v1") read the per-group physical region_upper column. + try (OnheapIncrementalIndex index = virtualClusteringIndex()) { + final IncrementalIndexCursorFactory factory = new IncrementalIndexCursorFactory(index); + final CursorBuildSpec buildSpec = CursorBuildSpec.builder() + .setVirtualColumns(VirtualColumns.create( + new ExpressionVirtualColumn("v1", "upper(region)", ColumnType.STRING, TestExprMacroTable.INSTANCE) + )) + .build(); + try (CursorHolder holder = factory.makeCursorHolder(buildSpec)) { + final Cursor cursor = holder.asCursor(); + final DimensionSelector v1Sel = + cursor.getColumnSelectorFactory().makeDimensionSelector(DefaultDimensionSpec.of("v1")); + final List out = new ArrayList<>(); + while (!cursor.isDone()) { + out.add(v1Sel.getRow().size() == 0 ? null : v1Sel.lookupName(v1Sel.getRow().get(0))); + cursor.advance(); + } + Assertions.assertEquals(List.of("US-EAST-1", "US-WEST-2", "EU-WEST-1"), out); + } + } + } + + @Test + void testQueryVirtualColumnWithoutEquivalentStillRecomputes() + { + // a query VC with NO materialized equivalent (upper(region_upper) -> a different expression) is not remapped and + // is recomputed normally from the stored region_upper column. + try (OnheapIncrementalIndex index = virtualClusteringIndex()) { + final IncrementalIndexCursorFactory factory = new IncrementalIndexCursorFactory(index); + final CursorBuildSpec buildSpec = CursorBuildSpec.builder() + .setVirtualColumns(VirtualColumns.create( + new ExpressionVirtualColumn("v2", "lower(region_upper)", ColumnType.STRING, TestExprMacroTable.INSTANCE) + )) + .build(); + try (CursorHolder holder = factory.makeCursorHolder(buildSpec)) { + final Cursor cursor = holder.asCursor(); + final DimensionSelector v2Sel = + cursor.getColumnSelectorFactory().makeDimensionSelector(DefaultDimensionSpec.of("v2")); + final List out = new ArrayList<>(); + while (!cursor.isDone()) { + out.add(v2Sel.getRow().size() == 0 ? null : v2Sel.lookupName(v2Sel.getRow().get(0))); + cursor.advance(); + } + // lower(upper(region)) recomputed from the materialized region_upper column. + Assertions.assertEquals(List.of("us-east-1", "us-west-2", "eu-west-1"), out); + } + } + } + @Test void testRowSignatureExposesClusteringAndNonClusteringColumns() { @@ -232,6 +354,112 @@ void testFilterOnClusteringColumnPrunesNonMatchingGroups() } } + @Test + void testRangeFilterOnClusteringColumnResolvesViaClusteringConstant() + { + try (OnheapIncrementalIndex index = standardTwoGroup()) { + final IncrementalIndexCursorFactory factory = new IncrementalIndexCursorFactory(index); + // A RangeFilter on the clustering column is NOT folded by the pruner (only Equality/In/Null fold), so it + // survives to the per-group cursor still referencing "tenant". OnHeapClusterGroup doesn't store the clustering + // column, so the per-group selector factory must expose it as a per-group constant (mirroring the historical + // fabricated-constant path) or the filter matches a missing/all-null column and returns nothing. tenant in + // ['a','b'] selects the acme group and excludes globex. + final Filter filter = new RangeFilter("tenant", ColumnType.STRING, "a", "b", false, false, null); + try (CursorHolder holder = factory.makeCursorHolder(specWith(filter))) { + Assertions.assertEquals( + List.of( + List.of("acme", "us-east-1"), + List.of("acme", "us-west-2") + ), + scanTenantRegion(holder) + ); + } + } + } + + @Test + void testVirtualColumnReadingClusteringColumnResolvesViaClusteringConstant() + { + try (OnheapIncrementalIndex index = standardTwoGroup()) { + final IncrementalIndexCursorFactory factory = new IncrementalIndexCursorFactory(index); + // v := concat(tenant, '_x') reads the clustering column `tenant` (not stored per-row in the group). It has no + // materialized equivalent so it is retained (recomputed at read). The VC's `tenant` input must resolve to the + // group's clustering constant, not a missing column. + final CursorBuildSpec spec = CursorBuildSpec.builder() + .setVirtualColumns(VirtualColumns.create( + new ExpressionVirtualColumn("v", "concat(tenant, '_x')", ColumnType.STRING, TestExprMacroTable.INSTANCE) + )) + .build(); + try (CursorHolder holder = factory.makeCursorHolder(spec)) { + final Cursor cursor = holder.asCursor(); + final DimensionSelector sel = + cursor.getColumnSelectorFactory().makeDimensionSelector(DefaultDimensionSpec.of("v")); + final List out = new ArrayList<>(); + while (!cursor.isDone()) { + out.add(sel.getRow().size() == 0 ? null : sel.lookupName(sel.getRow().get(0))); + cursor.advance(); + } + Assertions.assertEquals(List.of("acme_x", "acme_x", "globex_x"), out); + } + } + } + + @Test + void testQueryVirtualColumnShadowingClusteringColumnWinsOverConstant() + { + try (OnheapIncrementalIndex index = standardTwoGroup()) { + final IncrementalIndexCursorFactory factory = new IncrementalIndexCursorFactory(index); + // A query VC named "tenant" shadows the clustering column of the same name (computed from region instead), and a + // retained VC "label" reads it. The superclass resolves virtual columns before physical columns, so "label" must + // see the VC's value (upper(region)), NOT the per-group clustering constant. If the clustering-constant + // interception fired first, "label" would silently read "acme"/"globex" instead. + final CursorBuildSpec spec = CursorBuildSpec.builder() + .setVirtualColumns(VirtualColumns.create( + new ExpressionVirtualColumn("tenant", "upper(region)", ColumnType.STRING, TestExprMacroTable.INSTANCE), + new ExpressionVirtualColumn("label", "concat(tenant, '_x')", ColumnType.STRING, TestExprMacroTable.INSTANCE) + )) + .build(); + try (CursorHolder holder = factory.makeCursorHolder(spec)) { + final Cursor cursor = holder.asCursor(); + final DimensionSelector sel = + cursor.getColumnSelectorFactory().makeDimensionSelector(DefaultDimensionSpec.of("label")); + final List out = new ArrayList<>(); + while (!cursor.isDone()) { + out.add(sel.getRow().size() == 0 ? null : sel.lookupName(sel.getRow().get(0))); + cursor.advance(); + } + Assertions.assertEquals(List.of("US-EAST-1_x", "US-WEST-2_x", "EU-WEST-1_x"), out); + } + } + } + + @Test + void testFilterOnVirtualColumnReadingClusteringColumnResolvesViaClusteringConstant() + { + try (OnheapIncrementalIndex index = standardTwoGroup()) { + final IncrementalIndexCursorFactory factory = new IncrementalIndexCursorFactory(index); + // A filter on a retained VC that reads the clustering column (v := upper(tenant)) must evaluate the VC against + // the group's clustering constant, not a missing column. The VC resolves its `tenant` input through its own + // selector factory, so the clustering constants must be visible below the virtual-column layer or the filter + // silently rejects every realtime row. upper(tenant)='ACME' selects the acme group. + final CursorBuildSpec spec = CursorBuildSpec.builder() + .setVirtualColumns(VirtualColumns.create( + new ExpressionVirtualColumn("v", "upper(tenant)", ColumnType.STRING, TestExprMacroTable.INSTANCE) + )) + .setFilter(new EqualityFilter("v", ColumnType.STRING, "ACME", null)) + .build(); + try (CursorHolder holder = factory.makeCursorHolder(spec)) { + Assertions.assertEquals( + List.of( + List.of("acme", "us-east-1"), + List.of("acme", "us-west-2") + ), + scanTenantRegion(holder) + ); + } + } + } + @Test void testFilterOnNonClusteringColumnRunsOnEverySurvivingGroup() { diff --git a/processing/src/test/java/org/apache/druid/segment/projections/NoRewriteFilter.java b/processing/src/test/java/org/apache/druid/segment/projections/NoRewriteFilter.java new file mode 100644 index 000000000000..f6be66cb60d9 --- /dev/null +++ b/processing/src/test/java/org/apache/druid/segment/projections/NoRewriteFilter.java @@ -0,0 +1,62 @@ +/* + * 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.segment.projections; + +import org.apache.druid.query.filter.ColumnIndexSelector; +import org.apache.druid.query.filter.Filter; +import org.apache.druid.query.filter.ValueMatcher; +import org.apache.druid.segment.ColumnSelectorFactory; +import org.apache.druid.segment.index.BitmapColumnIndex; + +import java.util.Set; + +/** + * A test {@link Filter} that does not support required-column rewriting: + * {@link Filter#supportsRequiredColumnRewrite()} is the default {@code false} and {@link Filter#rewriteRequiredColumns} + * the default throwing implementation. Its index / matcher methods are unused by the projection-planning code under + * test and throw if called. + */ +final class NoRewriteFilter implements Filter +{ + private final Set requiredColumns; + + NoRewriteFilter(String... columns) + { + this.requiredColumns = Set.of(columns); + } + + @Override + public Set getRequiredColumns() + { + return requiredColumns; + } + + @Override + public BitmapColumnIndex getBitmapColumnIndex(ColumnIndexSelector selector) + { + throw new UnsupportedOperationException(); + } + + @Override + public ValueMatcher makeMatcher(ColumnSelectorFactory factory) + { + throw new UnsupportedOperationException(); + } +} diff --git a/processing/src/test/java/org/apache/druid/segment/projections/ProjectionsPlanClusterGroupQueryTest.java b/processing/src/test/java/org/apache/druid/segment/projections/ProjectionsPlanClusterGroupQueryTest.java index b013feb333f7..d355287e4938 100644 --- a/processing/src/test/java/org/apache/druid/segment/projections/ProjectionsPlanClusterGroupQueryTest.java +++ b/processing/src/test/java/org/apache/druid/segment/projections/ProjectionsPlanClusterGroupQueryTest.java @@ -20,6 +20,7 @@ package org.apache.druid.segment.projections; import com.google.common.collect.ImmutableList; +import org.apache.druid.java.util.common.granularity.Granularities; import org.apache.druid.query.OrderBy; import org.apache.druid.query.expression.TestExprMacroTable; import org.apache.druid.query.filter.EqualityFilter; @@ -45,6 +46,8 @@ import java.util.Arrays; import java.util.LinkedHashSet; import java.util.List; +import java.util.Map; +import java.util.Set; /** * Coverage for {@link Projections#planClusterGroupQuery}, exercised through both facets of the returned @@ -184,7 +187,94 @@ private static TableClusterGroupSpec virtualClusteringGroup(String loweredTenant TestExprMacroTable.INSTANCE ) ), - List.of("tenant_lower", ColumnHolder.TIME_COLUMN_NAME, "metric"), + // raw `tenant` is retained, so lower(tenant) is a pure-optimization remap (recompute from tenant is also valid) + List.of("tenant_lower", "tenant", ColumnHolder.TIME_COLUMN_NAME, "metric"), + List.of(OrderBy.ascending("tenant_lower"), OrderBy.ascending(ColumnHolder.TIME_COLUMN_NAME)), + clustering, + null, + built.dictionaries(), + built.specs() + ); + return built.specs().get(0); + } + + /** + * Group clustered on {@code tenant_lower := lower(tenant)} (a clustering column produced by a group VC) with a + * non-clustering materialized column {@code region_upper := upper(region)}. The raw inputs {@code tenant} and + * {@code region} are retained as stored columns, so both equivalences are pure-optimization remaps (the query VC + * could also be correctly recomputed from the retained input). + */ + private static TableClusterGroupSpec materializedVcGroup(String loweredTenant) + { + final RowSignature clustering = RowSignature.builder().add("tenant_lower", ColumnType.STRING).build(); + final ClusterGroupSchemaTestHelpers.Built built = ClusterGroupSchemaTestHelpers.buildClusterGroups( + clustering, + List.of(Arrays.asList(loweredTenant)) + ); + new ClusteredValueGroupsBaseTableSchema( + VirtualColumns.create( + new ExpressionVirtualColumn("tenant_lower", "lower(tenant)", ColumnType.STRING, TestExprMacroTable.INSTANCE), + new ExpressionVirtualColumn("region_upper", "upper(region)", ColumnType.STRING, TestExprMacroTable.INSTANCE) + ), + List.of("tenant_lower", "region_upper", "tenant", "region", ColumnHolder.TIME_COLUMN_NAME, "metric"), + List.of(OrderBy.ascending("tenant_lower"), OrderBy.ascending(ColumnHolder.TIME_COLUMN_NAME)), + clustering, + null, + built.dictionaries(), + built.specs() + ); + return built.specs().get(0); + } + + /** + * Like {@link #materializedVcGroup} but also materializes a NESTED virtual column + * {@code tenant_upper_lower := upper(tenant_lower)} (derived from the clustering column), so a query VC chain + * {@code v0 := lower(tenant)} -> {@code v1 := upper(v0)} can be materialized end-to-end. + */ + private static TableClusterGroupSpec nestedMaterializedVcGroup(String loweredTenant) + { + final RowSignature clustering = RowSignature.builder().add("tenant_lower", ColumnType.STRING).build(); + final ClusterGroupSchemaTestHelpers.Built built = ClusterGroupSchemaTestHelpers.buildClusterGroups( + clustering, + List.of(Arrays.asList(loweredTenant)) + ); + new ClusteredValueGroupsBaseTableSchema( + VirtualColumns.create( + new ExpressionVirtualColumn("tenant_lower", "lower(tenant)", ColumnType.STRING, TestExprMacroTable.INSTANCE), + new ExpressionVirtualColumn("tenant_upper_lower", "upper(tenant_lower)", ColumnType.STRING, TestExprMacroTable.INSTANCE) + ), + // raw `tenant` is retained so the whole chain (lower(tenant) -> upper(...)) is recomputable, hence remappable + List.of("tenant_lower", "tenant_upper_lower", "tenant", ColumnHolder.TIME_COLUMN_NAME, "metric"), + List.of(OrderBy.ascending("tenant_lower"), OrderBy.ascending(ColumnHolder.TIME_COLUMN_NAME)), + clustering, + null, + built.dictionaries(), + built.specs() + ); + return built.specs().get(0); + } + + /** + * Like {@link #materializedVcGroup} but the summary also carries the metadata-only {@code __virtualGranularity} + * query-granularity carrier virtual column (HOUR). The carrier is deliberately NOT listed in the stored columns, so + * it must never be a remap target even though {@link VirtualColumns#findEquivalent} can match a query VC against its + * floor expression. + */ + private static TableClusterGroupSpec granularityCarrierVcGroup(String loweredTenant) + { + final RowSignature clustering = RowSignature.builder().add("tenant_lower", ColumnType.STRING).build(); + final ClusterGroupSchemaTestHelpers.Built built = ClusterGroupSchemaTestHelpers.buildClusterGroups( + clustering, + List.of(Arrays.asList(loweredTenant)) + ); + new ClusteredValueGroupsBaseTableSchema( + VirtualColumns.create( + new ExpressionVirtualColumn("tenant_lower", "lower(tenant)", ColumnType.STRING, TestExprMacroTable.INSTANCE), + Granularities.toVirtualColumn(Granularities.HOUR, Granularities.GRANULARITY_VIRTUAL_COLUMN_NAME) + ), + // __virtualGranularity is intentionally absent from the stored columns; it is a metadata carrier only. Raw + // `tenant` is retained so lower(tenant) is remappable while the granularity carrier still isn't. + List.of("tenant_lower", "tenant", ColumnHolder.TIME_COLUMN_NAME, "metric"), List.of(OrderBy.ascending("tenant_lower"), OrderBy.ascending(ColumnHolder.TIME_COLUMN_NAME)), clustering, null, @@ -900,4 +990,236 @@ void testQueryVcShadowingClusteringNameWithoutEquivalenceLeavesLeafUnchanged() Projections.planClusterGroupQuery(List.of(group), buildSpec(f, queryVcs)).rewriteFor(group) ); } + + @Test + void testVirtualColumnRemapMapsQueryVcEquivalentToClusteringColumn() + { + // Query VC v0 := lower(tenant) is equivalent to the group's clustering column tenant_lower (also lower(tenant)). + // The remap should map v0 -> tenant_lower so grouping/select reads the materialized clustering column. + final VirtualColumns queryVcs = VirtualColumns.create( + new ExpressionVirtualColumn("v0", "lower(tenant)", ColumnType.STRING, TestExprMacroTable.INSTANCE) + ); + final ClusterGroupQueryPlan plan = Projections.planClusterGroupQuery( + List.of(materializedVcGroup("acme")), + buildSpec(null, queryVcs) + ); + Assertions.assertEquals(Map.of("v0", "tenant_lower"), plan.virtualColumnRemap()); + } + + @Test + void testVirtualColumnRemapMapsQueryVcEquivalentToNonClusteringMaterializedColumn() + { + // Query VC v1 := upper(region) is equivalent to the group's non-clustering materialized column region_upper. + final VirtualColumns queryVcs = VirtualColumns.create( + new ExpressionVirtualColumn("v1", "upper(region)", ColumnType.STRING, TestExprMacroTable.INSTANCE) + ); + final ClusterGroupQueryPlan plan = Projections.planClusterGroupQuery( + List.of(materializedVcGroup("acme")), + buildSpec(null, queryVcs) + ); + Assertions.assertEquals(Map.of("v1", "region_upper"), plan.virtualColumnRemap()); + } + + @Test + void testVirtualColumnRemapExcludesTargetShadowedByAnotherQueryVc() + { + // v1 := upper(region) is equivalent to the materialized region_upper, but the query also has a VC NAMED + // region_upper (:= lower(region)). Remapping v1 -> region_upper would route through the per-group delegate, which + // resolves "region_upper" to the shadowing query VC (lowercase) instead of the stored column -- so v1 must NOT be + // remapped (it recomputes upper(region) instead). See the shadow guard in buildClusterVirtualColumnRemap. + final VirtualColumns queryVcs = VirtualColumns.create( + new ExpressionVirtualColumn("v1", "upper(region)", ColumnType.STRING, TestExprMacroTable.INSTANCE), + new ExpressionVirtualColumn("region_upper", "lower(region)", ColumnType.STRING, TestExprMacroTable.INSTANCE) + ); + final ClusterGroupQueryPlan plan = Projections.planClusterGroupQuery( + List.of(materializedVcGroup("acme")), + buildSpec(null, queryVcs) + ); + Assertions.assertTrue(plan.virtualColumnRemap().isEmpty()); + } + + @Test + void testRebuildCursorBuildSpecDeclaresRemapTargetsInPhysicalColumns() + { + // v1 := upper(region) remaps to the materialized region_upper. A populated physicalColumns set carried the VC's + // raw input (region), not the target, so the rebuilt spec must add region_upper -- otherwise partial-segment + // prefetch omits it and the remapped read hits deep storage synchronously (and incremental capability checks + // reject the undeclared column). The raw input stays too. + final VirtualColumns queryVcs = VirtualColumns.create( + new ExpressionVirtualColumn("v1", "upper(region)", ColumnType.STRING, TestExprMacroTable.INSTANCE) + ); + final TableClusterGroupSpec group = materializedVcGroup("acme"); + final CursorBuildSpec spec = CursorBuildSpec.builder() + .setVirtualColumns(queryVcs) + .setPhysicalColumns(Set.of("region")) + .build(); + final ClusterGroupQueryPlan plan = Projections.planClusterGroupQuery(List.of(group), spec); + Assertions.assertEquals(Map.of("v1", "region_upper"), plan.virtualColumnRemap()); + + final CursorBuildSpec rebuilt = plan.rebuildCursorBuildSpec(spec, group); + Assertions.assertTrue(rebuilt.getPhysicalColumns().contains("region_upper")); + Assertions.assertTrue(rebuilt.getPhysicalColumns().contains("region")); + } + + @Test + void testRebuildCursorBuildSpecLeavesNullPhysicalColumnsNull() + { + // A null physicalColumns means "all columns"; rebuild must not fabricate a (now-incomplete) set from the remap. + final VirtualColumns queryVcs = VirtualColumns.create( + new ExpressionVirtualColumn("v1", "upper(region)", ColumnType.STRING, TestExprMacroTable.INSTANCE) + ); + final TableClusterGroupSpec group = materializedVcGroup("acme"); + final CursorBuildSpec spec = buildSpec(null, queryVcs); + final ClusterGroupQueryPlan plan = Projections.planClusterGroupQuery(List.of(group), spec); + Assertions.assertFalse(plan.virtualColumnRemap().isEmpty()); + + Assertions.assertNull(plan.rebuildCursorBuildSpec(spec, group).getPhysicalColumns()); + } + + @Test + void testRebuildCursorBuildSpecLeavesUnrewritableFilterUntouchedWhenDisjointFromRemap() + { + // A filter that can't rewrite its required columns, used alongside a remappable VC it does NOT reference, must be + // left untouched: v0 := lower(tenant) still remaps to tenant_lower, but the filter (on region) does not intersect + // the remap, so rebuildCursorBuildSpec must not call rewriteRequiredColumns on it (which would throw). + final VirtualColumns queryVcs = VirtualColumns.create( + new ExpressionVirtualColumn("v0", "lower(tenant)", ColumnType.STRING, TestExprMacroTable.INSTANCE) + ); + final Filter filter = new NoRewriteFilter("region"); + final TableClusterGroupSpec group = materializedVcGroup("acme"); + final CursorBuildSpec spec = buildSpec(filter, queryVcs); + final ClusterGroupQueryPlan plan = Projections.planClusterGroupQuery(List.of(group), spec); + Assertions.assertEquals(Map.of("v0", "tenant_lower"), plan.virtualColumnRemap()); + + Assertions.assertSame(filter, plan.rebuildCursorBuildSpec(spec, group).getFilter()); + } + + @Test + void testVirtualColumnRemapExcludesRetainedInputVcReferencedByUnrewritableFilter() + { + // v0 := lower(tenant) is remappable (tenant retained), but an unrewritable filter references it directly. The + // filter can't be rewritten to the materialized column, so v0 is left unremapped and recomputes from the retained + // tenant column -- the filter reads that value rather than throwing on an unsupported rewrite. + final VirtualColumns queryVcs = VirtualColumns.create( + new ExpressionVirtualColumn("v0", "lower(tenant)", ColumnType.STRING, TestExprMacroTable.INSTANCE) + ); + final Filter filter = new NoRewriteFilter("v0"); + final TableClusterGroupSpec group = materializedVcGroup("acme"); + final CursorBuildSpec spec = buildSpec(filter, queryVcs); + final ClusterGroupQueryPlan plan = Projections.planClusterGroupQuery(List.of(group), spec); + Assertions.assertTrue(plan.virtualColumnRemap().isEmpty()); + + Assertions.assertSame(filter, plan.rebuildCursorBuildSpec(spec, group).getFilter()); + } + + @Test + void testVirtualColumnRemapExcludesMetadataOnlyGranularityCarrier() + { + // The summary carries a metadata-only __virtualGranularity VC (records the query granularity; not a stored + // column). A query VC equivalent to the carrier's floor expression must NOT be remapped to __virtualGranularity -- + // the per-group / clustering selector can't serve it -- so it stays in place to recompute from __time. A sibling + // query VC equivalent to the (stored) clustering column is still remapped, proving only the carrier is excluded. + final VirtualColumns queryVcs = VirtualColumns.create( + Granularities.toVirtualColumn(Granularities.HOUR, "q_floor"), + new ExpressionVirtualColumn("q_lower", "lower(tenant)", ColumnType.STRING, TestExprMacroTable.INSTANCE) + ); + final ClusterGroupQueryPlan plan = Projections.planClusterGroupQuery( + List.of(granularityCarrierVcGroup("acme")), + buildSpec(null, queryVcs) + ); + Assertions.assertEquals(Map.of("q_lower", "tenant_lower"), plan.virtualColumnRemap()); + } + + @Test + void testVirtualColumnRemapEmptyWhenNoEquivalent() + { + // Query VC has no materialized equivalent in the group (no length(region) column / VC) → no remap. + final VirtualColumns queryVcs = VirtualColumns.create( + new ExpressionVirtualColumn("v0", "strlen(region)", ColumnType.LONG, TestExprMacroTable.INSTANCE) + ); + final ClusterGroupQueryPlan plan = Projections.planClusterGroupQuery( + List.of(materializedVcGroup("acme")), + buildSpec(null, queryVcs) + ); + Assertions.assertTrue(plan.virtualColumnRemap().isEmpty()); + } + + @Test + void testVirtualColumnRemapEmptyWhenNoQueryVirtualColumns() + { + final ClusterGroupQueryPlan plan = Projections.planClusterGroupQuery( + List.of(materializedVcGroup("acme")), + buildSpec(null) + ); + Assertions.assertTrue(plan.virtualColumnRemap().isEmpty()); + } + + @Test + void testVirtualColumnRemapBuiltEvenWithFilter() + { + // The remap must be computed even when the query has a filter (grouping/select substitution applies regardless). + final VirtualColumns queryVcs = VirtualColumns.create( + new ExpressionVirtualColumn("v0", "lower(tenant)", ColumnType.STRING, TestExprMacroTable.INSTANCE) + ); + final Filter f = new EqualityFilter("v0", ColumnType.STRING, "acme", null); + final ClusterGroupQueryPlan plan = Projections.planClusterGroupQuery( + List.of(materializedVcGroup("acme")), + buildSpec(f, queryVcs) + ); + Assertions.assertEquals(Map.of("v0", "tenant_lower"), plan.virtualColumnRemap()); + } + + @Test + void testVirtualColumnRemapRespectsDependencyGuard() + { + // v0 := lower(tenant) (equivalent to clustering tenant_lower) AND v1 := concat(v0, 'x') which DEPENDS on v0. + // v0 must NOT be remapped/dropped because v1 still needs it as an input; v1 itself has no equivalent → no remap. + final VirtualColumns queryVcs = VirtualColumns.create( + new ExpressionVirtualColumn("v0", "lower(tenant)", ColumnType.STRING, TestExprMacroTable.INSTANCE), + new ExpressionVirtualColumn("v1", "concat(v0, 'x')", ColumnType.STRING, TestExprMacroTable.INSTANCE) + ); + final ClusterGroupQueryPlan plan = Projections.planClusterGroupQuery( + List.of(materializedVcGroup("acme")), + buildSpec(null, queryVcs) + ); + Assertions.assertTrue(plan.virtualColumnRemap().isEmpty()); + } + + @Test + void testVirtualColumnRemapSubstitutesFullyMaterializedDependencyChain() + { + // v0 := lower(tenant) (equivalent to clustering tenant_lower) AND v1 := upper(v0) which DEPENDS on v0 but is + // ALSO materialized (equivalent to tenant_upper_lower := upper(tenant_lower)). Because v1 is substituted (dropped, + // read from its materialized column) it never recomputes, so it does not force v0 to be kept -> BOTH substitute. + final VirtualColumns queryVcs = VirtualColumns.create( + new ExpressionVirtualColumn("v0", "lower(tenant)", ColumnType.STRING, TestExprMacroTable.INSTANCE), + new ExpressionVirtualColumn("v1", "upper(v0)", ColumnType.STRING, TestExprMacroTable.INSTANCE) + ); + final ClusterGroupQueryPlan plan = Projections.planClusterGroupQuery( + List.of(nestedMaterializedVcGroup("acme")), + buildSpec(null, queryVcs) + ); + Assertions.assertEquals( + Map.of("v0", "tenant_lower", "v1", "tenant_upper_lower"), + plan.virtualColumnRemap() + ); + } + + @Test + void testVirtualColumnRemapKeepsMaterializedChainNeededByNonMaterializedDependent() + { + // Transitive keep: v0 := lower(tenant) and v1 := upper(v0) are both materialized, but v2 := strlen(v1) is NOT + // (no strlen column). v2 is kept and recomputes -> needs v1 -> v1 kept -> v1 recomputes -> needs v0 -> v0 kept. + // So nothing substitutes, even though v0/v1's only direct dependents are themselves materialized. + final VirtualColumns queryVcs = VirtualColumns.create( + new ExpressionVirtualColumn("v0", "lower(tenant)", ColumnType.STRING, TestExprMacroTable.INSTANCE), + new ExpressionVirtualColumn("v1", "upper(v0)", ColumnType.STRING, TestExprMacroTable.INSTANCE), + new ExpressionVirtualColumn("v2", "strlen(v1)", ColumnType.LONG, TestExprMacroTable.INSTANCE) + ); + final ClusterGroupQueryPlan plan = Projections.planClusterGroupQuery( + List.of(nestedMaterializedVcGroup("acme")), + buildSpec(null, queryVcs) + ); + Assertions.assertTrue(plan.virtualColumnRemap().isEmpty()); + } } diff --git a/processing/src/test/java/org/apache/druid/segment/projections/ProjectionsTest.java b/processing/src/test/java/org/apache/druid/segment/projections/ProjectionsTest.java index 01b391001b77..c4caa1ed858a 100644 --- a/processing/src/test/java/org/apache/druid/segment/projections/ProjectionsTest.java +++ b/processing/src/test/java/org/apache/druid/segment/projections/ProjectionsTest.java @@ -262,6 +262,52 @@ void testSchemaMatchFilter() Assertions.assertEquals(expected, projectionMatch); } + @Test + void testSchemaFilterRejectedWhenQueryFilterCannotBeRewritten() + { + // The query VC v0 := upper(b) is equivalent to the projection's b_upper, so matchQueryVirtualColumns remaps + // v0 -> b_upper. The query's filter references v0 but can't rewrite its required columns, so it can't be remapped + // into the projection's column namespace: the match must be rejected (fall back to the base table) rather than + // throwing from rewriteRequiredColumns. + RowSignature baseTable = RowSignature.builder() + .addTimeColumn() + .add("a", ColumnType.LONG) + .add("b", ColumnType.STRING) + .add("c", ColumnType.LONG) + .build(); + AggregateProjectionMetadata spec = new AggregateProjectionMetadata( + AggregateProjectionSpec.builder("some_projection") + .filter(new EqualityFilter("b", ColumnType.STRING, "foo", null)) + .virtualColumns( + new ExpressionVirtualColumn("b_upper", "upper(b)", ColumnType.STRING, TestExprMacroTable.INSTANCE) + ) + .groupingColumns(new StringDimensionSchema("b_upper"), new LongDimensionSchema("a")) + .aggregators(new LongSumAggregatorFactory("c_sum", "c")) + .build() + .toMetadataSchema(), + 12345 + ); + CursorBuildSpec query = CursorBuildSpec.builder() + .setVirtualColumns( + VirtualColumns.create( + new ExpressionVirtualColumn("v0", "upper(b)", ColumnType.STRING, TestExprMacroTable.INSTANCE) + ) + ) + .setFilter(new NoRewriteFilter("v0")) + .setPhysicalColumns(Set.of("b", "c")) + .setPreferredOrdering(List.of()) + .build(); + + Assertions.assertNull( + Projections.matchAggregateProjection( + spec.getSchema(), + query, + Intervals.ETERNITY, + new RowSignatureChecker(baseTable) + ) + ); + } + @Test void testSchemaMatchFilterIncludedInProjection() { diff --git a/processing/src/test/java/org/apache/druid/segment/vector/ConcatenatingVectorCursorTest.java b/processing/src/test/java/org/apache/druid/segment/vector/ConcatenatingVectorCursorTest.java index 5a9ebf1ba2c2..80fa379a849c 100644 --- a/processing/src/test/java/org/apache/druid/segment/vector/ConcatenatingVectorCursorTest.java +++ b/processing/src/test/java/org/apache/druid/segment/vector/ConcatenatingVectorCursorTest.java @@ -36,6 +36,7 @@ import javax.annotation.Nullable; import java.util.ArrayList; import java.util.List; +import java.util.Map; class ConcatenatingVectorCursorTest { @@ -58,7 +59,8 @@ void testWalksTwoNonEmptyGroupsBackToBack() ConcatenatingVectorCursor c = new ConcatenatingVectorCursor( List.of(holderSupplier(a), holderSupplier(b)), List.of(new Object[]{"acme"}, new Object[]{"globex"}), - wrapper + wrapper, + Map.of() ); VectorObjectSelector tenant = c.getColumnSelectorFactory().makeObjectSelector("tenant"); @@ -105,7 +107,8 @@ void testSkipsLeadingEmptyGroup() ConcatenatingVectorCursor c = new ConcatenatingVectorCursor( List.of(holderSupplier(empty), holderSupplier(full)), List.of(new Object[]{"a"}, new Object[]{"b"}), - wrapper + wrapper, + Map.of() ); VectorObjectSelector tenant = c.getColumnSelectorFactory().makeObjectSelector("tenant"); @@ -133,7 +136,8 @@ void testSkipsTrailingEmptyGroup() ConcatenatingVectorCursor c = new ConcatenatingVectorCursor( List.of(holderSupplier(full), holderSupplier(empty)), List.of(new Object[]{"a"}, new Object[]{"b"}), - wrapper + wrapper, + Map.of() ); VectorObjectSelector tenant = c.getColumnSelectorFactory().makeObjectSelector("tenant"); @@ -158,7 +162,8 @@ void testAllEmptyGroups() ConcatenatingVectorCursor c = new ConcatenatingVectorCursor( List.of(holderSupplier(e1), holderSupplier(e2)), List.of(new Object[]{"a"}, new Object[]{"b"}), - wrapper + wrapper, + Map.of() ); Assertions.assertTrue(c.isDone()); @@ -179,7 +184,8 @@ void testSingleGroupDegenerateCase() ConcatenatingVectorCursor c = new ConcatenatingVectorCursor( List.of(holderSupplier(only)), List.of(new Object[]{"a"}), - wrapper + wrapper, + Map.of() ); VectorObjectSelector tenant = c.getColumnSelectorFactory().makeObjectSelector("tenant"); @@ -210,7 +216,8 @@ void testPartialVectorAtGroupBoundary() ConcatenatingVectorCursor c = new ConcatenatingVectorCursor( List.of(holderSupplier(a), holderSupplier(b)), List.of(new Object[]{"acme"}, new Object[]{"globex"}), - wrapper + wrapper, + Map.of() ); VectorObjectSelector tenant = c.getColumnSelectorFactory().makeObjectSelector("tenant"); @@ -245,7 +252,8 @@ void testCloserClosesAllOpenedHolders() ConcatenatingVectorCursor c = new ConcatenatingVectorCursor( List.of(holderSupplier(a), holderSupplier(b)), List.of(new Object[]{"x"}, new Object[]{"y"}), - wrapper + wrapper, + Map.of() ); c.getColumnSelectorFactory(); @@ -279,7 +287,8 @@ void testGroupsAreOpenedLazilyOnTransitionNotEagerly() ConcatenatingVectorCursor c = new ConcatenatingVectorCursor( suppliers, List.of(new Object[]{"a"}, new Object[]{"b"}), - wrapper + wrapper, + Map.of() ); c.getColumnSelectorFactory(); @@ -307,7 +316,8 @@ void testMaxVectorSizeIsConfiguredValueAcrossAllStates() ConcatenatingVectorCursor c = new ConcatenatingVectorCursor( List.of(holderSupplier(a)), List.of(new Object[]{"acme"}), - wrapper + wrapper, + Map.of() ); // Pre-init. diff --git a/server/src/test/java/org/apache/druid/catalog/model/ClusteredValueGroupsBaseTableMetadataTest.java b/server/src/test/java/org/apache/druid/catalog/model/ClusteredValueGroupsBaseTableMetadataTest.java index 2d99f03a8e2a..af3a956970af 100644 --- a/server/src/test/java/org/apache/druid/catalog/model/ClusteredValueGroupsBaseTableMetadataTest.java +++ b/server/src/test/java/org/apache/druid/catalog/model/ClusteredValueGroupsBaseTableMetadataTest.java @@ -35,6 +35,7 @@ import org.apache.druid.segment.VirtualColumns; import org.apache.druid.segment.column.ColumnType; import org.apache.druid.segment.virtual.ExpressionVirtualColumn; +import org.apache.druid.testing.InitializedNullHandlingTest; import org.junit.Assert; import org.junit.Test; @@ -43,7 +44,7 @@ import java.util.List; import java.util.Map; -public class ClusteredValueGroupsBaseTableMetadataTest +public class ClusteredValueGroupsBaseTableMetadataTest extends InitializedNullHandlingTest { private final ObjectMapper mapper = new DefaultObjectMapper().setInjectableValues( new InjectableValues.Std().addValue(ExprMacroTable.class, ExprMacroTable.nil())