xiangfu0 commented on code in PR #19264:
URL: https://github.com/apache/pinot/pull/19264#discussion_r4225290532
##########
pinot-core/src/main/java/org/apache/pinot/core/plan/maker/InstancePlanMakerImplV2.java:
##########
@@ -422,6 +427,89 @@ protected CombinePlanNode
createCombinePlanNode(List<PlanNode> planNodes, QueryC
return new CombinePlanNode(planNodes, queryContext, executorService,
streamer);
}
+ /// Only emit BASE records when every local BASE key and every derived
grouping-set key can fit the combine
+ /// table. Summing each segment's dictionary product is conservative even
when keys overlap across segments.
+ /// Unknown, raw, transformed and multi-value columns use the per-row
expansion path. This also makes a larger
+ /// `groupingSetsBaseAggregationMaxGroups` option unable to bypass the
actual `numGroupsLimit` memory guard.
+ @VisibleForTesting
+ static void boundGroupingSetsBaseAggregation(List<SegmentContext>
segmentContexts, QueryContext queryContext) {
+ if (queryContext.isGroupingSetsBaseAggregation()) {
+ queryContext.setGroupingSetsBaseAggregationAllowed(
+ fitsGroupingSetsBaseAggregationLimit(segmentContexts, queryContext));
+ }
+ }
+
+ @VisibleForTesting
+ static boolean fitsGroupingSetsBaseAggregationLimit(List<SegmentContext>
segmentContexts, QueryContext queryContext) {
+ List<ExpressionContext> groupByExpressions =
queryContext.getGroupByExpressions();
+ List<int[]> groupingSets = queryContext.getGroupingSets();
+ int limit = queryContext.getNumGroupsLimit();
+ if (groupByExpressions == null || groupingSets == null || limit <= 0) {
+ return false;
+ }
+
+ long remainingBase = limit;
+ long remainingDerived = limit;
+ for (SegmentContext segmentContext : segmentContexts) {
+ IndexSegment segment = segmentContext.getIndexSegment();
+ if (segment.getSegmentMetadata().getTotalDocs() == 0) {
+ continue;
+ }
+ long[] cardinalities = new long[groupByExpressions.size()];
+ for (int i = 0; i < cardinalities.length; i++) {
+ ExpressionContext expression = groupByExpressions.get(i);
+ if (expression.getType() != ExpressionContext.Type.IDENTIFIER) {
+ return false;
+ }
+ DataSource dataSource =
segment.getDataSourceNullable(expression.getIdentifier());
+ if (dataSource == null ||
!dataSource.getDataSourceMetadata().isSingleValue()) {
+ return false;
+ }
+ Dictionary dictionary = dataSource.getDictionary();
+ ForwardIndexReader<?> forwardIndex = dataSource.getForwardIndex();
+ if (dictionary == null || forwardIndex == null ||
!forwardIndex.isDictionaryEncoded()) {
+ return false;
+ }
+ long cardinality = dictionary.length();
Review Comment:
Good catch — the bound has to hold at execution time, and a consuming
segment's dictionaries keep growing until `FilterPlanNode.run()` snapshots the
doc count, so a proof taken at plan time is not stable for it. Went with your
first suggestion: `fitsGroupingSetsBaseAggregationLimit` now returns `false`
whenever any segment is a `MutableSegment`, so those queries use the exact
per-row expansion path (headroom would just be a heuristic — a steady flow of
new keys can outrun any fixed margin, and your repro shape is exactly that).
This also closes the same staleness hole in the zero-doc skip: a consuming
segment with 0 docs at plan time could have docs by execution.
Added `testMutableSegmentDisablesBaseAggregation` covering a lone mutable
segment and a mixed immutable+mutable list under a limit with plenty of
headroom.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
