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 80ae6bc7d0f feat: support time ordered cursor for clustered segments
when possible (#19726)
80ae6bc7d0f is described below
commit 80ae6bc7d0f7e4f5907d7623f40d0842bde90dde
Author: Clint Wylie <[email protected]>
AuthorDate: Thu Jul 23 17:40:33 2026 -0700
feat: support time ordered cursor for clustered segments when possible
(#19726)
---
.../druid/segment/MergingClusterGroupCursor.java | 208 +++++++++++++
.../druid/segment/QueryableIndexCursorFactory.java | 340 ++++++++++++--------
.../projections/MergingColumnSelectorFactory.java | 342 +++++++++++++++++++++
.../ClusteredSegmentTimeOrderedQueryTest.java | 320 +++++++++++++++++++
.../druid/segment/IndexMergerV10ClusteredTest.java | 128 ++++++++
.../segment/MergingClusterGroupCursorTest.java | 302 ++++++++++++++++++
.../MergingColumnSelectorFactoryTest.java | 170 ++++++++++
7 files changed, 1688 insertions(+), 122 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
new file mode 100644
index 00000000000..a3a22fe55e3
--- /dev/null
+++
b/processing/src/main/java/org/apache/druid/segment/MergingClusterGroupCursor.java
@@ -0,0 +1,208 @@
+/*
+ * 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;
+
+import com.google.common.base.Supplier;
+import it.unimi.dsi.fastutil.ints.IntHeapPriorityQueue;
+import org.apache.druid.error.DruidException;
+import org.apache.druid.segment.column.ColumnHolder;
+import org.apache.druid.segment.projections.MergingColumnSelectorFactory;
+
+import java.util.List;
+
+/**
+ * {@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}.
+ * <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.
+ * <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()}
+ * re-settles it after its cursor advances, and it is dropped once exhausted.
Ties on {@code __time} across groups break
+ * by group index (arbitrary but deterministic); the advertised ordering is
only {@code [__time]}, so secondary sort
+ * columns are not preserved across groups. The outer {@link CursorHolder}
owns the lifecycle of the per-group holders.
+ */
+public final class MergingClusterGroupCursor implements Cursor
+{
+ private final List<Supplier<CursorHolder>> holderSuppliers;
+ private final boolean descending;
+
+ private boolean initialized;
+ // Indexed by group. groupCursors[i] is null if that group's holder produced
no cursor.
+ private Cursor[] groupCursors;
+ private BaseLongColumnValueSelector[] timeSelectors;
+ private long[] currentTimes;
+ // Group indices ordered by current __time (min for ascending, max for
descending); the head is the winning group.
+ private IntHeapPriorityQueue heap;
+ // heap head while not done; retains the last winner once done (selectors
are undefined after isDone() anyway).
+ 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;
+
+ public MergingClusterGroupCursor(
+ List<Supplier<CursorHolder>> holderSuppliers,
+ boolean descending
+ )
+ {
+ if (holderSuppliers.isEmpty()) {
+ throw DruidException.defensive("MergingClusterGroupCursor requires at
least one cluster group");
+ }
+ this.holderSuppliers = holderSuppliers;
+ this.descending = descending;
+ }
+
+ private void initializeIfNeeded()
+ {
+ if (initialized) {
+ return;
+ }
+ initialized = true;
+ final int n = holderSuppliers.size();
+ groupCursors = new Cursor[n];
+ timeSelectors = new BaseLongColumnValueSelector[n];
+ currentTimes = new long[n];
+ heap = new IntHeapPriorityQueue(this::compareGroups);
+ outputRowId = 0;
+ final ColumnSelectorFactory[] groupFactories = new
ColumnSelectorFactory[n];
+ for (int i = 0; i < n; i++) {
+ final CursorHolder holder = holderSuppliers.get(i).get();
+ final Cursor cursor = holder.asCursor();
+ groupCursors[i] = cursor;
+ if (cursor == null) {
+ continue;
+ }
+ // A per-group cursor is still queryable for its factory/capabilities
even when empty (done); only non-empty
+ // groups join the heap.
+ groupFactories[i] = cursor.getColumnSelectorFactory();
+ timeSelectors[i] =
groupFactories[i].makeColumnValueSelector(ColumnHolder.TIME_COLUMN_NAME);
+ if (!cursor.isDone()) {
+ currentTimes[i] = timeSelectors[i].getLong();
+ heap.enqueue(i);
+ }
+ }
+ currentGroup = heap.isEmpty() ? 0 : heap.firstInt();
+ factory = new MergingColumnSelectorFactory(groupFactories, () ->
currentGroup, () -> outputRowId);
+ }
+
+ @Override
+ public ColumnSelectorFactory getColumnSelectorFactory()
+ {
+ initializeIfNeeded();
+ return factory;
+ }
+
+ @Override
+ public void advance()
+ {
+ if (isDone()) {
+ return;
+ }
+ final int root = heap.firstInt();
+ groupCursors[root].advance();
+ settleRoot(root);
+ outputRowId++;
+ }
+
+ @Override
+ public void advanceUninterruptibly()
+ {
+ if (isDone()) {
+ return;
+ }
+ final int root = heap.firstInt();
+ groupCursors[root].advanceUninterruptibly();
+ settleRoot(root);
+ outputRowId++;
+ }
+
+ /**
+ * Re-settle the heap after the winning group's cursor has been advanced:
drop the group if it is now exhausted,
+ * otherwise re-key it on its new {@code __time} and let the queue re-sift
the head. Refreshes {@link #currentGroup}.
+ */
+ private void settleRoot(int root)
+ {
+ if (groupCursors[root].isDone()) {
+ heap.dequeueInt();
+ } else {
+ currentTimes[root] = timeSelectors[root].getLong();
+ heap.changed();
+ }
+ if (!heap.isEmpty()) {
+ currentGroup = heap.firstInt();
+ }
+ }
+
+ @Override
+ public boolean isDone()
+ {
+ initializeIfNeeded();
+ return heap.isEmpty();
+ }
+
+ @Override
+ public boolean isDoneOrInterrupted()
+ {
+ return isDone() || Thread.currentThread().isInterrupted();
+ }
+
+ @Override
+ public void reset()
+ {
+ if (!initialized) {
+ return;
+ }
+ // Reset the held per-group cursors in place (do NOT re-fetch via
asCursor(), which would produce fresh factories
+ // and invalidate the selectors already handed out through the
MergingColumnSelectorFactory), then rebuild the heap.
+ heap.clear();
+ outputRowId = 0;
+ for (int i = 0; i < groupCursors.length; i++) {
+ final Cursor cursor = groupCursors[i];
+ if (cursor == null) {
+ continue;
+ }
+ cursor.reset();
+ if (!cursor.isDone()) {
+ currentTimes[i] = timeSelectors[i].getLong();
+ heap.enqueue(i);
+ }
+ }
+ currentGroup = heap.isEmpty() ? 0 : heap.firstInt();
+ }
+
+ /**
+ * Heap ordering over group indices: earlier {@code __time} sits at the head
for ascending, later for descending,
+ * breaking ties by group index for determinism.
+ */
+ private int compareGroups(int a, int b)
+ {
+ final int cmp = descending
+ ? Long.compare(currentTimes[b], currentTimes[a])
+ : Long.compare(currentTimes[a], currentTimes[b]);
+ return cmp != 0 ? cmp : Integer.compare(a, b);
+ }
+}
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 a29b2b57072..ac0e843260a 100644
---
a/processing/src/main/java/org/apache/druid/segment/QueryableIndexCursorFactory.java
+++
b/processing/src/main/java/org/apache/druid/segment/QueryableIndexCursorFactory.java
@@ -23,6 +23,7 @@ import com.google.common.base.Supplier;
import com.google.common.base.Suppliers;
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.query.dimension.DimensionSpec;
@@ -48,6 +49,7 @@ import org.apache.druid.segment.vector.VectorCursor;
import org.apache.druid.segment.vector.VectorObjectSelector;
import org.apache.druid.segment.vector.VectorOffset;
import org.apache.druid.segment.vector.VectorValueSelector;
+import org.apache.druid.utils.CloseableUtils;
import javax.annotation.Nullable;
import java.io.IOException;
@@ -116,59 +118,6 @@ public class QueryableIndexCursorFactory implements
ResidentCursorFactory
return new QueryableIndexCursorHolder(index, spec, timeBoundaryInspector);
}
- private CursorHolder
makeAggregateProjectionCursorHolder(QueryableProjection<QueryableIndex>
projection)
- {
- return new QueryableIndexCursorHolder(
- projection.getRowSelector(),
- projection.getCursorBuildSpec(),
- QueryableIndexTimeBoundaryInspector.create(projection.getRowSelector())
- )
- {
- @Override
- protected ColumnSelectorFactory makeColumnSelectorFactoryForOffset(
- ColumnCache columnCache,
- Offset baseOffset
- )
- {
- return projection.wrapColumnSelectorFactory(
- super.makeColumnSelectorFactoryForOffset(columnCache, baseOffset)
- );
- }
-
- @Override
- protected VectorColumnSelectorFactory
makeVectorColumnSelectorFactoryForOffset(
- ColumnCache columnCache,
- VectorOffset baseOffset
- )
- {
- return projection.wrapVectorColumnSelectorFactory(
- super.makeVectorColumnSelectorFactoryForOffset(columnCache,
baseOffset)
- );
- }
-
- @Override
- public boolean isPreAggregated()
- {
- return true;
- }
-
- @Nullable
- @Override
- public List<AggregatorFactory> getAggregatorsForPreAggregated()
- {
- return projection.getCursorBuildSpec().getAggregators();
- }
- };
- }
-
- private CursorHolder makeClusteredCursorHolder(CursorBuildSpec spec)
- {
- return makeClusteredCursorHolder(
- spec,
- Projections.planClusterGroupQuery(new
ArrayList<>(index.getClusterGroupSchemas()), spec)
- );
- }
-
/**
* Build a clustered-base-table cursor holder from an already-computed
{@link ClusterGroupQueryPlan}. Exposed so the
* partial (on-demand) cursor factory can plan the cluster groups once — to
decide which group bundles to download —
@@ -186,6 +135,53 @@ public class QueryableIndexCursorFactory implements
ResidentCursorFactory
return makeMultiGroupClusteredCursorHolder(spec, plan);
}
+ @Override
+ public RowSignature getRowSignature()
+ {
+ final ClusteredValueGroupsBaseTableSchema clusterSummary =
index.getClusteredBaseSummary();
+ if (clusterSummary != null) {
+ return getClusteredRowSignature(clusterSummary);
+ }
+
+ final LinkedHashSet<String> columns = new LinkedHashSet<>();
+
+ for (final OrderBy orderBy : index.getOrdering()) {
+ columns.add(orderBy.getColumnName());
+ }
+
+ // Add __time after the defined ordering, if __time wasn't part of it.
+ columns.add(ColumnHolder.TIME_COLUMN_NAME);
+ columns.addAll(index.getColumnNames());
+
+ final RowSignature.Builder builder = RowSignature.builder();
+ for (final String column : columns) {
+ final ColumnType columnType =
ColumnType.fromCapabilities(index.getColumnCapabilities(column));
+
+ // index.getOrdering() may include columns that don't exist, such as if
they were omitted due to
+ // being 100% nulls. Don't add those to the row signature.
+ if (columnType != null) {
+ builder.add(column, columnType);
+ }
+ }
+
+ return builder.build();
+ }
+
+ @Nullable
+ @Override
+ public ColumnCapabilities getColumnCapabilities(String column)
+ {
+ return index.getColumnCapabilities(column);
+ }
+
+ private CursorHolder makeClusteredCursorHolder(CursorBuildSpec spec)
+ {
+ return makeClusteredCursorHolder(
+ spec,
+ Projections.planClusterGroupQuery(new
ArrayList<>(index.getClusterGroupSchemas()), spec)
+ );
+ }
+
private CursorHolder makeSingleGroupClusteredCursorHolder(
CursorBuildSpec spec,
ClusterGroupQueryPlan plan,
@@ -200,12 +196,18 @@ public class QueryableIndexCursorFactory implements
ResidentCursorFactory
);
}
+ // Omit cluster key from ordering if the caller requests (and is granted)
time ordering.
+ // 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();
+
// groupIndex exposes the group's clustering columns as constant columns,
no selector wrapper is needed
return new QueryableIndexCursorHolder(
groupIndex,
plan.rebuildCursorBuildSpec(spec, valueGroup),
QueryableIndexTimeBoundaryInspector.create(groupIndex),
- valueGroup.getSummary().getOrdering()
+ ordering
);
}
@@ -251,10 +253,132 @@ public class QueryableIndexCursorFactory implements
ResidentCursorFactory
);
}
+ // Use k-way merged group cursors for time ordering, or concatenated
cursors otherwise.
+ if (useTimeOrderedCursors(spec, clusterSummary)) {
+ return makeTimeMergedClusteredCursorHolder(
+ holderSuppliers,
+ closer,
+ Cursors.getTimeOrdering(spec.getPreferredOrdering())
+ );
+ }
+
+ return makeConcatenatedClusteredCursorHolder(
+ spec,
+ this,
+ clusteringColumns,
+ clusteringValuesByGroup,
+ holderSuppliers,
+ clusterSummary,
+ closer
+ );
+ }
+
+ /**
+ * Build the row signature for a clustered segment. Top-level columns are
empty, so column types are sourced from:
+ * - the summary's clustering {@link RowSignature} for clustering columns;
+ * - the first cluster group's sub-index for everything else (all groups
share the same data-column shape).
+ */
+ private RowSignature
getClusteredRowSignature(ClusteredValueGroupsBaseTableSchema clusterSummary)
+ {
+ final LinkedHashSet<String> columns = new LinkedHashSet<>();
+
+ for (final OrderBy orderBy : clusterSummary.getOrdering()) {
+ columns.add(orderBy.getColumnName());
+ }
+ columns.add(ColumnHolder.TIME_COLUMN_NAME);
+ columns.addAll(clusterSummary.getColumnNames());
+
+ final RowSignature.Builder builder = RowSignature.builder();
+ for (final String column : columns) {
+ final ColumnType columnType =
ColumnType.fromCapabilities(index.getColumnCapabilities(column));
+ if (columnType != null) {
+ builder.add(column, columnType);
+ }
+ }
+ 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(
+ projection.getRowSelector(),
+ projection.getCursorBuildSpec(),
+ QueryableIndexTimeBoundaryInspector.create(projection.getRowSelector())
+ )
+ {
+ @Override
+ protected ColumnSelectorFactory makeColumnSelectorFactoryForOffset(
+ ColumnCache columnCache,
+ Offset baseOffset
+ )
+ {
+ return projection.wrapColumnSelectorFactory(
+ super.makeColumnSelectorFactoryForOffset(columnCache, baseOffset)
+ );
+ }
+
+ @Override
+ protected VectorColumnSelectorFactory
makeVectorColumnSelectorFactoryForOffset(
+ ColumnCache columnCache,
+ VectorOffset baseOffset
+ )
+ {
+ return projection.wrapVectorColumnSelectorFactory(
+ super.makeVectorColumnSelectorFactoryForOffset(columnCache,
baseOffset)
+ );
+ }
+
+ @Override
+ public boolean isPreAggregated()
+ {
+ return true;
+ }
+
+ @Nullable
+ @Override
+ public List<AggregatorFactory> getAggregatorsForPreAggregated()
+ {
+ return projection.getCursorBuildSpec().getAggregators();
+ }
+ };
+ }
+
+ /**
+ * Builds a {@link CursorHolder} that concatenates cluster group cursors
together.
+ */
+ private static CursorHolder makeConcatenatedClusteredCursorHolder(
+ CursorBuildSpec spec,
+ ColumnInspector inspector,
+ RowSignature clusteringColumns,
+ List<Object[]> clusteringValuesByGroup,
+ List<Supplier<CursorHolder>> holderSuppliers,
+ ClusteredValueGroupsBaseTableSchema clusterSummary,
+ Closer closer
+ )
+ {
// Initial wrapper state uses the first group's clustering values + a
throwing placeholder delegate. The
- // ConcatenatingCursor immediately calls setDelegate on init (before any
selector is exposed). The vector
- // wrapper carries the query-level max vector size from the build spec,
the placeholder delegate can't be
- // queried for sizing, and the value is constant across groups anyway.
+ // ConcatenatingCursor immediately calls setDelegate on init (before any
selector is exposed).
final int vectorSize = spec.getQueryContext().getVectorSize();
final ClusteringColumnSelectorFactory wrapperFactory = new
ClusteringColumnSelectorFactory(
ClusteringColumnSelectorFactory.UNINITIALIZED_DELEGATE,
@@ -284,7 +408,7 @@ public class QueryableIndexCursorFactory implements
ResidentCursorFactory
// original filter (clustering leaves fold to constant TRUE/FALSE, other
leaves pass through unchanged)
final Filter queryFilter = spec.getFilter();
final boolean filterCanVectorize =
- queryFilter == null ||
queryFilter.canVectorizeMatcher(spec.getVirtualColumns().wrapInspector(this));
+ 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();
@@ -318,6 +442,42 @@ public class QueryableIndexCursorFactory implements
ResidentCursorFactory
return clusterSummary.getOrdering();
}
+ @Override
+ public void close()
+ {
+ CloseableUtils.closeAndWrapExceptions(closer);
+ }
+ };
+ }
+
+ /**
+ * 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).
+ */
+ private static CursorHolder makeTimeMergedClusteredCursorHolder(
+ List<Supplier<CursorHolder>> holderSuppliers,
+ Closer closer,
+ Order timeOrder
+ )
+ {
+ final boolean descending = timeOrder == Order.DESCENDING;
+ final MergingClusterGroupCursor cursor = new
MergingClusterGroupCursor(holderSuppliers, descending);
+ 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()
{
@@ -374,68 +534,4 @@ public class QueryableIndexCursorFactory implements
ResidentCursorFactory
return null;
}
};
-
- @Override
- public RowSignature getRowSignature()
- {
- final ClusteredValueGroupsBaseTableSchema clusterSummary =
index.getClusteredBaseSummary();
- if (clusterSummary != null) {
- return getClusteredRowSignature(clusterSummary);
- }
-
- final LinkedHashSet<String> columns = new LinkedHashSet<>();
-
- for (final OrderBy orderBy : index.getOrdering()) {
- columns.add(orderBy.getColumnName());
- }
-
- // Add __time after the defined ordering, if __time wasn't part of it.
- columns.add(ColumnHolder.TIME_COLUMN_NAME);
- columns.addAll(index.getColumnNames());
-
- final RowSignature.Builder builder = RowSignature.builder();
- for (final String column : columns) {
- final ColumnType columnType =
ColumnType.fromCapabilities(index.getColumnCapabilities(column));
-
- // index.getOrdering() may include columns that don't exist, such as if
they were omitted due to
- // being 100% nulls. Don't add those to the row signature.
- if (columnType != null) {
- builder.add(column, columnType);
- }
- }
-
- return builder.build();
- }
-
- /**
- * Build the row signature for a clustered segment. Top-level columns are
empty, so column types are sourced from:
- * - the summary's clustering {@link RowSignature} for clustering columns;
- * - the first cluster group's sub-index for everything else (all groups
share the same data-column shape).
- */
- private RowSignature
getClusteredRowSignature(ClusteredValueGroupsBaseTableSchema clusterSummary)
- {
- final LinkedHashSet<String> columns = new LinkedHashSet<>();
-
- for (final OrderBy orderBy : clusterSummary.getOrdering()) {
- columns.add(orderBy.getColumnName());
- }
- columns.add(ColumnHolder.TIME_COLUMN_NAME);
- columns.addAll(clusterSummary.getColumnNames());
-
- final RowSignature.Builder builder = RowSignature.builder();
- for (final String column : columns) {
- final ColumnType columnType =
ColumnType.fromCapabilities(index.getColumnCapabilities(column));
- if (columnType != null) {
- builder.add(column, columnType);
- }
- }
- return builder.build();
- }
-
- @Nullable
- @Override
- public ColumnCapabilities getColumnCapabilities(String column)
- {
- return index.getColumnCapabilities(column);
- }
}
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
new file mode 100644
index 00000000000..b6224e8d0e7
--- /dev/null
+++
b/processing/src/main/java/org/apache/druid/segment/projections/MergingColumnSelectorFactory.java
@@ -0,0 +1,342 @@
+/*
+ * 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.error.DruidException;
+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.DimensionDictionarySelector;
+import org.apache.druid.segment.DimensionSelector;
+import org.apache.druid.segment.IdLookup;
+import org.apache.druid.segment.RowIdSupplier;
+import org.apache.druid.segment.column.ColumnCapabilities;
+import org.apache.druid.segment.column.ColumnCapabilitiesImpl;
+import org.apache.druid.segment.data.IndexedInts;
+
+import javax.annotation.Nullable;
+import java.util.function.IntSupplier;
+import java.util.function.LongSupplier;
+
+/**
+ * {@link ColumnSelectorFactory} for a time-ordered k-way merge across
per-cluster-group cursors (see
+ * {@code MergingClusterGroupCursor}). Unlike {@link
ClusteringColumnSelectorFactory}, which swaps a single delegate
+ * on each <em>group</em> transition and rebuilds selectors via a generation
counter, this factory pre-builds one
+ * 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>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.
+ */
+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;
+ // 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).
+ @Nullable
+ private final ColumnSelectorFactory representative;
+ // Row id minted from the merge's output-row counter (one per emitted row),
forwarded to callers for caching.
+ private final RowIdSupplier rowIdSupplier;
+
+ public MergingColumnSelectorFactory(
+ ColumnSelectorFactory[] groupFactories,
+ IntSupplier currentGroup,
+ LongSupplier currentRowId
+ )
+ {
+ this.groupFactories = groupFactories;
+ this.currentGroup = currentGroup;
+ this.representative = firstNonNull(groupFactories);
+ this.rowIdSupplier = currentRowId::getAsLong;
+ }
+
+ @Nullable
+ private static ColumnSelectorFactory firstNonNull(ColumnSelectorFactory[]
factories)
+ {
+ for (ColumnSelectorFactory factory : factories) {
+ if (factory != null) {
+ return factory;
+ }
+ }
+ return null;
+ }
+
+ /**
+ * Resolve the current winning group's per-group entry (selector/matcher).
{@code currentGroup} only ever points at
+ * a non-empty (hence non-null) group while the merge cursor is not done,
and the {@link org.apache.druid.segment.Cursor}
+ * contract requires callers to check {@code isDone()} before reading
selectors, so a null here means that contract
+ * was violated (a read past exhaustion). Fail fast rather than NPE opaquely.
+ */
+ private <T> T requireCurrent(T[] perGroup)
+ {
+ final int group = currentGroup.getAsInt();
+ final T current = perGroup[group];
+ if (current == null) {
+ throw DruidException.defensive(
+ "No entry for current cluster group [%s]; merge selectors must not
be read after isDone()",
+ group
+ );
+ }
+ return current;
+ }
+
+ @Override
+ public DimensionSelector makeDimensionSelector(DimensionSpec dimensionSpec)
+ {
+ final DimensionSelector[] perGroup = new
DimensionSelector[groupFactories.length];
+ for (int i = 0; i < groupFactories.length; i++) {
+ if (groupFactories[i] != null) {
+ perGroup[i] = groupFactories[i].makeDimensionSelector(dimensionSpec);
+ }
+ }
+ return new MergingDimensionSelector(perGroup, dimensionSpec);
+ }
+
+ @Override
+ public ColumnValueSelector makeColumnValueSelector(String 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);
+ }
+ }
+ return new MergingColumnValueSelector(perGroup, columnName);
+ }
+
+ @Nullable
+ @Override
+ public ColumnCapabilities getColumnCapabilities(String column)
+ {
+ if (representative == null) {
+ return null;
+ }
+ // Precondition: every cluster group shares one schema (the sub-indexes
are the same table split by clustering
+ // key), so the first non-null group is a valid stand-in for all groups'
capabilities of any given column.
+ final ColumnCapabilities capabilities =
representative.getColumnCapabilities(column);
+ if (capabilities == null) {
+ return null;
+ }
+ // Per-group-local dictionary ids are not stable across the merged stream
(the same id means different values in
+ // different groups), so advertise non-dictionary-encoded to force
value-based grouping, which is correct across
+ // groups.
+ return ColumnCapabilitiesImpl.copyOf(capabilities)
+ .setDictionaryEncoded(false)
+ .setDictionaryValuesSorted(false)
+ .setDictionaryValuesUnique(false)
+ .setHasBitmapIndexes(false);
+ }
+
+ @Nullable
+ @Override
+ public RowIdSupplier getRowIdSupplier()
+ {
+ return rowIdSupplier;
+ }
+
+ private final class MergingDimensionSelector implements DimensionSelector
+ {
+ private final DimensionSelector[] perGroup;
+ private final DimensionSpec spec;
+
+ private MergingDimensionSelector(DimensionSelector[] perGroup,
DimensionSpec spec)
+ {
+ this.perGroup = perGroup;
+ this.spec = spec;
+ }
+
+ private DimensionSelector current()
+ {
+ return requireCurrent(perGroup);
+ }
+
+ @Override
+ public IndexedInts getRow()
+ {
+ return current().getRow();
+ }
+
+ @Override
+ public ValueMatcher makeValueMatcher(@Nullable String value)
+ {
+ final ValueMatcher[] matchers = new ValueMatcher[perGroup.length];
+ for (int i = 0; i < perGroup.length; i++) {
+ if (perGroup[i] != null) {
+ matchers[i] = perGroup[i].makeValueMatcher(value);
+ }
+ }
+ return new MergingValueMatcher(matchers);
+ }
+
+ @Override
+ public ValueMatcher makeValueMatcher(DruidPredicateFactory
predicateFactory)
+ {
+ final ValueMatcher[] matchers = new ValueMatcher[perGroup.length];
+ for (int i = 0; i < perGroup.length; i++) {
+ if (perGroup[i] != null) {
+ matchers[i] = perGroup[i].makeValueMatcher(predicateFactory);
+ }
+ }
+ return new MergingValueMatcher(matchers);
+ }
+
+ @Override
+ public int getValueCardinality()
+ {
+ // Per-group dictionaries are not stable across the merged stream;
CARDINALITY_UNKNOWN forces value-based
+ // grouping (see class javadoc and ClusteringColumnSelectorFactory).
+ return DimensionDictionarySelector.CARDINALITY_UNKNOWN;
+ }
+
+ @Nullable
+ @Override
+ public String lookupName(int id)
+ {
+ return current().lookupName(id);
+ }
+
+ @Override
+ public boolean nameLookupPossibleInAdvance()
+ {
+ return false;
+ }
+
+ @Nullable
+ @Override
+ public IdLookup idLookup()
+ {
+ return null;
+ }
+
+ @Nullable
+ @Override
+ public Object getObject()
+ {
+ return current().getObject();
+ }
+
+ @Override
+ public Class<?> classOfObject()
+ {
+ return current().classOfObject();
+ }
+
+ @Override
+ public void inspectRuntimeShape(RuntimeShapeInspector inspector)
+ {
+ inspector.visit("merging", spec.getDimension());
+ }
+
+ /**
+ * Dispatches to the winning group's matcher per row. Pre-built per group
so a matcher held across the merge
+ * observes each row's winning group without rebuilding.
+ */
+ private final class MergingValueMatcher implements ValueMatcher
+ {
+ private final ValueMatcher[] matchers;
+
+ private MergingValueMatcher(ValueMatcher[] matchers)
+ {
+ this.matchers = matchers;
+ }
+
+ @Override
+ public boolean matches(boolean includeUnknown)
+ {
+ return requireCurrent(matchers).matches(includeUnknown);
+ }
+
+ @Override
+ public void inspectRuntimeShape(RuntimeShapeInspector inspector)
+ {
+ inspector.visit("merging-matcher", spec.getDimension());
+ }
+ }
+ }
+
+ private final class MergingColumnValueSelector implements
ColumnValueSelector<Object>
+ {
+ private final ColumnValueSelector[] perGroup;
+ private final String columnName;
+
+ private MergingColumnValueSelector(ColumnValueSelector[] perGroup, String
columnName)
+ {
+ this.perGroup = perGroup;
+ this.columnName = columnName;
+ }
+
+ private ColumnValueSelector current()
+ {
+ return requireCurrent(perGroup);
+ }
+
+ @Override
+ public double getDouble()
+ {
+ return current().getDouble();
+ }
+
+ @Override
+ public float getFloat()
+ {
+ return current().getFloat();
+ }
+
+ @Override
+ public long getLong()
+ {
+ return current().getLong();
+ }
+
+ @Override
+ public boolean isNull()
+ {
+ return current().isNull();
+ }
+
+ @Nullable
+ @Override
+ public Object getObject()
+ {
+ return current().getObject();
+ }
+
+ @Override
+ public Class<?> classOfObject()
+ {
+ return current().classOfObject();
+ }
+
+ @Override
+ public void inspectRuntimeShape(RuntimeShapeInspector inspector)
+ {
+ inspector.visit("merging", columnName);
+ }
+ }
+}
diff --git
a/processing/src/test/java/org/apache/druid/segment/ClusteredSegmentTimeOrderedQueryTest.java
b/processing/src/test/java/org/apache/druid/segment/ClusteredSegmentTimeOrderedQueryTest.java
new file mode 100644
index 00000000000..e3eb0b0778d
--- /dev/null
+++
b/processing/src/test/java/org/apache/druid/segment/ClusteredSegmentTimeOrderedQueryTest.java
@@ -0,0 +1,320 @@
+/*
+ * 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;
+
+import org.apache.druid.data.input.InputRow;
+import org.apache.druid.data.input.MapBasedInputRow;
+import
org.apache.druid.data.input.impl.ClusteredValueGroupsBaseTableProjectionSpec;
+import org.apache.druid.data.input.impl.DimensionsSpec;
+import org.apache.druid.data.input.impl.LongDimensionSchema;
+import org.apache.druid.data.input.impl.StringDimensionSchema;
+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.query.DefaultGenericQueryMetricsFactory;
+import org.apache.druid.query.Druids;
+import org.apache.druid.query.Order;
+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.scan.ScanQuery;
+import org.apache.druid.query.scan.ScanQueryConfig;
+import org.apache.druid.query.scan.ScanQueryEngine;
+import org.apache.druid.query.scan.ScanQueryQueryToolChest;
+import org.apache.druid.query.scan.ScanQueryRunnerFactory;
+import org.apache.druid.query.scan.ScanResultValue;
+import org.apache.druid.query.spec.MultipleIntervalSegmentSpec;
+import org.apache.druid.query.timeseries.TimeseriesQuery;
+import org.apache.druid.query.timeseries.TimeseriesQueryEngine;
+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.incremental.IncrementalIndexSchema;
+import
org.apache.druid.segment.writeout.OffHeapMemorySegmentWriteOutMediumFactory;
+import org.apache.druid.testing.InitializedNullHandlingTest;
+import org.apache.druid.timeline.SegmentId;
+import org.joda.time.Interval;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.io.File;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Functional coverage that the time-ordering query engines work over a
CLUSTERED base-table segment whose {@code
+ * __time} is the first non-clustering column (so the read path serves them
via the k-way {@code __time} merge; see
+ * {@link MergingClusterGroupCursor}). Runs a granular {@code timeseries} and
an ascending/descending native {@code
+ * scan} — scan hard-enforces cursor time ordering, so a passing time-ordered
scan over a multi-group clustered segment
+ * could not happen without the merge. Each result is also compared against an
equivalent non-clustered (naturally
+ * {@code __time}-sorted) segment to pin correctness.
+ */
+class ClusteredSegmentTimeOrderedQueryTest extends InitializedNullHandlingTest
+{
+ private static final String DATA_SOURCE = "clustered-time-ordered";
+ private static final TimestampSpec TIMESTAMP_SPEC = new TimestampSpec("ts",
"millis", null);
+ private static final long T0 =
DateTimes.of("2026-01-01T00:00:00.000Z").getMillis();
+ private static final long MINUTE = 60_000L;
+ private static final Interval INTERVAL = Intervals.utc(T0, T0 + 4 * MINUTE);
+
+ // Two tenants => two cluster groups; timestamps interleave across the
groups, ingested out of order. So the
+ // clustering-first concatenation is NOT globally time-ordered, and the
merge must interleave by __time.
+ private static final List<InputRow> ROWS = List.of(
+ row(T0, "acme", 1),
+ row(T0 + 2 * MINUTE, "acme", 4),
+ row(T0 + MINUTE, "globex", 2),
+ row(T0 + 3 * MINUTE, "globex", 8)
+ );
+
+ // A single tenant => a single cluster group, so queries take the
single-group clustered path.
+ private static final List<InputRow> SINGLE_GROUP_ROWS = List.of(
+ row(T0, "acme", 1),
+ row(T0 + MINUTE, "acme", 2),
+ row(T0 + 2 * MINUTE, "acme", 4)
+ );
+
+ @TempDir
+ File tempDir;
+
+ private Segment clusteredSegment;
+ private Segment nonClusteredSegment;
+
+ @BeforeEach
+ void setUp()
+ {
+ clusteredSegment = new QueryableIndexSegment(buildClustered("clustered",
ROWS), SegmentId.dummy(DATA_SOURCE));
+ nonClusteredSegment = new QueryableIndexSegment(buildNonClustered("plain",
ROWS), SegmentId.dummy(DATA_SOURCE));
+ }
+
+ @Test
+ void testGranularTimeseriesOverClusteredSegment()
+ {
+ // MINUTE-granular sum requires a globally __time-ordered cursor (the
granularizer advances buckets as time
+ // increases); each bucket here comes from a different group, so the merge
is what makes it correct.
+ final List<List<Long>> expected = List.of(
+ List.of(T0, 1L),
+ List.of(T0 + MINUTE, 2L),
+ List.of(T0 + 2 * MINUTE, 4L),
+ List.of(T0 + 3 * MINUTE, 8L)
+ );
+ Assertions.assertEquals(expected, runTimeseries(clusteredSegment));
+ // ... and identical to the equivalent non-clustered, naturally
time-sorted segment.
+ Assertions.assertEquals(runTimeseries(nonClusteredSegment),
runTimeseries(clusteredSegment));
+ }
+
+ @Test
+ void testAscendingScanOverClusteredSegment()
+ {
+ 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, runScanRows(clusteredSegment,
Order.ASCENDING));
+ Assertions.assertEquals(
+ runScanRows(nonClusteredSegment, Order.ASCENDING),
+ runScanRows(clusteredSegment, Order.ASCENDING)
+ );
+ }
+
+ @Test
+ void testDescendingScanOverClusteredSegment()
+ {
+ final List<List<Object>> expected = 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)
+ );
+ Assertions.assertEquals(expected, runScanRows(clusteredSegment,
Order.DESCENDING));
+ Assertions.assertEquals(
+ runScanRows(nonClusteredSegment, Order.DESCENDING),
+ runScanRows(clusteredSegment, Order.DESCENDING)
+ );
+ }
+
+ @Test
+ void testAscendingScanOverSingleGroupClusteredSegment()
+ {
+ // A clustered segment that resolves to a SINGLE cluster group must also
serve time-ordered scans. Previously the
+ // single-group path advertised the clustering-first ordering
(getTimeOrder()==NONE) and the scan engine rejected
+ // it, so the same query succeeded with >=2 groups but failed with one.
Scan throws unless the cursor is
+ // time-ordered, so this passing is the proof the single-group path now
reports __time ordering.
+ final Segment clustered =
+ new QueryableIndexSegment(buildClustered("clustered-single",
SINGLE_GROUP_ROWS), SegmentId.dummy(DATA_SOURCE));
+ final Segment plain =
+ new QueryableIndexSegment(buildNonClustered("plain-single",
SINGLE_GROUP_ROWS), SegmentId.dummy(DATA_SOURCE));
+ final List<List<Object>> expected = List.of(
+ Arrays.asList(T0, "acme", 1L),
+ Arrays.asList(T0 + MINUTE, "acme", 2L),
+ Arrays.asList(T0 + 2 * MINUTE, "acme", 4L)
+ );
+ Assertions.assertEquals(expected, runScanRows(clustered, Order.ASCENDING));
+ Assertions.assertEquals(runScanRows(plain, Order.ASCENDING),
runScanRows(clustered, Order.ASCENDING));
+ }
+
+ @Test
+ void testDescendingScanOverSingleGroupClusteredSegment()
+ {
+ final Segment clustered = new QueryableIndexSegment(
+ buildClustered("clustered-single-desc", SINGLE_GROUP_ROWS),
+ SegmentId.dummy(DATA_SOURCE)
+ );
+ final Segment plain = new QueryableIndexSegment(
+ buildNonClustered("plain-single-desc", SINGLE_GROUP_ROWS),
+ SegmentId.dummy(DATA_SOURCE)
+ );
+ final List<List<Object>> expected = List.of(
+ Arrays.asList(T0 + 2 * MINUTE, "acme", 4L),
+ Arrays.asList(T0 + MINUTE, "acme", 2L),
+ Arrays.asList(T0, "acme", 1L)
+ );
+ Assertions.assertEquals(expected, runScanRows(clustered,
Order.DESCENDING));
+ Assertions.assertEquals(runScanRows(plain, Order.DESCENDING),
runScanRows(clustered, Order.DESCENDING));
+ }
+
+ private static List<List<Long>> runTimeseries(Segment segment)
+ {
+ final TimeseriesQuery query = Druids.newTimeseriesQueryBuilder()
+ .dataSource(DATA_SOURCE)
+ .granularity(Granularities.MINUTE)
+ .intervals(new
MultipleIntervalSegmentSpec(List.of(INTERVAL)))
+ .aggregators(new
LongSumAggregatorFactory("sum_m", "m"))
+ .build();
+ final TimeseriesQueryRunnerFactory factory = new
TimeseriesQueryRunnerFactory(
+ new TimeseriesQueryQueryToolChest(),
+ new TimeseriesQueryEngine(),
+ QueryRunnerTestHelper.NOOP_QUERYWATCHER
+ );
+ final List<Result<TimeseriesResultValue>> results =
+ factory.createRunner(segment).run(QueryPlus.wrap(query)).toList();
+ final List<List<Long>> out = new ArrayList<>();
+ for (Result<TimeseriesResultValue> result : results) {
+ final long sum = ((Number)
result.getValue().getMetric("sum_m")).longValue();
+ out.add(List.of(result.getTimestamp().getMillis(), sum));
+ }
+ return out;
+ }
+
+ private static List<List<Object>> runScanRows(Segment segment, Order order)
+ {
+ final ScanQuery query = Druids.newScanQueryBuilder()
+ .dataSource(DATA_SOURCE)
+ .intervals(new
MultipleIntervalSegmentSpec(List.of(INTERVAL)))
+ .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();
+ // Extract full [__time, tenant, m] rows so the comparison against the
non-clustered baseline verifies per-column
+ // dispatch (winning group's tenant/m paired with the right __time), not
just the time column.
+ 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 buildClustered(String dirName, List<InputRow> rows)
+ {
+ // columns [tenant, __time, m], clustering [tenant] => ordering [tenant,
__time, m], so __time is the first
+ // non-clustering column and each cluster group is individually
__time-sorted.
+ 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();
+ return IndexBuilder.create()
+ .useV10()
+ .tmpDir(new File(tempDir, dirName))
+
.segmentWriteOutMediumFactory(OffHeapMemorySegmentWriteOutMediumFactory.instance())
+ .schema(schema)
+ .rows(rows)
+ .buildMMappedIndex();
+ }
+
+ private QueryableIndex buildNonClustered(String dirName, List<InputRow> rows)
+ {
+ final IncrementalIndexSchema schema =
+ IncrementalIndexSchema.builder()
+ .withMinTimestamp(T0)
+ .withTimestampSpec(TIMESTAMP_SPEC)
+ .withQueryGranularity(Granularities.NONE)
+ .withDimensionsSpec(
+ DimensionsSpec.builder()
+ .setDimensions(List.of(
+ new
StringDimensionSchema("tenant"),
+ new
LongDimensionSchema("m")
+ ))
+ .build()
+ )
+ .withRollup(false)
+ .build();
+ return IndexBuilder.create()
+ .tmpDir(new File(tempDir, dirName))
+ .schema(schema)
+ .rows(rows)
+ .buildMMappedIndex();
+ }
+
+ private static InputRow row(long ts, String tenant, long m)
+ {
+ final Map<String, Object> event = new HashMap<>();
+ event.put("ts", ts);
+ event.put("tenant", tenant);
+ event.put("m", m);
+ return new MapBasedInputRow(ts, List.of("tenant", "m"), event);
+ }
+}
diff --git
a/processing/src/test/java/org/apache/druid/segment/IndexMergerV10ClusteredTest.java
b/processing/src/test/java/org/apache/druid/segment/IndexMergerV10ClusteredTest.java
index d19d174d812..93eaf89d0ed 100644
---
a/processing/src/test/java/org/apache/druid/segment/IndexMergerV10ClusteredTest.java
+++
b/processing/src/test/java/org/apache/druid/segment/IndexMergerV10ClusteredTest.java
@@ -29,6 +29,8 @@ 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.query.Order;
+import org.apache.druid.query.OrderBy;
import org.apache.druid.query.aggregation.AggregatorFactory;
import org.apache.druid.query.aggregation.CountAggregatorFactory;
import org.apache.druid.query.aggregation.LongSumAggregatorFactory;
@@ -36,6 +38,7 @@ import org.apache.druid.query.dimension.DefaultDimensionSpec;
import org.apache.druid.query.groupby.GroupByQuery;
import org.apache.druid.query.groupby.GroupingEngine;
import org.apache.druid.query.groupby.orderby.OrderByColumnSpec.Direction;
+import org.apache.druid.segment.column.ColumnHolder;
import org.apache.druid.segment.incremental.IncrementalIndexSchema;
import
org.apache.druid.segment.projections.ClusteredValueGroupsBaseTableSchema;
import org.apache.druid.segment.projections.TableClusterGroupSpec;
@@ -139,6 +142,51 @@ class IndexMergerV10ClusteredTest extends
InitializedNullHandlingTest
return out;
}
+ private static ClusteredValueGroupsBaseTableProjectionSpec
timeOrderedClusterSpec()
+ {
+ // __time declared as the first non-clustering column, so the segment
ordering is [tenant, __time, region] and each
+ // cluster group is individually __time-sorted -- the precondition for the
time-ordered merge cursor.
+ return ClusteredValueGroupsBaseTableProjectionSpec.builder()
+ .columns(
+ new StringDimensionSchema("tenant"),
+ new LongDimensionSchema("__time"),
+ new StringDimensionSchema("region")
+ )
+ .clusteringColumns("tenant")
+ .build();
+ }
+
+ private QueryableIndex buildTimeOrderedSegment(String dirName,
List<InputRow> rows)
+ {
+ return IndexBuilder.create()
+ .useV10()
+ .tmpDir(new File(tempDir, dirName))
+
.segmentWriteOutMediumFactory(OffHeapMemorySegmentWriteOutMediumFactory.instance())
+ .schema(clusteredSchema(timeOrderedClusterSpec()))
+ .rows(rows)
+ .buildMMappedIndex();
+ }
+
+ /**
+ * Walk (@code __time}, {@code region}) pairs from a holder's scalar cursor,
in whatever order the holder produces.
+ */
+ private static List<List<Object>> scanTimeRegion(CursorHolder holder)
+ {
+ final Cursor cursor = holder.asCursor();
+ final ColumnValueSelector timeSelector =
+
cursor.getColumnSelectorFactory().makeColumnValueSelector(ColumnHolder.TIME_COLUMN_NAME);
+ final DimensionSelector regionSelector =
+
cursor.getColumnSelectorFactory().makeDimensionSelector(DefaultDimensionSpec.of("region"));
+ final List<List<Object>> out = new ArrayList<>();
+ while (!cursor.isDone()) {
+ final String region =
+ regionSelector.getRow().size() == 0 ? null :
regionSelector.lookupName(regionSelector.getRow().get(0));
+ out.add(Arrays.asList(timeSelector.getLong(), region));
+ cursor.advance();
+ }
+ return out;
+ }
+
/**
* Cluster spec with a numeric column {@code x} (in addition to {@code
tenant}/{@code region}) so an aggregate
* projection can sum it. Clustering is still on {@code tenant}.
@@ -453,6 +501,86 @@ class IndexMergerV10ClusteredTest extends
InitializedNullHandlingTest
);
}
+ @Test
+ void testTimeOrderedMergeAcrossGroups()
+ {
+ // Two tenants (=> two cluster groups) with interleaved timestamps,
ingested out of order.
+ final QueryableIndex index = buildTimeOrderedSegment(
+ "time-merge",
+ List.of(
+ row(T0 + 4, "acme", "a4"),
+ row(T0 + 1, "acme", "a1"),
+ row(T0 + 3, "globex", "g3"),
+ row(T0 + 2, "globex", "g2")
+ )
+ );
+
+ // Layout precondition: __time is the first non-clustering column, so each
group is individually __time-sorted.
+ final ClusteredValueGroupsBaseTableSchema summary =
index.getClusteredBaseSummary();
+ Assertions.assertEquals(
+ ColumnHolder.TIME_COLUMN_NAME,
+ summary.getGroupOrdering().get(0).getColumnName()
+ );
+
+ final QueryableIndexCursorFactory factory = new
QueryableIndexCursorFactory(
+ index,
+ QueryableIndexTimeBoundaryInspector.create(index)
+ );
+
+ // No preferred ordering => concatenation: clustering-first ordering, NOT
globally __time-ordered (each group's
+ // rows are contiguous: acme's two rows, then globex's two rows).
+ try (CursorHolder holder =
factory.makeCursorHolder(CursorBuildSpec.FULL_SCAN)) {
+ Assertions.assertEquals("tenant",
holder.getOrdering().get(0).getColumnName());
+ Assertions.assertEquals(
+ List.of(
+ Arrays.asList(T0 + 1, "a1"),
+ Arrays.asList(T0 + 4, "a4"),
+ Arrays.asList(T0 + 2, "g2"),
+ Arrays.asList(T0 + 3, "g3")
+ ),
+ scanTimeRegion(holder)
+ );
+ }
+
+ // Ascending __time preferred => the time-ordered merge cursor: holder
advertises __time ASC and rows are globally
+ // time-ordered across groups.
+ final CursorBuildSpec ascending =
CursorBuildSpec.builder(CursorBuildSpec.FULL_SCAN)
+
.setPreferredOrdering(List.of(OrderBy.ascending(ColumnHolder.TIME_COLUMN_NAME)))
+ .build();
+ try (CursorHolder holder = factory.makeCursorHolder(ascending)) {
+ Assertions.assertEquals(Cursors.ascendingTimeOrder(),
holder.getOrdering());
+ Assertions.assertEquals(Order.ASCENDING, holder.getTimeOrder());
+ Assertions.assertFalse(holder.canVectorize(), "scalar-first: the merge
holder is not vectorizable");
+ Assertions.assertEquals(
+ List.of(
+ Arrays.asList(T0 + 1, "a1"),
+ Arrays.asList(T0 + 2, "g2"),
+ Arrays.asList(T0 + 3, "g3"),
+ Arrays.asList(T0 + 4, "a4")
+ ),
+ scanTimeRegion(holder)
+ );
+ }
+
+ // Descending __time preferred => descending merge (each per-group cursor
reverses, merged with a max-heap).
+ final CursorBuildSpec descending =
CursorBuildSpec.builder(CursorBuildSpec.FULL_SCAN)
+
.setPreferredOrdering(List.of(OrderBy.descending(ColumnHolder.TIME_COLUMN_NAME)))
+ .build();
+ try (CursorHolder holder = factory.makeCursorHolder(descending)) {
+ Assertions.assertEquals(Cursors.descendingTimeOrder(),
holder.getOrdering());
+ Assertions.assertEquals(Order.DESCENDING, holder.getTimeOrder());
+ Assertions.assertEquals(
+ List.of(
+ Arrays.asList(T0 + 4, "a4"),
+ Arrays.asList(T0 + 3, "g3"),
+ Arrays.asList(T0 + 2, "g2"),
+ Arrays.asList(T0 + 1, "a1")
+ ),
+ scanTimeRegion(holder)
+ );
+ }
+ }
+
@Test
void testMergeClusteredSegmentsAlignsGroupsAcrossSegments() throws Exception
{
diff --git
a/processing/src/test/java/org/apache/druid/segment/MergingClusterGroupCursorTest.java
b/processing/src/test/java/org/apache/druid/segment/MergingClusterGroupCursorTest.java
new file mode 100644
index 00000000000..daf38d3a2cc
--- /dev/null
+++
b/processing/src/test/java/org/apache/druid/segment/MergingClusterGroupCursorTest.java
@@ -0,0 +1,302 @@
+/*
+ * 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;
+
+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.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import javax.annotation.Nullable;
+import java.util.ArrayList;
+import java.util.List;
+
+class MergingClusterGroupCursorTest
+{
+ @Test
+ void testAscendingMergeAcrossGroups()
+ {
+ // Three individually time-sorted groups; the merge interleaves them into
one globally ascending stream, and each
+ // emitted value must come from the group that owns that timestamp
(verifies per-row dispatch to the winner).
+ final MergingClusterGroupCursor cursor = cursor(
+ false,
+ group(new long[]{1, 5, 9}, new String[]{"g0@1", "g0@5", "g0@9"}),
+ group(new long[]{2, 3, 10}, new String[]{"g1@2", "g1@3", "g1@10"}),
+ group(new long[]{4, 6}, new String[]{"g2@4", "g2@6"})
+ );
+ final List<Object[]> rows = drain(cursor);
+ assertRows(
+ rows,
+ new long[]{1, 2, 3, 4, 5, 6, 9, 10},
+ new String[]{"g0@1", "g1@2", "g1@3", "g2@4", "g0@5", "g2@6", "g0@9",
"g1@10"}
+ );
+ }
+
+ @Test
+ void testDescendingMerge()
+ {
+ // Descending: each per-group cursor is itself descending (mirrors
QueryableIndexCursorHolder reversing offsets),
+ // and the merge uses a max-heap.
+ final MergingClusterGroupCursor cursor = cursor(
+ true,
+ group(new long[]{9, 5, 1}, new String[]{"g0@9", "g0@5", "g0@1"}),
+ group(new long[]{10, 3, 2}, new String[]{"g1@10", "g1@3", "g1@2"}),
+ group(new long[]{6, 4}, new String[]{"g2@6", "g2@4"})
+ );
+ final List<Object[]> rows = drain(cursor);
+ assertRows(
+ rows,
+ new long[]{10, 9, 6, 5, 4, 3, 2, 1},
+ new String[]{"g1@10", "g0@9", "g2@6", "g0@5", "g2@4", "g1@3", "g1@2",
"g0@1"}
+ );
+ }
+
+ @Test
+ void testTiesBreakByGroupIndex()
+ {
+ // Equal timestamps across groups break by ascending group index
(deterministic). Within a group, rows are emitted
+ // in the group's own order.
+ final MergingClusterGroupCursor cursor = cursor(
+ false,
+ group(new long[]{5, 5}, new String[]{"g0a", "g0b"}),
+ group(new long[]{5}, new String[]{"g1a"})
+ );
+ final List<Object[]> rows = drain(cursor);
+ assertRows(rows, new long[]{5, 5, 5}, new String[]{"g0a", "g0b", "g1a"});
+ }
+
+ @Test
+ void testEmptyGroupsSkipped()
+ {
+ final MergingClusterGroupCursor cursor = cursor(
+ false,
+ group(new long[]{}, new String[]{}),
+ group(new long[]{1, 2}, new String[]{"x", "y"}),
+ group(new long[]{}, new String[]{})
+ );
+ assertRows(drain(cursor), new long[]{1, 2}, new String[]{"x", "y"});
+ }
+
+ @Test
+ void testSingleNonEmptyGroup()
+ {
+ final MergingClusterGroupCursor cursor = cursor(
+ false,
+ group(new long[]{}, new String[]{}),
+ group(new long[]{7, 8, 9}, new String[]{"a", "b", "c"})
+ );
+ assertRows(drain(cursor), new long[]{7, 8, 9}, new String[]{"a", "b",
"c"});
+ }
+
+ @Test
+ void testAllEmptyIsDoneImmediately()
+ {
+ final MergingClusterGroupCursor cursor = cursor(
+ false,
+ group(new long[]{}, new String[]{}),
+ group(new long[]{}, new String[]{})
+ );
+ Assertions.assertTrue(cursor.isDone());
+ }
+
+ @Test
+ void testResetReplaysSameSequence()
+ {
+ final MergingClusterGroupCursor cursor = cursor(
+ false,
+ group(new long[]{1, 4}, new String[]{"g0@1", "g0@4"}),
+ group(new long[]{2, 3}, new String[]{"g1@2", "g1@3"})
+ );
+ final List<Object[]> first = drain(cursor);
+ cursor.reset();
+ final List<Object[]> second = drain(cursor);
+ assertRows(first, new long[]{1, 2, 3, 4}, new String[]{"g0@1", "g1@2",
"g1@3", "g0@4"});
+ assertRows(second, new long[]{1, 2, 3, 4}, new String[]{"g0@1", "g1@2",
"g1@3", "g0@4"});
+ }
+
+ @Test
+ void testRowIdMonotonicAndStableWithinRow()
+ {
+ final MergingClusterGroupCursor cursor = cursor(
+ false,
+ group(new long[]{1, 3}, new String[]{"g0@1", "g0@3"}),
+ group(new long[]{2, 4}, new String[]{"g1@2", "g1@4"})
+ );
+ final RowIdSupplier rowIdSupplier =
cursor.getColumnSelectorFactory().getRowIdSupplier();
+ Assertions.assertNotNull(rowIdSupplier);
+ long previous = -1;
+ while (!cursor.isDone()) {
+ final long id = rowIdSupplier.getRowId();
+ Assertions.assertEquals(id, rowIdSupplier.getRowId(), "row id must be
stable within a row");
+ Assertions.assertTrue(id > previous, "row id must strictly increase
across rows");
+ previous = id;
+ cursor.advance();
+ }
+ }
+
+ @SafeVarargs
+ private static MergingClusterGroupCursor cursor(boolean descending,
Supplier<CursorHolder>... groups)
+ {
+ return new MergingClusterGroupCursor(new ArrayList<>(List.of(groups)),
descending);
+ }
+
+ private static List<Object[]> drain(MergingClusterGroupCursor cursor)
+ {
+ final ColumnSelectorFactory factory = cursor.getColumnSelectorFactory();
+ final ColumnValueSelector timeSelector =
factory.makeColumnValueSelector(ColumnHolder.TIME_COLUMN_NAME);
+ final ColumnValueSelector valueSelector =
factory.makeColumnValueSelector("v");
+ final List<Object[]> rows = new ArrayList<>();
+ while (!cursor.isDone()) {
+ rows.add(new Object[]{timeSelector.getLong(),
valueSelector.getObject()});
+ cursor.advance();
+ }
+ return rows;
+ }
+
+ private static void assertRows(List<Object[]> rows, long[] expectedTimes,
String[] expectedValues)
+ {
+ Assertions.assertEquals(expectedTimes.length, rows.size(), "row count");
+ for (int i = 0; i < expectedTimes.length; i++) {
+ Assertions.assertEquals(expectedTimes[i], (long) (Long) rows.get(i)[0],
"time at row " + i);
+ Assertions.assertEquals(expectedValues[i], rows.get(i)[1], "value at row
" + i);
+ }
+ }
+
+ private static Supplier<CursorHolder> group(long[] times, String[] values)
+ {
+ final Cursor cursor = new ListCursor(times, values);
+ final CursorHolder holder = new CursorHolder()
+ {
+ @Override
+ public Cursor asCursor()
+ {
+ return cursor;
+ }
+ };
+ return () -> holder;
+ }
+
+ /**
+ * A minimal scalar {@link Cursor} over an in-memory {@code (times, values)}
pair, exposing {@code __time} (long) and
+ * {@code v} (object) selectors backed by the current position. Stands in
for a per-group cluster cursor.
+ */
+ private static final class ListCursor implements Cursor
+ {
+ private final long[] times;
+ private final String[] values;
+ private final ColumnSelectorFactory factory;
+ private int pos;
+
+ private ListCursor(long[] times, String[] values)
+ {
+ this.times = times;
+ this.values = values;
+ this.factory = new ColumnSelectorFactory()
+ {
+ @Override
+ public DimensionSelector makeDimensionSelector(DimensionSpec
dimensionSpec)
+ {
+ throw new UnsupportedOperationException("not used");
+ }
+
+ @Override
+ public ColumnValueSelector makeColumnValueSelector(String columnName)
+ {
+ if (ColumnHolder.TIME_COLUMN_NAME.equals(columnName)) {
+ return new TestLongColumnSelector()
+ {
+ @Override
+ public long getLong()
+ {
+ return times[pos];
+ }
+
+ @Override
+ public boolean isNull()
+ {
+ return false;
+ }
+ };
+ }
+ return new TestObjectColumnSelector<String>()
+ {
+ @Override
+ public Class<String> classOfObject()
+ {
+ return String.class;
+ }
+
+ @Nullable
+ @Override
+ public String getObject()
+ {
+ return values[pos];
+ }
+ };
+ }
+
+ @Nullable
+ @Override
+ public ColumnCapabilities getColumnCapabilities(String column)
+ {
+ return null;
+ }
+ };
+ }
+
+ @Override
+ public ColumnSelectorFactory getColumnSelectorFactory()
+ {
+ return factory;
+ }
+
+ @Override
+ public void advance()
+ {
+ pos++;
+ }
+
+ @Override
+ public void advanceUninterruptibly()
+ {
+ pos++;
+ }
+
+ @Override
+ public boolean isDone()
+ {
+ return pos >= times.length;
+ }
+
+ @Override
+ public boolean isDoneOrInterrupted()
+ {
+ return isDone();
+ }
+
+ @Override
+ public void reset()
+ {
+ pos = 0;
+ }
+ }
+}
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
new file mode 100644
index 00000000000..a8ea723a979
--- /dev/null
+++
b/processing/src/test/java/org/apache/druid/segment/projections/MergingColumnSelectorFactoryTest.java
@@ -0,0 +1,170 @@
+/*
+ * 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.dimension.DefaultDimensionSpec;
+import org.apache.druid.query.dimension.DimensionSpec;
+import org.apache.druid.segment.ColumnSelectorFactory;
+import org.apache.druid.segment.ColumnValueSelector;
+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.column.ColumnCapabilities;
+import org.apache.druid.segment.column.ColumnCapabilitiesImpl;
+import org.apache.druid.segment.column.ValueType;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import javax.annotation.Nullable;
+import java.util.function.IntSupplier;
+import java.util.function.LongSupplier;
+
+class MergingColumnSelectorFactoryTest
+{
+ private final int[] currentGroup = {0};
+ private final long[] rowId = {0};
+
+ private MergingColumnSelectorFactory factory(ColumnSelectorFactory... groups)
+ {
+ final IntSupplier currentGroupSupplier = () -> currentGroup[0];
+ final LongSupplier rowIdSupplier = () -> rowId[0];
+ return new MergingColumnSelectorFactory(groups, currentGroupSupplier,
rowIdSupplier);
+ }
+
+ @Test
+ void testGetColumnCapabilitiesStripsDictionaryEncoding()
+ {
+ // The representative group advertises "v" as a dictionary-encoded string
with bitmap indexes; because per-group
+ // dictionary ids are not stable across the merged stream, the factory
must strip those flags (forcing value-based
+ // grouping) while keeping the type.
+ final MergingColumnSelectorFactory factory = factory(groupFactory("g0"),
groupFactory("g1"));
+ final ColumnCapabilities caps = factory.getColumnCapabilities("v");
+ Assertions.assertNotNull(caps);
+ Assertions.assertTrue(caps.is(ValueType.STRING));
+ Assertions.assertTrue(caps.isDictionaryEncoded().isFalse());
+ Assertions.assertTrue(caps.areDictionaryValuesSorted().isFalse());
+ Assertions.assertTrue(caps.areDictionaryValuesUnique().isFalse());
+ Assertions.assertFalse(caps.hasBitmapIndexes());
+ }
+
+ @Test
+ void testGetColumnCapabilitiesNullWhenRepresentativeHasNone()
+ {
+ // Unknown column -> representative returns null -> factory returns null
(caller skips it).
+ final MergingColumnSelectorFactory factory = factory(groupFactory("g0"));
+ Assertions.assertNull(factory.getColumnCapabilities("nope"));
+ // No non-null group factory at all -> null capabilities.
+ Assertions.assertNull(factory(new
ColumnSelectorFactory[]{null}).getColumnCapabilities("v"));
+ }
+
+ @Test
+ void testColumnValueSelectorDispatchesToCurrentGroup()
+ {
+ final MergingColumnSelectorFactory factory = factory(groupFactory("g0"),
groupFactory("g1"));
+ final ColumnValueSelector selector = factory.makeColumnValueSelector("v");
+ currentGroup[0] = 0;
+ Assertions.assertEquals("g0", selector.getObject());
+ currentGroup[0] = 1;
+ Assertions.assertEquals("g1", selector.getObject());
+ currentGroup[0] = 0;
+ Assertions.assertEquals("g0", selector.getObject());
+ }
+
+ @Test
+ void testDimensionSelectorDispatchesAndReportsUnstableDictionary()
+ {
+ final MergingColumnSelectorFactory factory = factory(groupFactory("g0"),
groupFactory("g1"));
+ final DimensionSelector selector =
factory.makeDimensionSelector(DefaultDimensionSpec.of("v"));
+
+ currentGroup[0] = 0;
+ Assertions.assertEquals("g0", selector.getObject());
+ Assertions.assertEquals("g0",
selector.lookupName(selector.getRow().get(0)));
+ currentGroup[0] = 1;
+ Assertions.assertEquals("g1", selector.getObject());
+
+ // Cross-group dictionary instability is advertised so engines use
value-based grouping.
+ Assertions.assertEquals(DimensionDictionarySelector.CARDINALITY_UNKNOWN,
selector.getValueCardinality());
+ Assertions.assertNull(selector.idLookup());
+ Assertions.assertFalse(selector.nameLookupPossibleInAdvance());
+ }
+
+ @Test
+ void testRowIdSupplierReflectsMintedOutputRowId()
+ {
+ final MergingColumnSelectorFactory factory = factory(groupFactory("g0"));
+ final RowIdSupplier supplier = factory.getRowIdSupplier();
+ Assertions.assertNotNull(supplier);
+ rowId[0] = 0;
+ Assertions.assertEquals(0L, supplier.getRowId());
+ rowId[0] = 7;
+ Assertions.assertEquals(7L, supplier.getRowId());
+ }
+
+ /**
+ * A stub group factory that reports column "v" as a dictionary-encoded
string (to exercise stripping) and returns
+ * constant selectors carrying {@code tag} (to verify per-group dispatch).
+ */
+ private static ColumnSelectorFactory groupFactory(String tag)
+ {
+ return new ColumnSelectorFactory()
+ {
+ @Override
+ public DimensionSelector makeDimensionSelector(DimensionSpec
dimensionSpec)
+ {
+ return DimensionSelector.constant(tag);
+ }
+
+ @Override
+ public ColumnValueSelector makeColumnValueSelector(String columnName)
+ {
+ return new TestObjectColumnSelector<String>()
+ {
+ @Override
+ public Class<String> classOfObject()
+ {
+ return String.class;
+ }
+
+ @Nullable
+ @Override
+ public String getObject()
+ {
+ return tag;
+ }
+ };
+ }
+
+ @Nullable
+ @Override
+ public ColumnCapabilities getColumnCapabilities(String column)
+ {
+ if (!"v".equals(column)) {
+ return null;
+ }
+ return
ColumnCapabilitiesImpl.createSimpleSingleValueStringColumnCapabilities()
+ .setDictionaryEncoded(true)
+ .setDictionaryValuesSorted(true)
+ .setDictionaryValuesUnique(true)
+ .setHasBitmapIndexes(true);
+ }
+ };
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
