/*
 * Licensed 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 com.facebook.presto.localfile;

import com.facebook.presto.spi.PrestoException;
import com.fasterxml.jackson.annotation.JsonCreator;
import com.fasterxml.jackson.annotation.JsonProperty;
import com.google.common.collect.ImmutableList;

import java.io.File;
import java.util.Arrays;
import java.util.List;
import java.util.Optional;
import java.util.regex.Pattern;

import static com.facebook.presto.localfile.LocalFileErrorCode.LOCAL_FILE_ERROR_CODE;
import static com.google.common.base.Preconditions.checkArgument;
import static com.google.common.base.Preconditions.checkState;
import static java.lang.String.format;
import static java.util.Objects.requireNonNull;

final class DataLocation
{
    private final File location;
    private final Optional<String> pattern;
    private final Optional<Pattern> compiledPattern;

    @SuppressWarnings("ResultOfMethodCallIgnored")
    @JsonCreator
    public DataLocation(
            @JsonProperty("location") String location,
            @JsonProperty("pattern") Optional<String> pattern)
    {
        requireNonNull(location, "location is null");
        requireNonNull(pattern, "pattern is null");

        File file = new File(location);
        if (!file.exists() && pattern.isPresent()) {
            file.mkdirs();
        }

        checkArgument(file.exists(), "location does not exist");
        if (pattern.isPresent() && !file.isDirectory()) {
            throw new IllegalArgumentException("pattern may be specified only if location is a directory");
        }

        this.location = file;
        this.pattern = (!pattern.isPresent() && file.isDirectory()) ? Optional.of(".*") : pattern;
        this.compiledPattern = file.isDirectory() ? Optional.of(Pattern.compile(pattern.orElse(".*"))) : Optional.empty();
    }

    @JsonProperty
    public File getLocation()
    {
        return location;
    }

    @JsonProperty
    public Optional<String> getPattern()
    {
        return pattern;
    }

    public List<File> files()
    {
        checkState(location.exists(), "location %s doesn't exist", location);
        if (!pattern.isPresent()) {
            return ImmutableList.of(location);
        }

        checkState(location.isDirectory(), "location %s is not a directory", location);
        File[] files = location.listFiles(file -> compiledPattern.get().matcher(file.getName()).matches());

        if (files == null) {
            throw new PrestoException(LOCAL_FILE_ERROR_CODE, format("Failed to list files at %s", location));
        }

        Arrays.sort(files, (o1, o2) -> Long.compare(o2.lastModified(), o1.lastModified()));
        return ImmutableList.copyOf(files);
    }
}

@@ -836,19 +836,87 @@
 ava.
-util.Arrays
+io.IOException;%0Aimport java.nio.file.DirectoryStream;%0Aimport java.nio.file.Path
 ;%0Aim
@@ -984,21 +984,25 @@
 til.
-regex.Pattern
+stream.Collectors
 ;%0A%0Ai
@@ -1254,16 +1254,70 @@
 format;%0A
+import static java.nio.file.Files.newDirectoryStream;%0A
 import s
@@ -1464,61 +1464,8 @@
 ern;
-%0A    private final Optional%3CPattern%3E compiledPattern;
 %0A%0A  
@@ -2272,17 +2272,16 @@
 nal.of(%22
-.
 *%22) : pa
@@ -2291,131 +2291,8 @@
 rn;%0A
-        this.compiledPattern = file.isDirectory() ? Optional.of(Pattern.compile(pattern.orElse(%22.*%22))) : Optional.empty();%0A
     
@@ -2772,24 +2772,25 @@
 ation);%0A
+%0A
         
 File%5B%5D f
@@ -2785,138 +2785,364 @@
     
-File%5B%5D files = location.listFiles(file -%3E compiledPattern.get().matcher(file.getName()).matches());%0A%0A        if (files == null) %7B%0A
+try (DirectoryStream%3CPath%3E paths = newDirectoryStream(location.toPath(), pattern.get())) %7B%0A            ImmutableList.Builder%3CFile%3E builder = ImmutableList.%3CFile%3Ebuilder();%0A            for (Path path : paths) %7B%0A                builder.add(path.toFile());%0A            %7D%0A            List%3CFile%3E files = builder.build();%0A%0A            if (files.isEmpty()) %7B%0A    
     
@@ -3247,34 +3247,37 @@
 tion));%0A        
+    
 %7D
-%0A
 %0A        Arrays.
@@ -3273,27 +3273,62 @@
     
-Arrays.sort(files, 
+    return files.stream()%0A                    .sorted(
 (o1,
@@ -3390,53 +3390,231 @@
 ()))
-;%0A        return ImmutableList.copyOf(files);
+%0A                    .collect(Collectors.toList());%0A        %7D%0A        catch (IOException e) %7B%0A            throw new PrestoException(LOCAL_FILE_ERROR_CODE, format(%22Error getting list of files at %25s%22, location), e);%0A        %7D
 %0A   

Update localfile connector to use glob for file name patterns