DRILL-2618: handle queries over empty folders consistently so that th…

…ey report table not found rather than failing.

Refactor FileSelection to eliminate redundancy, make it more managable
Fix WorkspaceSchemaFactory to handle empty folders.
Introduce ParquetFileSelection, a sub-class of FileSelection that carries along metadata cache
Fix MagicStringMatcher so that it operate on files only.
Unit test file selection

common/src/main/java/org/apache/drill/common/util/TestTools.java

@@ -49,4 +49,27 @@ public static String getWorkingPath() {
     return WORKING_PATH;
   }
 
+  private static final String PATH_SEPARATOR = System.getProperty("file.separator");
+  private static final String[] STRUCTURE = {"drill", "exec", "java-exec", "src", "test", "resources"};
+
+  /**
+   * Returns fully qualified path where test resources reside if current working directory is at any level in the
+   * following root->exec->java-exec->src->test->resources, throws an {@link IllegalStateException} otherwise.
+   */
+  public static String getTestResourcesPath() {
+    final StringBuilder builder = new StringBuilder(WORKING_PATH);
+    for (int i=0; i< STRUCTURE.length; i++) {
+      if (WORKING_PATH.endsWith(STRUCTURE[i])) {
+        for (int j=i+1; j< STRUCTURE.length; j++) {
+          builder.append(PATH_SEPARATOR).append(STRUCTURE[j]);
+        }
+        return builder.toString();
+      }
+    }
+    final String msg = String.format("Unable to recognize working directory[%s]. The workspace must be root or exec " +
+        "module.", WORKING_PATH);
+    throw new IllegalStateException(msg);
+  }
+
+
 }

exec/java-exec/src/main/java/org/apache/drill/exec/planner/FileSystemPartitionDescriptor.java

@@ -84,8 +84,8 @@ public int getMaxHierarchyLevel() {
 
   @Override
   public GroupScan createNewGroupScan(List<String> newFiles) throws IOException {
-    final FileSelection newFileSelection = new FileSelection(newFiles, getBaseTableLocation(), true);
-    final FileGroupScan newScan = ((FileGroupScan)scanRel.getGroupScan()).clone(newFileSelection);
+    final FileSelection newSelection = FileSelection.create(null, newFiles, getBaseTableLocation());
+    final FileGroupScan newScan = ((FileGroupScan)scanRel.getGroupScan()).clone(newSelection);
     return newScan;
   }
 

exec/java-exec/src/main/java/org/apache/drill/exec/planner/ParquetPartitionDescriptor.java

@@ -46,7 +46,6 @@ public class ParquetPartitionDescriptor extends AbstractPartitionDescriptor {
 
   private final List<SchemaPath> partitionColumns;
   private final DrillScanRel scanRel;
-  static final int MAX_NESTED_SUBDIRS = 10;
 
   public ParquetPartitionDescriptor(PlannerSettings settings, DrillScanRel scanRel) {
     ParquetGroupScan scan = (ParquetGroupScan) scanRel.getGroupScan();
@@ -81,8 +80,8 @@ public int getMaxHierarchyLevel() {
 
   @Override
   public GroupScan createNewGroupScan(List<String> newFiles) throws IOException {
-    final FileSelection newFileSelection = new FileSelection(newFiles, getBaseTableLocation(), true);
-    final FileGroupScan newScan = ((FileGroupScan)scanRel.getGroupScan()).clone(newFileSelection);
+    final FileSelection newSelection = FileSelection.create(null, newFiles, getBaseTableLocation());
+    final FileGroupScan newScan = ((FileGroupScan)scanRel.getGroupScan()).clone(newSelection);
     return newScan;
   }
 

exec/java-exec/src/main/java/org/apache/drill/exec/store/dfs/BasicFormatMatcher.java

@@ -138,9 +138,19 @@ public MagicStringMatcher(List<MagicString> magicStrings) {
     }
 
     public boolean matches(DrillFileSystem fs, FileStatus status) throws IOException{
-      if (ranges.isEmpty()) {
+      if (ranges.isEmpty() || status.isDirectory()) {
         return false;
       }
+      // walk all the way down in the symlinks until a hard entry is reached
+      FileStatus current = status;
+      while (current.isSymlink()) {
+        current = fs.getFileStatus(status.getSymlink());
+      }
+      // if hard entry is not a file nor can it be a symlink then it is not readable simply deny matching.
+      if (!current.isFile()) {
+        return false;
+      }
+
       final Range<Long> fileRange = Range.closedOpen( 0L, status.getLen());
 
       try (FSDataInputStream is = fs.open(status.getPath())) {