DRILL-6952: Host compliant text reader on the row set framework

The result set loader allows controlling batch sizes. The new scan framework
built on top of that framework handles projection, implicit columns, null
columns and more. This commit converts the "new" ("compliant") text reader
to use the new framework. Options select the use of the V2 ("new") or V3
(row-set based) versions. Unit tests demonstrate V3 functionality.

closes #1683

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

@@ -713,6 +713,14 @@ private ExecConstants() {
   public static final OptionValidator ENABLE_NEW_TEXT_READER = new BooleanValidator(ENABLE_NEW_TEXT_READER_KEY,
       new OptionDescription("Enables the text reader that complies with the RFC 4180 standard for text/csv files."));
 
+  public static final String ENABLE_V3_TEXT_READER_KEY = "exec.storage.enable_v3_text_reader";
+  public static final OptionValidator ENABLE_V3_TEXT_READER = new BooleanValidator(ENABLE_V3_TEXT_READER_KEY,
+      new OptionDescription("Enables the row set based version of the text/csv reader."));
+
+  public static final String MIN_READER_WIDTH_KEY = "exec.storage.min_width";
+  public static final OptionValidator MIN_READER_WIDTH = new LongValidator(MIN_READER_WIDTH_KEY,
+      new OptionDescription("Min width for text readers, mostly for testing."));
+
   public static final String BOOTSTRAP_STORAGE_PLUGINS_FILE = "bootstrap-storage-plugins.json";
 
   public static final String DRILL_SYS_FILE_SUFFIX = ".sys.drill";

exec/java-exec/src/main/java/org/apache/drill/exec/physical/base/AbstractExchange.java

@@ -31,7 +31,6 @@
 import org.apache.drill.shaded.guava.com.google.common.collect.Maps;
 
 public abstract class AbstractExchange extends AbstractSingle implements Exchange {
-  static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(AbstractExchange.class);
 
   // Ephemeral info for generating execution fragments.
   protected int senderMajorFragmentId;
@@ -77,7 +76,7 @@ public ParallelizationInfo getReceiverParallelizationInfo(List<DrillbitEndpoint>
   }
 
   /**
-   * Get a default endpoint affinity map where affinity of a Drillbit is proportional to the number of its occurrances
+   * Get a default endpoint affinity map where affinity of a Drillbit is proportional to the number of its occurrences
    * in given endpoint list.
    *
    * @param fragmentEndpoints Drillbit endpoint assignments of fragments.
@@ -111,7 +110,6 @@ public final void setupSenders(int majorFragmentId, List<DrillbitEndpoint> sende
     setupSenders(senderLocations);
   }
 
-
   @Override
   public final void setupReceivers(int majorFragmentId, List<DrillbitEndpoint> receiverLocations) throws PhysicalOperatorSetupException {
     this.receiverMajorFragmentId = majorFragmentId;

exec/java-exec/src/main/java/org/apache/drill/exec/physical/base/AbstractFileGroupScan.java

@@ -22,7 +22,6 @@
 import org.apache.drill.exec.store.dfs.FileSelection;
 
 public abstract class AbstractFileGroupScan extends AbstractGroupScan implements FileGroupScan {
-  static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(AbstractFileGroupScan.class);
 
   public AbstractFileGroupScan(String userName) {
     super(userName);
@@ -46,5 +45,4 @@ public FileGroupScan clone(FileSelection selection) throws IOException {
   public boolean supportsPartitionFilterPushdown() {
     return true;
   }
-
 }

exec/java-exec/src/main/java/org/apache/drill/exec/physical/base/AbstractSubScan.java

@@ -25,8 +25,7 @@
 import org.apache.drill.common.graph.GraphVisitor;
 import org.apache.drill.exec.record.BatchSchema.SelectionVectorMode;
 
-public abstract class AbstractSubScan extends AbstractBase implements SubScan{
-  static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(AbstractSubScan.class);
+public abstract class AbstractSubScan extends AbstractBase implements SubScan {
 
   public AbstractSubScan(String userName) {
     super(userName);

exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/ScanBatch.java

@@ -499,7 +499,6 @@ public void clear() {
       schemaChanged = false;
     }
 
-    @SuppressWarnings("resource")
     private <T extends ValueVector> T addField(MaterializedField field,
         Class<T> clazz, boolean isImplicitField) throws SchemaChangeException {
       Map<String, ValueVector> fieldVectorMap;

exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/project/ProjectRecordBatch.java

@@ -42,7 +42,6 @@
 import org.apache.drill.exec.exception.OutOfMemoryException;
 import org.apache.drill.exec.exception.SchemaChangeException;
 import org.apache.drill.exec.expr.ClassGenerator;
-import org.apache.drill.exec.expr.ClassGenerator.HoldingContainer;
 import org.apache.drill.exec.expr.CodeGenerator;
 import org.apache.drill.exec.expr.DrillFuncHolderExpr;
 import org.apache.drill.exec.expr.ExpressionTreeMaterializer;
@@ -77,21 +76,21 @@
 import static org.apache.drill.exec.record.RecordBatch.IterOutcome.EMIT;
 
 public class ProjectRecordBatch extends AbstractSingleRecordBatch<Project> {
-  static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(ProjectRecordBatch.class);
+  private static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(ProjectRecordBatch.class);
+
+  private static final String EMPTY_STRING = "";
+
   private Projector projector;
   private List<ValueVector> allocationVectors;
   private List<ComplexWriter> complexWriters;
   private List<FieldReference> complexFieldReferencesList;
   private boolean hasRemainder = false;
-  private int remainderIndex = 0;
+  private int remainderIndex;
   private int recordCount;
-
   private ProjectMemoryManager memoryManager;
-
-
-  private static final String EMPTY_STRING = "";
   private boolean first = true;
   private boolean wasNone = false; // whether a NONE iter outcome was already seen
+  private ColumnExplorer columnExplorer;
 
   private class ClassifierResult {
     public boolean isStar = false;
@@ -114,21 +113,20 @@ private void clear() {
 
   public ProjectRecordBatch(final Project pop, final RecordBatch incoming, final FragmentContext context) throws OutOfMemoryException {
     super(pop, context, incoming);
+    columnExplorer = new ColumnExplorer(context.getOptions());
   }
 
   @Override
   public int getRecordCount() {
     return recordCount;
   }
 
-
   @Override
   protected void killIncoming(final boolean sendUpstream) {
     super.killIncoming(sendUpstream);
     hasRemainder = false;
   }
 
-
   @Override
   public IterOutcome innerNext() {
     if (wasNone) {
@@ -145,7 +143,7 @@ public IterOutcome innerNext() {
 
   @Override
   public VectorContainer getOutgoingContainer() {
-    return this.container;
+    return container;
   }
 
   @Override
@@ -204,22 +202,21 @@ protected IterOutcome doWork() {
           }
           incomingRecordCount = incoming.getRecordCount();
           memoryManager.update();
-          logger.trace("doWork():[1] memMgr RC {}, incoming rc {},  incoming {}, Project {}",
+          logger.trace("doWork():[1] memMgr RC {}, incoming rc {}, incoming {}, Project {}",
                        memoryManager.getOutputRowCount(), incomingRecordCount, incoming, this);
         }
       }
     }
 
     if (complexWriters != null && getLastKnownOutcome() == EMIT) {
-      throw new UnsupportedOperationException("Currently functions producing complex types as output is not " +
+      throw new UnsupportedOperationException("Currently functions producing complex types as output are not " +
         "supported in project list for subquery between LATERAL and UNNEST. Please re-write the query using this " +
         "function in the projection list of outermost query.");
     }
 
     first = false;
     container.zeroVectors();
 
-
     int maxOuputRecordCount = memoryManager.getOutputRowCount();
     logger.trace("doWork():[2] memMgr RC {}, incoming rc {}, incoming {}, project {}",
                  memoryManager.getOutputRowCount(), incomingRecordCount, incoming, this);
@@ -233,7 +230,6 @@ protected IterOutcome doWork() {
     long projectEndTime = System.currentTimeMillis();
     logger.trace("doWork(): projection: records {}, time {} ms", outputRecords, (projectEndTime - projectStartTime));
 
-
     if (outputRecords < incomingRecordCount) {
       setValueCount(outputRecords);
       hasRemainder = true;
@@ -277,7 +273,7 @@ private void handleRemainder() {
     final int projRecords = projector.projectRecords(this.incoming, remainderIndex, recordsToProcess, 0);
     long projectEndTime = System.currentTimeMillis();
 
-    logger.trace("handleRemainder: projection: " + "records {}, time {} ms", projRecords,(projectEndTime - projectStartTime));
+    logger.trace("handleRemainder: projection: records {}, time {} ms", projRecords,(projectEndTime - projectStartTime));
 
     if (projRecords < remainingRecordCount) {
       setValueCount(projRecords);
@@ -463,7 +459,7 @@ private void setupNewSchemaFromInput(RecordBatch incomingBatch) throws SchemaCha
               final TypedFieldId fid = container.getValueVectorId(SchemaPath.getSimplePath(outputField.getName()));
               final ValueVectorWriteExpression write = new ValueVectorWriteExpression(fid, expr, true);
               memoryManager.addNewField(vv, write);
-              final HoldingContainer hc = cg.addExpr(write, ClassGenerator.BlkCreateMode.TRUE_IF_BOUND);
+              cg.addExpr(write, ClassGenerator.BlkCreateMode.TRUE_IF_BOUND);
             }
           }
           continue;
@@ -546,7 +542,7 @@ private void setupNewSchemaFromInput(RecordBatch incomingBatch) throws SchemaCha
         final TypedFieldId fid = container.getValueVectorId(SchemaPath.getSimplePath(outputField.getName()));
         final boolean useSetSafe = !(ouputVector instanceof FixedWidthVector);
         final ValueVectorWriteExpression write = new ValueVectorWriteExpression(fid, expr, useSetSafe);
-        final HoldingContainer hc = cg.addExpr(write, ClassGenerator.BlkCreateMode.TRUE_IF_BOUND);
+        cg.addExpr(write, ClassGenerator.BlkCreateMode.TRUE_IF_BOUND);
         memoryManager.addNewField(ouputVector, write);
 
         // We cannot do multiple transfers from the same vector. However we still need to instantiate the output vector.
@@ -590,7 +586,7 @@ protected boolean setupNewSchema() throws SchemaChangeException {
   }
 
   private boolean isImplicitFileColumn(ValueVector vvIn) {
-    return ColumnExplorer.initImplicitFileColumns(context.getOptions()).get(vvIn.getField().getName()) != null;
+    return columnExplorer.isImplicitFileColumn(vvIn.getField().getName());
   }
 
   private List<NamedExpression> getExpressionList() {

exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/scan/columns/ColumnsArrayParser.java

@@ -24,7 +24,7 @@
 import org.apache.drill.exec.physical.impl.scan.project.UnresolvedColumn;
 import org.apache.drill.exec.physical.rowSet.project.RequestedColumnImpl;
 import org.apache.drill.exec.physical.rowSet.project.RequestedTuple.RequestedColumn;
-import org.apache.drill.exec.store.easy.text.compliant.RepeatedVarCharOutput;
+import org.apache.drill.exec.store.easy.text.compliant.v3.TextReader;
 
 /**
  * Parses the `columns` array. Doing so is surprisingly complex.
@@ -113,14 +113,14 @@ public boolean parse(RequestedColumn inCol) {
 
     if (inCol.isArray()) {
       int maxIndex = inCol.maxIndex();
-      if (maxIndex > RepeatedVarCharOutput.MAXIMUM_NUMBER_COLUMNS) {
+      if (maxIndex > TextReader.MAXIMUM_NUMBER_COLUMNS) {
         throw UserException
           .validationError()
           .message(String.format(
               "`columns`[%d] index out of bounds, max supported size is %d",
-              maxIndex, RepeatedVarCharOutput.MAXIMUM_NUMBER_COLUMNS))
+              maxIndex, TextReader.MAXIMUM_NUMBER_COLUMNS))
           .addContext("Column", inCol.name())
-          .addContext("Maximum index", RepeatedVarCharOutput.MAXIMUM_NUMBER_COLUMNS)
+          .addContext("Maximum index", TextReader.MAXIMUM_NUMBER_COLUMNS)
           .build(logger);
       }
     }

exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/scan/file/BaseFileScanFramework.java

@@ -69,6 +69,7 @@ public interface FileSchemaNegotiator extends SchemaNegotiator {
   private List<FileSplit> spilts = new ArrayList<>();
   private Iterator<FileSplit> splitIter;
   private Path scanRootDir;
+  private int partitionDepth;
   protected DrillFileSystem dfs;
   private FileMetadataManager metadataManager;
 
@@ -82,12 +83,18 @@ public BaseFileScanFramework(List<SchemaPath> projection,
 
   /**
    * Specify the selection root for a directory scan, if any.
-   * Used to populate partition columns.
+   * Used to populate partition columns. Also, specify the maximum
+   * partition depth.
+   *
    * @param rootPath Hadoop file path for the directory
+   * @param partitionDepth maximum partition depth across all files
+   * within this logical scan operator (files in this scan may be
+   * shallower)
    */
 
-  public void setSelectionRoot(Path rootPath) {
+  public void setSelectionRoot(Path rootPath, int partitionDepth) {
     this.scanRootDir = rootPath;
+    this.partitionDepth = partitionDepth;
   }
 
   @Override
@@ -122,7 +129,10 @@ protected void configure() {
 
     metadataManager = new FileMetadataManager(
         context.getFragmentContext().getOptions(),
+        true, // Expand partition columns with wildcard
+        false, // Put partition columns after table columns
         scanRootDir,
+        partitionDepth,
         paths);
     scanOrchestrator.withMetadata(metadataManager);
   }

exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/scan/file/FileMetadataColumnsParser.java

@@ -46,6 +46,8 @@ public class FileMetadataColumnsParser implements ScanProjectionParser {
 
   private boolean hasImplicitCols;
 
+  private boolean expandPartitionsAtEnd;
+
   public FileMetadataColumnsParser(FileMetadataManager metadataManager) {
     this.metadataManager = metadataManager;
     partitionPattern = Pattern.compile(metadataManager.partitionDesignator + "(\\d+)", Pattern.CASE_INSENSITIVE);
@@ -123,8 +125,10 @@ private boolean buildMetadataColumn(FileMetadataColumnDefn defn,
   }
 
   private void buildWildcard() {
-    if (metadataManager.useLegacyWildcardExpansion &&
-        metadataManager.useLegacyExpansionLocation) {
+    if (!metadataManager.useLegacyWildcardExpansion) {
+      return;
+    }
+    if (metadataManager.useLegacyExpansionLocation) {
 
       // Star column: this is a SELECT * query.
 
@@ -134,6 +138,8 @@ private void buildWildcard() {
       // set is constant across all files.
 
       expandPartitions();
+    } else {
+      expandPartitionsAtEnd = true;
     }
   }
 
@@ -144,8 +150,7 @@ public void validate() {
     // feature to expand partitions for wildcards, and we want the
     // partitions after data columns.
 
-    if (builder.hasWildcard() && metadataManager.useLegacyWildcardExpansion &&
-        ! metadataManager.useLegacyExpansionLocation) {
+    if (expandPartitionsAtEnd) {
       expandPartitions();
     }
   }