DRILL-2879: Part 2 - Enhancing extended json support for date in millies and binary with type info

Addressing review comments

Updated unit test to remove timezone that was being pulled from the local system
(and thus failed to match the baseline if run from a different timezone)

4 files changed, 52 insertions, 4 deletions

diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/util/VectorUtil.java b/exec/java-exec/src/main/java/org/apache/drill/exec/util/VectorUtil.java
index cd1d79b26..efbd30d56 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/util/VectorUtil.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/util/VectorUtil.java
@@ -27,6 +27,8 @@ import org.apache.drill.exec.record.VectorWrapper;
 
 import com.google.common.base.Joiner;
 import com.google.common.collect.Lists;
+import org.joda.time.DateTime;
+import org.joda.time.format.DateTimeFormat;
 
 public class VectorUtil {
 
@@ -96,6 +98,10 @@ public class VectorUtil {
           rowValues.add("null");
         } else if (o instanceof byte[]) {
           rowValues.add(new String((byte[]) o));
+        } else if (o instanceof DateTime) {
+          // TODO(DRILL-3882) - remove this once the datetime is not returned in an
+          // object needlessly holding a timezone
+          rowValues.add(DateTimeFormat.forPattern("yyyy-MM-dd'T'HH:mm:ss.SSS").print((DateTime) o));
         } else {
           rowValues.add(o.toString());
         }
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/vector/complex/fn/VectorOutput.java b/exec/java-exec/src/main/java/org/apache/drill/exec/vector/complex/fn/VectorOutput.java
index 769f341d5..bf1448e27 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/vector/complex/fn/VectorOutput.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/vector/complex/fn/VectorOutput.java
@@ -19,6 +19,7 @@ package org.apache.drill.exec.vector.complex.fn;
 
 import java.io.IOException;
 
+import org.apache.drill.common.exceptions.UserException;
 import org.apache.drill.exec.expr.fn.impl.DateUtility;
 import org.apache.drill.exec.expr.fn.impl.StringFunctionHelpers;
 import org.apache.drill.exec.expr.holders.BigIntHolder;
@@ -42,6 +43,8 @@ import org.joda.time.Period;
 import org.joda.time.format.DateTimeFormatter;
 import org.joda.time.format.ISODateTimeFormat;
 import org.joda.time.format.ISOPeriodFormat;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import com.fasterxml.jackson.core.JsonParseException;
 import com.fasterxml.jackson.core.JsonParser;
@@ -49,6 +52,7 @@ import com.fasterxml.jackson.core.JsonToken;
 
 abstract class VectorOutput {
 
+  private static final Logger LOG = LoggerFactory.getLogger(VectorOutput.class);
   final VarBinaryHolder binary = new VarBinaryHolder();
   final TimeHolder time = new TimeHolder();
   final DateHolder date = new DateHolder();
@@ -83,6 +87,15 @@ abstract class VectorOutput {
         writeBinary(checkNextToken(JsonToken.VALUE_STRING));
         checkCurrentToken(JsonToken.END_OBJECT);
         return true;
+      case ExtendedTypeName.TYPE:
+        if(checkNextToken(JsonToken.VALUE_NUMBER_INT) || !hasBinary()) {
+          throw UserException.parseError()
+          .message("Either $type is not an integer or has no $binary")
+          .build(LOG);
+        }
+        writeBinary(checkNextToken(JsonToken.VALUE_STRING));
+        checkCurrentToken(JsonToken.END_OBJECT);
+        return true;
       case ExtendedTypeName.DATE:
         writeDate(checkNextToken(JsonToken.VALUE_STRING));
         checkNextToken(JsonToken.END_OBJECT);
@@ -133,6 +146,11 @@ abstract class VectorOutput {
     return token == JsonToken.FIELD_NAME && parser.getText().equals(ExtendedTypeName.TYPE);
   }
 
+  boolean hasBinary() throws JsonParseException, IOException {
+    JsonToken token = parser.nextToken();
+    return token == JsonToken.FIELD_NAME && parser.getText().equals(ExtendedTypeName.BINARY);
+  }
+
   long getType() throws JsonParseException, IOException {
     if (!checkNextToken(JsonToken.VALUE_NUMBER_INT, JsonToken.VALUE_STRING)) {
       long type = parser.getValueAsLong();
@@ -184,7 +202,12 @@ abstract class VectorOutput {
         byte[] binaryData = parser.getBinaryValue();
         if (hasType()) {
           //Ignoring type info as of now.
-          getType();
+          long type = getType();
+          if (type < 0 || type > 255) {
+            throw UserException.validationError()
+            .message("$type should be between 0 to 255")
+            .build(LOG);
+          }
         }
         work.prepareBinary(binaryData, binary);
         bin.write(binary);
@@ -223,7 +246,9 @@ abstract class VectorOutput {
           ts.writeTimeStamp(DateTime.parse(parser.getValueAsString(), f).withZoneRetainFields(org.joda.time.DateTimeZone.UTC).getMillis());
           break;
         default:
-          break;
+          throw UserException.unsupportedError()
+              .message(parser.getCurrentToken().toString())
+              .build(LOG);
         }
       }
     }
@@ -277,7 +302,12 @@ abstract class VectorOutput {
         byte[] binaryData = parser.getBinaryValue();
         if (hasType()) {
           //Ignoring type info as of now.
-          getType();
+          long type = getType();
+          if (type < 0 || type > 255) {
+            throw UserException.validationError()
+            .message("$type should be between 0 to 255")
+            .build(LOG);
+          }
         }
         work.prepareBinary(binaryData, binary);
         bin.write(binary);
@@ -317,7 +347,9 @@ abstract class VectorOutput {
           ts.writeTimeStamp(DateTime.parse(parser.getValueAsString(), f).withZoneRetainFields(org.joda.time.DateTimeZone.UTC).getMillis());
           break;
         default:
-          break;
+          throw UserException.unsupportedError()
+          .message(parser.getCurrentToken().toString())
+          .build(LOG);
         }
       }
     }
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/vector/complex/writer/TestExtendedTypes.java b/exec/java-exec/src/test/java/org/apache/drill/exec/vector/complex/writer/TestExtendedTypes.java
index 51ecec521..6d6f35a6f 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/vector/complex/writer/TestExtendedTypes.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/vector/complex/writer/TestExtendedTypes.java
@@ -21,12 +21,15 @@ import static org.junit.Assert.assertEquals;
 
 import java.nio.file.Files;
 import java.nio.file.Paths;
+import java.util.List;
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
 
 import org.apache.drill.BaseTestQuery;
 import org.apache.drill.common.util.TestTools;
 import org.apache.drill.exec.ExecConstants;
+import org.apache.drill.exec.rpc.user.QueryDataBatch;
+import org.junit.Assert;
 import org.junit.Test;
 
 public class TestExtendedTypes extends BaseTestQuery {
@@ -81,6 +84,10 @@ public class TestExtendedTypes extends BaseTestQuery {
           String.format(
               "Received unexpected number of rows in output: expected=%d, received=%s",
               1, actualRecordCount), 1, actualRecordCount);
+      List<QueryDataBatch> resultList = testSqlWithResults(String.format("select * from dfs.`%s`", originalFile));
+      String actual = getResultString(resultList, ",");
+      String expected = "drill_timestamp_millies,bin,bin1\n2015-07-07T03:59:43.488,drill,drill\n";
+      Assert.assertEquals(expected, actual);
     } finally {
       testNoResult(String.format("ALTER SESSION SET `%s` = '%s'",
           ExecConstants.OUTPUT_FORMAT_VALIDATOR.getOptionName(),
diff --git a/exec/java-exec/src/test/resources/vector/complex/mongo_extended.json b/exec/java-exec/src/test/resources/vector/complex/mongo_extended.json
index a38a83f48..576be0d92 100644
--- a/exec/java-exec/src/test/resources/vector/complex/mongo_extended.json
+++ b/exec/java-exec/src/test/resources/vector/complex/mongo_extended.json
@@ -4,5 +4,8 @@
   },
   "bin" : {
       "$binary" : "ZHJpbGw=", "$type" : 1
+  },
+  "bin1" : {
+      "$type" : 1, "$binary" : "ZHJpbGw="
   }
 }