drill-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From ar...@apache.org
Subject [7/8] drill git commit: DRILL-6331: Revisit Hive Drill native parquet implementation to be exposed to Drill optimizations (filter / limit push down, count to direct scan)
Date Fri, 27 Apr 2018 09:41:49 GMT
http://git-wip-us.apache.org/repos/asf/drill/blob/c6549e58/contrib/storage-hive/core/src/test/java/org/apache/drill/exec/sql/hive/TestViewSupportOnHiveTables.java
----------------------------------------------------------------------
diff --git a/contrib/storage-hive/core/src/test/java/org/apache/drill/exec/sql/hive/TestViewSupportOnHiveTables.java b/contrib/storage-hive/core/src/test/java/org/apache/drill/exec/sql/hive/TestViewSupportOnHiveTables.java
index 3706ff2..64680c0 100644
--- a/contrib/storage-hive/core/src/test/java/org/apache/drill/exec/sql/hive/TestViewSupportOnHiveTables.java
+++ b/contrib/storage-hive/core/src/test/java/org/apache/drill/exec/sql/hive/TestViewSupportOnHiveTables.java
@@ -35,7 +35,7 @@ public class TestViewSupportOnHiveTables extends TestBaseViewSupport {
 
   @BeforeClass
   public static void generateHive() throws Exception{
-    hiveTest = HiveTestDataGenerator.getInstance(dirTestWatcher.getRootDir());
+    hiveTest = HiveTestDataGenerator.getInstance(dirTestWatcher);
     hiveTest.addHiveTestPlugin(getDrillbitContext().getStorage());
   }
 

http://git-wip-us.apache.org/repos/asf/drill/blob/c6549e58/contrib/storage-hive/core/src/test/java/org/apache/drill/exec/store/hive/HiveTestDataGenerator.java
----------------------------------------------------------------------
diff --git a/contrib/storage-hive/core/src/test/java/org/apache/drill/exec/store/hive/HiveTestDataGenerator.java b/contrib/storage-hive/core/src/test/java/org/apache/drill/exec/store/hive/HiveTestDataGenerator.java
index 78e5b39..f206999 100644
--- a/contrib/storage-hive/core/src/test/java/org/apache/drill/exec/store/hive/HiveTestDataGenerator.java
+++ b/contrib/storage-hive/core/src/test/java/org/apache/drill/exec/store/hive/HiveTestDataGenerator.java
@@ -21,6 +21,7 @@ import java.io.File;
 import java.io.IOException;
 import java.io.PrintWriter;
 import java.nio.file.Files;
+import java.nio.file.Paths;
 import java.nio.file.attribute.PosixFilePermission;
 import java.sql.Date;
 import java.sql.Timestamp;
@@ -31,6 +32,7 @@ import com.google.common.collect.Sets;
 import com.google.common.io.Resources;
 import org.apache.commons.io.FileUtils;
 import org.apache.commons.lang3.StringUtils;
+import org.apache.drill.test.BaseDirTestWatcher;
 import org.apache.drill.test.BaseTestQuery;
 import org.apache.drill.common.exceptions.DrillException;
 import org.apache.drill.exec.store.StoragePluginRegistry;
@@ -52,9 +54,11 @@ public class HiveTestDataGenerator {
 
   private final String dbDir;
   private final String whDir;
+  private final BaseDirTestWatcher dirTestWatcher;
   private final Map<String, String> config;
 
-  public static synchronized HiveTestDataGenerator getInstance(File baseDir) throws Exception {
+  public static synchronized HiveTestDataGenerator getInstance(BaseDirTestWatcher dirTestWatcher) throws Exception {
+    File baseDir = dirTestWatcher.getRootDir();
     if (instance == null || !HiveTestDataGenerator.baseDir.equals(baseDir)) {
       HiveTestDataGenerator.baseDir = baseDir;
 
@@ -64,19 +68,20 @@ public class HiveTestDataGenerator {
       final String dbDir = dbDirFile.getAbsolutePath();
       final String whDir = whDirFile.getAbsolutePath();
 
-      instance = new HiveTestDataGenerator(dbDir, whDir);
+      instance = new HiveTestDataGenerator(dbDir, whDir, dirTestWatcher);
       instance.generateTestData();
     }
 
     return instance;
   }
 
-  private HiveTestDataGenerator(final String dbDir, final String whDir) {
+  private HiveTestDataGenerator(final String dbDir, final String whDir, final BaseDirTestWatcher dirTestWatcher) {
     this.dbDir = dbDir;
     this.whDir = whDir;
+    this.dirTestWatcher = dirTestWatcher;
 
     config = Maps.newHashMap();
-    config.put("hive.metastore.uris", "");
+    config.put(ConfVars.METASTOREURIS.toString(), "");
     config.put("javax.jdo.option.ConnectionURL", String.format("jdbc:derby:;databaseName=%s;create=true", dbDir));
     config.put("hive.metastore.warehouse.dir", whDir);
     config.put(FileSystem.FS_DEFAULT_NAME_KEY, FileSystem.DEFAULT_FS);
@@ -128,7 +133,7 @@ public class HiveTestDataGenerator {
     try {
       Files.setPosixFilePermissions(dir.toPath(), perms);
     } catch (IOException e) {
-      new RuntimeException(e);
+      throw new RuntimeException(e);
     }
 
     return dir;
@@ -494,22 +499,6 @@ public class HiveTestDataGenerator {
     executeQuery(hiveDriver, "INSERT INTO TABLE kv_parquet PARTITION(part1) SELECT key, value, key FROM default.kv");
     executeQuery(hiveDriver, "ALTER TABLE kv_parquet ADD COLUMNS (newcol string)");
 
-    executeQuery(hiveDriver,
-        "CREATE TABLE countStar_Parquet (int_field INT) STORED AS parquet");
-
-    final int numOfRows = 200;
-    final StringBuffer sb = new StringBuffer();
-    sb.append("VALUES ");
-    for(int i = 0; i < numOfRows; ++i) {
-      if(i != 0) {
-        sb.append(",");
-      }
-      sb.append("(").append(i).append(")");
-    }
-
-    executeQuery(hiveDriver, "INSERT INTO TABLE countStar_Parquet \n" +
-        sb.toString());
-
     // Create a StorageHandler based table (DRILL-3739)
     executeQuery(hiveDriver, "CREATE TABLE kv_sh(key INT, value STRING) STORED BY " +
         "'org.apache.hadoop.hive.ql.metadata.DefaultStorageHandler'");
@@ -551,9 +540,60 @@ public class HiveTestDataGenerator {
         Resources.getResource("simple.json") + "' into table default.simple_json";
     executeQuery(hiveDriver, loadData);
 
+    createTestDataForDrillNativeParquetReaderTests(hiveDriver);
+
     ss.close();
   }
 
+  private void createTestDataForDrillNativeParquetReaderTests(Driver hiveDriver) {
+    // Hive managed table that has data qualified for Drill native filter push down
+    executeQuery(hiveDriver, "create table kv_native(key int, sub_key int) stored as parquet");
+    // each insert is created in separate file
+    executeQuery(hiveDriver, "insert into table kv_native values (1, 1), (1, 2)");
+    executeQuery(hiveDriver, "insert into table kv_native values (1, 3), (1, 4)");
+    executeQuery(hiveDriver, "insert into table kv_native values (2, 5), (2, 6)");
+    executeQuery(hiveDriver, "insert into table kv_native values (null, 9), (null, 10)");
+
+    // Hive external table which has three partitions
+
+    // copy external table with data from test resources
+    dirTestWatcher.copyResourceToRoot(Paths.get("external"));
+
+    File external = new File (baseDir, "external");
+    String tableLocation = new File(external, "kv_native_ext").toURI().getPath();
+
+    executeQuery(hiveDriver, String.format("create external table kv_native_ext(key int) " +
+        "partitioned by (part_key int) " +
+        "stored as parquet location '%s'",
+        tableLocation));
+
+    /*
+      DATA:
+      key, part_key
+      1, 1
+      2, 1
+      3, 2
+      4, 2
+     */
+
+    // add partitions
+
+    // partition in the same location as table
+    String firstPartition = new File(tableLocation, "part_key=1").toURI().getPath();
+    executeQuery(hiveDriver, String.format("alter table kv_native_ext add partition (part_key = '1') " +
+      "location '%s'", firstPartition));
+
+    // partition in different location with table
+    String secondPartition = new File(external, "part_key=2").toURI().getPath();
+    executeQuery(hiveDriver, String.format("alter table kv_native_ext add partition (part_key = '2') " +
+      "location '%s'", secondPartition));
+
+    // add empty partition
+    String thirdPartition = new File(dirTestWatcher.makeSubDir(Paths.get("empty_part")), "part_key=3").toURI().getPath();
+    executeQuery(hiveDriver, String.format("alter table kv_native_ext add partition (part_key = '3') " +
+      "location '%s'", thirdPartition));
+  }
+
   private File getTempFile() throws Exception {
     return java.nio.file.Files.createTempFile("drill-hive-test", ".txt").toFile();
   }

http://git-wip-us.apache.org/repos/asf/drill/blob/c6549e58/contrib/storage-hive/core/src/test/resources/external/kv_native_ext/part_key=1/kv_1.parquet
----------------------------------------------------------------------
diff --git a/contrib/storage-hive/core/src/test/resources/external/kv_native_ext/part_key=1/kv_1.parquet b/contrib/storage-hive/core/src/test/resources/external/kv_native_ext/part_key=1/kv_1.parquet
new file mode 100755
index 0000000..f641402
Binary files /dev/null and b/contrib/storage-hive/core/src/test/resources/external/kv_native_ext/part_key=1/kv_1.parquet differ

http://git-wip-us.apache.org/repos/asf/drill/blob/c6549e58/contrib/storage-hive/core/src/test/resources/external/part_key=2/kv_2.parquet
----------------------------------------------------------------------
diff --git a/contrib/storage-hive/core/src/test/resources/external/part_key=2/kv_2.parquet b/contrib/storage-hive/core/src/test/resources/external/part_key=2/kv_2.parquet
new file mode 100755
index 0000000..c34c267
Binary files /dev/null and b/contrib/storage-hive/core/src/test/resources/external/part_key=2/kv_2.parquet differ

http://git-wip-us.apache.org/repos/asf/drill/blob/c6549e58/exec/java-exec/src/main/java/org/apache/drill/exec/ops/BaseOperatorContext.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/ops/BaseOperatorContext.java b/exec/java-exec/src/main/java/org/apache/drill/exec/ops/BaseOperatorContext.java
index 211bd65..dfa41e8 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/ops/BaseOperatorContext.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/ops/BaseOperatorContext.java
@@ -18,8 +18,11 @@
 package org.apache.drill.exec.ops;
 
 import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
 import java.util.concurrent.ExecutorService;
 
+import com.google.common.base.Preconditions;
 import org.apache.drill.common.exceptions.UserException;
 import org.apache.drill.exec.memory.BufferAllocator;
 import org.apache.drill.exec.physical.base.PhysicalOperator;
@@ -28,8 +31,6 @@ import org.apache.drill.exec.testing.ControlsInjector;
 import org.apache.drill.exec.testing.ExecutionControls;
 import org.apache.hadoop.conf.Configuration;
 
-import com.google.common.base.Preconditions;
-
 import io.netty.buffer.DrillBuf;
 
 /**
@@ -46,8 +47,9 @@ public abstract class BaseOperatorContext implements OperatorContext {
   protected final BufferAllocator allocator;
   protected final PhysicalOperator popConfig;
   protected final BufferManager manager;
-  private DrillFileSystem fs;
+  private List<DrillFileSystem> fileSystems;
   private ControlsInjector injector;
+  private boolean allowCreatingFileSystem = true;
 
   public BaseOperatorContext(FragmentContext context, BufferAllocator allocator,
                PhysicalOperator popConfig) {
@@ -55,6 +57,7 @@ public abstract class BaseOperatorContext implements OperatorContext {
     this.allocator = allocator;
     this.popConfig = popConfig;
     this.manager = new BufferManagerImpl(allocator);
+    this.fileSystems = new ArrayList<>();
   }
 
   @Override
@@ -158,35 +161,46 @@ public abstract class BaseOperatorContext implements OperatorContext {
     } catch (RuntimeException e) {
       ex = ex == null ? e : ex;
     }
-    try {
-      if (fs != null) {
+
+    for (DrillFileSystem fs : fileSystems) {
+      try {
         fs.close();
-        fs = null;
-      }
-    } catch (IOException e) {
+      } catch (IOException e) {
       throw UserException.resourceError(e)
-        .addContext("Failed to close the Drill file system for " + getName())
-        .build(logger);
+          .addContext("Failed to close the Drill file system for " + getName())
+          .build(logger);
+      }
     }
+
     if (ex != null) {
       throw ex;
     }
   }
 
+  /**
+   * Creates DrillFileSystem that automatically tracks operator stats.
+   * Only one tracking and no non-tracking file system per operator context.
+   */
   @Override
   public DrillFileSystem newFileSystem(Configuration conf) throws IOException {
-    Preconditions.checkState(fs == null, "Tried to create a second FileSystem. Can only be called once per OperatorContext");
-    fs = new DrillFileSystem(conf, getStats());
+    Preconditions.checkState(allowCreatingFileSystem, "Only one tracking file system is allowed per Operator Context and it is already created.");
+    Preconditions.checkState(fileSystems.isEmpty(), "Non-tracking file system(-s) is(are) already created.");
+    DrillFileSystem fs = new DrillFileSystem(conf, getStats());
+    fileSystems.add(fs);
+    allowCreatingFileSystem = false;
     return fs;
   }
 
   /**
    * Creates a DrillFileSystem that does not automatically track operator stats.
+   * Multiple non-tracking file system are allowed.
    */
   @Override
   public DrillFileSystem newNonTrackingFileSystem(Configuration conf) throws IOException {
-    Preconditions.checkState(fs == null, "Tried to create a second FileSystem. Can only be called once per OperatorContext");
-    fs = new DrillFileSystem(conf, null);
+    Preconditions.checkState(allowCreatingFileSystem, "Only one tracking file system is allowed per Operator Context and it is already created.");
+    DrillFileSystem fs = new DrillFileSystem(conf, null);
+    fileSystems.add(fs);
     return fs;
   }
+
 }

http://git-wip-us.apache.org/repos/asf/drill/blob/c6549e58/exec/java-exec/src/main/java/org/apache/drill/exec/physical/base/AbstractGroupScan.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/base/AbstractGroupScan.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/base/AbstractGroupScan.java
index b2ddf68..e89a1f9 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/base/AbstractGroupScan.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/base/AbstractGroupScan.java
@@ -24,12 +24,16 @@ import java.util.List;
 
 import com.google.common.collect.Lists;
 
+import org.apache.drill.common.expression.LogicalExpression;
 import org.apache.drill.common.expression.SchemaPath;
+import org.apache.drill.exec.expr.fn.FunctionImplementationRegistry;
+import org.apache.drill.exec.ops.UdfUtilities;
 import org.apache.drill.exec.physical.EndpointAffinity;
 import org.apache.drill.exec.planner.fragment.DistributionAffinity;
 import org.apache.drill.exec.planner.physical.PlannerSettings;
 
 import com.fasterxml.jackson.annotation.JsonIgnore;
+import org.apache.drill.exec.server.options.OptionManager;
 
 public abstract class AbstractGroupScan extends AbstractBase implements GroupScan {
 
@@ -164,4 +168,14 @@ public abstract class AbstractGroupScan extends AbstractBase implements GroupSca
   public DistributionAffinity getDistributionAffinity() {
     return DistributionAffinity.SOFT;
   }
+
+  @Override
+  public LogicalExpression getFilter() {
+    return null;
+  }
+
+  @Override
+  public GroupScan applyFilter(LogicalExpression filterExpr, UdfUtilities udfUtilities, FunctionImplementationRegistry functionImplementationRegistry, OptionManager optionManager) {
+    return null;
+  }
 }

http://git-wip-us.apache.org/repos/asf/drill/blob/c6549e58/exec/java-exec/src/main/java/org/apache/drill/exec/physical/base/GroupScan.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/base/GroupScan.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/base/GroupScan.java
index fc63c77..33ab13d 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/base/GroupScan.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/base/GroupScan.java
@@ -21,13 +21,17 @@ import java.util.Collection;
 import java.util.List;
 
 import org.apache.drill.common.exceptions.ExecutionSetupException;
+import org.apache.drill.common.expression.LogicalExpression;
 import org.apache.drill.common.expression.SchemaPath;
+import org.apache.drill.exec.expr.fn.FunctionImplementationRegistry;
+import org.apache.drill.exec.ops.UdfUtilities;
 import org.apache.drill.exec.physical.PhysicalOperatorSetupException;
 import org.apache.drill.exec.planner.physical.PlannerSettings;
 import org.apache.drill.exec.proto.CoordinationProtos.DrillbitEndpoint;
 
 import com.fasterxml.jackson.annotation.JsonIgnore;
 import com.google.common.collect.ImmutableList;
+import org.apache.drill.exec.server.options.OptionManager;
 
 /**
  * A GroupScan operator represents all data which will be scanned by a given physical
@@ -134,4 +138,10 @@ public interface GroupScan extends Scan, HasAffinity{
    */
   Collection<String> getFiles();
 
+  @JsonIgnore
+  LogicalExpression getFilter();
+
+  GroupScan applyFilter(LogicalExpression filterExpr, UdfUtilities udfUtilities,
+                        FunctionImplementationRegistry functionImplementationRegistry, OptionManager optionManager);
+
 }

http://git-wip-us.apache.org/repos/asf/drill/blob/c6549e58/exec/java-exec/src/main/java/org/apache/drill/exec/physical/base/ScanStats.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/base/ScanStats.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/base/ScanStats.java
index 4404a98..f2974e1 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/base/ScanStats.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/base/ScanStats.java
@@ -18,7 +18,6 @@
 package org.apache.drill.exec.physical.base;
 
 public class ScanStats {
-//  private static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(ScanStats.class);
 
   public static final ScanStats TRIVIAL_TABLE = new ScanStats(GroupScanProperty.NO_EXACT_ROW_COUNT, 20, 1, 1);
 
@@ -30,7 +29,6 @@ public class ScanStats {
   private final GroupScanProperty property;
 
   public ScanStats(GroupScanProperty property, long recordCount, float cpuCost, float diskCost) {
-    super();
     this.recordCount = recordCount;
     this.cpuCost = cpuCost;
     this.diskCost = diskCost;
@@ -49,6 +47,11 @@ public class ScanStats {
     return diskCost;
   }
 
+  @Override
+  public String toString() {
+    return "ScanStats{" + "recordCount=" + recordCount + ", cpuCost=" + cpuCost + ", diskCost=" + diskCost + ", property=" + property + '}';
+  }
+
   /**
    * Return if GroupScan knows the exact row count in the result of getSize() call.
    * By default, groupscan does not know the exact row count, before it scans every rows.
@@ -60,7 +63,7 @@ public class ScanStats {
 
 
 
-  public static enum GroupScanProperty {
+  public enum GroupScanProperty {
     NO_EXACT_ROW_COUNT(false, false),
     EXACT_ROW_COUNT(true, true);
 

http://git-wip-us.apache.org/repos/asf/drill/blob/c6549e58/exec/java-exec/src/main/java/org/apache/drill/exec/planner/ParquetPartitionDescriptor.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/ParquetPartitionDescriptor.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/ParquetPartitionDescriptor.java
index a183e51..cbc530b 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/ParquetPartitionDescriptor.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/ParquetPartitionDescriptor.java
@@ -21,7 +21,6 @@ import org.apache.calcite.rel.core.TableScan;
 import org.apache.calcite.util.BitSets;
 import org.apache.drill.common.expression.SchemaPath;
 import org.apache.drill.common.types.TypeProtos;
-import org.apache.drill.exec.physical.base.FileGroupScan;
 import org.apache.drill.exec.physical.base.GroupScan;
 import org.apache.drill.exec.planner.logical.DrillRel;
 import org.apache.drill.exec.planner.logical.DrillScanRel;
@@ -29,31 +28,57 @@ import org.apache.drill.exec.planner.physical.PlannerSettings;
 import org.apache.drill.exec.store.dfs.FileSelection;
 import org.apache.drill.exec.store.dfs.FormatSelection;
 import org.apache.drill.exec.store.dfs.MetadataContext;
-import org.apache.drill.exec.store.parquet.ParquetGroupScan;
+import org.apache.drill.exec.store.parquet.AbstractParquetGroupScan;
+import org.apache.drill.exec.store.parquet.ParquetReaderUtility;
+import org.apache.drill.exec.util.DecimalUtility;
+import org.apache.drill.exec.vector.NullableBigIntVector;
+import org.apache.drill.exec.vector.NullableBitVector;
+import org.apache.drill.exec.vector.NullableDateVector;
+import org.apache.drill.exec.vector.NullableDecimal18Vector;
+import org.apache.drill.exec.vector.NullableFloat4Vector;
+import org.apache.drill.exec.vector.NullableFloat8Vector;
+import org.apache.drill.exec.vector.NullableIntVector;
+import org.apache.drill.exec.vector.NullableIntervalVector;
+import org.apache.drill.exec.vector.NullableSmallIntVector;
+import org.apache.drill.exec.vector.NullableTimeStampVector;
+import org.apache.drill.exec.vector.NullableTimeVector;
+import org.apache.drill.exec.vector.NullableTinyIntVector;
+import org.apache.drill.exec.vector.NullableUInt1Vector;
+import org.apache.drill.exec.vector.NullableUInt2Vector;
+import org.apache.drill.exec.vector.NullableUInt4Vector;
+import org.apache.drill.exec.vector.NullableVarBinaryVector;
+import org.apache.drill.exec.vector.NullableVarCharVector;
 import org.apache.drill.exec.vector.ValueVector;
 
 import com.google.common.collect.Lists;
+import org.apache.hadoop.fs.Path;
+import org.apache.parquet.io.api.Binary;
+import org.joda.time.DateTimeConstants;
 
 import java.io.IOException;
+import java.util.ArrayList;
 import java.util.BitSet;
 import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
 
-
 /**
  * PartitionDescriptor that describes partitions based on column names instead of directory structure
  */
 public class ParquetPartitionDescriptor extends AbstractPartitionDescriptor {
 
-  private final List<SchemaPath> partitionColumns;
+  private static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(ParquetPartitionDescriptor.class);
+
   private final DrillScanRel scanRel;
+  private final AbstractParquetGroupScan groupScan;
+  private final List<SchemaPath> partitionColumns;
 
   public ParquetPartitionDescriptor(PlannerSettings settings, DrillScanRel scanRel) {
-    ParquetGroupScan scan = (ParquetGroupScan) scanRel.getGroupScan();
-    this.partitionColumns = scan.getPartitionColumns();
     this.scanRel = scanRel;
+    assert scanRel.getGroupScan() instanceof AbstractParquetGroupScan;
+    this.groupScan = (AbstractParquetGroupScan) scanRel.getGroupScan();
+    this.partitionColumns = groupScan.getPartitionColumns();
   }
 
   @Override
@@ -81,15 +106,6 @@ public class ParquetPartitionDescriptor extends AbstractPartitionDescriptor {
     return partitionColumns.size();
   }
 
-  private GroupScan createNewGroupScan(List<String> newFiles, String cacheFileRoot,
-      boolean wasAllPartitionsPruned, MetadataContext metaContext) throws IOException {
-    final FileSelection newSelection = FileSelection.create(null, newFiles, getBaseTableLocation(),
-        cacheFileRoot, wasAllPartitionsPruned);
-    newSelection.setMetaContext(metaContext);
-    final FileGroupScan newScan = ((FileGroupScan)scanRel.getGroupScan()).clone(newSelection);
-    return newScan;
-  }
-
   @Override
   public void populatePartitionVectors(ValueVector[] vectors, List<PartitionLocation> partitions,
                                        BitSet partitionColumnBitSet, Map<Integer, String> fieldNameMap) {
@@ -97,8 +113,7 @@ public class ParquetPartitionDescriptor extends AbstractPartitionDescriptor {
     for (PartitionLocation partitionLocation: partitions) {
       for (int partitionColumnIndex : BitSets.toIter(partitionColumnBitSet)) {
         SchemaPath column = SchemaPath.getSimplePath(fieldNameMap.get(partitionColumnIndex));
-        ((ParquetGroupScan) scanRel.getGroupScan()).populatePruningVector(vectors[partitionColumnIndex], record, column,
-            partitionLocation.getEntirePartitionLocation());
+        populatePruningVector(vectors[partitionColumnIndex], record, column, partitionLocation.getEntirePartitionLocation());
       }
       record++;
     }
@@ -114,7 +129,7 @@ public class ParquetPartitionDescriptor extends AbstractPartitionDescriptor {
 
   @Override
   public TypeProtos.MajorType getVectorType(SchemaPath column, PlannerSettings plannerSettings) {
-    return ((ParquetGroupScan) scanRel.getGroupScan()).getTypeForColumn(column);
+    return groupScan.getTypeForColumn(column);
   }
 
   @Override
@@ -124,26 +139,22 @@ public class ParquetPartitionDescriptor extends AbstractPartitionDescriptor {
   }
 
   @Override
-  protected void createPartitionSublists() {
-    Set<String> fileLocations = ((ParquetGroupScan) scanRel.getGroupScan()).getFileSet();
-    List<PartitionLocation> locations = new LinkedList<>();
-    for (String file: fileLocations) {
-      locations.add(new ParquetPartitionLocation(file));
-    }
-    locationSuperList = Lists.partition(locations, PartitionDescriptor.PARTITION_BATCH_SIZE);
-    sublistsCreated = true;
-  }
-
-  @Override
-  public TableScan createTableScan(List<PartitionLocation> newPartitionLocation, String cacheFileRoot,
-      boolean wasAllPartitionsPruned, MetadataContext metaContext) throws Exception {
-    List<String> newFiles = Lists.newArrayList();
+  public TableScan createTableScan(List<PartitionLocation> newPartitionLocation,
+                                   String cacheFileRoot,
+                                   boolean wasAllPartitionsPruned,
+                                   MetadataContext metaContext) throws Exception {
+    List<String> newFiles = new ArrayList<>();
     for (final PartitionLocation location : newPartitionLocation) {
       newFiles.add(location.getEntirePartitionLocation());
     }
 
     final GroupScan newGroupScan = createNewGroupScan(newFiles, cacheFileRoot, wasAllPartitionsPruned, metaContext);
 
+    if (newGroupScan == null) {
+      logger.warn("Unable to create new group scan, returning original table scan.");
+      return scanRel;
+    }
+
     return new DrillScanRel(scanRel.getCluster(),
         scanRel.getTraitSet().plus(DrillRel.DRILL_LOGICAL),
         scanRel.getTable(),
@@ -154,9 +165,261 @@ public class ParquetPartitionDescriptor extends AbstractPartitionDescriptor {
   }
 
   @Override
-  public TableScan createTableScan(List<PartitionLocation> newPartitionLocation,
-      boolean wasAllPartitionsPruned) throws Exception {
+  public TableScan createTableScan(List<PartitionLocation> newPartitionLocation, boolean wasAllPartitionsPruned) throws Exception {
     return createTableScan(newPartitionLocation, null, wasAllPartitionsPruned, null);
   }
 
+  @Override
+  protected void createPartitionSublists() {
+    Set<String> fileLocations = groupScan.getFileSet();
+    List<PartitionLocation> locations = new LinkedList<>();
+    for (String file : fileLocations) {
+      locations.add(new ParquetPartitionLocation(file));
+    }
+    locationSuperList = Lists.partition(locations, PartitionDescriptor.PARTITION_BATCH_SIZE);
+    sublistsCreated = true;
+  }
+
+  private GroupScan createNewGroupScan(List<String> newFiles,
+                                       String cacheFileRoot,
+                                       boolean wasAllPartitionsPruned,
+                                       MetadataContext metaContext) throws IOException {
+
+    FileSelection newSelection = FileSelection.create(null, newFiles, getBaseTableLocation(), cacheFileRoot, wasAllPartitionsPruned);
+    if (newSelection == null) {
+      return null;
+    }
+    newSelection.setMetaContext(metaContext);
+    return groupScan.clone(newSelection);
+  }
+
+  private void populatePruningVector(ValueVector v, int index, SchemaPath column, String file) {
+    String path = Path.getPathWithoutSchemeAndAuthority(new Path(file)).toString();
+    TypeProtos.MajorType majorType = getVectorType(column, null);
+    TypeProtos.MinorType type = majorType.getMinorType();
+    switch (type) {
+      case BIT: {
+        NullableBitVector bitVector = (NullableBitVector) v;
+        Boolean value = groupScan.getPartitionValue(path, column, Boolean.class);
+        if (value == null) {
+          bitVector.getMutator().setNull(index);
+        } else {
+          bitVector.getMutator().setSafe(index, value ? 1 : 0);
+        }
+        return;
+      }
+      case INT: {
+        NullableIntVector intVector = (NullableIntVector) v;
+        Integer value = groupScan.getPartitionValue(path, column, Integer.class);
+        if (value == null) {
+          intVector.getMutator().setNull(index);
+        } else {
+          intVector.getMutator().setSafe(index, value);
+        }
+        return;
+      }
+      case SMALLINT: {
+        NullableSmallIntVector smallIntVector = (NullableSmallIntVector) v;
+        Integer value = groupScan.getPartitionValue(path, column, Integer.class);
+        if (value == null) {
+          smallIntVector.getMutator().setNull(index);
+        } else {
+          smallIntVector.getMutator().setSafe(index, value.shortValue());
+        }
+        return;
+      }
+      case TINYINT: {
+        NullableTinyIntVector tinyIntVector = (NullableTinyIntVector) v;
+        Integer value = groupScan.getPartitionValue(path, column, Integer.class);
+        if (value == null) {
+          tinyIntVector.getMutator().setNull(index);
+        } else {
+          tinyIntVector.getMutator().setSafe(index, value.byteValue());
+        }
+        return;
+      }
+      case UINT1: {
+        NullableUInt1Vector intVector = (NullableUInt1Vector) v;
+        Integer value = groupScan.getPartitionValue(path, column, Integer.class);
+        if (value == null) {
+          intVector.getMutator().setNull(index);
+        } else {
+          intVector.getMutator().setSafe(index, value.byteValue());
+        }
+        return;
+      }
+      case UINT2: {
+        NullableUInt2Vector intVector = (NullableUInt2Vector) v;
+        Integer value = groupScan.getPartitionValue(path, column, Integer.class);
+        if (value == null) {
+          intVector.getMutator().setNull(index);
+        } else {
+          intVector.getMutator().setSafe(index, (char) value.shortValue());
+        }
+        return;
+      }
+      case UINT4: {
+        NullableUInt4Vector intVector = (NullableUInt4Vector) v;
+        Integer value = groupScan.getPartitionValue(path, column, Integer.class);
+        if (value == null) {
+          intVector.getMutator().setNull(index);
+        } else {
+          intVector.getMutator().setSafe(index, value);
+        }
+        return;
+      }
+      case BIGINT: {
+        NullableBigIntVector bigIntVector = (NullableBigIntVector) v;
+        Long value = groupScan.getPartitionValue(path, column, Long.class);
+        if (value == null) {
+          bigIntVector.getMutator().setNull(index);
+        } else {
+          bigIntVector.getMutator().setSafe(index, value);
+        }
+        return;
+      }
+      case FLOAT4: {
+        NullableFloat4Vector float4Vector = (NullableFloat4Vector) v;
+        Float value = groupScan.getPartitionValue(path, column, Float.class);
+        if (value == null) {
+          float4Vector.getMutator().setNull(index);
+        } else {
+          float4Vector.getMutator().setSafe(index, value);
+        }
+        return;
+      }
+      case FLOAT8: {
+        NullableFloat8Vector float8Vector = (NullableFloat8Vector) v;
+        Double value = groupScan.getPartitionValue(path, column, Double.class);
+        if (value == null) {
+          float8Vector.getMutator().setNull(index);
+        } else {
+          float8Vector.getMutator().setSafe(index, value);
+        }
+        return;
+      }
+      case VARBINARY: {
+        NullableVarBinaryVector varBinaryVector = (NullableVarBinaryVector) v;
+        Object s = groupScan.getPartitionValue(path, column, Object.class);
+        byte[] bytes;
+        if (s == null) {
+          varBinaryVector.getMutator().setNull(index);
+          return;
+        } else {
+          bytes = getBytes(type, s);
+        }
+        varBinaryVector.getMutator().setSafe(index, bytes, 0, bytes.length);
+        return;
+      }
+      case DECIMAL18: {
+        NullableDecimal18Vector decimalVector = (NullableDecimal18Vector) v;
+        Object s = groupScan.getPartitionValue(path, column, Object.class);
+        byte[] bytes;
+        if (s == null) {
+          decimalVector.getMutator().setNull(index);
+          return;
+        } else if (s instanceof Integer) {
+          long value = DecimalUtility.getBigDecimalFromPrimitiveTypes(
+              (Integer) s,
+              majorType.getScale(),
+              majorType.getPrecision()).longValue();
+          decimalVector.getMutator().setSafe(index, value);
+          return;
+        } else if (s instanceof Long) {
+          long value = DecimalUtility.getBigDecimalFromPrimitiveTypes(
+              (Long) s,
+              majorType.getScale(),
+              majorType.getPrecision()).longValue();
+          decimalVector.getMutator().setSafe(index, value);
+          return;
+        } else {
+          bytes = getBytes(type, s);
+        }
+        long value = DecimalUtility.getBigDecimalFromByteArray(bytes, 0, bytes.length, majorType.getScale()).longValue();
+        decimalVector.getMutator().setSafe(index, value);
+        return;
+      }
+      case DATE: {
+        NullableDateVector dateVector = (NullableDateVector) v;
+        Integer value = groupScan.getPartitionValue(path, column, Integer.class);
+        if (value == null) {
+          dateVector.getMutator().setNull(index);
+        } else {
+          dateVector.getMutator().setSafe(index, value * (long) DateTimeConstants.MILLIS_PER_DAY);
+        }
+        return;
+      }
+      case TIME: {
+        NullableTimeVector timeVector = (NullableTimeVector) v;
+        Integer value = groupScan.getPartitionValue(path, column, Integer.class);
+        if (value == null) {
+          timeVector.getMutator().setNull(index);
+        } else {
+          timeVector.getMutator().setSafe(index, value);
+        }
+        return;
+      }
+      case TIMESTAMP: {
+        NullableTimeStampVector timeStampVector = (NullableTimeStampVector) v;
+        Long value = groupScan.getPartitionValue(path, column, Long.class);
+        if (value == null) {
+          timeStampVector.getMutator().setNull(index);
+        } else {
+          timeStampVector.getMutator().setSafe(index, value);
+        }
+        return;
+      }
+      case VARCHAR: {
+        NullableVarCharVector varCharVector = (NullableVarCharVector) v;
+        Object s = groupScan.getPartitionValue(path, column, Object.class);
+        byte[] bytes;
+        if (s == null) {
+          varCharVector.getMutator().setNull(index);
+          return;
+        } else {
+          bytes = getBytes(type, s);
+        }
+        varCharVector.getMutator().setSafe(index, bytes, 0, bytes.length);
+        return;
+      }
+      case INTERVAL: {
+        NullableIntervalVector intervalVector = (NullableIntervalVector) v;
+        Object s = groupScan.getPartitionValue(path, column, Object.class);
+        byte[] bytes;
+        if (s == null) {
+          intervalVector.getMutator().setNull(index);
+          return;
+        } else {
+          bytes = getBytes(type, s);
+        }
+        intervalVector.getMutator().setSafe(index, 1,
+            ParquetReaderUtility.getIntFromLEBytes(bytes, 0),
+            ParquetReaderUtility.getIntFromLEBytes(bytes, 4),
+            ParquetReaderUtility.getIntFromLEBytes(bytes, 8));
+        return;
+      }
+      default:
+        throw new UnsupportedOperationException("Unsupported type: " + type);
+    }
+  }
+
+  /**
+   * Returns the sequence of bytes received from {@code Object source}.
+   *
+   * @param type the column type
+   * @param source the source of the bytes sequence
+   * @return bytes sequence obtained from {@code Object source}
+   */
+  private byte[] getBytes(TypeProtos.MinorType type, Object source) {
+    byte[] bytes;
+    if (source instanceof Binary) {
+      bytes = ((Binary) source).getBytes();
+    } else if (source instanceof byte[]) {
+      bytes = (byte[]) source;
+    } else {
+      throw new UnsupportedOperationException("Unable to create column data for type: " + type);
+    }
+    return bytes;
+  }
+
 }

http://git-wip-us.apache.org/repos/asf/drill/blob/c6549e58/exec/java-exec/src/main/java/org/apache/drill/exec/planner/PlannerPhase.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/PlannerPhase.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/PlannerPhase.java
index f71c281..96cfa8a 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/PlannerPhase.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/PlannerPhase.java
@@ -365,6 +365,7 @@ public enum PlannerPhase {
             // Ideally this should be done in logical planning, before join order planning is done.
             // Before we can make such change, we have to figure out how to adjust the selectivity
             // estimation of filter operator, after filter is pushed down to scan.
+
             ParquetPushDownFilter.getFilterOnProject(optimizerRulesContext),
             ParquetPushDownFilter.getFilterOnScan(optimizerRulesContext)
         )
@@ -426,6 +427,9 @@ public enum PlannerPhase {
     ruleList.add(ValuesPrule.INSTANCE);
     ruleList.add(DirectScanPrule.INSTANCE);
 
+    ruleList.add(DrillPushLimitToScanRule.LIMIT_ON_PROJECT);
+    ruleList.add(DrillPushLimitToScanRule.LIMIT_ON_SCAN);
+
     if (ps.isHashAggEnabled()) {
       ruleList.add(HashAggPrule.INSTANCE);
     }

http://git-wip-us.apache.org/repos/asf/drill/blob/c6549e58/exec/java-exec/src/main/java/org/apache/drill/exec/planner/logical/DrillFilterItemStarReWriterRule.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/logical/DrillFilterItemStarReWriterRule.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/logical/DrillFilterItemStarReWriterRule.java
index 27f8c49..b7cdcfc 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/logical/DrillFilterItemStarReWriterRule.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/logical/DrillFilterItemStarReWriterRule.java
@@ -32,7 +32,7 @@ import org.apache.drill.common.expression.PathSegment;
 import org.apache.drill.common.expression.SchemaPath;
 import org.apache.drill.exec.planner.types.RelDataTypeDrillImpl;
 import org.apache.drill.exec.planner.types.RelDataTypeHolder;
-import org.apache.drill.exec.store.parquet.ParquetGroupScan;
+import org.apache.drill.exec.store.parquet.AbstractParquetGroupScan;
 
 import java.util.ArrayList;
 import java.util.Collection;
@@ -73,7 +73,7 @@ public class DrillFilterItemStarReWriterRule {
     @Override
     public boolean matches(RelOptRuleCall call) {
       DrillScanRel scan = call.rel(1);
-      return scan.getGroupScan() instanceof ParquetGroupScan && super.matches(call);
+      return scan.getGroupScan() instanceof AbstractParquetGroupScan && super.matches(call);
     }
 
     @Override
@@ -129,7 +129,7 @@ public class DrillFilterItemStarReWriterRule {
     @Override
     public boolean matches(RelOptRuleCall call) {
       DrillScanRel scan = call.rel(1);
-      return scan.getGroupScan() instanceof ParquetGroupScan && super.matches(call);
+      return scan.getGroupScan() instanceof AbstractParquetGroupScan && super.matches(call);
     }
 
     @Override
@@ -149,7 +149,7 @@ public class DrillFilterItemStarReWriterRule {
     @Override
     public boolean matches(RelOptRuleCall call) {
       DrillScanRel scan = call.rel(2);
-      return scan.getGroupScan() instanceof ParquetGroupScan && super.matches(call);
+      return scan.getGroupScan() instanceof AbstractParquetGroupScan && super.matches(call);
     }
 
     @Override

http://git-wip-us.apache.org/repos/asf/drill/blob/c6549e58/exec/java-exec/src/main/java/org/apache/drill/exec/planner/logical/partition/ParquetPruneScanRule.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/logical/partition/ParquetPruneScanRule.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/logical/partition/ParquetPruneScanRule.java
index 3153b9d..6e44383 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/logical/partition/ParquetPruneScanRule.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/logical/partition/ParquetPruneScanRule.java
@@ -19,10 +19,8 @@ package org.apache.drill.exec.planner.logical.partition;
 
 import org.apache.calcite.plan.RelOptRule;
 import org.apache.calcite.plan.RelOptRuleCall;
-import org.apache.calcite.rel.RelNode;
 import org.apache.calcite.rel.core.TableScan;
 import org.apache.drill.exec.ops.OptimizerRulesContext;
-import org.apache.drill.exec.physical.base.FileGroupScan;
 import org.apache.drill.exec.physical.base.GroupScan;
 import org.apache.drill.exec.planner.ParquetPartitionDescriptor;
 import org.apache.drill.exec.planner.PartitionDescriptor;
@@ -32,11 +30,11 @@ import org.apache.drill.exec.planner.logical.DrillScanRel;
 import org.apache.drill.exec.planner.logical.RelOptHelper;
 import org.apache.drill.exec.planner.physical.PlannerSettings;
 import org.apache.drill.exec.planner.physical.PrelUtil;
-import org.apache.drill.exec.store.parquet.ParquetGroupScan;
+import org.apache.drill.exec.store.parquet.AbstractParquetGroupScan;
 
 public class ParquetPruneScanRule {
 
-  public static final RelOptRule getFilterOnProjectParquet(OptimizerRulesContext optimizerRulesContext) {
+  public static RelOptRule getFilterOnProjectParquet(OptimizerRulesContext optimizerRulesContext) {
     return new PruneScanRule(
         RelOptHelper.some(DrillFilterRel.class, RelOptHelper.some(DrillProjectRel.class, RelOptHelper.any(DrillScanRel.class))),
         "PruneScanRule:Filter_On_Project_Parquet",
@@ -53,9 +51,9 @@ public class ParquetPruneScanRule {
         GroupScan groupScan = scan.getGroupScan();
         // this rule is applicable only for parquet based partition pruning
         if (PrelUtil.getPlannerSettings(scan.getCluster().getPlanner()).isHepPartitionPruningEnabled()) {
-          return groupScan instanceof ParquetGroupScan && groupScan.supportsPartitionFilterPushdown() && !scan.partitionFilterPushdown();
+          return groupScan instanceof AbstractParquetGroupScan && groupScan.supportsPartitionFilterPushdown() && !scan.partitionFilterPushdown();
         } else {
-          return groupScan instanceof ParquetGroupScan && groupScan.supportsPartitionFilterPushdown();
+          return groupScan instanceof AbstractParquetGroupScan && groupScan.supportsPartitionFilterPushdown();
         }
       }
 
@@ -69,7 +67,7 @@ public class ParquetPruneScanRule {
     };
   }
 
-  public static final RelOptRule getFilterOnScanParquet(OptimizerRulesContext optimizerRulesContext) {
+  public static RelOptRule getFilterOnScanParquet(OptimizerRulesContext optimizerRulesContext) {
     return new PruneScanRule(
         RelOptHelper.some(DrillFilterRel.class, RelOptHelper.any(DrillScanRel.class)),
         "PruneScanRule:Filter_On_Scan_Parquet", optimizerRulesContext) {
@@ -85,9 +83,9 @@ public class ParquetPruneScanRule {
         GroupScan groupScan = scan.getGroupScan();
         // this rule is applicable only for parquet based partition pruning
         if (PrelUtil.getPlannerSettings(scan.getCluster().getPlanner()).isHepPartitionPruningEnabled()) {
-          return groupScan instanceof ParquetGroupScan && groupScan.supportsPartitionFilterPushdown() && !scan.partitionFilterPushdown();
+          return groupScan instanceof AbstractParquetGroupScan && groupScan.supportsPartitionFilterPushdown() && !scan.partitionFilterPushdown();
         } else {
-          return groupScan instanceof ParquetGroupScan && groupScan.supportsPartitionFilterPushdown();
+          return groupScan instanceof AbstractParquetGroupScan && groupScan.supportsPartitionFilterPushdown();
         }
       }
 

http://git-wip-us.apache.org/repos/asf/drill/blob/c6549e58/exec/java-exec/src/main/java/org/apache/drill/exec/planner/logical/partition/PruneScanRule.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/logical/partition/PruneScanRule.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/logical/partition/PruneScanRule.java
index 5f679a4..7fa1794 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/logical/partition/PruneScanRule.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/logical/partition/PruneScanRule.java
@@ -24,7 +24,6 @@ import java.util.Map;
 import java.util.concurrent.TimeUnit;
 
 import com.google.common.base.Stopwatch;
-
 import org.apache.calcite.adapter.enumerable.EnumerableTableScan;
 import org.apache.calcite.rel.core.Filter;
 import org.apache.calcite.rel.core.Project;
@@ -147,8 +146,8 @@ public abstract class PruneScanRule extends StoragePluginOptimizerRule {
   protected void doOnMatch(RelOptRuleCall call, Filter filterRel, Project projectRel, TableScan scanRel) {
 
     final String pruningClassName = getClass().getName();
-    logger.info("Beginning partition pruning, pruning class: {}", pruningClassName);
-    Stopwatch totalPruningTime = Stopwatch.createStarted();
+    logger.debug("Beginning partition pruning, pruning class: {}", pruningClassName);
+    Stopwatch totalPruningTime = logger.isDebugEnabled() ? Stopwatch.createStarted() : null;
 
     final PlannerSettings settings = PrelUtil.getPlannerSettings(call.getPlanner());
     PartitionDescriptor descriptor = getPartitionDescriptor(settings, scanRel);
@@ -191,30 +190,33 @@ public abstract class PruneScanRule extends StoragePluginOptimizerRule {
     }
 
     if (partitionColumnBitSet.isEmpty()) {
-      logger.info("No partition columns are projected from the scan..continue. " +
-          "Total pruning elapsed time: {} ms", totalPruningTime.elapsed(TimeUnit.MILLISECONDS));
+      if (totalPruningTime != null) {
+        logger.debug("No partition columns are projected from the scan..continue. Total pruning elapsed time: {} ms",
+            totalPruningTime.elapsed(TimeUnit.MILLISECONDS));
+      }
       setPruneStatus(metaContext, PruneStatus.NOT_PRUNED);
       return;
     }
 
     // stop watch to track how long we spend in different phases of pruning
-    Stopwatch miscTimer = Stopwatch.createUnstarted();
-
-    // track how long we spend building the filter tree
-    miscTimer.start();
+    // first track how long we spend building the filter tree
+    Stopwatch miscTimer = logger.isDebugEnabled() ? Stopwatch.createStarted() : null;
 
     FindPartitionConditions c = new FindPartitionConditions(columnBitset, filterRel.getCluster().getRexBuilder());
     c.analyze(condition);
     RexNode pruneCondition = c.getFinalCondition();
     BitSet referencedDirsBitSet = c.getReferencedDirs();
 
-    logger.info("Total elapsed time to build and analyze filter tree: {} ms",
-        miscTimer.elapsed(TimeUnit.MILLISECONDS));
-    miscTimer.reset();
+    if (miscTimer != null) {
+      logger.debug("Total elapsed time to build and analyze filter tree: {} ms", miscTimer.elapsed(TimeUnit.MILLISECONDS));
+      miscTimer.reset();
+    }
 
     if (pruneCondition == null) {
-      logger.info("No conditions were found eligible for partition pruning." +
-          "Total pruning elapsed time: {} ms", totalPruningTime.elapsed(TimeUnit.MILLISECONDS));
+      if (totalPruningTime != null) {
+        logger.debug("No conditions were found eligible for partition pruning. Total pruning elapsed time: {} ms",
+            totalPruningTime.elapsed(TimeUnit.MILLISECONDS));
+      }
       setPruneStatus(metaContext, PruneStatus.NOT_PRUNED);
       return;
     }
@@ -251,15 +253,19 @@ public abstract class PruneScanRule extends StoragePluginOptimizerRule {
           container.add(v);
         }
 
-        // track how long we spend populating partition column vectors
-        miscTimer.start();
+        if (miscTimer != null) {
+          // track how long we spend populating partition column vectors
+          miscTimer.start();
+        }
 
         // populate partition vectors.
         descriptor.populatePartitionVectors(vectors, partitions, partitionColumnBitSet, fieldNameMap);
 
-        logger.info("Elapsed time to populate partitioning column vectors: {} ms within batchIndex: {}",
-            miscTimer.elapsed(TimeUnit.MILLISECONDS), batchIndex);
-        miscTimer.reset();
+        if (miscTimer != null) {
+          logger.debug("Elapsed time to populate partitioning column vectors: {} ms within batchIndex: {}",
+              miscTimer.elapsed(TimeUnit.MILLISECONDS), batchIndex);
+          miscTimer.reset();
+        }
 
         // materialize the expression; only need to do this once
         if (batchIndex == 0) {
@@ -267,8 +273,9 @@ public abstract class PruneScanRule extends StoragePluginOptimizerRule {
           if (materializedExpr == null) {
             // continue without partition pruning; no need to log anything here since
             // materializePruneExpr logs it already
-            logger.info("Total pruning elapsed time: {} ms",
-                totalPruningTime.elapsed(TimeUnit.MILLISECONDS));
+            if (totalPruningTime != null) {
+              logger.debug("Total pruning elapsed time: {} ms", totalPruningTime.elapsed(TimeUnit.MILLISECONDS));
+            }
             setPruneStatus(metaContext, PruneStatus.NOT_PRUNED);
             return;
           }
@@ -276,14 +283,18 @@ public abstract class PruneScanRule extends StoragePluginOptimizerRule {
 
         output.allocateNew(partitions.size());
 
-        // start the timer to evaluate how long we spend in the interpreter evaluation
-        miscTimer.start();
+        if (miscTimer != null) {
+          // start the timer to evaluate how long we spend in the interpreter evaluation
+          miscTimer.start();
+        }
 
         InterpreterEvaluator.evaluate(partitions.size(), optimizerContext, container, output, materializedExpr);
 
-        logger.info("Elapsed time in interpreter evaluation: {} ms within batchIndex: {} with # of partitions : {}",
-            miscTimer.elapsed(TimeUnit.MILLISECONDS), batchIndex, partitions.size());
-        miscTimer.reset();
+        if (miscTimer != null) {
+          logger.debug("Elapsed time in interpreter evaluation: {} ms within batchIndex: {} with # of partitions : {}",
+              miscTimer.elapsed(TimeUnit.MILLISECONDS), batchIndex, partitions.size());
+          miscTimer.reset();
+        }
 
         int recordCount = 0;
         int qualifiedCount = 0;
@@ -338,7 +349,9 @@ public abstract class PruneScanRule extends StoragePluginOptimizerRule {
         batchIndex++;
       } catch (Exception e) {
         logger.warn("Exception while trying to prune partition.", e);
-        logger.info("Total pruning elapsed time: {} ms", totalPruningTime.elapsed(TimeUnit.MILLISECONDS));
+        if (totalPruningTime != null) {
+          logger.debug("Total pruning elapsed time: {} ms", totalPruningTime.elapsed(TimeUnit.MILLISECONDS));
+        }
 
         setPruneStatus(metaContext, PruneStatus.NOT_PRUNED);
         return; // continue without partition pruning
@@ -352,7 +365,7 @@ public abstract class PruneScanRule extends StoragePluginOptimizerRule {
 
     try {
       if (newPartitions.size() == numTotal) {
-        logger.info("No partitions were eligible for pruning");
+        logger.debug("No partitions were eligible for pruning");
         return;
       }
 
@@ -371,7 +384,7 @@ public abstract class PruneScanRule extends StoragePluginOptimizerRule {
         // directories first and the non-composite partition location will still return
         // directories, not files.  So, additional processing is done depending on this flag
         wasAllPartitionsPruned = true;
-        logger.info("All {} partitions were pruned; added back a single partition to allow creating a schema", numTotal);
+        logger.debug("All {} partitions were pruned; added back a single partition to allow creating a schema", numTotal);
 
         // set the cacheFileRoot appropriately
         if (firstLocation.isCompositePartition()) {
@@ -379,7 +392,7 @@ public abstract class PruneScanRule extends StoragePluginOptimizerRule {
         }
       }
 
-      logger.info("Pruned {} partitions down to {}", numTotal, newPartitions.size());
+      logger.debug("Pruned {} partitions down to {}", numTotal, newPartitions.size());
 
       List<RexNode> conjuncts = RelOptUtil.conjunctions(condition);
       List<RexNode> pruneConjuncts = RelOptUtil.conjunctions(pruneCondition);
@@ -439,7 +452,9 @@ public abstract class PruneScanRule extends StoragePluginOptimizerRule {
     } catch (Exception e) {
       logger.warn("Exception while using the pruned partitions.", e);
     } finally {
-      logger.info("Total pruning elapsed time: {} ms", totalPruningTime.elapsed(TimeUnit.MILLISECONDS));
+      if (totalPruningTime != null) {
+        logger.debug("Total pruning elapsed time: {} ms", totalPruningTime.elapsed(TimeUnit.MILLISECONDS));
+      }
     }
   }
 

http://git-wip-us.apache.org/repos/asf/drill/blob/c6549e58/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/handlers/RefreshMetadataHandler.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/handlers/RefreshMetadataHandler.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/handlers/RefreshMetadataHandler.java
index 6bfceb4..f463e6d 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/handlers/RefreshMetadataHandler.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/handlers/RefreshMetadataHandler.java
@@ -37,7 +37,7 @@ import org.apache.drill.exec.store.dfs.FileSelection;
 import org.apache.drill.exec.store.dfs.FileSystemPlugin;
 import org.apache.drill.exec.store.dfs.FormatSelection;
 import org.apache.drill.exec.store.dfs.NamedFormatPluginConfig;
-import org.apache.drill.exec.store.parquet.Metadata;
+import org.apache.drill.exec.store.parquet.metadata.Metadata;
 import org.apache.drill.exec.store.parquet.ParquetFormatConfig;
 import org.apache.drill.exec.work.foreman.ForemanSetupException;
 import org.apache.hadoop.fs.Path;

http://git-wip-us.apache.org/repos/asf/drill/blob/c6549e58/exec/java-exec/src/main/java/org/apache/drill/exec/store/ColumnExplorer.java
----------------------------------------------------------------------
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 fb86783..5ab67f1 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
@@ -17,24 +17,25 @@
  */
 package org.apache.drill.exec.store;
 
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
 
-import org.apache.commons.lang3.ArrayUtils;
 import org.apache.drill.common.expression.SchemaPath;
 import org.apache.drill.common.map.CaseInsensitiveMap;
 import org.apache.drill.exec.ExecConstants;
 import org.apache.drill.exec.server.options.OptionManager;
 import org.apache.drill.exec.server.options.OptionValue;
 import org.apache.drill.exec.store.dfs.FileSelection;
-import org.apache.drill.exec.store.dfs.easy.FileWork;
 import org.apache.drill.exec.util.Utilities;
 import org.apache.hadoop.fs.Path;
 
 import com.google.common.collect.Lists;
-import com.google.common.collect.Maps;
 import com.google.common.io.Files;
 
 public class ColumnExplorer {
@@ -156,43 +157,74 @@ public class ColumnExplorer {
   }
 
   /**
-   * Compares selection root and actual file path to determine partition columns values.
-   * Adds implicit file columns according to columns list.
+   * Creates map with implicit columns where key is column name, value is columns actual value.
+   * This map contains partition and implicit file columns (if requested).
+   * Partition columns names are formed based in partition designator and value index.
    *
-   * @return map with columns names as keys and their values
+   * @param filePath file path, used to populate file implicit columns
+   * @param partitionValues list of partition values
+   * @param includeFileImplicitColumns if file implicit columns should be included into the result
+   * @return implicit columns map
    */
-  public Map<String, String> populateImplicitColumns(FileWork work, String selectionRoot) {
-    return populateImplicitColumns(work.getPath(), selectionRoot);
-  }
+  public Map<String, String> populateImplicitColumns(String filePath,
+                                                     List<String> partitionValues,
+                                                     boolean includeFileImplicitColumns) {
+    Map<String, String> implicitValues = new LinkedHashMap<>();
 
-  /**
-   * Compares selection root and actual file path to determine partition columns values.
-   * Adds implicit file columns according to columns list.
-   *
-   * @return map with columns names as keys and their values
-   */
-  public Map<String, String> populateImplicitColumns(String filePath, String selectionRoot) {
-    Map<String, String> implicitValues = Maps.newLinkedHashMap();
-    if (selectionRoot != null) {
-      String[] r = Path.getPathWithoutSchemeAndAuthority(new Path(selectionRoot)).toString().split("/");
-      Path path = Path.getPathWithoutSchemeAndAuthority(new Path(filePath));
-      String[] p = path.toString().split("/");
-      if (p.length > r.length) {
-        String[] q = ArrayUtils.subarray(p, r.length, p.length - 1);
-        for (int a = 0; a < q.length; a++) {
-          if (isStarQuery || selectedPartitionColumns.contains(a)) {
-            implicitValues.put(partitionDesignator + a, q[a]);
-          }
-        }
+    for (int i = 0; i < partitionValues.size(); i++) {
+      if (isStarQuery || selectedPartitionColumns.contains(i)) {
+        implicitValues.put(partitionDesignator + i, partitionValues.get(i));
       }
-      //add implicit file columns
+    }
+
+    if (includeFileImplicitColumns) {
+      Path path = Path.getPathWithoutSchemeAndAuthority(new Path(filePath));
       for (Map.Entry<String, ImplicitFileColumns> entry : selectedImplicitColumns.entrySet()) {
         implicitValues.put(entry.getKey(), entry.getValue().getValue(path));
       }
     }
+
     return implicitValues;
   }
 
+  /**
+   * Compares root and file path to determine directories
+   * that are present in the file path but absent in root.
+   * Example: root - a/b/c, filePath - a/b/c/d/e/0_0_0.parquet, result - d/e.
+   * Stores different directory names in the list in successive order.
+   *
+   *
+   * @param filePath file path
+   * @param root root directory
+   * @return list of directory names
+   */
+  public static List<String> listPartitionValues(String filePath, String root) {
+    if (filePath == null || root == null) {
+      return Collections.emptyList();
+    }
+
+    int rootDepth = new Path(root).depth();
+    Path path = new Path(filePath);
+    int parentDepth = path.getParent().depth();
+
+    int diffCount = parentDepth - rootDepth;
+
+    if (diffCount < 1) {
+      return Collections.emptyList();
+    }
+
+    String[] diffDirectoryNames = new String[diffCount];
+
+    // start filling in array from the end
+    for (int i = rootDepth; parentDepth > i; i++) {
+      path = path.getParent();
+      // place in the end of array
+      diffDirectoryNames[parentDepth - i - 1] = path.getName();
+    }
+
+    return Arrays.asList(diffDirectoryNames);
+  }
+
   public boolean isStarQuery() {
     return isStarQuery;
   }

http://git-wip-us.apache.org/repos/asf/drill/blob/c6549e58/exec/java-exec/src/main/java/org/apache/drill/exec/store/TimedRunnable.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/store/TimedRunnable.java b/exec/java-exec/src/main/java/org/apache/drill/exec/store/TimedRunnable.java
index fe0cae1..7cce2ad 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/store/TimedRunnable.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/store/TimedRunnable.java
@@ -20,18 +20,15 @@ package org.apache.drill.exec.store;
 import java.io.IOException;
 import java.util.List;
 import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.ExecutionException;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
-import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
-import java.util.concurrent.TimeoutException;
 
+import com.google.common.base.Stopwatch;
 import org.apache.drill.common.concurrent.ExtendedLatch;
 import org.apache.drill.common.exceptions.UserException;
 import org.slf4j.Logger;
 
-import com.google.common.base.Stopwatch;
 import com.google.common.collect.Lists;
 
 /**
@@ -115,7 +112,7 @@ public abstract class TimedRunnable<V> implements Runnable {
    * @throws IOException All exceptions are coerced to IOException since this was build for storage system tasks initially.
    */
   public static <V> List<V> run(final String activity, final Logger logger, final List<TimedRunnable<V>> runnables, int parallelism) throws IOException {
-    Stopwatch watch = Stopwatch.createStarted();
+    Stopwatch watch = logger.isDebugEnabled() ? Stopwatch.createStarted() : null;
     long timedRunnableStart=System.nanoTime();
     if(runnables.size() == 1){
       parallelism = 1;
@@ -186,21 +183,22 @@ public abstract class TimedRunnable<V> implements Runnable {
       }
     }
 
-    if(logger.isInfoEnabled()){
+    if (watch != null) {
       double avg = (sum/1000.0/1000.0)/(count*1.0d);
       double avgStart = (totalStart/1000.0)/(count*1.0d);
 
-      logger.info(
+      logger.debug(
           String.format("%s: Executed %d out of %d using %d threads. "
               + "Time: %dms total, %fms avg, %dms max.",
               activity, count, runnables.size(), parallelism, watch.elapsed(TimeUnit.MILLISECONDS), avg, max/1000/1000));
-      logger.info(
+      logger.debug(
               String.format("%s: Executed %d out of %d using %d threads. "
                               + "Earliest start: %f \u03BCs, Latest start: %f \u03BCs, Average start: %f \u03BCs .",
                       activity, count, runnables.size(), parallelism, earliestStart/1000.0, latestStart/1000.0, avgStart));
+      watch.stop();
     }
 
-    if(excep != null) {
+    if (excep != null) {
       throw excep;
     }
 

http://git-wip-us.apache.org/repos/asf/drill/blob/c6549e58/exec/java-exec/src/main/java/org/apache/drill/exec/store/dfs/FileSelection.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/store/dfs/FileSelection.java b/exec/java-exec/src/main/java/org/apache/drill/exec/store/dfs/FileSelection.java
index 7edb327..f5bcced 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/store/dfs/FileSelection.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/store/dfs/FileSelection.java
@@ -95,7 +95,7 @@ public class FileSelection {
       final String cacheFileRoot, final boolean wasAllPartitionsPruned, final StatusType dirStatus) {
     this.statuses = statuses;
     this.files = files;
-    this.selectionRoot = Preconditions.checkNotNull(selectionRoot);
+    this.selectionRoot = selectionRoot;
     this.dirStatus = dirStatus;
     this.cacheFileRoot = cacheFileRoot;
     this.wasAllPartitionsPruned = wasAllPartitionsPruned;
@@ -121,7 +121,7 @@ public class FileSelection {
   }
 
   public List<FileStatus> getStatuses(final DrillFileSystem fs) throws IOException {
-    Stopwatch timer = Stopwatch.createStarted();
+    Stopwatch timer = logger.isDebugEnabled() ? Stopwatch.createStarted() : null;
 
     if (statuses == null)  {
       final List<FileStatus> newStatuses = Lists.newArrayList();
@@ -130,8 +130,11 @@ public class FileSelection {
       }
       statuses = newStatuses;
     }
-    logger.info("FileSelection.getStatuses() took {} ms, numFiles: {}",
-        timer.elapsed(TimeUnit.MILLISECONDS), statuses == null ? 0 : statuses.size());
+    if (timer != null) {
+      logger.debug("FileSelection.getStatuses() took {} ms, numFiles: {}",
+          timer.elapsed(TimeUnit.MILLISECONDS), statuses == null ? 0 : statuses.size());
+      timer.stop();
+    }
 
     return statuses;
   }
@@ -164,7 +167,7 @@ public class FileSelection {
     if (isExpandedFully()) {
       return this;
     }
-    Stopwatch timer = Stopwatch.createStarted();
+    Stopwatch timer = logger.isDebugEnabled() ? Stopwatch.createStarted() : null;
     List<FileStatus> statuses = getStatuses(fs);
 
     List<FileStatus> nonDirectories = Lists.newArrayList();
@@ -173,8 +176,10 @@ public class FileSelection {
     }
 
     final FileSelection fileSel = create(nonDirectories, null, selectionRoot);
-    logger.debug("FileSelection.minusDirectories() took {} ms, numFiles: {}",
-        timer.elapsed(TimeUnit.MILLISECONDS), statuses.size());
+    if (timer != null) {
+      logger.debug("FileSelection.minusDirectories() took {} ms, numFiles: {}", timer.elapsed(TimeUnit.MILLISECONDS), statuses.size());
+      timer.stop();
+    }
 
     // fileSel will be null if we query an empty folder
     if (fileSel != null) {
@@ -259,7 +264,7 @@ public class FileSelection {
 
   public static FileSelection create(final DrillFileSystem fs, final String parent, final String path,
       final boolean allowAccessOutsideWorkspace) throws IOException {
-    Stopwatch timer = Stopwatch.createStarted();
+    Stopwatch timer = logger.isDebugEnabled() ? Stopwatch.createStarted() : null;
     boolean hasWildcard = path.contains(WILD_CARD);
 
     final Path combined = new Path(parent, removeLeadingSlash(path));
@@ -271,7 +276,10 @@ public class FileSelection {
       return null;
     }
     final FileSelection fileSel = create(Lists.newArrayList(statuses), null, combined.toUri().getPath());
-    logger.debug("FileSelection.create() took {} ms ", timer.elapsed(TimeUnit.MILLISECONDS));
+    if (timer != null) {
+      logger.debug("FileSelection.create() took {} ms ", timer.elapsed(TimeUnit.MILLISECONDS));
+      timer.stop();
+    }
     if (fileSel == null) {
       return null;
     }
@@ -322,7 +330,7 @@ public class FileSelection {
 
   public static FileSelection createFromDirectories(final List<String> dirPaths, final FileSelection selection,
       final String cacheFileRoot) {
-    Stopwatch timer = Stopwatch.createStarted();
+    Stopwatch timer = logger.isDebugEnabled() ? Stopwatch.createStarted() : null;
     final String root = selection.getSelectionRoot();
     if (Strings.isNullOrEmpty(root)) {
       throw new DrillRuntimeException("Selection root is null or empty" + root);
@@ -338,9 +346,7 @@ public class FileSelection {
         dirs.add(status.getPath().toString());
       }
     } else {
-      for (String s : dirPaths) {
-        dirs.add(s);
-      }
+      dirs.addAll(dirPaths);
     }
 
     final Path rootPath = handleWildCard(root);
@@ -349,7 +355,10 @@ public class FileSelection {
     final Path path = new Path(uri.getScheme(), uri.getAuthority(), rootPath.toUri().getPath());
     FileSelection fileSel = new FileSelection(null, dirs, path.toString(), cacheFileRoot, false);
     fileSel.setHadWildcard(selection.hadWildcard());
-    logger.info("FileSelection.createFromDirectories() took {} ms ", timer.elapsed(TimeUnit.MILLISECONDS));
+    if (timer != null) {
+      logger.debug("FileSelection.createFromDirectories() took {} ms ", timer.elapsed(TimeUnit.MILLISECONDS));
+      timer.stop();
+    }
     return fileSel;
   }
 

http://git-wip-us.apache.org/repos/asf/drill/blob/c6549e58/exec/java-exec/src/main/java/org/apache/drill/exec/store/dfs/ReadEntryFromHDFS.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/store/dfs/ReadEntryFromHDFS.java b/exec/java-exec/src/main/java/org/apache/drill/exec/store/dfs/ReadEntryFromHDFS.java
index e02d841..15107ac 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/store/dfs/ReadEntryFromHDFS.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/store/dfs/ReadEntryFromHDFS.java
@@ -22,14 +22,14 @@ import org.apache.drill.exec.store.dfs.easy.FileWork;
 import com.fasterxml.jackson.annotation.JsonCreator;
 import com.fasterxml.jackson.annotation.JsonProperty;
 
-public class ReadEntryFromHDFS extends ReadEntryWithPath implements FileWork{
+public class ReadEntryFromHDFS extends ReadEntryWithPath implements FileWork {
 
   private long start;
   private long length;
 
   @JsonCreator
   public ReadEntryFromHDFS(@JsonProperty("path") String path,@JsonProperty("start") long start, @JsonProperty("length") long length) {
-    this.path = path;
+    super(path);
     this.start = start;
     this.length = length;
   }

http://git-wip-us.apache.org/repos/asf/drill/blob/c6549e58/exec/java-exec/src/main/java/org/apache/drill/exec/store/dfs/easy/EasyFormatPlugin.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/store/dfs/easy/EasyFormatPlugin.java b/exec/java-exec/src/main/java/org/apache/drill/exec/store/dfs/easy/EasyFormatPlugin.java
index 678569f..5af1091 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/store/dfs/easy/EasyFormatPlugin.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/store/dfs/easy/EasyFormatPlugin.java
@@ -147,10 +147,12 @@ public abstract class EasyFormatPlugin<T extends FormatPluginConfig> implements
     List<RecordReader> readers = new LinkedList<>();
     List<Map<String, String>> implicitColumns = Lists.newArrayList();
     Map<String, String> mapWithMaxColumns = Maps.newLinkedHashMap();
-    for(FileWork work : scan.getWorkUnits()){
+    boolean supportsFileImplicitColumns = scan.getSelectionRoot() != null;
+    for (FileWork work : scan.getWorkUnits()){
       RecordReader recordReader = getRecordReader(context, dfs, work, scan.getColumns(), scan.getUserName());
       readers.add(recordReader);
-      Map<String, String> implicitValues = columnExplorer.populateImplicitColumns(work, scan.getSelectionRoot());
+      List<String> partitionValues = ColumnExplorer.listPartitionValues(work.getPath(), scan.getSelectionRoot());
+      Map<String, String> implicitValues = columnExplorer.populateImplicitColumns(work.getPath(), partitionValues, supportsFileImplicitColumns);
       implicitColumns.add(implicitValues);
       if (implicitValues.size() > mapWithMaxColumns.size()) {
         mapWithMaxColumns = implicitValues;

http://git-wip-us.apache.org/repos/asf/drill/blob/c6549e58/exec/java-exec/src/main/java/org/apache/drill/exec/store/ischema/InfoSchemaGroupScan.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/store/ischema/InfoSchemaGroupScan.java b/exec/java-exec/src/main/java/org/apache/drill/exec/store/ischema/InfoSchemaGroupScan.java
index f1da0f7..aa3f4ae 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/store/ischema/InfoSchemaGroupScan.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/store/ischema/InfoSchemaGroupScan.java
@@ -70,7 +70,7 @@ public class InfoSchemaGroupScan extends AbstractGroupScan{
   }
 
   @JsonProperty("filter")
-  public InfoSchemaFilter getFilter() {
+  public InfoSchemaFilter getSchemaFilter() {
     return filter;
   }
 


Mime
View raw message