[drill] 01/04: DRILL-6918: Skip ensureAtLeastOneField when there are no records

[so...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

sorabh pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/drill.git

commit 4b66226dc7e33d9173e6e714d5cf85beea1e4e46
Author: Abhishek Ravi <ab...@gmail.com>
AuthorDate: Fri Jan 4 00:04:32 2019 -0800

    DRILL-6918: Skip ensureAtLeastOneField when there are no records
    
    If none of the project / filter columns, exist in the vector, ensureAtLeastOneField (or the Scan operator) adds at least one field as nullable integer (or nullable varchar if `allTextmode` is enabled).
    
    The downstream Filter operator would then go on to fail with `NumberFormatException` because it tries to convert empty fields to integers.
    
    Since ensureAtLeastOneField is called after reading all the messages in a batch, it can be skipped if the batch is empty.
    
    closes #1595
---
 .../java/org/apache/drill/exec/store/kafka/KafkaRecordReader.java     | 4 +++-
 1 file changed, 3 insertions(+), 1 deletion(-)

diff --git a/contrib/storage-kafka/src/main/java/org/apache/drill/exec/store/kafka/KafkaRecordReader.java b/contrib/storage-kafka/src/main/java/org/apache/drill/exec/store/kafka/KafkaRecordReader.java
index 9559c3d..62e588c 100644
--- a/contrib/storage-kafka/src/main/java/org/apache/drill/exec/store/kafka/KafkaRecordReader.java
+++ b/contrib/storage-kafka/src/main/java/org/apache/drill/exec/store/kafka/KafkaRecordReader.java
@@ -119,7 +119,9 @@ public class KafkaRecordReader extends AbstractRecordReader {
         }
       }
 
-      messageReader.ensureAtLeastOneField();
+      if (currentMessageCount > 0) {
+        messageReader.ensureAtLeastOneField();
+      }
       writer.setValueCount(currentMessageCount);
       logger.debug("Took {} ms to process {} records.", watch.elapsed(TimeUnit.MILLISECONDS), currentMessageCount);
       logger.debug("Last offset consumed for {}:{} is {}", subScanSpec.getTopicName(), subScanSpec.getPartitionId(),