You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@druid.apache.org by cw...@apache.org on 2026/07/25 11:05:09 UTC
(druid) branch master updated: feat: restrict virtual column usage in clustered segments, allow remapping them at query time (#19659)
This is an automated email from the ASF dual-hosted git repository.
clintropolis pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/druid.git
The following commit(s) were added to refs/heads/master by this push:
new 13f2f583e2c feat: restrict virtual column usage in clustered segments, allow remapping them at query time (#19659)
13f2f583e2c is described below
commit 13f2f583e2cd53b83044a22fda149d39bf96921a
Author: Clint Wylie <cw...@apache.org>
AuthorDate: Sat Jul 25 04:05:02 2026 -0700
feat: restrict virtual column usage in clustered segments, allow remapping them at query time (#19659)
---
...lusteredValueGroupsBaseTableProjectionSpec.java | 102 +++-
.../apache/druid/segment/ConcatenatingCursor.java | 10 +-
.../druid/segment/MergingClusterGroupCursor.java | 24 +-
.../druid/segment/QueryableIndexCursorFactory.java | 67 ++-
...AwareIncrementalIndexColumnSelectorFactory.java | 112 +++++
.../incremental/IncrementalIndexCursorFactory.java | 40 +-
.../segment/projections/ClusterGroupQueryPlan.java | 101 +++-
.../ClusteringColumnSelectorFactory.java | 46 +-
.../ClusteringVectorColumnSelectorFactory.java | 43 +-
.../druid/segment/projections/Projections.java | 166 ++++++-
.../segment/vector/ConcatenatingVectorCursor.java | 10 +-
...eredValueGroupsBaseTableProjectionSpecTest.java | 217 +++++++++
.../ClusteredSegmentTimeOrderedQueryTest.java | 103 ++++
.../druid/segment/ConcatenatingCursorTest.java | 25 +-
.../segment/MergingClusterGroupCursorTest.java | 34 +-
.../QueryableIndexCursorFactoryClusteredTest.java | 524 +++++++++++++++++----
...IncrementalIndexCursorFactoryClusteredTest.java | 228 +++++++++
.../druid/segment/projections/NoRewriteFilter.java | 62 +++
.../ProjectionsPlanClusterGroupQueryTest.java | 324 ++++++++++++-
.../druid/segment/projections/ProjectionsTest.java | 46 ++
.../vector/ConcatenatingVectorCursorTest.java | 28 +-
.../ClusteredValueGroupsBaseTableMetadataTest.java | 3 +-
22 files changed, 2129 insertions(+), 186 deletions(-)
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 9685ab03c07..401816421dc 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.OrderBy;
import org.apache.druid.query.aggregation.AggregatorFactory;
import org.apache.druid.segment.VirtualColumn;
import org.apache.druid.segment.VirtualColumns;
+import org.apache.druid.segment.column.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 javax.annotation.Nullable;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
+import java.util.HashSet;
import java.util.List;
import java.util.Objects;
import java.util.Set;
@@ -48,9 +52,10 @@ import java.util.Set;
/**
* {@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.
* <p>
- * 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 final class ClusteredValueGroupsBaseTableProjectionSpec implements BaseTa
{
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 @@ public final class ClusteredValueGroupsBaseTableProjectionSpec implements BaseTa
}
}
+ /**
+ * Rules to keep virtual columns definitions reasonable:
+ * <ul>
+ * <li><b>dot notation</b>: a virtual column that {@link VirtualColumn#usesDotNotation()} is referenced as
+ * {@code name.subfield} rather than by a fixed output name, so it has no stable identity to materialize or cluster
+ * on; it is rejected outright.</li>
+ * <li><b>inputs</b>: every input of a virtual column must be a stored column (declared in {@code columns}) or
+ * another virtual column in the spec.</li>
+ * <li><b>outputs</b>: every virtual column must either be materialized (its output declared in {@code columns}) or
+ * be an intermediary that feeds another virtual column. A virtual column that is neither materializes nothing and
+ * is used by nothing and dead metadata and so it is rejected.</li>
+ * <li><b>output type</b>: a materialized virtual column must produce the {@link ColumnType} declared for its column
+ * in {@code columns}. This is intentionally strict to avoid a virtual column silently filling a column with a
+ * coerced/mismatched type.</li>
+ * </ul>
+ * 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<DimensionSchema> 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<String> 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<DimensionSchema> columns,
List<String> 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 1208c3df202..d23069c3137 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 org.apache.druid.segment.projections.ClusteringColumnSelectorFactory;
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<Supplier<CursorHolder>> holderSuppliers;
private final List<Object[]> 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<Supplier<CursorHolder>> holderSuppliers,
List<Object[]> clusteringValuesByGroup,
- ClusteringColumnSelectorFactory wrapperFactory
+ ClusteringColumnSelectorFactory wrapperFactory,
+ Map<String, String> virtualColumnRemap
)
{
if (holderSuppliers.size() != clusteringValuesByGroup.size()) {
@@ -65,6 +68,9 @@ public final class ConcatenatingCursor implements Cursor
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 @@ public final class ConcatenatingCursor implements Cursor
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 a3a22fe55e3..33404f95958 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.column.ColumnHolder;
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 @@ import java.util.List;
* <em>all</em> 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.
* <p>
* 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<Supplier<CursorHolder>> holderSuppliers;
private final boolean descending;
+ private final Map<String, String> 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<Supplier<CursorHolder>> holderSuppliers,
- boolean descending
+ boolean descending,
+ Map<String, String> virtualColumnRemap
)
{
if (holderSuppliers.isEmpty()) {
@@ -74,6 +83,7 @@ public final class MergingClusterGroupCursor implements Cursor
}
this.holderSuppliers = holderSuppliers;
this.descending = descending;
+ this.virtualColumnRemap = virtualColumnRemap;
}
private void initializeIfNeeded()
@@ -106,14 +116,18 @@ public final class MergingClusterGroupCursor implements Cursor
}
}
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 ac0e843260a..1206000e6cc 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.projections.TableClusterGroupSpec;
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.ArrayList;
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 @@ public class QueryableIndexCursorFactory implements ResidentCursorFactory
final List<OrderBy> 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 @@ public class QueryableIndexCursorFactory implements ResidentCursorFactory
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 class QueryableIndexCursorFactory implements ResidentCursorFactory
*/
private static CursorHolder makeConcatenatedClusteredCursorHolder(
CursorBuildSpec spec,
+ ClusterGroupQueryPlan plan,
ColumnInspector inspector,
RowSignature clusteringColumns,
List<Object[]> clusteringValuesByGroup,
@@ -383,24 +421,28 @@ public class QueryableIndexCursorFactory implements ResidentCursorFactory
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 @@ public class QueryableIndexCursorFactory implements ResidentCursorFactory
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 class QueryableIndexCursorFactory implements ResidentCursorFactory
/**
* 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<Supplier<CursorHolder>> holderSuppliers,
Closer closer,
- Order timeOrder
+ Order timeOrder,
+ Map<String, String> virtualColumnRemap
)
{
final boolean descending = timeOrder == Order.DESCENDING;
- final MergingClusterGroupCursor cursor = new MergingClusterGroupCursor(holderSuppliers, descending);
+ final MergingClusterGroupCursor cursor = new MergingClusterGroupCursor(holderSuppliers, descending, virtualColumnRemap);
final List<OrderBy> 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 00000000000..2703e720ea6
--- /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.
+ * <p>
+ * 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 07cebffa2c9..74f2ddaa014 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 @@ public class IncrementalIndexCursorFactory implements ResidentCursorFactory
final List<Supplier<CursorHolder>> 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 @@ public class IncrementalIndexCursorFactory implements ResidentCursorFactory
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 6f87c1984df..e145433f265 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 @@ package org.apache.druid.segment.projections;
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<TableClusterGroupSpec> survivingGroups;
private final Function<TableClusterGroupSpec, Filter> rewriter;
+ private final Map<String, String> virtualColumnRemap;
ClusterGroupQueryPlan(
List<TableClusterGroupSpec> survivingGroups,
- Function<TableClusterGroupSpec, Filter> rewriter
+ Function<TableClusterGroupSpec, Filter> rewriter,
+ Map<String, String> virtualColumnRemap
)
{
this.survivingGroups = survivingGroups;
this.rewriter = rewriter;
+ this.virtualColumnRemap = virtualColumnRemap;
}
/**
@@ -59,10 +69,21 @@ public final class ClusterGroupQueryPlan
}
/**
- * 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<String, String> 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.
* <p>
* 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 final class ClusterGroupQueryPlan
/**
* 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.
+ * <p>
+ * 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<VirtualColumn> 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<String> physicalColumns = spec.getPhysicalColumns();
+ if (physicalColumns != null) {
+ final Set<String> 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 ae719a96fb2..b224bd3b2e3 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.DimensionDictionarySelector;
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 @@ import java.util.function.Supplier;
* 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.
+ * <p>
+ * 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 class ClusteringColumnSelectorFactory implements ColumnSelectorFactory
};
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 class ClusteringColumnSelectorFactory implements ColumnSelectorFactory
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 @@ public class ClusteringColumnSelectorFactory implements ColumnSelectorFactory
@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 class ClusteringColumnSelectorFactory implements ColumnSelectorFactory
@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 class ClusteringColumnSelectorFactory implements ColumnSelectorFactory
.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 e82c0a76722..4ff67e9d617 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 @@ package org.apache.druid.segment.projections;
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 @@ import javax.annotation.Nullable;
* 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 @@ import javax.annotation.Nullable;
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 class ClusteringVectorColumnSelectorFactory implements VectorColumnSelect
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 class ClusteringVectorColumnSelectorFactory implements VectorColumnSelect
@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 class ClusteringVectorColumnSelectorFactory implements VectorColumnSelect
.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 54c214f4b37..ef211ae6e24 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.DateTime;
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 class Projections
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 class Projections
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 class Projections
{
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 class Projections
final RowSignature clusteringColumns = summary.getClusteringColumns();
final VirtualColumns groupVcs = summary.getVirtualColumns();
+ final Set<String> materializedColumns = new HashSet<>(summary.getColumns());
+ final Map<String, String> 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 class Projections
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<String, String> buildClusterVirtualColumnRemap(
+ VirtualColumns queryVcs,
+ VirtualColumns groupVcs,
+ Set<String> 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<String> unrewritableFilterColumns =
+ queryFilter != null && !queryFilter.supportsRequiredColumnRewrite()
+ ? queryFilter.getRequiredColumns()
+ : Set.of();
+ // Candidate substitutions: query VCs that have a differently-named equivalent materialized (stored) column.
+ final Map<String, String> candidates = new HashMap<>();
+ final Map<String, VirtualColumn> 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<String> keep = new HashSet<>();
+ final Deque<String> 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<String, String> remap = new HashMap<>();
+ for (Map.Entry<String, String> 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.
* <p/>
* 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 @@ public class Projections
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}).
+ * <p>
+ * 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<String, String> remap)
+ {
+ final Set<String> requiredColumns = filter.getRequiredColumns();
+ if (Collections.disjoint(requiredColumns, remap.keySet())) {
+ return filter;
+ }
final Map<String, String> 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 c2cfe44efd7..73328d4d21b 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.CursorHolder;
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<Supplier<CursorHolder>> holderSuppliers;
private final List<Object[]> 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<Supplier<CursorHolder>> holderSuppliers,
List<Object[]> clusteringValuesByGroup,
- ClusteringVectorColumnSelectorFactory wrapperFactory
+ ClusteringVectorColumnSelectorFactory wrapperFactory,
+ Map<String, String> virtualColumnRemap
)
{
if (holderSuppliers.size() != clusteringValuesByGroup.size()) {
@@ -69,6 +72,9 @@ public final class ConcatenatingVectorCursor implements VectorCursor
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 @@ public final class ConcatenatingVectorCursor implements VectorCursor
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 f6edc088dfd..603919e7112 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 @@ package org.apache.druid.data.input.impl;
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 @@ class ClusteredValueGroupsBaseTableProjectionSpecTest extends InitializedNullHan
)
);
}
+
+ @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<String> 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 e3eb0b0778d..b067b4ccc94 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.data.input.impl.TimestampSpec;
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.QueryPlus;
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.TimeseriesQueryQueryToolChest;
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 @@ class ClusteredSegmentTimeOrderedQueryTest extends InitializedNullHandlingTest
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<List<Object>> 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<List<Long>> runTimeseries(Segment segment)
{
final TimeseriesQuery query = Druids.newTimeseriesQueryBuilder()
@@ -254,6 +280,83 @@ class ClusteredSegmentTimeOrderedQueryTest extends InitializedNullHandlingTest
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<List<Object>> 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<ScanResultValue> results = factory.createRunner(segment).run(QueryPlus.wrap(query)).toList();
+ final List<List<Object>> rows = new ArrayList<>();
+ for (ScanResultValue result : results) {
+ for (Object event : (List<?>) result.getEvents()) {
+ @SuppressWarnings("unchecked")
+ final Map<String, Object> row = (Map<String, Object>) 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<InputRow> 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 36d2f8c547f..29a6cc54c26 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 org.junit.jupiter.api.Test;
import javax.annotation.Nullable;
import java.util.ArrayList;
import java.util.List;
+import java.util.Map;
class ConcatenatingCursorTest
{
@@ -56,7 +57,8 @@ class ConcatenatingCursorTest
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 @@ class ConcatenatingCursorTest
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 @@ class ConcatenatingCursorTest
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 @@ class ConcatenatingCursorTest
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 @@ class ConcatenatingCursorTest
ConcatenatingCursor c = new ConcatenatingCursor(
List.of(holderSupplier(only)),
List.<Object[]>of(new Object[]{"a"}),
- wrapper
+ wrapper,
+ Map.of()
);
ColumnValueSelector tenant = c.getColumnSelectorFactory().makeColumnValueSelector("tenant");
@@ -202,7 +208,8 @@ class ConcatenatingCursorTest
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 @@ class ConcatenatingCursorTest
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 @@ class ConcatenatingCursorTest
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 daf38d3a2cc..9c3f3b187f1 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 org.junit.jupiter.api.Test;
import javax.annotation.Nullable;
import java.util.ArrayList;
import java.util.List;
+import java.util.Map;
class MergingClusterGroupCursorTest
{
@@ -153,10 +154,36 @@ class MergingClusterGroupCursorTest
}
}
+ @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<Object[]> 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<CursorHolder>... groups)
{
- return new MergingClusterGroupCursor(new ArrayList<>(List.of(groups)), descending);
+ return new MergingClusterGroupCursor(new ArrayList<>(List.of(groups)), descending, Map.of());
}
private static List<Object[]> drain(MergingClusterGroupCursor cursor)
@@ -237,6 +264,11 @@ class MergingClusterGroupCursorTest
}
};
}
+ 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<String>()
{
@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 e89123ac9f9..05cd220ec8d 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.QueryContexts;
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.AndFilter;
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 @@ import java.util.Set;
* 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 @@ class QueryableIndexCursorFactoryClusteredTest
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<InputRow> 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 @@ class QueryableIndexCursorFactoryClusteredTest
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<List<String>> collectTenantRegionRows(Cursor cursor)
- {
- final DimensionSelector tenantSel =
- cursor.getColumnSelectorFactory().makeDimensionSelector(DefaultDimensionSpec.of("tenant"));
- final DimensionSelector regionSel =
- cursor.getColumnSelectorFactory().makeDimensionSelector(DefaultDimensionSpec.of("region"));
- final List<List<String>> 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 @@ class QueryableIndexCursorFactoryClusteredTest
}
}
- /**
- * 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<String> getRequiredColumns()
- {
- return delegate.getRequiredColumns();
- }
- }
-
@Test
void testAllGroupsPrunedTimeseriesReturnsEmptyResult()
{
@@ -764,6 +969,66 @@ class QueryableIndexCursorFactoryClusteredTest
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<InputRow> 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 @@ class QueryableIndexCursorFactoryClusteredTest
.intervals(List.of(Intervals.ETERNITY))
.aggregators(new CountAggregatorFactory("count"));
}
+
+ private static List<String> collectDimension(Cursor cursor, String column)
+ {
+ final DimensionSelector sel =
+ cursor.getColumnSelectorFactory().makeDimensionSelector(DefaultDimensionSpec.of(column));
+ final List<String> 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<String> collectObjectVector(VectorCursor cursor, String column)
+ {
+ final VectorObjectSelector sel = cursor.getColumnSelectorFactory().makeObjectSelector(column);
+ final List<String> 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<List<String>> collectTenantRegionRows(Cursor cursor)
+ {
+ final DimensionSelector tenantSel =
+ cursor.getColumnSelectorFactory().makeDimensionSelector(DefaultDimensionSpec.of("tenant"));
+ final DimensionSelector regionSel =
+ cursor.getColumnSelectorFactory().makeDimensionSelector(DefaultDimensionSpec.of("region"));
+ final List<List<String>> 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<String> 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 36b663d4505..027813f7e64 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.dimension.DefaultDimensionSpec;
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 @@ class IncrementalIndexCursorFactoryClusteredTest extends InitializedNullHandling
}
}
+ /**
+ * 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<String> 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<String> 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<String> 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 @@ class IncrementalIndexCursorFactoryClusteredTest extends InitializedNullHandling
}
}
+ @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<String> 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<String> 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 00000000000..f6be66cb60d
--- /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<String> requiredColumns;
+
+ NoRewriteFilter(String... columns)
+ {
+ this.requiredColumns = Set.of(columns);
+ }
+
+ @Override
+ public Set<String> 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 b013feb333f..d355287e493 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 javax.annotation.Nullable;
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 @@ class ProjectionsPlanClusterGroupQueryTest
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 @@ class ProjectionsPlanClusterGroupQueryTest
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 01b391001b7..c4caa1ed858 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 @@ class ProjectionsTest
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 5a9ebf1ba2c..80fa379a849 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 org.junit.jupiter.api.Test;
import javax.annotation.Nullable;
import java.util.ArrayList;
import java.util.List;
+import java.util.Map;
class ConcatenatingVectorCursorTest
{
@@ -58,7 +59,8 @@ class ConcatenatingVectorCursorTest
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 @@ class ConcatenatingVectorCursorTest
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 @@ class ConcatenatingVectorCursorTest
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 @@ class ConcatenatingVectorCursorTest
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 @@ class ConcatenatingVectorCursorTest
ConcatenatingVectorCursor c = new ConcatenatingVectorCursor(
List.of(holderSupplier(only)),
List.<Object[]>of(new Object[]{"a"}),
- wrapper
+ wrapper,
+ Map.of()
);
VectorObjectSelector tenant = c.getColumnSelectorFactory().makeObjectSelector("tenant");
@@ -210,7 +216,8 @@ class ConcatenatingVectorCursorTest
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 @@ class ConcatenatingVectorCursorTest
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 @@ class ConcatenatingVectorCursorTest
ConcatenatingVectorCursor c = new ConcatenatingVectorCursor(
suppliers,
List.of(new Object[]{"a"}, new Object[]{"b"}),
- wrapper
+ wrapper,
+ Map.of()
);
c.getColumnSelectorFactory();
@@ -307,7 +316,8 @@ class ConcatenatingVectorCursorTest
ConcatenatingVectorCursor c = new ConcatenatingVectorCursor(
List.of(holderSupplier(a)),
List.<Object[]>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 2d99f03a8e2..af3a956970a 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.NestedDataColumnSchema;
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.Collections;
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())
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org