[timeseries] Part-3.1: Add Support for Partial Aggregate and Complex Intermediate Type #14631
Conversation
ankitsultana
added
the
timeseries-engine
Codecov ReportAttention: Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## master #14631 +/- ##
============================================
+ Coverage 61.75% 63.97% +2.22%
- Complexity 207 1600 +1393
============================================
Files 2436 2696 +260
Lines 133233 148442 +15209
Branches 20636 22751 +2115
============================================
+ Hits 82274 94971 +12697
- Misses 44911 46502 +1591
- Partials 6048 6969 +921
Flags with carried forward coverage won't be shown. Click here to find out more. ☔ View full report in Codecov by Sentry. |
pinot-timeseries/pinot-timeseries-spi/src/main/java/org/apache/pinot/tsdb/spi/AggInfo.java
Show resolved
Hide resolved
| } else { | ||
| Preconditions.checkState(!currentAggInfo.getIsPartial(), | ||
| "Leaf node in the logical plan should not have partial agg"); | ||
| context._fragments.add(leafNode.withAggInfo(currentAggInfo.withPartialAggregation())); |
There was a problem hiding this comment.
If currentAggInfo isPartial is false, then shouldn't we use currentAggInfo.withFullAggregation here?
There was a problem hiding this comment.
The leaf node created by the users should have agg-mode=full, but the leaf node generated by Pinot should have partial aggregation since the TimeSeriesExchangeNode will have the full aggregation.
There's a cleanup pending here: since users are returning Logical plan, we shouldn't allow them to choose between partial or full aggregation. But implementation wise it's a bit tricky.. but I should add a TODO for this (I thought I already did).
...lang/pinot-timeseries-m3ql/src/main/java/org/apache/pinot/tsdb/m3ql/M3TimeSeriesPlanner.java
Show resolved
Hide resolved
ankitsultana
changed the title
| @@ -102,8 +104,15 @@ public static List<BaseTimeSeriesPlanNode> getFragments(BaseTimeSeriesPlanNode r | |||
| private static BaseTimeSeriesPlanNode fragmentRecursively(BaseTimeSeriesPlanNode planNode, Context context) { | |||
| if (planNode instanceof LeafTimeSeriesPlanNode) { | |||
| LeafTimeSeriesPlanNode leafNode = (LeafTimeSeriesPlanNode) planNode; | |||
| context._fragments.add(leafNode.withInputs(Collections.emptyList())); | |||
| return new TimeSeriesExchangeNode(planNode.getId(), Collections.emptyList(), leafNode.getAggInfo()); | |||
| AggInfo currentAggInfo = leafNode.getAggInfo(); | |||
There was a problem hiding this comment.
Do Users have to specify or set isPartial flag while creating LeafPlanNode?
There was a problem hiding this comment.
You have to specify, but you should always set it to false. There's a TODO in AggInfo regarding this. Ideally logical plan shouldn't be aware of this since this is a physical detail. I'll fix it as part of MSE integration.
| * TODO(timeseries): Ideally we should remove this from the logical plan completely. | ||
| * </p> | ||
| */ | ||
| private final boolean _isPartial; |
There was a problem hiding this comment.
Based on the _isPartial flag Users are expected to run SeriesBuilderoperations. For example if _isPartial = true merge series based on function which would produce final result for leaf phase.
Am I correct?
There was a problem hiding this comment.
Yup. Note that you should set isPartial to false in the logical plan that you return. Pinot will automatically create a partial AggInfo when the execution is split between the brokers and servers.
|
@ankitsultana One more change we discussed was passing raw timestamps along with values to SeriesBuilder for each time bucket . I do not see that change are you planning to do it in next iteration? |
Yup that'll be in a follow-up. I'll wrap up the broker reduce support first. |
Adds support for partial aggregates and complex types in
TimeSeries. This is required now because we will be soon adding support for data transfer from the server to the broker, and certain aggregation functions require storing data in complex types until the final aggregate can be run.Examples of these functions are
rate,irate, etc. in PromQL.Summary of Changes
isPartialparameter toAggInfo.AggInfowithisPartial=true, and theTimeSeriesExchangeNodewill have the same but withisPartial=false.TimeSeriescontainsbyte[][]values, then theTimeSeriesBlockSerdewill Hex encode them and send them asString[]to the broker. This is inefficient but should be good enough for now.How to Build Complex Series Builders
If you want to build a complex series builder, where complex refers to the fact that the intermediate aggregate will need to be stored in a non-double type, you can do the following:
BaseTimeSeriesBuilder, in thebuildcall, generate theTimeSerieswithbyte[][]values if the aggregate was partial.mergeAlignedSeriesfor such series builders to handlebyte[][]values.Test Plan
Added unit tests to cover testing for TimeSeries serde. Also ran quickstart to verify E2E integration works as before.