beam-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From j...@apache.org
Subject [1/2] beam git commit: Moves Match into FileIO.match()/matchAll()
Date Sun, 03 Sep 2017 23:30:49 GMT
Repository: beam
Updated Branches:
  refs/heads/master f7d4583bd -> 80b9cf9c2


Moves Match into FileIO.match()/matchAll()

FileIO will later gain other methods, such as read()/write().

Also introduces FileIO.MatchConfiguration - a common type to use
by various file-based IOs to reduce boilerplate, and uses it in TextIO.


Project: http://git-wip-us.apache.org/repos/asf/beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/beam/commit/58e8a01b
Tree: http://git-wip-us.apache.org/repos/asf/beam/tree/58e8a01b
Diff: http://git-wip-us.apache.org/repos/asf/beam/diff/58e8a01b

Branch: refs/heads/master
Commit: 58e8a01b351a80ca233983c8ccfd4b2699c86a3a
Parents: f7d4583
Author: Eugene Kirpichov <kirpichov@google.com>
Authored: Thu Aug 24 16:31:41 2017 -0700
Committer: Eugene Kirpichov <ekirpichov@gmail.com>
Committed: Sun Sep 3 16:14:54 2017 -0700

----------------------------------------------------------------------
 .../java/org/apache/beam/sdk/io/AvroIO.java     | 232 +++++++---------
 .../java/org/apache/beam/sdk/io/FileIO.java     | 265 +++++++++++++++++++
 .../main/java/org/apache/beam/sdk/io/Match.java | 156 -----------
 .../beam/sdk/io/ReadAllViaFileBasedSource.java  |   2 +-
 .../java/org/apache/beam/sdk/io/TextIO.java     | 147 ++++------
 .../java/org/apache/beam/sdk/io/AvroIOTest.java |   4 +-
 .../org/apache/beam/sdk/io/TextIOReadTest.java  |   5 +-
 7 files changed, 418 insertions(+), 393 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/beam/blob/58e8a01b/sdks/java/core/src/main/java/org/apache/beam/sdk/io/AvroIO.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/io/AvroIO.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/io/AvroIO.java
index 6eeeac9..c4711e8 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/io/AvroIO.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/io/AvroIO.java
@@ -19,7 +19,6 @@ package org.apache.beam.sdk.io;
 
 import static com.google.common.base.Preconditions.checkArgument;
 import static com.google.common.base.Preconditions.checkNotNull;
-import static org.apache.beam.sdk.transforms.Watch.Growth.ignoreInput;
 
 import com.google.auto.value.AutoValue;
 import com.google.common.annotations.VisibleForTesting;
@@ -40,6 +39,7 @@ import org.apache.beam.sdk.coders.Coder;
 import org.apache.beam.sdk.coders.CoderRegistry;
 import org.apache.beam.sdk.coders.StringUtf8Coder;
 import org.apache.beam.sdk.io.FileBasedSink.FilenamePolicy;
+import org.apache.beam.sdk.io.FileIO.MatchConfiguration;
 import org.apache.beam.sdk.io.fs.EmptyMatchTreatment;
 import org.apache.beam.sdk.io.fs.ResourceId;
 import org.apache.beam.sdk.options.ValueProvider;
@@ -266,7 +266,7 @@ public class AvroIO {
    */
   public static <T> Read<T> read(Class<T> recordClass) {
     return new AutoValue_AvroIO_Read.Builder<T>()
-        .setEmptyMatchTreatment(EmptyMatchTreatment.DISALLOW)
+        .setMatchConfiguration(MatchConfiguration.create(EmptyMatchTreatment.DISALLOW))
         .setRecordClass(recordClass)
         .setSchema(ReflectData.get().getSchema(recordClass))
         .setHintMatchesManyFiles(false)
@@ -276,7 +276,7 @@ public class AvroIO {
   /** Like {@link #read}, but reads each filepattern in the input {@link PCollection}. */
   public static <T> ReadAll<T> readAll(Class<T> recordClass) {
     return new AutoValue_AvroIO_ReadAll.Builder<T>()
-        .setEmptyMatchTreatment(EmptyMatchTreatment.ALLOW_IF_WILDCARD)
+        .setMatchConfiguration(MatchConfiguration.create(EmptyMatchTreatment.ALLOW_IF_WILDCARD))
         .setRecordClass(recordClass)
         .setSchema(ReflectData.get().getSchema(recordClass))
         // 64MB is a reasonable value that allows to amortize the cost of opening files,
@@ -289,7 +289,7 @@ public class AvroIO {
   /** Reads Avro file(s) containing records of the specified schema. */
   public static Read<GenericRecord> readGenericRecords(Schema schema) {
     return new AutoValue_AvroIO_Read.Builder<GenericRecord>()
-        .setEmptyMatchTreatment(EmptyMatchTreatment.DISALLOW)
+        .setMatchConfiguration(MatchConfiguration.create(EmptyMatchTreatment.DISALLOW))
         .setRecordClass(GenericRecord.class)
         .setSchema(schema)
         .setHintMatchesManyFiles(false)
@@ -302,7 +302,7 @@ public class AvroIO {
    */
   public static ReadAll<GenericRecord> readAllGenericRecords(Schema schema) {
     return new AutoValue_AvroIO_ReadAll.Builder<GenericRecord>()
-        .setEmptyMatchTreatment(EmptyMatchTreatment.ALLOW_IF_WILDCARD)
+        .setMatchConfiguration(MatchConfiguration.create(EmptyMatchTreatment.ALLOW_IF_WILDCARD))
         .setRecordClass(GenericRecord.class)
         .setSchema(schema)
         .setDesiredBundleSizeBytes(64 * 1024 * 1024L)
@@ -331,7 +331,7 @@ public class AvroIO {
    */
   public static <T> Parse<T> parseGenericRecords(SerializableFunction<GenericRecord, T> parseFn) {
     return new AutoValue_AvroIO_Parse.Builder<T>()
-        .setEmptyMatchTreatment(EmptyMatchTreatment.DISALLOW)
+        .setMatchConfiguration(MatchConfiguration.create(EmptyMatchTreatment.DISALLOW))
         .setParseFn(parseFn)
         .setHintMatchesManyFiles(false)
         .build();
@@ -344,7 +344,7 @@ public class AvroIO {
   public static <T> ParseAll<T> parseAllGenericRecords(
       SerializableFunction<GenericRecord, T> parseFn) {
     return new AutoValue_AvroIO_ParseAll.Builder<T>()
-        .setEmptyMatchTreatment(EmptyMatchTreatment.ALLOW_IF_WILDCARD)
+        .setMatchConfiguration(MatchConfiguration.create(EmptyMatchTreatment.ALLOW_IF_WILDCARD))
         .setParseFn(parseFn)
         .setDesiredBundleSizeBytes(64 * 1024 * 1024L)
         .build();
@@ -425,9 +425,7 @@ public class AvroIO {
   @AutoValue
   public abstract static class Read<T> extends PTransform<PBegin, PCollection<T>> {
     @Nullable abstract ValueProvider<String> getFilepattern();
-    abstract EmptyMatchTreatment getEmptyMatchTreatment();
-    @Nullable abstract Duration getWatchForNewFilesInterval();
-    @Nullable abstract TerminationCondition<?, ?> getWatchForNewFilesTerminationCondition();
+    abstract MatchConfiguration getMatchConfiguration();
     @Nullable abstract Class<T> getRecordClass();
     @Nullable abstract Schema getSchema();
     abstract boolean getHintMatchesManyFiles();
@@ -437,10 +435,7 @@ public class AvroIO {
     @AutoValue.Builder
     abstract static class Builder<T> {
       abstract Builder<T> setFilepattern(ValueProvider<String> filepattern);
-      abstract Builder<T> setEmptyMatchTreatment(EmptyMatchTreatment emptyMatchTreatment);
-      abstract Builder<T> setWatchForNewFilesInterval(Duration watchForNewFilesInterval);
-      abstract Builder<T> setWatchForNewFilesTerminationCondition(
-              TerminationCondition<?, ?> condition);
+      abstract Builder<T> setMatchConfiguration(MatchConfiguration matchConfiguration);
       abstract Builder<T> setRecordClass(Class<T> recordClass);
       abstract Builder<T> setSchema(Schema schema);
       abstract Builder<T> setHintMatchesManyFiles(boolean hintManyFiles);
@@ -463,11 +458,15 @@ public class AvroIO {
       return from(StaticValueProvider.of(filepattern));
     }
 
-    /**
-     * Configures whether or not a filepattern matching no files is allowed.
-     */
+
+    /** Sets the {@link MatchConfiguration}. */
+    public Read<T> withMatchConfiguration(MatchConfiguration matchConfiguration) {
+      return toBuilder().setMatchConfiguration(matchConfiguration).build();
+    }
+
+    /** Configures whether or not a filepattern matching no files is allowed. */
     public Read<T> withEmptyMatchTreatment(EmptyMatchTreatment treatment) {
-      return toBuilder().setEmptyMatchTreatment(treatment).build();
+      return withMatchConfiguration(getMatchConfiguration().withEmptyMatchTreatment(treatment));
     }
 
     /**
@@ -476,16 +475,12 @@ public class AvroIO {
      * is unbounded.
      *
      * <p>This works only in runners supporting {@link Kind#SPLITTABLE_DO_FN}.
-     *
-     * @see TerminationCondition
      */
     @Experimental(Kind.SPLITTABLE_DO_FN)
     public Read<T> watchForNewFiles(
-        Duration pollInterval, TerminationCondition<?, ?> terminationCondition) {
-      return toBuilder()
-          .setWatchForNewFilesInterval(pollInterval)
-          .setWatchForNewFilesTerminationCondition(terminationCondition)
-          .build();
+        Duration pollInterval, TerminationCondition<String, ?> terminationCondition) {
+      return withMatchConfiguration(
+              getMatchConfiguration().continuously(pollInterval, terminationCondition));
     }
 
     /**
@@ -506,12 +501,15 @@ public class AvroIO {
       checkNotNull(getFilepattern(), "filepattern");
       checkNotNull(getSchema(), "schema");
 
-      if (getWatchForNewFilesInterval() == null && !getHintMatchesManyFiles()) {
+      if (getMatchConfiguration().getWatchInterval() == null && !getHintMatchesManyFiles()) {
         return input.apply(
             "Read",
             org.apache.beam.sdk.io.Read.from(
                 createSource(
-                    getFilepattern(), getEmptyMatchTreatment(), getRecordClass(), getSchema())));
+                    getFilepattern(),
+                    getMatchConfiguration().getEmptyMatchTreatment(),
+                    getRecordClass(),
+                    getSchema())));
       }
       // All other cases go through ReadAll.
 
@@ -519,12 +517,7 @@ public class AvroIO {
           (getRecordClass() == GenericRecord.class)
               ? (ReadAll<T>) readAllGenericRecords(getSchema())
               : readAll(getRecordClass());
-      readAll = readAll.withEmptyMatchTreatment(getEmptyMatchTreatment());
-      if (getWatchForNewFilesInterval() != null) {
-        TerminationCondition<String, ?> readAllCondition =
-            ignoreInput(getWatchForNewFilesTerminationCondition());
-        readAll = readAll.watchForNewFiles(getWatchForNewFilesInterval(), readAllCondition);
-      }
+      readAll = readAll.withMatchConfiguration(getMatchConfiguration());
       return input
           .apply("Create filepattern", Create.ofProvider(getFilepattern(), StringUtf8Coder.of()))
           .apply("Via ReadAll", readAll);
@@ -536,12 +529,7 @@ public class AvroIO {
       builder
           .addIfNotNull(
               DisplayData.item("filePattern", getFilepattern()).withLabel("Input File Pattern"))
-          .add(
-              DisplayData.item("emptyMatchTreatment", getEmptyMatchTreatment().toString())
-                  .withLabel("Treatment of filepatterns that match no files"))
-          .addIfNotNull(
-              DisplayData.item("watchForNewFilesInterval", getWatchForNewFilesInterval())
-                  .withLabel("Interval to watch for new files"));
+          .include("matchConfiguration", getMatchConfiguration());
     }
 
     @SuppressWarnings("unchecked")
@@ -563,9 +551,7 @@ public class AvroIO {
   /** Implementation of {@link #readAll}. */
   @AutoValue
   public abstract static class ReadAll<T> extends PTransform<PCollection<String>, PCollection<T>> {
-    abstract EmptyMatchTreatment getEmptyMatchTreatment();
-    @Nullable abstract Duration getWatchForNewFilesInterval();
-    @Nullable abstract TerminationCondition<String, ?> getWatchForNewFilesTerminationCondition();
+    abstract MatchConfiguration getMatchConfiguration();
     @Nullable abstract Class<T> getRecordClass();
     @Nullable abstract Schema getSchema();
     abstract long getDesiredBundleSizeBytes();
@@ -574,10 +560,7 @@ public class AvroIO {
 
     @AutoValue.Builder
     abstract static class Builder<T> {
-      abstract Builder<T> setEmptyMatchTreatment(EmptyMatchTreatment emptyMatchTreatment);
-      abstract Builder<T> setWatchForNewFilesInterval(Duration watchForNewFilesInterval);
-      abstract Builder<T> setWatchForNewFilesTerminationCondition(
-              TerminationCondition<String, ?> condition);
+      abstract Builder<T> setMatchConfiguration(MatchConfiguration matchConfiguration);
       abstract Builder<T> setRecordClass(Class<T> recordClass);
       abstract Builder<T> setSchema(Schema schema);
       abstract Builder<T> setDesiredBundleSizeBytes(long desiredBundleSizeBytes);
@@ -585,19 +568,23 @@ public class AvroIO {
       abstract ReadAll<T> build();
     }
 
+
+    /** Sets the {@link MatchConfiguration}. */
+    public ReadAll<T> withMatchConfiguration(MatchConfiguration configuration) {
+      return toBuilder().setMatchConfiguration(configuration).build();
+    }
+
     /** Like {@link Read#withEmptyMatchTreatment}. */
     public ReadAll<T> withEmptyMatchTreatment(EmptyMatchTreatment treatment) {
-      return toBuilder().setEmptyMatchTreatment(treatment).build();
+      return withMatchConfiguration(getMatchConfiguration().withEmptyMatchTreatment(treatment));
     }
 
     /** Like {@link Read#watchForNewFiles}. */
     @Experimental(Kind.SPLITTABLE_DO_FN)
     public ReadAll<T> watchForNewFiles(
         Duration pollInterval, TerminationCondition<String, ?> terminationCondition) {
-      return toBuilder()
-              .setWatchForNewFilesInterval(pollInterval)
-              .setWatchForNewFilesTerminationCondition(terminationCondition)
-              .build();
+      return withMatchConfiguration(
+          getMatchConfiguration().continuously(pollInterval, terminationCondition));
     }
 
     @VisibleForTesting
@@ -608,48 +595,30 @@ public class AvroIO {
     @Override
     public PCollection<T> expand(PCollection<String> input) {
       checkNotNull(getSchema(), "schema");
-      Match.Filepatterns matchFilepatterns =
-          Match.filepatterns().withEmptyMatchTreatment(getEmptyMatchTreatment());
-      if (getWatchForNewFilesInterval() != null) {
-        matchFilepatterns =
-            matchFilepatterns.continuously(
-                getWatchForNewFilesInterval(), getWatchForNewFilesTerminationCondition());
-      }
-
       return input
-          .apply(matchFilepatterns)
+          .apply(FileIO.matchAll().withConfiguration(getMatchConfiguration()))
           .apply(
               "Read all via FileBasedSource",
               new ReadAllViaFileBasedSource<>(
                   SerializableFunctions.<String, Boolean>constant(true) /* isSplittable */,
                   getDesiredBundleSizeBytes(),
-                  new CreateSourceFn<>(
-                      getEmptyMatchTreatment(), getRecordClass(), getSchema().toString())))
+                  new CreateSourceFn<>(getRecordClass(), getSchema().toString())))
           .setCoder(AvroCoder.of(getRecordClass(), getSchema()));
     }
 
     @Override
     public void populateDisplayData(DisplayData.Builder builder) {
       super.populateDisplayData(builder);
-      builder
-          .add(
-              DisplayData.item("emptyMatchTreatment", getEmptyMatchTreatment().toString())
-                  .withLabel("Treatment of filepatterns that match no files"))
-          .addIfNotNull(
-              DisplayData.item("watchForNewFilesInterval", getWatchForNewFilesInterval())
-                  .withLabel("Interval to watch for new files"));
+      builder.include("matchConfiguration", getMatchConfiguration());
     }
   }
 
   private static class CreateSourceFn<T>
       implements SerializableFunction<String, FileBasedSource<T>> {
-    private final EmptyMatchTreatment emptyMatchTreatment;
     private final Class<T> recordClass;
     private final Supplier<Schema> schemaSupplier;
 
-    public CreateSourceFn(
-        EmptyMatchTreatment emptyMatchTreatment, Class<T> recordClass, String jsonSchema) {
-      this.emptyMatchTreatment = emptyMatchTreatment;
+    public CreateSourceFn(Class<T> recordClass, String jsonSchema) {
       this.recordClass = recordClass;
       this.schemaSupplier = AvroUtils.serializableSchemaSupplier(jsonSchema);
     }
@@ -657,7 +626,10 @@ public class AvroIO {
     @Override
     public FileBasedSource<T> apply(String input) {
       return Read.createSource(
-          StaticValueProvider.of(input), emptyMatchTreatment, recordClass, schemaSupplier.get());
+          StaticValueProvider.of(input),
+          EmptyMatchTreatment.DISALLOW,
+          recordClass,
+          schemaSupplier.get());
     }
   }
 
@@ -667,9 +639,7 @@ public class AvroIO {
   @AutoValue
   public abstract static class Parse<T> extends PTransform<PBegin, PCollection<T>> {
     @Nullable abstract ValueProvider<String> getFilepattern();
-    abstract EmptyMatchTreatment getEmptyMatchTreatment();
-    @Nullable abstract Duration getWatchForNewFilesInterval();
-    @Nullable abstract TerminationCondition<?, ?> getWatchForNewFilesTerminationCondition();
+    abstract MatchConfiguration getMatchConfiguration();
     abstract SerializableFunction<GenericRecord, T> getParseFn();
     @Nullable abstract Coder<T> getCoder();
     abstract boolean getHintMatchesManyFiles();
@@ -679,10 +649,7 @@ public class AvroIO {
     @AutoValue.Builder
     abstract static class Builder<T> {
       abstract Builder<T> setFilepattern(ValueProvider<String> filepattern);
-      abstract Builder<T> setEmptyMatchTreatment(EmptyMatchTreatment emptyMatchTreatment);
-      abstract Builder<T> setWatchForNewFilesInterval(Duration watchForNewFilesInterval);
-      abstract Builder<T> setWatchForNewFilesTerminationCondition(
-              TerminationCondition<?, ?> condition);
+      abstract Builder<T> setMatchConfiguration(MatchConfiguration matchConfiguration);
       abstract Builder<T> setParseFn(SerializableFunction<GenericRecord, T> parseFn);
       abstract Builder<T> setCoder(Coder<T> coder);
       abstract Builder<T> setHintMatchesManyFiles(boolean hintMatchesManyFiles);
@@ -700,19 +667,22 @@ public class AvroIO {
       return toBuilder().setFilepattern(filepattern).build();
     }
 
+    /** Sets the {@link MatchConfiguration}. */
+    public Parse<T> withMatchConfiguration(MatchConfiguration configuration) {
+      return toBuilder().setMatchConfiguration(configuration).build();
+    }
+
     /** Like {@link Read#withEmptyMatchTreatment}. */
     public Parse<T> withEmptyMatchTreatment(EmptyMatchTreatment treatment) {
-      return toBuilder().setEmptyMatchTreatment(treatment).build();
+      return withMatchConfiguration(getMatchConfiguration().withEmptyMatchTreatment(treatment));
     }
 
     /** Like {@link Read#watchForNewFiles}. */
     @Experimental(Kind.SPLITTABLE_DO_FN)
     public Parse<T> watchForNewFiles(
-        Duration pollInterval, TerminationCondition<?, ?> terminationCondition) {
-      return toBuilder()
-          .setWatchForNewFilesInterval(pollInterval)
-          .setWatchForNewFilesTerminationCondition(terminationCondition)
-          .build();
+        Duration pollInterval, TerminationCondition<String, ?> terminationCondition) {
+      return withMatchConfiguration(
+          getMatchConfiguration().continuously(pollInterval, terminationCondition));
     }
 
     /** Sets a coder for the result of the parse function. */
@@ -730,24 +700,19 @@ public class AvroIO {
       checkNotNull(getFilepattern(), "filepattern");
       Coder<T> coder = inferCoder(getCoder(), getParseFn(), input.getPipeline().getCoderRegistry());
 
-      if (getWatchForNewFilesInterval() == null && !getHintMatchesManyFiles()) {
+      if (getMatchConfiguration().getWatchInterval() == null && !getHintMatchesManyFiles()) {
         return input.apply(
                 org.apache.beam.sdk.io.Read.from(
                         AvroSource.from(getFilepattern()).withParseFn(getParseFn(), coder)));
       }
       // All other cases go through ParseAllGenericRecords.
-      ParseAll<T> parseAll =
-          parseAllGenericRecords(getParseFn())
-              .withCoder(coder)
-              .withEmptyMatchTreatment(getEmptyMatchTreatment());
-      if (getWatchForNewFilesInterval() != null) {
-        TerminationCondition<String, ?> parseAllCondition =
-            ignoreInput(getWatchForNewFilesTerminationCondition());
-        parseAll = parseAll.watchForNewFiles(getWatchForNewFilesInterval(), parseAllCondition);
-      }
       return input
           .apply("Create filepattern", Create.ofProvider(getFilepattern(), StringUtf8Coder.of()))
-          .apply("Via ParseAll", parseAll);
+          .apply(
+              "Via ParseAll",
+              parseAllGenericRecords(getParseFn())
+                  .withCoder(coder)
+                  .withMatchConfiguration(getMatchConfiguration()));
     }
 
     private static <T> Coder<T> inferCoder(
@@ -776,12 +741,7 @@ public class AvroIO {
           .addIfNotNull(
               DisplayData.item("filePattern", getFilepattern()).withLabel("Input File Pattern"))
           .add(DisplayData.item("parseFn", getParseFn().getClass()).withLabel("Parse function"))
-          .add(
-              DisplayData.item("emptyMatchTreatment", getEmptyMatchTreatment().toString())
-                  .withLabel("Treatment of filepatterns that match no files"))
-          .addIfNotNull(
-              DisplayData.item("watchForNewFilesInterval", getWatchForNewFilesInterval())
-                  .withLabel("Interval to watch for new files"));
+          .include("matchConfiguration", getMatchConfiguration());
     }
   }
 
@@ -790,9 +750,7 @@ public class AvroIO {
   /** Implementation of {@link #parseAllGenericRecords}. */
   @AutoValue
   public abstract static class ParseAll<T> extends PTransform<PCollection<String>, PCollection<T>> {
-    abstract EmptyMatchTreatment getEmptyMatchTreatment();
-    @Nullable abstract Duration getWatchForNewFilesInterval();
-    @Nullable abstract TerminationCondition<String, ?> getWatchForNewFilesTerminationCondition();
+    abstract MatchConfiguration getMatchConfiguration();
     abstract SerializableFunction<GenericRecord, T> getParseFn();
     @Nullable abstract Coder<T> getCoder();
     abstract long getDesiredBundleSizeBytes();
@@ -801,10 +759,7 @@ public class AvroIO {
 
     @AutoValue.Builder
     abstract static class Builder<T> {
-      abstract Builder<T> setEmptyMatchTreatment(EmptyMatchTreatment emptyMatchTreatment);
-      abstract Builder<T> setWatchForNewFilesInterval(Duration watchForNewFilesInterval);
-      abstract Builder<T> setWatchForNewFilesTerminationCondition(
-          TerminationCondition<String, ?> condition);
+      abstract Builder<T> setMatchConfiguration(MatchConfiguration matchConfiguration);
       abstract Builder<T> setParseFn(SerializableFunction<GenericRecord, T> parseFn);
       abstract Builder<T> setCoder(Coder<T> coder);
       abstract Builder<T> setDesiredBundleSizeBytes(long desiredBundleSizeBytes);
@@ -812,19 +767,22 @@ public class AvroIO {
       abstract ParseAll<T> build();
     }
 
+    /** Sets the {@link MatchConfiguration}. */
+    public ParseAll<T> withMatchConfiguration(MatchConfiguration configuration) {
+      return toBuilder().setMatchConfiguration(configuration).build();
+    }
+
     /** Like {@link Read#withEmptyMatchTreatment}. */
     public ParseAll<T> withEmptyMatchTreatment(EmptyMatchTreatment treatment) {
-      return toBuilder().setEmptyMatchTreatment(treatment).build();
+      return withMatchConfiguration(getMatchConfiguration().withEmptyMatchTreatment(treatment));
     }
 
     /** Like {@link Read#watchForNewFiles}. */
     @Experimental(Kind.SPLITTABLE_DO_FN)
     public ParseAll<T> watchForNewFiles(
         Duration pollInterval, TerminationCondition<String, ?> terminationCondition) {
-      return toBuilder()
-          .setWatchForNewFilesInterval(pollInterval)
-          .setWatchForNewFilesTerminationCondition(terminationCondition)
-          .build();
+      return withMatchConfiguration(
+              getMatchConfiguration().continuously(pollInterval, terminationCondition));
     }
 
     /** Specifies the coder for the result of the {@code parseFn}. */
@@ -842,25 +800,10 @@ public class AvroIO {
       final Coder<T> coder =
           Parse.inferCoder(getCoder(), getParseFn(), input.getPipeline().getCoderRegistry());
       final SerializableFunction<GenericRecord, T> parseFn = getParseFn();
-      final EmptyMatchTreatment emptyMatchTreatment = getEmptyMatchTreatment();
       final SerializableFunction<String, FileBasedSource<T>> createSource =
-          new SerializableFunction<String, FileBasedSource<T>>() {
-            @Override
-            public FileBasedSource<T> apply(String input) {
-              return AvroSource.from(input)
-                  .withParseFn(parseFn, coder)
-                  .withEmptyMatchTreatment(emptyMatchTreatment);
-            }
-          };
-      Match.Filepatterns matchFilepatterns =
-              Match.filepatterns().withEmptyMatchTreatment(emptyMatchTreatment);
-      if (getWatchForNewFilesInterval() != null) {
-        matchFilepatterns =
-                matchFilepatterns.continuously(
-                        getWatchForNewFilesInterval(), getWatchForNewFilesTerminationCondition());
-      }
+              new CreateParseSourceFn<>(parseFn, coder);
       return input
-          .apply(matchFilepatterns)
+          .apply(FileIO.matchAll().withConfiguration(getMatchConfiguration()))
           .apply(
               "Parse all via FileBasedSource",
               new ReadAllViaFileBasedSource<>(
@@ -875,12 +818,23 @@ public class AvroIO {
       super.populateDisplayData(builder);
       builder
           .add(DisplayData.item("parseFn", getParseFn().getClass()).withLabel("Parse function"))
-          .add(
-              DisplayData.item("emptyMatchTreatment", getEmptyMatchTreatment().toString())
-                  .withLabel("Treatment of filepatterns that match no files"))
-          .addIfNotNull(
-              DisplayData.item("watchForNewFilesInterval", getWatchForNewFilesInterval())
-                  .withLabel("Interval to watch for new files"));
+          .include("matchConfiguration", getMatchConfiguration());
+    }
+
+    private static class CreateParseSourceFn<T>
+        implements SerializableFunction<String, FileBasedSource<T>> {
+      private final SerializableFunction<GenericRecord, T> parseFn;
+      private final Coder<T> coder;
+
+      public CreateParseSourceFn(SerializableFunction<GenericRecord, T> parseFn, Coder<T> coder) {
+        this.parseFn = parseFn;
+        this.coder = coder;
+      }
+
+      @Override
+      public FileBasedSource<T> apply(String input) {
+        return AvroSource.from(input).withParseFn(parseFn, coder);
+      }
     }
   }
 

http://git-wip-us.apache.org/repos/asf/beam/blob/58e8a01b/sdks/java/core/src/main/java/org/apache/beam/sdk/io/FileIO.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/io/FileIO.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/io/FileIO.java
new file mode 100644
index 0000000..1eb81df
--- /dev/null
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/io/FileIO.java
@@ -0,0 +1,265 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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 org.apache.beam.sdk.io;
+
+import com.google.auto.value.AutoValue;
+import java.io.Serializable;
+import javax.annotation.Nullable;
+import org.apache.beam.sdk.annotations.Experimental;
+import org.apache.beam.sdk.coders.StringUtf8Coder;
+import org.apache.beam.sdk.io.fs.EmptyMatchTreatment;
+import org.apache.beam.sdk.io.fs.MatchResult;
+import org.apache.beam.sdk.options.ValueProvider;
+import org.apache.beam.sdk.transforms.Create;
+import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.PTransform;
+import org.apache.beam.sdk.transforms.ParDo;
+import org.apache.beam.sdk.transforms.Values;
+import org.apache.beam.sdk.transforms.Watch;
+import org.apache.beam.sdk.transforms.Watch.Growth.TerminationCondition;
+import org.apache.beam.sdk.transforms.display.DisplayData;
+import org.apache.beam.sdk.transforms.display.HasDisplayData;
+import org.apache.beam.sdk.values.PBegin;
+import org.apache.beam.sdk.values.PCollection;
+import org.joda.time.Duration;
+import org.joda.time.Instant;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Transforms for working with files. Currently includes matching of filepatterns via {@link #match}
+ * and {@link #matchAll}.
+ */
+public class FileIO {
+  private static final Logger LOG = LoggerFactory.getLogger(FileIO.class);
+
+  /**
+   * Matches a filepattern using {@link FileSystems#match} and produces a collection of matched
+   * resources (both files and directories) as {@link MatchResult.Metadata}.
+   *
+   * <p>By default, matches the filepattern once and produces a bounded {@link PCollection}. To
+   * continuously watch the filepattern for new matches, use {@link MatchAll#continuously(Duration,
+   * TerminationCondition)} - this will produce an unbounded {@link PCollection}.
+   *
+   * <p>By default, a filepattern matching no resources is treated according to {@link
+   * EmptyMatchTreatment#DISALLOW}. To configure this behavior, use {@link
+   * Match#withEmptyMatchTreatment}.
+   */
+  public static Match match() {
+    return new AutoValue_FileIO_Match.Builder()
+        .setConfiguration(MatchConfiguration.create(EmptyMatchTreatment.DISALLOW))
+        .build();
+  }
+
+  /**
+   * Like {@link #match}, but matches each filepattern in a collection of filepatterns.
+   *
+   * <p>Resources are not deduplicated between filepatterns, i.e. if the same resource matches
+   * multiple filepatterns, it will be produced multiple times.
+   *
+   * <p>By default, a filepattern matching no resources is treated according to {@link
+   * EmptyMatchTreatment#ALLOW_IF_WILDCARD}. To configure this behavior, use {@link
+   * MatchAll#withEmptyMatchTreatment}.
+   */
+  public static MatchAll matchAll() {
+    return new AutoValue_FileIO_MatchAll.Builder()
+        .setConfiguration(MatchConfiguration.create(EmptyMatchTreatment.ALLOW_IF_WILDCARD))
+        .build();
+  }
+
+  /**
+   * Describes configuration for matching filepatterns, such as {@link EmptyMatchTreatment}
+   * and continuous watching for matching files.
+   */
+  @AutoValue
+  public abstract static class MatchConfiguration implements HasDisplayData, Serializable {
+    /** Creates a {@link MatchConfiguration} with the given {@link EmptyMatchTreatment}. */
+    public static MatchConfiguration create(EmptyMatchTreatment emptyMatchTreatment) {
+      return new AutoValue_FileIO_MatchConfiguration.Builder()
+          .setEmptyMatchTreatment(emptyMatchTreatment)
+          .build();
+    }
+
+    abstract EmptyMatchTreatment getEmptyMatchTreatment();
+    @Nullable abstract Duration getWatchInterval();
+    @Nullable abstract TerminationCondition<String, ?> getWatchTerminationCondition();
+
+    abstract Builder toBuilder();
+
+    @AutoValue.Builder
+    abstract static class Builder {
+      abstract Builder setEmptyMatchTreatment(EmptyMatchTreatment treatment);
+      abstract Builder setWatchInterval(Duration watchInterval);
+      abstract Builder setWatchTerminationCondition(TerminationCondition<String, ?> condition);
+      abstract MatchConfiguration build();
+    }
+
+    /** Sets the {@link EmptyMatchTreatment}. */
+    public MatchConfiguration withEmptyMatchTreatment(EmptyMatchTreatment treatment) {
+      return toBuilder().setEmptyMatchTreatment(treatment).build();
+    }
+
+    /**
+     * Continuously watches for new files at the given interval until the given termination
+     * condition is reached, where the input to the condition is the filepattern.
+     */
+    public MatchConfiguration continuously(
+        Duration interval, TerminationCondition<String, ?> condition) {
+      return toBuilder().setWatchInterval(interval).setWatchTerminationCondition(condition).build();
+    }
+
+    @Override
+    public void populateDisplayData(DisplayData.Builder builder) {
+      builder
+          .add(
+              DisplayData.item("emptyMatchTreatment", getEmptyMatchTreatment().toString())
+                  .withLabel("Treatment of filepatterns that match no files"))
+          .addIfNotNull(
+              DisplayData.item("watchForNewFilesInterval", getWatchInterval())
+                  .withLabel("Interval to watch for new files"));
+    }
+  }
+
+  /** Implementation of {@link #match}. */
+  @AutoValue
+  public abstract static class Match extends PTransform<PBegin, PCollection<MatchResult.Metadata>> {
+    abstract ValueProvider<String> getFilepattern();
+    abstract MatchConfiguration getConfiguration();
+    abstract Builder toBuilder();
+
+    @AutoValue.Builder
+    abstract static class Builder {
+      abstract Builder setFilepattern(ValueProvider<String> filepattern);
+      abstract Builder setConfiguration(MatchConfiguration configuration);
+      abstract Match build();
+    }
+
+    /** Matches the given filepattern. */
+    public Match filepattern(String filepattern) {
+      return this.filepattern(ValueProvider.StaticValueProvider.of(filepattern));
+    }
+
+    /** Like {@link #filepattern(String)} but using a {@link ValueProvider}. */
+    public Match filepattern(ValueProvider<String> filepattern) {
+      return toBuilder().setFilepattern(filepattern).build();
+    }
+
+    /** Sets the {@link MatchConfiguration}. */
+    public Match withConfiguration(MatchConfiguration configuration) {
+      return toBuilder().setConfiguration(configuration).build();
+    }
+
+    /** See {@link MatchConfiguration#withEmptyMatchTreatment(EmptyMatchTreatment)}. */
+    public Match withEmptyMatchTreatment(EmptyMatchTreatment treatment) {
+      return withConfiguration(getConfiguration().withEmptyMatchTreatment(treatment));
+    }
+
+    /**
+     * See {@link MatchConfiguration#continuously}. The returned {@link PCollection} is unbounded.
+     *
+     * <p>This works only in runners supporting {@link Experimental.Kind#SPLITTABLE_DO_FN}.
+     */
+    @Experimental(Experimental.Kind.SPLITTABLE_DO_FN)
+    public Match continuously(
+        Duration pollInterval, TerminationCondition<String, ?> terminationCondition) {
+      return withConfiguration(getConfiguration().continuously(pollInterval, terminationCondition));
+    }
+
+    @Override
+    public PCollection<MatchResult.Metadata> expand(PBegin input) {
+      return input
+          .apply("Create filepattern", Create.ofProvider(getFilepattern(), StringUtf8Coder.of()))
+          .apply("Via MatchAll", matchAll().withConfiguration(getConfiguration()));
+    }
+  }
+
+  /** Implementation of {@link #matchAll}. */
+  @AutoValue
+  public abstract static class MatchAll
+      extends PTransform<PCollection<String>, PCollection<MatchResult.Metadata>> {
+    abstract MatchConfiguration getConfiguration();
+    abstract Builder toBuilder();
+
+    @AutoValue.Builder
+    abstract static class Builder {
+      abstract Builder setConfiguration(MatchConfiguration configuration);
+      abstract MatchAll build();
+    }
+
+    /** Like {@link Match#withConfiguration}. */
+    public MatchAll withConfiguration(MatchConfiguration configuration) {
+      return toBuilder().setConfiguration(configuration).build();
+    }
+
+    /** Like {@link Match#withEmptyMatchTreatment}. */
+    public MatchAll withEmptyMatchTreatment(EmptyMatchTreatment treatment) {
+      return withConfiguration(getConfiguration().withEmptyMatchTreatment(treatment));
+    }
+
+    /** Like {@link Match#continuously}. */
+    @Experimental(Experimental.Kind.SPLITTABLE_DO_FN)
+    public MatchAll continuously(
+        Duration pollInterval, TerminationCondition<String, ?> terminationCondition) {
+      return withConfiguration(getConfiguration().continuously(pollInterval, terminationCondition));
+    }
+
+    @Override
+    public PCollection<MatchResult.Metadata> expand(PCollection<String> input) {
+      if (getConfiguration().getWatchInterval() == null) {
+        return input.apply(
+            "Match filepatterns",
+            ParDo.of(new MatchFn(getConfiguration().getEmptyMatchTreatment())));
+      } else {
+        return input
+            .apply(
+                "Continuously match filepatterns",
+                Watch.growthOf(new MatchPollFn())
+                    .withPollInterval(getConfiguration().getWatchInterval())
+                    .withTerminationPerInput(getConfiguration().getWatchTerminationCondition()))
+            .apply(Values.<MatchResult.Metadata>create());
+      }
+    }
+
+    private static class MatchFn extends DoFn<String, MatchResult.Metadata> {
+      private final EmptyMatchTreatment emptyMatchTreatment;
+
+      public MatchFn(EmptyMatchTreatment emptyMatchTreatment) {
+        this.emptyMatchTreatment = emptyMatchTreatment;
+      }
+
+      @ProcessElement
+      public void process(ProcessContext c) throws Exception {
+        String filepattern = c.element();
+        MatchResult match = FileSystems.match(filepattern, emptyMatchTreatment);
+        LOG.info("Matched {} files for pattern {}", match.metadata().size(), filepattern);
+        for (MatchResult.Metadata metadata : match.metadata()) {
+          c.output(metadata);
+        }
+      }
+    }
+
+    private static class MatchPollFn implements Watch.Growth.PollFn<String, MatchResult.Metadata> {
+      @Override
+      public Watch.Growth.PollResult<MatchResult.Metadata> apply(String input, Instant timestamp)
+          throws Exception {
+        return Watch.Growth.PollResult.incomplete(
+            Instant.now(), FileSystems.match(input, EmptyMatchTreatment.ALLOW).metadata());
+      }
+    }
+  }
+}

http://git-wip-us.apache.org/repos/asf/beam/blob/58e8a01b/sdks/java/core/src/main/java/org/apache/beam/sdk/io/Match.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/io/Match.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/io/Match.java
deleted file mode 100644
index bb44fac..0000000
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/io/Match.java
+++ /dev/null
@@ -1,156 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  The ASF licenses this file
- * to you 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 org.apache.beam.sdk.io;
-
-import com.google.auto.value.AutoValue;
-import javax.annotation.Nullable;
-import org.apache.beam.sdk.annotations.Experimental;
-import org.apache.beam.sdk.io.fs.EmptyMatchTreatment;
-import org.apache.beam.sdk.io.fs.MatchResult;
-import org.apache.beam.sdk.io.fs.MatchResult.Metadata;
-import org.apache.beam.sdk.transforms.DoFn;
-import org.apache.beam.sdk.transforms.PTransform;
-import org.apache.beam.sdk.transforms.ParDo;
-import org.apache.beam.sdk.transforms.Values;
-import org.apache.beam.sdk.transforms.Watch;
-import org.apache.beam.sdk.transforms.Watch.Growth.PollResult;
-import org.apache.beam.sdk.transforms.Watch.Growth.TerminationCondition;
-import org.apache.beam.sdk.values.PCollection;
-import org.joda.time.Duration;
-import org.joda.time.Instant;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-/**
- * Matches each filepattern in a collection of filepatterns using {@link FileSystems#match}, and
- * produces a collection of matched resources (both files and directories) as {@link Metadata}.
- * Resources are not deduplicated between filepatterns, i.e. if the same resource matches multiple
- * filepatterns, it will be produced multiple times.
- *
- * <p>By default, this transform matches each filepattern once and produces a bounded {@link
- * PCollection}. To continuously watch each filepattern for new matches, use {@link
- * Filepatterns#continuously(Duration, TerminationCondition)} - this will produce an unbounded
- * {@link PCollection}.
- *
- * <p>By default, filepatterns matching no resources are treated according to {@link
- * EmptyMatchTreatment#ALLOW_IF_WILDCARD}. To configure this behavior, use {@link
- * Filepatterns#withEmptyMatchTreatment}.
- */
-public class Match {
-  private static final Logger LOG = LoggerFactory.getLogger(Match.class);
-
-  /** See {@link Match}. */
-  public static Filepatterns filepatterns() {
-    return new AutoValue_Match_Filepatterns.Builder()
-        .setEmptyMatchTreatment(EmptyMatchTreatment.ALLOW_IF_WILDCARD)
-        .build();
-  }
-
-  /** Implementation of {@link #filepatterns}. */
-  @AutoValue
-  public abstract static class Filepatterns
-      extends PTransform<PCollection<String>, PCollection<Metadata>> {
-    abstract EmptyMatchTreatment getEmptyMatchTreatment();
-
-    @Nullable
-    abstract Duration getWatchInterval();
-
-    @Nullable
-    abstract TerminationCondition<String, ?> getWatchTerminationCondition();
-
-    abstract Builder toBuilder();
-
-    @AutoValue.Builder
-    abstract static class Builder {
-      abstract Builder setEmptyMatchTreatment(EmptyMatchTreatment treatment);
-
-      abstract Builder setWatchInterval(Duration watchInterval);
-
-      abstract Builder setWatchTerminationCondition(TerminationCondition<String, ?> condition);
-
-      abstract Filepatterns build();
-    }
-
-    /**
-     * Sets whether or not filepatterns matching no files are allowed. When using {@link
-     * #continuously}, they are always allowed, and this parameter is ignored.
-     */
-    public Filepatterns withEmptyMatchTreatment(EmptyMatchTreatment treatment) {
-      return toBuilder().setEmptyMatchTreatment(treatment).build();
-    }
-
-    /**
-     * Continuously watches for new resources matching the filepattern, repeatedly matching it at
-     * the given interval, until the given termination condition is reached. The returned {@link
-     * PCollection} is unbounded.
-     *
-     * <p>This works only in runners supporting {@link Experimental.Kind#SPLITTABLE_DO_FN}.
-     *
-     * @see TerminationCondition
-     */
-    @Experimental(Experimental.Kind.SPLITTABLE_DO_FN)
-    public Filepatterns continuously(
-        Duration pollInterval, TerminationCondition<String, ?> terminationCondition) {
-      return toBuilder()
-          .setWatchInterval(pollInterval)
-          .setWatchTerminationCondition(terminationCondition)
-          .build();
-    }
-
-    @Override
-    public PCollection<Metadata> expand(PCollection<String> input) {
-      if (getWatchInterval() == null) {
-        return input.apply("Match filepatterns", ParDo.of(new MatchFn(getEmptyMatchTreatment())));
-      } else {
-        return input
-            .apply(
-                "Continuously match filepatterns",
-                Watch.growthOf(new MatchPollFn())
-                    .withPollInterval(getWatchInterval())
-                    .withTerminationPerInput(getWatchTerminationCondition()))
-            .apply(Values.<Metadata>create());
-      }
-    }
-
-    private static class MatchFn extends DoFn<String, Metadata> {
-      private final EmptyMatchTreatment emptyMatchTreatment;
-
-      public MatchFn(EmptyMatchTreatment emptyMatchTreatment) {
-        this.emptyMatchTreatment = emptyMatchTreatment;
-      }
-
-      @ProcessElement
-      public void process(ProcessContext c) throws Exception {
-        String filepattern = c.element();
-        MatchResult match = FileSystems.match(filepattern, emptyMatchTreatment);
-        LOG.info("Matched {} files for pattern {}", match.metadata().size(), filepattern);
-        for (Metadata metadata : match.metadata()) {
-          c.output(metadata);
-        }
-      }
-    }
-
-    private static class MatchPollFn implements Watch.Growth.PollFn<String, Metadata> {
-      @Override
-      public PollResult<Metadata> apply(String input, Instant timestamp) throws Exception {
-        return PollResult.incomplete(
-            Instant.now(), FileSystems.match(input, EmptyMatchTreatment.ALLOW).metadata());
-      }
-    }
-  }
-}

http://git-wip-us.apache.org/repos/asf/beam/blob/58e8a01b/sdks/java/core/src/main/java/org/apache/beam/sdk/io/ReadAllViaFileBasedSource.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/io/ReadAllViaFileBasedSource.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/io/ReadAllViaFileBasedSource.java
index 990f508..0cd7105 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/io/ReadAllViaFileBasedSource.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/io/ReadAllViaFileBasedSource.java
@@ -39,7 +39,7 @@ import org.apache.beam.sdk.values.PCollection;
  * input {@link PCollection} must not contain {@link ResourceId#isDirectory directories}.
  *
  * <p>To obtain the collection of {@link Metadata} from a filepattern, use {@link
- * Match#filepatterns()}.
+ * FileIO#match} or {@link FileIO#matchAll}.
  */
 class ReadAllViaFileBasedSource<T> extends PTransform<PCollection<Metadata>, PCollection<T>> {
   private final SerializableFunction<String, Boolean> isSplittable;

http://git-wip-us.apache.org/repos/asf/beam/blob/58e8a01b/sdks/java/core/src/main/java/org/apache/beam/sdk/io/TextIO.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/io/TextIO.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/io/TextIO.java
index eba5ab5..57bfaa9 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/io/TextIO.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/io/TextIO.java
@@ -20,7 +20,6 @@ package org.apache.beam.sdk.io;
 import static com.google.common.base.Preconditions.checkArgument;
 import static com.google.common.base.Preconditions.checkNotNull;
 import static com.google.common.base.Preconditions.checkState;
-import static org.apache.beam.sdk.transforms.Watch.Growth.ignoreInput;
 
 import com.google.auto.value.AutoValue;
 import com.google.common.annotations.VisibleForTesting;
@@ -39,6 +38,7 @@ import org.apache.beam.sdk.io.DefaultFilenamePolicy.Params;
 import org.apache.beam.sdk.io.FileBasedSink.DynamicDestinations;
 import org.apache.beam.sdk.io.FileBasedSink.FilenamePolicy;
 import org.apache.beam.sdk.io.FileBasedSink.WritableByteChannelFactory;
+import org.apache.beam.sdk.io.FileIO.MatchConfiguration;
 import org.apache.beam.sdk.io.fs.EmptyMatchTreatment;
 import org.apache.beam.sdk.io.fs.ResourceId;
 import org.apache.beam.sdk.options.ValueProvider;
@@ -193,7 +193,7 @@ public class TextIO {
     return new AutoValue_TextIO_Read.Builder()
         .setCompression(Compression.AUTO)
         .setHintMatchesManyFiles(false)
-        .setEmptyMatchTreatment(EmptyMatchTreatment.DISALLOW)
+        .setMatchConfiguration(MatchConfiguration.create(EmptyMatchTreatment.DISALLOW))
         .build();
   }
 
@@ -214,7 +214,7 @@ public class TextIO {
         // but is not so large as to exhaust a typical runner's maximum amount of output per
         // ProcessElement call.
         .setDesiredBundleSizeBytes(64 * 1024 * 1024L)
-        .setEmptyMatchTreatment(EmptyMatchTreatment.ALLOW_IF_WILDCARD)
+        .setMatchConfiguration(MatchConfiguration.create(EmptyMatchTreatment.ALLOW_IF_WILDCARD))
         .build();
   }
 
@@ -259,32 +259,19 @@ public class TextIO {
   /** Implementation of {@link #read}. */
   @AutoValue
   public abstract static class Read extends PTransform<PBegin, PCollection<String>> {
-    @Nullable
-    abstract ValueProvider<String> getFilepattern();
-    abstract Compression getCompression();
-
-    @Nullable
-    abstract Duration getWatchForNewFilesInterval();
-
-    @Nullable
-    abstract TerminationCondition<?, ?> getWatchForNewFilesTerminationCondition();
-
+    @Nullable abstract ValueProvider<String> getFilepattern();
+    abstract MatchConfiguration getMatchConfiguration();
     abstract boolean getHintMatchesManyFiles();
-    abstract EmptyMatchTreatment getEmptyMatchTreatment();
-    @Nullable
-    abstract byte[] getDelimiter();
-
+    abstract Compression getCompression();
+    @Nullable abstract byte[] getDelimiter();
     abstract Builder toBuilder();
 
     @AutoValue.Builder
     abstract static class Builder {
       abstract Builder setFilepattern(ValueProvider<String> filepattern);
-      abstract Builder setCompression(Compression compression);
-      abstract Builder setWatchForNewFilesInterval(Duration watchForNewFilesInterval);
-      abstract Builder setWatchForNewFilesTerminationCondition(
-              TerminationCondition<?, ?> condition);
+      abstract Builder setMatchConfiguration(MatchConfiguration matchConfiguration);
       abstract Builder setHintMatchesManyFiles(boolean hintManyFiles);
-      abstract Builder setEmptyMatchTreatment(EmptyMatchTreatment treatment);
+      abstract Builder setCompression(Compression compression);
       abstract Builder setDelimiter(byte[] delimiter);
 
       abstract Read build();
@@ -314,6 +301,11 @@ public class TextIO {
       return toBuilder().setFilepattern(filepattern).build();
     }
 
+    /** Sets the {@link MatchConfiguration}. */
+    public Read withMatchConfiguration(MatchConfiguration matchConfiguration) {
+      return toBuilder().setMatchConfiguration(matchConfiguration).build();
+    }
+
     /** @deprecated Use {@link #withCompression}. */
     @Deprecated
     public Read withCompressionType(TextIO.CompressionType compressionType) {
@@ -330,21 +322,15 @@ public class TextIO {
     }
 
     /**
-     * Continuously watches for new files matching the filepattern, polling it at the given
-     * interval, until the given termination condition is reached. The returned {@link PCollection}
-     * is unbounded.
+     * See {@link MatchConfiguration#continuously}.
      *
      * <p>This works only in runners supporting {@link Kind#SPLITTABLE_DO_FN}.
-     *
-     * @see TerminationCondition
      */
     @Experimental(Kind.SPLITTABLE_DO_FN)
     public Read watchForNewFiles(
-        Duration pollInterval, TerminationCondition<?, ?> terminationCondition) {
-      return toBuilder()
-          .setWatchForNewFilesInterval(pollInterval)
-          .setWatchForNewFilesTerminationCondition(terminationCondition)
-          .build();
+        Duration pollInterval, TerminationCondition<String, ?> terminationCondition) {
+      return withMatchConfiguration(
+          getMatchConfiguration().continuously(pollInterval, terminationCondition));
     }
 
     /**
@@ -360,12 +346,9 @@ public class TextIO {
       return toBuilder().setHintMatchesManyFiles(true).build();
     }
 
-    /**
-     * Configures whether or not a filepattern matching no files is allowed. When using {@link
-     * #watchForNewFiles}, it is always allowed and this parameter is ignored.
-     */
+    /** See {@link MatchConfiguration#withEmptyMatchTreatment}. */
     public Read withEmptyMatchTreatment(EmptyMatchTreatment treatment) {
-      return toBuilder().setEmptyMatchTreatment(treatment).build();
+      return withMatchConfiguration(getMatchConfiguration().withEmptyMatchTreatment(treatment));
     }
 
     /**
@@ -390,29 +373,27 @@ public class TextIO {
     @Override
     public PCollection<String> expand(PBegin input) {
       checkNotNull(getFilepattern(), "need to set the filepattern of a TextIO.Read transform");
-      if (getWatchForNewFilesInterval() == null && !getHintMatchesManyFiles()) {
+      if (getMatchConfiguration().getWatchInterval() == null && !getHintMatchesManyFiles()) {
         return input.apply("Read", org.apache.beam.sdk.io.Read.from(getSource()));
       }
       // All other cases go through ReadAll.
-      ReadAll readAll =
-          readAll()
-              .withCompression(getCompression())
-              .withEmptyMatchTreatment(getEmptyMatchTreatment())
-              .withDelimiter(getDelimiter());
-      if (getWatchForNewFilesInterval() != null) {
-        TerminationCondition<String, ?> readAllCondition =
-            ignoreInput(getWatchForNewFilesTerminationCondition());
-        readAll = readAll.watchForNewFiles(getWatchForNewFilesInterval(), readAllCondition);
-      }
       return input
           .apply("Create filepattern", Create.ofProvider(getFilepattern(), StringUtf8Coder.of()))
-          .apply("Via ReadAll", readAll);
+          .apply(
+              "Via ReadAll",
+              readAll()
+                  .withCompression(getCompression())
+                  .withMatchConfiguration(getMatchConfiguration())
+                  .withDelimiter(getDelimiter()));
     }
 
     // Helper to create a source specific to the requested compression type.
     protected FileBasedSource<String> getSource() {
-      return CompressedSource
-          .from(new TextSource(getFilepattern(), getEmptyMatchTreatment(), getDelimiter()))
+      return CompressedSource.from(
+              new TextSource(
+                  getFilepattern(),
+                  getMatchConfiguration().getEmptyMatchTreatment(),
+                  getDelimiter()))
           .withCompression(getCompression());
     }
 
@@ -425,16 +406,10 @@ public class TextIO {
                   .withLabel("Compression Type"))
           .addIfNotNull(
               DisplayData.item("filePattern", getFilepattern()).withLabel("File Pattern"))
-          .add(
-              DisplayData.item("emptyMatchTreatment", getEmptyMatchTreatment().toString())
-                  .withLabel("Treatment of filepatterns that match no files"))
-          .addIfNotNull(
-              DisplayData.item("watchForNewFilesInterval", getWatchForNewFilesInterval())
-                  .withLabel("Interval to watch for new files"))
+          .include("matchConfiguration", getMatchConfiguration())
           .addIfNotNull(
               DisplayData.item("delimiter", Arrays.toString(getDelimiter()))
               .withLabel("Custom delimiter to split records"));
-
     }
   }
 
@@ -444,15 +419,8 @@ public class TextIO {
   @AutoValue
   public abstract static class ReadAll
       extends PTransform<PCollection<String>, PCollection<String>> {
+    abstract MatchConfiguration getMatchConfiguration();
     abstract Compression getCompression();
-
-    @Nullable
-    abstract Duration getWatchForNewFilesInterval();
-
-    @Nullable
-    abstract TerminationCondition<String, ?> getWatchForNewFilesTerminationCondition();
-
-    abstract EmptyMatchTreatment getEmptyMatchTreatment();
     abstract long getDesiredBundleSizeBytes();
     @Nullable
     abstract byte[] getDelimiter();
@@ -461,16 +429,18 @@ public class TextIO {
 
     @AutoValue.Builder
     abstract static class Builder {
+      abstract Builder setMatchConfiguration(MatchConfiguration matchConfiguration);
       abstract Builder setCompression(Compression compression);
-      abstract Builder setWatchForNewFilesInterval(Duration watchForNewFilesInterval);
-      abstract Builder setWatchForNewFilesTerminationCondition(
-          TerminationCondition<String, ?> condition);
-      abstract Builder setEmptyMatchTreatment(EmptyMatchTreatment treatment);
       abstract Builder setDesiredBundleSizeBytes(long desiredBundleSizeBytes);
       abstract Builder setDelimiter(byte[] delimiter);
       abstract ReadAll build();
     }
 
+    /** Sets the {@link MatchConfiguration}. */
+    public ReadAll withMatchConfiguration(MatchConfiguration configuration) {
+      return toBuilder().setMatchConfiguration(configuration).build();
+    }
+
     /** @deprecated Use {@link #withCompression}. */
     @Deprecated
     public ReadAll withCompressionType(TextIO.CompressionType compressionType) {
@@ -488,17 +458,15 @@ public class TextIO {
 
     /** Same as {@link Read#withEmptyMatchTreatment}. */
     public ReadAll withEmptyMatchTreatment(EmptyMatchTreatment treatment) {
-      return toBuilder().setEmptyMatchTreatment(treatment).build();
+      return withMatchConfiguration(getMatchConfiguration().withEmptyMatchTreatment(treatment));
     }
 
     /** Same as {@link Read#watchForNewFiles(Duration, TerminationCondition)}. */
     @Experimental(Kind.SPLITTABLE_DO_FN)
     public ReadAll watchForNewFiles(
         Duration pollInterval, TerminationCondition<String, ?> terminationCondition) {
-      return toBuilder()
-          .setWatchForNewFilesInterval(pollInterval)
-          .setWatchForNewFilesTerminationCondition(terminationCondition)
-          .build();
+      return withMatchConfiguration(
+          getMatchConfiguration().continuously(pollInterval, terminationCondition));
     }
 
     @VisibleForTesting
@@ -512,22 +480,15 @@ public class TextIO {
 
     @Override
     public PCollection<String> expand(PCollection<String> input) {
-      Match.Filepatterns matchFilepatterns =
-          Match.filepatterns().withEmptyMatchTreatment(getEmptyMatchTreatment());
-      if (getWatchForNewFilesInterval() != null) {
-        matchFilepatterns =
-            matchFilepatterns.continuously(
-                getWatchForNewFilesInterval(), getWatchForNewFilesTerminationCondition());
-      }
       return input
-          .apply(matchFilepatterns)
+          .apply(FileIO.matchAll().withConfiguration(getMatchConfiguration()))
           .apply(
               "Read all via FileBasedSource",
               new ReadAllViaFileBasedSource<>(
                   new IsSplittableFn(getCompression()),
                   getDesiredBundleSizeBytes(),
-                  new CreateTextSourceFn(getCompression(), getEmptyMatchTreatment(),
-                      getDelimiter()))).setCoder(StringUtf8Coder.of());
+                  new CreateTextSourceFn(getCompression(), getDelimiter())))
+          .setCoder(StringUtf8Coder.of());
     }
 
     @Override
@@ -536,30 +497,30 @@ public class TextIO {
 
       builder
           .add(
-          DisplayData.item("compressionType", getCompression().toString())
-              .withLabel("Compression Type"))
+              DisplayData.item("compressionType", getCompression().toString())
+                  .withLabel("Compression Type"))
           .addIfNotNull(
-          DisplayData.item("delimiter", Arrays.toString(getDelimiter()))
-              .withLabel("Custom delimiter to split records"));
+              DisplayData.item("delimiter", Arrays.toString(getDelimiter()))
+                  .withLabel("Custom delimiter to split records"))
+          .include("matchConfiguration", getMatchConfiguration());
     }
 
     private static class CreateTextSourceFn
         implements SerializableFunction<String, FileBasedSource<String>> {
       private final Compression compression;
-      private final EmptyMatchTreatment emptyMatchTreatment;
       private byte[] delimiter;
 
       private CreateTextSourceFn(
-          Compression compression, EmptyMatchTreatment emptyMatchTreatment, byte[] delimiter) {
+          Compression compression, byte[] delimiter) {
         this.compression = compression;
-        this.emptyMatchTreatment = emptyMatchTreatment;
         this.delimiter = delimiter;
       }
 
       @Override
       public FileBasedSource<String> apply(String input) {
         return CompressedSource.from(
-                new TextSource(StaticValueProvider.of(input), emptyMatchTreatment, delimiter))
+                new TextSource(
+                    StaticValueProvider.of(input), EmptyMatchTreatment.DISALLOW, delimiter))
             .withCompression(compression);
       }
     }

http://git-wip-us.apache.org/repos/asf/beam/blob/58e8a01b/sdks/java/core/src/test/java/org/apache/beam/sdk/io/AvroIOTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/io/AvroIOTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/io/AvroIOTest.java
index 8c68f57..695e196 100644
--- a/sdks/java/core/src/test/java/org/apache/beam/sdk/io/AvroIOTest.java
+++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/io/AvroIOTest.java
@@ -335,7 +335,7 @@ public class AvroIOTest {
                     .from(tmpFolder.getRoot().getAbsolutePath() + "/first*")
                     .watchForNewFiles(
                         Duration.millis(100),
-                        Watch.Growth.<Void>afterTimeSinceNewOutput(Duration.standardSeconds(3)))))
+                        Watch.Growth.<String>afterTimeSinceNewOutput(Duration.standardSeconds(3)))))
         .containsInAnyOrder(firstValues);
     PAssert.that(
             readPipeline.apply(
@@ -344,7 +344,7 @@ public class AvroIOTest {
                     .from(tmpFolder.getRoot().getAbsolutePath() + "/first*")
                     .watchForNewFiles(
                         Duration.millis(100),
-                        Watch.Growth.<Void>afterTimeSinceNewOutput(Duration.standardSeconds(3)))))
+                        Watch.Growth.<String>afterTimeSinceNewOutput(Duration.standardSeconds(3)))))
         .containsInAnyOrder(firstValues);
 
     PCollection<String> paths =

http://git-wip-us.apache.org/repos/asf/beam/blob/58e8a01b/sdks/java/core/src/test/java/org/apache/beam/sdk/io/TextIOReadTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/io/TextIOReadTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/io/TextIOReadTest.java
index e55a820..e708b46 100644
--- a/sdks/java/core/src/test/java/org/apache/beam/sdk/io/TextIOReadTest.java
+++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/io/TextIOReadTest.java
@@ -26,7 +26,6 @@ import static org.apache.beam.sdk.io.Compression.DEFLATE;
 import static org.apache.beam.sdk.io.Compression.GZIP;
 import static org.apache.beam.sdk.io.Compression.UNCOMPRESSED;
 import static org.apache.beam.sdk.io.Compression.ZIP;
-import static org.apache.beam.sdk.transforms.Watch.Growth.afterTimeSinceNewOutput;
 import static org.apache.beam.sdk.transforms.display.DisplayDataMatchers.hasDisplayItem;
 import static org.apache.beam.sdk.transforms.display.DisplayDataMatchers.hasValue;
 import static org.hamcrest.Matchers.containsInAnyOrder;
@@ -78,6 +77,7 @@ import org.apache.beam.sdk.testing.TestPipeline;
 import org.apache.beam.sdk.testing.UsesSplittableParDo;
 import org.apache.beam.sdk.testing.ValidatesRunner;
 import org.apache.beam.sdk.transforms.Create;
+import org.apache.beam.sdk.transforms.Watch;
 import org.apache.beam.sdk.transforms.display.DisplayData;
 import org.apache.beam.sdk.transforms.display.DisplayDataEvaluator;
 import org.apache.beam.sdk.util.CoderUtils;
@@ -951,7 +951,8 @@ public class TextIOReadTest {
                 // Make sure that compression type propagates into readAll()
                 .withCompression(ZIP)
                 .watchForNewFiles(
-                    Duration.millis(100), afterTimeSinceNewOutput(Duration.standardSeconds(3))));
+                    Duration.millis(100),
+                    Watch.Growth.<String>afterTimeSinceNewOutput(Duration.standardSeconds(3))));
 
     Thread writer =
         new Thread() {


Mime
View raw message