public static List<String> listMatchedFilesWithRecursiveOption(PinotFS pinotFs, URI fileUri,
      @Nullable String includePattern, @Nullable String excludePattern, boolean searchRecursively)
      throws Exception {
    String[] files;
    // listFiles throws IOException
    files = pinotFs.listFiles(fileUri, searchRecursively);
    //TODO: sort input files based on creation time
    PathMatcher includeFilePathMatcher = null;
    if (includePattern != null) {
      includeFilePathMatcher = FileSystems.getDefault().getPathMatcher(includePattern);
    }
    PathMatcher excludeFilePathMatcher = null;
    if (excludePattern != null) {
      excludeFilePathMatcher = FileSystems.getDefault().getPathMatcher(excludePattern);
    }
    List<String> filteredFiles = new ArrayList<>();
    for (String file : files) {
      if (includeFilePathMatcher != null) {
        if (!includeFilePathMatcher.matches(Paths.get(file))) {
          continue;
        }
      }
      if (excludeFilePathMatcher != null) {
        if (excludeFilePathMatcher.matches(Paths.get(file))) {
          continue;
        }
      }
      if (!pinotFs.isDirectory(new URI(sanitizeURIString(file)))) {
        // In case PinotFS implementations list files without a scheme (e.g. hdfs://), then we may lose it in the
        // input file path. Call SegmentGenerationUtils.getFileURI() to fix this up.
        // getFileURI throws URISyntaxException
        filteredFiles.add(SegmentGenerationUtils.getFileURI(file, fileUri).toString());
      }
    }
    if (filteredFiles.isEmpty()) {
      throw new RuntimeException(String.format(
          "No file found in the input directory: %s matching includeFileNamePattern: %s,"
              + " excludeFileNamePattern: %s", fileUri, includePattern, excludePattern));
    }
    return filteredFiles;
  }
@Test
  public void testMatchFilesRecursiveSearchOnRecursiveInputFilePattern()
      throws Exception {
    File testDir = makeTestDir();
    File inputDir = new File(testDir, "input");
    File inputSubDir1 = new File(inputDir, "2009");
    inputSubDir1.mkdirs();

    File inputFile1 = new File(inputDir, "input.csv");
    FileUtils.writeLines(inputFile1, Lists.newArrayList("col1,col2", "value1,1", "value2,2"));

    File inputFile2 = new File(inputSubDir1, "input.csv");
    FileUtils.writeLines(inputFile2, Lists.newArrayList("col1,col2", "value3,3", "value4,4"));
    URI inputDirURI = new URI(inputDir.getAbsolutePath());
    if (inputDirURI.getScheme() == null) {
      inputDirURI = new File(inputDir.getAbsolutePath()).toURI();
    }
    PinotFS inputDirFS = PinotFSFactory.create(inputDirURI.getScheme());
    String includePattern = "glob:" + inputDir.getAbsolutePath() + "/**.csv";
    List<String> files =
        SegmentGenerationUtils.listMatchedFilesWithRecursiveOption(inputDirFS, inputDirURI, includePattern, null, true);
    Assert.assertEquals(files.size(), 2);
  }