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_r199980234
##########
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.");
+ }
Review comment:
Should `previousIndex` be set to -1 here ? (instead of inside
`setOkAndReturn()`, where it seems "out of place" )
----------------------------------------------------------------
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:
[email protected]
With regards,
Apache Git Services
