[ https://issues.apache.org/jira/browse/DRILL-6516?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16535491#comment-16535491 ]
ASF GitHub Bot commented on DRILL-6516: --------------------------------------- Ben-Zvi commented on a change in pull request #1358: DRILL-6516: EMIT support in streaming agg URL: https://github.com/apache/drill/pull/1358#discussion_r199973893 ########## File path: exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/aggregate/StreamingAggTemplate.java ########## @@ -189,83 +209,128 @@ else if (isSame( previousIndex, currentIndex )) { logger.debug("Received IterOutcome of {}", out); } switch (out) { - case NONE: - done = true; - lastOutcome = out; - if (first && addedRecordCount == 0) { - return setOkAndReturn(); - } else if (addedRecordCount > 0) { - outputToBatchPrev(previous, previousIndex, outputCount); // No need to check the return value - // (output container full or not) as we are not going to insert any more records. - if (EXTRA_DEBUG) { - logger.debug("Received no more batches, returning."); + case NONE: + done = true; + lastOutcome = out; + if (firstBatchForDataSet && addedRecordCount == 0) { + return setOkAndReturn(out); + } else if (addedRecordCount > 0) { + outputToBatchPrev(previous, previousIndex, outputCount); // No need to check the return value + // (output container full or not) as we are not going to insert any more records. + if (EXTRA_DEBUG) { + logger.debug("Received no more batches, returning."); + } + return setOkAndReturn(out); + } else { + // not first batch and record Count == 0 + outcome = out; + return AggOutcome.CLEANUP_AND_RETURN; } - return setOkAndReturn(); - } else { - if (first && out == IterOutcome.OK) { - out = IterOutcome.OK_NEW_SCHEMA; + // EMIT is handled like OK, except that we do not loop back to process the + // next incoming batch; we return instead + case EMIT: + if (incoming.getRecordCount() == 0) { + if (addedRecordCount > 0) { + outputToBatchPrev(previous, previousIndex, outputCount); + } + resetIndex(); + return setOkAndReturn(out); + } else { + resetIndex(); + if (previousIndex != -1 && isSamePrev(previousIndex, previous, currentIndex)) { + if (EXTRA_DEBUG) { + logger.debug("New value was same as last value of previous batch, adding."); + } + addRecordInc(currentIndex); + previousIndex = currentIndex; + incIndex(); + if (EXTRA_DEBUG) { + logger.debug("Continuing outside"); + } + processRemainingRecordsInBatch(); + // currentIndex has been reset to int_max so use previous index. + outputToBatch(previousIndex); + resetIndex(); + return setOkAndReturn(out); + } else { // not the same + if (EXTRA_DEBUG) { + logger.debug("This is not the same as the previous, add record and continue outside."); + } + if (addedRecordCount > 0) { + if (outputToBatchPrev(previous, previousIndex, outputCount)) { + if (EXTRA_DEBUG) { + logger.debug("Output container is full. flushing it."); + } + return setOkAndReturn(out); + } + } + previousIndex = -1; + processRemainingRecordsInBatch(); + outputToBatch(previousIndex); // currentIndex has been reset to int_max so use previous index. + resetIndex(); + return setOkAndReturn(out); Review comment: The above 4 lines of code are identical in the two sides of the if() - can take them out outside the if(). Actually the last 2 are even common with the outer if() - can be called only once at the end of the case EMIT. ---------------------------------------------------------------- This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org > Support for EMIT outcome in streaming agg > ----------------------------------------- > > Key: DRILL-6516 > URL: https://issues.apache.org/jira/browse/DRILL-6516 > Project: Apache Drill > Issue Type: Improvement > Reporter: Parth Chandra > Assignee: Parth Chandra > Priority: Major > Fix For: 1.14.0 > > > Update the streaming aggregator to recognize the EMIT outcome -- This message was sent by Atlassian JIRA (v7.6.3#76005)