You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@druid.apache.org by "clintropolis (via GitHub)" <gi...@apache.org> on 2024/09/19 20:13:24 UTC
[PR] projections part 1 - realtime segments (druid)
clintropolis opened a new pull request, #17117:
URL: https://github.com/apache/druid/pull/17117
### Description
This PR begins to introduce the concept of projections to Druid datasources, which are similar to materialized views but are built into a segment, and which can automatically be used during query execution if the projection fits the query. This PR only contains the logic to build and query them for realtime queries, and does not contain the ability to serialize and actually store them in persisted segments, so it is effectively a toy right now.
changes:
* Adds ProjectionSpec interface, AggregateProjectionSpec implementation for defining rollup projections on druid datasources
* Adds projections to DataSchema
* Adds projection building and querying support to OnHeapIncrementalIndex
<hr>
This PR has:
- [ ] been self-reviewed.
- [ ] using the [concurrency checklist](https://github.com/apache/druid/blob/master/dev/code-review/concurrency.md) (Remove this item if the PR doesn't have any relation to concurrency.)
- [ ] added documentation for new or modified features or behaviors.
- [ ] a release note entry in the PR description.
- [ ] added Javadocs for most classes and all non-trivial methods. Linked related entities via Javadoc links.
- [ ] added or updated version, license, or notice information in [licenses.yaml](https://github.com/apache/druid/blob/master/dev/license.md)
- [ ] added comments explaining the "why" and the intent of the code wherever would not be obvious for an unfamiliar reader.
- [ ] added unit tests or modified existing tests to cover new code paths, ensuring the threshold for [code coverage](https://github.com/apache/druid/blob/master/dev/code-review/code-coverage.md) is met.
- [ ] added integration tests.
- [ ] been tested in a test Druid cluster.
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] projections part 1 - realtime segments (druid)
Posted by "clintropolis (via GitHub)" <gi...@apache.org>.
clintropolis commented on code in PR #17117:
URL: https://github.com/apache/druid/pull/17117#discussion_r1771914548
##########
processing/src/main/java/org/apache/druid/data/input/ProjectionSpec.java:
##########
@@ -0,0 +1,41 @@
+/*
+ * 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.data.input;
+
+import com.fasterxml.jackson.annotation.JsonSubTypes;
+import com.fasterxml.jackson.annotation.JsonTypeInfo;
+import org.apache.druid.data.input.impl.AggregateProjectionSpec;
+import org.apache.druid.query.OrderBy;
+import org.apache.druid.segment.VirtualColumns;
+
+import java.util.List;
+
+@JsonTypeInfo(use = JsonTypeInfo.Id.NAME, property = "type")
+@JsonSubTypes({
+ @JsonSubTypes.Type(name = "aggregate", value = AggregateProjectionSpec.class)
+})
+public interface ProjectionSpec
Review Comment:
The other interface I had in mind is projections which are not using a rollup facts holder, which came from some offline discussions I have had with @gianm, but I have not built this yet. The thinking was that since we've lifted the restriction that segments have to be time ordered, there seems likely to be some use cases for projections which filter, transform, or order the data differently than what the 'base' table does but without grouping the data. I suspect we would always want these to be filtered, or else represent them in persisted segments quite a bit differently (maybe just an index for ordering with no filtering).
The ordering stuff is also not fully realized yet so maybe the utility of these kinds of projections isn't quite apparent at this time. We were thinking that query engines should be able to take advantage of the cases where `CursorHolder.getOrdering()` is the same as the order specified by the `CursorBuildSpec.getPreferredOrdering()`, but right now its use is primarily limited to making cursors for ascending or descending time ordered data for the timeseries, timeboundary, and scan with time ordering queries (which are the only things which currently set preferredOrdering on the build spec today).
I'll back out the interface for now until we add other implementations.
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] projections part 1 - realtime segments (druid)
Posted by "clintropolis (via GitHub)" <gi...@apache.org>.
clintropolis commented on code in PR #17117:
URL: https://github.com/apache/druid/pull/17117#discussion_r1771918768
##########
processing/src/main/java/org/apache/druid/segment/incremental/OnheapIncrementalIndex.java:
##########
@@ -158,6 +185,82 @@ public class OnheapIncrementalIndex extends IncrementalIndex
this.maxBytesPerRowForAggregators =
useMaxMemoryEstimates ? getMaxBytesPerRowForAggregators(incrementalIndexSchema) : 0;
this.useMaxMemoryEstimates = useMaxMemoryEstimates;
+
+ initializeProjections(incrementalIndexSchema, useMaxMemoryEstimates);
+ }
+
+ private void initializeProjections(IncrementalIndexSchema incrementalIndexSchema, boolean useMaxMemoryEstimates)
+ {
+ this.aggregateProjections = new ObjectAVLTreeSet<>(OnHeapAggregateProjection.COMPARATOR);
+ for (ProjectionSpec projectionSpec : incrementalIndexSchema.getProjections()) {
+ if (projectionSpec instanceof AggregateProjectionSpec) {
+ AggregateProjectionSpec schema = (AggregateProjectionSpec) projectionSpec;
+ final List<DimensionDesc> descs = new ArrayList<>();
+ final int[] parentDimIndex = new int[schema.getGroupingColumns().size()];
+ int i = 0;
+ int foundTimePosition = -1;
+ String timeColumnName = null;
+ for (DimensionSchema dimension : schema.getGroupingColumns()) {
+ if (ColumnHolder.TIME_COLUMN_NAME.equals(dimension.getName())) {
+ timeColumnName = dimension.getName();
+ foundTimePosition = i;
+ parentDimIndex[i] = -1;
+ } else {
+ VirtualColumn vc = schema.getVirtualColumns().getVirtualColumn(dimension.getName());
+ Granularity granularity = Granularities.fromVirtualColumn(vc);
+ if (granularity != null) {
+ timeColumnName = dimension.getName();
+ foundTimePosition = i;
+ parentDimIndex[i] = -1;
+ } else {
+ final DimensionDesc parent = getDimension(dimension.getName());
+ if (parent == null) {
+ final DimensionDesc childOnly = new DimensionDesc(
+ i++,
+ dimension.getName(),
+ dimension.getDimensionHandler(),
+ useMaxMemoryEstimates
+ );
+ descs.add(childOnly);
+ parentDimIndex[childOnly.getIndex()] = -1;
+ } else {
+
+ if (!dimension.getColumnType().equals(parent.getCapabilities().toColumnType())) {
+ throw InvalidInput.exception(
+ "projection[%s] contains dimension[%s] with different type[%s] than type[%s] in base table",
+ schema.getName(),
+ dimension.getName(),
+ dimension.getColumnType(),
+ parent.getCapabilities().toColumnType()
+ );
+ }
+ final DimensionDesc rewrite = new DimensionDesc(
+ i++,
+ parent.getName(),
+ parent.getHandler(),
+ parent.getIndexer()
+ );
+ descs.add(rewrite);
Review Comment:
this was a bad name, it should have been called `child`. It is a `DimensionDesc` that doesn't make a new handler or indexer, but instead just contains the contents of its parent `DimensionDesc` except with the position of the child (`DimensionDesc.getIndex()` which should probably be renamed `DimensionDesc.getPosition()`...)
##########
processing/src/main/java/org/apache/druid/segment/incremental/OnheapIncrementalIndex.java:
##########
@@ -254,6 +357,10 @@ protected AddToFactsResult addToFacts(
final List<String> parseExceptionMessages = new ArrayList<>();
final AtomicLong totalSizeInBytes = getBytesInMemory();
+ for (OnHeapAggregateProjection projection : aggregateProjections) {
+ projection.addToFacts(key, inputRowHolder.getRow(), parseExceptionMessages, totalSizeInBytes);
+ }
+
Review Comment:
it doesn't really make a difference either way and works fine before or after. My initial prototype was just like.. inlined through `OnHeapIncrementalIndex` and i had the projection logic first because i wanted my debugger to get to the code asap so had it at the front of the method.
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] projections part 1 - realtime segments (druid)
Posted by "clintropolis (via GitHub)" <gi...@apache.org>.
clintropolis commented on code in PR #17117:
URL: https://github.com/apache/druid/pull/17117#discussion_r1773981386
##########
processing/src/main/java/org/apache/druid/segment/incremental/OnheapIncrementalIndex.java:
##########
@@ -1038,4 +1182,398 @@ public void clear()
facts.clear();
}
}
+
+ public static class OnHeapAggregateProjection implements IncrementalIndexRowSelector
Review Comment:
yea, i was (and am still) considering this, especially since `IncrementalIndex` itself is kind of a strange abstraction in the first place and I think it could let us clean some stuff up. The main reason I have not tackled this yet is because it would require being able to represent no rollup projections, which is left out for now, the other was that it would be a bit more dramatic of a change for `IncrementalIndex`, and I was trying to be conservative on overall changes for this PR. I do think it would be a lot nicer though if it was just all projections, so I plan to keep thinking about this.
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] projections part 1 - realtime segments (druid)
Posted by "clintropolis (via GitHub)" <gi...@apache.org>.
clintropolis commented on code in PR #17117:
URL: https://github.com/apache/druid/pull/17117#discussion_r1771918300
##########
processing/src/main/java/org/apache/druid/segment/incremental/IncrementalIndexCursorFactory.java:
##########
@@ -77,7 +78,20 @@ public IncrementalIndexCursorFactory(IncrementalIndex index)
@Override
public CursorHolder makeCursorHolder(CursorBuildSpec spec)
{
- return new IncrementalIndexCursorHolder(index, spec);
+ final IncrementalIndexRowSelector rowSelector;
+ final CursorBuildSpec buildSpec;
+ final String timeColumnName;
+ final IncrementalIndex.ProjectionRowSelector projection = index.getProjection(spec);
+ if (projection == null) {
+ rowSelector = index;
+ buildSpec = spec;
+ timeColumnName = ColumnHolder.TIME_COLUMN_NAME;
+ } else {
+ rowSelector = projection.getRowSelector();
+ buildSpec = projection.getRewrittenBuildSpec();
+ timeColumnName = projection.getTimeColumnName();
+ }
+ return new IncrementalIndexCursorHolder(rowSelector, buildSpec, timeColumnName, projection != null);
Review Comment:
In practice I don't think `PreAggregatedIncrementalIndexCursorHolder` makes much difference unless it also makes its own version of `IncrementalIndexCursor` and/or `IncrementalIndexColumnSelectorFactory` which are the real things that get used by the engines. However, implementing these things doesn't really seem necessary, since its all basically just selecting stuff from `FactsHolder` + aggs, which is why I just kept it in the same holder for now, because i think it would basically be just overriding this method
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] projections part 1 - realtime segments (druid)
Posted by "clintropolis (via GitHub)" <gi...@apache.org>.
clintropolis commented on code in PR #17117:
URL: https://github.com/apache/druid/pull/17117#discussion_r1771920459
##########
processing/src/main/java/org/apache/druid/segment/incremental/OnheapIncrementalIndex.java:
##########
@@ -1038,4 +1182,398 @@ public void clear()
facts.clear();
}
}
+
+ public static class OnHeapAggregateProjection implements IncrementalIndexRowSelector
+ {
+ static Comparator<OnHeapAggregateProjection> COMPARATOR = (o1, o2) -> {
+ // coarsest granularity first
+ if (o1.projectionGranularity.isFinerThan(o2.projectionGranularity)) {
+ return 1;
+ }
+ if (o2.projectionGranularity.isFinerThan(o1.projectionGranularity)) {
+ return -1;
+ }
+ // fewer dimensions first
+ final int dimsCompare = Integer.compare(
+ o1.schema.getGroupingColumns().size(),
+ o2.schema.getGroupingColumns().size()
+ );
+ if (dimsCompare != 0) {
+ return dimsCompare;
+ }
+ // more metrics first
+ int metCompare = Integer.compare(o2.schema.getAggregators().length, o1.schema.getAggregators().length);
+ if (metCompare != 0) {
+ return metCompare;
+ }
+ // more virtual columns first
+ final int virtCompare = Integer.compare(
+ o2.schema.getVirtualColumns().getVirtualColumns().length,
+ o1.schema.getVirtualColumns().getVirtualColumns().length
+ );
+ if (virtCompare != 0) {
+ return virtCompare;
+ }
+ return o1.schema.getName().compareTo(o2.schema.getName());
+ };
+
+ private final AggregateProjectionSpec schema;
+ private final Granularity projectionGranularity;
+ private final boolean hasTimeColumn;
+ @Nullable
+ private final String timeColumnName;
+ private final long minTimestamp;
+ private final List<DimensionDesc> dimensions;
+ private final int[] parentDimensionIndex;
+ private final Map<String, DimensionDesc> dimensionsMap;
+ private final Map<String, MetricDesc> aggregatorsMap;
+ private final List<OrderBy> ordering;
+ private final FactsHolder factsHolder;
+ private final InputRowHolder inputRowHolder = new InputRowHolder();
+ private final Map<String, ColumnSelectorFactory> aggSelectors;
+ private final ColumnSelectorFactory virtualSelectorFactory;
+ private final ConcurrentHashMap<Integer, Aggregator[]> aggregators = new ConcurrentHashMap<>();
+ private final ColumnInspector baseTableInspector;
+ private final AtomicInteger rowCounter = new AtomicInteger(0);
+ private final boolean useMaxMemoryEstimates;
+ private final long maxBytesPerRowForAggregators;
+
+
+ public OnHeapAggregateProjection(
+ AggregateProjectionSpec schema,
+ List<DimensionDesc> dimensions,
+ int[] parentDimensionIndex,
+ ColumnInspector baseTableInspector,
+ @Nullable String timeColumnName,
+ int foundTimePosition,
+ long minTimestamp,
+ boolean useMaxMemoryEstimates,
+ long maxBytesPerRowForAggregators
+ )
+ {
+ this.schema = schema;
+ this.dimensions = dimensions;
+ this.parentDimensionIndex = parentDimensionIndex;
+ this.dimensionsMap = new HashMap<>();
+ for (DimensionDesc desc : dimensions) {
+ dimensionsMap.put(desc.getName(), desc);
+ }
+ this.baseTableInspector = baseTableInspector;
+ this.timeColumnName = timeColumnName;
+ Granularity virtualGranularity = null;
+ for (VirtualColumn vc : schema.getVirtualColumns().getVirtualColumns()) {
+ if (vc.requiresColumn(ColumnHolder.TIME_COLUMN_NAME) && vc.requiredColumns().size() == 1) {
+ virtualGranularity = Granularities.fromVirtualColumn(vc);
+ break;
+ }
+ }
Review Comment:
this was actually incorrect, we need to use the finest granularity time_floor so i have updated the logic to do that and added comments
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] projections part 1 - realtime segments (druid)
Posted by "clintropolis (via GitHub)" <gi...@apache.org>.
clintropolis commented on code in PR #17117:
URL: https://github.com/apache/druid/pull/17117#discussion_r1771925490
##########
processing/src/main/java/org/apache/druid/segment/incremental/IncrementalIndexCursorHolder.java:
##########
@@ -73,10 +79,17 @@ public Cursor asCursor()
return new IncrementalIndexCursor(
rowSelector,
spec,
- Cursors.getTimeOrdering(ordering)
+ timeColumnName,
+ getTimeOrder(ordering, timeColumnName)
);
}
+ @Override
+ public boolean isPreAggregated()
+ {
+ return isPreaggregated;
+ }
+
Review Comment:
The flow is this, the query engines make a `CursorBuildSpec` that is given to the `CursorFactory` to produce a `CursorHolder`. If the `CursorBuildSpec` is aggregating, the `CursorFactory` can see if it has all of the required grouping and filtering columns, equivalent virtual columns (or all required inputs) and equivalent aggregators. If all of these things are true, then the `CursorFactory` returns a `CursorHolder` with `isPreAggregated` set to true, and the selector factories expose the aggregator columns as the output name of the agg factories on the `CursorBuildSpec` (it also does this for virtual columns if they are pre-existing and don't need recomputed). Because these agg columns were built by the same agg factory otherwise, the engine can safely check `isPreAggregated` and use the combining factory to process the segment instead.
I think this pattern should safe and has worked for everything I have tried, because its also how merging of per segment results works.
I couldn't find any cases where the aggregator renames its output when calling `getCombiningFactory` and i looked at all of the implementations. I'm not sure it would be allowed, because it would break any result format that was map based, so it seems to me that the output name must be stable for both the initial pass and also merging results from the per segment results with the combining aggregator. As far as I can tell, the factory stores its output at output name, so the combining factory will always make selectors that reads inputs from that output name to do segment merging.
ColumnType alone isn't enough to do it because the combining operation might be different than the initial segment scan, the simplest case is the Count aggregator, but others can fit this pattern too (most of the sketches split into build and merge aggregators for example)
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] projections part 1 - realtime segments (druid)
Posted by "clintropolis (via GitHub)" <gi...@apache.org>.
clintropolis commented on code in PR #17117:
URL: https://github.com/apache/druid/pull/17117#discussion_r1771915240
##########
processing/src/main/java/org/apache/druid/segment/Cursors.java:
##########
@@ -64,6 +64,26 @@ public static Order getTimeOrdering(final List<OrderBy> ordering)
}
}
+ public static boolean compatibleOrdering(CursorBuildSpec buildSpec, List<OrderBy> ordering)
Review Comment:
moved
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] projections part 1 - realtime segments (druid)
Posted by "clintropolis (via GitHub)" <gi...@apache.org>.
clintropolis commented on code in PR #17117:
URL: https://github.com/apache/druid/pull/17117#discussion_r1771914944
##########
processing/src/main/java/org/apache/druid/segment/CursorBuildSpec.java:
##########
@@ -166,6 +167,11 @@ public QueryMetrics<?> getQueryMetrics()
return queryMetrics;
}
+ public boolean isAggregate()
+ {
+ return !CollectionUtils.isNullOrEmpty(groupingColumns) || !CollectionUtils.isNullOrEmpty(aggregators);
Review Comment:
yea, it seems reasonable to just precompute it in the `CursorBuildSpec` constructor (and add javadocs too)
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] projections part 1 - realtime segments (druid)
Posted by "clintropolis (via GitHub)" <gi...@apache.org>.
clintropolis commented on code in PR #17117:
URL: https://github.com/apache/druid/pull/17117#discussion_r1771919145
##########
processing/src/main/java/org/apache/druid/segment/incremental/OnheapIncrementalIndex.java:
##########
@@ -431,6 +538,43 @@ private void closeAggregators()
}
}
+ @Nullable
+ @Override
+ public ProjectionRowSelector getProjection(CursorBuildSpec buildSpec)
+ {
+ if (buildSpec.getQueryContext().getBoolean(QueryContexts.CTX_NO_PROJECTION, false)) {
+ return null;
+ }
Review Comment:
I think that it is likely the only ever direct caller of this method is going to be `IncrementalIndexCursorFactory.makeCursorHolder`, which seems also a reasonable place to check, I can move it there if you prefer. The main reason I did it here is because i wired up both context flags I added at the same time when I wrote it, since here is also where checking the context flag for forcing a specific projection be used by name is happening.
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] projections part 1 - realtime segments (druid)
Posted by "clintropolis (via GitHub)" <gi...@apache.org>.
clintropolis commented on code in PR #17117:
URL: https://github.com/apache/druid/pull/17117#discussion_r1771925490
##########
processing/src/main/java/org/apache/druid/segment/incremental/IncrementalIndexCursorHolder.java:
##########
@@ -73,10 +79,17 @@ public Cursor asCursor()
return new IncrementalIndexCursor(
rowSelector,
spec,
- Cursors.getTimeOrdering(ordering)
+ timeColumnName,
+ getTimeOrder(ordering, timeColumnName)
);
}
+ @Override
+ public boolean isPreAggregated()
+ {
+ return isPreaggregated;
+ }
+
Review Comment:
The flow is this, the query engines make a `CursorBuildSpec` that is given to the `CursorFactory` to produce a `CursorHolder`. If the `CursorBuildSpec` is aggregating, the `CursorFactory` can see if it has all of the required grouping and filtering columns, equivalent virtual columns (or all required inputs) and equivalent aggregators. If all of these things are true, then the `CursorFactory` returns a `CursorHolder` with `isPreAggregated` set to true, and the selector factories expose the aggregator columns as the output name of the agg factories on the `CursorBuildSpec` (it also does this for virtual columns if they are pre-existing and don't need recomputed). Because these agg columns were built by the same agg factory otherwise, the engine can safely check `isPreAggregated` and use the combining factory to process the segment instead.
I think this pattern should safe and has worked for everything I have tried, because its also how merging of per segment results works.
I couldn't find any cases where the aggregator renames its output when calling `getCombiningFactory` and i looked at all of the implementations. I'm not sure it would be allowed, because it would break any result format that was map based, so it seems to me that the output name must be stable for both the initial pass and also merging results from the per segment results with the combining aggregator. As far as I can tell, the factory stores its output at output name, so the combining factory will always make selectors that reads inputs from that output name to do segment merging.
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] projections part 1 - realtime segments (druid)
Posted by "github-advanced-security[bot] (via GitHub)" <gi...@apache.org>.
github-advanced-security[bot] commented on code in PR #17117:
URL: https://github.com/apache/druid/pull/17117#discussion_r1767565512
##########
processing/src/test/java/org/apache/druid/segment/incremental/IncrementalIndexCursorFactoryTest.java:
##########
@@ -0,0 +1,1281 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.segment.incremental;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.Iterables;
+import com.google.common.collect.Lists;
+import org.apache.druid.collections.CloseableDefaultBlockingPool;
+import org.apache.druid.collections.CloseableStupidPool;
+import org.apache.druid.collections.NonBlockingPool;
+import org.apache.druid.common.config.NullHandling;
+import org.apache.druid.data.input.InputRow;
+import org.apache.druid.data.input.ListBasedInputRow;
+import org.apache.druid.data.input.MapBasedInputRow;
+import org.apache.druid.data.input.impl.AggregateProjectionSpec;
+import org.apache.druid.data.input.impl.DimensionsSpec;
+import org.apache.druid.data.input.impl.DoubleDimensionSchema;
+import org.apache.druid.data.input.impl.FloatDimensionSchema;
+import org.apache.druid.data.input.impl.LongDimensionSchema;
+import org.apache.druid.data.input.impl.StringDimensionSchema;
+import org.apache.druid.guice.BuiltInTypesModule;
+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.java.util.common.guava.Sequence;
+import org.apache.druid.js.JavaScriptConfig;
+import org.apache.druid.query.DruidProcessingConfig;
+import org.apache.druid.query.QueryContexts;
+import org.apache.druid.query.Result;
+import org.apache.druid.query.aggregation.AggregatorFactory;
+import org.apache.druid.query.aggregation.CountAggregatorFactory;
+import org.apache.druid.query.aggregation.DoubleSumAggregatorFactory;
+import org.apache.druid.query.aggregation.FloatSumAggregatorFactory;
+import org.apache.druid.query.aggregation.JavaScriptAggregatorFactory;
+import org.apache.druid.query.aggregation.LongSumAggregatorFactory;
+import org.apache.druid.query.dimension.DefaultDimensionSpec;
+import org.apache.druid.query.expression.TestExprMacroTable;
+import org.apache.druid.query.filter.ColumnIndexSelector;
+import org.apache.druid.query.filter.DimFilters;
+import org.apache.druid.query.filter.DruidDoublePredicate;
+import org.apache.druid.query.filter.DruidFloatPredicate;
+import org.apache.druid.query.filter.DruidLongPredicate;
+import org.apache.druid.query.filter.DruidObjectPredicate;
+import org.apache.druid.query.filter.DruidPredicateFactory;
+import org.apache.druid.query.filter.Filter;
+import org.apache.druid.query.filter.ValueMatcher;
+import org.apache.druid.query.groupby.GroupByQuery;
+import org.apache.druid.query.groupby.GroupByQueryConfig;
+import org.apache.druid.query.groupby.GroupByResourcesReservationPool;
+import org.apache.druid.query.groupby.GroupingEngine;
+import org.apache.druid.query.groupby.ResultRow;
+import org.apache.druid.query.topn.TopNQueryBuilder;
+import org.apache.druid.query.topn.TopNQueryEngine;
+import org.apache.druid.query.topn.TopNResultValue;
+import org.apache.druid.segment.CloserRule;
+import org.apache.druid.segment.ColumnSelectorFactory;
+import org.apache.druid.segment.Cursor;
+import org.apache.druid.segment.CursorBuildSpec;
+import org.apache.druid.segment.CursorFactory;
+import org.apache.druid.segment.CursorHolder;
+import org.apache.druid.segment.Cursors;
+import org.apache.druid.segment.DimensionSelector;
+import org.apache.druid.segment.IncrementalIndexSegment;
+import org.apache.druid.segment.IncrementalIndexTimeBoundaryInspector;
+import org.apache.druid.segment.TestHelper;
+import org.apache.druid.segment.VirtualColumns;
+import org.apache.druid.segment.column.ColumnType;
+import org.apache.druid.segment.column.RowSignature;
+import org.apache.druid.segment.data.IndexedInts;
+import org.apache.druid.segment.filter.Filters;
+import org.apache.druid.segment.filter.SelectorFilter;
+import org.apache.druid.segment.index.AllTrueBitmapColumnIndex;
+import org.apache.druid.segment.index.BitmapColumnIndex;
+import org.apache.druid.segment.virtual.ExpressionVirtualColumn;
+import org.apache.druid.testing.InitializedNullHandlingTest;
+import org.apache.druid.timeline.SegmentId;
+import org.joda.time.DateTime;
+import org.joda.time.Interval;
+import org.junit.Assert;
+import org.junit.Assume;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+import javax.annotation.Nullable;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.Set;
+
+/**
+ */
+@RunWith(Parameterized.class)
+public class IncrementalIndexCursorFactoryTest extends InitializedNullHandlingTest
+{
+ public final IncrementalIndexCreator indexCreator;
+ public final IncrementalIndexCursorFactory projectionsCursorFactory;
+ public final IncrementalIndexTimeBoundaryInspector projectionsTimeBoundaryInspector;
+
+ private final GroupingEngine groupingEngine;
+ private final TopNQueryEngine topnQueryEngine;
+
+ private final NonBlockingPool<ByteBuffer> nonBlockingPool;
+
+ /**
+ * If true, sort by [billy, __time]. If false, sort by [__time].
+ */
+ public final boolean sortByDim;
+
+ @Rule
+ public final CloserRule closer = new CloserRule(false);
+
+ public IncrementalIndexCursorFactoryTest(String indexType, boolean sortByDim)
+ throws JsonProcessingException, IndexSizeExceededException
+ {
+ BuiltInTypesModule.registerHandlersAndSerde();
+ this.sortByDim = sortByDim;
+ this.indexCreator = closer.closeLater(
+ new IncrementalIndexCreator(
+ indexType,
+ (builder, args) -> {
+ final DimensionsSpec dimensionsSpec;
+
+ if (sortByDim) {
+ dimensionsSpec =
+ DimensionsSpec.builder()
+ .setDimensions(Collections.singletonList(new StringDimensionSchema("billy")))
+ .setForceSegmentSortByTime(false)
+ .setIncludeAllDimensions(true)
+ .build();
+ } else {
+ dimensionsSpec = DimensionsSpec.EMPTY;
+ }
+
+ return builder
+ .setIndexSchema(
+ IncrementalIndexSchema
+ .builder()
+ .withDimensionsSpec(dimensionsSpec)
+ .withMetrics(new CountAggregatorFactory("cnt"))
+ .build()
+ )
+ .setMaxRowCount(1_000)
+ .build();
+ }
+ )
+ );
+
+ final DateTime timestamp = Granularities.DAY.bucket(DateTimes.nowUtc()).getStart();
+ final IncrementalIndexCreator projectionsCreator = closer.closeLater(
+ new IncrementalIndexCreator(
+ indexType,
+ (builder, args) -> {
+ DimensionsSpec.Builder bob = DimensionsSpec.builder()
+ .setDimensions(
+ Arrays.asList(
+ new StringDimensionSchema("a"),
+ new StringDimensionSchema("b"),
+ new LongDimensionSchema("c"),
+ new DoubleDimensionSchema("d"),
+ new FloatDimensionSchema("e")
+ )
+ );
+ if (sortByDim) {
+ bob.setForceSegmentSortByTime(false);
+ }
+
+ return builder
+ .setIndexSchema(
+ IncrementalIndexSchema
+ .builder()
+ .withRollup(false)
+ .withMinTimestamp(timestamp.getMillis())
+ .withDimensionsSpec(bob.build())
+ .withProjections(
+ Arrays.asList(
+ new AggregateProjectionSpec(
+ "ab_hourly_cd_sum",
+ Arrays.asList(
+ new StringDimensionSchema("a"),
+ new StringDimensionSchema("b"),
+ new LongDimensionSchema("__gran")
+ ),
+ VirtualColumns.create(
+ Granularities.toVirtualColumn(Granularities.HOUR, "__gran")
+ ),
+ new AggregatorFactory[]{
+ new LongSumAggregatorFactory("_c_sum", "c"),
+ new DoubleSumAggregatorFactory("d", "d")
+ }
+ ),
+ new AggregateProjectionSpec(
+ "a_hourly_c_sum",
+ Arrays.asList(
+ new LongDimensionSchema("__gran"),
+ new StringDimensionSchema("a")
+ ),
+ VirtualColumns.create(
+ Granularities.toVirtualColumn(Granularities.HOUR, "__gran")
+ ),
+ new AggregatorFactory[]{
+ new LongSumAggregatorFactory("_c_sum", "c")
+ }
+ ),
+ new AggregateProjectionSpec(
+ "bf_daily_c_sum",
+ Arrays.asList(
+ new LongDimensionSchema("__gran"),
+ new StringDimensionSchema("b"),
+ new FloatDimensionSchema("e")
+ ),
+ VirtualColumns.create(
+ Granularities.toVirtualColumn(Granularities.DAY, "__gran")
+ ),
+ new AggregatorFactory[]{
+ new LongSumAggregatorFactory("_c_sum", "c")
+ }
+ ),
+ new AggregateProjectionSpec(
+ "ab_daily",
+ Arrays.asList(
+ new StringDimensionSchema("a"),
+ new StringDimensionSchema("b")
+ ),
+ null,
+ null
+ ),
+ new AggregateProjectionSpec(
+ "abfoo_daily",
+ Arrays.asList(
+ new StringDimensionSchema("a"),
+ new StringDimensionSchema("bfoo")
+ ),
+ VirtualColumns.create(
+ new ExpressionVirtualColumn(
+ "bfoo",
+ "concat(b, 'foo')",
+ ColumnType.STRING,
+ TestExprMacroTable.INSTANCE
+ )
+ ),
+ null
+ )
+ )
+ )
+ .build()
+ )
+ .setMaxRowCount(1_000)
+ .build();
+ }
+ )
+ );
+
+ final IncrementalIndex projectionsIndex = projectionsCreator.createIndex();
+ final RowSignature signature = RowSignature.builder()
+ .add("a", ColumnType.STRING)
+ .add("b", ColumnType.STRING)
+ .add("c", ColumnType.LONG)
+ .add("d", ColumnType.DOUBLE)
+ .build();
+ final List<InputRow> rows = Arrays.asList(
+ new ListBasedInputRow(signature, timestamp, signature.getColumnNames(), Arrays.asList("a", "aa", 1L, 1.0)),
+ new ListBasedInputRow(
+ signature,
+ timestamp.plusMinutes(2),
+ signature.getColumnNames(),
+ Arrays.asList("a", "bb", 1L, 1.1, 1.1f)
+ ),
+ new ListBasedInputRow(
+ signature,
+ timestamp.plusMinutes(4),
+ signature.getColumnNames(),
+ Arrays.asList("a", "cc", 2L, 2.2, 2.2f)
+ ),
+ new ListBasedInputRow(
+ signature,
+ timestamp.plusMinutes(6),
+ signature.getColumnNames(),
+ Arrays.asList("b", "aa", 3L, 3.3, 3.3f)
+ ),
+ new ListBasedInputRow(
+ signature,
+ timestamp.plusMinutes(8),
+ signature.getColumnNames(),
+ Arrays.asList("b", "aa", 4L, 4.4, 4.4f)
+ ),
+ new ListBasedInputRow(
+ signature,
+ timestamp.plusMinutes(10),
+ signature.getColumnNames(),
+ Arrays.asList("b", "bb", 5L, 5.5, 5.5f)
+ ),
+ new ListBasedInputRow(
+ signature,
+ timestamp.plusHours(1),
+ signature.getColumnNames(),
+ Arrays.asList("a", "aa", 1L, 1.1, 1.1f)
+ ),
+ new ListBasedInputRow(
+ signature,
+ timestamp.plusHours(1).plusMinutes(1),
+ signature.getColumnNames(),
+ Arrays.asList("a", "dd", 2L, 2.2, 2.2f)
+ )
+ );
+
+ for (InputRow row : rows) {
+ projectionsIndex.add(row);
+ }
+
+ projectionsCursorFactory = new IncrementalIndexCursorFactory(projectionsIndex);
+ projectionsTimeBoundaryInspector = new IncrementalIndexTimeBoundaryInspector(projectionsIndex);
+
+ nonBlockingPool = closer.closeLater(
+ new CloseableStupidPool<>(
+ "GroupByQueryEngine-bufferPool",
+ () -> ByteBuffer.allocate(50000)
+ )
+ );
+ groupingEngine = new GroupingEngine(
+ new DruidProcessingConfig(),
+ GroupByQueryConfig::new,
+ new GroupByResourcesReservationPool(
+ closer.closeLater(
+ new CloseableDefaultBlockingPool<>(
+ () -> ByteBuffer.allocate(50000),
+ 5
+ )
+ ),
+ new GroupByQueryConfig()
+ ),
+ TestHelper.makeJsonMapper(),
+ TestHelper.makeSmileMapper(),
+ (query, future) -> {}
+ );
+ topnQueryEngine = new TopNQueryEngine(nonBlockingPool);
+ }
+
+ @Parameterized.Parameters(name = "{index}: {0}, sortByDim: {1}")
+ public static Collection<?> constructorFeeder()
+ {
+ return IncrementalIndexCreator.indexTypeCartesianProduct(
+ ImmutableList.of(true, false) // sortByDim
+ );
+ }
+
+ @Test
+ public void testSanity() throws Exception
+ {
+ IncrementalIndex index = indexCreator.createIndex();
+ index.add(
+ new MapBasedInputRow(
+ System.currentTimeMillis() - 1,
+ Collections.singletonList("billy"),
+ ImmutableMap.of("billy", "hi")
+ )
+ );
+ index.add(
+ new MapBasedInputRow(
+ System.currentTimeMillis() - 1,
+ Collections.singletonList("sally"),
+ ImmutableMap.of("sally", "bo")
+ )
+ );
+
+
+ final GroupByQuery query = GroupByQuery.builder()
+ .setDataSource("test")
+ .setGranularity(Granularities.ALL)
+ .setInterval(new Interval(DateTimes.EPOCH, DateTimes.nowUtc()))
+ .addDimension("billy")
+ .addDimension("sally")
+ .addAggregator(new LongSumAggregatorFactory("cnt", "cnt"))
+ .addOrderByColumn("billy")
+ .build();
+ final IncrementalIndexCursorFactory cursorFactory = new IncrementalIndexCursorFactory(index);
+ final Sequence<ResultRow> rows = groupingEngine.process(
+ query,
+ cursorFactory,
+ new IncrementalIndexTimeBoundaryInspector(index),
+ nonBlockingPool,
+ null
+ );
+
+ final List<ResultRow> results = rows.toList();
+
+ Assert.assertEquals(2, results.size());
+
+ ResultRow row = results.get(0);
+ Assert.assertArrayEquals(new Object[]{NullHandling.defaultStringValue(), "bo", 1L}, row.getArray());
+
+ row = results.get(1);
+ Assert.assertArrayEquals(new Object[]{"hi", NullHandling.defaultStringValue(), 1L}, row.getArray());
+ }
+
+ @Test
+ public void testObjectColumnSelectorOnVaryingColumnSchema() throws Exception
+ {
+ IncrementalIndex index = indexCreator.createIndex();
+ index.add(
+ new MapBasedInputRow(
+ DateTimes.of("2014-09-01T00:00:00"),
+ Collections.singletonList("billy"),
+ ImmutableMap.of("billy", "hi")
+ )
+ );
+ index.add(
+ new MapBasedInputRow(
+ DateTimes.of("2014-09-01T01:00:00"),
+ Lists.newArrayList("billy", "sally"),
+ ImmutableMap.of(
+ "billy", "hip",
+ "sally", "hop"
+ )
+ )
+ );
+
+ final GroupByQuery query = GroupByQuery.builder()
+ .setDataSource("test")
+ .setGranularity(Granularities.ALL)
+ .setInterval(new Interval(DateTimes.EPOCH, DateTimes.nowUtc()))
+ .addDimension("billy")
+ .addDimension("sally")
+ .addAggregator(
+ new LongSumAggregatorFactory("cnt", "cnt")
+ )
+ .addAggregator(
+ new JavaScriptAggregatorFactory(
+ "fieldLength",
+ Arrays.asList("sally", "billy"),
+ "function(current, s, b) { return current + (s == null ? 0 : s.length) + (b == null ? 0 : b.length); }",
+ "function() { return 0; }",
+ "function(a,b) { return a + b; }",
+ JavaScriptConfig.getEnabledInstance()
+ )
+ )
+ .addOrderByColumn("billy")
+ .build();
+ final IncrementalIndexCursorFactory cursorFactory = new IncrementalIndexCursorFactory(index);
+ final Sequence<ResultRow> rows = groupingEngine.process(
+ query,
+ cursorFactory,
+ new IncrementalIndexTimeBoundaryInspector(index),
+ nonBlockingPool,
+ null
+ );
+
+ final List<ResultRow> results = rows.toList();
+
+ Assert.assertEquals(2, results.size());
+
+ ResultRow row = results.get(0);
+ Assert.assertArrayEquals(new Object[]{"hi", NullHandling.defaultStringValue(), 1L, 2.0}, row.getArray());
+
+ row = results.get(1);
+ Assert.assertArrayEquals(
+ new Object[]{"hip", "hop", 1L, 6.0},
+ row.getArray()
+ );
+ }
+
+ @Test
+ public void testResetSanity() throws IOException
+ {
+ // Test is only valid when sortByDim = false, due to usage of Granularities.NONE.
+ Assume.assumeFalse(sortByDim);
+
+ IncrementalIndex index = indexCreator.createIndex();
+ DateTime t = DateTimes.nowUtc();
+ Interval interval = new Interval(t.minusMinutes(1), t.plusMinutes(1));
+
+ index.add(
+ new MapBasedInputRow(
+ t.minus(1).getMillis(),
+ Collections.singletonList("billy"),
+ ImmutableMap.of("billy", "hi")
+ )
+ );
+ index.add(
+ new MapBasedInputRow(
+ t.minus(1).getMillis(),
+ Collections.singletonList("sally"),
+ ImmutableMap.of("sally", "bo")
+ )
+ );
+
+ IncrementalIndexCursorFactory cursorFactory = new IncrementalIndexCursorFactory(index);
+
+ for (boolean descending : Arrays.asList(false, true)) {
+ final CursorBuildSpec buildSpec = CursorBuildSpec
+ .builder()
+ .setFilter(new SelectorFilter("sally", "bo"))
+ .setInterval(interval)
+ .setPreferredOrdering(descending ? Cursors.descendingTimeOrder() : Cursors.ascendingTimeOrder())
+ .build();
+
+ try (final CursorHolder cursorHolder = cursorFactory.makeCursorHolder(buildSpec)) {
+ Cursor cursor = cursorHolder.asCursor();
+ DimensionSelector dimSelector;
+
+ dimSelector = cursor
+ .getColumnSelectorFactory()
+ .makeDimensionSelector(new DefaultDimensionSpec("sally", "sally"));
+ Assert.assertEquals("bo", dimSelector.lookupName(dimSelector.getRow().get(0)));
+
+ index.add(
+ new MapBasedInputRow(
+ t.minus(1).getMillis(),
+ Collections.singletonList("sally"),
+ ImmutableMap.of("sally", "ah")
+ )
+ );
+
+ // Cursor reset should not be affected by out of order values
+ cursor.reset();
+
+ dimSelector = cursor
+ .getColumnSelectorFactory()
+ .makeDimensionSelector(new DefaultDimensionSpec("sally", "sally"));
+ Assert.assertEquals("bo", dimSelector.lookupName(dimSelector.getRow().get(0)));
+ }
+ }
+ }
+
+ @Test
+ public void testSingleValueTopN() throws IOException
+ {
+ IncrementalIndex index = indexCreator.createIndex();
+ DateTime t = DateTimes.nowUtc();
+ index.add(
+ new MapBasedInputRow(
+ t.minus(1).getMillis(),
+ Collections.singletonList("sally"),
+ ImmutableMap.of("sally", "bo")
+ )
+ );
+
+ final Iterable<Result<TopNResultValue>> results = topnQueryEngine
+ .query(
+ new TopNQueryBuilder()
+ .dataSource("test")
+ .granularity(Granularities.ALL)
+ .intervals(Collections.singletonList(new Interval(DateTimes.EPOCH, DateTimes.nowUtc())))
+ .dimension("sally")
+ .metric("cnt")
+ .threshold(10)
+ .aggregators(new LongSumAggregatorFactory("cnt", "cnt"))
+ .build(),
+ new IncrementalIndexSegment(index, SegmentId.dummy("test")),
+ null
+ )
+ .toList();
+
+ Assert.assertEquals(1, Iterables.size(results));
+ Assert.assertEquals(1, results.iterator().next().getValue().getValue().size());
+ }
+
+ @Test
+ public void testFilterByNull() throws Exception
+ {
+ IncrementalIndex index = indexCreator.createIndex();
+ index.add(
+ new MapBasedInputRow(
+ System.currentTimeMillis() - 1,
+ Collections.singletonList("billy"),
+ ImmutableMap.of("billy", "hi")
+ )
+ );
+ index.add(
+ new MapBasedInputRow(
+ System.currentTimeMillis() - 1,
+ Collections.singletonList("sally"),
+ ImmutableMap.of("sally", "bo")
+ )
+ );
+
+
+ final GroupByQuery query = GroupByQuery.builder()
+ .setDataSource("test")
+ .setGranularity(Granularities.ALL)
+ .setInterval(new Interval(DateTimes.EPOCH, DateTimes.nowUtc()))
+ .addDimension("billy")
+ .addDimension("sally")
+ .addAggregator(new LongSumAggregatorFactory("cnt", "cnt"))
+ .setDimFilter(DimFilters.dimEquals("sally", (String) null))
+ .build();
+ final IncrementalIndexCursorFactory cursorFactory = new IncrementalIndexCursorFactory(index);
+
+ final Sequence<ResultRow> rows = groupingEngine.process(
+ query,
+ cursorFactory,
+ new IncrementalIndexTimeBoundaryInspector(index),
+ nonBlockingPool,
+ null
+ );
+
+ final List<ResultRow> results = rows.toList();
+
+ Assert.assertEquals(1, results.size());
+
+ ResultRow row = results.get(0);
+ Assert.assertArrayEquals(new Object[]{"hi", NullHandling.defaultStringValue(), 1L}, row.getArray());
+ }
+
+ @Test
+ public void testCursoringAndIndexUpdationInterleaving() throws Exception
+ {
+ final IncrementalIndex index = indexCreator.createIndex();
+ final long timestamp = System.currentTimeMillis();
+
+ for (int i = 0; i < 2; i++) {
+ index.add(
+ new MapBasedInputRow(
+ timestamp,
+ Collections.singletonList("billy"),
+ ImmutableMap.of("billy", "v1" + i)
+ )
+ );
+ }
+
+ final CursorFactory cursorFactory = new IncrementalIndexCursorFactory(index);
+
+ final CursorBuildSpec buildSpec = CursorBuildSpec.builder()
+ .setInterval(Intervals.utc(timestamp - 60_000, timestamp + 60_000))
+ .build();
+ try (final CursorHolder cursorHolder = cursorFactory.makeCursorHolder(buildSpec)) {
+ Cursor cursor = cursorHolder.asCursor();
+ DimensionSelector dimSelector = cursor
+ .getColumnSelectorFactory()
+ .makeDimensionSelector(new DefaultDimensionSpec("billy", "billy"));
+ int cardinality = dimSelector.getValueCardinality();
+
+ //index gets more rows at this point, while other thread is iterating over the cursor
+ try {
+ for (int i = 0; i < 1; i++) {
+ index.add(new MapBasedInputRow(
+ timestamp,
+ Collections.singletonList("billy"),
+ ImmutableMap.of("billy", "v2" + i)
+ ));
+ }
+ }
+ catch (Exception ex) {
+ throw new RuntimeException(ex);
+ }
+
+ int rowNumInCursor = 0;
+ // and then, cursoring continues in the other thread
+ while (!cursor.isDone()) {
+ IndexedInts row = dimSelector.getRow();
+ row.forEach(i -> Assert.assertTrue(i < cardinality));
+ cursor.advance();
+ rowNumInCursor++;
+ }
+ Assert.assertEquals(2, rowNumInCursor);
+ }
+ }
+
+ @Test
+ public void testCursorDictionaryRaceConditionFix() throws Exception
+ {
+ // Tests the dictionary ID race condition bug described at https://github.com/apache/druid/pull/6340
+
+ final IncrementalIndex index = indexCreator.createIndex();
+ final long timestamp = System.currentTimeMillis();
+
+ for (int i = 0; i < 5; i++) {
+ index.add(
+ new MapBasedInputRow(
+ timestamp,
+ Collections.singletonList("billy"),
+ ImmutableMap.of("billy", "v1" + i)
+ )
+ );
+ }
+
+ final CursorFactory cursorFactory = new IncrementalIndexCursorFactory(index);
+
+ final CursorBuildSpec buildSpec = CursorBuildSpec.builder()
+ .setFilter(new DictionaryRaceTestFilter(index, timestamp))
+ .setInterval(Intervals.utc(timestamp - 60_000, timestamp + 60_000))
+ .build();
+ try (final CursorHolder cursorHolder = cursorFactory.makeCursorHolder(buildSpec)) {
+ Cursor cursor = cursorHolder.asCursor();
+ DimensionSelector dimSelector = cursor
+ .getColumnSelectorFactory()
+ .makeDimensionSelector(new DefaultDimensionSpec("billy", "billy"));
+ int cardinality = dimSelector.getValueCardinality();
+
+ int rowNumInCursor = 0;
+ while (!cursor.isDone()) {
+ IndexedInts row = dimSelector.getRow();
+ row.forEach(i -> Assert.assertTrue(i < cardinality));
+ cursor.advance();
+ rowNumInCursor++;
+ }
+ Assert.assertEquals(5, rowNumInCursor);
+ }
+ }
+
+ @Test
+ public void testCursoringAndSnapshot() throws Exception
+ {
+ final IncrementalIndex index = indexCreator.createIndex();
+ final long timestamp = System.currentTimeMillis();
+
+ final List<InputRow> rows = ImmutableList.of(
+ new MapBasedInputRow(timestamp, Collections.singletonList("billy"), ImmutableMap.of("billy", "v00")),
+ new MapBasedInputRow(timestamp, Collections.singletonList("billy"), ImmutableMap.of("billy", "v01")),
+ new MapBasedInputRow(timestamp, Collections.singletonList("billy"), ImmutableMap.of("billy", "v1")),
+ new MapBasedInputRow(timestamp, Collections.singletonList("billy"), ImmutableMap.of("billy", "v2")),
+ new MapBasedInputRow(timestamp, Collections.singletonList("billy2"), ImmutableMap.of("billy2", "v3")),
+ new MapBasedInputRow(timestamp, Collections.singletonList("billy"), ImmutableMap.of("billy", "v3")),
+ new MapBasedInputRow(timestamp, Collections.singletonList("billy3"), ImmutableMap.of("billy3", ""))
+ );
+
+ // Add first two rows.
+ for (int i = 0; i < 2; i++) {
+ index.add(rows.get(i));
+ }
+
+ final CursorFactory cursorFactory = new IncrementalIndexCursorFactory(index);
+
+ final CursorBuildSpec buildSpec = CursorBuildSpec.builder()
+ .setInterval(Intervals.utc(timestamp - 60_000, timestamp + 60_000))
+ .build();
+ try (final CursorHolder cursorHolder = cursorFactory.makeCursorHolder(buildSpec)) {
+ Cursor cursor = cursorHolder.asCursor();
+
+ DimensionSelector dimSelector1A = cursor
+ .getColumnSelectorFactory()
+ .makeDimensionSelector(new DefaultDimensionSpec("billy", "billy"));
+ int cardinalityA = dimSelector1A.getValueCardinality();
+
+ //index gets more rows at this point, while other thread is iterating over the cursor
+ try {
+ index.add(rows.get(2));
+ }
+ catch (Exception ex) {
+ throw new RuntimeException(ex);
+ }
+
+ DimensionSelector dimSelector1B = cursor
+ .getColumnSelectorFactory()
+ .makeDimensionSelector(new DefaultDimensionSpec("billy", "billy"));
+ //index gets more rows at this point, while other thread is iterating over the cursor
+ try {
+ index.add(rows.get(3));
+ index.add(rows.get(4));
+ }
+ catch (Exception ex) {
+ throw new RuntimeException(ex);
+ }
+
+ DimensionSelector dimSelector1C = cursor
+ .getColumnSelectorFactory()
+ .makeDimensionSelector(new DefaultDimensionSpec("billy", "billy"));
+
+ DimensionSelector dimSelector2D = cursor
+ .getColumnSelectorFactory()
+ .makeDimensionSelector(new DefaultDimensionSpec("billy2", "billy2"));
+ //index gets more rows at this point, while other thread is iterating over the cursor
+ try {
+ index.add(rows.get(5));
+ index.add(rows.get(6));
+ }
+ catch (Exception ex) {
+ throw new RuntimeException(ex);
+ }
+
+ DimensionSelector dimSelector3E = cursor
+ .getColumnSelectorFactory()
+ .makeDimensionSelector(new DefaultDimensionSpec("billy3", "billy3"));
+
+ int rowNumInCursor = 0;
+ // and then, cursoring continues in the other thread
+ while (!cursor.isDone()) {
+ IndexedInts rowA = dimSelector1A.getRow();
+ rowA.forEach(i -> Assert.assertTrue(i < cardinalityA));
+ IndexedInts rowB = dimSelector1B.getRow();
+ rowB.forEach(i -> Assert.assertTrue(i < cardinalityA));
+ IndexedInts rowC = dimSelector1C.getRow();
+ rowC.forEach(i -> Assert.assertTrue(i < cardinalityA));
+ IndexedInts rowD = dimSelector2D.getRow();
+ // no null id, so should get empty dims array
+ Assert.assertEquals(0, rowD.size());
+ IndexedInts rowE = dimSelector3E.getRow();
+ if (NullHandling.replaceWithDefault()) {
+ Assert.assertEquals(1, rowE.size());
+ // the null id
+ Assert.assertEquals(0, rowE.get(0));
+ } else {
+ Assert.assertEquals(0, rowE.size());
+ }
+ cursor.advance();
+ rowNumInCursor++;
+ }
+ Assert.assertEquals(2, rowNumInCursor);
+ }
+ }
+
+ @Test
+ public void testProjectionSelectionTwoDims()
+ {
+ // this query can use the projection with 2 dims, which has 7 rows instead of the total of 8
+ final GroupByQuery query =
+ GroupByQuery.builder()
+ .setDataSource("test")
+ .setGranularity(Granularities.ALL)
+ .setInterval(Intervals.ETERNITY)
+ .addDimension("a")
+ .addDimension("b")
+ .build();
+
+ final CursorBuildSpec buildSpec = GroupingEngine.makeCursorBuildSpec(query, null);
+
+ try (final CursorHolder cursorHolder = projectionsCursorFactory.makeCursorHolder(buildSpec)) {
+ final Cursor cursor = cursorHolder.asCursor();
+ int rowCount = 0;
+ while (!cursor.isDone()) {
+ rowCount++;
+ cursor.advance();
+ }
+ Assert.assertEquals(6, rowCount);
+ }
+
+ final Sequence<ResultRow> resultRows = groupingEngine.process(
+ query,
+ projectionsCursorFactory,
+ projectionsTimeBoundaryInspector,
+ nonBlockingPool,
+ null
+ );
+
+ final List<ResultRow> results = resultRows.toList();
+ Assert.assertEquals(6, results.size());
+ Assert.assertArrayEquals(new Object[]{"a", "aa"}, results.get(0).getArray());
+ Assert.assertArrayEquals(new Object[]{"a", "bb"}, results.get(1).getArray());
+ Assert.assertArrayEquals(new Object[]{"a", "cc"}, results.get(2).getArray());
+ Assert.assertArrayEquals(new Object[]{"a", "dd"}, results.get(3).getArray());
+ Assert.assertArrayEquals(new Object[]{"b", "aa"}, results.get(4).getArray());
+ Assert.assertArrayEquals(new Object[]{"b", "bb"}, results.get(5).getArray());
+ }
+
+ @Test
+ public void testProjectionSelectionTwoDimsVirtual()
+ {
+ // this query can use the projection with 2 dims, which has 7 rows instead of the total of 8
+ final GroupByQuery query =
+ GroupByQuery.builder()
+ .setDataSource("test")
+ .setGranularity(Granularities.ALL)
+ .setInterval(Intervals.ETERNITY)
+ .addDimension("a")
+ .addDimension("v0")
+ .setVirtualColumns(
+ new ExpressionVirtualColumn(
+ "v0",
+ "concat(b, 'foo')",
+ ColumnType.STRING,
+ TestExprMacroTable.INSTANCE
+ )
+ )
+ .setContext(ImmutableMap.of(QueryContexts.CTX_USE_PROJECTION, "abfoo_daily"))
+ .build();
+
+ final CursorBuildSpec buildSpec = GroupingEngine.makeCursorBuildSpec(query, null);
+
+ try (final CursorHolder cursorHolder = projectionsCursorFactory.makeCursorHolder(buildSpec)) {
+ final Cursor cursor = cursorHolder.asCursor();
+ int rowCount = 0;
+ while (!cursor.isDone()) {
+ rowCount++;
+ cursor.advance();
+ }
+ Assert.assertEquals(6, rowCount);
+ }
+
+ final Sequence<ResultRow> resultRows = groupingEngine.process(
+ query,
+ projectionsCursorFactory,
+ projectionsTimeBoundaryInspector,
+ nonBlockingPool,
+ null
+ );
+
+ final List<ResultRow> results = resultRows.toList();
+ Assert.assertEquals(6, results.size());
+ Assert.assertArrayEquals(new Object[]{"a", "aafoo"}, results.get(0).getArray());
+ Assert.assertArrayEquals(new Object[]{"a", "bbfoo"}, results.get(1).getArray());
+ Assert.assertArrayEquals(new Object[]{"a", "ccfoo"}, results.get(2).getArray());
+ Assert.assertArrayEquals(new Object[]{"a", "ddfoo"}, results.get(3).getArray());
+ Assert.assertArrayEquals(new Object[]{"b", "aafoo"}, results.get(4).getArray());
+ Assert.assertArrayEquals(new Object[]{"b", "bbfoo"}, results.get(5).getArray());
+ }
+
+ @Test
+ public void testProjectionSkipContext()
+ {
+ // setting context flag to skip projection
+ final GroupByQuery query =
+ GroupByQuery.builder()
+ .setDataSource("test")
+ .setGranularity(Granularities.ALL)
+ .setInterval(Intervals.ETERNITY)
+ .addDimension("a")
+ .addDimension("b")
+ .setContext(ImmutableMap.of(QueryContexts.CTX_NO_PROJECTION, true))
+ .addAggregator(new LongSumAggregatorFactory("c_sum", "c"))
+ .build();
+ CursorBuildSpec buildSpecSkip = GroupingEngine.makeCursorBuildSpec(query, null);
+ try (final CursorHolder cursorHolder = projectionsCursorFactory.makeCursorHolder(buildSpecSkip)) {
+ final Cursor cursor = cursorHolder.asCursor();
+ int rowCount = 0;
+ while (!cursor.isDone()) {
+ rowCount++;
+ cursor.advance();
+ }
+ // has to scan full 8 rows because context ensures projections not used
+ Assert.assertEquals(8, rowCount);
+ }
+ }
+
+ @Test
+ public void testProjectionSingleDim()
+ {
+ // test can use the single dimension projection
+ final GroupByQuery query =
+ GroupByQuery.builder()
+ .setDataSource("test")
+ .setGranularity(Granularities.ALL)
+ .setInterval(Intervals.ETERNITY)
+ .addDimension("a")
+ .addAggregator(new LongSumAggregatorFactory("c_sum", "c"))
+ .build();
+ final CursorBuildSpec buildSpec = GroupingEngine.makeCursorBuildSpec(query, null);
+ try (final CursorHolder cursorHolder = projectionsCursorFactory.makeCursorHolder(buildSpec)) {
+ final Cursor cursor = cursorHolder.asCursor();
+ int rowCount = 0;
+ while (!cursor.isDone()) {
+ rowCount++;
+ cursor.advance();
+ }
+ Assert.assertEquals(3, rowCount);
+ }
+ final Sequence<ResultRow> resultRows = groupingEngine.process(
+ query,
+ projectionsCursorFactory,
+ projectionsTimeBoundaryInspector,
+ nonBlockingPool,
+ null
+ );
+ final List<ResultRow> results = resultRows.toList();
+ Assert.assertEquals(2, results.size());
+ Assert.assertArrayEquals(new Object[]{"a", 7L}, results.get(0).getArray());
+ Assert.assertArrayEquals(new Object[]{"b", 12L}, results.get(1).getArray());
+ }
+
+ @Test
+ public void testQueryGranularityFinerThanProjectionGranularity()
+ {
+ final GroupByQuery.Builder queryBuilder =
+ GroupByQuery.builder()
+ .setDataSource("test")
+ .setInterval(Intervals.ETERNITY)
+ .addAggregator(new LongSumAggregatorFactory("c_sum", "c"));
+ if (sortByDim) {
+ queryBuilder.setVirtualColumns(Granularities.toVirtualColumn(Granularities.MINUTE, "__gran"))
+ .setDimensions(
+ DefaultDimensionSpec.of("__gran", ColumnType.LONG),
+ DefaultDimensionSpec.of("a")
+ )
+ .setGranularity(Granularities.ALL);
+ } else {
+ queryBuilder.addDimension("a")
+ .setGranularity(Granularities.MINUTE);
+ }
+ final GroupByQuery query = queryBuilder.build();
+ final CursorBuildSpec buildSpec = GroupingEngine.makeCursorBuildSpec(query, null);
+ try (final CursorHolder cursorHolder = projectionsCursorFactory.makeCursorHolder(buildSpec)) {
+ final Cursor cursor = cursorHolder.asCursor();
+ int rowCount = 0;
+ while (!cursor.isDone()) {
+ rowCount++;
+ cursor.advance();
+ }
+ Assert.assertEquals(8, rowCount);
+ }
+
+ final Sequence<ResultRow> resultRows = groupingEngine.process(
+ query,
+ projectionsCursorFactory,
+ projectionsTimeBoundaryInspector,
+ nonBlockingPool,
+ null
+ );
+
+ final List<ResultRow> results = resultRows.toList();
+ Assert.assertEquals(8, results.size());
+
+ Assert.assertArrayEquals(new Object[]{1726704000000L, "a", 1L}, results.get(0).getArray());
+ Assert.assertArrayEquals(new Object[]{1726704120000L, "a", 1L}, results.get(1).getArray());
+ Assert.assertArrayEquals(new Object[]{1726704240000L, "a", 2L}, results.get(2).getArray());
+ Assert.assertArrayEquals(new Object[]{1726704360000L, "b", 3L}, results.get(3).getArray());
+ Assert.assertArrayEquals(new Object[]{1726704480000L, "b", 4L}, results.get(4).getArray());
+ Assert.assertArrayEquals(new Object[]{1726704600000L, "b", 5L}, results.get(5).getArray());
+ Assert.assertArrayEquals(new Object[]{1726707600000L, "a", 1L}, results.get(6).getArray());
+ Assert.assertArrayEquals(new Object[]{1726707660000L, "a", 2L}, results.get(7).getArray());
+ }
+
+ @Test
+ public void testQueryGranularityFitsProjectionGranularity()
+ {
+ final GroupByQuery.Builder queryBuilder =
+ GroupByQuery.builder()
+ .setDataSource("test")
+ .setInterval(Intervals.ETERNITY)
+ .addAggregator(new LongSumAggregatorFactory("c_sum", "c"));
+ if (sortByDim) {
+ queryBuilder.setGranularity(Granularities.ALL)
+ .setDimensions(
+ DefaultDimensionSpec.of("__gran", ColumnType.LONG),
+ DefaultDimensionSpec.of("a")
+ )
+ .setVirtualColumns(Granularities.toVirtualColumn(Granularities.HOUR, "__gran"));
+ } else {
+ queryBuilder.addDimension("a")
+ .setGranularity(Granularities.HOUR);
+ }
+ final GroupByQuery query = queryBuilder.build();
+ final CursorBuildSpec buildSpec = GroupingEngine.makeCursorBuildSpec(query, null);
+ try (final CursorHolder cursorHolder = projectionsCursorFactory.makeCursorHolder(buildSpec)) {
+ final Cursor cursor = cursorHolder.asCursor();
+ int rowCount = 0;
+ while (!cursor.isDone()) {
+ rowCount++;
+ cursor.advance();
+ }
+ Assert.assertEquals(3, rowCount);
+ }
+
+ final Sequence<ResultRow> resultRows = groupingEngine.process(
+ query,
+ projectionsCursorFactory,
+ projectionsTimeBoundaryInspector,
+ nonBlockingPool,
+ null
+ );
+
+ final List<ResultRow> results = resultRows.toList();
+ Assert.assertEquals(3, results.size());
+ Assert.assertArrayEquals(new Object[]{1726704000000L, "a", 4L}, results.get(0).getArray());
+ Assert.assertArrayEquals(new Object[]{1726704000000L, "b", 12L}, results.get(1).getArray());
+ Assert.assertArrayEquals(new Object[]{1726707600000L, "a", 3L}, results.get(2).getArray());
+ }
+
+ @Test
+ public void testProjectionSelectionMissingAggregatorButHasAggregatorInput()
+ {
+ // d is present as a column on the projection, but since its an aggregate projection we cannot use it
+ final GroupByQuery query =
+ GroupByQuery.builder()
+ .setDataSource("test")
+ .setGranularity(Granularities.ALL)
+ .setInterval(Intervals.ETERNITY)
+ .addDimension("b")
+ .addAggregator(new LongSumAggregatorFactory("c_sum", "c"))
+ .addAggregator(new FloatSumAggregatorFactory("e_sum", "e"))
+ .build();
+ final CursorBuildSpec buildSpec = GroupingEngine.makeCursorBuildSpec(query, null);
+ try (final CursorHolder cursorHolder = projectionsCursorFactory.makeCursorHolder(buildSpec)) {
+ final Cursor cursor = cursorHolder.asCursor();
+ int rowCount = 0;
+ while (!cursor.isDone()) {
+ rowCount++;
+ cursor.advance();
+ }
+ Assert.assertEquals(8, rowCount);
+ }
+
+ final Sequence<ResultRow> resultRows = groupingEngine.process(
+ query,
+ projectionsCursorFactory,
+ projectionsTimeBoundaryInspector,
+ nonBlockingPool,
+ null
+ );
+
+ final List<ResultRow> results = resultRows.toList();
+ Assert.assertEquals(4, results.size());
+ Assert.assertArrayEquals(new Object[]{"aa", 9L, null}, results.get(0).getArray());
+ Assert.assertArrayEquals(new Object[]{"bb", 6L, null}, results.get(1).getArray());
+ Assert.assertArrayEquals(new Object[]{"cc", 2L, null}, results.get(2).getArray());
+ Assert.assertArrayEquals(new Object[]{"dd", 2L, null}, results.get(3).getArray());
+ }
+
+ @Test
+ public void testProjectionSelectionMissingAggregatorAndAggregatorInput()
+ {
+ // since d isn't present on the smaller projection, cant use it, but can still use the larger projection
+ final GroupByQuery query =
+ GroupByQuery.builder()
+ .setDataSource("test")
+ .setGranularity(Granularities.ALL)
+ .setInterval(Intervals.ETERNITY)
+ .addDimension("a")
+ .addAggregator(new LongSumAggregatorFactory("c_sum", "c"))
+ .addAggregator(new DoubleSumAggregatorFactory("d_sum", "d"))
+ .build();
+ final CursorBuildSpec buildSpec = GroupingEngine.makeCursorBuildSpec(query, null);
+ try (final CursorHolder cursorHolder = projectionsCursorFactory.makeCursorHolder(buildSpec)) {
+ final Cursor cursor = cursorHolder.asCursor();
+ int rowCount = 0;
+ while (!cursor.isDone()) {
+ rowCount++;
+ cursor.advance();
+ }
+ Assert.assertEquals(7, rowCount);
+ }
+
+ final Sequence<ResultRow> resultRows = groupingEngine.process(
+ query,
+ projectionsCursorFactory,
+ projectionsTimeBoundaryInspector,
+ nonBlockingPool,
+ null
+ );
+
+ final List<ResultRow> results = resultRows.toList();
+ Assert.assertEquals(2, results.size());
+ Assert.assertArrayEquals(new Object[]{"a", 7L, 7.6000000000000005}, results.get(0).getArray());
+ Assert.assertArrayEquals(new Object[]{"b", 12L, 13.2}, results.get(1).getArray());
+ }
+
+ @Test
+ public void testProjectionSelectionMissingColumnOnBaseTableToo()
+ {
+ // since d isn't present on the smaller projection, cant use it, but can still use the larger projection
+ final GroupByQuery query =
+ GroupByQuery.builder()
+ .setDataSource("test")
+ .setGranularity(Granularities.ALL)
+ .setInterval(Intervals.ETERNITY)
+ .addDimension("a")
+ .addDimension("z")
+ .addAggregator(new LongSumAggregatorFactory("c_sum", "c"))
+ .addAggregator(new DoubleSumAggregatorFactory("d_sum", "d"))
+ .build();
+ final CursorBuildSpec buildSpec = GroupingEngine.makeCursorBuildSpec(query, null);
+ try (final CursorHolder cursorHolder = projectionsCursorFactory.makeCursorHolder(buildSpec)) {
+ final Cursor cursor = cursorHolder.asCursor();
+ int rowCount = 0;
+ while (!cursor.isDone()) {
+ rowCount++;
+ cursor.advance();
+ }
+ Assert.assertEquals(7, rowCount);
+ }
+
+ final Sequence<ResultRow> resultRows = groupingEngine.process(
+ query,
+ projectionsCursorFactory,
+ projectionsTimeBoundaryInspector,
+ nonBlockingPool,
+ null
+ );
+
+ final List<ResultRow> results = resultRows.toList();
+ Assert.assertEquals(2, results.size());
+ Assert.assertArrayEquals(new Object[]{"a", null, 7L, 7.6000000000000005}, results.get(0).getArray());
+ Assert.assertArrayEquals(new Object[]{"b", null, 12L, 13.2}, results.get(1).getArray());
+ }
+
+ private static class DictionaryRaceTestFilter implements Filter
Review Comment:
## Inconsistent equals and hashCode
Class DictionaryRaceTestFilter overrides [hashCode](1) but not equals.
[Show more details](https://github.com/apache/druid/security/code-scanning/7914)
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] projections part 1 - realtime segments (druid)
Posted by "imply-cheddar (via GitHub)" <gi...@apache.org>.
imply-cheddar commented on code in PR #17117:
URL: https://github.com/apache/druid/pull/17117#discussion_r1767545355
##########
processing/src/main/java/org/apache/druid/java/util/common/granularity/Granularities.java:
##########
@@ -160,4 +161,20 @@ public static ExpressionVirtualColumn toVirtualColumn(Granularity granularity, S
ExprMacroTable.granularity()
);
}
+
+ @Nullable
+ public static Granularity fromVirtualColumn(VirtualColumn virtualColumn)
+ {
+ if (virtualColumn instanceof ExpressionVirtualColumn) {
+ final ExpressionVirtualColumn expressionVirtualColumn = (ExpressionVirtualColumn) virtualColumn;
+ final Expr expr = expressionVirtualColumn.getParsedExpression().get();
+ if (expr instanceof TimestampFloorExprMacro.TimestampFloorExpr) {
+ final TimestampFloorExprMacro.TimestampFloorExpr gran = (TimestampFloorExprMacro.TimestampFloorExpr) expr;
+ if (gran.getArg().getBindingIfIdentifier() != null) {
Review Comment:
This if statement is pretty obtuse, no clue what it's checking for. It reads like "If this one thing is not null, return this other thing". Perhaps a comment for why? Or the `TimestampFloorExpr` concrete class could gain a method that knows how to return a granularity?
##########
processing/src/main/java/org/apache/druid/segment/Cursors.java:
##########
@@ -64,6 +64,26 @@ public static Order getTimeOrdering(final List<OrderBy> ordering)
}
}
+ public static boolean compatibleOrdering(CursorBuildSpec buildSpec, List<OrderBy> ordering)
Review Comment:
Is there a reason that this isn't a method on `CursorBuildSpec`? That would seem pretty natural, I think?
##########
processing/src/main/java/org/apache/druid/segment/VirtualColumn.java:
##########
@@ -312,6 +312,10 @@ default ColumnCapabilities capabilities(ColumnInspector inspector, String column
*/
List<String> requiredColumns();
+ default boolean requiresColumn(String columnName)
+ {
+ return requiredColumns().contains(columnName);
+ }
Review Comment:
If someone were to ever naively iterate a bunch of column names and call this method thinking it was a good thing, they'd effectively get an `n^2` implementation, not to mention if `requiredColumns()` actually does work and doesn't memoize it (like, for example, building a list from a set or mapping the column names out of some nested objects in a list or something). These lists shouldn't be large, so perhaps not the end of the world, but also not a great thing to be hidden behind an interface like this (one would assume that this method would exist because it is actually efficient).
I think there are 3 options:
1) update the interface in a backwards incompatible manner and force concrete implementations to implement this method in a way that makes sense even if it's run in a loop
2) update the interface to `requiresColumns(Set<String> columnNames)` and have the default implementation iterate the list of required column names checking against the set
3) Don't add the method at all and have the places that need to check this do the check by calling `requiredColumns()` and then iterating that list.
##########
processing/src/main/java/org/apache/druid/segment/incremental/IncrementalIndexCursorHolder.java:
##########
@@ -73,10 +79,17 @@ public Cursor asCursor()
return new IncrementalIndexCursor(
rowSelector,
spec,
- Cursors.getTimeOrdering(ordering)
+ timeColumnName,
+ getTimeOrder(ordering, timeColumnName)
);
}
+ @Override
+ public boolean isPreAggregated()
+ {
+ return isPreaggregated;
+ }
+
Review Comment:
I was trying to understand what this did and looked at the code that's already been checked in for handling "isPreAggregated". I'm fairly certain that the code is buggy as it won't handle an aggregator that renames its output. The common operation that `getCombiningFactory()` does is rename the input to be the name of the output (because the aggregator would have renamed it), there is sometimes a secondary action that also specializes to the pre-aggregated type, but the aggregator itself must be able to handle both the pre-aggregated and the post-aggregation type in doing aggregations, so the way to handle the case where this is pre-aggregated is for the ColumnSelector to represent the correct type information to the aggregator rather than trying to latch onto some notion of things being pre-aggregated here...
##########
processing/src/main/java/org/apache/druid/segment/incremental/IncrementalIndexCursorFactory.java:
##########
@@ -77,7 +78,20 @@ public IncrementalIndexCursorFactory(IncrementalIndex index)
@Override
public CursorHolder makeCursorHolder(CursorBuildSpec spec)
{
- return new IncrementalIndexCursorHolder(index, spec);
+ final IncrementalIndexRowSelector rowSelector;
+ final CursorBuildSpec buildSpec;
+ final String timeColumnName;
+ final IncrementalIndex.ProjectionRowSelector projection = index.getProjection(spec);
+ if (projection == null) {
+ rowSelector = index;
+ buildSpec = spec;
+ timeColumnName = ColumnHolder.TIME_COLUMN_NAME;
+ } else {
+ rowSelector = projection.getRowSelector();
+ buildSpec = projection.getRewrittenBuildSpec();
+ timeColumnName = projection.getTimeColumnName();
+ }
+ return new IncrementalIndexCursorHolder(rowSelector, buildSpec, timeColumnName, projection != null);
Review Comment:
A style nit, but tbh, I think that it's much nicer to have the code look like
```
final IncrementalIndex.ProjectionRowSelector projection = index.getProjection(spec);
if (projection == null) {
return new IncrementalIndexCursorHolder(index, spec, ColumnHolder.TIME_COLUMN_NAME, false);
} else {
return new IncrementalIndexCursorHolder(
projection.getRowSelector(),
projection.getRewrittenBuildSpec(),
projection.getTimeColumnName(),
true
);
}
```
Additionally, I'd probably create a `PreAggregatedIncrementalIndexCursorHolder` to use for the pre-aggregated case instead of the boolean. Though, I'd then argue with myself about keeping the boolean and using methods to build a pre-aggregated or not (the passing of the boolean makes it hard to know what's the intended meaning of `true` or `false`). I think I'd fall on the side of creating the class for it because then the class name shows up in stacktraces and provides a hint that we are working with a projection.
##########
processing/src/main/java/org/apache/druid/segment/incremental/OnheapIncrementalIndex.java:
##########
@@ -158,6 +185,82 @@ public class OnheapIncrementalIndex extends IncrementalIndex
this.maxBytesPerRowForAggregators =
useMaxMemoryEstimates ? getMaxBytesPerRowForAggregators(incrementalIndexSchema) : 0;
this.useMaxMemoryEstimates = useMaxMemoryEstimates;
+
+ initializeProjections(incrementalIndexSchema, useMaxMemoryEstimates);
+ }
+
+ private void initializeProjections(IncrementalIndexSchema incrementalIndexSchema, boolean useMaxMemoryEstimates)
+ {
+ this.aggregateProjections = new ObjectAVLTreeSet<>(OnHeapAggregateProjection.COMPARATOR);
+ for (ProjectionSpec projectionSpec : incrementalIndexSchema.getProjections()) {
+ if (projectionSpec instanceof AggregateProjectionSpec) {
+ AggregateProjectionSpec schema = (AggregateProjectionSpec) projectionSpec;
+ final List<DimensionDesc> descs = new ArrayList<>();
+ final int[] parentDimIndex = new int[schema.getGroupingColumns().size()];
+ int i = 0;
+ int foundTimePosition = -1;
+ String timeColumnName = null;
+ for (DimensionSchema dimension : schema.getGroupingColumns()) {
+ if (ColumnHolder.TIME_COLUMN_NAME.equals(dimension.getName())) {
+ timeColumnName = dimension.getName();
+ foundTimePosition = i;
+ parentDimIndex[i] = -1;
+ } else {
+ VirtualColumn vc = schema.getVirtualColumns().getVirtualColumn(dimension.getName());
+ Granularity granularity = Granularities.fromVirtualColumn(vc);
+ if (granularity != null) {
+ timeColumnName = dimension.getName();
+ foundTimePosition = i;
+ parentDimIndex[i] = -1;
+ } else {
+ final DimensionDesc parent = getDimension(dimension.getName());
+ if (parent == null) {
+ final DimensionDesc childOnly = new DimensionDesc(
+ i++,
+ dimension.getName(),
+ dimension.getDimensionHandler(),
+ useMaxMemoryEstimates
+ );
+ descs.add(childOnly);
+ parentDimIndex[childOnly.getIndex()] = -1;
+ } else {
+
+ if (!dimension.getColumnType().equals(parent.getCapabilities().toColumnType())) {
+ throw InvalidInput.exception(
+ "projection[%s] contains dimension[%s] with different type[%s] than type[%s] in base table",
Review Comment:
Good error message!
##########
processing/src/main/java/org/apache/druid/segment/incremental/OnheapIncrementalIndex.java:
##########
@@ -1038,4 +1182,398 @@ public void clear()
facts.clear();
}
}
+
+ public static class OnHeapAggregateProjection implements IncrementalIndexRowSelector
Review Comment:
A curiosity question: the name of this seems to be specific to a Projection. But, like, you could argue that even the "primary" could also be a projection of the incoming data, which *could* show up in the code as having one more projection for the primary along with the other defined projections. Instead, you've taken the approach of having a different Projection class than the main IncrementalIndex. Is there a specific reason why *not* modeling the "primary" as just-another-projection ended up being the favored approach?
##########
processing/src/main/java/org/apache/druid/segment/CursorBuildSpec.java:
##########
@@ -166,6 +167,11 @@ public QueryMetrics<?> getQueryMetrics()
return queryMetrics;
}
+ public boolean isAggregate()
+ {
+ return !CollectionUtils.isNullOrEmpty(groupingColumns) || !CollectionUtils.isNullOrEmpty(aggregators);
Review Comment:
I find this boolean hard to keep track of. After thinking about it a lot, understand it. It's essentially the same as "are there aggregators? if so, return true, if not, check if there are grouping columns, if there are grouping columns and no aggregators, then assume that we are just naively dedupping with a grouping and moving on"
As I thought about if there was an easier to grok way of dealing with this, I think it might be easier to actually add a `boolean isAggregate` field that gets set to true when either groupingColumns or aggregators are set. This then just returns that. I think that'd be easier to understand as it's basically saying "if you set the groupingColumns or give me aggregators, then I'm an aggregate" which is essentially what we are trying to say. It also effectively memoizes the value, though I don't know that we will really build this over and over and over again...
##########
processing/src/main/java/org/apache/druid/segment/incremental/IncrementalIndexColumnSelectorFactory.java:
##########
@@ -47,18 +47,22 @@ class IncrementalIndexColumnSelectorFactory implements ColumnSelectorFactory, Ro
private final Order timeOrder;
private final IncrementalIndexRowHolder rowHolder;
private final IncrementalIndexRowSelector rowSelector;
+ @Nullable
+ private final String timeColumnName;
IncrementalIndexColumnSelectorFactory(
IncrementalIndexRowSelector rowSelector,
VirtualColumns virtualColumns,
+ @Nullable String timeColumnName,
Review Comment:
What is this stuff for? Is it documented why this will be non-null and what it's used for? If it is documented somewhere lower, maybe I'll see it...
##########
processing/src/main/java/org/apache/druid/data/input/ProjectionSpec.java:
##########
@@ -0,0 +1,41 @@
+/*
+ * 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.data.input;
+
+import com.fasterxml.jackson.annotation.JsonSubTypes;
+import com.fasterxml.jackson.annotation.JsonTypeInfo;
+import org.apache.druid.data.input.impl.AggregateProjectionSpec;
+import org.apache.druid.query.OrderBy;
+import org.apache.druid.segment.VirtualColumns;
+
+import java.util.List;
+
+@JsonTypeInfo(use = JsonTypeInfo.Id.NAME, property = "type")
+@JsonSubTypes({
+ @JsonSubTypes.Type(name = "aggregate", value = AggregateProjectionSpec.class)
+})
+public interface ProjectionSpec
Review Comment:
This is a stream-of-consciousness comment, but this interface has only one concrete implementation, right? Is there a reason it needs to be an interface? You could make it just a concrete class (and still force a `type` parameter to exist if you want) for now, until we have a second implementation that we want? Or, do you already have another implementation of this interface in mind that you are thinking we will use?
##########
processing/src/main/java/org/apache/druid/segment/Cursors.java:
##########
@@ -64,6 +64,26 @@ public static Order getTimeOrdering(final List<OrderBy> ordering)
}
}
+ public static boolean compatibleOrdering(CursorBuildSpec buildSpec, List<OrderBy> ordering)
+ {
+ // if the build spec doesn't prefer an ordering, any order is ok
+ if (buildSpec.getPreferredOrdering().isEmpty()) {
+ return true;
+ }
+ // all columns must be present in ordering if the build spec specifies them
+ if (ordering.size() < buildSpec.getPreferredOrdering().size()) {
+ return false;
+ }
+ for (int i = 0; i < buildSpec.getPreferredOrdering().size(); i++) {
+ final OrderBy preferredOrderBy = buildSpec.getPreferredOrdering().get(i);
Review Comment:
if `buildSpec.getPreferredOrdering()` were to ever do lazy computation of things, this is going to call that a lot of times and basically keep doing the same thing over and over again. Get it once and reuse it pls.
##########
processing/src/main/java/org/apache/druid/segment/incremental/OnheapIncrementalIndex.java:
##########
@@ -254,6 +357,10 @@ protected AddToFactsResult addToFacts(
final List<String> parseExceptionMessages = new ArrayList<>();
final AtomicLong totalSizeInBytes = getBytesInMemory();
+ for (OnHeapAggregateProjection projection : aggregateProjections) {
+ projection.addToFacts(key, inputRowHolder.getRow(), parseExceptionMessages, totalSizeInBytes);
+ }
+
Review Comment:
I haven't thought through if this generates a meaningful semantic issue or not, but I had been imagining that the projections are updated after the primary rather than before. Was this placed first for a specific reason?
##########
processing/src/main/java/org/apache/druid/segment/incremental/IncrementalIndex.java:
##########
@@ -1367,4 +1379,120 @@ public void inspectRuntimeShape(RuntimeShapeInspector inspector)
inspector.visit("index", rowSelector);
}
}
+
+ public static final class ProjectionRowSelector
+ {
+ private final IncrementalIndexRowSelector rowSelector;
+ private final CursorBuildSpec rewrittenBuildSpec;
+ private final Map<String, String> rewriteColumns;
+ private final String timeColumnName;
+
+ ProjectionRowSelector(
+ IncrementalIndexRowSelector rowSelector,
+ CursorBuildSpec rewrittenBuildSpec,
+ Map<String, String> rewriteColumns,
+ @Nullable String timeColumnName
+ )
+ {
+ this.rowSelector = rowSelector;
+ this.rewrittenBuildSpec = rewrittenBuildSpec;
+ this.rewriteColumns = rewriteColumns;
+ this.timeColumnName = timeColumnName == null ? ColumnHolder.TIME_COLUMN_NAME : timeColumnName;
+ }
+
+ public IncrementalIndexRowSelector getRowSelector()
+ {
+ if (rewriteColumns.isEmpty()) {
+ return rowSelector;
+ }
+ return new IncrementalIndexRowSelector()
+ {
+ @Nullable
+ @Override
+ public ColumnCapabilities getColumnCapabilities(String column)
+ {
+ return rowSelector.getColumnCapabilities(column);
+ }
+
+ @Override
+ public DimensionDesc getDimension(String columnName)
+ {
+ final String columnToUse = rewriteColumns.getOrDefault(columnName, columnName);
+ return rowSelector.getDimension(columnToUse);
+ }
+
+ @Override
+ public MetricDesc getMetric(String s)
+ {
+ final String columnToUse = rewriteColumns.getOrDefault(s, s);
+ return rowSelector.getMetric(columnToUse);
+ }
+
+ @Override
+ public List<OrderBy> getOrdering()
+ {
+ return rowSelector.getOrdering();
+ }
+
+ @Override
+ public boolean isEmpty()
+ {
+ return rowSelector.isEmpty();
+ }
+
+ @Override
+ public FactsHolder getFacts()
+ {
+ return rowSelector.getFacts();
+ }
+
+ @Override
+ public int getLastRowIndex()
+ {
+ return rowSelector.getLastRowIndex();
+ }
+
+ @Override
+ public float getMetricFloatValue(int rowOffset, int aggOffset)
+ {
+ return rowSelector.getMetricFloatValue(rowOffset, aggOffset);
+ }
+
+ @Override
+ public long getMetricLongValue(int rowOffset, int aggOffset)
+ {
+ return rowSelector.getMetricLongValue(rowOffset, aggOffset);
+ }
+
+ @Override
+ public double getMetricDoubleValue(int rowOffset, int aggOffset)
+ {
+ return rowSelector.getMetricDoubleValue(rowOffset, aggOffset);
+ }
+
+ @Nullable
+ @Override
+ public Object getMetricObjectValue(int rowOffset, int aggOffset)
+ {
+ return rowSelector.getMetricObjectValue(rowOffset, aggOffset);
+ }
+
+ @Override
+ public boolean isNull(int rowOffset, int aggOffset)
+ {
+ return rowSelector.isNull(rowOffset, aggOffset);
+ }
+ };
+ }
+
+ public CursorBuildSpec getRewrittenBuildSpec()
+ {
+ return rewrittenBuildSpec;
+ }
Review Comment:
This method feels weird on this object? Why would it be assumed that there is a rewritten build spec and what did the rewriting to what aim?
##########
processing/src/main/java/org/apache/druid/segment/incremental/OnheapIncrementalIndex.java:
##########
@@ -158,6 +185,82 @@ public class OnheapIncrementalIndex extends IncrementalIndex
this.maxBytesPerRowForAggregators =
useMaxMemoryEstimates ? getMaxBytesPerRowForAggregators(incrementalIndexSchema) : 0;
this.useMaxMemoryEstimates = useMaxMemoryEstimates;
+
+ initializeProjections(incrementalIndexSchema, useMaxMemoryEstimates);
+ }
+
+ private void initializeProjections(IncrementalIndexSchema incrementalIndexSchema, boolean useMaxMemoryEstimates)
+ {
+ this.aggregateProjections = new ObjectAVLTreeSet<>(OnHeapAggregateProjection.COMPARATOR);
+ for (ProjectionSpec projectionSpec : incrementalIndexSchema.getProjections()) {
+ if (projectionSpec instanceof AggregateProjectionSpec) {
Review Comment:
What's the point of the interface if it's just going to be used via `instanceof` checks here? I left a comment earlier asking if we really needed an interface, this is just more evidence that it's probably pre-mature for an interface at this point.
##########
processing/src/main/java/org/apache/druid/segment/incremental/OnheapIncrementalIndex.java:
##########
@@ -431,6 +538,43 @@ private void closeAggregators()
}
}
+ @Nullable
+ @Override
+ public ProjectionRowSelector getProjection(CursorBuildSpec buildSpec)
+ {
+ if (buildSpec.getQueryContext().getBoolean(QueryContexts.CTX_NO_PROJECTION, false)) {
+ return null;
+ }
Review Comment:
This seems like a weird place to check this. I think it's okay to check here, but I'd actually say that instead of returning `null` this should throw a defensive exception and the developer should change the code to stop calling this method if it's not supposed to use a projection. Is there a reason that it is hard for the calling code to know whether it's allowed to use projections or not?
##########
processing/src/main/java/org/apache/druid/segment/incremental/OnheapIncrementalIndex.java:
##########
@@ -158,6 +185,82 @@ public class OnheapIncrementalIndex extends IncrementalIndex
this.maxBytesPerRowForAggregators =
useMaxMemoryEstimates ? getMaxBytesPerRowForAggregators(incrementalIndexSchema) : 0;
this.useMaxMemoryEstimates = useMaxMemoryEstimates;
+
+ initializeProjections(incrementalIndexSchema, useMaxMemoryEstimates);
+ }
+
+ private void initializeProjections(IncrementalIndexSchema incrementalIndexSchema, boolean useMaxMemoryEstimates)
+ {
+ this.aggregateProjections = new ObjectAVLTreeSet<>(OnHeapAggregateProjection.COMPARATOR);
+ for (ProjectionSpec projectionSpec : incrementalIndexSchema.getProjections()) {
+ if (projectionSpec instanceof AggregateProjectionSpec) {
+ AggregateProjectionSpec schema = (AggregateProjectionSpec) projectionSpec;
+ final List<DimensionDesc> descs = new ArrayList<>();
+ final int[] parentDimIndex = new int[schema.getGroupingColumns().size()];
+ int i = 0;
+ int foundTimePosition = -1;
+ String timeColumnName = null;
+ for (DimensionSchema dimension : schema.getGroupingColumns()) {
+ if (ColumnHolder.TIME_COLUMN_NAME.equals(dimension.getName())) {
+ timeColumnName = dimension.getName();
+ foundTimePosition = i;
+ parentDimIndex[i] = -1;
+ } else {
+ VirtualColumn vc = schema.getVirtualColumns().getVirtualColumn(dimension.getName());
+ Granularity granularity = Granularities.fromVirtualColumn(vc);
+ if (granularity != null) {
+ timeColumnName = dimension.getName();
+ foundTimePosition = i;
+ parentDimIndex[i] = -1;
+ } else {
+ final DimensionDesc parent = getDimension(dimension.getName());
+ if (parent == null) {
+ final DimensionDesc childOnly = new DimensionDesc(
+ i++,
+ dimension.getName(),
+ dimension.getDimensionHandler(),
+ useMaxMemoryEstimates
+ );
+ descs.add(childOnly);
+ parentDimIndex[childOnly.getIndex()] = -1;
+ } else {
+
+ if (!dimension.getColumnType().equals(parent.getCapabilities().toColumnType())) {
+ throw InvalidInput.exception(
+ "projection[%s] contains dimension[%s] with different type[%s] than type[%s] in base table",
+ schema.getName(),
+ dimension.getName(),
+ dimension.getColumnType(),
+ parent.getCapabilities().toColumnType()
+ );
+ }
+ final DimensionDesc rewrite = new DimensionDesc(
+ i++,
+ parent.getName(),
+ parent.getHandler(),
+ parent.getIndexer()
+ );
+ descs.add(rewrite);
Review Comment:
Why is it called "rewrite"? Is it getting renamed or something? As I read the code, I feel like there's some game going on with naming of things, but I'm unsure of what it is. Is there something to be aware of with naming?
##########
processing/src/main/java/org/apache/druid/segment/incremental/OnheapIncrementalIndex.java:
##########
@@ -1038,4 +1182,398 @@ public void clear()
facts.clear();
}
}
+
+ public static class OnHeapAggregateProjection implements IncrementalIndexRowSelector
+ {
+ static Comparator<OnHeapAggregateProjection> COMPARATOR = (o1, o2) -> {
+ // coarsest granularity first
+ if (o1.projectionGranularity.isFinerThan(o2.projectionGranularity)) {
+ return 1;
+ }
+ if (o2.projectionGranularity.isFinerThan(o1.projectionGranularity)) {
+ return -1;
+ }
+ // fewer dimensions first
+ final int dimsCompare = Integer.compare(
+ o1.schema.getGroupingColumns().size(),
+ o2.schema.getGroupingColumns().size()
+ );
+ if (dimsCompare != 0) {
+ return dimsCompare;
+ }
+ // more metrics first
+ int metCompare = Integer.compare(o2.schema.getAggregators().length, o1.schema.getAggregators().length);
+ if (metCompare != 0) {
+ return metCompare;
+ }
+ // more virtual columns first
+ final int virtCompare = Integer.compare(
+ o2.schema.getVirtualColumns().getVirtualColumns().length,
+ o1.schema.getVirtualColumns().getVirtualColumns().length
+ );
+ if (virtCompare != 0) {
+ return virtCompare;
+ }
+ return o1.schema.getName().compareTo(o2.schema.getName());
+ };
+
+ private final AggregateProjectionSpec schema;
+ private final Granularity projectionGranularity;
+ private final boolean hasTimeColumn;
+ @Nullable
+ private final String timeColumnName;
+ private final long minTimestamp;
+ private final List<DimensionDesc> dimensions;
+ private final int[] parentDimensionIndex;
+ private final Map<String, DimensionDesc> dimensionsMap;
+ private final Map<String, MetricDesc> aggregatorsMap;
+ private final List<OrderBy> ordering;
+ private final FactsHolder factsHolder;
+ private final InputRowHolder inputRowHolder = new InputRowHolder();
+ private final Map<String, ColumnSelectorFactory> aggSelectors;
+ private final ColumnSelectorFactory virtualSelectorFactory;
+ private final ConcurrentHashMap<Integer, Aggregator[]> aggregators = new ConcurrentHashMap<>();
+ private final ColumnInspector baseTableInspector;
+ private final AtomicInteger rowCounter = new AtomicInteger(0);
+ private final boolean useMaxMemoryEstimates;
+ private final long maxBytesPerRowForAggregators;
+
+
+ public OnHeapAggregateProjection(
+ AggregateProjectionSpec schema,
+ List<DimensionDesc> dimensions,
+ int[] parentDimensionIndex,
+ ColumnInspector baseTableInspector,
+ @Nullable String timeColumnName,
+ int foundTimePosition,
+ long minTimestamp,
+ boolean useMaxMemoryEstimates,
+ long maxBytesPerRowForAggregators
+ )
+ {
+ this.schema = schema;
+ this.dimensions = dimensions;
+ this.parentDimensionIndex = parentDimensionIndex;
+ this.dimensionsMap = new HashMap<>();
+ for (DimensionDesc desc : dimensions) {
+ dimensionsMap.put(desc.getName(), desc);
+ }
+ this.baseTableInspector = baseTableInspector;
+ this.timeColumnName = timeColumnName;
+ Granularity virtualGranularity = null;
+ for (VirtualColumn vc : schema.getVirtualColumns().getVirtualColumns()) {
+ if (vc.requiresColumn(ColumnHolder.TIME_COLUMN_NAME) && vc.requiredColumns().size() == 1) {
+ virtualGranularity = Granularities.fromVirtualColumn(vc);
+ break;
+ }
+ }
Review Comment:
If there are multiple time floors in the same projection, the rule is that the first one is the virtual granularity? That sounds correct as this is also the sort-order for the rows right? so the first one will take precedence. It might be worth it to explain that in a comment or move this to a static method that can have method-level javadoc or something like that.
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] projections part 1 - realtime segments (druid)
Posted by "clintropolis (via GitHub)" <gi...@apache.org>.
clintropolis commented on code in PR #17117:
URL: https://github.com/apache/druid/pull/17117#discussion_r1771916421
##########
processing/src/main/java/org/apache/druid/segment/Cursors.java:
##########
@@ -64,6 +64,26 @@ public static Order getTimeOrdering(final List<OrderBy> ordering)
}
}
+ public static boolean compatibleOrdering(CursorBuildSpec buildSpec, List<OrderBy> ordering)
+ {
+ // if the build spec doesn't prefer an ordering, any order is ok
+ if (buildSpec.getPreferredOrdering().isEmpty()) {
+ return true;
+ }
+ // all columns must be present in ordering if the build spec specifies them
+ if (ordering.size() < buildSpec.getPreferredOrdering().size()) {
+ return false;
+ }
+ for (int i = 0; i < buildSpec.getPreferredOrdering().size(); i++) {
+ final OrderBy preferredOrderBy = buildSpec.getPreferredOrdering().get(i);
Review Comment:
i don't imagine that it will ever be lazy since its just populated by a builder from the query engines, but this problem went away moving the function to `CursorBuildSpec`
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org