gianm commented on code in PR #19726: URL: https://github.com/apache/druid/pull/19726#discussion_r3636357628
########## 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 Review Comment: Should this be in `org.apache.druid.segment.projections` too? Also no reason to be `final`. ########## processing/src/main/java/org/apache/druid/segment/QueryableIndexCursorFactory.java: ########## @@ -200,15 +201,49 @@ private CursorHolder makeSingleGroupClusteredCursorHolder( ); } + // A single cluster group is physically sorted by its group ordering (the segment ordering with the constant Review Comment: This comment is too long for what needs to be communicated. Try something like: ```java // Omit cluster key from ordering if the caller requests (and is granted) time ordering. // This way, the returned ordering will begin with {@code __time}. ``` ########## processing/src/main/java/org/apache/druid/segment/QueryableIndexCursorFactory.java: ########## @@ -331,6 +379,49 @@ public void close() }; } + /** + * Builds a {@link CursorHolder} whose scalar 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). Scalar only for now: {@code + * canVectorize()} and {@code asVectorCursor()} keep their {@link CursorHolder} defaults (false / null), so a + * vectorizing engine falls back to the scalar path. + */ + 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() + { + try { Review Comment: `CloseableUtils.closeAndWrapExceptions(closer)` ########## processing/src/main/java/org/apache/druid/segment/QueryableIndexCursorFactory.java: ########## @@ -331,6 +379,49 @@ public void close() }; } + /** + * Builds a {@link CursorHolder} whose scalar cursor is a globally {@code __time}-ordered {@link Review Comment: I don't usually see "scalar" used to refer to nonvectorized execution. Usually "nonvectorized" or maybe "row-at-a-time". -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
