drill-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From a..@apache.org
Subject [45/50] [abbrv] drill git commit: MD-813: Improve count(*) queries against MapR-DB Json tables.
Date Tue, 13 Sep 2016 01:32:32 GMT
MD-813: Improve count(*) queries against MapR-DB Json tables.

+ Fail query on schema change.
+ Added a configuration option 'ignoreSchemaChange', which when enabled, drops the rows from
the result


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

Branch: refs/heads/master
Commit: 12cbd27ef8cf23db15645453d815a631926bd8cd
Parents: 391027a
Author: Aditya <aditya@mapr.com>
Authored: Wed Mar 16 01:58:56 2016 -0700
Committer: Aditya Kishore <adi@apache.org>
Committed: Fri Sep 9 10:08:40 2016 -0700

----------------------------------------------------------------------
 .../store/mapr/db/MapRDBFormatPluginConfig.java | 14 ++-
 .../mapr/db/json/MaprDBJsonRecordReader.java    | 90 +++++++++++++-------
 .../store/mapr/streams/StreamsFormatPlugin.java |  3 +-
 3 files changed, 73 insertions(+), 34 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/drill/blob/12cbd27e/contrib/format-maprdb/src/main/java/org/apache/drill/exec/store/mapr/db/MapRDBFormatPluginConfig.java
----------------------------------------------------------------------
diff --git a/contrib/format-maprdb/src/main/java/org/apache/drill/exec/store/mapr/db/MapRDBFormatPluginConfig.java
b/contrib/format-maprdb/src/main/java/org/apache/drill/exec/store/mapr/db/MapRDBFormatPluginConfig.java
index 7295265..1bb07ed 100644
--- a/contrib/format-maprdb/src/main/java/org/apache/drill/exec/store/mapr/db/MapRDBFormatPluginConfig.java
+++ b/contrib/format-maprdb/src/main/java/org/apache/drill/exec/store/mapr/db/MapRDBFormatPluginConfig.java
@@ -28,8 +28,9 @@ import com.fasterxml.jackson.annotation.JsonTypeName;
 public class MapRDBFormatPluginConfig extends TableFormatPluginConfig {
 
   public boolean allTextMode = false;
-  public boolean readAllNumbersAsDouble = false;
   public boolean enablePushdown = true;
+  public boolean ignoreSchemaChange = false;
+  public boolean readAllNumbersAsDouble = false;
 
   @Override
   public int hashCode() {
@@ -43,10 +44,11 @@ public class MapRDBFormatPluginConfig extends TableFormatPluginConfig
{
       return false;
     } else if (allTextMode != other.allTextMode) {
       return false;
+    } else if (isIgnoreSchemaChange() != other.isIgnoreSchemaChange()) {
+      return false;
     } else if (enablePushdown != other.enablePushdown) {
       return false;
     }
-
     return true;
   }
 
@@ -77,4 +79,12 @@ public class MapRDBFormatPluginConfig extends TableFormatPluginConfig {
     this.enablePushdown = enablePushdown;
   }
 
+  public boolean isIgnoreSchemaChange() {
+    return ignoreSchemaChange;
+  }
+
+  public void setIgnoreSchemaChange(boolean ignoreSchemaChange) {
+    this.ignoreSchemaChange = ignoreSchemaChange;
+  }
+
 }

http://git-wip-us.apache.org/repos/asf/drill/blob/12cbd27e/contrib/format-maprdb/src/main/java/org/apache/drill/exec/store/mapr/db/json/MaprDBJsonRecordReader.java
----------------------------------------------------------------------
diff --git a/contrib/format-maprdb/src/main/java/org/apache/drill/exec/store/mapr/db/json/MaprDBJsonRecordReader.java
b/contrib/format-maprdb/src/main/java/org/apache/drill/exec/store/mapr/db/json/MaprDBJsonRecordReader.java
index cb86e32..9bb7daa 100644
--- a/contrib/format-maprdb/src/main/java/org/apache/drill/exec/store/mapr/db/json/MaprDBJsonRecordReader.java
+++ b/contrib/format-maprdb/src/main/java/org/apache/drill/exec/store/mapr/db/json/MaprDBJsonRecordReader.java
@@ -32,10 +32,12 @@ import org.apache.drill.common.exceptions.UserException;
 import org.apache.drill.common.expression.PathSegment;
 import org.apache.drill.common.expression.SchemaPath;
 import org.apache.drill.exec.ExecConstants;
+import org.apache.drill.exec.exception.SchemaChangeException;
 import org.apache.drill.exec.ops.FragmentContext;
 import org.apache.drill.exec.ops.OperatorContext;
 import org.apache.drill.exec.ops.OperatorStats;
 import org.apache.drill.exec.physical.impl.OutputMutator;
+import org.apache.drill.exec.proto.UserBitShared.DrillPBError.ErrorType;
 import org.apache.drill.exec.store.AbstractRecordReader;
 import org.apache.drill.exec.store.mapr.db.MapRDBFormatPluginConfig;
 import org.apache.drill.exec.store.mapr.db.MapRDBSubScanSpec;
@@ -90,6 +92,7 @@ public class MaprDBJsonRecordReader extends AbstractRecordReader {
   private final boolean readNumbersAsDouble;
   private boolean disablePushdown;
   private final boolean allTextMode;
+  private final boolean ignoreSchemaChange;
 
   public MaprDBJsonRecordReader(MapRDBSubScanSpec subScanSpec,
       MapRDBFormatPluginConfig formatPluginConfig,
@@ -111,6 +114,7 @@ public class MaprDBJsonRecordReader extends AbstractRecordReader {
     unionEnabled = context.getOptions().getOption(ExecConstants.ENABLE_UNION_TYPE);
     readNumbersAsDouble = formatPluginConfig.isReadAllNumbersAsDouble();
     allTextMode = formatPluginConfig.isAllTextMode();
+    ignoreSchemaChange = formatPluginConfig.isIgnoreSchemaChange();
     disablePushdown = !formatPluginConfig.isEnablePushdown();
   }
 
@@ -177,35 +181,23 @@ public class MaprDBJsonRecordReader extends AbstractRecordReader {
     DBDocumentReaderBase reader = null;
 
     while(recordCount < BaseValueVector.INITIAL_VALUE_ALLOCATION) {
+      vectorWriter.setPosition(recordCount);
       try {
         reader = nextDocumentReader();
-        if (reader == null) break;
-
-        vectorWriter.setPosition(recordCount);
-        MapOrListWriterImpl writer = new MapOrListWriterImpl(vectorWriter.rootAsMap());
-        if (idOnly) {
-          Value id = reader.getId();
-          try {
-            switch(id.getType()) {
-            case STRING:
-              writeString(writer, ID_KEY, id.getString());
-              break;
-            case BINARY:
-              writeBinary(writer, ID_KEY, id.getBinary());
-              break;
-            default:
-              throw new UnsupportedOperationException(id.getType() +
-                  " is not a supported type for _id field.");
-            }
-          } catch (IllegalStateException | IllegalArgumentException e) {
-            logger.warn(String.format("Possible schema change at _id: '%s'",
-                IdCodec.asString(id)), e);
-          }
+        if (reader == null) {
+          break; // no more documents for this scanner
+        } else if (isSkipQuery()) {
+          vectorWriter.rootAsMap().bit("count").writeBit(1);
         } else {
-          if (reader.next() != EventType.START_MAP) {
-            throw dataReadError("The document did not start with START_MAP!");
+          MapOrListWriterImpl writer = new MapOrListWriterImpl(vectorWriter.rootAsMap());
+          if (idOnly) {
+            writeId(writer, reader.getId());
+          } else {
+            if (reader.next() != EventType.START_MAP) {
+              throw dataReadError("The document did not start with START_MAP!");
+            }
+            writeToListOrMap(writer, reader);
           }
-          writeToListOrMap(writer, reader);
         }
         recordCount++;
       } catch (UserException e) {
@@ -214,6 +206,13 @@ public class MaprDBJsonRecordReader extends AbstractRecordReader {
                 table.getPath(),
                 reader == null ? null : IdCodec.asString(reader.getId())))
             .build(logger);
+      } catch (SchemaChangeException e) {
+        if (ignoreSchemaChange) {
+          logger.warn("{}. Dropping the row from result.", e.getMessage());
+          logger.debug("Stack trace:", e);
+        } else {
+          throw dataReadError(e);
+        }
       }
     }
 
@@ -222,7 +221,25 @@ public class MaprDBJsonRecordReader extends AbstractRecordReader {
     return recordCount;
   }
 
-  private void writeToListOrMap(MapOrListWriterImpl writer, DBDocumentReaderBase reader)
{
+  private void writeId(MapOrListWriterImpl writer, Value id) throws SchemaChangeException
{
+    try {
+      switch(id.getType()) {
+      case STRING:
+        writeString(writer, ID_KEY, id.getString());
+        break;
+      case BINARY:
+        writeBinary(writer, ID_KEY, id.getBinary());
+        break;
+      default:
+        throw new UnsupportedOperationException(id.getType() +
+            " is not a supported type for _id field.");
+      }
+    } catch (IllegalStateException | IllegalArgumentException e) {
+      throw schemaChangeException(e, "Possible schema change at _id: '%s'", IdCodec.asString(id));
+    }
+  }
+
+  private void writeToListOrMap(MapOrListWriterImpl writer, DBDocumentReaderBase reader)
throws SchemaChangeException {
     String fieldName = null;
     writer.start();
     outside: while (true) {
@@ -289,8 +306,7 @@ public class MaprDBJsonRecordReader extends AbstractRecordReader {
           throw unsupportedError("Unsupported type: %s encountered during the query.", event);
         }
       } catch (IllegalStateException | IllegalArgumentException e) {
-        logger.warn(String.format("Possible schema change at _id: '%s', field: '%s'",
-            IdCodec.asString(reader.getId()), fieldName), e);
+        throw schemaChangeException(e, "Possible schema change at _id: '%s', field: '%s'",
IdCodec.asString(reader.getId()), fieldName);
       }
     }
     writer.end();
@@ -410,12 +426,24 @@ public class MaprDBJsonRecordReader extends AbstractRecordReader {
         .build(logger);
   }
 
+  private UserException dataReadError(Throwable t) {
+    return dataReadError(t, null);
+  }
+
   private UserException dataReadError(String format, Object... args) {
-    return UserException.dataReadError()
-        .message(String.format(format, args))
+    return dataReadError(null, format, args);
+  }
+
+  private UserException dataReadError(Throwable t, String format, Object... args) {
+    return UserException.dataReadError(t)
+        .message(format == null ? null : String.format(format, args))
         .build(logger);
   }
 
+  private SchemaChangeException schemaChangeException(Throwable t, String format, Object...
args) {
+    return new SchemaChangeException(format, t, args);
+  }
+
   private DBDocumentReaderBase nextDocumentReader() {
     final OperatorStats operatorStats = operatorContext == null ? null : operatorContext.getStats();
     try {
@@ -434,7 +462,7 @@ public class MaprDBJsonRecordReader extends AbstractRecordReader {
         }
       }
     } catch (DBException e) {
-      throw UserException.dataReadError(e).build(logger);
+      throw dataReadError(e);
     }
   }
 

http://git-wip-us.apache.org/repos/asf/drill/blob/12cbd27e/contrib/format-maprdb/src/main/java/org/apache/drill/exec/store/mapr/streams/StreamsFormatPlugin.java
----------------------------------------------------------------------
diff --git a/contrib/format-maprdb/src/main/java/org/apache/drill/exec/store/mapr/streams/StreamsFormatPlugin.java
b/contrib/format-maprdb/src/main/java/org/apache/drill/exec/store/mapr/streams/StreamsFormatPlugin.java
index 811e245..f7c76b5 100644
--- a/contrib/format-maprdb/src/main/java/org/apache/drill/exec/store/mapr/streams/StreamsFormatPlugin.java
+++ b/contrib/format-maprdb/src/main/java/org/apache/drill/exec/store/mapr/streams/StreamsFormatPlugin.java
@@ -20,6 +20,7 @@ package org.apache.drill.exec.store.mapr.streams;
 import java.io.IOException;
 import java.util.List;
 
+import org.apache.drill.common.exceptions.UserException;
 import org.apache.drill.common.expression.SchemaPath;
 import org.apache.drill.common.logical.StoragePluginConfig;
 import org.apache.drill.exec.physical.base.AbstractGroupScan;
@@ -73,7 +74,7 @@ public class StreamsFormatPlugin extends TableFormatPlugin {
     List<String> files = selection.getFiles();
     assert (files.size() == 1);
     //TableProperties props = getMaprFS().getTableProperties(new Path(files.get(0)));
-    throw new UnsupportedOperationException("not implemented");
+    throw UserException.unsupportedError().message("MapR streams can not be querried at this
time.").build(logger);
   }
 
 }


Mime
View raw message