Add path_match_pattern option

This option is used to skip files if path name doesn't match with
pattern.

Author: frsyuki

README.md

@@ -53,6 +53,9 @@
 
     - **session_token**: session token (string, required)
 
+* **path_match_pattern**: regexp to match file paths. If a file path doesn't match with this pattern, the file will be skipped (regexp string, optional)
+
+* **total_file_count_limit**: maximum number of files to read (integer, optional)
 
 ## Example
 
@@ -66,6 +69,22 @@ in:
   secret_access_key: AbCxYz123aBcXyZ123
 ```
 
+To skip files using regexp:
+
+```yaml
+in:
+  type: s3
+  bucket: my-s3-bucket
+  path_prefix: logs/csv-
+  # ...
+  path_match_pattern: \.csv$   # a file will be skipped if its path doesn't match with this pattern
+
+  ## some examples of regexp:
+  #path_match_pattern: /archive/         # match files in .../archive/... directory
+  #path_match_pattern: /data1/|/data2/   # match files in .../data1/... or .../data2/... directory
+  #path_match_pattern: .csv$|.csv.gz$    # match files whose suffix is .csv or .csv.gz
+```
+
 To use AWS_ACCESS_KEY_ID and AWS_SECRET_ACCESS_KEY environment variables:
 
 ```yaml
@@ -100,4 +119,4 @@ To run unit tests, we need to configure the following environment variables.
 EMBULK_S3_TEST_BUCKET
 EMBULK_S3_TEST_ACCESS_KEY_ID
 EMBULK_S3_TEST_SECRET_ACCESS_KEY
-```
+```

embulk-input-s3/src/main/java/org/embulk/input/s3/FileList.java

@@ -5,6 +5,8 @@
 import java.util.ArrayList;
 import java.util.zip.GZIPInputStream;
 import java.util.zip.GZIPOutputStream;
+import java.util.regex.Pattern;
+import java.util.regex.Matcher;
 import java.io.InputStream;
 import java.io.OutputStream;
 import java.io.BufferedOutputStream;
@@ -29,6 +31,10 @@ public class FileList
 {
     public interface Task
     {
+        @Config("path_match_pattern")
+        @ConfigDefault("\".*\"")
+        String getPathMatchPattern();
+
         @Config("total_file_count_limit")
         @ConfigDefault("2147483647")
         int getTotalFileCountLimit();
@@ -63,17 +69,21 @@ public static class Builder
         private String last = null;
 
         private int limitCount = Integer.MAX_VALUE;
+        private Pattern pathMatchPattern;
+
         private final ByteBuffer castBuffer = ByteBuffer.allocate(4);
 
         public Builder(Task task)
         {
             this();
             this.limitCount = task.getTotalFileCountLimit();
+            this.pathMatchPattern = Pattern.compile(task.getPathMatchPattern());
         }
 
         public Builder(ConfigSource config)
         {
             this();
+            this.pathMatchPattern = Pattern.compile(config.get(String.class, "path_match_pattern", ".*"));
             this.limitCount = config.get(int.class, "total_file_count_limit", Integer.MAX_VALUE);
         }
 
@@ -94,6 +104,12 @@ public Builder limitTotalFileCount(int limitCount)
             return this;
         }
 
+        public Builder pathMatchPattern(String pattern)
+        {
+            this.pathMatchPattern = Pattern.compile(pattern);
+            return this;
+        }
+
         public int size()
         {
             return entries.size();
@@ -104,6 +120,7 @@ public boolean needsMore()
             return size() < limitCount;
         }
 
+        // returns true if this file is used
         public synchronized boolean add(String path, long size)
         {
             // TODO throw IllegalStateException if stream is already closed
@@ -112,8 +129,9 @@ public synchronized boolean add(String path, long size)
                 return false;
             }
 
-            // TODO in the future, support some other filtering parameters (file name suffix filter, regex filter, etc)
-            //      and return false if filtered out.
+            if (!pathMatchPattern.matcher(path).matches()) {
+                return false;
+            }
 
             int index = entries.size();
             entries.add(new Entry(index, size));

embulk-input-s3/src/test/java/org/embulk/input/s3/TestS3FileInputPlugin.java

@@ -115,6 +115,17 @@ public void useTotalFileCountLimit()
         assertEquals(0, getRecords(config, output).size());
     }
 
+    @Test
+    public void usePathMatchPattern()
+            throws Exception
+    {
+        ConfigSource config = this.config.deepCopy().set("path_match_pattern", "/match/");
+        ConfigDiff configDiff = runner.transaction(config, new Control(runner, output));
+
+        assertNull(configDiff.get(String.class, "last_path"));
+        assertEquals(0, getRecords(config, output).size());
+    }
+
     static class Control
             implements InputPlugin.Control
     {