This is an automated email from the ASF dual-hosted git repository.
capistrant 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 957121248f3 feat: clustered segment time ordering for realtime queries
(#19753)
957121248f3 is described below
commit 957121248f31e957d9f261215e99548558d50ebd
Author: Clint Wylie <[email protected]>
AuthorDate: Sun Jul 26 16:28:44 2026 -0700
feat: clustered segment time ordering for realtime queries (#19753)
---
.../druid/segment/MergingClusterGroupCursor.java | 82 +++++++++++-
.../druid/segment/QueryableIndexCursorFactory.java | 83 ++-----------
.../incremental/IncrementalIndexCursorFactory.java | 27 ++--
.../projections/MergingColumnSelectorFactory.java | 137 +++++++++++++++++++--
.../druid/segment/projections/Projections.java | 15 +++
.../ClusteredSegmentTimeOrderedQueryTest.java | 134 ++++++++++++++++++++
.../segment/MergingClusterGroupCursorTest.java | 25 +++-
.../MergingColumnSelectorFactoryTest.java | 106 +++++++++++++++-
8 files changed, 509 insertions(+), 100 deletions(-)
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 33404f95958..16786a12cc3 100644
---
a/processing/src/main/java/org/apache/druid/segment/MergingClusterGroupCursor.java
+++
b/processing/src/main/java/org/apache/druid/segment/MergingClusterGroupCursor.java
@@ -22,8 +22,12 @@ package org.apache.druid.segment;
import com.google.common.base.Supplier;
import it.unimi.dsi.fastutil.ints.IntHeapPriorityQueue;
import org.apache.druid.error.DruidException;
+import org.apache.druid.java.util.common.io.Closer;
+import org.apache.druid.query.OrderBy;
import org.apache.druid.segment.column.ColumnHolder;
+import org.apache.druid.segment.column.RowSignature;
import org.apache.druid.segment.projections.MergingColumnSelectorFactory;
+import org.apache.druid.utils.CloseableUtils;
import java.util.List;
import java.util.Map;
@@ -32,14 +36,14 @@ import java.util.Map;
* {@link Cursor} that presents a set of individually {@code __time}-sorted
per-cluster-group cursors as a single
* globally {@code __time}-ordered cursor, via a streaming k-way merge. Used
for clustered base tables where
* {@code __time} is the first non-clustering column (so each group, whose
clustering prefix is constant, is sorted on
- * {@code __time}); built by {@code
QueryableIndexCursorFactory#makeTimeMergedClusteredCursorHolder}.
+ * {@code __time}).
* <p>
* This is the time-ordered sibling of {@link ConcatenatingCursor}: where the
concatenating cursor walks groups
* back-to-back (order = {@code [clustering…, __time, …]}), this cursor
interleaves them by {@code __time}. It opens
* <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. When the plan carries a
+ * {@code __time} across the groups. The {@link MergingColumnSelectorFactory}
exposes each row's clustering values
+ * (constant per group) and dispatches non-clustering columns 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
@@ -54,7 +58,10 @@ import java.util.Map;
public final class MergingClusterGroupCursor implements Cursor
{
private final List<Supplier<CursorHolder>> holderSuppliers;
+ private final RowSignature clusteringColumns;
+ private final List<Object[]> clusteringValuesByGroup;
private final boolean descending;
+ private final VirtualColumns queryVirtualColumns;
private final Map<String, String> virtualColumnRemap;
private boolean initialized;
@@ -74,18 +81,77 @@ public final class MergingClusterGroupCursor implements
Cursor
public MergingClusterGroupCursor(
List<Supplier<CursorHolder>> holderSuppliers,
+ RowSignature clusteringColumns,
+ List<Object[]> clusteringValuesByGroup,
boolean descending,
+ VirtualColumns queryVirtualColumns,
Map<String, String> virtualColumnRemap
)
{
if (holderSuppliers.isEmpty()) {
throw DruidException.defensive("MergingClusterGroupCursor requires at
least one cluster group");
}
+ if (holderSuppliers.size() != clusteringValuesByGroup.size()) {
+ throw DruidException.defensive(
+ "holderSuppliers size [%s] must equal clusteringValuesByGroup size
[%s]",
+ holderSuppliers.size(),
+ clusteringValuesByGroup.size()
+ );
+ }
this.holderSuppliers = holderSuppliers;
+ this.clusteringColumns = clusteringColumns;
+ this.clusteringValuesByGroup = clusteringValuesByGroup;
this.descending = descending;
+ this.queryVirtualColumns = queryVirtualColumns;
this.virtualColumnRemap = virtualColumnRemap;
}
+ /**
+ * Build a {@link CursorHolder} over a globally {@code __time}-ordered merge
of the per-group holders, advertising
+ * ascending/descending {@code __time} ordering per {@code timeOrder}.The
supplied {@link Closer} owns the per-group
+ * holders and is closed with the returned holder.
+ */
+ public static CursorHolder makeCursorHolder(
+ List<Supplier<CursorHolder>> holderSuppliers,
+ RowSignature clusteringColumns,
+ List<Object[]> clusteringValuesByGroup,
+ boolean descending,
+ VirtualColumns queryVirtualColumns,
+ Map<String, String> virtualColumnRemap,
+ Closer closer
+ )
+ {
+ final MergingClusterGroupCursor cursor = new MergingClusterGroupCursor(
+ holderSuppliers,
+ clusteringColumns,
+ clusteringValuesByGroup,
+ descending,
+ queryVirtualColumns,
+ virtualColumnRemap
+ );
+ final List<OrderBy> ordering = descending ? Cursors.descendingTimeOrder()
: Cursors.ascendingTimeOrder();
+ return new CursorHolder()
+ {
+ @Override
+ public Cursor asCursor()
+ {
+ return cursor;
+ }
+
+ @Override
+ public List<OrderBy> getOrdering()
+ {
+ return ordering;
+ }
+
+ @Override
+ public void close()
+ {
+ CloseableUtils.closeAndWrapExceptions(closer);
+ }
+ };
+ }
+
private void initializeIfNeeded()
{
if (initialized) {
@@ -116,8 +182,14 @@ public final class MergingClusterGroupCursor implements
Cursor
}
}
currentGroup = heap.isEmpty() ? 0 : heap.firstInt();
- final MergingColumnSelectorFactory mergingFactory =
- new MergingColumnSelectorFactory(groupFactories, () -> currentGroup,
() -> outputRowId);
+ final MergingColumnSelectorFactory mergingFactory = new
MergingColumnSelectorFactory(
+ groupFactories,
+ clusteringColumns,
+ clusteringValuesByGroup,
+ queryVirtualColumns,
+ () -> currentGroup,
+ () -> outputRowId
+ );
exposedFactory = virtualColumnRemap.isEmpty()
? mergingFactory
: new RemapColumnSelectorFactory(mergingFactory,
virtualColumnRemap);
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 1206000e6cc..96cc9d6fb6b 100644
---
a/processing/src/main/java/org/apache/druid/segment/QueryableIndexCursorFactory.java
+++
b/processing/src/main/java/org/apache/druid/segment/QueryableIndexCursorFactory.java
@@ -53,12 +53,10 @@ import org.apache.druid.segment.vector.VectorValueSelector;
import org.apache.druid.utils.CloseableUtils;
import javax.annotation.Nullable;
-import java.io.IOException;
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,7 +200,7 @@ public class QueryableIndexCursorFactory implements
ResidentCursorFactory
// This way, the returned ordering will begin with {@code __time}.
final ClusteredValueGroupsBaseTableSchema summary =
valueGroup.getSummary();
final List<OrderBy> ordering =
- useTimeOrderedCursors(spec, summary) ? summary.getGroupOrdering() :
summary.getOrdering();
+ Projections.useTimeOrderedCursors(spec, summary) ?
summary.getGroupOrdering() : summary.getOrdering();
if (plan.virtualColumnRemap().isEmpty()) {
return new QueryableIndexCursorHolder(
@@ -289,12 +287,16 @@ public class QueryableIndexCursorFactory implements
ResidentCursorFactory
}
// Use k-way merged group cursors for time ordering, or concatenated
cursors otherwise.
- if (useTimeOrderedCursors(spec, clusterSummary)) {
- return makeTimeMergedClusteredCursorHolder(
+ if (Projections.useTimeOrderedCursors(spec, clusterSummary)) {
+ final boolean descending =
Cursors.getTimeOrdering(spec.getPreferredOrdering()) == Order.DESCENDING;
+ return MergingClusterGroupCursor.makeCursorHolder(
holderSuppliers,
- closer,
- Cursors.getTimeOrdering(spec.getPreferredOrdering()),
- plan.virtualColumnRemap()
+ clusteringColumns,
+ clusteringValuesByGroup,
+ descending,
+ spec.getVirtualColumns(),
+ plan.virtualColumnRemap(),
+ closer
);
}
@@ -335,27 +337,6 @@ public class QueryableIndexCursorFactory implements
ResidentCursorFactory
return builder.build();
}
- /**
- * Whether the query requests {@code __time} ordering and each cluster group
is individually time-ordered. In that
- * case, we return time ordered cursors.
- */
- private static boolean useTimeOrderedCursors(CursorBuildSpec spec,
ClusteredValueGroupsBaseTableSchema summary)
- {
- if (Cursors.getTimeOrdering(spec.getPreferredOrdering()) == Order.NONE) {
- return false;
- }
- final List<OrderBy> groupOrdering = summary.getGroupOrdering();
- if (groupOrdering.isEmpty()) {
- return false;
- }
- final OrderBy first = groupOrdering.get(0);
- // Require __time to be the first non-clustering column AND natively
ASCENDING. Each per-group cursor can flip
- // ascending->descending on request but never the reverse, so an ascending
group ordering guarantees the per-group
- // cursors emit the direction the merge's heap (and the single-group
holder) assume. Druid always writes __time
- // ascending; guarding here keeps a hypothetical descending-written group
from being mis-ordered rather than served.
- return ColumnHolder.TIME_COLUMN_NAME.equals(first.getColumnName()) &&
first.getOrder() == Order.ASCENDING;
- }
-
private static CursorHolder
makeAggregateProjectionCursorHolder(QueryableProjection<QueryableIndex>
projection)
{
return new QueryableIndexCursorHolder(
@@ -492,50 +473,6 @@ 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). 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,
- Map<String, String> virtualColumnRemap
- )
- {
- final boolean descending = timeOrder == Order.DESCENDING;
- final MergingClusterGroupCursor cursor = new
MergingClusterGroupCursor(holderSuppliers, descending, virtualColumnRemap);
- final List<OrderBy> ordering = descending ? Cursors.descendingTimeOrder()
: Cursors.ascendingTimeOrder();
- return new CursorHolder()
- {
- @Override
- public Cursor asCursor()
- {
- return cursor;
- }
-
- @Override
- public List<OrderBy> getOrdering()
- {
- return ordering;
- }
-
- @Override
- public void close()
- {
- try {
- closer.close();
- }
- catch (IOException e) {
- throw new RuntimeException(e);
- }
- }
- };
- }
-
/**
* Vector counterpart of {@link
ClusteringColumnSelectorFactory#UNINITIALIZED_DELEGATE}. Replaced by
* {@link ConcatenatingVectorCursor}'s lazy init before the wrapper is
exposed.
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 74f2ddaa014..ac90cf6fb66 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
@@ -24,6 +24,7 @@ import com.google.common.base.Suppliers;
import com.google.common.collect.Iterables;
import org.apache.druid.error.DruidException;
import org.apache.druid.java.util.common.io.Closer;
+import org.apache.druid.query.Order;
import org.apache.druid.query.OrderBy;
import org.apache.druid.query.aggregation.AggregatorFactory;
import org.apache.druid.segment.ColumnSelectorFactory;
@@ -31,7 +32,9 @@ import org.apache.druid.segment.ConcatenatingCursor;
import org.apache.druid.segment.Cursor;
import org.apache.druid.segment.CursorBuildSpec;
import org.apache.druid.segment.CursorHolder;
+import org.apache.druid.segment.Cursors;
import org.apache.druid.segment.EmptyCursorHolder;
+import org.apache.druid.segment.MergingClusterGroupCursor;
import org.apache.druid.segment.ResidentCursorFactory;
import org.apache.druid.segment.column.ColumnCapabilities;
import org.apache.druid.segment.column.ColumnCapabilitiesImpl;
@@ -43,9 +46,9 @@ import
org.apache.druid.segment.projections.ClusteringColumnSelectorFactory;
import org.apache.druid.segment.projections.Projections;
import org.apache.druid.segment.projections.QueryableProjection;
import org.apache.druid.segment.projections.TableClusterGroupSpec;
+import org.apache.druid.utils.CloseableUtils;
import javax.annotation.Nullable;
-import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
@@ -188,6 +191,21 @@ public class IncrementalIndexCursorFactory implements
ResidentCursorFactory
);
}
+ // when the query asks for __time ordering and each group is individually
__time-sorted (__time first
+ // non-clustering column), present a globally __time-ordered cursor via a
k-way merge across groups
+ if (Projections.useTimeOrderedCursors(spec, summary)) {
+ final boolean descending =
Cursors.getTimeOrdering(spec.getPreferredOrdering()) == Order.DESCENDING;
+ return MergingClusterGroupCursor.makeCursorHolder(
+ holderSuppliers,
+ clusteringColumns,
+ clusteringValuesByGroup,
+ descending,
+ spec.getVirtualColumns(),
+ plan.virtualColumnRemap(),
+ closer
+ );
+ }
+
// The wrapper starts with a throwing placeholder delegate;
ConcatenatingCursor swaps in each group's real
// selector factory (and clustering constants) on init, before any
selector is exposed.
final ClusteringColumnSelectorFactory wrapperFactory = new
ClusteringColumnSelectorFactory(
@@ -222,12 +240,7 @@ public class IncrementalIndexCursorFactory implements
ResidentCursorFactory
@Override
public void close()
{
- try {
- closer.close();
- }
- catch (IOException e) {
- throw new RuntimeException(e);
- }
+ CloseableUtils.closeAndWrapExceptions(closer);
}
};
}
diff --git
a/processing/src/main/java/org/apache/druid/segment/projections/MergingColumnSelectorFactory.java
b/processing/src/main/java/org/apache/druid/segment/projections/MergingColumnSelectorFactory.java
index b6224e8d0e7..79484fe4ded 100644
---
a/processing/src/main/java/org/apache/druid/segment/projections/MergingColumnSelectorFactory.java
+++
b/processing/src/main/java/org/apache/druid/segment/projections/MergingColumnSelectorFactory.java
@@ -20,21 +20,29 @@
package org.apache.druid.segment.projections;
import org.apache.druid.error.DruidException;
+import org.apache.druid.math.expr.ExprEval;
+import org.apache.druid.math.expr.ExpressionType;
import org.apache.druid.query.dimension.DimensionSpec;
import org.apache.druid.query.filter.DruidPredicateFactory;
import org.apache.druid.query.filter.ValueMatcher;
import org.apache.druid.query.monomorphicprocessing.RuntimeShapeInspector;
import org.apache.druid.segment.ColumnSelectorFactory;
import org.apache.druid.segment.ColumnValueSelector;
+import org.apache.druid.segment.ConstantExprEvalSelector;
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;
+import org.apache.druid.segment.column.RowSignature;
+import org.apache.druid.segment.column.ValueType;
import org.apache.druid.segment.data.IndexedInts;
import javax.annotation.Nullable;
+import java.util.List;
import java.util.function.IntSupplier;
import java.util.function.LongSupplier;
@@ -45,20 +53,31 @@ import java.util.function.LongSupplier;
* inner selector <em>per group</em> for each requested column and, on every
access, dispatches to whichever group is
* currently winning the merge (which changes per <em>row</em>).
*
+ * <p>Clustering columns are handled directly rather than dispatched as they
are constant within a group, so this
+ * factory returns the winning group's clustering value as a per-group
constant. Non-clustering columns dispatch to the
+ * winning group's selector. A clustering column whose name is also a query
virtual column's output is NOT served as the
+ * constant: it dispatches to the winning group (whose factory resolves the
virtual column), so a shadowing VC observes
+ * the computed value rather than the constant. Names remapped away (a query
VC equivalent to a materialized column)
+ * never reach this factory: the enclosing {@code RemapColumnSelectorFactory}
rewrites them to their materialized
+ * target first.
+ *
* <p>Because the merge interleaves groups row-by-row, per-group-local
dictionary ids are never stable across the
- * merged stream, so {@link #getColumnCapabilities} strips dictionary encoding
for every column (forcing value-based
- * grouping, correct across groups) exactly as {@link
ClusteringColumnSelectorFactory} does for non-clustering columns.
- * The row id is minted from the merge's output-row counter rather than any
delegate's, since the merge emits exactly
- * one output row per advance.
+ * merged stream, so {@link #getColumnCapabilities} advertises
non-dictionary-encoded for every column (forcing
+ * value-based grouping, correct across groups) exactly as {@link
ClusteringColumnSelectorFactory} does. The row id is
+ * set from the merge's output-row counter rather than any delegate's, since
the merge emits exactly one output row
+ * per advance.
*/
public class MergingColumnSelectorFactory implements ColumnSelectorFactory
{
// Per-group factories, indexed by group. Entries may be null for groups
whose cursor was null/absent; such groups
// never win the merge, so their slots are never dispatched to.
private final ColumnSelectorFactory[] groupFactories;
+ private final RowSignature clusteringColumns;
+ private final List<Object[]> clusteringValuesByGroup;
+ private final VirtualColumns queryVirtualColumns;
// Index of the group currently winning the merge (the row being exposed).
Valid while the cursor is not done.
private final IntSupplier currentGroup;
- // First non-null group factory; a valid stand-in for every group's
capabilities (see getColumnCapabilities).
+ // First non-null group factory; a valid stand-in for every group's
non-clustering capabilities (schema-homogeneous).
@Nullable
private final ColumnSelectorFactory representative;
// Row id minted from the merge's output-row counter (one per emitted row),
forwarded to callers for caching.
@@ -66,16 +85,34 @@ public class MergingColumnSelectorFactory implements
ColumnSelectorFactory
public MergingColumnSelectorFactory(
ColumnSelectorFactory[] groupFactories,
+ RowSignature clusteringColumns,
+ List<Object[]> clusteringValuesByGroup,
+ VirtualColumns queryVirtualColumns,
IntSupplier currentGroup,
LongSupplier currentRowId
)
{
this.groupFactories = groupFactories;
+ this.clusteringColumns = clusteringColumns;
+ this.clusteringValuesByGroup = clusteringValuesByGroup;
+ this.queryVirtualColumns = queryVirtualColumns;
this.currentGroup = currentGroup;
this.representative = firstNonNull(groupFactories);
this.rowIdSupplier = currentRowId::getAsLong;
}
+ /**
+ * Index of {@code name} in the clustering columns when it should be served
as this group's clustering constant, or
+ * {@code -1} otherwise. A clustering column shadowed by a query virtual
column of the same output name is NOT served
+ * as the constant (returns {@code -1}); it dispatches to the winning group
so the computed value wins. Mirrors
+ * {@link ClusteringColumnSelectorFactory}'s {@code
servesClusteringConstant}.
+ */
+ private int clusteringConstantIndex(String name)
+ {
+ final int idx = clusteringColumns.indexOf(name);
+ return idx >= 0 && !queryVirtualColumns.exists(name) ? idx : -1;
+ }
+
@Nullable
private static ColumnSelectorFactory firstNonNull(ColumnSelectorFactory[]
factories)
{
@@ -109,9 +146,17 @@ public class MergingColumnSelectorFactory implements
ColumnSelectorFactory
@Override
public DimensionSelector makeDimensionSelector(DimensionSpec dimensionSpec)
{
+ final int clusteringIdx =
clusteringConstantIndex(dimensionSpec.getDimension());
final DimensionSelector[] perGroup = new
DimensionSelector[groupFactories.length];
for (int i = 0; i < groupFactories.length; i++) {
- if (groupFactories[i] != null) {
+ if (clusteringIdx >= 0) {
+ // Clustering column: the winning group's constant value, decorated
with any extraction fn.
+ final Object value = clusteringValuesByGroup.get(i)[clusteringIdx];
+ perGroup[i] = DimensionSelector.constant(
+ value == null ? null : String.valueOf(value),
+ dimensionSpec.getExtractionFn()
+ );
+ } else if (groupFactories[i] != null) {
perGroup[i] = groupFactories[i].makeDimensionSelector(dimensionSpec);
}
}
@@ -121,19 +166,93 @@ public class MergingColumnSelectorFactory implements
ColumnSelectorFactory
@Override
public ColumnValueSelector makeColumnValueSelector(String columnName)
{
+ final int clusteringIdx = clusteringConstantIndex(columnName);
final ColumnValueSelector[] perGroup = new
ColumnValueSelector[groupFactories.length];
- for (int i = 0; i < groupFactories.length; i++) {
- if (groupFactories[i] != null) {
- perGroup[i] = groupFactories[i].makeColumnValueSelector(columnName);
+ if (clusteringIdx >= 0) {
+ // Clustering column: the winning group's constant value, as a typed
value selector.
+ final ExpressionType type =
+
ExpressionType.fromColumnTypeStrict(clusteringColumns.getColumnType(clusteringIdx).orElseThrow());
+ for (int i = 0; i < groupFactories.length; i++) {
+ perGroup[i] = constantClusteringValueSelector(type,
clusteringValuesByGroup.get(i)[clusteringIdx]);
+ }
+ } else {
+ for (int i = 0; i < groupFactories.length; i++) {
+ if (groupFactories[i] != null) {
+ perGroup[i] = groupFactories[i].makeColumnValueSelector(columnName);
+ }
}
}
return new MergingColumnValueSelector(perGroup, columnName);
}
+ /**
+ * A constant {@link ColumnValueSelector} for a clustering column's
per-group value, unwrapping the {@link ExprEval}
+ * so {@code getObject()} yields the raw typed value (matching an ordinary
column selector).
+ */
+ private static ColumnValueSelector<?>
constantClusteringValueSelector(ExpressionType type, @Nullable Object value)
+ {
+ final ConstantExprEvalSelector eval = new
ConstantExprEvalSelector(ExprEval.ofType(type, value));
+ return new ColumnValueSelector<>()
+ {
+ @Override
+ public double getDouble()
+ {
+ return eval.getDouble();
+ }
+
+ @Override
+ public float getFloat()
+ {
+ return eval.getFloat();
+ }
+
+ @Override
+ public long getLong()
+ {
+ return eval.getLong();
+ }
+
+ @Override
+ public boolean isNull()
+ {
+ return eval.isNull();
+ }
+
+ @Nullable
+ @Override
+ public Object getObject()
+ {
+ return eval.getObject().value();
+ }
+
+ @Override
+ public Class<?> classOfObject()
+ {
+ return Object.class;
+ }
+
+ @Override
+ public void inspectRuntimeShape(RuntimeShapeInspector inspector)
+ {
+ eval.inspectRuntimeShape(inspector);
+ }
+ };
+ }
+
@Nullable
@Override
public ColumnCapabilities getColumnCapabilities(String column)
{
+ final int clusteringIdx = clusteringConstantIndex(column);
+ if (clusteringIdx >= 0) {
+ // Clustering columns are exposed as per-group constants; report simple
type-based capabilities (never
+ // dictionary-encoded across the merge), exactly as
ClusteringColumnSelectorFactory does.
+ final ColumnType type =
clusteringColumns.getColumnType(clusteringIdx).orElseThrow();
+ if (type.is(ValueType.STRING)) {
+ return
ColumnCapabilitiesImpl.createSimpleSingleValueStringColumnCapabilities();
+ }
+ return
ColumnCapabilitiesImpl.createSimpleNumericColumnCapabilities(type);
+ }
if (representative == null) {
return null;
}
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 ef211ae6e24..3649071d70e 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
@@ -27,6 +27,7 @@ import
org.apache.druid.java.util.common.granularity.Granularities;
import org.apache.druid.java.util.common.granularity.Granularity;
import org.apache.druid.java.util.common.granularity.PeriodGranularity;
import org.apache.druid.java.util.common.logger.Logger;
+import org.apache.druid.query.Order;
import org.apache.druid.query.QueryContext;
import org.apache.druid.query.QueryContexts;
import org.apache.druid.query.aggregation.AggregatorFactory;
@@ -40,6 +41,7 @@ import org.apache.druid.query.filter.TypedInFilter;
import org.apache.druid.segment.AggregateProjectionMetadata;
import org.apache.druid.segment.CursorBuildSpec;
import org.apache.druid.segment.CursorHolder;
+import org.apache.druid.segment.Cursors;
import org.apache.druid.segment.VirtualColumn;
import org.apache.druid.segment.VirtualColumns;
import org.apache.druid.segment.column.ColumnHolder;
@@ -792,6 +794,19 @@ public class Projections
return remap;
}
+ /**
+ * Whether a clustered read should serve a globally {@code __time}-ordered
cursor for {@code spec}. True when the
+ * query requests {@code __time} ordering AND each cluster group is
individually {@code __time}-sorted, i.e.
+ * {@code __time} is the first non-clustering column.
+ */
+ public static boolean useTimeOrderedCursors(CursorBuildSpec spec,
ClusteredValueGroupsBaseTableSchema summary)
+ {
+ if (Cursors.getTimeOrdering(spec.getPreferredOrdering()) == Order.NONE) {
+ return false;
+ }
+ return Cursors.getTimeOrdering(summary.getGroupOrdering()) ==
Order.ASCENDING;
+ }
+
/**
* Walk the filter tree against {@code group}'s constant clustering tuple
and return a rewritten filter where each
* recognized equality / in / null leaf whose column resolves to a
clustering column (physical or virtual) is folded
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 b067b4ccc94..d5fa466bf2e 100644
---
a/processing/src/test/java/org/apache/druid/segment/ClusteredSegmentTimeOrderedQueryTest.java
+++
b/processing/src/test/java/org/apache/druid/segment/ClusteredSegmentTimeOrderedQueryTest.java
@@ -52,7 +52,9 @@ import
org.apache.druid.query.timeseries.TimeseriesQueryRunnerFactory;
import org.apache.druid.query.timeseries.TimeseriesResultValue;
import org.apache.druid.segment.column.ColumnHolder;
import org.apache.druid.segment.column.ColumnType;
+import org.apache.druid.segment.incremental.IncrementalIndex;
import org.apache.druid.segment.incremental.IncrementalIndexSchema;
+import org.apache.druid.segment.incremental.OnheapIncrementalIndex;
import org.apache.druid.segment.virtual.ExpressionVirtualColumn;
import
org.apache.druid.segment.writeout.OffHeapMemorySegmentWriteOutMediumFactory;
import org.apache.druid.testing.InitializedNullHandlingTest;
@@ -225,6 +227,64 @@ class ClusteredSegmentTimeOrderedQueryTest extends
InitializedNullHandlingTest
Assertions.assertEquals(expected, runScanWithLowerTenantVc(clustered,
Order.ASCENDING));
}
+ @Test
+ void testAscendingScanWithQueryVcShadowingClusteringColumn()
+ {
+ // A query VC whose OUTPUT name shadows the clustering column `tenant` but
computes a different value (from `m`, not
+ // equivalent to any materialized column, so it is NOT remapped away). On
the __time merge path the merge factory
+ // must dispatch reads of `tenant` to the winning group (which resolves
the VC) rather than serve the raw clustering
+ // constant. Without the shadowing guard this returned the group constant
("acme"/"globex") instead of the computed
+ // value, diverging from the concatenating path.
+ final List<List<Object>> expected = List.of(
+ Arrays.asList(T0, "t_1", 1L),
+ Arrays.asList(T0 + MINUTE, "t_2", 2L),
+ Arrays.asList(T0 + 2 * MINUTE, "t_4", 4L),
+ Arrays.asList(T0 + 3 * MINUTE, "t_8", 8L)
+ );
+ Assertions.assertEquals(expected,
runScanWithShadowingTenantVc(clusteredSegment, Order.ASCENDING));
+ }
+
+ @Test
+ void testTimeOrderedQueriesOverIncrementalClusteredSegment()
+ {
+ // The realtime path (IncrementalIndexCursorFactory) must also k-way-merge
groups for a __time-ordered query.
+ // Incremental per-group cursors do not carry clustering columns, so this
also exercises the merge factory
+ // injecting the winning group's clustering value (tenant). Results must
match the non-clustered baseline.
+ final Segment incremental = buildClusteredIncremental(ROWS);
+
+ Assertions.assertEquals(
+ List.of(
+ List.of(T0, 1L),
+ List.of(T0 + MINUTE, 2L),
+ List.of(T0 + 2 * MINUTE, 4L),
+ List.of(T0 + 3 * MINUTE, 8L)
+ ),
+ runTimeseries(incremental)
+ );
+ Assertions.assertEquals(runTimeseries(nonClusteredSegment),
runTimeseries(incremental));
+
+ Assertions.assertEquals(
+ 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)
+ ),
+ runScanRows(incremental, Order.ASCENDING)
+ );
+ Assertions.assertEquals(runScanRows(nonClusteredSegment, Order.ASCENDING),
runScanRows(incremental, Order.ASCENDING));
+
+ Assertions.assertEquals(
+ List.of(
+ Arrays.asList(T0 + 3 * MINUTE, "globex", 8L),
+ Arrays.asList(T0 + 2 * MINUTE, "acme", 4L),
+ Arrays.asList(T0 + MINUTE, "globex", 2L),
+ Arrays.asList(T0, "acme", 1L)
+ ),
+ runScanRows(incremental, Order.DESCENDING)
+ );
+ }
+
private static List<List<Long>> runTimeseries(Segment segment)
{
final TimeseriesQuery query = Druids.newTimeseriesQueryBuilder()
@@ -321,6 +381,48 @@ class ClusteredSegmentTimeOrderedQueryTest extends
InitializedNullHandlingTest
return rows;
}
+ /**
+ * Time-ordered scan projecting a query virtual column {@code tenant :=
concat('t_', m)} whose output name shadows the
+ * clustering column {@code tenant} (plus {@code __time} and {@code m}),
extracting {@code [__time, tenant, m]} rows.
+ * Exercises the shadowing-VC guard on the {@code __time} merge path: reads
of {@code tenant} must resolve to the
+ * computed VC value, not the group's clustering constant.
+ */
+ private static List<List<Object>> runScanWithShadowingTenantVc(Segment
segment, Order order)
+ {
+ final ScanQuery query = Druids.newScanQueryBuilder()
+ .dataSource(DATA_SOURCE)
+ .intervals(new
MultipleIntervalSegmentSpec(List.of(INTERVAL)))
+ .virtualColumns(new ExpressionVirtualColumn(
+ "tenant",
+ "concat('t_', m)",
+ ColumnType.STRING,
+ TestExprMacroTable.INSTANCE
+ ))
+ .columns(ColumnHolder.TIME_COLUMN_NAME,
"tenant", "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("tenant"),
+ 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
@@ -412,6 +514,38 @@ class ClusteredSegmentTimeOrderedQueryTest extends
InitializedNullHandlingTest
.buildMMappedIndex();
}
+ private static Segment buildClusteredIncremental(List<InputRow> rows)
+ {
+ // Realtime counterpart of buildClustered: an OnheapIncrementalIndex
clustered by tenant with __time as the first
+ // non-clustering column, served through IncrementalIndexCursorFactory
(via IncrementalIndexSegment).
+ final ClusteredValueGroupsBaseTableProjectionSpec clusterSpec =
+ ClusteredValueGroupsBaseTableProjectionSpec.builder()
+ .columns(
+ new StringDimensionSchema("tenant"),
+ new LongDimensionSchema("__time"),
+ new LongDimensionSchema("m")
+ )
+ .clusteringColumns("tenant")
+ .build();
+ final IncrementalIndexSchema schema =
+ IncrementalIndexSchema.builder()
+ .withMinTimestamp(T0)
+ .withTimestampSpec(TIMESTAMP_SPEC)
+ .withQueryGranularity(Granularities.NONE)
+
.withDimensionsSpec(clusterSpec.getDimensionsSpec())
+ .withRollup(false)
+ .withClusterSpec(clusterSpec)
+ .build();
+ final IncrementalIndex index = new OnheapIncrementalIndex.Builder()
+ .setIndexSchema(schema)
+ .setMaxRowCount(10_000)
+ .build();
+ for (final InputRow row : rows) {
+ index.add(row);
+ }
+ return new IncrementalIndexSegment(index, SegmentId.dummy(DATA_SOURCE));
+ }
+
private static InputRow row(long ts, String tenant, long m)
{
final Map<String, Object> event = new HashMap<>();
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 9c3f3b187f1..836f4072014 100644
---
a/processing/src/test/java/org/apache/druid/segment/MergingClusterGroupCursorTest.java
+++
b/processing/src/test/java/org/apache/druid/segment/MergingClusterGroupCursorTest.java
@@ -23,11 +23,13 @@ import com.google.common.base.Supplier;
import org.apache.druid.query.dimension.DimensionSpec;
import org.apache.druid.segment.column.ColumnCapabilities;
import org.apache.druid.segment.column.ColumnHolder;
+import org.apache.druid.segment.column.RowSignature;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
import javax.annotation.Nullable;
import java.util.ArrayList;
+import java.util.Collections;
import java.util.List;
import java.util.Map;
@@ -161,12 +163,16 @@ class MergingClusterGroupCursorTest
// 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 List<Supplier<CursorHolder>> suppliers = 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"})
+ ));
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"})
- )),
+ suppliers,
+ RowSignature.empty(),
+ List.of(new Object[0], new Object[0]),
false,
+ VirtualColumns.EMPTY,
Map.of("aliased", "v")
);
final ColumnSelectorFactory factory = cursor.getColumnSelectorFactory();
@@ -183,7 +189,16 @@ class MergingClusterGroupCursorTest
@SafeVarargs
private static MergingClusterGroupCursor cursor(boolean descending,
Supplier<CursorHolder>... groups)
{
- return new MergingClusterGroupCursor(new ArrayList<>(List.of(groups)),
descending, Map.of());
+ // These cases only read non-clustering columns ("v"), so no clustering
columns/values are needed.
+ final List<Supplier<CursorHolder>> suppliers = new
ArrayList<>(List.of(groups));
+ return new MergingClusterGroupCursor(
+ suppliers,
+ RowSignature.empty(),
+ Collections.nCopies(suppliers.size(), new Object[0]),
+ descending,
+ VirtualColumns.EMPTY,
+ Map.of()
+ );
}
private static List<Object[]> drain(MergingClusterGroupCursor cursor)
diff --git
a/processing/src/test/java/org/apache/druid/segment/projections/MergingColumnSelectorFactoryTest.java
b/processing/src/test/java/org/apache/druid/segment/projections/MergingColumnSelectorFactoryTest.java
index a8ea723a979..236b81b049e 100644
---
a/processing/src/test/java/org/apache/druid/segment/projections/MergingColumnSelectorFactoryTest.java
+++
b/processing/src/test/java/org/apache/druid/segment/projections/MergingColumnSelectorFactoryTest.java
@@ -19,6 +19,7 @@
package org.apache.druid.segment.projections;
+import org.apache.druid.math.expr.ExprMacroTable;
import org.apache.druid.query.dimension.DefaultDimensionSpec;
import org.apache.druid.query.dimension.DimensionSpec;
import org.apache.druid.segment.ColumnSelectorFactory;
@@ -27,13 +28,19 @@ import org.apache.druid.segment.DimensionDictionarySelector;
import org.apache.druid.segment.DimensionSelector;
import org.apache.druid.segment.RowIdSupplier;
import org.apache.druid.segment.TestObjectColumnSelector;
+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;
+import org.apache.druid.segment.column.RowSignature;
import org.apache.druid.segment.column.ValueType;
+import org.apache.druid.segment.virtual.ExpressionVirtualColumn;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
import javax.annotation.Nullable;
+import java.util.Collections;
+import java.util.List;
import java.util.function.IntSupplier;
import java.util.function.LongSupplier;
@@ -43,10 +50,37 @@ class MergingColumnSelectorFactoryTest
private final long[] rowId = {0};
private MergingColumnSelectorFactory factory(ColumnSelectorFactory... groups)
+ {
+ // No clustering columns: every requested column dispatches to the
per-group factories.
+ return factory(RowSignature.empty(), Collections.nCopies(groups.length,
new Object[0]), groups);
+ }
+
+ private MergingColumnSelectorFactory factory(
+ RowSignature clusteringColumns,
+ List<Object[]> clusteringValuesByGroup,
+ ColumnSelectorFactory... groups
+ )
+ {
+ return factory(clusteringColumns, clusteringValuesByGroup,
VirtualColumns.EMPTY, groups);
+ }
+
+ private MergingColumnSelectorFactory factory(
+ RowSignature clusteringColumns,
+ List<Object[]> clusteringValuesByGroup,
+ VirtualColumns queryVirtualColumns,
+ ColumnSelectorFactory... groups
+ )
{
final IntSupplier currentGroupSupplier = () -> currentGroup[0];
final LongSupplier rowIdSupplier = () -> rowId[0];
- return new MergingColumnSelectorFactory(groups, currentGroupSupplier,
rowIdSupplier);
+ return new MergingColumnSelectorFactory(
+ groups,
+ clusteringColumns,
+ clusteringValuesByGroup,
+ queryVirtualColumns,
+ currentGroupSupplier,
+ rowIdSupplier
+ );
}
@Test
@@ -106,6 +140,76 @@ class MergingColumnSelectorFactoryTest
Assertions.assertFalse(selector.nameLookupPossibleInAdvance());
}
+ @Test
+ void testClusteringColumnExposedAsWinningGroupConstant()
+ {
+ // Clustering columns are not read from the per-group cursors (which may
not carry them); the factory injects the
+ // winning group's clustering value. Verify both the value selector and
dimension selector track the current group.
+ final RowSignature clusteringColumns =
RowSignature.builder().add("tenant", ColumnType.STRING).build();
+ final List<Object[]> clusteringValuesByGroup = List.of(new
Object[]{"acme"}, new Object[]{"globex"});
+ final MergingColumnSelectorFactory factory =
+ factory(clusteringColumns, clusteringValuesByGroup,
groupFactory("g0"), groupFactory("g1"));
+
+ final ColumnValueSelector valueSelector =
factory.makeColumnValueSelector("tenant");
+ final DimensionSelector dimensionSelector =
factory.makeDimensionSelector(DefaultDimensionSpec.of("tenant"));
+ currentGroup[0] = 0;
+ Assertions.assertEquals("acme", valueSelector.getObject());
+ Assertions.assertEquals("acme", dimensionSelector.getObject());
+ currentGroup[0] = 1;
+ Assertions.assertEquals("globex", valueSelector.getObject());
+ Assertions.assertEquals("globex", dimensionSelector.getObject());
+
+ // Clustering capabilities are type-based and never dictionary-encoded
across the merge.
+ final ColumnCapabilities caps = factory.getColumnCapabilities("tenant");
+ Assertions.assertNotNull(caps);
+ Assertions.assertTrue(caps.is(ValueType.STRING));
+ Assertions.assertTrue(caps.isDictionaryEncoded().isFalse());
+ Assertions.assertEquals(DimensionDictionarySelector.CARDINALITY_UNKNOWN,
dimensionSelector.getValueCardinality());
+ }
+
+ @Test
+ void testNumericClusteringColumnExposedAsWinningGroupConstant()
+ {
+ final RowSignature clusteringColumns =
RowSignature.builder().add("priority", ColumnType.LONG).build();
+ final List<Object[]> clusteringValuesByGroup = List.of(new Object[]{5L},
new Object[]{9L});
+ final MergingColumnSelectorFactory factory =
+ factory(clusteringColumns, clusteringValuesByGroup,
groupFactory("g0"), groupFactory("g1"));
+
+ final ColumnValueSelector selector =
factory.makeColumnValueSelector("priority");
+ currentGroup[0] = 0;
+ Assertions.assertEquals(5L, selector.getLong());
+ Assertions.assertEquals(5L, selector.getObject());
+ currentGroup[0] = 1;
+ Assertions.assertEquals(9L, selector.getLong());
+
+
Assertions.assertTrue(factory.getColumnCapabilities("priority").is(ValueType.LONG));
+ }
+
+ @Test
+ void testClusteringColumnShadowedByQueryVcDispatchesToGroup()
+ {
+ // A query virtual column whose output name equals a clustering column
must NOT be served as the group's clustering
+ // constant; it dispatches to the winning group (whose factory resolves
the VC), mirroring
+ // ClusteringColumnSelectorFactory. The stub group factory answers
"g0"/"g1" for any column, standing in for the
+ // computed VC value; the assertion is that we get the dispatched value,
not the "acme"/"globex" constant.
+ final RowSignature clusteringColumns =
RowSignature.builder().add("tenant", ColumnType.STRING).build();
+ final List<Object[]> clusteringValuesByGroup = List.of(new
Object[]{"acme"}, new Object[]{"globex"});
+ final VirtualColumns shadowingVc = VirtualColumns.create(
+ new ExpressionVirtualColumn("tenant", "concat('t_', m)",
ColumnType.STRING, ExprMacroTable.nil())
+ );
+ final MergingColumnSelectorFactory factory =
+ factory(clusteringColumns, clusteringValuesByGroup, shadowingVc,
groupFactory("g0"), groupFactory("g1"));
+
+ final ColumnValueSelector valueSelector =
factory.makeColumnValueSelector("tenant");
+ final DimensionSelector dimensionSelector =
factory.makeDimensionSelector(DefaultDimensionSpec.of("tenant"));
+ currentGroup[0] = 0;
+ Assertions.assertEquals("g0", valueSelector.getObject());
+ Assertions.assertEquals("g0", dimensionSelector.getObject());
+ currentGroup[0] = 1;
+ Assertions.assertEquals("g1", valueSelector.getObject());
+ Assertions.assertEquals("g1", dimensionSelector.getObject());
+ }
+
@Test
void testRowIdSupplierReflectsMintedOutputRowId()
{
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
