DRILL-4653.json - Malformed JSON should not stop the entire query from progressing

exec/java-exec/src/main/java/org/apache/drill/exec/ExecConstants.java

@@ -135,6 +135,9 @@ public interface ExecConstants {
   BooleanValidator JSON_EXTENDED_TYPES = new BooleanValidator("store.json.extended_types", false);
   BooleanValidator JSON_WRITER_UGLIFY = new BooleanValidator("store.json.writer.uglify", false);
   BooleanValidator JSON_WRITER_SKIPNULLFIELDS = new BooleanValidator("store.json.writer.skip_null_fields", true);
+  String JSON_READER_SKIP_MALFORMED_RECORDS_FLAG = "store.json.reader.skip_malformed_records";
+  BooleanValidator JSON_SKIP_MALFORMED_RECORDS_VALIDATOR = new BooleanValidator(JSON_READER_SKIP_MALFORMED_RECORDS_FLAG, false);
+
 
   DoubleValidator TEXT_ESTIMATED_ROW_SIZE = new RangeDoubleValidator(
       "store.text.estimated_row_size_bytes", 1, Long.MAX_VALUE, 100.0);

exec/java-exec/src/main/java/org/apache/drill/exec/server/options/SystemOptionManager.java

@@ -101,6 +101,7 @@ public class SystemOptionManager extends BaseOptionManager implements AutoClosea
       ExecConstants.JSON_WRITER_UGLIFY,
       ExecConstants.JSON_WRITER_SKIPNULLFIELDS,
       ExecConstants.JSON_READ_NUMBERS_AS_DOUBLE_VALIDATOR,
+      ExecConstants.JSON_SKIP_MALFORMED_RECORDS_VALIDATOR,
       ExecConstants.FILESYSTEM_PARTITION_COLUMN_LABEL_VALIDATOR,
       ExecConstants.MONGO_READER_ALL_TEXT_MODE_VALIDATOR,
       ExecConstants.MONGO_READER_READ_NUMBERS_AS_DOUBLE_VALIDATOR,

exec/java-exec/src/main/java/org/apache/drill/exec/store/easy/json/JSONRecordReader.java

@@ -64,6 +64,8 @@ public class JSONRecordReader extends AbstractRecordReader {
   private final boolean enableAllTextMode;
   private final boolean readNumbersAsDouble;
   private final boolean unionEnabled;
+  private int parseErrorCount;
+  private final boolean skipMalformedJSONRecords;
 
   /**
    * Create a JSON Record Reader that uses a file based input stream.
@@ -114,6 +116,7 @@ private JSONRecordReader(final FragmentContext fragmentContext, final String inp
     this.enableAllTextMode = embeddedContent == null && fragmentContext.getOptions().getOption(ExecConstants.JSON_READER_ALL_TEXT_MODE_VALIDATOR);
     this.readNumbersAsDouble = embeddedContent == null && fragmentContext.getOptions().getOption(ExecConstants.JSON_READ_NUMBERS_AS_DOUBLE_VALIDATOR);
     this.unionEnabled = embeddedContent == null && fragmentContext.getOptions().getOption(ExecConstants.ENABLE_UNION_TYPE);
+    this.skipMalformedJSONRecords = fragmentContext.getOptions().getOption(ExecConstants.JSON_SKIP_MALFORMED_RECORDS_VALIDATOR);
     setColumns(columns);
   }
 
@@ -122,7 +125,8 @@ public String toString() {
     return super.toString()
         + "[hadoopPath = " + hadoopPath
         + ", recordCount = " + recordCount
-        + ", runningRecordCount = " + runningRecordCount + ", ...]";
+        + ", parseErrorCount = " + parseErrorCount
+         + ", runningRecordCount = " + runningRecordCount + ", ...]";
   }
 
   @Override
@@ -189,26 +193,33 @@ private long currentRecordNumberInFile() {
   public int next() {
     writer.allocate();
     writer.reset();
-
     recordCount = 0;
     ReadState write = null;
 //    Stopwatch p = new Stopwatch().start();
-    try{
-      outside: while(recordCount < DEFAULT_ROWS_PER_BATCH) {
-        writer.setPosition(recordCount);
-        write = jsonReader.write(writer);
-
-        if(write == ReadState.WRITE_SUCCEED) {
+   // try
+   // {
+      outside: while(recordCount < DEFAULT_ROWS_PER_BATCH){
+      try{
+            writer.setPosition(recordCount);
+            write = jsonReader.write(writer);
+            if(write == ReadState.WRITE_SUCCEED) {
 //          logger.debug("Wrote record.");
-          recordCount++;
-        }else{
+              recordCount++;
+            }else{
 //          logger.debug("Exiting.");
-          break outside;
-        }
-
+              break outside;
+            }
+      }
+      catch(Exception ex)
+      {
+           if(skipMalformedJSONRecords == false){
+              handleAndRaise("Error parsing JSON", ex);
+           }
+           ++parseErrorCount;
       }
+     }
 
-      jsonReader.ensureAtLeastOneField(writer);
+     jsonReader.ensureAtLeastOneField(writer);
 
       writer.setValueCount(recordCount);
 //      p.stop();
@@ -217,11 +228,11 @@ public int next() {
       updateRunningCount();
       return recordCount;
 
-    } catch (final Exception e) {
-      handleAndRaise("Error parsing JSON", e);
-    }
+   // } catch (final Exception e) {
+   //   handleAndRaise("Error parsing JSON", e);
+   // }
     // this is never reached
-    return 0;
+    //return 0;
   }
 
   private void updateRunningCount() {

exec/java-exec/src/test/java/org/apache/drill/exec/store/json/TestJsonRecordReader.java

@@ -20,6 +20,7 @@
 import org.apache.drill.BaseTestQuery;
 import org.apache.drill.common.exceptions.UserException;
 import org.apache.drill.exec.proto.UserBitShared;
+import org.apache.drill.exec.ExecConstants;
 import org.junit.Test;
 import org.junit.Assert;
 
@@ -179,4 +180,27 @@ public void testNestedFilter() throws Exception {
         .sqlBaselineQuery(baselineQuery)
         .go();
   }
+
+ @Test // See DRILL-4653
+  public void testSkippingInvalidJSONRecords() throws Exception {
+    String set = "alter session set `" + ExecConstants.JSON_READER_SKIP_MALFORMED_RECORDS_FLAG+ "` = true";
+    testNoResult(set);
+    test("select count(*) from cp.`jsoninput/DRILL-4653.json`");
+    set = "alter session set `" + ExecConstants.JSON_READER_SKIP_MALFORMED_RECORDS_FLAG+ "` = false";
+    testNoResult(set);
+  }
+
+  @Test // See DRILL-4653
+  public void testNotSkippingInvalidJSONRecords() throws Exception {
+      try
+      {
+      test("select count(*) from cp.`jsoninput/DRILL-4653.json`");
+      }
+      catch(Exception ex)
+      {
+        // do nothing just return
+        return;
+      }
+      throw new Exception("testNotSkippingInvalidJSONRecords");
+  }
 }

exec/java-exec/src/test/resources/jsoninput/DRILL-4653.json

@@ -0,0 +1,40 @@
+{ "integer" : 2010,
+  "float"   : 17.4,
+  "x": {
+    "y": "kevin",
+    "z": "paul"
+  },
+  "z": [
+    {"orange" : "yellow" , "pink": "red"},
+    {"pink" : "purple" }
+  ],
+  "l": [4,2],
+  "rl": [ [2,1], [4,6] ]
+}
+{ "integer : -2002,
+  "float"   : -1.2
+}
+{ "integer" : 2001,
+  "float"   : 1.2,
+  "x": {
+    "y": "bill",
+    "z": "peter"
+  },
+  "z": [
+    {"pink" : "lilac" }
+  ],
+  "l": [4,2],
+  "rl": [ [2,1], [4,6] ]
+}
+{ "integer" : 6005,
+  "float"   : 1.2,
+  "x": {
+    "y": "mike",
+    "z": "mary"
+  },
+  "z": [
+    {"orange" : "stucco" }
+  ],
+  "l": [4,2],
+  "rl": [ [2,1], [4,6] ]
+}