drill-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From volody...@apache.org
Subject [drill] 04/04: DRILL-7725: Updates to the EVF2 framework
Date Mon, 25 May 2020 22:34:53 GMT
This is an automated email from the ASF dual-hosted git repository.

volodymyr pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/drill.git

commit d280d0697f89570611aed7ae0359251b6242633d
Author: Paul Rogers <par0328@yahoo.com>
AuthorDate: Thu Apr 30 16:06:55 2020 -0700

    DRILL-7725: Updates to the EVF2 framework
    
    * Supports internal implicit columns
    * Better support for standard conversions
    * Handle several reader corner cases
    * Simplified file reader
    
    closes #2073
---
 .../categories/{EvfTests.java => EvfTest.java}     |   2 +-
 .../drill/exec/store/hdf5/writers/WriterSpec.java  |   6 +-
 .../exec/physical/impl/scan/ScanOperatorExec.java  |   2 +-
 .../impl/scan/columns/ColumnsScanFramework.java    |   3 +-
 .../scan/convert/AbstractConvertFromString.java    |   4 +-
 .../impl/scan/convert/StandardConversions.java     | 149 ++++++--------
 .../exec/physical/impl/scan/v3/FixedReceiver.java  | 208 +++++++++++++++++++
 .../impl/scan/v3/ScanLifecycleBuilder.java         |  36 +++-
 .../physical/impl/scan/v3/SchemaNegotiator.java    |  10 +
 .../physical/impl/scan/v3/file/FileDescrip.java    | 102 +++++++++-
 .../impl/scan/v3/file/FileScanLifecycle.java       |   5 +-
 .../scan/v3/file/FileScanLifecycleBuilder.java     |  37 ++--
 .../impl/scan/v3/file/FileSchemaNegotiator.java    |  22 +--
 .../scan/v3/file/FileSchemaNegotiatorImpl.java     |  47 +++--
 .../impl/scan/v3/file/ImplicitColumnMarker.java    |  99 ++++++++++
 .../{schema => file}/ImplicitColumnResolver.java   | 191 ++++++++++--------
 .../scan/v3/file/ImplicitFileColumnsHandler.java   |  37 ++--
 .../impl/scan/v3/lifecycle/ReaderLifecycle.java    | 191 ++++++++++--------
 .../impl/scan/v3/lifecycle/ScanLifecycle.java      |  20 +-
 .../scan/v3/lifecycle/SchemaNegotiatorImpl.java    |   9 +
 .../impl/scan/v3/lifecycle/StaticBatchBuilder.java |   4 +-
 .../impl/scan/v3/schema/AbstractSchemaTracker.java |  31 ++-
 .../impl/scan/v3/schema/MutableTupleMetadata.java  | 220 ---------------------
 .../impl/scan/v3/schema/MutableTupleSchema.java    |  71 ++++---
 .../scan/v3/schema/ProjectionSchemaTracker.java    |  44 +++--
 .../scan/v3/schema/ScanSchemaConfigBuilder.java    |  28 ++-
 .../impl/scan/v3/schema/ScanSchemaResolver.java    |  66 ++++---
 .../impl/scan/v3/schema/ScanSchemaTracker.java     |   8 +-
 .../impl/scan/v3/schema/SchemaBasedTracker.java    |  27 ++-
 .../exec/physical/resultSet/RowSetLoader.java      |   7 -
 .../apache/drill/exec/store/ColumnExplorer.java    |   9 +-
 .../drill/exec/store/avro/AvroBatchReader.java     |   4 +-
 .../exec/store/avro/ColumnConverterFactory.java    |   7 +-
 .../easy/text/reader/CompliantTextBatchReader.java |  18 +-
 .../drill/exec/store/log/LogBatchReader.java       |   8 +-
 .../exec/physical/impl/scan/ScanTestUtils.java     |  45 +++--
 .../physical/impl/scan/TestFileScanFramework.java  |   2 +-
 .../impl/scan/TestScanOperExecOuputSchema.java     |   3 +-
 .../scan/TestScanOrchestratorImplicitColumns.java  |   2 +-
 .../impl/scan/convert/TestDirectConverter.java     | 187 +++++++++---------
 .../impl/scan/project/TestScanLevelProjection.java |   4 +-
 .../impl/scan/project/TestSchemaSmoothing.java     |  52 +----
 .../physical/impl/scan/v3/TestFixedReceiver.java   | 124 ++++++++++++
 .../exec/physical/impl/scan/v3/TestScanBasics.java |   4 +-
 .../physical/impl/scan/v3/TestScanEarlySchema.java |   4 +-
 .../physical/impl/scan/v3/TestScanLateSchema.java  |   4 +-
 .../physical/impl/scan/v3/TestScanOuputSchema.java |   7 +-
 .../physical/impl/scan/v3/TestScanOverflow.java    |   4 +-
 .../physical/impl/scan/v3/file/MockFileNames.java  |   1 -
 .../impl/scan/v3/file/TestFileDescrip.java         |  84 ++++----
 .../physical/impl/scan/v3/file/TestFileScan.java   |   9 +-
 .../impl/scan/v3/file/TestFileScanLifecycle.java   |   4 +-
 .../scan/v3/file/TestImplicitColumnLoader.java     | 121 ++++++++++--
 .../scan/v3/file/TestImplicitColumnResolver.java   |  84 ++++++--
 .../scan/v3/lifecycle/BaseTestScanLifecycle.java   |   1 -
 .../scan/v3/lifecycle/TestMissingColumnLoader.java |   4 +-
 .../scan/v3/lifecycle/TestOutputBatchBuilder.java  |   4 +-
 .../impl/scan/v3/lifecycle/TestReaderErrors.java   |   4 +-
 .../scan/v3/lifecycle/TestScanLifecycleBasics.java |  15 +-
 .../scan/v3/lifecycle/TestScanLifecycleSchema.java |   4 +-
 .../v3/lifecycle/TestScanLifecycleTwoReaders.java  |   4 +-
 .../scan/v3/schema/TestDynamicSchemaFilter.java    |   4 +-
 .../impl/scan/v3/schema/TestProjectedPath.java     |   4 +-
 .../impl/scan/v3/schema/TestProjectionParser.java  |   4 +-
 .../impl/scan/v3/schema/TestScanSchemaTracker.java |  90 ++++++++-
 .../scan/v3/schema/TestScanSchemaTrackerMaps.java  |   4 +-
 .../scan/v3/schema/TestSchemaTrackerDefined.java   |   4 +-
 .../schema/TestSchemaTrackerEarlyReaderSchema.java |  27 ++-
 .../v3/schema/TestSchemaTrackerInputSchema.java    |  23 +--
 .../v3/schema/TestSchemaTrackerProjection.java     |  28 ++-
 .../impl/TestResultSetLoaderProjection.java        |   4 +-
 .../store/easy/text/compliant/TestCsvHeader.java   |   4 +-
 .../easy/text/compliant/TestCsvIgnoreHeaders.java  |   4 +-
 .../text/compliant/TestCsvTableProperties.java     |   4 +-
 .../easy/text/compliant/TestCsvWithHeaders.java    |   4 +-
 .../easy/text/compliant/TestCsvWithSchema.java     |   4 +-
 .../easy/text/compliant/TestCsvWithoutHeaders.java |   4 +-
 .../easy/text/compliant/TestPartitionRace.java     |   4 +-
 .../java/org/apache/drill/test/QueryBuilder.java   |   3 +-
 .../drill/exec/record/metadata/MetadataUtils.java  |   4 +
 80 files changed, 1763 insertions(+), 977 deletions(-)

diff --git a/common/src/test/java/org/apache/drill/categories/EvfTests.java b/common/src/test/java/org/apache/drill/categories/EvfTest.java
similarity index 97%
rename from common/src/test/java/org/apache/drill/categories/EvfTests.java
rename to common/src/test/java/org/apache/drill/categories/EvfTest.java
index 7bc771e..e2fc1a7 100644
--- a/common/src/test/java/org/apache/drill/categories/EvfTests.java
+++ b/common/src/test/java/org/apache/drill/categories/EvfTest.java
@@ -21,6 +21,6 @@ package org.apache.drill.categories;
  * A category for tests that test the "Extended Vector Framework"
  * (EVF): the mechanism that drives the plugin-based scan operator.
  */
-public interface EvfTests {
+public interface EvfTest {
   // Junit category marker
 }
diff --git a/contrib/format-hdf5/src/main/java/org/apache/drill/exec/store/hdf5/writers/WriterSpec.java b/contrib/format-hdf5/src/main/java/org/apache/drill/exec/store/hdf5/writers/WriterSpec.java
index 6d45e06..6690df8 100644
--- a/contrib/format-hdf5/src/main/java/org/apache/drill/exec/store/hdf5/writers/WriterSpec.java
+++ b/contrib/format-hdf5/src/main/java/org/apache/drill/exec/store/hdf5/writers/WriterSpec.java
@@ -44,7 +44,9 @@ public class WriterSpec {
     this.tupleWriter = tupleWriter;
     this.providedSchema = providedSchema;
     this.errorContext = errorContext;
-    this.conversions = new StandardConversions(providedSchema);
+    this.conversions = StandardConversions.builder()
+        .withSchema(providedSchema)
+        .build();
   }
 
   public ValueWriter makeWriter(String name, MinorType type, DataMode mode) {
@@ -56,7 +58,7 @@ public class WriterSpec {
       return tupleWriter.scalar(index);
     } else {
       int index = tupleWriter.addColumn(providedCol);
-      return conversions.converter(tupleWriter.scalar(index), type);
+      return conversions.converterFor(tupleWriter.scalar(index), type);
     }
   }
 }
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/scan/ScanOperatorExec.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/scan/ScanOperatorExec.java
index b6c77ca..699eb1f 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/scan/ScanOperatorExec.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/scan/ScanOperatorExec.java
@@ -263,7 +263,7 @@ public class ScanOperatorExec implements OperatorExec {
       }
 
       // Another reader available?
-      if (! nextReader()) {
+      if (!nextReader()) {
         finalizeResults();
         return;
       }
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/scan/columns/ColumnsScanFramework.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/scan/columns/ColumnsScanFramework.java
index 17b3f68..842c798 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/scan/columns/ColumnsScanFramework.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/scan/columns/ColumnsScanFramework.java
@@ -110,5 +110,6 @@ public class ColumnsScanFramework extends FileScanFramework {
   public static TupleMetadata columnsSchema() {
     return new SchemaBuilder()
       .addArray(ColumnsScanFramework.COLUMNS_COL, MinorType.VARCHAR)
-      .buildSchema();  }
+      .buildSchema();
+  }
 }
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/scan/convert/AbstractConvertFromString.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/scan/convert/AbstractConvertFromString.java
index f31ee69..76d2616 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/scan/convert/AbstractConvertFromString.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/scan/convert/AbstractConvertFromString.java
@@ -95,10 +95,10 @@ public abstract class AbstractConvertFromString extends DirectConverter {
    * @param properties optional framework-specific properties
    * @return a function to call to prepare each string value for conversion
    */
-  private Function<String,String> buildPrepare(ColumnMetadata schema,
+  private Function<String, String> buildPrepare(ColumnMetadata schema,
       Map<String, String> properties) {
 
-    String blankProp = properties.get(ColumnMetadata.BLANK_AS_PROP);
+    String blankProp = properties == null ? null : properties.get(ColumnMetadata.BLANK_AS_PROP);
     if (blankProp != null) {
       switch (blankProp.toLowerCase()) {
         case ColumnMetadata.BLANK_AS_ZERO:
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/scan/convert/StandardConversions.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/scan/convert/StandardConversions.java
index 7abaac5..396e741 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/scan/convert/StandardConversions.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/scan/convert/StandardConversions.java
@@ -24,21 +24,13 @@ import java.util.Map;
 import org.apache.drill.common.types.TypeProtos.MinorType;
 import org.apache.drill.exec.record.metadata.ColumnMetadata;
 import org.apache.drill.exec.record.metadata.TupleMetadata;
-import org.apache.drill.exec.record.metadata.TupleSchema;
 import org.apache.drill.exec.vector.accessor.ScalarWriter;
 import org.apache.drill.exec.vector.accessor.ValueWriter;
 
 /**
  * Factory for standard conversions as outlined in the package header.
- * Can be used in two ways:
- * <ul>
- * <li>As an object with a default set of properties that apply to
- * all columns.</li>
- * <li>As a set of static functions which operate on each column
- * individually.</li>
- * </ul>
- * The object form is more typical: it allows the provided schema to
- * set general blank-handling behavior at the tuple level.
+ * Use the builder to add schema-wide properties from a provided schema,
+ * from context-specific properties or both.
  * <p>
  * The class provides two kinds of information:
  * <p>
@@ -111,6 +103,42 @@ public class StandardConversions {
     }
   }
 
+  public static class Builder {
+
+    private final Map<String, String> properties = new HashMap<>();
+
+    public Builder withSchema(TupleMetadata providedSchema) {
+      if (providedSchema != null && providedSchema.hasProperties()) {
+        properties.putAll(providedSchema.properties());
+      }
+      return this;
+    }
+
+    public Builder withProperties(Map<String, String> properties) {
+      if (properties != null) {
+        this.properties.putAll(properties);
+      }
+      return this;
+    }
+
+    public Builder property(String key, String value) {
+      if (value == null) {
+        properties.remove(key);
+      } else {
+        properties.put(key, value);
+      }
+      return this;
+    }
+
+    public Builder blankAs(String value) {
+      return property(ColumnMetadata.BLANK_AS_PROP, value);
+    }
+
+    public StandardConversions build() {
+      return new StandardConversions(properties);
+    }
+  }
+
   public static final ConversionDefn IMPLICIT =
       new ConversionDefn(ConversionType.IMPLICIT);
   public static final ConversionDefn IMPLICIT_UNSAFE =
@@ -120,32 +148,14 @@ public class StandardConversions {
 
   private final Map<String, String> properties;
 
-  public StandardConversions() {
-    this.properties = null;
-  }
-
-  public StandardConversions(Map<String,String> properties) {
+  private StandardConversions(Map<String, String> properties) {
     this.properties = properties;
   }
 
-  public StandardConversions(String blankAsProp) {
-    if (blankAsProp == null) {
-      this.properties = null;
-    } else {
-      this.properties = new HashMap<>();
-      this.properties.put(ColumnMetadata.BLANK_AS_PROP, blankAsProp);
-    }
-  }
-
-  public StandardConversions(TupleMetadata providedSchema) {
-    if (providedSchema == null || !providedSchema.hasProperties()) {
-      this.properties = null;
-    } else {
-      this.properties = providedSchema.properties();
-    }
-  }
+  public static Builder builder() { return new Builder(); }
 
-  private Map<String,String> merge(Map<String,String> specificProps) {
+  private static Map<String, String> mergeProperties(Map<String, String> properties,
+      Map<String, String> specificProps) {
     if (properties == null) {
       return specificProps;
     } else if (specificProps == null) {
@@ -157,15 +167,19 @@ public class StandardConversions {
     return merged;
   }
 
-  public static DirectConverter newInstance(
+  private Map<String, String> mergeProperties(Map<String, String> specificProps) {
+    return mergeProperties(properties, specificProps);
+  }
+
+  public DirectConverter newInstance(
       Class<? extends DirectConverter> conversionClass, ScalarWriter baseWriter,
-      Map<String,String> properties) {
+      Map<String, String> properties) {
 
     // Try the Converter(ScalerWriter writer, Map<String, String> props) constructor first.
     // This first form is optional.
     try {
       final Constructor<? extends DirectConverter> ctor = conversionClass.getDeclaredConstructor(ScalarWriter.class, Map.class);
-      return ctor.newInstance(baseWriter, properties);
+      return ctor.newInstance(baseWriter, mergeProperties(properties));
     } catch (final ReflectiveOperationException e) {
       // Ignore
     }
@@ -174,7 +188,7 @@ public class StandardConversions {
     return newInstance(conversionClass, baseWriter);
   }
 
-  public static DirectConverter newInstance(
+  public DirectConverter newInstance(
       Class<? extends DirectConverter> conversionClass, ScalarWriter baseWriter) {
     try {
       final Constructor<? extends DirectConverter> ctor = conversionClass.getDeclaredConstructor(ScalarWriter.class);
@@ -196,11 +210,11 @@ public class StandardConversions {
    * @return a description of the conversion needed (if any), along with the
    * standard conversion class, if available
    */
-  public static ConversionDefn analyze(ColumnMetadata inputSchema, ColumnMetadata outputSchema) {
+  public ConversionDefn analyze(ColumnMetadata inputSchema, ColumnMetadata outputSchema) {
     return analyze(inputSchema.type(), outputSchema);
   }
 
-  public static ConversionDefn analyze(MinorType inputType, ColumnMetadata outputSchema) {
+  public ConversionDefn analyze(MinorType inputType, ColumnMetadata outputSchema) {
     if (inputType == outputSchema.type()) {
       return new ConversionDefn(ConversionType.NONE);
     }
@@ -364,8 +378,7 @@ public class StandardConversions {
     return EXPLICIT;
   }
 
-  public static Class<? extends DirectConverter> convertFromVarchar(
-      ColumnMetadata outputDefn) {
+  public Class<? extends DirectConverter> convertFromVarchar(ColumnMetadata outputDefn) {
     switch (outputDefn.type()) {
       case BIT:
         return ConvertStringToBoolean.class;
@@ -404,75 +417,33 @@ public class StandardConversions {
    *
    * @param scalarWriter the output column writer
    * @param inputType the type of the input data
-   * @param properties optional properties for some string-based conversions
+   * @param columnProps optional properties for some string-based conversions
    * @return a column converter, if needed and available, the input writer if
    * no conversion is needed, or null if there is no conversion available
    */
-  public static ValueWriter converterFor(ScalarWriter scalarWriter, MinorType inputType, Map<String, String> properties) {
+  public ValueWriter converterFor(ScalarWriter scalarWriter, MinorType inputType, Map<String, String> columnProps) {
     ConversionDefn defn = analyze(inputType, scalarWriter.schema());
     switch (defn.type) {
       case EXPLICIT:
         if (defn.conversionClass != null) {
-          return newInstance(defn.conversionClass, scalarWriter, properties);
+          return newInstance(defn.conversionClass, scalarWriter, columnProps);
         } else {
           return null;
         }
       case IMPLICIT:
       case IMPLICIT_UNSAFE:
-        return newInstance(defn.conversionClass, scalarWriter, properties);
+        return newInstance(defn.conversionClass, scalarWriter, columnProps);
       default:
         return scalarWriter;
     }
   }
 
-  public static ValueWriter converterFor(ScalarWriter scalarWriter, MinorType inputType) {
+  public ValueWriter converterFor(ScalarWriter scalarWriter, MinorType inputType) {
     return converterFor(scalarWriter, inputType, null);
   }
 
-  public static ValueWriter converterFor(ScalarWriter scalarWriter, ColumnMetadata inputSchema) {
-    return converterFor(scalarWriter, inputSchema.type(), inputSchema.properties());
-  }
-
-  public ValueWriter converter(ScalarWriter scalarWriter, ColumnMetadata inputSchema) {
+  public ValueWriter converterFor(ScalarWriter scalarWriter, ColumnMetadata inputSchema) {
     return converterFor(scalarWriter, inputSchema.type(),
-        merge(inputSchema.properties()));
-  }
-
-  public ValueWriter converter(ScalarWriter scalarWriter, MinorType inputType) {
-    return converterFor(scalarWriter, inputType, properties);
-  }
-
-  /**
-   * Given a desired provided schema and an actual reader schema, create a merged
-   * schema that contains the provided column where available, but the reader
-   * column otherwise. Copies provided properties to the output schema.
-   * <p>
-   * The result is the schema to use when creating column writers: it reflects
-   * the type of the target vector. The reader is responsible for converting from
-   * the (possibly different) reader column type to the provided column type.
-   * <p>
-   * Note: the provided schema should only contain types that the reader is prepared
-   * to offer: there is no requirement that the reader support every possible conversion,
-   * only those that make sense for that one reader.
-   *
-   * @param providedSchema the provided schema from {@code CREATE SCHEMA}
-   * @param readerSchema the set of column types that the reader can provide
-   * "natively"
-   * @return a merged schema to use when creating the {@code ResultSetLoader}
-   */
-  public static TupleMetadata mergeSchemas(TupleMetadata providedSchema,
-      TupleMetadata readerSchema) {
-    if (providedSchema == null) {
-      return readerSchema;
-    }
-    final TupleMetadata tableSchema = new TupleSchema();
-    for (ColumnMetadata readerCol : readerSchema) {
-      final ColumnMetadata providedCol = providedSchema.metadata(readerCol.name());
-      tableSchema.addColumn(providedCol == null ? readerCol : providedCol);
-    }
-    if (providedSchema.hasProperties()) {
-      tableSchema.properties().putAll(providedSchema.properties());
-    }
-    return tableSchema;
+        inputSchema.hasProperties() ? inputSchema.properties() : null);
   }
 }
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/scan/v3/FixedReceiver.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/scan/v3/FixedReceiver.java
new file mode 100644
index 0000000..923214a
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/scan/v3/FixedReceiver.java
@@ -0,0 +1,208 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.drill.exec.physical.impl.scan.v3;
+
+import org.apache.drill.common.exceptions.CustomErrorContext;
+import org.apache.drill.common.exceptions.UserException;
+import org.apache.drill.common.types.TypeProtos.DataMode;
+import org.apache.drill.exec.physical.impl.scan.convert.StandardConversions;
+import org.apache.drill.exec.physical.resultSet.ResultSetLoader;
+import org.apache.drill.exec.physical.resultSet.RowSetLoader;
+import org.apache.drill.exec.record.metadata.ColumnMetadata;
+import org.apache.drill.exec.record.metadata.MetadataUtils;
+import org.apache.drill.exec.record.metadata.TupleMetadata;
+import org.apache.drill.exec.record.metadata.TupleNameSpace;
+import org.apache.drill.exec.record.metadata.TupleSchema;
+import org.apache.drill.exec.vector.accessor.ScalarWriter;
+import org.apache.drill.exec.vector.accessor.ValueWriter;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Layer above the {@code ResultSetLoader} which handles standard conversions
+ * for scalar columns where the schema is known up front (i.e. "early schema".)
+ * Columns access is by both name and position, though access by position is
+ * faster and is preferred where possible for performance.
+ */
+public class FixedReceiver {
+  private static final Logger logger = LoggerFactory.getLogger(FixedReceiver.class);
+
+  public static class Builder {
+    private final SchemaNegotiator negotiator;
+    private final TupleMetadata providedSchema;
+    private final StandardConversions.Builder conversionBuilder = StandardConversions.builder();
+    private boolean isComplete;
+    private RowSetLoader rowWriter;
+
+    public Builder(SchemaNegotiator negotiator) {
+      this.negotiator = negotiator;
+      this.providedSchema = negotiator.providedSchema();
+    }
+
+    /**
+     * Provides access to the conversion builder to add custom properties.
+      */
+    public StandardConversions.Builder conversionBuilder() {
+      return conversionBuilder;
+    }
+
+    /**
+     * Mark that the reader schema provided to {@link #build(TupleMetadata)}
+     * contains all columns that this reader will deliver. Allows some
+     * optimizations. See {@link SchemaNegotiator#schemaIsComplete(boolean)}.
+     */
+    public Builder schemaIsComplete() {
+      isComplete = true;
+      return this;
+    }
+
+    /**
+     * Create a fixed receiver for the provided schema (if any) in the
+     * scan plan, and the given reader schema. Assumes no new columns will
+     * be added later in the read.
+     */
+    public FixedReceiver build(TupleMetadata readerSchema) {
+      StandardConversions conversions = conversionBuilder.build();
+      TupleMetadata writerSchema = mergeSchemas(negotiator.providedSchema(), readerSchema);
+      negotiator.tableSchema(writerSchema);
+      negotiator.schemaIsComplete(isComplete);
+      ResultSetLoader loader = negotiator.build();
+      rowWriter = loader.writer();
+      TupleNameSpace<ValueWriter> writers = new TupleNameSpace<>();
+      for (ColumnMetadata col : readerSchema) {
+        writers.add(col.name(), writerFor(col, conversions));
+      }
+      return new FixedReceiver(rowWriter, writers);
+    }
+
+    /**
+     * Given a desired provided schema and an actual reader schema, create a merged
+     * schema that contains the provided column where available, but the reader
+     * column otherwise. Copies provided properties to the output schema.
+     * <p>
+     * The result is the schema to use when creating column writers: it reflects
+     * the type of the target vector. The reader is responsible for converting from
+     * the (possibly different) reader column type to the provided column type.
+     * <p>
+     * Note: the provided schema should only contain types that the reader is prepared
+     * to offer: there is no requirement that the reader support every possible conversion,
+     * only those that make sense for that one reader.
+     *
+     * @param providedSchema the provided schema from {@code CREATE SCHEMA}
+     * @param readerSchema the set of column types that the reader can provide
+     * "natively"
+     * @return a merged schema to use when creating the {@code ResultSetLoader}
+     */
+    public static TupleMetadata mergeSchemas(TupleMetadata providedSchema,
+        TupleMetadata readerSchema) {
+      if (providedSchema == null) {
+        return readerSchema;
+      }
+      final TupleMetadata tableSchema = new TupleSchema();
+      for (ColumnMetadata readerCol : readerSchema) {
+        final ColumnMetadata providedCol = providedSchema.metadata(readerCol.name());
+        tableSchema.addColumn(providedCol == null ? readerCol : providedCol);
+      }
+      if (providedSchema.hasProperties()) {
+        tableSchema.properties().putAll(providedSchema.properties());
+      }
+      return tableSchema;
+    }
+
+    private ValueWriter writerFor(ColumnMetadata readerCol, StandardConversions conversions) {
+      if (!MetadataUtils.isScalar(readerCol)) {
+        throw UserException.internalError()
+            .message("FixedReceiver only works with scalar columns, reader column is not scalar")
+            .addContext("Column name", readerCol.name())
+            .addContext("Column type", readerCol.type().name())
+            .addContext(errorContext())
+            .build(logger);
+      }
+      ScalarWriter baseWriter = rowWriter.scalar(readerCol.name());
+      if (!rowWriter.isProjected(readerCol.name())) {
+        return baseWriter;
+      }
+      ColumnMetadata providedCol = providedCol(readerCol.name());
+      if (providedCol == null) {
+        return baseWriter;
+      }
+      if (!MetadataUtils.isScalar(providedCol)) {
+        throw UserException.validationError()
+          .message("FixedReceiver only works with scalar columns, provided column is not scalar")
+          .addContext("Provided column name", providedCol.name())
+          .addContext("Provided column type", providedCol.type().name())
+          .addContext(errorContext())
+          .build(logger);
+      }
+      if (!compatibleModes(readerCol.mode(), providedCol.mode())) {
+        throw UserException.validationError()
+          .message("Reader and provided columns have incompatible cardinality")
+          .addContext("Column name", providedCol.name())
+          .addContext("Provided column mode", providedCol.mode().name())
+          .addContext("Reader column mode", readerCol.mode().name())
+          .addContext(errorContext())
+          .build(logger);
+      }
+      return conversions.converterFor(baseWriter, readerCol.type());
+    }
+
+    private boolean compatibleModes(DataMode source, DataMode dest) {
+      return source == dest ||
+             dest == DataMode.OPTIONAL && source == DataMode.REQUIRED;
+    }
+
+    private ColumnMetadata providedCol(String name) {
+      return providedSchema == null ? null : providedSchema.metadata(name);
+    }
+
+    private CustomErrorContext errorContext( ) {
+      return negotiator.errorContext();
+    }
+  }
+
+  private final RowSetLoader rowWriter;
+  private final TupleNameSpace<ValueWriter> writers;
+
+  private FixedReceiver(RowSetLoader rowWriter,
+      TupleNameSpace<ValueWriter> writers) {
+    this.rowWriter = rowWriter;
+    this.writers = writers;
+  }
+
+  public static Builder builderFor(SchemaNegotiator negotiator) {
+    return new Builder(negotiator);
+  }
+
+  public boolean start() {
+    return rowWriter.start();
+  }
+
+  public ValueWriter scalar(int index) {
+    return writers.get(index);
+  }
+
+  public ValueWriter scalar(String name) {
+    return writers.get(name);
+  }
+
+  public void save() {
+    rowWriter.save();
+  }
+
+  public RowSetLoader rowWriter() { return rowWriter; }
+}
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/scan/v3/ScanLifecycleBuilder.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/scan/v3/ScanLifecycleBuilder.java
index f4393f1..393fc23 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/scan/v3/ScanLifecycleBuilder.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/scan/v3/ScanLifecycleBuilder.java
@@ -24,12 +24,17 @@ import org.apache.drill.common.exceptions.CustomErrorContext;
 import org.apache.drill.common.exceptions.EmptyErrorContext;
 import org.apache.drill.common.expression.SchemaPath;
 import org.apache.drill.common.types.TypeProtos.MajorType;
+import org.apache.drill.exec.ops.FragmentContext;
 import org.apache.drill.exec.ops.OperatorContext;
+import org.apache.drill.exec.physical.base.PhysicalOperator;
 import org.apache.drill.exec.physical.impl.protocol.OperatorDriver;
+import org.apache.drill.exec.physical.impl.protocol.OperatorRecordBatch;
 import org.apache.drill.exec.physical.impl.scan.ScanOperatorExec;
 import org.apache.drill.exec.physical.impl.scan.v3.lifecycle.ScanEventListener;
 import org.apache.drill.exec.physical.impl.scan.v3.lifecycle.ScanLifecycle;
+import org.apache.drill.exec.physical.impl.scan.v3.schema.ScanSchemaTracker;
 import org.apache.drill.exec.record.metadata.TupleMetadata;
+import org.apache.drill.exec.server.options.OptionSet;
 import org.apache.drill.exec.vector.ValueVector;
 import org.apache.drill.shaded.guava.com.google.common.annotations.VisibleForTesting;
 
@@ -80,6 +85,11 @@ public class ScanLifecycleBuilder {
     }
   }
 
+  public interface SchemaValidator {
+    void validate(ScanSchemaTracker schema);
+  }
+
+  private OptionSet options;
   private ReaderFactory<?> readerFactory;
   protected String userName;
   protected MajorType nullType;
@@ -133,10 +143,24 @@ public class ScanLifecycleBuilder {
   protected boolean allowSchemaChange = true;
 
   /**
+   * Optional schema validator to perform per-scan checks of the
+   * projection or resolved schema.
+   */
+  protected SchemaValidator schemaValidator;
+
+  /**
    * Context for error messages.
    */
   protected CustomErrorContext errorContext;
 
+  public void options(OptionSet options) {
+    this.options = options;
+  }
+
+  public OptionSet options() {
+    return options;
+  }
+
   public void readerFactory(ReaderFactory<?> readerFactory) {
     this.readerFactory = readerFactory;
   }
@@ -258,6 +282,12 @@ public class ScanLifecycleBuilder {
     return readerFactory;
   }
 
+  public void schemaValidator(SchemaValidator schemaValidator) {
+    this.schemaValidator = schemaValidator;
+  }
+
+  public SchemaValidator schemaValidator() { return schemaValidator; }
+
   public ScanLifecycle build(OperatorContext context) {
     return new ScanLifecycle(context, this);
   }
@@ -266,6 +296,10 @@ public class ScanLifecycleBuilder {
   public ScanOperatorExec buildScan() {
     return new ScanOperatorExec(
         new ScanEventListener(this),
-        !disableEmptyResults);
+             !disableEmptyResults);
+  }
+
+  public OperatorRecordBatch buildScanOperator(FragmentContext fragContext, PhysicalOperator pop) {
+    return new OperatorRecordBatch(fragContext, pop, buildScan(), enableSchemaBatch);
   }
 }
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/scan/v3/SchemaNegotiator.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/scan/v3/SchemaNegotiator.java
index 01d9b4c..9dee1d7 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/scan/v3/SchemaNegotiator.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/scan/v3/SchemaNegotiator.java
@@ -19,6 +19,7 @@ package org.apache.drill.exec.physical.impl.scan.v3;
 
 import org.apache.drill.common.exceptions.CustomErrorContext;
 import org.apache.drill.exec.ops.OperatorContext;
+import org.apache.drill.exec.physical.impl.scan.v3.schema.ProjectedColumn;
 import org.apache.drill.exec.physical.resultSet.ResultSetLoader;
 import org.apache.drill.exec.physical.resultSet.RowSetLoader;
 import org.apache.drill.exec.record.metadata.TupleMetadata;
@@ -116,6 +117,13 @@ public interface SchemaNegotiator {
   void setErrorContext(CustomErrorContext context);
 
   /**
+   * Returns the error context to use for this reader: either the
+   * parent or the reader-specific context set in
+   * {@link #setErrorContext(CustomErrorContext)}.
+   */
+  CustomErrorContext errorContext();
+
+  /**
    * Name of the user running the query.
    */
   String userName();
@@ -134,6 +142,8 @@ public interface SchemaNegotiator {
    */
   boolean isProjectionEmpty();
 
+  ProjectedColumn projectionFor(String colName);
+
   /**
    * Returns the provided schema, if defined. The provided schema is a
    * description of the source schema viewed as a Drill schema.
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/scan/v3/file/FileDescrip.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/scan/v3/file/FileDescrip.java
index bb92772..c6fff11 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/scan/v3/file/FileDescrip.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/scan/v3/file/FileDescrip.java
@@ -17,10 +17,18 @@
  */
 package org.apache.drill.exec.physical.impl.scan.v3.file;
 
+import java.io.IOException;
+import java.io.InputStream;
+
 import org.apache.drill.exec.store.ColumnExplorer;
+import org.apache.drill.exec.store.dfs.DrillFileSystem;
+import org.apache.drill.exec.store.dfs.easy.FileWork;
+import org.apache.drill.shaded.guava.com.google.common.annotations.VisibleForTesting;
 import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.mapred.FileSplit;
 
 /**
+ * Describes one file within a scan and is used to populate implicit columns.
  * Specify the file name and optional selection root. If the selection root
  * is provided, then partitions are defined as the portion of the file name
  * that is not also part of the selection root. That is, if selection root is
@@ -28,13 +36,35 @@ import org.apache.hadoop.fs.Path;
  */
 public class FileDescrip {
 
-  private final Path filePath;
+  private final DrillFileSystem dfs;
+  private final FileWork fileWork;
+  private final FileSplit split;
   private final String[] dirPath;
 
-  public FileDescrip(Path filePath, Path selectionRoot) {
-    this.filePath = filePath;
+  // Option to open the file as optionally compressed
+  private boolean isCompressible;
+
+  // Parquet-related attributes
+  protected Integer rowGroupIndex;
+  protected Long rowGroupStart;
+  protected Long rowGroupLength;
+
+  // Cached modification time. Cached as a string because
+  // that's the odd way we return the value.
+  private String modTime;
+
+  // Flag to indicate that the file turned out to be empty.
+  // Used to set one of the internal implicit columns.
+  protected boolean isEmpty;
+
+  public FileDescrip(DrillFileSystem dfs, FileWork fileWork, Path selectionRoot) {
+    this.dfs = dfs;
+    this.fileWork = fileWork;
+    Path path = dfs.makeQualified(fileWork.getPath());
+    this.split = new FileSplit(path, fileWork.getStart(), fileWork.getLength(), new String[]{""});
 
     // If the data source is not a file, no file metadata is available.
+    Path filePath = fileWork.getPath();
     if (selectionRoot == null || filePath == null) {
       dirPath = null;
       return;
@@ -57,7 +87,25 @@ public class FileDescrip {
     }
   }
 
-  public Path filePath() { return filePath; }
+  /**
+   * Gives the Drill file system for this operator.
+   */
+  public DrillFileSystem fileSystem() { return dfs; }
+
+  /**
+   * Returns Drill's version of the Hadoop file split.
+   */
+  public Path filePath() { return fileWork.getPath(); }
+
+  /**
+   * Describes the file split (path and block offset) for this scan.
+   *
+   * @return Hadoop file split object with the file path, block
+   * offset, and length.
+   */
+  public FileSplit split() { return split; }
+
+  public FileWork fileWork() { return fileWork; }
 
   public String partition(int index) {
     if (dirPath == null ||  dirPath.length <= index) {
@@ -70,5 +118,49 @@ public class FileDescrip {
     return dirPath == null ? 0 : dirPath.length;
   }
 
-  public boolean isSet() { return filePath != null; }
+  public void setRowGroupAttribs(int index, long start, long length) {
+    this.rowGroupIndex = index;
+    this.rowGroupStart = start;
+    this.rowGroupLength = length;
+  }
+
+  public String getModTime() {
+    if (modTime == null) {
+      try {
+        modTime = String.valueOf(dfs.getFileStatus(filePath()).getModificationTime());
+      } catch (IOException e) {
+
+        // This is an odd place to catch and report errors. Assume that, if the file
+        // has problems, the call to open the file will fail and will return a better
+        // error message than we can provide here.
+      }
+    }
+    return modTime;
+  }
+
+  /**
+   * Explicitly set the cached modification time. For testing only.
+   */
+  @VisibleForTesting
+  public void setModTime(String modTime) {
+    this.modTime = modTime;
+  }
+
+  public void setCompressible(boolean isCompressed) {
+    this.isCompressible = isCompressed;
+  }
+
+  public boolean isCompressible() { return isCompressible; }
+
+  public InputStream open() throws IOException {
+    if (isCompressible) {
+      return dfs.openPossiblyCompressedStream(filePath());
+    } else {
+      return dfs.open(filePath());
+    }
+  }
+
+  public void markEmpty() {
+    isEmpty = true;
+  }
 }
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/scan/v3/file/FileScanLifecycle.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/scan/v3/file/FileScanLifecycle.java
index eb4fc07..0db2cdc 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/scan/v3/file/FileScanLifecycle.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/scan/v3/file/FileScanLifecycle.java
@@ -77,9 +77,10 @@ public class FileScanLifecycle extends ScanLifecycle {
 
     // Create the implicit columns manager
     this.implicitColumnsHandler = new ImplicitFileColumnsHandler(
-        context, options, vectorCache(), schemaTracker());
+        dfs, context.getFragmentContext().getOptions(),
+        options, vectorCache(), schemaTracker());
 
-    // Bind the reader factory which intializes the list
+    // Bind the reader factory which initializes the list
     // of splits from the builder.
     FileReaderFactory readerFactory = (FileReaderFactory) readerFactory();
     readerFactory.bind(this);
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/scan/v3/file/FileScanLifecycleBuilder.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/scan/v3/file/FileScanLifecycleBuilder.java
index a791a35..85f9dbf 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/scan/v3/file/FileScanLifecycleBuilder.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/scan/v3/file/FileScanLifecycleBuilder.java
@@ -17,22 +17,25 @@
  */
 package org.apache.drill.exec.physical.impl.scan.v3.file;
 
+import java.util.ArrayList;
 import java.util.List;
 
 import org.apache.drill.exec.ops.OperatorContext;
 import org.apache.drill.exec.physical.impl.scan.v3.ScanLifecycleBuilder;
 import org.apache.drill.exec.physical.impl.scan.v3.lifecycle.ScanLifecycle;
 import org.apache.drill.exec.store.dfs.easy.FileWork;
+import org.apache.drill.exec.store.schedule.CompleteFileWork.FileWorkImpl;
 import org.apache.drill.shaded.guava.com.google.common.base.Preconditions;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.Path;
 
 public class FileScanLifecycleBuilder extends ScanLifecycleBuilder {
   protected int maxPartitionDepth;
-  protected boolean useLegacyWildcardExpansion;
+  protected boolean useLegacyWildcardExpansion = true;
   protected Path rootDir;
   private List<FileWork> splits;
   private Configuration fsConf;
+  private boolean compressible;
 
   public void fileSystemConfig(Configuration fsConf) {
     this.fsConf = fsConf;
@@ -42,6 +45,14 @@ public class FileScanLifecycleBuilder extends ScanLifecycleBuilder {
     this.splits = splits;
   }
 
+  /**
+   * Legacy version because the file scan operator exposes the
+   * implementation, not the interface.
+   */
+  public void fileSplitImpls(List<FileWorkImpl> splits) {
+    this.splits = new ArrayList<>(splits);
+  }
+
   public void maxPartitionDepth(int maxPartitionDepth) {
     this.maxPartitionDepth = maxPartitionDepth;
   }
@@ -54,6 +65,10 @@ public class FileScanLifecycleBuilder extends ScanLifecycleBuilder {
     this.rootDir = rootDir;
   }
 
+  public void compressible(boolean compressible) {
+    this.compressible = compressible;
+  }
+
   public List<FileWork> splits() {
     return Preconditions.checkNotNull(splits);
   }
@@ -65,20 +80,16 @@ public class FileScanLifecycleBuilder extends ScanLifecycleBuilder {
     return fsConf;
   }
 
-  @Override
-  public ScanLifecycle build(OperatorContext context) {
-    return new FileScanLifecycle(context, this);
-  }
+  public int maxPartitionDepth() { return maxPartitionDepth; }
 
-  public int maxPartitionDepth() {
-    return maxPartitionDepth;
-  }
+  public boolean useLegacyWildcardExpansion() { return useLegacyWildcardExpansion; }
 
-  public boolean useLegacyWildcardExpansion() {
-    return useLegacyWildcardExpansion;
-  }
+  public Path rootDir() { return rootDir; }
+
+  public boolean isCompressible() { return compressible; }
 
-  public Path rootDir() {
-    return rootDir;
+  @Override
+  public ScanLifecycle build(OperatorContext context) {
+    return new FileScanLifecycle(context, this);
   }
 }
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/scan/v3/file/FileSchemaNegotiator.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/scan/v3/file/FileSchemaNegotiator.java
index 156b5e4..86dd2b3 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/scan/v3/file/FileSchemaNegotiator.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/scan/v3/file/FileSchemaNegotiator.java
@@ -18,9 +18,6 @@
 package org.apache.drill.exec.physical.impl.scan.v3.file;
 
 import org.apache.drill.exec.physical.impl.scan.v3.SchemaNegotiator;
-import org.apache.drill.exec.store.dfs.DrillFileSystem;
-import org.apache.drill.exec.store.dfs.easy.FileWork;
-import org.apache.hadoop.mapred.FileSplit;
 
 /**
  * The file schema negotiator provides access to the Drill file system
@@ -29,20 +26,9 @@ import org.apache.hadoop.mapred.FileSplit;
 public interface FileSchemaNegotiator extends SchemaNegotiator {
 
   /**
-   * Gives the Drill file system for this operator.
+   * Gives the file description which holds the Drill file system,
+   * split, file work and format-specific properties. Can open the
+   * file and provides information used to populate implicit columns.
    */
-  DrillFileSystem fileSystem();
-
-  /**
-   * Describes the file split (path and block offset) for this scan.
-   *
-   * @return Hadoop file split object with the file path, block
-   * offset, and length.
-   */
-  FileSplit split();
-
-  /**
-   * Returns Drill's version of the Hadoop file split.
-   */
-  FileWork fileWork();
+  FileDescrip file();
 }
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/scan/v3/file/FileSchemaNegotiatorImpl.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/scan/v3/file/FileSchemaNegotiatorImpl.java
index 8457b2f..9d69a97 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/scan/v3/file/FileSchemaNegotiatorImpl.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/scan/v3/file/FileSchemaNegotiatorImpl.java
@@ -21,14 +21,13 @@ import org.apache.drill.common.exceptions.ChildErrorContext;
 import org.apache.drill.common.exceptions.CustomErrorContext;
 import org.apache.drill.common.exceptions.UserException.Builder;
 import org.apache.drill.exec.physical.impl.scan.v3.ManagedReader;
-import org.apache.drill.exec.physical.impl.scan.v3.ReaderFactory;
 import org.apache.drill.exec.physical.impl.scan.v3.ManagedReader.EarlyEofException;
+import org.apache.drill.exec.physical.impl.scan.v3.ReaderFactory;
 import org.apache.drill.exec.physical.impl.scan.v3.lifecycle.ReaderLifecycle;
 import org.apache.drill.exec.physical.impl.scan.v3.lifecycle.SchemaNegotiatorImpl;
 import org.apache.drill.exec.physical.impl.scan.v3.lifecycle.StaticBatchBuilder;
-import org.apache.drill.exec.store.dfs.DrillFileSystem;
+import org.apache.drill.exec.physical.resultSet.ResultSetLoader;
 import org.apache.drill.exec.store.dfs.easy.FileWork;
-import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.mapred.FileSplit;
 
 /**
@@ -49,7 +48,8 @@ public class FileSchemaNegotiatorImpl extends SchemaNegotiatorImpl
     @Override
     public void addContext(Builder builder) {
       super.addContext(builder);
-      builder.addContext("File", Path.getPathWithoutSchemeAndAuthority(split.getPath()).toString());
+      FileSplit split = fileDescrip.split();
+      builder.addContext("File", split.getPath().toString());
       if (split.getStart() != 0) {
         builder.addContext("Offset", split.getStart());
         builder.addContext("Length", split.getLength());
@@ -57,8 +57,7 @@ public class FileSchemaNegotiatorImpl extends SchemaNegotiatorImpl
     }
   }
 
-  private FileWork fileWork;
-  private FileSplit split;
+  private FileDescrip fileDescrip;
 
   public FileSchemaNegotiatorImpl(ReaderLifecycle readerLifecycle) {
     super(readerLifecycle);
@@ -67,19 +66,11 @@ public class FileSchemaNegotiatorImpl extends SchemaNegotiatorImpl
   }
 
   public void bindSplit(FileWork fileWork) {
-    this.fileWork = fileWork;
-    Path path = fileScan().fileSystem().makeQualified(fileWork.getPath());
-    split = new FileSplit(path, fileWork.getStart(), fileWork.getLength(), new String[]{""});
+    fileDescrip = fileScan().implicitColumnsHandler().makeDescrip(fileWork);
   }
 
   @Override
-  public DrillFileSystem fileSystem() { return fileScan().fileSystem(); }
-
-  @Override
-  public FileSplit split() { return split; }
-
-  @Override
-  public FileWork fileWork() { return fileWork; }
+  public FileDescrip file() { return fileDescrip; }
 
   @Override
   @SuppressWarnings("unchecked")
@@ -89,10 +80,32 @@ public class FileSchemaNegotiatorImpl extends SchemaNegotiatorImpl
 
   @Override
   public StaticBatchBuilder implicitColumnsLoader() {
-    return fileScan().implicitColumnsHandler().forFile(split.getPath());
+    return fileScan().implicitColumnsHandler().forFile(fileDescrip);
   }
 
   private FileScanLifecycle fileScan() {
     return (FileScanLifecycle) readerLifecycle.scanLifecycle();
   }
+
+  @Override
+  protected void onEndBatch() {
+
+    // If this is is a metadata scan, and this file has no rows (this is
+    // the first batch and contains no data), then add a dummy row so
+    // we have something to aggregate upon.
+    ImplicitFileColumnsHandler handler = fileScan().implicitColumnsHandler();
+    if (!handler.isMetadataScan()) {
+      return;
+    }
+    ResultSetLoader tableLoader = readerLifecycle.tableLoader();
+    if (tableLoader.batchCount() == 0 && !tableLoader.hasRows()) {
+
+      // This is admittedly a hack. The table may contain non-nullable
+      // columns, but we are asking for null values for those columns.
+      // We'll fill in defaults, with is not ideal.
+      tableLoader.writer().start();
+      tableLoader.writer().save();
+      fileDescrip.markEmpty();
+    }
+  }
 }
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/scan/v3/file/ImplicitColumnMarker.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/scan/v3/file/ImplicitColumnMarker.java
new file mode 100644
index 0000000..e8c9790
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/scan/v3/file/ImplicitColumnMarker.java
@@ -0,0 +1,99 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.drill.exec.physical.impl.scan.v3.file;
+
+import org.apache.drill.exec.store.ColumnExplorer.ImplicitFileColumns;
+import org.apache.drill.exec.store.ColumnExplorer.ImplicitInternalFileColumns;
+
+/**
+ * Marks a column as implicit and provides a function to resolve an
+ * implicit column given a description of the input file.
+ */
+public interface ImplicitColumnMarker {
+  String resolve(FileDescrip fileInfo);
+
+  /**
+   * Marker for a file-based, non-internal implicit column that
+   * extracts parts of the file name as defined by the implicit
+   * column definition.
+   */
+  public class FileImplicitMarker implements ImplicitColumnMarker {
+    public final ImplicitFileColumns defn;
+
+    public FileImplicitMarker(ImplicitFileColumns defn) {
+      this.defn = defn;
+    }
+
+    @Override
+    public String resolve(FileDescrip fileInfo) {
+      return defn.getValue(fileInfo.filePath());
+    }
+  }
+
+  /**
+   * Partition column defined by a partition depth from the scan
+   * root folder. Partitions that reference non-existent directory levels
+   * are null.
+   */
+  public class PartitionColumnMarker implements ImplicitColumnMarker {
+    private final int partition;
+
+    public PartitionColumnMarker(int partition) {
+      this.partition = partition;
+    }
+
+    @Override
+    public String resolve(FileDescrip fileInfo) {
+      return fileInfo.partition(partition);
+    }
+  }
+
+  public class InternalColumnMarker implements ImplicitColumnMarker {
+    public final ImplicitInternalFileColumns defn;
+
+    public InternalColumnMarker(ImplicitInternalFileColumns defn) {
+      this.defn = defn;
+    }
+
+    @Override
+    public String resolve(FileDescrip fileInfo) {
+      switch (defn) {
+      case ROW_GROUP_INDEX:
+        return valueOf(fileInfo.rowGroupIndex);
+      case ROW_GROUP_START:
+        return valueOf(fileInfo.rowGroupStart);
+      case ROW_GROUP_LENGTH:
+        return valueOf(fileInfo.rowGroupLength);
+      case PROJECT_METADATA:
+      case USE_METADATA:
+
+        // Per Metadata code: if file is empty (and record is a placeholder)
+        // return "false", else return null for valid rows.
+        return fileInfo.isEmpty ? Boolean.FALSE.toString() : null;
+      case LAST_MODIFIED_TIME:
+        return fileInfo.getModTime();
+      default:
+        throw new IllegalStateException(defn.name());
+      }
+    }
+
+    private String valueOf(Object value) {
+      return value == null ? null : String.valueOf(value);
+    }
+  }
+}
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/scan/v3/schema/ImplicitColumnResolver.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/scan/v3/file/ImplicitColumnResolver.java
similarity index 70%
rename from exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/scan/v3/schema/ImplicitColumnResolver.java
rename to exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/scan/v3/file/ImplicitColumnResolver.java
index fcc3ed8..3ec7f35 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/scan/v3/schema/ImplicitColumnResolver.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/scan/v3/file/ImplicitColumnResolver.java
@@ -15,7 +15,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.apache.drill.exec.physical.impl.scan.v3.schema;
+package org.apache.drill.exec.physical.impl.scan.v3.file;
 
 import java.util.ArrayList;
 import java.util.HashSet;
@@ -33,13 +33,22 @@ import org.apache.drill.common.types.TypeProtos.MajorType;
 import org.apache.drill.common.types.TypeProtos.MinorType;
 import org.apache.drill.common.types.Types;
 import org.apache.drill.exec.ExecConstants;
-import org.apache.drill.exec.physical.impl.scan.v3.file.FileDescrip;
+import org.apache.drill.exec.physical.impl.scan.v3.file.ImplicitColumnMarker.FileImplicitMarker;
+import org.apache.drill.exec.physical.impl.scan.v3.file.ImplicitColumnMarker.InternalColumnMarker;
+import org.apache.drill.exec.physical.impl.scan.v3.file.ImplicitColumnMarker.PartitionColumnMarker;
+import org.apache.drill.exec.physical.impl.scan.v3.schema.MutableTupleSchema;
 import org.apache.drill.exec.physical.impl.scan.v3.schema.MutableTupleSchema.ColumnHandle;
+import org.apache.drill.exec.physical.impl.scan.v3.schema.ProjectedColumn;
+import org.apache.drill.exec.physical.impl.scan.v3.schema.ScanSchemaTracker;
+import org.apache.drill.exec.physical.impl.scan.v3.schema.SchemaUtils;
 import org.apache.drill.exec.record.metadata.ColumnMetadata;
 import org.apache.drill.exec.record.metadata.MetadataUtils;
 import org.apache.drill.exec.record.metadata.TupleMetadata;
 import org.apache.drill.exec.server.options.OptionSet;
+import org.apache.drill.exec.store.ColumnExplorer.ImplicitFileColumn;
 import org.apache.drill.exec.store.ColumnExplorer.ImplicitFileColumns;
+import org.apache.drill.exec.store.ColumnExplorer.ImplicitInternalFileColumns;
+import org.apache.drill.exec.store.dfs.DrillFileSystem;
 import org.apache.drill.shaded.guava.com.google.common.base.Strings;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -83,6 +92,8 @@ public class ImplicitColumnResolver {
      */
     protected boolean useLegacyWildcardExpansion = true;
 
+    protected DrillFileSystem dfs;
+
     public ImplicitColumnOptions optionSet(OptionSet optionSet) {
       this.optionSet = optionSet;
       return this;
@@ -111,47 +122,10 @@ public class ImplicitColumnResolver {
       useLegacyWildcardExpansion = flag;
       return this;
     }
-  }
-
-  /**
-   * Provides a function to resolve an implicit column given a description
-   * of the input file.
-   */
-  public interface ColumnMarker {
-    String resolve(FileDescrip fileInfo);
-  }
-
-  /**
-   * Implicit column defined by an {@link ImplicitFileColumns}.
-   */
-  public static class ImplicitColumnMarker implements ColumnMarker {
-    private final ImplicitFileColumns defn;
-
-    public ImplicitColumnMarker(ImplicitFileColumns defn) {
-      this.defn = defn;
-    }
-
-    @Override
-    public String resolve(FileDescrip fileInfo) {
-      return defn.getValue(fileInfo.filePath());
-    }
-  }
-
-  /**
-   * Partition column defined by a partition depth from the scan
-   * root folder. Partitions that reference non-existent directory levels
-   * are null.
-   */
-  public static class PartitionColumnMarker implements ColumnMarker {
-    private final int partition;
-
-    private PartitionColumnMarker(int partition) {
-      this.partition = partition;
-    }
 
-    @Override
-    public String resolve(FileDescrip fileInfo) {
-      return fileInfo.partition(partition);
+    public ImplicitColumnOptions dfs(DrillFileSystem dfs) {
+      this.dfs = dfs;
+      return this;
     }
   }
 
@@ -161,16 +135,20 @@ public class ImplicitColumnResolver {
    * column markers which resolve the columns for each file.
    */
   public static class ParseResult {
-    private final List<ColumnMarker> columns;
+    private final List<ImplicitColumnMarker> columns;
     private final TupleMetadata schema;
+    private final boolean isMetadataScan;
 
-    protected ParseResult(List<ColumnMarker> columns, TupleMetadata schema) {
+    protected ParseResult(List<ImplicitColumnMarker> columns, TupleMetadata schema,
+        boolean isMetadataScan) {
       this.columns = columns;
       this.schema = schema;
+      this.isMetadataScan = isMetadataScan;
     }
 
     public TupleMetadata schema() { return schema; }
-    public List<ColumnMarker> columns() { return columns; }
+    public List<ImplicitColumnMarker> columns() { return columns; }
+    public boolean isMetadataScan() { return isMetadataScan; }
 
     public Object[] resolve(FileDescrip fileInfo) {
       Object values[] = new Object[columns.size()];
@@ -185,8 +163,9 @@ public class ImplicitColumnResolver {
     private final ImplicitColumnResolver parser;
     private final ScanSchemaTracker tracker;
     private final MutableTupleSchema scanSchema;
-    private final List<ColumnMarker> columns = new ArrayList<>();
+    private final List<ImplicitColumnMarker> columns = new ArrayList<>();
     private final Set<Integer> referencedPartitions = new HashSet<>();
+    private boolean isMetadataScan;
 
     protected ImplicitColumnParser(ImplicitColumnResolver parser, ScanSchemaTracker tracker) {
       this.parser = parser;
@@ -196,7 +175,7 @@ public class ImplicitColumnResolver {
 
     protected ParseResult parse() {
       for (ColumnHandle col : tracker.internalSchema().columns()) {
-        matchColumn(parser, col);
+        matchColumn(col);
       }
       if (tracker.internalSchema().projectionType() == ScanSchemaTracker.ProjectionType.ALL) {
         expandWildcard();
@@ -207,7 +186,7 @@ public class ImplicitColumnResolver {
       // appears out-of-order:
       // SELECT *, fileName
       // SELECT fileName, *
-      return new ParseResult(columns, tracker.applyImplicitCols());
+      return new ParseResult(columns, tracker.applyImplicitCols(), isMetadataScan);
     }
 
     private void expandWildcard() {
@@ -224,21 +203,21 @@ public class ImplicitColumnResolver {
         if (referencedPartitions.contains(i)) {
           continue;
         }
+        ImplicitColumnMarker marker = new PartitionColumnMarker(i);
         ColumnMetadata resolved = MetadataUtils.newScalar(parser.partitionName(i), PARTITION_COL_TYPE);
-        SchemaUtils.markAsPartition(resolved, i);
-        columns.add(new PartitionColumnMarker(i));
-        tracker.expandImplicitCol(resolved);
+        columns.add(marker);
+        tracker.expandImplicitCol(resolved, marker);
         referencedPartitions.add(i);
       }
     }
 
-    private void matchColumn(ImplicitColumnResolver parser, ColumnHandle col) {
+    private void matchColumn(ColumnHandle col) {
       String colType = SchemaUtils.implicitColType(col.column());
       if (colType != null) {
         resolveTaggedColumn(parser, col, colType);
         return;
       } else if (col.column().isDynamic()) {
-        matchByName(parser, col);
+        matchByName(col);
       }
     }
 
@@ -250,22 +229,23 @@ public class ImplicitColumnResolver {
         return;
       }
 
-      ImplicitFileColumns defn = parser.typeDefs.get(colType);
+      ImplicitFileColumn defn = parser.typeDefs.get(colType);
       if (defn != null) {
-        resolveImplicitColumn(defn, parser, col);
+        resolveImplicitColumn((ImplicitFileColumns) defn, col, colType);
         return;
       }
-      resolveUnknownColumn(parser, col, colType);
+      resolveUnknownColumn(col, colType);
     }
 
-    private void resolvePartitionColumn(Matcher m, ImplicitColumnResolver parser, ColumnHandle col) {
+    private void resolvePartitionColumn(Matcher m, ImplicitColumnResolver parser,
+        ColumnHandle col) {
 
       // The provided schema column must be of the correct type and mode.
       ColumnMetadata colSchema = col.column();
       if (colSchema.type() != MinorType.VARCHAR ||
           colSchema.mode() != DataMode.OPTIONAL) {
         throw UserException.validationError()
-            .message("Provided column %s is marked as a parition column, but is of the wrong type",
+            .message("Provided column `%s` is marked as a parition column, but is of the wrong type",
                 colSchema.columnString())
             .addContext("Expected type", MinorType.VARCHAR.name())
             .addContext("Expected cardinality", DataMode.OPTIONAL.name())
@@ -275,22 +255,21 @@ public class ImplicitColumnResolver {
 
       // Partition column
       int partitionIndex = Integer.parseInt(m.group(1));
-      columns.add(new PartitionColumnMarker(partitionIndex));
-      col.markImplicit();
+      markImplicit(col, new PartitionColumnMarker(partitionIndex));
 
       // Remember the partition for later wildcard expansion
       referencedPartitions.add(partitionIndex);
     }
 
     private void resolveImplicitColumn(ImplicitFileColumns defn,
-        ImplicitColumnResolver parser, ColumnHandle col) {
+        ColumnHandle col, String colType) {
 
       // The provided schema column must be of the correct type and mode.
       ColumnMetadata colSchema = col.column();
       if (colSchema.type() != MinorType.VARCHAR ||
           colSchema.mode() == DataMode.REPEATED) {
         throw UserException.validationError()
-            .message("Provided column %s is marked as an implicit column '%s', but is of the wrong type",
+            .message("Provided column `%s` is marked as implicit '%s', but is of the wrong type",
                 colSchema.columnString(), defn.propertyValue())
             .addContext("Expected type", MinorType.VARCHAR.name())
             .addContext("Expected cardinality", String.format("%s or %s",
@@ -298,12 +277,15 @@ public class ImplicitColumnResolver {
             .addContext(parser.errorContext)
             .build(logger);
       }
-      columns.add(new ImplicitColumnMarker(defn));
-      col.markImplicit();
+      markImplicit(col, new FileImplicitMarker(defn));
     }
 
-    private void resolveUnknownColumn(ImplicitColumnResolver parser,
-        ColumnHandle col, String colType) {
+    private void markImplicit(ColumnHandle col, ImplicitColumnMarker marker) {
+      columns.add(marker);
+      col.markImplicit(marker);
+    }
+
+    private void resolveUnknownColumn(ColumnHandle col, String colType) {
       throw UserException.validationError()
           .message("Provided column %s references an undefined implicit column type '%s'",
               col.column().columnString(), colType)
@@ -314,20 +296,20 @@ public class ImplicitColumnResolver {
           .build(logger);
     }
 
-    private void matchByName(ImplicitColumnResolver parser, ColumnHandle col) {
+    private void matchByName(ColumnHandle col) {
       Matcher m = parser.partitionPattern.matcher(col.name());
       if (m.matches()) {
-        buildPartitionColumn(m, parser, col);
+        buildPartitionColumn(m, col);
         return;
       }
 
-      ImplicitFileColumns defn = parser.colDefs.get(col.name());
+      ImplicitFileColumn defn = parser.colDefs.get(col.name());
       if (defn != null) {
-        buildImplicitColumn(defn, parser, col);
+        buildImplicitColumn(defn, col);
       }
     }
 
-    private void buildPartitionColumn(Matcher m, ImplicitColumnResolver parser, ColumnHandle col) {
+    private void buildPartitionColumn(Matcher m, ColumnHandle col) {
 
       // If the projected column is a map or array, then it shadows the
       // partition column. Example: dir0.x, dir0[2].
@@ -340,17 +322,21 @@ public class ImplicitColumnResolver {
 
       // Partition column
       int partitionIndex = Integer.parseInt(m.group(1));
-      ColumnMetadata resolved = MetadataUtils.newScalar(col.name(), PARTITION_COL_TYPE);
-      SchemaUtils.markAsPartition(resolved, partitionIndex);
-      columns.add(new PartitionColumnMarker(partitionIndex));
-      scanSchema.resolveImplicit(col, resolved);
+      resolve(col,
+          MetadataUtils.newScalar(col.name(), PARTITION_COL_TYPE),
+          new PartitionColumnMarker(partitionIndex));
 
       // Remember the partition for later wildcard expansion
       referencedPartitions.add(partitionIndex);
     }
 
-    private void buildImplicitColumn(ImplicitFileColumns defn,
-        ImplicitColumnResolver parser, ColumnHandle col) {
+    private void resolve(ColumnHandle col, ColumnMetadata resolved, ImplicitColumnMarker marker) {
+      columns.add(marker);
+      scanSchema.resolveImplicit(col, resolved, marker);
+    }
+
+    private void buildImplicitColumn(ImplicitFileColumn defn,
+        ColumnHandle col) {
 
       // If the projected column is a map or array, then it shadows the
       // metadata column. Example: filename.x, filename[2].
@@ -358,31 +344,58 @@ public class ImplicitColumnResolver {
       if (!projCol.isSimple()) {
         logger.warn("Projected column {} shadows implicit column {}",
             projCol.projectString(), col.name());
+      } else if (defn instanceof ImplicitInternalFileColumns) {
+        resolveInternalColumn(col, (ImplicitInternalFileColumns) defn);
       } else {
-        ColumnMetadata resolved = MetadataUtils.newScalar(col.name(), IMPLICIT_COL_TYPE);
-        SchemaUtils.markImplicit(resolved, defn.propertyValue());
-        columns.add(new ImplicitColumnMarker(defn));
-        scanSchema.resolveImplicit(col, resolved);
+        resolve(col,
+            MetadataUtils.newScalar(col.name(), IMPLICIT_COL_TYPE),
+            new FileImplicitMarker((ImplicitFileColumns) defn));
+      }
+    }
+
+    private void resolveInternalColumn(ColumnHandle col,
+        ImplicitInternalFileColumns defn) {
+
+      // Tests may not provide the DFS, real code must
+      if (defn == ImplicitInternalFileColumns.LAST_MODIFIED_TIME &&
+          parser.dfs == null) {
+        throw new IllegalStateException(
+            "Must provide a file system to use " + defn.name());
       }
+
+      // Check if this is an implied metadata scan
+      if (defn == ImplicitInternalFileColumns.PROJECT_METADATA) {
+        isMetadataScan = true;
+      }
+
+      // TODO: Internal columns are VARCHAR for historical reasons.
+      // Better to use a type that fits the column purposes.
+      resolve(col,
+          MetadataUtils.newScalar(col.name(),
+              defn.isOptional() ? OPTIONAL_INTERNAL_COL_TYPE : IMPLICIT_COL_TYPE),
+          new InternalColumnMarker(defn));
     }
   }
 
   public static final MajorType IMPLICIT_COL_TYPE = Types.required(MinorType.VARCHAR);
   public static final MajorType PARTITION_COL_TYPE =  Types.optional(MinorType.VARCHAR);
+  public static final MajorType OPTIONAL_INTERNAL_COL_TYPE =  Types.optional(MinorType.VARCHAR);
 
   private final int maxPartitionDepth;
   private final boolean useLegacyWildcardExpansion;
   private final String partitionDesignator;
   private final Pattern partitionPattern;
   private final Pattern partitionTypePattern;
-  private final Map<String, ImplicitFileColumns> colDefs = CaseInsensitiveMap.newHashMap();
-  private final Map<String, ImplicitFileColumns> typeDefs = CaseInsensitiveMap.newHashMap();
+  private final Map<String, ImplicitFileColumn> colDefs = CaseInsensitiveMap.newHashMap();
+  private final Map<String, ImplicitFileColumn> typeDefs = CaseInsensitiveMap.newHashMap();
   private final CustomErrorContext errorContext;
+  private final DrillFileSystem dfs;
 
   public ImplicitColumnResolver(ImplicitColumnOptions options, CustomErrorContext errorContext) {
     this.errorContext = errorContext;
     this.maxPartitionDepth = options.maxPartitionDepth;
     this.useLegacyWildcardExpansion = options.useLegacyWildcardExpansion;
+    this.dfs = options.dfs;
     this.partitionDesignator = options.optionSet.getString(ExecConstants.FILESYSTEM_PARTITION_COLUMN_LABEL);
     this.partitionPattern = Pattern.compile(partitionDesignator + "(\\d+)", Pattern.CASE_INSENSITIVE);
     if (partitionDesignator.equals(ColumnMetadata.IMPLICIT_PARTITION_PREFIX)) {
@@ -391,12 +404,22 @@ public class ImplicitColumnResolver {
       this.partitionTypePattern = Pattern.compile(ColumnMetadata.IMPLICIT_PARTITION_PREFIX + "(\\d+)",
           Pattern.CASE_INSENSITIVE);
     }
+
+    // File implicit columns: can be defined in the provided schema
     for (ImplicitFileColumns defn : ImplicitFileColumns.values()) {
       String colName = options.optionSet.getString(defn.optionName());
       if (!Strings.isNullOrEmpty(colName)) {
         this.colDefs.put(colName, defn);
       }
-      typeDefs.put(defn.propertyValue(), defn);
+      this.typeDefs.put(defn.propertyValue(), defn);
+    }
+
+    // Internal implicit cols: cannot be defined in the provided schema
+    for (ImplicitInternalFileColumns defn : ImplicitInternalFileColumns.values()) {
+      String colName = options.optionSet.getString(defn.optionName());
+      if (!Strings.isNullOrEmpty(colName)) {
+        this.colDefs.put(colName, defn);
+      }
     }
   }
 
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/scan/v3/file/ImplicitFileColumnsHandler.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/scan/v3/file/ImplicitFileColumnsHandler.java
index d6e7ef6..ee8e6b0 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/scan/v3/file/ImplicitFileColumnsHandler.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/scan/v3/file/ImplicitFileColumnsHandler.java
@@ -19,15 +19,15 @@ package org.apache.drill.exec.physical.impl.scan.v3.file;
 
 import java.util.List;
 
-import org.apache.drill.exec.ops.OperatorContext;
+import org.apache.drill.exec.physical.impl.scan.v3.file.ImplicitColumnResolver.ImplicitColumnOptions;
+import org.apache.drill.exec.physical.impl.scan.v3.file.ImplicitColumnResolver.ParseResult;
 import org.apache.drill.exec.physical.impl.scan.v3.lifecycle.StaticBatchBuilder;
 import org.apache.drill.exec.physical.impl.scan.v3.lifecycle.StaticBatchBuilder.RepeatedBatchBuilder;
-import org.apache.drill.exec.physical.impl.scan.v3.schema.ImplicitColumnResolver;
 import org.apache.drill.exec.physical.impl.scan.v3.schema.ScanSchemaTracker;
-import org.apache.drill.exec.physical.impl.scan.v3.schema.ImplicitColumnResolver.ColumnMarker;
-import org.apache.drill.exec.physical.impl.scan.v3.schema.ImplicitColumnResolver.ImplicitColumnOptions;
-import org.apache.drill.exec.physical.impl.scan.v3.schema.ImplicitColumnResolver.ParseResult;
 import org.apache.drill.exec.physical.resultSet.ResultVectorCache;
+import org.apache.drill.exec.server.options.OptionSet;
+import org.apache.drill.exec.store.dfs.DrillFileSystem;
+import org.apache.drill.exec.store.dfs.easy.FileWork;
 import org.apache.hadoop.fs.Path;
 
 /**
@@ -41,26 +41,37 @@ import org.apache.hadoop.fs.Path;
  */
 public class ImplicitFileColumnsHandler {
 
+  private final DrillFileSystem dfs;
   private final ImplicitColumnResolver parser;
   private final ResultVectorCache vectorCache;
   private final Path rootDir;
   private final ParseResult parseResult;
+  private final boolean isCompressible;
 
-  public ImplicitFileColumnsHandler(OperatorContext context, FileScanLifecycleBuilder scanOptions,
+  public ImplicitFileColumnsHandler(DrillFileSystem dfs, OptionSet options,
+      FileScanLifecycleBuilder scanOptions,
       ResultVectorCache vectorCache, ScanSchemaTracker schemaTracker) {
-    ImplicitColumnOptions options = new ImplicitColumnOptions()
-        .optionSet(context.getFragmentContext().getOptions())
+    ImplicitColumnOptions implicitOptions = new ImplicitColumnOptions()
+        .optionSet(options)
+        .dfs(dfs)
         .maxPartitionDepth(scanOptions.maxPartitionDepth())
         .useLegacyWildcardExpansion(scanOptions.useLegacyWildcardExpansion());
+    this.dfs = dfs;
     this.rootDir = scanOptions.rootDir();
-    this.parser = new ImplicitColumnResolver(options, scanOptions.errorContext());
+    this.parser = new ImplicitColumnResolver(implicitOptions, scanOptions.errorContext());
     this.vectorCache = vectorCache;
     this.parseResult = parser.parse(schemaTracker);
+    this.isCompressible = scanOptions.isCompressible();
   }
 
-  public StaticBatchBuilder forFile(Path filePath) {
-    FileDescrip fileInfo = new FileDescrip(filePath, rootDir);
-    List<ColumnMarker> columns = parseResult.columns();
+  public FileDescrip makeDescrip(FileWork fileWork) {
+    FileDescrip descrip = new FileDescrip(dfs, fileWork, rootDir);
+    descrip.setCompressible(isCompressible);
+    return descrip;
+  }
+
+  public StaticBatchBuilder forFile(FileDescrip fileInfo) {
+    List<ImplicitColumnMarker> columns = parseResult.columns();
     if (columns.isEmpty()) {
       return null;
     }
@@ -70,4 +81,6 @@ public class ImplicitFileColumnsHandler {
     }
     return new RepeatedBatchBuilder(vectorCache, parseResult.schema(), values);
   }
+
+  public boolean isMetadataScan() { return parseResult.isMetadataScan(); }
 }
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/scan/v3/lifecycle/ReaderLifecycle.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/scan/v3/lifecycle/ReaderLifecycle.java
index e758fc3..f322f51 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/scan/v3/lifecycle/ReaderLifecycle.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/scan/v3/lifecycle/ReaderLifecycle.java
@@ -34,70 +34,68 @@ import org.apache.drill.exec.physical.resultSet.impl.ResultSetLoaderImpl;
 import org.apache.drill.exec.physical.resultSet.impl.ResultSetOptionBuilder;
 import org.apache.drill.exec.record.VectorContainer;
 import org.apache.drill.exec.record.metadata.TupleMetadata;
+import org.apache.drill.shaded.guava.com.google.common.base.Preconditions;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 /**
- * Manages the schema and batch construction for a managed reader.
- * Allows the reader itself to be as simple as possible. This class
- * implements the basic {@link RowBatchReader} protocol based on
- * three methods, and converts it to the two-method protocol of
- * the managed reader. The {@code open()} call of the
+ * Manages the schema and batch construction for a managed reader. Allows the
+ * reader itself to be as simple as possible. This class implements the basic
+ * {@link RowBatchReader} protocol based on three methods, and converts it to
+ * the two-method protocol of the managed reader. The {@code open()} call of the
  * {@code RowBatchReader} is combined with the constructor of the
- * {@link ManagedReader}, enforcing the rule that the managed reader
- * is created just-in-time when it is to be used, which avoids
- * accidentally holding resources for the life of the scan.
+ * {@link ManagedReader}, enforcing the rule that the managed reader is created
+ * just-in-time when it is to be used, which avoids accidentally holding
+ * resources for the life of the scan. Also allows most of the reader's fields
+ * to be {@code final}.
  * <p>
- * Coordinates the components that wrap a reader to create the final
- * output batch:
+ * Coordinates the components that wrap a reader to create the final output
+ * batch:
  * <ul>
- * <li>The actual reader which load (possibly a subset of) the
- * columns requested from the input source.</li>
- * <li>Implicit columns manager instance which populates implicit
- * file columns, partition columns, and Drill's internal metadata
- * columns.</li>
- * <li>The missing columns handler which "makes up" values for projected
- * columns not read by the reader.</li>
- * <li>Batch assembler, which combines the three sources of vectors
- * to create the output batch with the schema specified by the
- * schema tracker.</li>
+ * <li>The actual reader which loads (possibly a subset of) the columns requested
+ * from the input source.</li>
+ * <li>Implicit columns manager instance which populates implicit file columns,
+ * partition columns, and Drill's internal implicit columns.</li>
+ * <li>The missing columns handler which "makes up" values for projected columns
+ * not read by the reader.</li>
+ * <li>Batch assembler, which combines the three sources of vectors to create
+ * the output batch with the schema specified by the schema tracker.</li>
  * </ul>
  * <p>
  * This class coordinates the reader-visible aspects of the scan:
  * <ul>
- * <li>The {@link SchemaNegotiator} (or subclass) which provides
- * schema-related input to the reader and which creates the reader's
- * {@link ResultSetLoader}, among other tasks. The schema negotiator
- * is specific to each kind of scan and is thus created via the
- * {@link ScanLifecycleBuilder}.</li>
- * <li>The reader, which is designed to be as simple as possible,
- * with all generic overhead tasks handled by this "shim" between
- * the scan operator and the actual reader implementation.</li>
+ * <li>The {@link SchemaNegotiator} (or subclass) which provides schema-related
+ * input to the reader and which creates the reader's {@link ResultSetLoader},
+ * among other tasks. The schema negotiator is specific to each kind of scan and
+ * is thus created via the {@link ScanLifecycleBuilder}.</li>
+ * <li>The reader, which is designed to be as simple as possible, with all
+ * generic overhead tasks handled by this "shim" between the scan operator and
+ * the actual reader implementation.</li>
  * </ul>
  * <p>
- * The reader is schema-driven. See {@link ScanSchemaTracker} for
- * an overview.
+ * The reader is schema-driven. See {@link ScanSchemaTracker} for an overview.
  * <ul>
- * <li>The reader is given a <i>reader input schema</i>, via the
- * schema negotiator, which specifies the desired output schema.
- * The schema can be fully dynamic (a wildcard), fully defined (a
- * prior reader already chose column types), or a hybrid.</li>
- * <li>The reader can load a subset of columns. Those that are
- * left out become "missing columns" to be filled in by this
- * class.</li>
- * <li>The <i>reader output schema</i> along with implicit and missing
- * columns, together define the scan's output schema.</li>
+ * <li>The reader is given a <i>reader input schema</i>, via the schema
+ * negotiator, which specifies the desired output schema. The schema can be
+ * fully dynamic (a wildcard), fully defined (a prior reader already chose
+ * column types), or a hybrid.</li>
+ * <li>The reader can load a subset of columns. Those that are left out become
+ * "missing columns" to be filled in by this class.</li>
+ * <li>The <i>reader output schema</i> along with implicit and missing columns,
+ * together define the scan's output schema.</li>
  * </ul>
  * <p>
- * The framework handles the projection task so the
- * reader does not have to worry about it. Reading an unwanted column
- * is low cost: the result set loader will have provided a "dummy" column
- * writer that simply discards the value. This is just as fast as having the
- * reader use if-statements or a table to determine which columns to save.
+ * The framework handles the projection task so the reader does not have to
+ * worry about it. Reading an unwanted column is low cost: the result set loader
+ * will have provided a "dummy" column writer that simply discards the value.
+ * This is just as fast as having the reader use if-statements or a table to
+ * determine which columns to save.
  */
 public class ReaderLifecycle implements RowBatchReader {
   private static final Logger logger = LoggerFactory.getLogger(ReaderLifecycle.class);
 
+  private enum State { START, DATA, FINAL, EOF }
+
   private final ScanLifecycle scanLifecycle;
   protected final TupleMetadata readerInputSchema;
   private ManagedReader reader;
@@ -107,12 +105,7 @@ public class ReaderLifecycle implements RowBatchReader {
   private StaticBatchBuilder implicitColumnsLoader;
   private StaticBatchBuilder missingColumnsHandler;
   private OutputBatchBuilder outputBuilder;
-
-  /**
-   * True once the reader reports EOF. This shim may keep going for another
-   * batch to handle any look-ahead row on the last batch.
-   */
-  private boolean eof;
+  private State state = State.START;
 
   public ReaderLifecycle(ScanLifecycle scanLifecycle) {
     this.scanLifecycle = scanLifecycle;
@@ -137,6 +130,8 @@ public class ReaderLifecycle implements RowBatchReader {
     return reader.getClass().getSimpleName();
   }
 
+  public ResultSetLoader tableLoader() { return tableLoader; }
+
   @Override
   public boolean open() {
     try {
@@ -168,6 +163,7 @@ public class ReaderLifecycle implements RowBatchReader {
   }
 
   public ResultSetLoader buildLoader() {
+    Preconditions.checkState(state == State.START);
     ResultSetOptionBuilder options = new ResultSetOptionBuilder()
         .rowCountLimit(Math.min(schemaNegotiator.batchSize, scanOptions().scanBatchRecordLimit()))
         .vectorCache(scanLifecycle.vectorCache())
@@ -176,12 +172,12 @@ public class ReaderLifecycle implements RowBatchReader {
         .projectionFilter(schemaTracker().projectionFilter(errorContext()))
         .readerSchema(schemaNegotiator.readerSchema);
 
-    // Resolve the scan scame if possible.
+    // Resolve the scan schema if possible.
     applyEarlySchema();
 
     // Create the table loader
     tableLoader = new ResultSetLoaderImpl(scanLifecycle.allocator(), options.build());
-    implicitColumnsLoader = schemaNegotiator.implicitColumnsLoader();
+    state = State.DATA;
     return tableLoader;
   }
 
@@ -205,12 +201,12 @@ public class ReaderLifecycle implements RowBatchReader {
 
   @Override
   public boolean defineSchema() {
-    if (!schemaNegotiator.isSchemaComplete()) {
-      return false;
+    boolean hasSchema = schemaNegotiator.isSchemaComplete() && schemaTracker().isResolved();
+    if (hasSchema) {
+      tableLoader.startBatch();
+      endBatch();
     }
-    tableLoader.startBatch();
-    endBatch();
-    return true;
+    return hasSchema;
   }
 
   @Override
@@ -218,7 +214,7 @@ public class ReaderLifecycle implements RowBatchReader {
 
     // The reader may report EOF, but the result set loader might
     // have a lookahead row.
-    if (isEof()) {
+    if (state == State.EOF) {
       return false;
     }
 
@@ -230,9 +226,11 @@ public class ReaderLifecycle implements RowBatchReader {
     // a new batch just to learn about EOF. Don't read if the reader
     // already reported EOF. In that case, we're just processing any last
     // lookahead row in the result set loader.
-    if (!eof) {
+    if (state == State.DATA) {
       try {
-        eof = !reader.next();
+        if (!reader.next()) {
+          state = State.FINAL;
+        }
       } catch (UserException e) {
         throw e;
       } catch (Exception e) {
@@ -252,11 +250,7 @@ public class ReaderLifecycle implements RowBatchReader {
     // Return EOF (false) only when the reader reports EOF
     // and the result set loader has drained its rows from either
     // this batch or lookahead rows.
-    return !isEof();
-  }
-
-  public boolean isEof() {
-    return eof && !tableLoader.hasRows();
+    return state != State.EOF;
   }
 
   /**
@@ -265,24 +259,21 @@ public class ReaderLifecycle implements RowBatchReader {
    * table row count. Then, merge the sources.
    */
   private void endBatch() {
-    VectorContainer readerOutput = tableLoader.harvest();
 
-    // Corner case: Reader produced no rows and defined no schema.
-    // We assume this is a null file and skip it as if we got an
-    // EOF from the constructor. There may be some actual case in
-    // which a file is known to have no records and no columns,
-    // which is an empty file. At present, there is no way to differentiate
-    // an intentional empty file from a null file.
-    if (readerOutput.getRecordCount() == 0 && tableLoader.schemaVersion() == 0) {
-      readerOutput.clear();
-      return;
-    }
+    // Let the schema negotiator finish up the batch. Needed for metadata
+    // scans on files.
+    // TODO: Modify the metadata system to handle non-file scans, then
+    // generalize the implicit columns parser, identify a new field to
+    // replace/augment fqn, and handle empty scans here.
+    schemaNegotiator.onEndBatch();
 
-    // If not the first batch, and there are no rows, discard the batch
-    // as it adds no new information.
-    if (readerOutput.getRecordCount() == 0 && tableLoader.batchCount() > 1) {
+    // Get the output batch
+    VectorContainer readerOutput = tableLoader.harvest();
+
+    if (readerOutput.getRecordCount() == 0 && !returnEmptyBatch(readerOutput)) {
       readerOutput.clear();
       outputBuilder = null;
+      state = State.EOF;
       return;
     }
 
@@ -292,6 +283,40 @@ public class ReaderLifecycle implements RowBatchReader {
       reviseOutputProjection(tableLoader.outputSchema());
     }
     buildOutputBatch(readerOutput);
+    scanLifecycle.tallyBatch();
+  }
+
+  /**
+   * The reader returned no data. Determine if this batch should be
+   * returned (return {@code true}), or if the empty batch should
+   * be returned to convey schema information. (return {@code false}).
+   */
+  private boolean returnEmptyBatch(VectorContainer readerOutput) {
+
+    // If the batch is not the first, then it conveys no new info.
+    if (scanLifecycle.batchCount() > 0) {
+      return false;
+    }
+
+    // Corner case: Reader produced no rows and defined no schema.
+    // There are three sub-cases. In the first, the reader is "late-schema",
+    // it discovers schema as it reads, and did not read anything. In
+    // this case, we assume this is a null file and skip it as if we got an
+    // EOF from the constructor.
+    //
+    // The second sub-case is as above, but a schema was provided. In this
+    // case, we can produce an empty result set to convey that schema.
+    //
+    // A third, possible, but very obscure case occurs in which a file
+    // is known to have no records and no columns. At present, there is
+    // no way to differentiate an intentional empty file from a null file.
+    if (tableLoader.schemaVersion() == 0) {
+      return schemaTracker().isResolved();
+    }
+
+    // Otherwise, we did define a schema on the first batch of the scan
+    // so we need to return it.
+    return true;
   }
 
   private void reviseOutputProjection(TupleMetadata readerOutputSchema) {
@@ -314,6 +339,12 @@ public class ReaderLifecycle implements RowBatchReader {
 
   private void buildOutputBatch(VectorContainer readerContainer) {
 
+    // Create the implicit columns loader loader after the first
+    // batch so we can report if the file is empty.
+    if (tableLoader.batchCount() == 1) {
+      implicitColumnsLoader = schemaNegotiator.implicitColumnsLoader();
+    }
+
     // Get the batch results in a container.
     int rowCount = readerContainer.getRecordCount();
     if (implicitColumnsLoader != null) {
@@ -353,7 +384,7 @@ public class ReaderLifecycle implements RowBatchReader {
 
   @Override
   public int schemaVersion() {
-    return tableLoader.schemaVersion();
+    return schemaTracker().schemaVersion();
   }
 
   @Override
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/scan/v3/lifecycle/ScanLifecycle.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/scan/v3/lifecycle/ScanLifecycle.java
index 039810f..3261a05 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/scan/v3/lifecycle/ScanLifecycle.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/scan/v3/lifecycle/ScanLifecycle.java
@@ -135,6 +135,7 @@ public class ScanLifecycle {
   private final ScanLifecycleBuilder options;
   private final ScanSchemaTracker schemaTracker;
   private final ReaderFactory<?> readerFactory;
+  private int batchCount;
 
   /**
    * Cache used to preserve the same vectors from one output batch to the
@@ -146,17 +147,20 @@ public class ScanLifecycle {
    */
   private final ResultVectorCacheImpl vectorCache;
 
-  public ScanLifecycle(OperatorContext context, ScanLifecycleBuilder options) {
+  public ScanLifecycle(OperatorContext context, ScanLifecycleBuilder builder) {
     this.context = context;
-    this.options = options;
+    this.options = builder;
     this.schemaTracker = new ScanSchemaConfigBuilder()
-        .projection(options.projection())
-        .definedSchema(options.definedSchema())
-        .providedSchema(options.providedSchema())
-        .allowSchemaChange(options.allowSchemaChange())
+        .projection(builder.projection())
+        .definedSchema(builder.definedSchema())
+        .providedSchema(builder.providedSchema())
+        .allowSchemaChange(builder.allowSchemaChange())
         .build();
+    if (builder.schemaValidator() != null) {
+      builder.schemaValidator().validate(schemaTracker);
+    }
     this.vectorCache = new ResultVectorCacheImpl(allocator(), false);
-    this.readerFactory = options.readerFactory();
+    this.readerFactory = builder.readerFactory();
   }
 
   public OperatorContext context() { return context; }
@@ -167,6 +171,8 @@ public class ScanLifecycle {
   public boolean hasOutputSchema() { return schemaTracker.isResolved(); }
   public CustomErrorContext errorContext() { return options.errorContext(); }
   public BufferAllocator allocator() { return context.getAllocator(); }
+  public void tallyBatch() { batchCount++; }
+  public int batchCount() { return batchCount; }
 
   public RowBatchReader nextReader() {
     if (readerFactory.hasNext()) {
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/scan/v3/lifecycle/SchemaNegotiatorImpl.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/scan/v3/lifecycle/SchemaNegotiatorImpl.java
index 20b2a02..7b94cc4 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/scan/v3/lifecycle/SchemaNegotiatorImpl.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/scan/v3/lifecycle/SchemaNegotiatorImpl.java
@@ -23,6 +23,7 @@ import org.apache.drill.exec.physical.impl.scan.v3.ManagedReader;
 import org.apache.drill.exec.physical.impl.scan.v3.ReaderFactory;
 import org.apache.drill.exec.physical.impl.scan.v3.SchemaNegotiator;
 import org.apache.drill.exec.physical.impl.scan.v3.ManagedReader.EarlyEofException;
+import org.apache.drill.exec.physical.impl.scan.v3.schema.ProjectedColumn;
 import org.apache.drill.exec.physical.impl.scan.v3.schema.ScanSchemaTracker.ProjectionType;
 import org.apache.drill.exec.physical.resultSet.ResultSetLoader;
 import org.apache.drill.exec.record.metadata.TupleMetadata;
@@ -73,6 +74,11 @@ public class SchemaNegotiatorImpl implements SchemaNegotiator {
   }
 
   @Override
+  public ProjectedColumn projectionFor(String colName) {
+    return readerLifecycle.scanLifecycle().schemaTracker().columnProjection(colName);
+  }
+
+  @Override
   public TupleMetadata providedSchema() {
     return readerLifecycle.scanOptions().providedSchema();
   }
@@ -92,6 +98,7 @@ public class SchemaNegotiatorImpl implements SchemaNegotiator {
     return baseErrorContext;
   }
 
+  @Override
   public CustomErrorContext errorContext() {
     return readerErrorContext == null ? baseErrorContext : readerErrorContext;
   }
@@ -151,4 +158,6 @@ public class SchemaNegotiatorImpl implements SchemaNegotiator {
   public ManagedReader newReader(ReaderFactory<?> readerFactory) throws EarlyEofException {
     return ((ReaderFactory<SchemaNegotiator>) readerFactory).next(this);
   }
+
+  protected void onEndBatch() { }
 }
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/scan/v3/lifecycle/StaticBatchBuilder.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/scan/v3/lifecycle/StaticBatchBuilder.java
index 1944916..064f4b6 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/scan/v3/lifecycle/StaticBatchBuilder.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/scan/v3/lifecycle/StaticBatchBuilder.java
@@ -74,7 +74,9 @@ public abstract class StaticBatchBuilder {
       for (int i = 0; i < rowCount; i++) {
         writer.start();
         for (int j = 0; j < n; j++) {
-          writers[j].setValue(values[j]);
+          if (values[j] != null) {
+            writers[j].setValue(values[j]);
+          }
         }
         writer.save();
       }
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/scan/v3/schema/AbstractSchemaTracker.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/scan/v3/schema/AbstractSchemaTracker.java
index 20de4f2..7646477 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/scan/v3/schema/AbstractSchemaTracker.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/scan/v3/schema/AbstractSchemaTracker.java
@@ -41,6 +41,13 @@ public abstract class AbstractSchemaTracker implements ScanSchemaTracker {
     this.errorContext = errorContext;
   }
 
+  /**
+   * Validate a projection list against a defined-schema tuple. Recursively walks
+   * the tree of maps to validate all nested tuples.
+   *
+   * @param projection the parsed projection list
+   * @param schema the defined schema to validate against
+   */
   protected static void validateProjection(TupleMetadata projection, TupleMetadata schema) {
     if (projection == null || SchemaUtils.isProjectAll(projection)) {
       return;
@@ -77,8 +84,25 @@ public abstract class AbstractSchemaTracker implements ScanSchemaTracker {
   @Override
   public int schemaVersion() { return schema.version(); }
 
+  /**
+   * Determine if the schema is resolved. It is resolved if the
+   * schema itself is resolved. Since an empty schema is resolved, for the
+   * {@code SELECT *} case, we require at least one column, which means
+   * that something (provided schema, early reader schema) has provided
+   * us with a schema. Once resolved, a schema can never become
+   * unresolved: readers are not allowed to add dynamic columns.
+   */
   protected void checkResolved() {
-    isResolved = schema.isResolved();
+    if (isResolved) {
+      return;
+    }
+    switch (projectionType()) {
+    case ALL:
+      isResolved = !schema.isEmpty() && schema.isResolved();
+      break;
+    default:
+      isResolved = schema.isResolved();
+    }
   }
 
   @Override
@@ -102,9 +126,8 @@ public abstract class AbstractSchemaTracker implements ScanSchemaTracker {
   private TupleMetadata implicitColumns() {
     TupleMetadata implicitCols = new TupleSchema();
     for (ColumnHandle handle : schema.columns()) {
-      ColumnMetadata col = handle.column();
-      if (SchemaUtils.isImplicit(col)) {
-        implicitCols.addColumn(col);
+      if (handle.isImplicit()) {
+        implicitCols.addColumn(handle.column());
       }
     }
     return implicitCols;
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/scan/v3/schema/MutableTupleMetadata.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/scan/v3/schema/MutableTupleMetadata.java
deleted file mode 100644
index d5f46d2..0000000
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/scan/v3/schema/MutableTupleMetadata.java
+++ /dev/null
@@ -1,220 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License.  You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.drill.exec.physical.impl.scan.v3.schema;
-
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Map;
-
-import org.apache.drill.common.map.CaseInsensitiveMap;
-import org.apache.drill.exec.physical.impl.scan.v3.schema.ScanSchemaTracker.ProjectionType;
-import org.apache.drill.exec.record.metadata.ColumnMetadata;
-import org.apache.drill.exec.record.metadata.TupleMetadata;
-import org.apache.drill.exec.record.metadata.TupleSchema;
-import org.apache.drill.shaded.guava.com.google.common.base.Preconditions;
-
-/**
- * A mutable form of a tuple schema. Allows insertions (at the wildcard position),
- * and replacing columns (as the schema becomes resolved). Tracks implicit columns
- * (those not filled in by the reader).
- * <p>
- * Does not implement the {@code TupleMetadata} interface because that interface
- * has far more functionality than is needed here, and assumes that column order
- * remains fixed (and hence columns can be addressed by position) which is not
- * true for this class.
- * <p>
- * This class represents the top-level tuple (the row.) Maps are also dynamic,
- * but provide a subset of resolution options:
- * map fields cannot be implicit. They can, however, be defined,
- * provided, discovered or missing. Map columns can start unresolved
- * if the map comes from projection. A map itself can be resolved,
- * but its members may be unresolved. New map members may only be added at the
- * end (there is no equivalent of a wildcard position.)
- */
-public class MutableTupleMetadata {
-
-  /**
-   * Holder for a column to allow inserting and replacing columns within
-   * the top-level project list. Changes to the column within the holder
-   * must go through the tuple itself so we can track schema versions.
-   * <p>
-   * Tracks the resolution status of each individual column as
-   * described for {@link ScanSchemaTracker}. Models a column throughout the
-   * projection lifecycle. Columns evolve from unresolved to resolved at
-   * different times. Columns are either implicit (defined by the framework)
-   * or normal (defined by the reader). Columns can be defined by the
-   * planner (via a defined schema), partially defined (via a provided
-   * schema), or discovered by the reader. Regardless of the path
-   * to definition, by the time the first batch is delivered downstream,
-   * each column has an output schema which describes the data.
-   */
-  public static class ColumnHandle {
-    private ColumnMetadata col;
-    private boolean isImplicit;
-
-    public ColumnHandle(ColumnMetadata col) {
-      this.col = col;
-      this.isImplicit = SchemaUtils.isImplicit(col);
-    }
-
-    public String name() {
-      return col.name();
-    }
-
-    private void replace(ColumnMetadata col) {
-      this.col = col;
-    }
-
-    private void resolve(ColumnMetadata col) {
-      SchemaUtils.mergeColProperties(this.col, col);
-      this.col = col;
-    }
-
-    private void resolveImplicit(ColumnMetadata col) {
-      SchemaUtils.mergeColProperties(this.col, col);
-      this.col = col;
-      markImplicit();
-    }
-
-    public void markImplicit() {
-      Preconditions.checkState(SchemaUtils.isImplicit(col));
-      isImplicit = true;
-    }
-
-    public ColumnMetadata column() { return col; }
-    public boolean isImplicit() { return isImplicit; }
-
-    @Override
-    public String toString() {
-      return col.toString();
-    }
-  }
-
-  protected final List<MutableTupleMetadata.ColumnHandle> columns = new ArrayList<>();
-  protected final Map<String, MutableTupleMetadata.ColumnHandle> nameIndex =
-      CaseInsensitiveMap.newHashMap();
-  private ProjectionType projType;
-  private int insertPoint = -1;
-  private int version;
-
-  public void setProjectionType(ScanSchemaTracker.ProjectionType type) {
-    this.projType = type;
-  }
-
-  public void setInsertPoint(int insertPoint) {
-    Preconditions.checkArgument(insertPoint == -1 ||
-        insertPoint >= 0 && insertPoint <= size());
-    this.insertPoint = insertPoint;
-  }
-
-  public ScanSchemaTracker.ProjectionType projectionType() { return projType; }
-  public int size() { return columns.size(); }
-  public int version() { return version; }
-
-  /**
-   * Provide the list of partially-resolved columns. Primarily for
-   * the implicit column parser.
-   */
-  public List<MutableTupleMetadata.ColumnHandle> columns() { return columns; }
-
-  public MutableTupleMetadata.ColumnHandle find(String colName) {
-    return nameIndex.get(colName);
-  }
-
-  public void copyFrom(TupleMetadata from) {
-    if (from.isEmpty()) {
-      return;
-    }
-    for (ColumnMetadata projCol : from) {
-      add(projCol.copy());
-    }
-    version++;
-  }
-
-  public void add(ColumnMetadata col) {
-    MutableTupleMetadata.ColumnHandle holder = new ColumnHandle(col);
-    columns.add(holder);
-    addIndex(holder);
-    version++;
-  }
-
-  public void addIndex(MutableTupleMetadata.ColumnHandle holder) {
-    if (nameIndex.put(holder.column().name(), holder) != null) {
-      throw new IllegalArgumentException("Duplicate scan projection column: " + holder.name());
-    }
-  }
-
-  public void insert(int posn, ColumnMetadata col) {
-    MutableTupleMetadata.ColumnHandle holder = new ColumnHandle(col);
-    columns.add(posn, holder);
-    addIndex(holder);
-    version++;
-  }
-
-  public void insert(ColumnMetadata col) {
-    insert(insertPoint++, col);
-  }
-
-  public boolean isResolved() {
-    for (MutableTupleMetadata.ColumnHandle handle : columns) {
-      if (!isColumnResolved(handle.column())) {
-        return false;
-      }
-    }
-    return true;
-  }
-
-  private boolean isColumnResolved(ColumnMetadata col) {
-    return !col.isDynamic() && (!col.isMap() || isMapResolved(col.tupleSchema()));
-  }
-
-  private boolean isMapResolved(TupleMetadata mapSchema) {
-    for (ColumnMetadata col : mapSchema) {
-      if (col.isDynamic()) {
-        return false;
-      }
-      if (col.isMap() && !isMapResolved(col.tupleSchema())) {
-        return false;
-      }
-    }
-    return true;
-  }
-
-  public TupleMetadata toSchema() {
-    TupleMetadata schema = new TupleSchema();
-    for (MutableTupleMetadata.ColumnHandle col : columns) {
-      schema.addColumn(col.column());
-    }
-    return schema;
-  }
-
-  public void resolveImplicit(ColumnHandle col, ColumnMetadata resolved) {
-    col.resolveImplicit(resolved);
-    version++;
-  }
-
-  public void replace(ColumnHandle col, ColumnMetadata resolved) {
-    col.replace(resolved);
-    version++;
-  }
-
-  public void resolve(ColumnHandle col, ColumnMetadata resolved) {
-    col.resolve(resolved);
-    version++;
-  }
-}
\ No newline at end of file
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/scan/v3/schema/MutableTupleSchema.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/scan/v3/schema/MutableTupleSchema.java
index ba3483f..89883b2 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/scan/v3/schema/MutableTupleSchema.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/scan/v3/schema/MutableTupleSchema.java
@@ -20,8 +20,10 @@ package org.apache.drill.exec.physical.impl.scan.v3.schema;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
 
 import org.apache.drill.common.map.CaseInsensitiveMap;
+import org.apache.drill.exec.physical.impl.scan.v3.file.ImplicitColumnMarker;
 import org.apache.drill.exec.physical.impl.scan.v3.schema.ScanSchemaTracker.ProjectionType;
 import org.apache.drill.exec.record.metadata.ColumnMetadata;
 import org.apache.drill.exec.record.metadata.TupleMetadata;
@@ -65,11 +67,10 @@ public class MutableTupleSchema {
    */
   public static class ColumnHandle {
     private ColumnMetadata col;
-    private boolean isImplicit;
+    private ImplicitColumnMarker marker;
 
     public ColumnHandle(ColumnMetadata col) {
       this.col = col;
-      this.isImplicit = SchemaUtils.isImplicit(col);
     }
 
     public String name() {
@@ -85,19 +86,18 @@ public class MutableTupleSchema {
       this.col = col;
     }
 
-    private void resolveImplicit(ColumnMetadata col) {
+    private void resolveImplicit(ColumnMetadata col, ImplicitColumnMarker marker) {
       SchemaUtils.mergeColProperties(this.col, col);
       this.col = col;
-      markImplicit();
+      markImplicit(marker);
     }
 
-    public void markImplicit() {
-      Preconditions.checkState(SchemaUtils.isImplicit(col));
-      isImplicit = true;
+    public void markImplicit(ImplicitColumnMarker marker) {
+      this.marker = marker;
     }
 
     public ColumnMetadata column() { return col; }
-    public boolean isImplicit() { return isImplicit; }
+    public boolean isImplicit() { return marker != null; }
 
     @Override
     public String toString() {
@@ -105,8 +105,8 @@ public class MutableTupleSchema {
     }
   }
 
-  protected final List<MutableTupleSchema.ColumnHandle> columns = new ArrayList<>();
-  protected final Map<String, MutableTupleSchema.ColumnHandle> nameIndex =
+  protected final List<ColumnHandle> columns = new ArrayList<>();
+  protected final Map<String, ColumnHandle> nameIndex =
       CaseInsensitiveMap.newHashMap();
   private ProjectionType projType;
   private int insertPoint = -1;
@@ -114,6 +114,10 @@ public class MutableTupleSchema {
 
   public void setProjectionType(ScanSchemaTracker.ProjectionType type) {
     this.projType = type;
+
+    // For project none, an empty schema is valid, so
+    // force a bump in schema version.
+    version++;
   }
 
   public void setInsertPoint(int insertPoint) {
@@ -130,9 +134,9 @@ public class MutableTupleSchema {
    * Provide the list of partially-resolved columns. Primarily for
    * the implicit column parser.
    */
-  public List<MutableTupleSchema.ColumnHandle> columns() { return columns; }
+  public List<ColumnHandle> columns() { return columns; }
 
-  public MutableTupleSchema.ColumnHandle find(String colName) {
+  public ColumnHandle find(String colName) {
     return nameIndex.get(colName);
   }
 
@@ -147,31 +151,52 @@ public class MutableTupleSchema {
   }
 
   public void add(ColumnMetadata col) {
-    MutableTupleSchema.ColumnHandle holder = new ColumnHandle(col);
+    ColumnHandle holder = new ColumnHandle(col);
     columns.add(holder);
     addIndex(holder);
     version++;
   }
 
-  public void addIndex(MutableTupleSchema.ColumnHandle holder) {
+  public void addIndex(ColumnHandle holder) {
     if (nameIndex.put(holder.column().name(), holder) != null) {
       throw new IllegalArgumentException("Duplicate scan projection column: " + holder.name());
     }
   }
 
-  public void insert(int posn, ColumnMetadata col) {
-    MutableTupleSchema.ColumnHandle holder = new ColumnHandle(col);
+  public ColumnHandle insert(int posn, ColumnMetadata col) {
+    ColumnHandle holder = new ColumnHandle(col);
     columns.add(posn, holder);
     addIndex(holder);
     version++;
+    return holder;
+  }
+
+  public ColumnHandle insert(ColumnMetadata col) {
+    return insert(insertPoint++, col);
   }
 
-  public void insert(ColumnMetadata col) {
-    insert(insertPoint++, col);
+  /**
+   * Move a column from its current position (which must be past the insert point)
+   * to the insert point. An index entry already exists. Special operation done
+   * when matching a provided schema to a projection list that includes a wildcard
+   * and explicitly projected columns. Works around unfortunate behavior in the
+   * planner.
+   */
+  public void moveIfExplicit(String colName) {
+    ColumnHandle holder = find(colName);
+    Objects.requireNonNull(holder);
+    int posn = columns.indexOf(holder);
+    if (posn == insertPoint) {
+      insertPoint++;
+    } else if (posn > insertPoint) {
+      columns.remove(posn);
+      columns.add(insertPoint++, holder);
+      version++;
+    }
   }
 
   public boolean isResolved() {
-    for (MutableTupleSchema.ColumnHandle handle : columns) {
+    for (ColumnHandle handle : columns) {
       if (!isColumnResolved(handle.column())) {
         return false;
       }
@@ -197,14 +222,14 @@ public class MutableTupleSchema {
 
   public TupleMetadata toSchema() {
     TupleMetadata schema = new TupleSchema();
-    for (MutableTupleSchema.ColumnHandle col : columns) {
+    for (ColumnHandle col : columns) {
       schema.addColumn(col.column());
     }
     return schema;
   }
 
-  public void resolveImplicit(ColumnHandle col, ColumnMetadata resolved) {
-    col.resolveImplicit(resolved);
+  public void resolveImplicit(ColumnHandle col, ColumnMetadata resolved, ImplicitColumnMarker marker) {
+    col.resolveImplicit(resolved, marker);
     version++;
   }
 
@@ -217,4 +242,6 @@ public class MutableTupleSchema {
     col.resolve(resolved);
     version++;
   }
+
+  public boolean isEmpty() { return columns.isEmpty(); }
 }
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/scan/v3/schema/ProjectionSchemaTracker.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/scan/v3/schema/ProjectionSchemaTracker.java
index 3e1373e..a11b675 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/scan/v3/schema/ProjectionSchemaTracker.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/scan/v3/schema/ProjectionSchemaTracker.java
@@ -18,13 +18,13 @@
 package org.apache.drill.exec.physical.impl.scan.v3.schema;
 
 import org.apache.drill.common.exceptions.CustomErrorContext;
+import org.apache.drill.exec.physical.impl.scan.v3.file.ImplicitColumnMarker;
 import org.apache.drill.exec.physical.impl.scan.v3.schema.DynamicSchemaFilter.RowSchemaFilter;
 import org.apache.drill.exec.physical.impl.scan.v3.schema.ScanProjectionParser.ProjectionParseResult;
 import org.apache.drill.exec.physical.impl.scan.v3.schema.ScanSchemaResolver.SchemaType;
 import org.apache.drill.exec.physical.resultSet.impl.ProjectionFilter;
 import org.apache.drill.exec.record.metadata.ColumnMetadata;
 import org.apache.drill.exec.record.metadata.TupleMetadata;
-import org.apache.drill.shaded.guava.com.google.common.base.Preconditions;
 
 /**
  * Schema tracker for the "normal" case in which schema starts from a simple
@@ -37,6 +37,7 @@ public class ProjectionSchemaTracker extends AbstractSchemaTracker {
   private final TupleMetadata projection;
   private final boolean allowSchemaChange;
   private int implicitInsertPoint;
+  private int readerSchemaCount;
   private boolean allowMapAdditions = true;
 
   public ProjectionSchemaTracker(TupleMetadata definedSchema,
@@ -47,16 +48,16 @@ public class ProjectionSchemaTracker extends AbstractSchemaTracker {
     this.allowSchemaChange = false;
     schema.copyFrom(definedSchema);
     validateProjection(parseResult.dynamicSchema, definedSchema);
-    checkResolved();
 
-    ScanSchemaTracker.ProjectionType projType;
+    ProjectionType projType;
     if (schema.size() == 0) {
-      projType = ScanSchemaTracker.ProjectionType.NONE;
+      projType = ProjectionType.NONE;
     } else {
-      projType = ScanSchemaTracker.ProjectionType.SOME;
+      projType = ProjectionType.SOME;
     }
     schema.setProjectionType(projType);
     this.implicitInsertPoint = -1;
+    checkResolved();
   }
 
   public ProjectionSchemaTracker(ProjectionParseResult parseResult, boolean allowSchemaChange,
@@ -67,15 +68,15 @@ public class ProjectionSchemaTracker extends AbstractSchemaTracker {
     this.schema.copyFrom(projection);
 
     // Work out the projection type: wildcard, empty, or explicit.
-    ScanSchemaTracker.ProjectionType projType;
+    ProjectionType projType;
     if (parseResult.isProjectAll()) {
-      projType = ScanSchemaTracker.ProjectionType.ALL;
+      projType = ProjectionType.ALL;
     } else if (projection.isEmpty()) {
-      projType = ScanSchemaTracker.ProjectionType.NONE;
+      projType = ProjectionType.NONE;
       this.isResolved = true;
       this.allowMapAdditions = false;
     } else {
-      projType = ScanSchemaTracker.ProjectionType.SOME;
+      projType = ProjectionType.SOME;
     }
     this.schema.setProjectionType(projType);
 
@@ -84,6 +85,11 @@ public class ProjectionSchemaTracker extends AbstractSchemaTracker {
     this.implicitInsertPoint = parseResult.wildcardPosn;
   }
 
+  @Override
+  public ProjectedColumn columnProjection(String colName) {
+    return (ProjectedColumn) projection.metadata(colName);
+  }
+
   public void applyProvidedSchema(TupleMetadata providedSchema) {
     boolean isStrict = SchemaUtils.isStrict(providedSchema);
     new ScanSchemaResolver(schema,
@@ -132,6 +138,8 @@ public class ProjectionSchemaTracker extends AbstractSchemaTracker {
   public ProjectionFilter projectionFilter(CustomErrorContext errorContext) {
     switch (projectionType()) {
       case ALL:
+
+        // Empty schema implies we've only seen the wildcard this far.
         if (schema.size() == 0) {
           return ProjectionFilter.PROJECT_ALL;
         }
@@ -146,7 +154,17 @@ public class ProjectionSchemaTracker extends AbstractSchemaTracker {
   @Override
   public void applyReaderSchema(TupleMetadata readerOutputSchema,
       CustomErrorContext errorContext) {
-    new ScanSchemaResolver(schema, SchemaType.READER_SCHEMA, allowMapAdditions, errorContext)
+    SchemaType schemaType;
+
+    // The first reader can reposition columns projected with a wildcard,
+    // other readers cannot as we want to preserve column order after the
+    // first batch.
+    if (readerSchemaCount == 0 && allowSchemaChange) {
+      schemaType = SchemaType.FIRST_READER_SCHEMA;
+    } else {
+      schemaType = SchemaType.READER_SCHEMA;
+    }
+    new ScanSchemaResolver(schema, schemaType, allowMapAdditions, errorContext)
         .applySchema(readerOutputSchema);
     if (!allowSchemaChange) {
       allowMapAdditions = false;
@@ -155,11 +173,11 @@ public class ProjectionSchemaTracker extends AbstractSchemaTracker {
       }
     }
     checkResolved();
+    readerSchemaCount++;
   }
 
   @Override
-  public void expandImplicitCol(ColumnMetadata resolved) {
-    Preconditions.checkArgument(SchemaUtils.isImplicit(resolved));
-    schema.insert(implicitInsertPoint++, resolved);
+  public void expandImplicitCol(ColumnMetadata resolved, ImplicitColumnMarker marker) {
+    schema.insert(implicitInsertPoint++, resolved).markImplicit(marker);
   }
 }
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/scan/v3/schema/ScanSchemaConfigBuilder.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/scan/v3/schema/ScanSchemaConfigBuilder.java
index 24511a1..9af575a 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/scan/v3/schema/ScanSchemaConfigBuilder.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/scan/v3/schema/ScanSchemaConfigBuilder.java
@@ -71,6 +71,8 @@ public class ScanSchemaConfigBuilder {
     if (errorContext == null) {
       errorContext = EmptyErrorContext.INSTANCE;
     }
+
+    // Parse the projection list
     ProjectionParseResult result;
     if (projectionList == null) {
       result = null;
@@ -78,21 +80,45 @@ public class ScanSchemaConfigBuilder {
       result = ScanProjectionParser.parse(projectionList);
     }
 
+    // If a strict schema is provided, then no schema changes are allowed.
     if (providedSchema != null && SchemaUtils.isStrict(providedSchema)) {
       allowSchemaChange = false;
     }
+
+    // Figure out the schema tracker to use
     if (definedSchema == null) {
+
+      // No defined schema: this is a projection-based tracker, possibly
+      // constrained by a provided schema.
       ProjectionSchemaTracker tracker = new ProjectionSchemaTracker(result, allowSchemaChange, errorContext);
+
+      // Apply the provided schema. Doing so forces resolution of the projection
+      // list just appled above.
       if (providedSchema != null) {
         tracker.applyProvidedSchema(providedSchema);
       }
       return tracker;
     } else {
+
+      // Defined schema case, which is supported only via tests at present;
+      // the planner can't yet produce a defined schema.
+
+      // A defined schema can include dynamic columns (those with no type.) If
+      // so, treat the dynamic schema as combination of a projection list and a
+      // provided schema.
       if (!MetadataUtils.hasDynamicColumns(definedSchema)) {
         SchemaBasedTracker tracker = new SchemaBasedTracker(definedSchema, errorContext);
-        tracker.validateProjection(result.dynamicSchema);
+
+        // A projection list is not required. But, if provided, it must be consistent
+        // with the defined schema.
+        if (result != null) {
+          tracker.validateProjection(result.dynamicSchema);
+        }
         return tracker;
       } else {
+
+        // The defined schema has not dynamic columns: it is fully defined, just like
+        // in a "classic" DB. Use a schema-driven schema tracker.
         return new ProjectionSchemaTracker(definedSchema, result, errorContext);
       }
     }
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/scan/v3/schema/ScanSchemaResolver.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/scan/v3/schema/ScanSchemaResolver.java
index 6d1ec40..34ceb9c 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/scan/v3/schema/ScanSchemaResolver.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/scan/v3/schema/ScanSchemaResolver.java
@@ -82,13 +82,14 @@ public class ScanSchemaResolver {
    * Indicates the source of the schema to be analyzed.
    * Each schema type has subtly different rules. The
    * schema type allows us to inject those differences inline
-   * within the resolution process. Also, each schema caries
+   * within the resolution process. Also, each schema carries
    * a tag used for error reporting.
    */
   public enum SchemaType {
     STRICT_PROVIDED_SCHEMA("Provided"),
     LENIENT_PROVIDED_SCHEMA("Provided"),
     EARLY_READER_SCHEMA("Reader"),
+    FIRST_READER_SCHEMA("Reader"),
     READER_SCHEMA("Reader"),
     MISSING_COLS("Missing columns");
 
@@ -105,41 +106,64 @@ public class ScanSchemaResolver {
 
   private final MutableTupleSchema schema;
   private final SchemaType mode;
-  private final boolean isProjectAll;
   private final boolean allowMapAdditions;
   private final String source;
   private final CustomErrorContext errorContext;
+  private final boolean allowColumnReorder;
 
   public ScanSchemaResolver(MutableTupleSchema schema, SchemaType mode,
       boolean allowMapAdditions,
       CustomErrorContext errorContext) {
     this.schema = schema;
-    this.isProjectAll = schema.projectionType() == ProjectionType.ALL;
     this.mode = mode;
     this.errorContext = errorContext;
     this.allowMapAdditions = allowMapAdditions;
     this.source = mode.source();
+    switch (mode) {
+      case STRICT_PROVIDED_SCHEMA:
+      case LENIENT_PROVIDED_SCHEMA:
+      case EARLY_READER_SCHEMA:
+      case FIRST_READER_SCHEMA:
+
+        // Allow reordering columns with projection is of the form
+        // *, foo. Move bar to its place in the schema (foo, bar) rather
+        // than at the end, as with implicit columns.
+        this.allowColumnReorder = schema.projectionType() == ProjectionType.ALL;
+        break;
+      default:
+        this.allowColumnReorder = false;
+    }
   }
 
   public void applySchema(TupleMetadata sourceSchema) {
     switch (schema.projectionType()) {
       case ALL:
+        projectSchema(sourceSchema);
+        if (mode == SchemaType.STRICT_PROVIDED_SCHEMA) {
+          schema.setProjectionType(ScanSchemaTracker.ProjectionType.SOME);
+        }
+        break;
       case SOME:
         projectSchema(sourceSchema);
         break;
       default:
         // Do nothing
     }
-    if (mode == SchemaType.STRICT_PROVIDED_SCHEMA && isProjectAll) {
-      schema.setProjectionType(ScanSchemaTracker.ProjectionType.SOME);
-    }
   }
 
   /**
-   * A project list can contain implicit columns in
-   * addition to the wildcard. The wildcard defines the
-   * <i>insert point</i>: the point at which reader-defined
-   * columns are inserted as found.
+   * A project list can contain implicit columns in addition to the wildcard.
+   * The wildcard defines the <i>insert point</i>: the point at which
+   * reader-defined columns are inserted as found. This version applies a
+   * provided schema to a projection. If we are given a query of the form
+   * {@code SELECT * FROM foo ORDER BY bar}, Drill will give us a projection
+   * list of the form {@code [`**`, `bar`]} and normal projection processing
+   * will project all provided columns, except {@code bar}, in place of the
+   * wildcard. Since this behavior differs from all other DBs, we apply special
+   * processing, we move the projection column into the next wildcard position
+   * as if Drill did not include the extra column projection. This is a hack,
+   * but one that helps with ease-of-use. We apply the same rule to the first
+   * reader schema for the same reason.
    */
   private void projectSchema(TupleMetadata sourceSchema) {
     for (ColumnMetadata colSchema : sourceSchema) {
@@ -148,6 +172,9 @@ public class ScanSchemaResolver {
         insertColumn(colSchema);
       } else {
         mergeColumn(existing, colSchema);
+        if (allowColumnReorder) {
+          schema.moveIfExplicit(colSchema.name());
+        }
       }
     }
   }
@@ -160,8 +187,9 @@ public class ScanSchemaResolver {
    */
   private void insertColumn(ColumnMetadata col) {
     switch (mode) {
+      case FIRST_READER_SCHEMA:
       case READER_SCHEMA:
-        if (!isProjectAll) {
+        if (schema.projectionType() != ProjectionType.ALL) {
           throw new IllegalStateException(
               "Reader should not have projected an unprojected column: " + col.name());
         }
@@ -169,7 +197,7 @@ public class ScanSchemaResolver {
       case EARLY_READER_SCHEMA:
       case LENIENT_PROVIDED_SCHEMA:
       case STRICT_PROVIDED_SCHEMA:
-        if (!isProjectAll || SchemaUtils.isExcludedFromWildcard(col)) {
+        if (schema.projectionType() != ProjectionType.ALL || SchemaUtils.isExcludedFromWildcard(col)) {
           return;
         }
         break;
@@ -200,16 +228,9 @@ public class ScanSchemaResolver {
     switch (mode) {
       case LENIENT_PROVIDED_SCHEMA:
       case STRICT_PROVIDED_SCHEMA:
-        // With a wilcard, there should be no existing column unless
-        // the planner projected an implicit column and the provided
-        // schema defines that same implicit column.
-        if (isProjectAll && !SchemaUtils.isImplicit(colSchema)) {
-          throw UserException.validationError()
-            .message("Provided schema column name conflicts with presumed implicit column name")
-            .addContext("Column", colSchema.name())
-            .addContext(errorContext)
-            .build(logger);
-        }
+        // Even with a wildcard, the planner may add additional columns.
+        // Example SELECT * FROM foo ORDER BY bar
+        // The planner will provide us with [`*`, `bar`]
         break;
       case EARLY_READER_SCHEMA:
         // If the reader offers a column which duplicates an implicit column,
@@ -341,6 +362,7 @@ public class ScanSchemaResolver {
       case LENIENT_PROVIDED_SCHEMA:
       case STRICT_PROVIDED_SCHEMA:
         break;
+      case FIRST_READER_SCHEMA:
       case READER_SCHEMA:
         if (!allowMapAdditions) {
           throw new IllegalStateException("Reader should not have projected column: " + readerCol.name());
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/scan/v3/schema/ScanSchemaTracker.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/scan/v3/schema/ScanSchemaTracker.java
index 7d1aa30..af5efa4 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/scan/v3/schema/ScanSchemaTracker.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/scan/v3/schema/ScanSchemaTracker.java
@@ -18,6 +18,7 @@
 package org.apache.drill.exec.physical.impl.scan.v3.schema;
 
 import org.apache.drill.common.exceptions.CustomErrorContext;
+import org.apache.drill.exec.physical.impl.scan.v3.file.ImplicitColumnMarker;
 import org.apache.drill.exec.physical.resultSet.ResultSetLoader;
 import org.apache.drill.exec.physical.resultSet.impl.ProjectionFilter;
 import org.apache.drill.exec.record.metadata.ColumnMetadata;
@@ -305,6 +306,11 @@ public interface ScanSchemaTracker {
   ProjectionType projectionType();
 
   /**
+   * Return the projection for a column, if any.
+   */
+  ProjectedColumn columnProjection(String colName);
+
+  /**
    * Is the scan schema resolved? The schema is resolved depending on the
    * complex lifecycle explained in the class comment. Resolution occurs
    * when the wildcard (if any) is expanded, and all explicit projection
@@ -349,7 +355,7 @@ public interface ScanSchemaTracker {
    * then determines which partition columns are needed and calls this
    * method to add each one.
    */
-  void expandImplicitCol(ColumnMetadata resolved);
+  void expandImplicitCol(ColumnMetadata resolved, ImplicitColumnMarker marker);
 
   /**
    * Indicate that implicit column parsing is complete. Returns the implicit
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/scan/v3/schema/SchemaBasedTracker.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/scan/v3/schema/SchemaBasedTracker.java
index 24beb38..8e5d6ca 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/scan/v3/schema/SchemaBasedTracker.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/scan/v3/schema/SchemaBasedTracker.java
@@ -18,6 +18,7 @@
 package org.apache.drill.exec.physical.impl.scan.v3.schema;
 
 import org.apache.drill.common.exceptions.CustomErrorContext;
+import org.apache.drill.exec.physical.impl.scan.v3.file.ImplicitColumnMarker;
 import org.apache.drill.exec.physical.impl.scan.v3.schema.DynamicSchemaFilter.RowSchemaFilter;
 import org.apache.drill.exec.physical.resultSet.impl.ProjectionFilter;
 import org.apache.drill.exec.record.metadata.ColumnMetadata;
@@ -39,10 +40,6 @@ public class SchemaBasedTracker extends AbstractSchemaTracker {
     super(errorContext);
     this.definedSchema = definedSchema;
     schema.copyFrom(definedSchema);
-    checkResolved();
-
-    // If not resolved, should not have used this tracker.
-    Preconditions.checkState(isResolved);
 
     ScanSchemaTracker.ProjectionType projType;
     if (schema.size() == 0) {
@@ -51,13 +48,24 @@ public class SchemaBasedTracker extends AbstractSchemaTracker {
       projType = ScanSchemaTracker.ProjectionType.SOME;
     }
     schema.setProjectionType(projType);
+    checkResolved();
+
+    // If not resolved, should not have used this tracker.
+    Preconditions.checkState(isResolved);
   }
 
+  /**
+   * Validate a projection list (provided as an argument) against a
+   * defined schema already held by this tracker. Ensures that, when we
+   * have both a defined schema and projection list, that they are
+   * consistent.
+   *
+   * @param projection the parsed projection list
+   */
   public void validateProjection(TupleMetadata projection) {
-    if (projection == null) {
-      return;
+    if (projection != null) {
+      validateProjection(projection, definedSchema);
     }
-    validateProjection(projection, definedSchema);
   }
 
   @Override
@@ -85,10 +93,13 @@ public class SchemaBasedTracker extends AbstractSchemaTracker {
   }
 
   @Override
-  public void expandImplicitCol(ColumnMetadata resolved) {
+  public void expandImplicitCol(ColumnMetadata resolved, ImplicitColumnMarker marker) {
     throw new IllegalStateException("Can't expand a defined schema.");
   }
 
   @Override
   public int schemaVersion() { return 1; }
+
+  @Override
+  public ProjectedColumn columnProjection(String colName) { return null; }
 }
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/resultSet/RowSetLoader.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/resultSet/RowSetLoader.java
index 08df4b3..bf9ba76 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/resultSet/RowSetLoader.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/resultSet/RowSetLoader.java
@@ -78,7 +78,6 @@ public interface RowSetLoader extends TupleWriter {
    *          variable-length argument list of column values
    * @return this writer
    */
-
   RowSetLoader addRow(Object... values);
 
   /**
@@ -89,7 +88,6 @@ public interface RowSetLoader extends TupleWriter {
    * @param value value of the one and only column
    * @return this writer
    */
-
   RowSetLoader addSingleCol(Object value);
 
   /**
@@ -107,7 +105,6 @@ public interface RowSetLoader extends TupleWriter {
    *
    * @return true if another row can be written, false if not
    */
-
   boolean isFull();
 
   /**
@@ -116,7 +113,6 @@ public interface RowSetLoader extends TupleWriter {
    *
    * @return number of rows to be sent downstream
    */
-
   int rowCount();
 
   /**
@@ -128,7 +124,6 @@ public interface RowSetLoader extends TupleWriter {
    *
    * @return the current write index
    */
-
   int rowIndex();
 
   /**
@@ -151,7 +146,6 @@ public interface RowSetLoader extends TupleWriter {
    *
    * @return true if another row can be added, false if the batch is full
    */
-
   boolean start();
 
   /**
@@ -160,6 +154,5 @@ public interface RowSetLoader extends TupleWriter {
    * to recover from partially-written rows that turn out to contain errors.
    * Done automatically if using <tt>setRow()</tt>.
    */
-
   void save();
 }
\ No newline at end of file
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/store/ColumnExplorer.java b/exec/java-exec/src/main/java/org/apache/drill/exec/store/ColumnExplorer.java
index 0467abc..8fe4ebd 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/store/ColumnExplorer.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/store/ColumnExplorer.java
@@ -317,7 +317,7 @@ public class ColumnExplorer {
    * @param length   row group length
    * @return implicit column value for specified implicit file column
    */
-  private static String getImplicitColumnValue(ImplicitFileColumn column, Path filePath,
+  public static String getImplicitColumnValue(ImplicitFileColumn column, Path filePath,
       FileSystem fs, Integer index, Long start, Long length) {
     if (column instanceof ImplicitFileColumns) {
       ImplicitFileColumns fileColumn = (ImplicitFileColumns) column;
@@ -359,9 +359,10 @@ public class ColumnExplorer {
   }
 
   /**
-   * Returns list of implicit file columns which includes all elements from {@link ImplicitFileColumns},
-   * {@link ImplicitInternalFileColumns#LAST_MODIFIED_TIME} and {@link ImplicitInternalFileColumns#USE_METADATA}
-   * columns.
+   * Returns list of implicit file columns which includes all elements from
+   * {@link ImplicitFileColumns},
+   * {@link ImplicitInternalFileColumns#LAST_MODIFIED_TIME} and
+   * {@link ImplicitInternalFileColumns#USE_METADATA} columns.
    *
    * @return list of implicit file columns
    */
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/store/avro/AvroBatchReader.java b/exec/java-exec/src/main/java/org/apache/drill/exec/store/avro/AvroBatchReader.java
index c22d453..580ba7d 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/store/avro/AvroBatchReader.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/store/avro/AvroBatchReader.java
@@ -23,9 +23,9 @@ import org.apache.avro.generic.GenericDatumReader;
 import org.apache.avro.generic.GenericRecord;
 import org.apache.avro.mapred.FsInput;
 import org.apache.drill.common.exceptions.UserException;
-import org.apache.drill.exec.physical.impl.scan.convert.StandardConversions;
 import org.apache.drill.exec.physical.impl.scan.file.FileScanFramework;
 import org.apache.drill.exec.physical.impl.scan.framework.ManagedReader;
+import org.apache.drill.exec.physical.impl.scan.v3.FixedReceiver;
 import org.apache.drill.exec.physical.resultSet.ResultSetLoader;
 import org.apache.drill.exec.physical.resultSet.RowSetLoader;
 import org.apache.drill.exec.record.metadata.TupleMetadata;
@@ -72,7 +72,7 @@ public class AvroBatchReader implements ManagedReader<FileScanFramework.FileSche
     TupleMetadata readerSchema = AvroSchemaUtil.convert(reader.getSchema());
     logger.debug("Avro file converted schema: {}", readerSchema);
     TupleMetadata providedSchema = negotiator.providedSchema();
-    TupleMetadata tableSchema = StandardConversions.mergeSchemas(providedSchema, readerSchema);
+    TupleMetadata tableSchema = FixedReceiver.Builder.mergeSchemas(providedSchema, readerSchema);
     logger.debug("Avro file table schema: {}", tableSchema);
     negotiator.tableSchema(tableSchema, true);
     loader = negotiator.build();
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/store/avro/ColumnConverterFactory.java b/exec/java-exec/src/main/java/org/apache/drill/exec/store/avro/ColumnConverterFactory.java
index 3c42227..527f21d 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/store/avro/ColumnConverterFactory.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/store/avro/ColumnConverterFactory.java
@@ -21,6 +21,7 @@ import org.apache.avro.generic.GenericFixed;
 import org.apache.avro.generic.GenericRecord;
 import org.apache.avro.util.Utf8;
 import org.apache.drill.exec.physical.impl.scan.convert.StandardConversions;
+import org.apache.drill.exec.physical.impl.scan.v3.FixedReceiver;
 import org.apache.drill.exec.physical.resultSet.RowSetLoader;
 import org.apache.drill.exec.record.metadata.ColumnMetadata;
 import org.apache.drill.exec.record.metadata.TupleMetadata;
@@ -57,7 +58,7 @@ public class ColumnConverterFactory {
     if (providedSchema == null) {
       standardConversions = null;
     } else {
-      standardConversions = new StandardConversions(providedSchema.properties());
+      standardConversions = StandardConversions.builder().withSchema(providedSchema).build();
     }
   }
 
@@ -157,7 +158,7 @@ public class ColumnConverterFactory {
     if (standardConversions == null) {
       valueWriter = scalarWriter;
     } else {
-      valueWriter = standardConversions.converter(scalarWriter, readerSchema);
+      valueWriter = standardConversions.converterFor(scalarWriter, readerSchema);
     }
     return buildScalar(readerSchema, valueWriter);
   }
@@ -246,7 +247,7 @@ public class ColumnConverterFactory {
       TupleWriter tupleWriter, List<ColumnConverter> converters) {
     // fill in tuple schema for cases when it contains recursive named record types
     TupleMetadata readerSchema = AvroSchemaUtil.convert(genericRecord.getSchema());
-    TupleMetadata tableSchema = StandardConversions.mergeSchemas(providedSchema, readerSchema);
+    TupleMetadata tableSchema = FixedReceiver.Builder.mergeSchemas(providedSchema, readerSchema);
     tableSchema.toMetadataList().forEach(tupleWriter::addColumn);
 
     IntStream.range(0, tableSchema.size())
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/store/easy/text/reader/CompliantTextBatchReader.java b/exec/java-exec/src/main/java/org/apache/drill/exec/store/easy/text/reader/CompliantTextBatchReader.java
index 8e3e69e..79840c1 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/store/easy/text/reader/CompliantTextBatchReader.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/store/easy/text/reader/CompliantTextBatchReader.java
@@ -19,8 +19,6 @@ package org.apache.drill.exec.store.easy.text.reader;
 
 import java.io.IOException;
 import java.io.InputStream;
-import java.util.HashMap;
-import java.util.Map;
 
 import org.apache.drill.common.exceptions.UserException;
 import org.apache.drill.common.types.TypeProtos.DataMode;
@@ -160,7 +158,7 @@ public class CompliantTextBatchReader implements ManagedReader<ColumnsSchemaNego
     for (int i = 0; i < fieldNames.length; i++) {
       ScalarWriter colWriter = writer.scalar(fieldNames[i]);
       if (writer.isProjected()) {
-        colWriters[i] = conversions.converter(colWriter, MinorType.VARCHAR);
+        colWriters[i] = conversions.converterFor(colWriter, MinorType.VARCHAR);
       } else {
         colWriters[i] = colWriter;
       }
@@ -220,7 +218,7 @@ public class CompliantTextBatchReader implements ManagedReader<ColumnsSchemaNego
     StandardConversions conversions = conversions(providedSchema);
     ValueWriter[] colWriters = new ValueWriter[providedSchema.size()];
     for (int i = 0; i < colWriters.length; i++) {
-      colWriters[i] = conversions.converter(
+      colWriters[i] = conversions.converterFor(
           writer.scalar(providedSchema.metadata(i).name()), MinorType.VARCHAR);
     }
     return new ConstrainedFieldOutput(writer, colWriters);
@@ -248,14 +246,10 @@ public class CompliantTextBatchReader implements ManagedReader<ColumnsSchemaNego
 
     // CSV maps blank columns to nulls (for nullable non-string columns),
     // or to the default value (for non-nullable non-string columns.)
-    Map<String, String> props = providedSchema.properties();
-    if (props == null) {
-      return new StandardConversions(ColumnMetadata.BLANK_AS_NULL);
-    } else {
-      props = new HashMap<>(props);
-      props.put(ColumnMetadata.BLANK_AS_PROP, ColumnMetadata.BLANK_AS_NULL);
-      return new StandardConversions(props);
-    }
+    return StandardConversions.builder()
+      .withSchema(providedSchema)
+      .blankAs(ColumnMetadata.BLANK_AS_NULL)
+      .build();
   }
 
   /**
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/store/log/LogBatchReader.java b/exec/java-exec/src/main/java/org/apache/drill/exec/store/log/LogBatchReader.java
index f8c512e..99efd76 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/store/log/LogBatchReader.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/store/log/LogBatchReader.java
@@ -90,7 +90,7 @@ public class LogBatchReader implements ManagedReader<FileSchemaNegotiator> {
       writers = new ValueWriter[readerSchema.size()];
       for (int i = 0; i < writers.length; i++) {
         ColumnMetadata colSchema = readerSchema.metadata(i);
-        writers[i] = conversions.converter(rowWriter.scalar(i), colSchema);
+        writers[i] = conversions.converterFor(rowWriter.scalar(i), colSchema);
       }
     }
 
@@ -178,9 +178,9 @@ public class LogBatchReader implements ManagedReader<FileSchemaNegotiator> {
       if (saveMatchedRows) {
         // Save using the defined columns
         TupleMetadata providedSchema = config.providedSchema;
-        StandardConversions conversions = new StandardConversions(
-            providedSchema == null || !providedSchema.hasProperties() ?
-                null : providedSchema.properties());
+        StandardConversions conversions = StandardConversions.builder()
+            .withSchema(providedSchema)
+            .build();
         vectorWriter = new ScalarGroupWriter(writer, config.readerSchema, conversions);
       }
     }
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/ScanTestUtils.java b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/ScanTestUtils.java
index 4f16e76..bfdff94 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/ScanTestUtils.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/ScanTestUtils.java
@@ -45,14 +45,23 @@ import org.apache.drill.test.OperatorFixture;
 
 public class ScanTestUtils {
 
-  // Default file metadata column names; primarily for testing.
+  // Default implicit file column names; primarily for testing.
 
   public static final String FILE_NAME_COL = "filename";
   public static final String FULLY_QUALIFIED_NAME_COL = "fqn";
   public static final String FILE_PATH_COL = "filepath";
   public static final String SUFFIX_COL = "suffix";
   public static final String PARTITION_COL = "dir";
+
+  // Default Internal implicit columns; primarily for testing.
+
   public static final String LAST_MODIFIED_TIME_COL = "lmt";
+  public static final String ROW_GROUP_INDEX_COL = "rgi";
+  public static final String ROW_GROUP_START_COL = "rgs";
+  public static final String ROW_GROUP_LENGTH_COL = "rgl";
+
+  // Yes, the following both have the same value.
+  public static final String USE_METADATA_COL ="$project_metadata$";
   public static final String PROJECT_METADATA_COL = "$project_metadata$";
 
   public static abstract class ScanFixtureBuilder {
@@ -71,8 +80,12 @@ public class ScanTestUtils {
       builder().projection(RowSetTestUtils.projectAll());
     }
 
-    public void projectAllWithMetadata(int dirs) {
-      builder().projection(ScanTestUtils.projectAllWithMetadata(dirs));
+    public void projectAllWithFileImplicit(int dirs) {
+      builder().projection(ScanTestUtils.projectAllWithFileImplicit(dirs));
+    }
+
+    public void projectAllWithAllImplicit(int dirs) {
+      builder().projection(ScanTestUtils.projectAllWithAllImplicit(dirs));
     }
 
     public void setProjection(String... projCols) {
@@ -101,7 +114,7 @@ public class ScanTestUtils {
 
   public static class ScanFixture {
 
-    private OperatorContext opContext;
+    private final OperatorContext opContext;
     public ScanOperatorExec scanOp;
 
     public ScanFixture(OperatorContext opContext, ScanOperatorExec scanOp) {
@@ -152,7 +165,7 @@ public class ScanTestUtils {
    * @return schema with the metadata columns appended to the table columns
    */
 
-  public static TupleMetadata expandMetadata(TupleMetadata base, ImplicitColumnManager metadataProj, int dirCount) {
+  public static TupleMetadata expandImplicit(TupleMetadata base, ImplicitColumnManager metadataProj, int dirCount) {
     TupleMetadata metadataSchema = new TupleSchema();
     for (ColumnMetadata col : base) {
       metadataSchema.addColumn(col);
@@ -188,24 +201,32 @@ public class ScanTestUtils {
     return schema;
   }
 
-  public static List<SchemaPath> expandMetadata(int dirCount) {
+  public static List<String> expandImplicit(boolean includeInternal, int dirCount) {
     List<String> selected = Lists.newArrayList(
         FULLY_QUALIFIED_NAME_COL,
         FILE_PATH_COL,
         FILE_NAME_COL,
-        SUFFIX_COL,
-        LAST_MODIFIED_TIME_COL,
-        PROJECT_METADATA_COL);
+        SUFFIX_COL);
 
+    if (includeInternal) {
+      selected.add(LAST_MODIFIED_TIME_COL);
+      selected.add(PROJECT_METADATA_COL);
+    }
     for (int i = 0; i < dirCount; i++) {
       selected.add(PARTITION_COL + i);
     }
-    return RowSetTestUtils.projectList(selected);
+    return selected;
+  }
+
+  public static List<SchemaPath> projectAllWithFileImplicit(int dirCount) {
+    return RowSetTestUtils.concat(
+        RowSetTestUtils.projectAll(),
+        RowSetTestUtils.projectList(expandImplicit(false, dirCount)));
   }
 
-  public static List<SchemaPath> projectAllWithMetadata(int dirCount) {
+  public static List<SchemaPath> projectAllWithAllImplicit(int dirCount) {
     return RowSetTestUtils.concat(
         RowSetTestUtils.projectAll(),
-        expandMetadata(dirCount));
+        RowSetTestUtils.projectList(expandImplicit(true, dirCount)));
   }
 }
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/TestFileScanFramework.java b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/TestFileScanFramework.java
index d4d806a..2efbe2a 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/TestFileScanFramework.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/TestFileScanFramework.java
@@ -308,7 +308,7 @@ public class TestFileScanFramework extends SubOperatorTest {
 
     // Create the scan operator
     FileScanFixtureBuilder builder = new FileScanFixtureBuilder();
-    builder.projectAllWithMetadata(2);
+    builder.projectAllWithAllImplicit(2);
     builder.addReader(reader);
     ScanFixture scanFixture = builder.build();
     ScanOperatorExec scan = scanFixture.scanOp;
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/TestScanOperExecOuputSchema.java b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/TestScanOperExecOuputSchema.java
index 65646bc..87ea958 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/TestScanOperExecOuputSchema.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/TestScanOperExecOuputSchema.java
@@ -82,6 +82,7 @@ public class TestScanOperExecOuputSchema extends BaseScanOperatorExecTest {
     private void buildWriters(TupleMetadata providedSchema,
         TupleMetadata schema) {
       RowSetLoader rowWriter = tableLoader.writer();
+      StandardConversions conversions = StandardConversions.builder().build();
       for (int i = 0; i < schema.size(); i++) {
         ColumnMetadata colSchema = schema.metadata(i);
         String colName = colSchema.name();
@@ -96,7 +97,7 @@ public class TestScanOperExecOuputSchema extends BaseScanOperatorExecTest {
           writers.add(colSchema.name(), rowWriter.scalar(colIndex));
         } else {
           writers.add(colSchema.name(),
-              StandardConversions.converterFor(rowWriter.scalar(colSchema.name()), colSchema));
+              conversions.converterFor(rowWriter.scalar(colSchema.name()), colSchema));
         }
       }
     }
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/TestScanOrchestratorImplicitColumns.java b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/TestScanOrchestratorImplicitColumns.java
index f57cc5c..36fa31a 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/TestScanOrchestratorImplicitColumns.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/TestScanOrchestratorImplicitColumns.java
@@ -131,7 +131,7 @@ public class TestScanOrchestratorImplicitColumns extends SubOperatorTest {
 
     // Verify
 
-    TupleMetadata expectedSchema = ScanTestUtils.expandMetadata(tableSchema, metadataManager, 2);
+    TupleMetadata expectedSchema = ScanTestUtils.expandImplicit(tableSchema, metadataManager, 2);
 
     String fqn = ImplicitFileColumns.FQN.getValue(filePath);
     String filePathValue = ImplicitFileColumns.FILEPATH.getValue(filePath);
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/convert/TestDirectConverter.java b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/convert/TestDirectConverter.java
index 808447d..aa106c2 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/convert/TestDirectConverter.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/convert/TestDirectConverter.java
@@ -32,6 +32,7 @@ import org.apache.drill.common.types.TypeProtos.MinorType;
 import org.apache.drill.exec.memory.BufferAllocator;
 import org.apache.drill.exec.physical.impl.scan.convert.StandardConversions.ConversionDefn;
 import org.apache.drill.exec.physical.impl.scan.convert.StandardConversions.ConversionType;
+import org.apache.drill.exec.physical.impl.scan.v3.FixedReceiver;
 import org.apache.drill.exec.physical.rowSet.RowSet;
 import org.apache.drill.exec.physical.rowSet.RowSet.SingleRowSet;
 import org.apache.drill.exec.physical.rowSet.RowSetBuilder;
@@ -70,13 +71,13 @@ public class TestDirectConverter extends SubOperatorTest {
     }
 
     public ConversionTestFixture withProperties(Map<String,String> props) {
-      conversions = new StandardConversions(props);
+      conversions = StandardConversions.builder().withProperties(props).build();
       return this;
     }
 
     private StandardConversions conversions() {
       if (conversions == null) {
-        conversions = new StandardConversions();
+        conversions = StandardConversions.builder().build();
       }
       return conversions;
     }
@@ -92,7 +93,7 @@ public class TestDirectConverter extends SubOperatorTest {
 
       // Test uses simple row writer; no support for adding columns.
       assertNotNull(colWriter);
-      ValueWriter converter = conversions().converter(colWriter, source);
+      ValueWriter converter = conversions().converterFor(colWriter, source);
       assertNotNull(converter);
       rowFormat.add(source.name(), converter);
     }
@@ -136,7 +137,7 @@ public class TestDirectConverter extends SubOperatorTest {
       .add("d", MinorType.VARCHAR)
       .build();
 
-    TupleMetadata mergedSchema = StandardConversions.mergeSchemas(providedSchema, readerSchema);
+    TupleMetadata mergedSchema = FixedReceiver.Builder.mergeSchemas(providedSchema, readerSchema);
     assertTrue(expected.isEquivalent(mergedSchema));
     assertTrue(mergedSchema.booleanProperty("foo"));
   }
@@ -680,6 +681,7 @@ public class TestDirectConverter extends SubOperatorTest {
    */
   @Test
   public void testBasicConversionType() {
+    StandardConversions conversions = StandardConversions.builder().build();
     TupleMetadata schema = new SchemaBuilder()
         .add("ti", MinorType.TINYINT)
         .add("si", MinorType.SMALLINT)
@@ -700,84 +702,84 @@ public class TestDirectConverter extends SubOperatorTest {
     ColumnMetadata stringCol = schema.metadata("str");
 
     // TinyInt --> x
-    expect(ConversionType.NONE, StandardConversions.analyze(tinyIntCol, tinyIntCol));
-    expect(ConversionType.IMPLICIT, StandardConversions.analyze(tinyIntCol, smallIntCol));
-    expect(ConversionType.IMPLICIT, StandardConversions.analyze(tinyIntCol, intCol));
-    expect(ConversionType.IMPLICIT, StandardConversions.analyze(tinyIntCol, bigIntCol));
-    expect(ConversionType.IMPLICIT, StandardConversions.analyze(tinyIntCol, float4Col));
-    expect(ConversionType.IMPLICIT, StandardConversions.analyze(tinyIntCol, float8Col));
-    expect(ConversionType.IMPLICIT, StandardConversions.analyze(tinyIntCol, decimalCol));
-    expect(ConversionType.EXPLICIT, StandardConversions.analyze(tinyIntCol, stringCol));
+    expect(ConversionType.NONE, conversions.analyze(tinyIntCol, tinyIntCol));
+    expect(ConversionType.IMPLICIT, conversions.analyze(tinyIntCol, smallIntCol));
+    expect(ConversionType.IMPLICIT, conversions.analyze(tinyIntCol, intCol));
+    expect(ConversionType.IMPLICIT, conversions.analyze(tinyIntCol, bigIntCol));
+    expect(ConversionType.IMPLICIT, conversions.analyze(tinyIntCol, float4Col));
+    expect(ConversionType.IMPLICIT, conversions.analyze(tinyIntCol, float8Col));
+    expect(ConversionType.IMPLICIT, conversions.analyze(tinyIntCol, decimalCol));
+    expect(ConversionType.EXPLICIT, conversions.analyze(tinyIntCol, stringCol));
 
     // SmallInt --> x
-    expect(ConversionType.IMPLICIT_UNSAFE, StandardConversions.analyze(smallIntCol, tinyIntCol));
-    expect(ConversionType.NONE, StandardConversions.analyze(smallIntCol, smallIntCol));
-    expect(ConversionType.IMPLICIT, StandardConversions.analyze(smallIntCol, intCol));
-    expect(ConversionType.IMPLICIT, StandardConversions.analyze(smallIntCol, bigIntCol));
-    expect(ConversionType.IMPLICIT, StandardConversions.analyze(smallIntCol, float4Col));
-    expect(ConversionType.IMPLICIT, StandardConversions.analyze(smallIntCol, float8Col));
-    expect(ConversionType.IMPLICIT, StandardConversions.analyze(smallIntCol, decimalCol));
-    expect(ConversionType.EXPLICIT, StandardConversions.analyze(smallIntCol, stringCol));
+    expect(ConversionType.IMPLICIT_UNSAFE, conversions.analyze(smallIntCol, tinyIntCol));
+    expect(ConversionType.NONE, conversions.analyze(smallIntCol, smallIntCol));
+    expect(ConversionType.IMPLICIT, conversions.analyze(smallIntCol, intCol));
+    expect(ConversionType.IMPLICIT, conversions.analyze(smallIntCol, bigIntCol));
+    expect(ConversionType.IMPLICIT, conversions.analyze(smallIntCol, float4Col));
+    expect(ConversionType.IMPLICIT, conversions.analyze(smallIntCol, float8Col));
+    expect(ConversionType.IMPLICIT, conversions.analyze(smallIntCol, decimalCol));
+    expect(ConversionType.EXPLICIT, conversions.analyze(smallIntCol, stringCol));
 
     // Int --> x
-    expect(ConversionType.IMPLICIT_UNSAFE, StandardConversions.analyze(intCol, tinyIntCol));
-    expect(ConversionType.IMPLICIT_UNSAFE, StandardConversions.analyze(intCol, smallIntCol));
-    expect(ConversionType.NONE, StandardConversions.analyze(intCol, intCol));
-    expect(ConversionType.IMPLICIT, StandardConversions.analyze(intCol, bigIntCol));
-    expect(ConversionType.IMPLICIT, StandardConversions.analyze(intCol, float4Col));
-    expect(ConversionType.IMPLICIT, StandardConversions.analyze(intCol, float8Col));
-    expect(ConversionType.IMPLICIT, StandardConversions.analyze(intCol, decimalCol));
-    expect(ConversionType.EXPLICIT, StandardConversions.analyze(intCol, stringCol));
+    expect(ConversionType.IMPLICIT_UNSAFE, conversions.analyze(intCol, tinyIntCol));
+    expect(ConversionType.IMPLICIT_UNSAFE, conversions.analyze(intCol, smallIntCol));
+    expect(ConversionType.NONE, conversions.analyze(intCol, intCol));
+    expect(ConversionType.IMPLICIT, conversions.analyze(intCol, bigIntCol));
+    expect(ConversionType.IMPLICIT, conversions.analyze(intCol, float4Col));
+    expect(ConversionType.IMPLICIT, conversions.analyze(intCol, float8Col));
+    expect(ConversionType.IMPLICIT, conversions.analyze(intCol, decimalCol));
+    expect(ConversionType.EXPLICIT, conversions.analyze(intCol, stringCol));
 
     // BigInt --> x
-    expect(ConversionType.IMPLICIT_UNSAFE, StandardConversions.analyze(bigIntCol, tinyIntCol));
-    expect(ConversionType.IMPLICIT_UNSAFE, StandardConversions.analyze(bigIntCol, smallIntCol));
-    expect(ConversionType.IMPLICIT_UNSAFE, StandardConversions.analyze(bigIntCol, intCol));
-    expect(ConversionType.NONE, StandardConversions.analyze(bigIntCol, bigIntCol));
-    expect(ConversionType.IMPLICIT, StandardConversions.analyze(bigIntCol, float4Col));
-    expect(ConversionType.IMPLICIT, StandardConversions.analyze(bigIntCol, float8Col));
-    expect(ConversionType.IMPLICIT, StandardConversions.analyze(bigIntCol, decimalCol));
-    expect(ConversionType.EXPLICIT, StandardConversions.analyze(bigIntCol, stringCol));
+    expect(ConversionType.IMPLICIT_UNSAFE, conversions.analyze(bigIntCol, tinyIntCol));
+    expect(ConversionType.IMPLICIT_UNSAFE, conversions.analyze(bigIntCol, smallIntCol));
+    expect(ConversionType.IMPLICIT_UNSAFE, conversions.analyze(bigIntCol, intCol));
+    expect(ConversionType.NONE, conversions.analyze(bigIntCol, bigIntCol));
+    expect(ConversionType.IMPLICIT, conversions.analyze(bigIntCol, float4Col));
+    expect(ConversionType.IMPLICIT, conversions.analyze(bigIntCol, float8Col));
+    expect(ConversionType.IMPLICIT, conversions.analyze(bigIntCol, decimalCol));
+    expect(ConversionType.EXPLICIT, conversions.analyze(bigIntCol, stringCol));
 
     // Float4 --> x
-    expect(ConversionType.IMPLICIT_UNSAFE, StandardConversions.analyze(float4Col, tinyIntCol));
-    expect(ConversionType.IMPLICIT_UNSAFE, StandardConversions.analyze(float4Col, smallIntCol));
-    expect(ConversionType.IMPLICIT_UNSAFE, StandardConversions.analyze(float4Col, intCol));
-    expect(ConversionType.IMPLICIT_UNSAFE, StandardConversions.analyze(float4Col, bigIntCol));
-    expect(ConversionType.NONE, StandardConversions.analyze(float4Col, float4Col));
-    expect(ConversionType.IMPLICIT, StandardConversions.analyze(float4Col, float8Col));
-    expect(ConversionType.IMPLICIT, StandardConversions.analyze(float4Col, decimalCol));
-    expect(ConversionType.EXPLICIT, StandardConversions.analyze(float4Col, stringCol));
+    expect(ConversionType.IMPLICIT_UNSAFE, conversions.analyze(float4Col, tinyIntCol));
+    expect(ConversionType.IMPLICIT_UNSAFE, conversions.analyze(float4Col, smallIntCol));
+    expect(ConversionType.IMPLICIT_UNSAFE, conversions.analyze(float4Col, intCol));
+    expect(ConversionType.IMPLICIT_UNSAFE, conversions.analyze(float4Col, bigIntCol));
+    expect(ConversionType.NONE, conversions.analyze(float4Col, float4Col));
+    expect(ConversionType.IMPLICIT, conversions.analyze(float4Col, float8Col));
+    expect(ConversionType.IMPLICIT, conversions.analyze(float4Col, decimalCol));
+    expect(ConversionType.EXPLICIT, conversions.analyze(float4Col, stringCol));
 
     // Float8 --> x
-    expect(ConversionType.IMPLICIT_UNSAFE, StandardConversions.analyze(float8Col, tinyIntCol));
-    expect(ConversionType.IMPLICIT_UNSAFE, StandardConversions.analyze(float8Col, smallIntCol));
-    expect(ConversionType.IMPLICIT_UNSAFE, StandardConversions.analyze(float8Col, intCol));
-    expect(ConversionType.IMPLICIT_UNSAFE, StandardConversions.analyze(float8Col, bigIntCol));
-    expect(ConversionType.IMPLICIT_UNSAFE, StandardConversions.analyze(float8Col, float4Col));
-    expect(ConversionType.NONE, StandardConversions.analyze(float8Col, float8Col));
-    expect(ConversionType.IMPLICIT, StandardConversions.analyze(float8Col, decimalCol));
-    expect(ConversionType.EXPLICIT, StandardConversions.analyze(float8Col, stringCol));
+    expect(ConversionType.IMPLICIT_UNSAFE, conversions.analyze(float8Col, tinyIntCol));
+    expect(ConversionType.IMPLICIT_UNSAFE, conversions.analyze(float8Col, smallIntCol));
+    expect(ConversionType.IMPLICIT_UNSAFE, conversions.analyze(float8Col, intCol));
+    expect(ConversionType.IMPLICIT_UNSAFE, conversions.analyze(float8Col, bigIntCol));
+    expect(ConversionType.IMPLICIT_UNSAFE, conversions.analyze(float8Col, float4Col));
+    expect(ConversionType.NONE, conversions.analyze(float8Col, float8Col));
+    expect(ConversionType.IMPLICIT, conversions.analyze(float8Col, decimalCol));
+    expect(ConversionType.EXPLICIT, conversions.analyze(float8Col, stringCol));
 
     // Decimal --> x
-    expect(ConversionType.EXPLICIT, StandardConversions.analyze(decimalCol, tinyIntCol));
-    expect(ConversionType.EXPLICIT, StandardConversions.analyze(decimalCol, smallIntCol));
-    expect(ConversionType.EXPLICIT, StandardConversions.analyze(decimalCol, intCol));
-    expect(ConversionType.EXPLICIT, StandardConversions.analyze(decimalCol, bigIntCol));
-    expect(ConversionType.EXPLICIT, StandardConversions.analyze(decimalCol, float4Col));
-    expect(ConversionType.EXPLICIT, StandardConversions.analyze(decimalCol, float8Col));
-    expect(ConversionType.NONE, StandardConversions.analyze(decimalCol, decimalCol));
-    expect(ConversionType.EXPLICIT, StandardConversions.analyze(decimalCol, stringCol));
+    expect(ConversionType.EXPLICIT, conversions.analyze(decimalCol, tinyIntCol));
+    expect(ConversionType.EXPLICIT, conversions.analyze(decimalCol, smallIntCol));
+    expect(ConversionType.EXPLICIT, conversions.analyze(decimalCol, intCol));
+    expect(ConversionType.EXPLICIT, conversions.analyze(decimalCol, bigIntCol));
+    expect(ConversionType.EXPLICIT, conversions.analyze(decimalCol, float4Col));
+    expect(ConversionType.EXPLICIT, conversions.analyze(decimalCol, float8Col));
+    expect(ConversionType.NONE, conversions.analyze(decimalCol, decimalCol));
+    expect(ConversionType.EXPLICIT, conversions.analyze(decimalCol, stringCol));
 
     // VarChar --> x
-    expect(ConversionType.EXPLICIT, StandardConversions.analyze(stringCol, tinyIntCol));
-    expect(ConversionType.EXPLICIT, StandardConversions.analyze(stringCol, smallIntCol));
-    expect(ConversionType.EXPLICIT, StandardConversions.analyze(stringCol, intCol));
-    expect(ConversionType.EXPLICIT, StandardConversions.analyze(stringCol, bigIntCol));
-    expect(ConversionType.EXPLICIT, StandardConversions.analyze(stringCol, float4Col));
-    expect(ConversionType.EXPLICIT, StandardConversions.analyze(stringCol, float8Col));
-    expect(ConversionType.EXPLICIT, StandardConversions.analyze(stringCol, decimalCol));
-    expect(ConversionType.NONE, StandardConversions.analyze(stringCol, stringCol));
+    expect(ConversionType.EXPLICIT, conversions.analyze(stringCol, tinyIntCol));
+    expect(ConversionType.EXPLICIT, conversions.analyze(stringCol, smallIntCol));
+    expect(ConversionType.EXPLICIT, conversions.analyze(stringCol, intCol));
+    expect(ConversionType.EXPLICIT, conversions.analyze(stringCol, bigIntCol));
+    expect(ConversionType.EXPLICIT, conversions.analyze(stringCol, float4Col));
+    expect(ConversionType.EXPLICIT, conversions.analyze(stringCol, float8Col));
+    expect(ConversionType.EXPLICIT, conversions.analyze(stringCol, decimalCol));
+    expect(ConversionType.NONE, conversions.analyze(stringCol, stringCol));
   }
 
   /**
@@ -785,6 +787,7 @@ public class TestDirectConverter extends SubOperatorTest {
    */
   @Test
   public void testSpecialConversionType() {
+    StandardConversions conversions = StandardConversions.builder().build();
     TupleMetadata schema = new SchemaBuilder()
         .add("time", MinorType.TIME)
         .add("date", MinorType.DATE)
@@ -807,40 +810,40 @@ public class TestDirectConverter extends SubOperatorTest {
     ColumnMetadata stringCol = schema.metadata("str");
 
     // TIME
-    expect(ConversionType.NONE, StandardConversions.analyze(timeCol, timeCol));
-    expect(ConversionType.EXPLICIT, StandardConversions.analyze(timeCol, stringCol));
-    expect(ConversionType.EXPLICIT, StandardConversions.analyze(stringCol, timeCol));
-    expect(ConversionType.IMPLICIT, StandardConversions.analyze(intCol, timeCol));
-    expect(ConversionType.IMPLICIT, StandardConversions.analyze(timeCol, intCol));
+    expect(ConversionType.NONE, conversions.analyze(timeCol, timeCol));
+    expect(ConversionType.EXPLICIT, conversions.analyze(timeCol, stringCol));
+    expect(ConversionType.EXPLICIT, conversions.analyze(stringCol, timeCol));
+    expect(ConversionType.IMPLICIT, conversions.analyze(intCol, timeCol));
+    expect(ConversionType.IMPLICIT, conversions.analyze(timeCol, intCol));
 
     // DATE
-    expect(ConversionType.NONE, StandardConversions.analyze(dateCol, dateCol));
-    expect(ConversionType.EXPLICIT, StandardConversions.analyze(dateCol, stringCol));
-    expect(ConversionType.EXPLICIT, StandardConversions.analyze(stringCol, dateCol));
-    expect(ConversionType.IMPLICIT, StandardConversions.analyze(bigIntCol, dateCol));
-    expect(ConversionType.IMPLICIT, StandardConversions.analyze(dateCol, bigIntCol));
+    expect(ConversionType.NONE, conversions.analyze(dateCol, dateCol));
+    expect(ConversionType.EXPLICIT, conversions.analyze(dateCol, stringCol));
+    expect(ConversionType.EXPLICIT, conversions.analyze(stringCol, dateCol));
+    expect(ConversionType.IMPLICIT, conversions.analyze(bigIntCol, dateCol));
+    expect(ConversionType.IMPLICIT, conversions.analyze(dateCol, bigIntCol));
 
     // TIMESTAMP
-    expect(ConversionType.NONE, StandardConversions.analyze(tsCol, tsCol));
-    expect(ConversionType.EXPLICIT, StandardConversions.analyze(tsCol, stringCol));
-    expect(ConversionType.EXPLICIT, StandardConversions.analyze(stringCol, tsCol));
-    expect(ConversionType.IMPLICIT, StandardConversions.analyze(bigIntCol, tsCol));
-    expect(ConversionType.IMPLICIT, StandardConversions.analyze(tsCol, bigIntCol));
+    expect(ConversionType.NONE, conversions.analyze(tsCol, tsCol));
+    expect(ConversionType.EXPLICIT, conversions.analyze(tsCol, stringCol));
+    expect(ConversionType.EXPLICIT, conversions.analyze(stringCol, tsCol));
+    expect(ConversionType.IMPLICIT, conversions.analyze(bigIntCol, tsCol));
+    expect(ConversionType.IMPLICIT, conversions.analyze(tsCol, bigIntCol));
 
     // INTERVAL
-    expect(ConversionType.NONE, StandardConversions.analyze(intervalCol, intervalCol));
-    expect(ConversionType.EXPLICIT, StandardConversions.analyze(intervalCol, stringCol));
-    expect(ConversionType.EXPLICIT, StandardConversions.analyze(stringCol, intervalCol));
+    expect(ConversionType.NONE, conversions.analyze(intervalCol, intervalCol));
+    expect(ConversionType.EXPLICIT, conversions.analyze(intervalCol, stringCol));
+    expect(ConversionType.EXPLICIT, conversions.analyze(stringCol, intervalCol));
 
     // INTERVALYEAR
-    expect(ConversionType.NONE, StandardConversions.analyze(yearCol, yearCol));
-    expect(ConversionType.EXPLICIT, StandardConversions.analyze(yearCol, stringCol));
-    expect(ConversionType.EXPLICIT, StandardConversions.analyze(stringCol, yearCol));
+    expect(ConversionType.NONE, conversions.analyze(yearCol, yearCol));
+    expect(ConversionType.EXPLICIT, conversions.analyze(yearCol, stringCol));
+    expect(ConversionType.EXPLICIT, conversions.analyze(stringCol, yearCol));
 
     // INTERVALDAY
-    expect(ConversionType.NONE, StandardConversions.analyze(dayCol, dayCol));
-    expect(ConversionType.EXPLICIT, StandardConversions.analyze(dayCol, stringCol));
-    expect(ConversionType.EXPLICIT, StandardConversions.analyze(stringCol, dayCol));
+    expect(ConversionType.NONE, conversions.analyze(dayCol, dayCol));
+    expect(ConversionType.EXPLICIT, conversions.analyze(dayCol, stringCol));
+    expect(ConversionType.EXPLICIT, conversions.analyze(stringCol, dayCol));
   }
 
   /**
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/project/TestScanLevelProjection.java b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/project/TestScanLevelProjection.java
index f4cee33..c5d92fe 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/project/TestScanLevelProjection.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/project/TestScanLevelProjection.java
@@ -24,7 +24,7 @@ import static org.junit.Assert.assertSame;
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
 
-import org.apache.drill.categories.EvfTests;
+import org.apache.drill.categories.EvfTest;
 import org.apache.drill.common.exceptions.UserException;
 import org.apache.drill.common.expression.SchemaPath;
 import org.apache.drill.common.types.TypeProtos.MinorType;
@@ -47,7 +47,7 @@ import org.junit.experimental.categories.Category;
  * Test the level of projection done at the level of the scan as a whole;
  * before knowledge of table "implicit" columns or the specific table schema.
  */
-@Category(EvfTests.class)
+@Category(EvfTest.class)
 public class TestScanLevelProjection extends SubOperatorTest {
 
   private boolean isProjected(ProjectionFilter filter, ColumnMetadata col) {
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/project/TestSchemaSmoothing.java b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/project/TestSchemaSmoothing.java
index ea55bc9..842f18e 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/project/TestSchemaSmoothing.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/project/TestSchemaSmoothing.java
@@ -87,7 +87,6 @@ import org.junit.experimental.categories.Category;
  * because such an algorithm would require time-travel: looking into
  * the future to know what data will be scanned.)
  */
-
 @Category(RowSetTests.class)
 public class TestSchemaSmoothing extends SubOperatorTest {
 
@@ -103,12 +102,10 @@ public class TestSchemaSmoothing extends SubOperatorTest {
    * discrete is just to run the basic lifecycle in a way that
    * is compatible with the schema-persistence version.
    */
-
   @Test
   public void testDiscrete() {
 
     // Set up the file metadata manager
-
     Path filePathA = new Path("hdfs:///w/x/y/a.csv");
     Path filePathB = new Path("hdfs:///w/x/y/b.csv");
     ImplicitColumnManager metadataManager = new ImplicitColumnManager(
@@ -116,18 +113,15 @@ public class TestSchemaSmoothing extends SubOperatorTest {
         standardOptions(Lists.newArrayList(filePathA, filePathB)));
 
     // Set up the scan level projection
-
     ScanLevelProjection scanProj = ScanLevelProjection.build(
         RowSetTestUtils.projectList(ScanTestUtils.FILE_NAME_COL, "a", "b"),
         ScanTestUtils.parsers(metadataManager.projectionParser()));
 
     {
       // Define a file a.csv
-
       metadataManager.startFile(filePathA);
 
       // Build the output schema from the (a, b) table schema
-
       TupleMetadata twoColSchema = new SchemaBuilder()
           .add("a", MinorType.INT)
           .addNullable("b", MinorType.VARCHAR, 10)
@@ -139,7 +133,6 @@ public class TestSchemaSmoothing extends SubOperatorTest {
           ScanTestUtils.resolvers(metadataManager));
 
       // Verify the full output schema
-
       TupleMetadata expectedSchema = new SchemaBuilder()
           .add("filename", MinorType.VARCHAR)
           .add("a", MinorType.INT)
@@ -147,21 +140,18 @@ public class TestSchemaSmoothing extends SubOperatorTest {
           .buildSchema();
 
       // Verify
-
       List<ResolvedColumn> columns = rootTuple.columns();
       assertEquals(3, columns.size());
       assertTrue(ScanTestUtils.schema(rootTuple).isEquivalent(expectedSchema));
       assertEquals(ScanTestUtils.FILE_NAME_COL, columns.get(0).name());
       assertEquals("a.csv", ((FileMetadataColumn) columns.get(0)).value());
       assertTrue(columns.get(1) instanceof ResolvedTableColumn);
-   }
+    }
     {
       // Define a file b.csv
-
       metadataManager.startFile(filePathB);
 
       // Build the output schema from the (a) table schema
-
       TupleMetadata oneColSchema = new SchemaBuilder()
           .add("a", MinorType.INT)
           .buildSchema();
@@ -177,7 +167,6 @@ public class TestSchemaSmoothing extends SubOperatorTest {
       // vector level as part of vector persistence.) During projection, it is
       // marked with type NULL so that the null column builder will fill in
       // the proper type.
-
       TupleMetadata expectedSchema = new SchemaBuilder()
           .add("filename", MinorType.VARCHAR)
           .add("a", MinorType.INT)
@@ -185,7 +174,6 @@ public class TestSchemaSmoothing extends SubOperatorTest {
           .buildSchema();
 
       // Verify
-
       List<ResolvedColumn> columns = rootTuple.columns();
       assertEquals(3, columns.size());
       assertTrue(ScanTestUtils.schema(rootTuple).isEquivalent(expectedSchema));
@@ -200,7 +188,6 @@ public class TestSchemaSmoothing extends SubOperatorTest {
    * Low-level test of the smoothing projection, including the exceptions
    * it throws when things are not going its way.
    */
-
   @Test
   public void testSmoothingProjection() {
     final ScanLevelProjection scanProj = ScanLevelProjection.build(
@@ -208,7 +195,6 @@ public class TestSchemaSmoothing extends SubOperatorTest {
         ScanTestUtils.parsers());
 
     // Table 1: (a: nullable bigint, b)
-
     final TupleMetadata schema1 = new SchemaBuilder()
         .addNullable("a", MinorType.BIGINT)
         .addNullable("b", MinorType.VARCHAR)
@@ -225,7 +211,6 @@ public class TestSchemaSmoothing extends SubOperatorTest {
     }
 
     // Table 2: (a: nullable bigint, c), column omitted, original schema preserved
-
     final TupleMetadata schema2 = new SchemaBuilder()
         .addNullable("a", MinorType.BIGINT)
         .add("c", MinorType.FLOAT8)
@@ -243,7 +228,6 @@ public class TestSchemaSmoothing extends SubOperatorTest {
     }
 
     // Table 3: (a, c, d), column added, must replan schema
-
     final TupleMetadata schema3 = new SchemaBuilder()
         .addNullable("a", MinorType.BIGINT)
         .addNullable("b", MinorType.VARCHAR)
@@ -262,7 +246,6 @@ public class TestSchemaSmoothing extends SubOperatorTest {
     }
 
     // Table 4: (a: double), change type must replan schema
-
     final TupleMetadata schema4 = new SchemaBuilder()
         .addNullable("a", MinorType.FLOAT8)
         .addNullable("b", MinorType.VARCHAR)
@@ -280,7 +263,6 @@ public class TestSchemaSmoothing extends SubOperatorTest {
     }
 
     // Table 5: Drop a non-nullable column, must replan
-
     final TupleMetadata schema6 = new SchemaBuilder()
         .addNullable("a", MinorType.BIGINT)
         .addNullable("b", MinorType.VARCHAR)
@@ -302,7 +284,6 @@ public class TestSchemaSmoothing extends SubOperatorTest {
    * schema. Discard prior schema. Turn off auto expansion of
    * metadata for a simpler test.
    */
-
   @Test
   public void testSmaller() {
     final ScanLevelProjection scanProj = ScanLevelProjection.build(
@@ -340,7 +321,6 @@ public class TestSchemaSmoothing extends SubOperatorTest {
    * Case in which the table schema and prior are disjoint
    * sets. Discard the prior schema.
    */
-
   @Test
   public void testDisjoint() {
     final ScanLevelProjection scanProj = ScanLevelProjection.build(
@@ -377,7 +357,6 @@ public class TestSchemaSmoothing extends SubOperatorTest {
   /**
    * Column names match, but types differ. Discard the prior schema.
    */
-
   @Test
   public void testDifferentTypes() {
     final ScanLevelProjection scanProj = ScanLevelProjection.build(
@@ -411,7 +390,6 @@ public class TestSchemaSmoothing extends SubOperatorTest {
    * schema (though, the output is no different than if we discarded
    * the prior schema...)
    */
-
   @Test
   public void testSameSchemas() {
     final ScanLevelProjection scanProj = ScanLevelProjection.build(
@@ -446,7 +424,6 @@ public class TestSchemaSmoothing extends SubOperatorTest {
    * The prior and table schemas are identical, but the cases of names differ.
    * Preserve the case of the first schema.
    */
-
   @Test
   public void testDifferentCase() {
     final ScanLevelProjection scanProj = ScanLevelProjection.build(
@@ -481,7 +458,6 @@ public class TestSchemaSmoothing extends SubOperatorTest {
    * Can't preserve the prior schema if it had required columns
    * where the new schema has no columns.
    */
-
   @Test
   public void testRequired() {
     final ScanLevelProjection scanProj = ScanLevelProjection.build(
@@ -513,7 +489,6 @@ public class TestSchemaSmoothing extends SubOperatorTest {
    * Preserve the prior schema if table is a subset and missing columns
    * are nullable or repeated.
    */
-
   @Test
   public void testMissingNullableColumns() {
     final ScanLevelProjection scanProj = ScanLevelProjection.build(
@@ -546,7 +521,6 @@ public class TestSchemaSmoothing extends SubOperatorTest {
    * Preserve the prior schema if table is a subset. Map the table
    * columns to the output using the prior schema ordering.
    */
-
   @Test
   public void testReordering() {
     final ScanLevelProjection scanProj = ScanLevelProjection.build(
@@ -580,12 +554,10 @@ public class TestSchemaSmoothing extends SubOperatorTest {
    * If using the legacy wildcard expansion, reuse schema if partition paths
    * are the same length.
    */
-
   @Test
   public void testSamePartitionLength() {
 
     // Set up the file metadata manager
-
     Path filePathA = new Path("hdfs:///w/x/y/a.csv");
     Path filePathB = new Path("hdfs:///w/x/y/b.csv");
     ImplicitColumnManager metadataManager = new ImplicitColumnManager(
@@ -593,9 +565,8 @@ public class TestSchemaSmoothing extends SubOperatorTest {
         standardOptions(Lists.newArrayList(filePathA, filePathB)));
 
     // Set up the scan level projection
-
     ScanLevelProjection scanProj = ScanLevelProjection.build(
-        ScanTestUtils.projectAllWithMetadata(2),
+        ScanTestUtils.projectAllWithAllImplicit(2),
         ScanTestUtils.parsers(metadataManager.projectionParser()));
 
     // Define the schema smoother
@@ -608,7 +579,7 @@ public class TestSchemaSmoothing extends SubOperatorTest {
         .add("b", MinorType.VARCHAR)
         .buildSchema();
 
-    TupleMetadata expectedSchema = ScanTestUtils.expandMetadata(tableSchema, metadataManager, 2);
+    TupleMetadata expectedSchema = ScanTestUtils.expandImplicit(tableSchema, metadataManager, 2);
     {
       metadataManager.startFile(filePathA);
       ResolvedRow rootTuple = doResolve(smoother, tableSchema);
@@ -627,12 +598,10 @@ public class TestSchemaSmoothing extends SubOperatorTest {
    * is shorter than the previous. (Unneeded partitions will be set to null by the
    * scan projector.)
    */
-
   @Test
   public void testShorterPartitionLength() {
 
     // Set up the file metadata manager
-
     Path filePathA = new Path("hdfs:///w/x/y/a.csv");
     Path filePathB = new Path("hdfs:///w/x/b.csv");
     ImplicitColumnManager metadataManager = new ImplicitColumnManager(
@@ -640,13 +609,11 @@ public class TestSchemaSmoothing extends SubOperatorTest {
         standardOptions(Lists.newArrayList(filePathA, filePathB)));
 
     // Set up the scan level projection
-
     ScanLevelProjection scanProj = ScanLevelProjection.build(
-        ScanTestUtils.projectAllWithMetadata(2),
+        ScanTestUtils.projectAllWithAllImplicit(2),
         ScanTestUtils.parsers(metadataManager.projectionParser()));
 
     // Define the schema smoother
-
     SchemaSmoother smoother = new SchemaSmoother(scanProj,
         ScanTestUtils.resolvers(metadataManager));
 
@@ -655,7 +622,7 @@ public class TestSchemaSmoothing extends SubOperatorTest {
         .add("b", MinorType.VARCHAR)
         .buildSchema();
 
-    TupleMetadata expectedSchema = ScanTestUtils.expandMetadata(tableSchema, metadataManager, 2);
+    TupleMetadata expectedSchema = ScanTestUtils.expandImplicit(tableSchema, metadataManager, 2);
     {
       metadataManager.startFile(filePathA);
       ResolvedRow rootTuple = doResolve(smoother, tableSchema);
@@ -674,12 +641,10 @@ public class TestSchemaSmoothing extends SubOperatorTest {
    * schema even if the new partition path is longer than the previous.
    * Because all file names are provided up front.
    */
-
   @Test
   public void testLongerPartitionLength() {
 
     // Set up the file metadata manager
-
     Path filePathA = new Path("hdfs:///w/x/a.csv");
     Path filePathB = new Path("hdfs:///w/x/y/b.csv");
     ImplicitColumnManager metadataManager = new ImplicitColumnManager(
@@ -687,13 +652,11 @@ public class TestSchemaSmoothing extends SubOperatorTest {
         standardOptions(Lists.newArrayList(filePathA, filePathB)));
 
     // Set up the scan level projection
-
     ScanLevelProjection scanProj = ScanLevelProjection.build(
-        ScanTestUtils.projectAllWithMetadata(2),
+        ScanTestUtils.projectAllWithAllImplicit(2),
         ScanTestUtils.parsers(metadataManager.projectionParser()));
 
     // Define the schema smoother
-
     SchemaSmoother smoother = new SchemaSmoother(scanProj,
         ScanTestUtils.resolvers(metadataManager));
 
@@ -702,7 +665,7 @@ public class TestSchemaSmoothing extends SubOperatorTest {
         .add("b", MinorType.VARCHAR)
         .buildSchema();
 
-    TupleMetadata expectedSchema = ScanTestUtils.expandMetadata(tableSchema, metadataManager, 2);
+    TupleMetadata expectedSchema = ScanTestUtils.expandImplicit(tableSchema, metadataManager, 2);
     {
       metadataManager.startFile(filePathA);
       ResolvedRow rootTuple = doResolve(smoother, tableSchema);
@@ -719,7 +682,6 @@ public class TestSchemaSmoothing extends SubOperatorTest {
   /**
    * Integrated test across multiple schemas at the batch level.
    */
-
   @Test
   public void testSmoothableSchemaBatches() {
     final ScanLevelProjection scanProj = ScanLevelProjection.build(
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/v3/TestFixedReceiver.java b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/v3/TestFixedReceiver.java
new file mode 100644
index 0000000..ef4dcd3
--- /dev/null
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/v3/TestFixedReceiver.java
@@ -0,0 +1,124 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.drill.exec.physical.impl.scan.v3;
+
+import static org.junit.Assert.assertTrue;
+
+import org.apache.drill.common.types.TypeProtos.MinorType;
+import org.apache.drill.exec.physical.impl.scan.ScanOperatorExec;
+import org.apache.drill.exec.physical.impl.scan.v3.FixedReceiver.Builder;
+import org.apache.drill.exec.physical.rowSet.RowSet.SingleRowSet;
+import org.apache.drill.exec.record.VectorContainer;
+import org.apache.drill.exec.record.metadata.SchemaBuilder;
+import org.apache.drill.exec.record.metadata.TupleMetadata;
+import org.apache.drill.test.rowSet.RowSetUtilities;
+import org.junit.Test;
+
+public class TestFixedReceiver extends BaseScanTest {
+
+  /**
+   * Mock reader which writes columns as strings, using
+   * standard conversions to convert to different types as
+   * specified in a provided schema.
+   */
+  private static class MockReader implements ManagedReader {
+
+    private final FixedReceiver receiver;
+
+    public MockReader(SchemaNegotiator negotiator) {
+      Builder builder = FixedReceiver.builderFor(negotiator)
+          .schemaIsComplete();
+      TupleMetadata readerSchema = new SchemaBuilder()
+          .add("ti", MinorType.VARCHAR)
+          .add("si", MinorType.VARCHAR)
+          .add("int", MinorType.VARCHAR)
+          .add("bi", MinorType.VARCHAR)
+          .add("fl", MinorType.VARCHAR)
+          .add("db", MinorType.VARCHAR)
+          .buildSchema();
+      receiver = builder.build(readerSchema);
+    }
+
+    @Override
+    public boolean next() {
+      if (receiver.rowWriter().loader().batchCount() > 0) {
+        return false;
+      }
+      receiver.start();
+      receiver.scalar(0).setString("11");
+      receiver.scalar(1).setString("12");
+      receiver.scalar(2).setString("13");
+      receiver.scalar(3).setString("14");
+      receiver.scalar(4).setString("15.5");
+      receiver.scalar(5).setString("16.25");
+      receiver.save();
+
+      receiver.start();
+      receiver.scalar("ti").setString("127");
+      receiver.scalar("si").setString("32757");
+      receiver.scalar("int").setString(Integer.toString(Integer.MAX_VALUE));
+      receiver.scalar("bi").setString(Long.toString(Long.MAX_VALUE));
+      receiver.scalar("fl").setString("10E6");
+      receiver.scalar("db").setString("10E200");
+      receiver.save();
+      return true;
+    }
+
+    @Override
+    public void close() { }
+  }
+
+  /**
+   * Test the standard string-to-type conversion using an ad-hoc conversion
+   * from the input type (the type used by the row set builder) to the output
+   * (vector) type.
+   */
+  @Test
+  public void testFixedReceiver() {
+
+    // Create the provided and output schemas
+    TupleMetadata outputSchema = new SchemaBuilder()
+        .add("ti", MinorType.TINYINT)
+        .add("si", MinorType.SMALLINT)
+        .add("int", MinorType.INT)
+        .add("bi", MinorType.BIGINT)
+        .add("fl", MinorType.FLOAT4)
+        .add("db", MinorType.FLOAT8)
+        .buildSchema();
+
+    // Create the scan
+    BaseScanFixtureBuilder builder = simpleBuilder(negotiator -> new MockReader(negotiator));
+    builder.builder.providedSchema(outputSchema);
+    ScanFixture scanFixture = builder.build();
+    ScanOperatorExec scan = scanFixture.scanOp;
+
+    // Load test data using converters
+    assertTrue(scan.next());
+
+    // Build the expected vector without a type converter.
+    final SingleRowSet expected = fixture.rowSetBuilder(outputSchema)
+        .addRow(11, 12, 13, 14L, 15.5F, 16.25D)
+        .addRow(127, 32757, Integer.MAX_VALUE, Long.MAX_VALUE, 10E6F, 10E200D)
+        .build();
+
+    // Compare
+    VectorContainer container = scan.batchAccessor().container();
+    RowSetUtilities.verify(expected, fixture.wrap(container));
+    scanFixture.close();
+  }
+}
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/v3/TestScanBasics.java b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/v3/TestScanBasics.java
index 2886d26..cba3a79 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/v3/TestScanBasics.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/v3/TestScanBasics.java
@@ -23,7 +23,7 @@ import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
 
-import org.apache.drill.categories.EvfTests;
+import org.apache.drill.categories.EvfTest;
 import org.apache.drill.common.exceptions.ExecutionSetupException;
 import org.apache.drill.common.exceptions.UserException;
 import org.apache.drill.exec.physical.impl.scan.ScanOperatorExec;
@@ -38,7 +38,7 @@ import org.slf4j.LoggerFactory;
  * Tests the basics of the scan operator protocol: error conditions,
  * etc.
  */
-@Category(EvfTests.class)
+@Category(EvfTest.class)
 public class TestScanBasics extends BaseScanTest {
   private static final Logger logger = LoggerFactory.getLogger(TestScanBasics.class);
 
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/v3/TestScanEarlySchema.java b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/v3/TestScanEarlySchema.java
index c0fbbfb..b68dc16 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/v3/TestScanEarlySchema.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/v3/TestScanEarlySchema.java
@@ -21,7 +21,7 @@ import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertTrue;
 
-import org.apache.drill.categories.EvfTests;
+import org.apache.drill.categories.EvfTest;
 import org.apache.drill.exec.physical.impl.scan.ScanOperatorExec;
 import org.apache.drill.exec.physical.rowSet.RowSet.SingleRowSet;
 import org.apache.drill.exec.physical.rowSet.RowSetBuilder;
@@ -34,7 +34,7 @@ import org.junit.experimental.categories.Category;
  * Test "early schema" readers: those that can declare a schema at
  * open time.
  */
-@Category(EvfTests.class)
+@Category(EvfTest.class)
 public class TestScanEarlySchema extends BaseScanTest {
 
   @Test
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/v3/TestScanLateSchema.java b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/v3/TestScanLateSchema.java
index cdbc3b1..cfd3d09 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/v3/TestScanLateSchema.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/v3/TestScanLateSchema.java
@@ -22,7 +22,7 @@ import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertTrue;
 
-import org.apache.drill.categories.EvfTests;
+import org.apache.drill.categories.EvfTest;
 import org.apache.drill.common.types.TypeProtos.DataMode;
 import org.apache.drill.common.types.TypeProtos.MinorType;
 import org.apache.drill.exec.physical.impl.scan.ScanOperatorExec;
@@ -40,7 +40,7 @@ import org.junit.experimental.categories.Category;
  * Test "late schema" readers: those like JSON that discover their schema
  * as they read data.
  */
-@Category(EvfTests.class)
+@Category(EvfTest.class)
 public class TestScanLateSchema extends BaseScanTest {
 
   /**
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/v3/TestScanOuputSchema.java b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/v3/TestScanOuputSchema.java
index 9d94e9f..7b14c41 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/v3/TestScanOuputSchema.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/v3/TestScanOuputSchema.java
@@ -20,7 +20,7 @@ package org.apache.drill.exec.physical.impl.scan.v3;
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertTrue;
 
-import org.apache.drill.categories.EvfTests;
+import org.apache.drill.categories.EvfTest;
 import org.apache.drill.common.types.TypeProtos.MinorType;
 import org.apache.drill.common.types.Types;
 import org.apache.drill.exec.physical.impl.scan.ScanOperatorExec;
@@ -43,7 +43,7 @@ import org.junit.experimental.categories.Category;
  * defines the schema to be output from the scan operator, and forces
  * conversions between reader and output data types.
  */
-@Category(EvfTests.class)
+@Category(EvfTest.class)
 public class TestScanOuputSchema extends BaseScanTest {
 
   private static class MockSimpleReader implements ManagedReader {
@@ -78,6 +78,7 @@ public class TestScanOuputSchema extends BaseScanTest {
     private void buildWriters(TupleMetadata providedSchema,
         TupleMetadata schema) {
       RowSetLoader rowWriter = tableLoader.writer();
+      StandardConversions conversions = StandardConversions.builder().build();
       for (int i = 0; i < schema.size(); i++) {
         ColumnMetadata colSchema = schema.metadata(i);
         String colName = colSchema.name();
@@ -92,7 +93,7 @@ public class TestScanOuputSchema extends BaseScanTest {
           writers.add(colSchema.name(), rowWriter.scalar(colIndex));
         } else {
           writers.add(colSchema.name(),
-              StandardConversions.converterFor(rowWriter.scalar(colSchema.name()), colSchema));
+              conversions.converterFor(rowWriter.scalar(colSchema.name()), colSchema));
         }
       }
     }
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/v3/TestScanOverflow.java b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/v3/TestScanOverflow.java
index efdcbbf..d813ec9 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/v3/TestScanOverflow.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/v3/TestScanOverflow.java
@@ -23,7 +23,7 @@ import static org.junit.Assert.assertTrue;
 
 import java.util.Arrays;
 
-import org.apache.drill.categories.EvfTests;
+import org.apache.drill.categories.EvfTest;
 import org.apache.drill.common.types.TypeProtos.MinorType;
 import org.apache.drill.exec.physical.impl.scan.ScanOperatorExec;
 import org.apache.drill.exec.physical.resultSet.RowSetLoader;
@@ -35,7 +35,7 @@ import org.junit.experimental.categories.Category;
 /**
  * Test vector overflow in the context of the scan operator.
  */
-@Category(EvfTests.class)
+@Category(EvfTest.class)
 public class TestScanOverflow extends BaseScanTest {
 
   /**
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/v3/file/MockFileNames.java b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/v3/file/MockFileNames.java
index ad3dc01..e35fc39 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/v3/file/MockFileNames.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/v3/file/MockFileNames.java
@@ -18,7 +18,6 @@
 package org.apache.drill.exec.physical.impl.scan.v3.file;
 
 import org.apache.drill.common.types.TypeProtos.MajorType;
-import org.apache.drill.exec.physical.impl.scan.v3.schema.ImplicitColumnResolver;
 import org.apache.hadoop.fs.Path;
 
 public interface MockFileNames {
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/v3/file/TestFileDescrip.java b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/v3/file/TestFileDescrip.java
index 24563c1..0dfdd9e 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/v3/file/TestFileDescrip.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/v3/file/TestFileDescrip.java
@@ -18,45 +18,41 @@
 package org.apache.drill.exec.physical.impl.scan.v3.file;
 
 import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertSame;
-import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
 
-import org.apache.drill.categories.EvfTests;
+import java.io.IOException;
+
+import org.apache.drill.categories.EvfTest;
+import org.apache.drill.exec.store.dfs.DrillFileSystem;
+import org.apache.drill.exec.store.schedule.CompleteFileWork.FileWorkImpl;
 import org.apache.drill.test.BaseTest;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
 import org.junit.Test;
 import org.junit.experimental.categories.Category;
 
-@Category(EvfTests.class)
+@Category(EvfTest.class)
 public class TestFileDescrip extends BaseTest {
 
-  /**
-   * Degenerate case: no file or root
-   */
-  @Test
-  public void testEmpty() {
-    FileDescrip fd = new FileDescrip(null, null);
-    assertFalse(fd.isSet());
-    assertNull(fd.filePath());
-    assertEquals(0, fd.dirPathLength());
-    assertNull(fd.partition(0));
+  private static final DrillFileSystem dfs;
+
+  static {
+    Configuration conf = new Configuration();
+    conf.set(FileSystem.FS_DEFAULT_NAME_KEY, FileSystem.DEFAULT_FS);
+    try {
+      dfs = new DrillFileSystem(conf);
+    } catch (IOException e) {
+      throw new IllegalStateException(e);
+    }
   }
 
-  /**
-   * Degenerate case: no file path, but with as selection root
-   * Should never occur in practice.
-   */
-  @Test
-  public void testNoPath() {
-    Path root = new Path("hdfs://a/b");
-    FileDescrip fd = new FileDescrip(null, root);
-    assertFalse(fd.isSet());
-    assertNull(fd.filePath());
-    assertEquals(0, fd.dirPathLength());
-    assertNull(fd.partition(0));
+  private FileDescrip fileDescrip(Path input, Path root) {
+    return new FileDescrip(dfs,
+        new FileWorkImpl(0, 1000, input),
+        root);
   }
 
   /**
@@ -65,9 +61,8 @@ public class TestFileDescrip extends BaseTest {
    */
   @Test
   public void testNoRoot() {
-    Path input = new Path("hdfs://foo.csv");
-    FileDescrip fd = new FileDescrip(input, null);
-    assertTrue(fd.isSet());
+    Path input = new Path("file:///foo.csv");
+    FileDescrip fd = fileDescrip(input, null);
     assertSame(input, fd.filePath());
     assertEquals(0, fd.dirPathLength());
     assertNull(fd.partition(0));
@@ -78,9 +73,8 @@ public class TestFileDescrip extends BaseTest {
    */
   @Test
   public void testSingleFile() {
-    Path input = new Path("hdfs://a/b/c/foo.csv");
-    FileDescrip fd = new FileDescrip(input, null);
-    assertTrue(fd.isSet());
+    Path input = new Path("file:///a/b/c/foo.csv");
+    FileDescrip fd = fileDescrip(input, null);
     assertSame(input, fd.filePath());
     assertEquals(0, fd.dirPathLength());
     assertNull(fd.partition(0));
@@ -91,10 +85,9 @@ public class TestFileDescrip extends BaseTest {
    */
   @Test
   public void testRootFile() {
-    Path root = new Path("hdfs://a/b");
-    Path input = new Path("hdfs://a/b/foo.csv");
-    FileDescrip fd = new FileDescrip(input, root);
-    assertTrue(fd.isSet());
+    Path root = new Path("file:///a/b");
+    Path input = new Path("file:///a/b/foo.csv");
+    FileDescrip fd = fileDescrip(input, root);
     assertSame(input, fd.filePath());
     assertEquals(0, fd.dirPathLength());
     assertNull(fd.partition(0));
@@ -105,10 +98,9 @@ public class TestFileDescrip extends BaseTest {
    */
   @Test
   public void testBelowRoot() {
-    Path root = new Path("hdfs://a/b");
-    Path input = new Path("hdfs://a/b/c/foo.csv");
-    FileDescrip fd = new FileDescrip(input, root);
-    assertTrue(fd.isSet());
+    Path root = new Path("file:///a/b");
+    Path input = new Path("file:///a/b/c/foo.csv");
+    FileDescrip fd = fileDescrip(input, root);
     assertSame(input, fd.filePath());
     assertEquals(1, fd.dirPathLength());
     assertEquals("c", fd.partition(0));
@@ -121,10 +113,10 @@ public class TestFileDescrip extends BaseTest {
    */
   @Test
   public void testAboveRoot() {
-    Path root = new Path("hdfs://a/b");
-    Path input = new Path("hdfs://a/foo.csv");
+    Path root = new Path("file:///a/b");
+    Path input = new Path("file:///a/foo.csv");
     try {
-      new FileDescrip(input, root);
+      fileDescrip(input, root);
       fail();
     } catch (IllegalArgumentException e) {
       // Expected
@@ -137,10 +129,10 @@ public class TestFileDescrip extends BaseTest {
    */
   @Test
   public void testDisjointPath() {
-    Path root = new Path("hdfs://a/b");
-    Path input = new Path("hdfs://d/foo.csv");
+    Path root = new Path("file:///a/b");
+    Path input = new Path("file:///d/foo.csv");
     try {
-      new FileDescrip(input, root);
+      fileDescrip(input, root);
       fail();
     } catch (IllegalArgumentException e) {
       // Expected
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/v3/file/TestFileScan.java b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/v3/file/TestFileScan.java
index 1f45f6b..9f7d4c1 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/v3/file/TestFileScan.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/v3/file/TestFileScan.java
@@ -21,7 +21,7 @@ import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertTrue;
 
-import org.apache.drill.categories.EvfTests;
+import org.apache.drill.categories.EvfTest;
 import org.apache.drill.common.types.TypeProtos.DataMode;
 import org.apache.drill.common.types.TypeProtos.MinorType;
 import org.apache.drill.common.types.Types;
@@ -43,7 +43,7 @@ import org.junit.experimental.categories.Category;
  * Focuses on the file implicit columns, assumes that other tests have
  * verified the underlying mechanisms.
  */
-@Category(EvfTests.class)
+@Category(EvfTest.class)
 public class TestFileScan extends BaseFileScanTest {
 
   /**
@@ -57,8 +57,9 @@ public class TestFileScan extends BaseFileScanTest {
     public MockLateSchemaReader(FileSchemaNegotiator negotiator) {
 
       // Verify the path
-      assertEquals(MOCK_FILE_SYSTEM_NAME, negotiator.fileWork().getPath().toString());
-      assertEquals(MOCK_FILE_SYSTEM_NAME, negotiator.split().getPath().toString());
+      assertEquals(MOCK_FILE_SYSTEM_NAME, negotiator.file().filePath().toString());
+      assertEquals(MOCK_FILE_SYSTEM_NAME, negotiator.file().fileWork().getPath().toString());
+      assertEquals(MOCK_FILE_SYSTEM_NAME, negotiator.file().split().getPath().toString());
 
       // No schema or file, just build the table loader.
       tableLoader = negotiator.build();
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/v3/file/TestFileScanLifecycle.java b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/v3/file/TestFileScanLifecycle.java
index 2b6f0cf..00931db 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/v3/file/TestFileScanLifecycle.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/v3/file/TestFileScanLifecycle.java
@@ -25,7 +25,7 @@ import static org.junit.Assert.fail;
 
 import java.util.Collections;
 
-import org.apache.drill.categories.EvfTests;
+import org.apache.drill.categories.EvfTest;
 import org.apache.drill.common.exceptions.UserException;
 import org.apache.drill.common.expression.SchemaPath;
 import org.apache.drill.exec.physical.impl.scan.RowBatchReader;
@@ -43,7 +43,7 @@ import org.apache.hadoop.fs.Path;
 import org.junit.Test;
 import org.junit.experimental.categories.Category;
 
-@Category(EvfTests.class)
+@Category(EvfTest.class)
 public class TestFileScanLifecycle extends BaseTestScanLifecycle implements MockFileNames {
 
   /**
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/v3/file/TestImplicitColumnLoader.java b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/v3/file/TestImplicitColumnLoader.java
index 4b64ef5..e064a20 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/v3/file/TestImplicitColumnLoader.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/v3/file/TestImplicitColumnLoader.java
@@ -20,13 +20,14 @@ package org.apache.drill.exec.physical.impl.scan.v3.file;
 import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertNull;
 
+import java.io.IOException;
 import java.util.List;
 
-import org.apache.drill.categories.EvfTests;
+import org.apache.drill.categories.EvfTest;
 import org.apache.drill.common.expression.SchemaPath;
+import org.apache.drill.common.types.TypeProtos.MinorType;
 import org.apache.drill.exec.physical.impl.scan.ScanTestUtils;
 import org.apache.drill.exec.physical.impl.scan.v3.lifecycle.StaticBatchBuilder;
-import org.apache.drill.exec.physical.impl.scan.v3.lifecycle.BaseTestScanLifecycle.DummySubScan;
 import org.apache.drill.exec.physical.impl.scan.v3.schema.ScanSchemaConfigBuilder;
 import org.apache.drill.exec.physical.impl.scan.v3.schema.ScanSchemaTracker;
 import org.apache.drill.exec.physical.resultSet.ResultVectorCache;
@@ -35,8 +36,13 @@ import org.apache.drill.exec.physical.rowSet.RowSet;
 import org.apache.drill.exec.physical.rowSet.RowSetTestUtils;
 import org.apache.drill.exec.record.metadata.SchemaBuilder;
 import org.apache.drill.exec.record.metadata.TupleMetadata;
+import org.apache.drill.exec.store.dfs.DrillFileSystem;
+import org.apache.drill.exec.store.dfs.easy.FileWork;
+import org.apache.drill.exec.store.schedule.CompleteFileWork.FileWorkImpl;
 import org.apache.drill.test.SubOperatorTest;
 import org.apache.drill.test.rowSet.RowSetUtilities;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
 import org.junit.Test;
 import org.junit.experimental.categories.Category;
@@ -45,19 +51,51 @@ import org.junit.experimental.categories.Category;
  * Tests the file implicit column handler which identifies implicit columns
  * and populates them. Assumes that the implicit column parser tests pass.
  */
-@Category(EvfTests.class)
+@Category(EvfTest.class)
 public class TestImplicitColumnLoader extends SubOperatorTest implements MockFileNames {
 
+  private static final DrillFileSystem dfs;
+
+  static {
+    Configuration conf = new Configuration();
+    conf.set(FileSystem.FS_DEFAULT_NAME_KEY, FileSystem.DEFAULT_FS);
+    try {
+      dfs = new DrillFileSystem(conf);
+    } catch (IOException e) {
+      throw new IllegalStateException(e);
+    }
+  }
+
+  public static class ImplicitFixture {
+    final FileScanLifecycleBuilder options;
+    final ScanSchemaTracker schemaTracker;
+    FileDescrip fileDescrip;
+    ImplicitFileColumnsHandler handler;
+
+    public ImplicitFixture(List<SchemaPath> projList, Path root) {
+      this.options = new FileScanLifecycleBuilder();
+      options.rootDir(root);
+      this.schemaTracker = new ScanSchemaConfigBuilder()
+          .projection(projList)
+          .build();
+    }
+
+    public void build(Path input) {
+      final ResultVectorCache cache = new NullResultVectorCacheImpl(fixture.allocator());
+      handler = new ImplicitFileColumnsHandler(
+          dfs, fixture.getOptionManager(), options, cache, schemaTracker);
+      FileWork fileWork = new FileWorkImpl(0, 1000, input);
+      fileDescrip = handler.makeDescrip(fileWork);
+    }
+
+    public StaticBatchBuilder batchBuilder() {
+      return handler.forFile(fileDescrip);
+    }
+  }
   public StaticBatchBuilder buildHandler(List<SchemaPath> projList, Path root, Path input) {
-    FileScanLifecycleBuilder options = new FileScanLifecycleBuilder();
-    options.rootDir(root);
-    final ResultVectorCache cache = new NullResultVectorCacheImpl(fixture.allocator());
-    final ScanSchemaTracker schemaTracker = new ScanSchemaConfigBuilder()
-        .projection(projList)
-        .build();
-    ImplicitFileColumnsHandler handler = new ImplicitFileColumnsHandler(
-        fixture.operatorContext(new DummySubScan()), options, cache, schemaTracker);
-    return handler.forFile(input);
+    ImplicitFixture fixture = new ImplicitFixture(projList, root);
+    fixture.build(input);
+    return fixture.batchBuilder();
   }
 
   @Test
@@ -84,9 +122,9 @@ public class TestImplicitColumnLoader extends SubOperatorTest implements MockFil
   }
 
   @Test
-  public void testAllColumns() {
+  public void testNonInternalColumns() {
     StaticBatchBuilder batchLoader = buildHandler(
-        ScanTestUtils.projectAllWithMetadata(3),
+        ScanTestUtils.projectAllWithFileImplicit(3),
         MOCK_ROOT_PATH, MOCK_FILE_PATH);
     assertNotNull(batchLoader);
     batchLoader.load(2);
@@ -106,4 +144,59 @@ public class TestImplicitColumnLoader extends SubOperatorTest implements MockFil
         .build();
     RowSetUtilities.verify(expected, fixture.wrap(batchLoader.outputContainer()));
   }
+
+  @Test
+  public void testInternalColumns() {
+    ImplicitFixture testFixture = new ImplicitFixture(
+        RowSetTestUtils.projectList(
+            ScanTestUtils.LAST_MODIFIED_TIME_COL,
+            ScanTestUtils.PROJECT_METADATA_COL,
+            ScanTestUtils.ROW_GROUP_INDEX_COL,
+            ScanTestUtils.ROW_GROUP_START_COL,
+            ScanTestUtils.ROW_GROUP_LENGTH_COL),
+        MOCK_ROOT_PATH);
+    testFixture.build(MOCK_FILE_PATH);
+    testFixture.fileDescrip.setRowGroupAttribs(10, 10_000, 5_000);
+    testFixture.fileDescrip.setModTime("123456789");
+    StaticBatchBuilder batchLoader = testFixture.batchBuilder();
+    assertNotNull(batchLoader);
+    batchLoader.load(2);
+
+    TupleMetadata expectedSchema = new SchemaBuilder()
+        .add(ScanTestUtils.LAST_MODIFIED_TIME_COL, MinorType.VARCHAR)
+        .addNullable(ScanTestUtils.PROJECT_METADATA_COL, MinorType.VARCHAR)
+        .add(ScanTestUtils.ROW_GROUP_INDEX_COL, MinorType.VARCHAR)
+        .add(ScanTestUtils.ROW_GROUP_START_COL, MinorType.VARCHAR)
+        .add(ScanTestUtils.ROW_GROUP_LENGTH_COL, MinorType.VARCHAR)
+        .build();
+    RowSet expected = fixture.rowSetBuilder(expectedSchema)
+        .addRow("123456789", null, "10", "10000", "5000")
+        .addRow("123456789", null, "10", "10000", "5000")
+        .build();
+    RowSetUtilities.verify(expected, fixture.wrap(batchLoader.outputContainer()));
+  }
+
+  @Test
+  public void testInternalEmptyFile() {
+    ImplicitFixture testFixture = new ImplicitFixture(
+        RowSetTestUtils.projectList(
+            ScanTestUtils.LAST_MODIFIED_TIME_COL,
+            ScanTestUtils.PROJECT_METADATA_COL),
+        MOCK_ROOT_PATH);
+    testFixture.build(MOCK_FILE_PATH);
+    testFixture.fileDescrip.setModTime("123456789");
+    testFixture.fileDescrip.markEmpty();
+    StaticBatchBuilder batchLoader = testFixture.batchBuilder();
+    assertNotNull(batchLoader);
+    batchLoader.load(1);
+
+    TupleMetadata expectedSchema = new SchemaBuilder()
+        .add(ScanTestUtils.LAST_MODIFIED_TIME_COL, MinorType.VARCHAR)
+        .addNullable(ScanTestUtils.PROJECT_METADATA_COL, MinorType.VARCHAR)
+        .build();
+    RowSet expected = fixture.rowSetBuilder(expectedSchema)
+        .addRow("123456789", "false")
+        .build();
+    RowSetUtilities.verify(expected, fixture.wrap(batchLoader.outputContainer()));
+  }
 }
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/v3/file/TestImplicitColumnResolver.java b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/v3/file/TestImplicitColumnResolver.java
index a87e330..f10083d 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/v3/file/TestImplicitColumnResolver.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/v3/file/TestImplicitColumnResolver.java
@@ -22,21 +22,20 @@ import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertSame;
 import static org.junit.Assert.assertTrue;
 
+import java.io.IOException;
 import java.util.Collection;
 import java.util.List;
 
-import org.apache.drill.categories.EvfTests;
+import org.apache.drill.categories.EvfTest;
 import org.apache.drill.common.exceptions.CustomErrorContext;
 import org.apache.drill.common.exceptions.EmptyErrorContext;
 import org.apache.drill.common.exceptions.UserException;
 import org.apache.drill.common.expression.SchemaPath;
 import org.apache.drill.common.types.TypeProtos.MinorType;
 import org.apache.drill.exec.physical.impl.scan.ScanTestUtils;
-import org.apache.drill.exec.physical.impl.scan.v3.schema.ImplicitColumnResolver;
-import org.apache.drill.exec.physical.impl.scan.v3.schema.ImplicitColumnResolver.ImplicitColumnOptions;
-import org.apache.drill.exec.physical.impl.scan.v3.schema.ImplicitColumnResolver.ParseResult;
-import org.apache.drill.exec.physical.impl.scan.v3.schema.MutableTupleSchema;
 import org.apache.drill.exec.physical.impl.scan.v3.schema.MutableTupleSchema.ColumnHandle;
+import org.apache.drill.exec.physical.impl.scan.v3.file.ImplicitColumnResolver.ImplicitColumnOptions;
+import org.apache.drill.exec.physical.impl.scan.v3.file.ImplicitColumnResolver.ParseResult;
 import org.apache.drill.exec.physical.impl.scan.v3.schema.ProjectionSchemaTracker;
 import org.apache.drill.exec.physical.impl.scan.v3.schema.ScanProjectionParser;
 import org.apache.drill.exec.physical.impl.scan.v3.schema.ScanProjectionParser.ProjectionParseResult;
@@ -49,11 +48,14 @@ import org.apache.drill.exec.physical.rowSet.RowSetTestUtils;
 import org.apache.drill.exec.record.metadata.ColumnMetadata;
 import org.apache.drill.exec.record.metadata.SchemaBuilder;
 import org.apache.drill.exec.record.metadata.TupleMetadata;
+import org.apache.drill.exec.store.dfs.DrillFileSystem;
 import org.apache.drill.test.SubOperatorTest;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
 import org.junit.Test;
 import org.junit.experimental.categories.Category;
 
-@Category(EvfTests.class)
+@Category(EvfTest.class)
 public class TestImplicitColumnResolver extends SubOperatorTest {
 
   private static final CustomErrorContext ERROR_CONTEXT = EmptyErrorContext.INSTANCE;
@@ -77,7 +79,7 @@ public class TestImplicitColumnResolver extends SubOperatorTest {
   }
 
   private boolean isImplicit(List<ColumnHandle> cols, int index) {
-    return SchemaUtils.isImplicit(cols.get(index).column());
+    return cols.get(index).isImplicit();
   }
 
   @Test
@@ -87,6 +89,7 @@ public class TestImplicitColumnResolver extends SubOperatorTest {
     ParseResult result = parseFixture.parseImplicit();
     assertTrue(result.columns().isEmpty());
     assertTrue(result.schema().isEmpty());
+    assertFalse(result.isMetadataScan());
   }
 
   /**
@@ -103,6 +106,7 @@ public class TestImplicitColumnResolver extends SubOperatorTest {
             ScanTestUtils.SUFFIX_COL));
     ParseResult result = parseFixture.parseImplicit();
 
+    assertFalse(result.isMetadataScan());
     assertEquals(4, result.columns().size());
 
     TupleMetadata expected = new SchemaBuilder()
@@ -132,6 +136,7 @@ public class TestImplicitColumnResolver extends SubOperatorTest {
     ParserFixture parseFixture = new ParserFixture(
         RowSetTestUtils.projectList(dir2, dir1, dir0, "a"));
     ParseResult result = parseFixture.parseImplicit();
+    assertFalse(result.isMetadataScan());
     assertEquals(3, result.columns().size());
 
     TupleMetadata expected = new SchemaBuilder()
@@ -156,6 +161,7 @@ public class TestImplicitColumnResolver extends SubOperatorTest {
         .maxPartitionDepth(3)
         .useLegacyWildcardExpansion(true);
     ParseResult result = parseFixture.parseImplicit();
+    assertFalse(result.isMetadataScan());
     assertEquals(3, result.columns().size());
 
     TupleMetadata expected = new SchemaBuilder()
@@ -535,12 +541,6 @@ public class TestImplicitColumnResolver extends SubOperatorTest {
     assertTrue(schemaTracker.isResolved());
     assertSame(ProjectionType.SOME, schemaTracker.projectionType());
 
-    // Implicit columns should be marked
-    MutableTupleSchema internalSchema = schemaTracker.internalSchema();
-    assertFalse(internalSchema.find("a").isImplicit());
-    assertTrue(internalSchema.find(ScanTestUtils.FILE_NAME_COL).isImplicit());
-    assertTrue(internalSchema.find(ScanTestUtils.partitionColName(2)).isImplicit());
-
     ImplicitColumnOptions options = new ImplicitColumnOptions()
         .optionSet(fixture.getOptionManager());
     ImplicitColumnResolver parser = new ImplicitColumnResolver(options, ERROR_CONTEXT);
@@ -552,4 +552,62 @@ public class TestImplicitColumnResolver extends SubOperatorTest {
 
     assertEquals(definedSchema, schemaTracker.outputSchema());
   }
+
+  /**
+   * Test including internal implicit columns in the project list.
+   * @throws IOException
+   */
+  @Test
+  public void testInternalImplicitColumnSelection() throws IOException {
+    // Simulate SELECT lmt, $project_metadata$", rgi, rgs, rgl ...
+    Configuration conf = new Configuration();
+    conf.set(FileSystem.FS_DEFAULT_NAME_KEY, FileSystem.DEFAULT_FS);
+    DrillFileSystem dfs = new DrillFileSystem(conf);
+    ParserFixture parseFixture = new ParserFixture(
+        RowSetTestUtils.projectList("a",
+            ScanTestUtils.LAST_MODIFIED_TIME_COL,
+            ScanTestUtils.PROJECT_METADATA_COL,
+            ScanTestUtils.ROW_GROUP_INDEX_COL,
+            ScanTestUtils.ROW_GROUP_START_COL,
+            ScanTestUtils.ROW_GROUP_LENGTH_COL));
+    parseFixture.options.dfs(dfs);
+    ParseResult result = parseFixture.parseImplicit();
+
+    assertTrue(result.isMetadataScan());
+    assertEquals(5, result.columns().size());
+
+    TupleMetadata expected = new SchemaBuilder()
+        .add(ScanTestUtils.LAST_MODIFIED_TIME_COL, MinorType.VARCHAR)
+        .addNullable(ScanTestUtils.PROJECT_METADATA_COL, MinorType.VARCHAR)
+        .add(ScanTestUtils.ROW_GROUP_INDEX_COL, MinorType.VARCHAR)
+        .add(ScanTestUtils.ROW_GROUP_START_COL, MinorType.VARCHAR)
+        .add(ScanTestUtils.ROW_GROUP_LENGTH_COL, MinorType.VARCHAR)
+        .build();
+    assertEquals(expected, result.schema());
+
+    List<ColumnHandle> cols = parseFixture.tracker.internalSchema().columns();
+    assertFalse(isImplicit(cols, 0));
+    assertTrue(isImplicit(cols, 1));
+    assertTrue(isImplicit(cols, 2));
+    assertTrue(isImplicit(cols, 3));
+    assertTrue(isImplicit(cols, 4));
+    assertTrue(isImplicit(cols, 5));
+  }
+
+  @Test
+  public void testProvidedImplicitColInternal() {
+    TupleMetadata providedSchema = new SchemaBuilder()
+        .add("myLmt", MinorType.INT)
+        .build();
+    SchemaUtils.markImplicit(providedSchema.metadata("myLmt"), ScanTestUtils.LAST_MODIFIED_TIME_COL);
+
+    ParserFixture parseFixture = new ParserFixture(
+        RowSetTestUtils.projectAll());
+    parseFixture.tracker.applyProvidedSchema(providedSchema);
+    try {
+      parseFixture.parseImplicit();
+    } catch (UserException e) {
+      assertTrue(e.getMessage().contains("references an undefined implicit column type"));
+    }
+  }
 }
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/v3/lifecycle/BaseTestScanLifecycle.java b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/v3/lifecycle/BaseTestScanLifecycle.java
index 5468175..66f10fc 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/v3/lifecycle/BaseTestScanLifecycle.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/v3/lifecycle/BaseTestScanLifecycle.java
@@ -135,7 +135,6 @@ public class BaseTestScanLifecycle extends SubOperatorTest {
     public void close() {}
   }
 
-
   /**
    * Mock reader with no data or schema, indicated by an early EOF
    * exception.
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/v3/lifecycle/TestMissingColumnLoader.java b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/v3/lifecycle/TestMissingColumnLoader.java
index 7e12260..b6becd0 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/v3/lifecycle/TestMissingColumnLoader.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/v3/lifecycle/TestMissingColumnLoader.java
@@ -21,7 +21,7 @@ import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertSame;
 
-import org.apache.drill.categories.EvfTests;
+import org.apache.drill.categories.EvfTest;
 import org.apache.drill.common.types.TypeProtos.DataMode;
 import org.apache.drill.common.types.TypeProtos.MajorType;
 import org.apache.drill.common.types.TypeProtos.MinorType;
@@ -49,7 +49,7 @@ import org.junit.experimental.categories.Category;
  * can create the classic nullable Int null column, or one of
  * any other type and mode.
  */
-@Category(EvfTests.class)
+@Category(EvfTest.class)
 public class TestMissingColumnLoader extends SubOperatorTest {
 
   /**
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/v3/lifecycle/TestOutputBatchBuilder.java b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/v3/lifecycle/TestOutputBatchBuilder.java
index 0ec3196..ff83b1d 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/v3/lifecycle/TestOutputBatchBuilder.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/v3/lifecycle/TestOutputBatchBuilder.java
@@ -22,7 +22,7 @@ import static org.apache.drill.test.rowSet.RowSetUtilities.mapValue;
 import java.util.Arrays;
 import java.util.Collections;
 
-import org.apache.drill.categories.EvfTests;
+import org.apache.drill.categories.EvfTest;
 import org.apache.drill.common.types.TypeProtos.MinorType;
 import org.apache.drill.exec.physical.impl.scan.v3.lifecycle.OutputBatchBuilder.BatchSource;
 import org.apache.drill.exec.physical.rowSet.RowSet;
@@ -36,7 +36,7 @@ import org.apache.drill.test.rowSet.RowSetUtilities;
 import org.junit.Test;
 import org.junit.experimental.categories.Category;
 
-@Category(EvfTests.class)
+@Category(EvfTest.class)
 public class TestOutputBatchBuilder extends SubOperatorTest {
 
   public TupleMetadata firstSchema() {
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/v3/lifecycle/TestReaderErrors.java b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/v3/lifecycle/TestReaderErrors.java
index 1500295..7809904 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/v3/lifecycle/TestReaderErrors.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/v3/lifecycle/TestReaderErrors.java
@@ -22,7 +22,7 @@ import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
 
-import org.apache.drill.categories.EvfTests;
+import org.apache.drill.categories.EvfTest;
 import org.apache.drill.common.exceptions.UserException;
 import org.apache.drill.exec.physical.impl.scan.RowBatchReader;
 import org.apache.drill.exec.physical.impl.scan.v3.ManagedReader;
@@ -35,7 +35,7 @@ import org.junit.experimental.categories.Category;
  * Verifies proper handling of errors from a reader, including use of the
  * scan and reader error contexts.
  */
-@Category(EvfTests.class)
+@Category(EvfTest.class)
 public class TestReaderErrors extends BaseTestScanLifecycle {
 
   @Test
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/v3/lifecycle/TestScanLifecycleBasics.java b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/v3/lifecycle/TestScanLifecycleBasics.java
index 75700c1..24a083d 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/v3/lifecycle/TestScanLifecycleBasics.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/v3/lifecycle/TestScanLifecycleBasics.java
@@ -23,7 +23,7 @@ import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertSame;
 import static org.junit.Assert.assertTrue;
 
-import org.apache.drill.categories.EvfTests;
+import org.apache.drill.categories.EvfTest;
 import org.apache.drill.common.types.TypeProtos.MinorType;
 import org.apache.drill.common.types.Types;
 import org.apache.drill.exec.physical.impl.scan.RowBatchReader;
@@ -34,13 +34,14 @@ import org.apache.drill.exec.physical.impl.scan.v3.SchemaNegotiator;
 import org.apache.drill.exec.physical.impl.scan.v3.schema.ScanSchemaTracker.ProjectionType;
 import org.apache.drill.exec.physical.rowSet.RowSet;
 import org.apache.drill.exec.physical.rowSet.RowSetTestUtils;
+import org.apache.drill.exec.record.VectorContainer;
 import org.apache.drill.exec.record.metadata.SchemaBuilder;
 import org.apache.drill.exec.record.metadata.TupleMetadata;
 import org.apache.drill.test.rowSet.RowSetUtilities;
 import org.junit.Test;
 import org.junit.experimental.categories.Category;
 
-@Category(EvfTests.class)
+@Category(EvfTest.class)
 public class TestScanLifecycleBasics extends BaseTestScanLifecycle {
 
   /**
@@ -122,11 +123,17 @@ public class TestScanLifecycleBasics extends BaseTestScanLifecycle {
     // Early schema: so output schema is available after open
     assertTrue(scan.hasOutputSchema());
     assertEquals(SCHEMA, scan.outputSchema());
-    assertFalse(reader.next());
-    reader.close();
+    assertTrue(reader.next());
+    VectorContainer result = reader.output();
+    assertEquals(0, result.getRecordCount());
+    result.zeroVectors();
 
     // Early schema with no additional columns discovered
     assertEquals(SCHEMA, scan.outputSchema());
+
+    // But, no second batch.
+    assertFalse(reader.next());
+    reader.close();
     scan.close();
   }
 
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/v3/lifecycle/TestScanLifecycleSchema.java b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/v3/lifecycle/TestScanLifecycleSchema.java
index 3e3b833..334753d 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/v3/lifecycle/TestScanLifecycleSchema.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/v3/lifecycle/TestScanLifecycleSchema.java
@@ -23,7 +23,7 @@ import static org.junit.Assert.assertSame;
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
 
-import org.apache.drill.categories.EvfTests;
+import org.apache.drill.categories.EvfTest;
 import org.apache.drill.common.exceptions.UserException;
 import org.apache.drill.exec.physical.impl.scan.RowBatchReader;
 import org.apache.drill.exec.physical.impl.scan.v3.ManagedReader;
@@ -37,7 +37,7 @@ import org.apache.drill.test.rowSet.RowSetUtilities;
 import org.junit.Test;
 import org.junit.experimental.categories.Category;
 
-@Category(EvfTests.class)
+@Category(EvfTest.class)
 public class TestScanLifecycleSchema extends BaseTestScanLifecycle {
 
   /**
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/v3/lifecycle/TestScanLifecycleTwoReaders.java b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/v3/lifecycle/TestScanLifecycleTwoReaders.java
index 9815b94..e9c571d 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/v3/lifecycle/TestScanLifecycleTwoReaders.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/v3/lifecycle/TestScanLifecycleTwoReaders.java
@@ -21,7 +21,7 @@ import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
 
-import org.apache.drill.categories.EvfTests;
+import org.apache.drill.categories.EvfTest;
 import org.apache.drill.common.exceptions.UserException;
 import org.apache.drill.exec.physical.impl.scan.RowBatchReader;
 import org.apache.drill.exec.physical.impl.scan.v3.ManagedReader;
@@ -37,7 +37,7 @@ import org.junit.experimental.categories.Category;
  * Test two readers in succession in various cases: empty readers, normal readers,
  * type conflicts, etc.
  */
-@Category(EvfTests.class)
+@Category(EvfTest.class)
 public class TestScanLifecycleTwoReaders extends BaseTestScanLifecycle {
 
   /**
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/v3/schema/TestDynamicSchemaFilter.java b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/v3/schema/TestDynamicSchemaFilter.java
index 4dcc4ec..b7fa328 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/v3/schema/TestDynamicSchemaFilter.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/v3/schema/TestDynamicSchemaFilter.java
@@ -24,7 +24,7 @@ import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertSame;
 import static org.junit.Assert.assertTrue;
 
-import org.apache.drill.categories.EvfTests;
+import org.apache.drill.categories.EvfTest;
 import org.apache.drill.common.exceptions.EmptyErrorContext;
 import org.apache.drill.common.types.TypeProtos.MinorType;
 import org.apache.drill.common.types.Types;
@@ -39,7 +39,7 @@ import org.apache.drill.exec.record.metadata.SchemaBuilder;
 import org.junit.Test;
 import org.junit.experimental.categories.Category;
 
-@Category(EvfTests.class)
+@Category(EvfTest.class)
 public class TestDynamicSchemaFilter {
 
   private static final ColumnMetadata A_COL =
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/v3/schema/TestProjectedPath.java b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/v3/schema/TestProjectedPath.java
index 491be8b..57277b5 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/v3/schema/TestProjectedPath.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/v3/schema/TestProjectedPath.java
@@ -20,7 +20,7 @@ package org.apache.drill.exec.physical.impl.scan.v3.schema;
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertTrue;
 
-import org.apache.drill.categories.EvfTests;
+import org.apache.drill.categories.EvfTest;
 import org.apache.drill.common.types.TypeProtos.MinorType;
 import org.apache.drill.common.types.Types;
 import org.apache.drill.exec.physical.impl.scan.v3.schema.ScanProjectionParser.ProjectionParseResult;
@@ -37,7 +37,7 @@ import org.junit.experimental.categories.Category;
  * to see if the projection path is consistent with the type. Tests here
  * verify the consistency checks.
  */
-@Category(EvfTests.class)
+@Category(EvfTest.class)
 public class TestProjectedPath {
 
   // INT is a proxy for all scalar columns.
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/v3/schema/TestProjectionParser.java b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/v3/schema/TestProjectionParser.java
index 02d05a1..d67d7e6 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/v3/schema/TestProjectionParser.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/v3/schema/TestProjectionParser.java
@@ -27,7 +27,7 @@ import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
 
-import org.apache.drill.categories.EvfTests;
+import org.apache.drill.categories.EvfTest;
 import org.apache.drill.common.expression.SchemaPath;
 import org.apache.drill.exec.physical.impl.scan.v3.schema.ScanProjectionParser.ProjectionParseResult;
 import org.apache.drill.exec.physical.rowSet.RowSetTestUtils;
@@ -46,7 +46,7 @@ import org.junit.experimental.categories.Category;
  * parsing; the only bits not tested here is that which is
  * inherently specific to some use case.
  */
-@Category(EvfTests.class)
+@Category(EvfTest.class)
 public class TestProjectionParser extends BaseTest {
 
   /**
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/v3/schema/TestScanSchemaTracker.java b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/v3/schema/TestScanSchemaTracker.java
index 6ccae71..075e774 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/v3/schema/TestScanSchemaTracker.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/v3/schema/TestScanSchemaTracker.java
@@ -23,10 +23,11 @@ import static org.junit.Assert.assertSame;
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
 
-import org.apache.drill.categories.EvfTests;
+import org.apache.drill.categories.EvfTest;
 import org.apache.drill.common.exceptions.CustomErrorContext;
 import org.apache.drill.common.exceptions.EmptyErrorContext;
 import org.apache.drill.common.exceptions.UserException;
+import org.apache.drill.common.expression.SchemaPath;
 import org.apache.drill.common.types.TypeProtos.MinorType;
 import org.apache.drill.common.types.Types;
 import org.apache.drill.exec.physical.impl.scan.v3.schema.ScanSchemaTracker.ProjectionType;
@@ -43,7 +44,7 @@ import org.junit.experimental.categories.Category;
  * Test the scan operator schema tracker which computes the final
  * output schema from a variety of sources.
  */
-@Category(EvfTests.class)
+@Category(EvfTest.class)
 public class TestScanSchemaTracker extends BaseTest {
   private static final CustomErrorContext ERROR_CONTEXT = EmptyErrorContext.INSTANCE;
 
@@ -294,6 +295,38 @@ public class TestScanSchemaTracker extends BaseTest {
     assertEquals(expected, outputSchema);
   }
 
+  /**
+   * Test for a query of the form:<br>
+   * {code SELECT * FROM t ORDER BY a}<br>
+   * in which we get a projection list of the form<br<
+   * {@code [`**`, `a`]<br>
+   * If we are given a provided schema of {@code (a, b, c)},
+   * the "natural" expansion will be @{code (b, c, a)}, but we
+   * add a hack to get what the user expects: @{code (a, b, c)}.
+   * The "natural" expansion occurs because the projection list says
+   * "all all columns to the wildcard position except those mentioned
+   * elsewhere". We essentially redefine it as "add or move all columns
+   * in provided schema order."
+   */
+  @Test
+  public void testWildcardWithExplicitWithProvided() {
+
+    // Simulate SELECT *, a ...
+    final ScanSchemaConfigBuilder builder = new ScanSchemaConfigBuilder()
+        .projection(RowSetTestUtils.projectList(
+            SchemaPath.DYNAMIC_STAR, "a"));
+
+    final TupleMetadata providedSchema = new SchemaBuilder()
+        .add("a", MinorType.INT)
+        .add("b", MinorType.BIGINT)
+        .add("c", MinorType.VARCHAR)
+        .buildSchema();
+    builder.providedSchema(providedSchema);
+    final ScanSchemaTracker schemaTracker = builder.build();
+    assertTrue(schemaTracker.isResolved());
+    assertEquals(providedSchema, schemaTracker.outputSchema());
+  }
+
   @Test
   public void testStrictProvidedSchemaWithWildcard() {
 
@@ -467,4 +500,57 @@ public class TestScanSchemaTracker extends BaseTest {
     final TupleMetadata outputSchema = schemaTracker.outputSchema();
     assertTrue(outputSchema.isEmpty());
   }
+
+  /**
+   * Test for a query of the form:<br>
+   * {code SELECT * FROM t ORDER BY a}<br>
+   * in which we get a projection list of the form<br<
+   * {@code [`**`, `a`]<br>
+   * If we are given a reader schema of {@code (a, b, c)},
+   * the "natural" expansion will be @{code (b, c, a)}, but we
+   * add a hack to get what the user expects: @{code (a, b, c)}.
+   * The "natural" expansion occurs because the projection list says
+   * "all all columns to the wildcard position except those mentioned
+   * elsewhere". We essentially redefine it as "add or move all columns
+   * in provided schema order."
+   */
+  @Test
+  public void testWildcardWithExplicitWithReaderSchema() {
+
+    // Simulate SELECT *, a ...
+    final ScanSchemaConfigBuilder builder = new ScanSchemaConfigBuilder()
+        .projection(RowSetTestUtils.projectList(
+            SchemaPath.DYNAMIC_STAR, "a"));
+
+    final TupleMetadata readerOutputSchema = new SchemaBuilder()
+        .add("a", MinorType.INT)
+        .add("b", MinorType.BIGINT)
+        .add("c", MinorType.VARCHAR)
+        .buildSchema();
+    final ScanSchemaTracker schemaTracker = builder.build();
+    schemaTracker.applyReaderSchema(readerOutputSchema, ERROR_CONTEXT);
+    assertTrue(schemaTracker.isResolved());
+    assertEquals(readerOutputSchema, schemaTracker.outputSchema());
+  }
+
+  @Test
+  public void testWildcardWithExplicitWithProvidedAndReaderSchema() {
+
+    // Simulate SELECT *, a ...
+    final ScanSchemaConfigBuilder builder = new ScanSchemaConfigBuilder()
+        .projection(RowSetTestUtils.projectList(
+            SchemaPath.DYNAMIC_STAR, "a"));
+
+    final TupleMetadata providedSchema = new SchemaBuilder()
+        .add("a", MinorType.INT)
+        .add("b", MinorType.BIGINT)
+        .add("c", MinorType.VARCHAR)
+        .buildSchema();
+    builder.providedSchema(providedSchema);
+    final ScanSchemaTracker schemaTracker = builder.build();
+    schemaTracker.applyReaderSchema(providedSchema, ERROR_CONTEXT);
+    assertTrue(schemaTracker.isResolved());
+    assertEquals(providedSchema, schemaTracker.outputSchema());
+  }
+
 }
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/v3/schema/TestScanSchemaTrackerMaps.java b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/v3/schema/TestScanSchemaTrackerMaps.java
index c7e4851..dda57c0 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/v3/schema/TestScanSchemaTrackerMaps.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/v3/schema/TestScanSchemaTrackerMaps.java
@@ -23,7 +23,7 @@ import static org.junit.Assert.assertSame;
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
 
-import org.apache.drill.categories.EvfTests;
+import org.apache.drill.categories.EvfTest;
 import org.apache.drill.common.exceptions.CustomErrorContext;
 import org.apache.drill.common.exceptions.EmptyErrorContext;
 import org.apache.drill.common.types.TypeProtos.MinorType;
@@ -44,7 +44,7 @@ import org.junit.experimental.categories.Category;
  * defined schema, scan schema, reader schema and missing columns schemas
  * must all be trees, and must all be kept in sync.
  */
-@Category(EvfTests.class)
+@Category(EvfTest.class)
 public class TestScanSchemaTrackerMaps extends BaseTest {
   private static final CustomErrorContext ERROR_CONTEXT = EmptyErrorContext.INSTANCE;
 
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/v3/schema/TestSchemaTrackerDefined.java b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/v3/schema/TestSchemaTrackerDefined.java
index 76aba7f..fc343bd 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/v3/schema/TestSchemaTrackerDefined.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/v3/schema/TestSchemaTrackerDefined.java
@@ -23,7 +23,7 @@ import static org.junit.Assert.assertSame;
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
 
-import org.apache.drill.categories.EvfTests;
+import org.apache.drill.categories.EvfTest;
 import org.apache.drill.common.exceptions.CustomErrorContext;
 import org.apache.drill.common.exceptions.EmptyErrorContext;
 import org.apache.drill.common.expression.SchemaPath;
@@ -38,7 +38,7 @@ import org.apache.drill.exec.record.metadata.TupleSchema;
 import org.junit.Test;
 import org.junit.experimental.categories.Category;
 
-@Category(EvfTests.class)
+@Category(EvfTest.class)
 public class TestSchemaTrackerDefined {
   private static final CustomErrorContext ERROR_CONTEXT = EmptyErrorContext.INSTANCE;
 
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/v3/schema/TestSchemaTrackerEarlyReaderSchema.java b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/v3/schema/TestSchemaTrackerEarlyReaderSchema.java
index af74205..5dab1ad 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/v3/schema/TestSchemaTrackerEarlyReaderSchema.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/v3/schema/TestSchemaTrackerEarlyReaderSchema.java
@@ -22,13 +22,15 @@ import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
 
-import org.apache.drill.categories.EvfTests;
+import org.apache.drill.categories.EvfTest;
 import org.apache.drill.common.exceptions.CustomErrorContext;
 import org.apache.drill.common.exceptions.EmptyErrorContext;
 import org.apache.drill.common.exceptions.UserException;
+import org.apache.drill.common.expression.SchemaPath;
 import org.apache.drill.common.types.TypeProtos.MinorType;
 import org.apache.drill.exec.physical.impl.scan.ScanTestUtils;
-import org.apache.drill.exec.physical.impl.scan.v3.schema.ImplicitColumnResolver.ImplicitColumnOptions;
+import org.apache.drill.exec.physical.impl.scan.v3.file.ImplicitColumnResolver;
+import org.apache.drill.exec.physical.impl.scan.v3.file.ImplicitColumnResolver.ImplicitColumnOptions;
 import org.apache.drill.exec.physical.rowSet.RowSetTestUtils;
 import org.apache.drill.exec.record.metadata.SchemaBuilder;
 import org.apache.drill.exec.record.metadata.TupleMetadata;
@@ -41,7 +43,7 @@ import org.junit.experimental.categories.Category;
  * Basic tests of early reader against a project list are in
  * {@link TestSchemaTrackerInputSchema}.
  */
-@Category(EvfTests.class)
+@Category(EvfTest.class)
 public class TestSchemaTrackerEarlyReaderSchema extends SubOperatorTest {
   protected static final CustomErrorContext ERROR_CONTEXT = EmptyErrorContext.INSTANCE;
   protected static final TupleMetadata SCHEMA = BaseTestSchemaTracker.SCHEMA;
@@ -169,4 +171,23 @@ public class TestSchemaTrackerEarlyReaderSchema extends SubOperatorTest {
       assertTrue(e.getMessage().contains("Reader column: `a` INT"));
     }
   }
+
+  @Test
+  public void testWildcardWithExplicitWithEarlyReaderSchema() {
+
+    // Simulate SELECT *, a ...
+    final ScanSchemaConfigBuilder builder = new ScanSchemaConfigBuilder()
+        .projection(RowSetTestUtils.projectList(
+            SchemaPath.DYNAMIC_STAR, "a"));
+
+    final TupleMetadata readerSchema = new SchemaBuilder()
+        .add("a", MinorType.INT)
+        .add("b", MinorType.BIGINT)
+        .add("c", MinorType.VARCHAR)
+        .buildSchema();
+    final ScanSchemaTracker schemaTracker = builder.build();
+    schemaTracker.applyEarlyReaderSchema(readerSchema);
+    assertTrue(schemaTracker.isResolved());
+    assertEquals(readerSchema, schemaTracker.outputSchema());
+  }
 }
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/v3/schema/TestSchemaTrackerInputSchema.java b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/v3/schema/TestSchemaTrackerInputSchema.java
index 40e860f..2113d01 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/v3/schema/TestSchemaTrackerInputSchema.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/v3/schema/TestSchemaTrackerInputSchema.java
@@ -27,7 +27,7 @@ import java.util.Collection;
 import java.util.Collections;
 import java.util.function.Consumer;
 
-import org.apache.drill.categories.EvfTests;
+import org.apache.drill.categories.EvfTest;
 import org.apache.drill.common.exceptions.UserException;
 import org.apache.drill.common.expression.SchemaPath;
 import org.apache.drill.common.types.TypeProtos.MinorType;
@@ -47,7 +47,7 @@ import org.junit.experimental.categories.Category;
  * before any data is seen. There are subtle differences explored
  * here and in {@link TestSchemaTrackerEarlyReaderSchema}.
  */
-@Category(EvfTests.class)
+@Category(EvfTest.class)
 public class TestSchemaTrackerInputSchema extends BaseTestSchemaTracker {
 
   private void testBoth(Collection<SchemaPath> projList, TupleMetadata schema,
@@ -175,25 +175,6 @@ public class TestSchemaTrackerInputSchema extends BaseTestSchemaTracker {
     });
   }
 
-  /**
-   * Verify a reasonable error if the name of an implied implicit
-   * column (one that appears with a wildcard) conflicts with a
-   * provided column name. We cannot project two columns with that
-   * same name. Does not occur for a reader schema.
-   */
-  @Test
-  public void testWithWildcardImplicitConflict() {
-    ProjectionSchemaTracker tracker = trackerFor(
-        RowSetTestUtils.projectList("a", SchemaPath.DYNAMIC_STAR, "y"));
-    try {
-      tracker.applyProvidedSchema(SCHEMA);
-      fail();
-    } catch (UserException e) {
-      assertTrue(e.getMessage().contains("implicit"));
-      assertTrue(e.getMessage().contains("Column: a"));
-    }
-  }
-
   @Test
   public void testProvidedMapProjectConflict() {
     ProjectionSchemaTracker tracker = trackerFor(
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/v3/schema/TestSchemaTrackerProjection.java b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/v3/schema/TestSchemaTrackerProjection.java
index 9587e9c..ffb1854 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/v3/schema/TestSchemaTrackerProjection.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/v3/schema/TestSchemaTrackerProjection.java
@@ -19,13 +19,14 @@ package org.apache.drill.exec.physical.impl.scan.v3.schema;
 
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertSame;
 import static org.junit.Assert.assertTrue;
 
 import java.util.Collection;
 import java.util.Collections;
 
-import org.apache.drill.categories.EvfTests;
+import org.apache.drill.categories.EvfTest;
 import org.apache.drill.common.exceptions.CustomErrorContext;
 import org.apache.drill.common.exceptions.EmptyErrorContext;
 import org.apache.drill.common.expression.SchemaPath;
@@ -41,7 +42,7 @@ import org.junit.experimental.categories.Category;
  * Test the first step of scan schema resolution: translating from the
  * projection parser to a dynamic schema ready for resolution.
  */
-@Category(EvfTests.class)
+@Category(EvfTest.class)
 public class TestSchemaTrackerProjection extends BaseTest {
   private static final CustomErrorContext ERROR_CONTEXT = EmptyErrorContext.INSTANCE;
 
@@ -55,7 +56,7 @@ public class TestSchemaTrackerProjection extends BaseTest {
     ProjectionSchemaTracker tracker = schemaTracker(
         Collections.emptyList());
     assertTrue(tracker.isResolved());
-    assertEquals(0, tracker.schemaVersion());
+    assertEquals(1, tracker.schemaVersion());
     assertSame(ScanSchemaTracker.ProjectionType.NONE, tracker.projectionType());
     assertTrue(tracker.internalSchema().toSchema().isEmpty());
     ProjectionFilter filter = tracker.projectionFilter(ERROR_CONTEXT);
@@ -67,7 +68,7 @@ public class TestSchemaTrackerProjection extends BaseTest {
     ProjectionSchemaTracker tracker = schemaTracker(
         RowSetTestUtils.projectAll());
     assertFalse(tracker.isResolved());
-    assertEquals(0, tracker.schemaVersion());
+    assertEquals(1, tracker.schemaVersion());
     assertSame(ScanSchemaTracker.ProjectionType.ALL, tracker.projectionType());
     assertTrue(tracker.internalSchema().toSchema().isEmpty());
     ProjectionFilter filter = tracker.projectionFilter(ERROR_CONTEXT);
@@ -81,6 +82,8 @@ public class TestSchemaTrackerProjection extends BaseTest {
     assertFalse(tracker.isResolved());
     assertTrue(0 < tracker.schemaVersion());
     assertSame(ScanSchemaTracker.ProjectionType.ALL, tracker.projectionType());
+    assertNotNull(tracker.columnProjection("a"));
+    assertNotNull(tracker.columnProjection("b"));
     TupleMetadata schema = tracker.internalSchema().toSchema();
     assertEquals(2, schema.size());
     assertTrue(schema.metadata(0).isDynamic());
@@ -100,4 +103,21 @@ public class TestSchemaTrackerProjection extends BaseTest {
     ProjectionFilter filter = tracker.projectionFilter(ERROR_CONTEXT);
     assertTrue(filter instanceof DynamicSchemaFilter);
   }
+
+  @Test
+  public void testExplicitArray() {
+    ProjectionSchemaTracker tracker = schemaTracker(
+        RowSetTestUtils.projectList("a[1]", "a[3]"));
+    assertSame(ScanSchemaTracker.ProjectionType.SOME, tracker.projectionType());
+
+    ProjectedColumn projCol = tracker.columnProjection("a");
+    assertNotNull(projCol);
+    boolean[] indexes = projCol.indexes();
+    assertNotNull(indexes);
+    assertEquals(4, indexes.length);
+    assertFalse(indexes[0]);
+    assertTrue(indexes[1]);
+    assertFalse(indexes[2]);
+    assertTrue(indexes[3]);
+  }
 }
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/resultSet/impl/TestResultSetLoaderProjection.java b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/resultSet/impl/TestResultSetLoaderProjection.java
index 1bee697..1c8ed95 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/resultSet/impl/TestResultSetLoaderProjection.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/resultSet/impl/TestResultSetLoaderProjection.java
@@ -30,7 +30,7 @@ import static org.junit.Assert.fail;
 import java.util.Arrays;
 import java.util.List;
 
-import org.apache.drill.categories.EvfTests;
+import org.apache.drill.categories.EvfTest;
 import org.apache.drill.common.exceptions.CustomErrorContext;
 import org.apache.drill.common.exceptions.EmptyErrorContext;
 import org.apache.drill.common.exceptions.UserException;
@@ -61,7 +61,7 @@ import org.junit.experimental.categories.Category;
 /**
  * Test of the basics of the projection mechanism.
  */
-@Category(EvfTests.class)
+@Category(EvfTest.class)
 public class TestResultSetLoaderProjection extends SubOperatorTest {
 
   /**
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/store/easy/text/compliant/TestCsvHeader.java b/exec/java-exec/src/test/java/org/apache/drill/exec/store/easy/text/compliant/TestCsvHeader.java
index e4ce405..d4ae1ea 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/store/easy/text/compliant/TestCsvHeader.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/store/easy/text/compliant/TestCsvHeader.java
@@ -23,7 +23,7 @@ import java.io.FileOutputStream;
 import java.nio.file.Paths;
 import java.util.List;
 
-import org.apache.drill.categories.EvfTests;
+import org.apache.drill.categories.EvfTest;
 import org.apache.drill.shaded.guava.com.google.common.collect.Lists;
 import org.apache.drill.test.BaseTestQuery;
 import org.apache.drill.test.TestBuilder;
@@ -31,7 +31,7 @@ import org.junit.Before;
 import org.junit.Test;
 import org.junit.experimental.categories.Category;
 
-@Category(EvfTests.class)
+@Category(EvfTest.class)
 public class TestCsvHeader extends BaseTestQuery{
 
   private static final String ROOT = "store/text/data/cars.csvh";
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/store/easy/text/compliant/TestCsvIgnoreHeaders.java b/exec/java-exec/src/test/java/org/apache/drill/exec/store/easy/text/compliant/TestCsvIgnoreHeaders.java
index 0b3c7d6..80c5ca4 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/store/easy/text/compliant/TestCsvIgnoreHeaders.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/store/easy/text/compliant/TestCsvIgnoreHeaders.java
@@ -22,7 +22,7 @@ import static org.apache.drill.test.rowSet.RowSetUtilities.strArray;
 import java.io.File;
 import java.io.IOException;
 
-import org.apache.drill.categories.EvfTests;
+import org.apache.drill.categories.EvfTest;
 import org.apache.drill.common.types.TypeProtos.MinorType;
 import org.apache.drill.exec.record.metadata.SchemaBuilder;
 import org.apache.drill.exec.record.metadata.TupleMetadata;
@@ -33,7 +33,7 @@ import org.junit.BeforeClass;
 import org.junit.Test;
 import org.junit.experimental.categories.Category;
 
-@Category(EvfTests.class)
+@Category(EvfTest.class)
 public class TestCsvIgnoreHeaders extends BaseCsvTest{
 
   private static String withHeaders[] = {
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/store/easy/text/compliant/TestCsvTableProperties.java b/exec/java-exec/src/test/java/org/apache/drill/exec/store/easy/text/compliant/TestCsvTableProperties.java
index 57ae662..9912ca4 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/store/easy/text/compliant/TestCsvTableProperties.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/store/easy/text/compliant/TestCsvTableProperties.java
@@ -18,7 +18,7 @@
 package org.apache.drill.exec.store.easy.text.compliant;
 
 import org.apache.drill.TestSelectWithOption;
-import org.apache.drill.categories.EvfTests;
+import org.apache.drill.categories.EvfTest;
 import org.apache.drill.common.types.TypeProtos.MinorType;
 import org.apache.drill.exec.physical.rowSet.RowSet;
 import org.apache.drill.exec.physical.rowSet.RowSetBuilder;
@@ -53,7 +53,7 @@ import static org.junit.Assert.assertTrue;
  * @see TestSelectWithOption for similar tests using table
  * properties within SQL
  */
-@Category(EvfTests.class)
+@Category(EvfTest.class)
 public class TestCsvTableProperties extends BaseCsvTest {
 
   @BeforeClass
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/store/easy/text/compliant/TestCsvWithHeaders.java b/exec/java-exec/src/test/java/org/apache/drill/exec/store/easy/text/compliant/TestCsvWithHeaders.java
index 03a98ac..a0bfbbf 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/store/easy/text/compliant/TestCsvWithHeaders.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/store/easy/text/compliant/TestCsvWithHeaders.java
@@ -23,7 +23,7 @@ import java.io.IOException;
 import java.io.PrintWriter;
 import java.util.Iterator;
 
-import org.apache.drill.categories.EvfTests;
+import org.apache.drill.categories.EvfTest;
 import org.apache.drill.common.exceptions.UserRemoteException;
 import org.apache.drill.common.types.TypeProtos.MinorType;
 import org.apache.drill.exec.physical.rowSet.DirectRowSet;
@@ -60,7 +60,7 @@ import static org.junit.Assert.fail;
  *
  * @see TestHeaderBuilder
  */
-@Category(EvfTests.class)
+@Category(EvfTest.class)
 public class TestCsvWithHeaders extends BaseCsvTest {
 
   private static final String TEST_FILE_NAME = "basic.csv";
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/store/easy/text/compliant/TestCsvWithSchema.java b/exec/java-exec/src/test/java/org/apache/drill/exec/store/easy/text/compliant/TestCsvWithSchema.java
index 0458aee..ac27d2f 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/store/easy/text/compliant/TestCsvWithSchema.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/store/easy/text/compliant/TestCsvWithSchema.java
@@ -24,7 +24,7 @@ import static org.junit.Assert.fail;
 
 import java.util.Iterator;
 
-import org.apache.drill.categories.EvfTests;
+import org.apache.drill.categories.EvfTest;
 import org.apache.drill.common.exceptions.UserRemoteException;
 import org.apache.drill.common.types.TypeProtos.MinorType;
 import org.apache.drill.exec.record.metadata.SchemaBuilder;
@@ -47,7 +47,7 @@ import org.junit.experimental.categories.Category;
  * The focus here is on the schema mechanism, which can be explored
  * with simple tables of just a few rows.
  */
-@Category(EvfTests.class)
+@Category(EvfTest.class)
 public class TestCsvWithSchema extends BaseCsvTest {
 
   protected static final String FILE1_NAME = "file1.csv";
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/store/easy/text/compliant/TestCsvWithoutHeaders.java b/exec/java-exec/src/test/java/org/apache/drill/exec/store/easy/text/compliant/TestCsvWithoutHeaders.java
index 872574c..c724a71 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/store/easy/text/compliant/TestCsvWithoutHeaders.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/store/easy/text/compliant/TestCsvWithoutHeaders.java
@@ -26,7 +26,7 @@ import java.io.File;
 import java.io.IOException;
 import java.util.Iterator;
 
-import org.apache.drill.categories.EvfTests;
+import org.apache.drill.categories.EvfTest;
 import org.apache.drill.common.exceptions.UserRemoteException;
 import org.apache.drill.common.types.TypeProtos.MinorType;
 import org.apache.drill.exec.TestEmptyInputSql;
@@ -47,7 +47,7 @@ import org.junit.experimental.categories.Category;
  * and without an external schema. Data is represented with the
  * `columns` array column.
  */
-@Category(EvfTests.class)
+@Category(EvfTest.class)
 public class TestCsvWithoutHeaders extends BaseCsvTest {
 
   private static final String TEST_FILE_NAME = "simple.csv";
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/store/easy/text/compliant/TestPartitionRace.java b/exec/java-exec/src/test/java/org/apache/drill/exec/store/easy/text/compliant/TestPartitionRace.java
index 76c9d5d..ab4ee01 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/store/easy/text/compliant/TestPartitionRace.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/store/easy/text/compliant/TestPartitionRace.java
@@ -25,7 +25,7 @@ import java.io.File;
 import java.io.IOException;
 import java.util.Iterator;
 
-import org.apache.drill.categories.EvfTests;
+import org.apache.drill.categories.EvfTest;
 import org.apache.drill.common.types.TypeProtos.MinorType;
 import org.apache.drill.exec.record.metadata.SchemaBuilder;
 import org.apache.drill.exec.record.metadata.TupleMetadata;
@@ -54,7 +54,7 @@ import org.junit.experimental.categories.Category;
  * current "V3" version. The tests here verify this behavior.
  */
 
-@Category(EvfTests.class)
+@Category(EvfTest.class)
 public class TestPartitionRace extends BaseCsvTest {
 
   @BeforeClass
diff --git a/exec/java-exec/src/test/java/org/apache/drill/test/QueryBuilder.java b/exec/java-exec/src/test/java/org/apache/drill/test/QueryBuilder.java
index e451dbb..0be54f3 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/test/QueryBuilder.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/test/QueryBuilder.java
@@ -329,12 +329,11 @@ public class QueryBuilder {
    * Run the query and return the first non-empty batch as a
    * {@link DirectRowSet} object that can be inspected directly
    * by the code using a {@link RowSetReader}.
-   * <p>
    *
    * @see #rowSetIterator() for a version that reads a series of
    * batches as row sets.
    * @return a row set that represents the first non-empty batch returned from
-   * the query
+   * the query, or {@code null} if the query returns no data (no batches)
    * @throws RpcException if anything goes wrong
    */
   public DirectRowSet rowSet() throws RpcException {
diff --git a/exec/vector/src/main/java/org/apache/drill/exec/record/metadata/MetadataUtils.java b/exec/vector/src/main/java/org/apache/drill/exec/record/metadata/MetadataUtils.java
index 9aee7eb..b21e22a 100644
--- a/exec/vector/src/main/java/org/apache/drill/exec/record/metadata/MetadataUtils.java
+++ b/exec/vector/src/main/java/org/apache/drill/exec/record/metadata/MetadataUtils.java
@@ -241,6 +241,10 @@ public class MetadataUtils {
     return new PrimitiveColumnMetadata(field);
   }
 
+  public static boolean isScalar(ColumnMetadata col) {
+    return isScalar(col.type());
+  }
+
   public static boolean isScalar(MinorType type) {
     return !isComplex(type);
   }


Mime
View raw message