drill-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From a..@apache.org
Subject [24/50] [abbrv] drill git commit: Enhance exception handling.
Date Tue, 13 Sep 2016 01:32:11 GMT
Enhance exception handling.


Project: http://git-wip-us.apache.org/repos/asf/drill/repo
Commit: http://git-wip-us.apache.org/repos/asf/drill/commit/16d4ab89
Tree: http://git-wip-us.apache.org/repos/asf/drill/tree/16d4ab89
Diff: http://git-wip-us.apache.org/repos/asf/drill/diff/16d4ab89

Branch: refs/heads/master
Commit: 16d4ab896df5cb368988220bc90f2834d9651d02
Parents: 48d195c
Author: Aditya <aditya@mapr.com>
Authored: Wed Feb 3 17:46:44 2016 -0800
Committer: Aditya Kishore <adi@apache.org>
Committed: Fri Sep 9 10:08:34 2016 -0700

----------------------------------------------------------------------
 .../maprdb/json/MaprDBJsonRecordReader.java     | 163 +++++++++++--------
 1 file changed, 91 insertions(+), 72 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/drill/blob/16d4ab89/contrib/format-maprdb/src/main/java/org/apache/drill/exec/store/maprdb/json/MaprDBJsonRecordReader.java
----------------------------------------------------------------------
diff --git a/contrib/format-maprdb/src/main/java/org/apache/drill/exec/store/maprdb/json/MaprDBJsonRecordReader.java
b/contrib/format-maprdb/src/main/java/org/apache/drill/exec/store/maprdb/json/MaprDBJsonRecordReader.java
index f033de4..2e0e0c1 100644
--- a/contrib/format-maprdb/src/main/java/org/apache/drill/exec/store/maprdb/json/MaprDBJsonRecordReader.java
+++ b/contrib/format-maprdb/src/main/java/org/apache/drill/exec/store/maprdb/json/MaprDBJsonRecordReader.java
@@ -28,8 +28,8 @@ import java.util.List;
 import java.util.Set;
 import java.util.concurrent.TimeUnit;
 
-import org.apache.drill.common.exceptions.DrillRuntimeException;
 import org.apache.drill.common.exceptions.ExecutionSetupException;
+import org.apache.drill.common.exceptions.UserException;
 import org.apache.drill.common.expression.SchemaPath;
 import org.apache.drill.exec.ops.FragmentContext;
 import org.apache.drill.exec.ops.OperatorContext;
@@ -85,8 +85,6 @@ public class MaprDBJsonRecordReader extends AbstractRecordReader {
 
   private boolean includeId;
 
-  private String currentFieldName;
-
   public MaprDBJsonRecordReader(MapRDBSubScanSpec subScanSpec,
       List<SchemaPath> projectedColumns, FragmentContext context) {
     buffer = context.getManagedBuffer();
@@ -144,20 +142,24 @@ public class MaprDBJsonRecordReader extends AbstractRecordReader {
     writer.reset();
 
     int recordCount = 0;
+    DBDocumentReaderBase reader = null;
 
     while(recordCount < BaseValueVector.INITIAL_VALUE_ALLOCATION) {
-      DBDocumentReaderBase reader = nextDocumentReader();
-      if (reader == null) break;
-      writer.setPosition(recordCount);
-      if (reader.next() != EventType.START_MAP) {
-        throw new IllegalStateException("The document did not start with START_MAP!");
-      }
       try {
+        reader = nextDocumentReader();
+        if (reader == null) break;
+        writer.setPosition(recordCount);
+        if (reader.next() != EventType.START_MAP) {
+          throw dataReadError("The document did not start with START_MAP!");
+        }
         writeToMap(reader, writer.rootAsMap());
         recordCount++;
-      } catch (IllegalArgumentException e) {
-        logger.warn(String.format("Possible schema change at _id: '%s', field: '%s'",
-            IdCodec.asString(reader.getId()), currentFieldName), e);
+      } catch (UserException e) {
+        throw UserException.unsupportedError(e)
+            .addContext(String.format("Table: %s, document id: '%s'",
+                table.getPath(),
+                reader == null ? null : IdCodec.asString(reader.getId())))
+            .build(logger);
       }
     }
 
@@ -172,60 +174,65 @@ public class MaprDBJsonRecordReader extends AbstractRecordReader {
       EventType event = reader.next();
       if (event == null || event == EventType.END_MAP) break outside;
 
-      currentFieldName = reader.getFieldName();
-      switch (event) {
-      case NULL:
-        map.varChar(currentFieldName).write(null); // treat as VARCHAR for now
-      case BINARY:
-        writeBinary(map.varBinary(currentFieldName), reader.getBinary());
-        break;
-      case BOOLEAN:
-        map.bit(currentFieldName).writeBit(reader.getBoolean() ? 1 : 0);
-        break;
-      case STRING:
-        writeString(map.varChar(currentFieldName), reader.getString());
-        break;
-      case BYTE:
-        map.tinyInt(currentFieldName).writeTinyInt(reader.getByte());
-        break;
-      case SHORT:
-        map.smallInt(currentFieldName).writeSmallInt(reader.getShort());
-        break;
-      case INT:
-        map.integer(currentFieldName).writeInt(reader.getInt());
-        break;
-      case LONG:
-        map.bigInt(currentFieldName).writeBigInt(reader.getLong());
-        break;
-      case FLOAT:
-        map.float4(currentFieldName).writeFloat4(reader.getFloat());
-        break;
-      case DOUBLE:
-        map.float8(currentFieldName).writeFloat8(reader.getDouble());
-        break;
-      case DECIMAL:
-        throw new UnsupportedOperationException("Decimals are currently not supported.");
-      case DATE:
-        map.date(currentFieldName).writeDate(reader.getDate().toDate().getTime());
-        break;
-      case TIME:
-        map.time(currentFieldName).writeTime(reader.getTimeInt());
-        break;
-      case TIMESTAMP:
-        map.timeStamp(currentFieldName).writeTimeStamp(reader.getTimestampLong());
-        break;
-      case INTERVAL:
-        throw new UnsupportedOperationException("Interval is currently not supported.");
-      case START_MAP:
-        writeToMap(reader, map.map(currentFieldName));
-        break;
-      case START_ARRAY:
-        writeToList(reader, map.list(currentFieldName));
-        break;
-      case END_ARRAY:
-        throw new IllegalStateException("Shouldn't get a END_ARRAY inside a map");
-      default:
-        throw new UnsupportedOperationException("Unsupported type: " + event);
+      String fieldName = reader.getFieldName();
+      try {
+        switch (event) {
+        case NULL:
+          break; // not setting the field will leave it as null
+        case BINARY:
+          writeBinary(map.varBinary(fieldName), reader.getBinary());
+          break;
+        case BOOLEAN:
+          map.bit(fieldName).writeBit(reader.getBoolean() ? 1 : 0);
+          break;
+        case STRING:
+          writeString(map.varChar(fieldName), reader.getString());
+          break;
+        case BYTE:
+          map.tinyInt(fieldName).writeTinyInt(reader.getByte());
+          break;
+        case SHORT:
+          map.smallInt(fieldName).writeSmallInt(reader.getShort());
+          break;
+        case INT:
+          map.integer(fieldName).writeInt(reader.getInt());
+          break;
+        case LONG:
+          map.bigInt(fieldName).writeBigInt(reader.getLong());
+          break;
+        case FLOAT:
+          map.float4(fieldName).writeFloat4(reader.getFloat());
+          break;
+        case DOUBLE:
+          map.float8(fieldName).writeFloat8(reader.getDouble());
+          break;
+        case DECIMAL:
+          throw unsupportedError("Decimal type is currently not supported.");
+        case DATE:
+          map.date(fieldName).writeDate(reader.getDate().toDate().getTime());
+          break;
+        case TIME:
+          map.time(fieldName).writeTime(reader.getTimeInt());
+          break;
+        case TIMESTAMP:
+          map.timeStamp(fieldName).writeTimeStamp(reader.getTimestampLong());
+          break;
+        case INTERVAL:
+          throw unsupportedError("Interval type is currently not supported.");
+        case START_MAP:
+          writeToMap(reader, map.map(fieldName));
+          break;
+        case START_ARRAY:
+          writeToList(reader, map.list(fieldName));
+          break;
+        case END_ARRAY:
+          throw dataReadError("Encountered an END_ARRAY event inside a map.");
+        default:
+          throw unsupportedError("Unsupported type: %s encountered during the query.", event);
+        }
+      } catch (IllegalArgumentException e) {
+        logger.warn(String.format("Possible schema change at _id: '%s', field: '%s'",
+            IdCodec.asString(reader.getId()), fieldName), e);
       }
     }
     map.end();
@@ -239,7 +246,7 @@ public class MaprDBJsonRecordReader extends AbstractRecordReader {
 
       switch (event) {
       case NULL:
-        list.varChar().write(null); // treat as VARCHAR for now
+        throw unsupportedError("Null values are not supported in lists.");
       case BINARY:
         writeBinary(list.varBinary(), reader.getBinary());
         break;
@@ -268,7 +275,7 @@ public class MaprDBJsonRecordReader extends AbstractRecordReader {
         list.float8().writeFloat8(reader.getDouble());
         break;
       case DECIMAL:
-        throw new UnsupportedOperationException("Decimals are currently not supported.");
+        throw unsupportedError("Decimals are currently not supported.");
       case DATE:
         list.date().writeDate(reader.getDate().toDate().getTime());
         break;
@@ -279,17 +286,17 @@ public class MaprDBJsonRecordReader extends AbstractRecordReader {
         list.timeStamp().writeTimeStamp(reader.getTimestampLong());
         break;
       case INTERVAL:
-        throw new UnsupportedOperationException("Interval is currently not supported.");
+        throw unsupportedError("Interval is currently not supported.");
       case START_MAP:
         writeToMap(reader, list.map());
         break;
       case END_MAP:
-        throw new IllegalStateException("Shouldn't get a END_MAP inside a list");
+        throw dataReadError("Encountered an END_MAP event inside a list.");
       case START_ARRAY:
         writeToList(reader, list.list());
         break;
       default:
-        throw new UnsupportedOperationException("Unsupported type: " + event);
+        throw unsupportedError("Unsupported type: %s encountered during the query.%s", event);
       }
     }
     list.endList();
@@ -308,6 +315,18 @@ public class MaprDBJsonRecordReader extends AbstractRecordReader {
     varCharWriter.writeVarChar(0, strBytes.length, buffer);
   }
 
+  private UserException unsupportedError(String format, Object... args) {
+    return UserException.unsupportedError()
+        .message(String.format(format, args))
+        .build(logger);
+  }
+
+  private UserException dataReadError(String format, Object... args) {
+    return UserException.dataReadError()
+        .message(String.format(format, args))
+        .build(logger);
+  }
+
   private DBDocumentReaderBase nextDocumentReader() {
     final OperatorStats operatorStats = operatorContext == null ? null : operatorContext.getStats();
     try {
@@ -326,7 +345,7 @@ public class MaprDBJsonRecordReader extends AbstractRecordReader {
         }
       }
     } catch (DBException e) {
-      throw new DrillRuntimeException(e);
+      throw UserException.dataReadError(e).build(logger);
     }
   }
 


Mime
View raw message