1 files changed, 25 insertions, 21 deletions

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 d5d9784b6..33b500018 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,6 +17,7 @@
  */
 package org.apache.drill.exec.store;
 
+import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.LinkedHashMap;
@@ -133,20 +134,20 @@ public class ColumnExplorer {
   public static List<String> getPartitionColumnNames(FileSelection selection, SchemaConfig schemaConfig) {
     int partitionsCount = 0;
     // a depth of table root path
-    int rootDepth = new Path(selection.getSelectionRoot()).depth();
+    int rootDepth = selection.getSelectionRoot().depth();
 
-    for (String file : selection.getFiles()) {
+    for (Path file : selection.getFiles()) {
       // Calculates partitions count for the concrete file:
       // depth of file path - depth of table root path - 1.
       // The depth of file path includes file itself,
       // so we should subtract 1 to consider only directories.
-      int currentPartitionsCount = new Path(file).depth() - rootDepth - 1;
+      int currentPartitionsCount = file.depth() - rootDepth - 1;
       // max depth of files path should be used to handle all partitions
       partitionsCount = Math.max(partitionsCount, currentPartitionsCount);
     }
 
     String partitionColumnLabel = schemaConfig.getOption(ExecConstants.FILESYSTEM_PARTITION_COLUMN_LABEL).string_val;
-    List<String> partitions = Lists.newArrayList();
+    List<String> partitions = new ArrayList<>();
 
     // generates partition column names: dir0, dir1 etc.
     for (int i = 0; i < partitionsCount; i++) {
@@ -165,7 +166,7 @@ public class ColumnExplorer {
    * @param includeFileImplicitColumns if file implicit columns should be included into the result
    * @return implicit columns map
    */
-  public Map<String, String> populateImplicitColumns(String filePath,
+  public Map<String, String> populateImplicitColumns(Path filePath,
                                                      List<String> partitionValues,
                                                      boolean includeFileImplicitColumns) {
     Map<String, String> implicitValues = new LinkedHashMap<>();
@@ -177,7 +178,7 @@ public class ColumnExplorer {
     }
 
     if (includeFileImplicitColumns) {
-      Path path = Path.getPathWithoutSchemeAndAuthority(new Path(filePath));
+      Path path = Path.getPathWithoutSchemeAndAuthority(filePath);
       for (Map.Entry<String, ImplicitFileColumns> entry : selectedImplicitColumns.entrySet()) {
         implicitValues.put(entry.getKey(), entry.getValue().getValue(path));
       }
@@ -189,15 +190,16 @@ public class ColumnExplorer {
   /**
    * 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.
+   * Example: root - a/b/c, file - 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 file file path
    * @param root root directory
+   * @param hasDirsOnly whether it is file or directory
    * @return list of directory names
    */
-  public static List<String> listPartitionValues(String filePath, String root) {
-    String[] dirs = parsePartitions(filePath, root);
+  public static List<String> listPartitionValues(Path file, Path root, boolean hasDirsOnly) {
+    String[] dirs = parsePartitions(file, root, hasDirsOnly);
     if (dirs == null) {
       return Collections.emptyList();
     }
@@ -208,21 +210,23 @@ public class ColumnExplorer {
    * Low-level parse of partitions, returned as a string array. Returns a
    * null array for invalid values.
    *
-   * @param filePath file path
+   * @param file file path
    * @param root root directory
+   * @param hasDirsOnly whether it is file or directory
    * @return array of directory names, or null if the arguments are invalid
    */
-  public static String[] parsePartitions(String filePath, String root) {
-    if (filePath == null || root == null) {
+  public static String[] parsePartitions(Path file, Path root, boolean hasDirsOnly) {
+    if (file == null || root == null) {
       return null;
     }
 
-    int rootDepth = new Path(root).depth();
-    Path path = new Path(filePath);
-    int parentDepth = path.getParent().depth();
-
-    int diffCount = parentDepth - rootDepth;
+    if (!hasDirsOnly) {
+      file = file.getParent();
+    }
 
+    int rootDepth = root.depth();
+    int fileDepth = file.depth();
+    int diffCount = fileDepth - rootDepth;
     if (diffCount < 0) {
       return null;
     }
@@ -230,10 +234,10 @@ public class ColumnExplorer {
     String[] diffDirectoryNames = new String[diffCount];
 
     // start filling in array from the end
-    for (int i = rootDepth; parentDepth > i; i++) {
-      path = path.getParent();
+    for (int i = rootDepth; fileDepth > i; i++) {
       // place in the end of array
-      diffDirectoryNames[parentDepth - i - 1] = path.getName();
+      diffDirectoryNames[fileDepth - i - 1] = file.getName();
+      file = file.getParent();
     }
 
     return diffDirectoryNames;