Add GroupNode and parallelize TableFunctionProcessorNode #15136
Conversation
Codecov ReportAttention: Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## master #15136 +/- ##
============================================
+ Coverage 39.40% 39.41% +0.01%
Complexity 193 193
============================================
Files 4612 4630 +18
Lines 296784 298303 +1519
Branches 37045 37220 +175
============================================
+ Hits 116938 117568 +630
- Misses 179846 180735 +889 ☔ View full report in Codecov by Sentry. 🚀 New features to boost your workflow:
|
Cpaulyz
changed the title
| compareKey.rowIndex = low + (high - low) / 2; | ||
| int cmp = partitionComparator.compare(currentPartitionKey, compareKey); | ||
| if (cmp == 0) { | ||
| low = compareKey.rowIndex + 1; |
There was a problem hiding this comment.
need update low? otherwise, it will be dead-loop
| private ListenableFuture<?> isBlocked; | ||
| private boolean finished = false; | ||
|
|
||
| private Queue<TsBlock> resultTsBlocks; |
There was a problem hiding this comment.
| private Queue<TsBlock> resultTsBlocks; | |
| private final Queue<TsBlock> resultTsBlocks; |
| @Override | ||
| protected void serializeAttributes(ByteBuffer byteBuffer) { | ||
| PlanNodeType.TABLE_GROUP_NODE.serialize(byteBuffer); | ||
| orderingScheme.serialize(byteBuffer); |
There was a problem hiding this comment.
better add a new protected method in SortNode, to extract these common code together, then if SortNode add some new attributes, we won't need to change subClass of it
There was a problem hiding this comment.
StreamSortNode has the same issue. Actually, there are some variables in SortNode is not used in StreamSortNode and GroupNode. So if SortNode add some new attributes, maybe it won't affact the logic of subClass?
GroupNode is a subClass of SortNode, so I think it projection push down will be executed in |
| private static class CountDataProcessor implements TableFunctionDataProcessor { | ||
|
|
||
| private final long size; | ||
| private final List<Long> currentRowIndexes = new ArrayList<>(); |
There was a problem hiding this comment.
there is no need to save this list? we can simplely record a curStartIndex and then [startStartIndex, curIndex] is for passThroughIndexBuilder
There was a problem hiding this comment.
Good idea! Change List<Long> currentRowIndexes to long currentStartIndex
| private long windowStart = -1; | ||
| private long windowEnd = -1; |
There was a problem hiding this comment.
what if time is negative? better using Long.MIN_VALUE
| private static class SessionDataProcessor implements TableFunctionDataProcessor { | ||
|
|
||
| private final long gap; | ||
| private final List<Long> currentRowIndexes = new ArrayList<>(); |
There was a problem hiding this comment.
no need to use list to save all
| for (Long currentRowIndex : currentRowIndexes) { | ||
| properColumnBuilders.get(0).writeLong(windowStart); | ||
| properColumnBuilders.get(1).writeLong(windowEnd - gap); | ||
| passThroughIndexBuilder.writeLong(currentRowIndex); | ||
| } |
There was a problem hiding this comment.
| for (Long currentRowIndex : currentRowIndexes) { | |
| properColumnBuilders.get(0).writeLong(windowStart); | |
| properColumnBuilders.get(1).writeLong(windowEnd - gap); | |
| passThroughIndexBuilder.writeLong(currentRowIndex); | |
| } | |
| long currentWindowEnd = windowEnd - gap; | |
| for (Long currentRowIndex : currentRowIndexes) { | |
| properColumnBuilders.get(0).writeLong(windowStart); | |
| properColumnBuilders.get(1).writeLong(currentWindowEnd); | |
| passThroughIndexBuilder.writeLong(currentRowIndex); | |
| } |
| throw new UDFException( | ||
| String.format("The type of the column [%s] is not as expected.", expectedFieldName)); |
There was a problem hiding this comment.
| throw new UDFException( | |
| String.format("The type of the column [%s] is not as expected.", expectedFieldName)); | |
| throw new IoTDBRuntimeException(String.format("The type of the column [%s] is not as expected.", expectedFieldName), TSStatusCode.SEMANTIC_ERROR.getStatusCode()); |
| throw new UDFException( | ||
| String.format( | ||
| "Required column [%s] not found in the source table argument.", expectedFieldName)); |
There was a problem hiding this comment.
| throw new UDFException( | |
| String.format( | |
| "Required column [%s] not found in the source table argument.", expectedFieldName)); | |
| throw new IoTDBRuntimeException( | |
| String.format( | |
| "Required column [%s] not found in the source table argument.", expectedFieldName), TSStatusCode.SEMANTIC_ERROR.getStatusCode()); |
| private static class VariationDataProcessor implements TableFunctionDataProcessor { | ||
|
|
||
| private final double gap; | ||
| private final List<Long> currentRowIndexes = new ArrayList<>(); |
| public class ParallelizeGrouping implements PlanOptimizer { | ||
| @Override | ||
| public PlanNode optimize(PlanNode plan, PlanOptimizer.Context context) { | ||
| if (!(context.getAnalysis().isQuery())) { |
There was a problem hiding this comment.
only query plan with GroupNode should continue
There was a problem hiding this comment.
How can I know if query plan is with GroupNode or not?
| List<ColumnBuilder> properColumnBuilders, | ||
| ColumnBuilder passThroughIndexBuilder) { | ||
| if (currentRowIndexes.size() >= size) { | ||
| if (windowIndex - curIndex == size) { |
There was a problem hiding this comment.
| if (windowIndex - curIndex == size) { | |
| if (curIndex - currentStartIndex == size) { |
|
2 participants
Description