beam-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From dhalp...@apache.org
Subject [1/2] incubator-beam git commit: [BEAM-1201] Remove BoundedSource.producesSortedKeys
Date Tue, 27 Dec 2016 17:29:21 GMT
Repository: incubator-beam
Updated Branches:
  refs/heads/master c2b27ff30 -> ae46f9a87


[BEAM-1201] Remove BoundedSource.producesSortedKeys

Unused and unclear; for more information see the linked JIRA.


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

Branch: refs/heads/master
Commit: 321058ff1fb5c0900de5a14ff2013f3285531488
Parents: c2b27ff
Author: Dan Halperin <dhalperi@google.com>
Authored: Wed Dec 21 15:32:38 2016 -0800
Committer: Dan Halperin <dhalperi@google.com>
Committed: Tue Dec 27 09:25:55 2016 -0800

----------------------------------------------------------------------
 .../runners/core/UnboundedReadFromBoundedSourceTest.java  |  5 -----
 .../runners/direct/BoundedReadEvaluatorFactoryTest.java   |  5 -----
 .../org/apache/beam/runners/direct/DirectRunnerTest.java  |  5 -----
 .../beam/runners/dataflow/internal/CustomSources.java     |  5 -----
 .../beam/runners/dataflow/internal/ReadTranslator.java    |  3 ---
 .../DataflowUnboundedReadFromBoundedSourceTest.java       |  4 ----
 .../apache/beam/runners/spark/io/MicrobatchSource.java    |  5 -----
 .../src/main/java/org/apache/beam/sdk/io/AvroSource.java  |  5 -----
 .../beam/sdk/io/BoundedReadFromUnboundedSource.java       |  5 -----
 .../main/java/org/apache/beam/sdk/io/BoundedSource.java   |  8 --------
 .../java/org/apache/beam/sdk/io/CompressedSource.java     |  8 --------
 .../main/java/org/apache/beam/sdk/io/CountingSource.java  |  5 -----
 .../core/src/main/java/org/apache/beam/sdk/io/TextIO.java |  5 -----
 .../src/main/java/org/apache/beam/sdk/io/XmlSource.java   |  5 -----
 .../java/org/apache/beam/sdk/testing/SourceTestUtils.java |  5 -----
 .../main/java/org/apache/beam/sdk/transforms/Create.java  |  5 -----
 .../java/org/apache/beam/sdk/io/CompressedSourceTest.java |  5 -----
 .../java/org/apache/beam/sdk/io/FileBasedSourceTest.java  |  5 -----
 .../org/apache/beam/sdk/io/OffsetBasedSourceTest.java     |  5 -----
 .../src/test/java/org/apache/beam/sdk/io/ReadTest.java    |  5 -----
 .../java/org/apache/beam/sdk/transforms/CreateTest.java   |  8 --------
 .../org/apache/beam/sdk/io/gcp/bigquery/BigQueryIO.java   | 10 ----------
 .../org/apache/beam/sdk/io/gcp/bigtable/BigtableIO.java   |  8 --------
 .../java/org/apache/beam/sdk/io/hdfs/HDFSFileSource.java  |  5 -----
 .../org/apache/beam/sdk/io/mongodb/MongoDbGridFSIO.java   |  5 -----
 .../java/org/apache/beam/sdk/io/mongodb/MongoDbIO.java    |  5 -----
 26 files changed, 144 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/321058ff/runners/core-java/src/test/java/org/apache/beam/runners/core/UnboundedReadFromBoundedSourceTest.java
----------------------------------------------------------------------
diff --git a/runners/core-java/src/test/java/org/apache/beam/runners/core/UnboundedReadFromBoundedSourceTest.java
b/runners/core-java/src/test/java/org/apache/beam/runners/core/UnboundedReadFromBoundedSourceTest.java
index 0f09cd1..e03793b 100644
--- a/runners/core-java/src/test/java/org/apache/beam/runners/core/UnboundedReadFromBoundedSourceTest.java
+++ b/runners/core-java/src/test/java/org/apache/beam/runners/core/UnboundedReadFromBoundedSourceTest.java
@@ -311,11 +311,6 @@ public class UnboundedReadFromBoundedSourceTest {
     }
 
     @Override
-    public boolean producesSortedKeys(PipelineOptions options) throws Exception {
-      return false;
-    }
-
-    @Override
     public Coder<Byte> getDefaultOutputCoder() {
       return SerializableCoder.of(Byte.class);
     }

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/321058ff/runners/direct-java/src/test/java/org/apache/beam/runners/direct/BoundedReadEvaluatorFactoryTest.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/BoundedReadEvaluatorFactoryTest.java
b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/BoundedReadEvaluatorFactoryTest.java
index 97eae27..8361bdc 100644
--- a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/BoundedReadEvaluatorFactoryTest.java
+++ b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/BoundedReadEvaluatorFactoryTest.java
@@ -376,11 +376,6 @@ public class BoundedReadEvaluatorFactoryTest {
     }
 
     @Override
-    public boolean producesSortedKeys(PipelineOptions options) throws Exception {
-      return false;
-    }
-
-    @Override
     public BoundedSource.BoundedReader<T> createReader(PipelineOptions options) throws
IOException {
       subrangesCompleted = new CountDownLatch(2);
       return new TestReader<>(this, firstSplitIndex, subrangesCompleted);

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/321058ff/runners/direct-java/src/test/java/org/apache/beam/runners/direct/DirectRunnerTest.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/DirectRunnerTest.java
b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/DirectRunnerTest.java
index eafb788..ac1689d 100644
--- a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/DirectRunnerTest.java
+++ b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/DirectRunnerTest.java
@@ -484,11 +484,6 @@ public class DirectRunnerTest implements Serializable {
     }
 
     @Override
-    public boolean producesSortedKeys(PipelineOptions options) throws Exception {
-      return underlying.producesSortedKeys(options);
-    }
-
-    @Override
     public BoundedReader<T> createReader(PipelineOptions options) throws IOException
{
       throw new IllegalStateException(
           "The MustSplitSource cannot create a reader without being split first");

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/321058ff/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/internal/CustomSources.java
----------------------------------------------------------------------
diff --git a/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/internal/CustomSources.java
b/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/internal/CustomSources.java
index aeb8103..4d5da32 100755
--- a/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/internal/CustomSources.java
+++ b/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/internal/CustomSources.java
@@ -83,11 +83,6 @@ public class CustomSources {
     SourceMetadata metadata = new SourceMetadata();
     if (source instanceof BoundedSource) {
       BoundedSource<?> boundedSource = (BoundedSource<?>) source;
-      try {
-        metadata.setProducesSortedKeys(boundedSource.producesSortedKeys(options));
-      } catch (Exception e) {
-        LOG.warn("Failed to check if the source produces sorted keys: " + source, e);
-      }
 
       // Size estimation is best effort so we continue even if it fails here.
       try {

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/321058ff/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/internal/ReadTranslator.java
----------------------------------------------------------------------
diff --git a/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/internal/ReadTranslator.java
b/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/internal/ReadTranslator.java
index b3af165..84950f7 100755
--- a/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/internal/ReadTranslator.java
+++ b/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/internal/ReadTranslator.java
@@ -93,9 +93,6 @@ public class ReadTranslator implements TransformTranslator<Read.Bounded<?>>
{
 
   private static Map<String, Object> cloudSourceMetadataToDictionary(SourceMetadata
metadata) {
     Map<String, Object> res = new HashMap<>();
-    if (metadata.getProducesSortedKeys() != null) {
-      addBoolean(res, PropertyNames.SOURCE_PRODUCES_SORTED_KEYS, metadata.getProducesSortedKeys());
-    }
     if (metadata.getEstimatedSizeBytes() != null) {
       addLong(res, PropertyNames.SOURCE_ESTIMATED_SIZE_BYTES, metadata.getEstimatedSizeBytes());
     }

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/321058ff/runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/internal/DataflowUnboundedReadFromBoundedSourceTest.java
----------------------------------------------------------------------
diff --git a/runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/internal/DataflowUnboundedReadFromBoundedSourceTest.java
b/runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/internal/DataflowUnboundedReadFromBoundedSourceTest.java
index 3a88935..d38428b 100644
--- a/runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/internal/DataflowUnboundedReadFromBoundedSourceTest.java
+++ b/runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/internal/DataflowUnboundedReadFromBoundedSourceTest.java
@@ -63,10 +63,6 @@ public class DataflowUnboundedReadFromBoundedSourceTest {
       return 0;
     }
     @Override
-    public boolean producesSortedKeys(PipelineOptions options) throws Exception {
-      return false;
-    }
-    @Override
     public BoundedReader<String> createReader(
         PipelineOptions options) throws IOException {
       return null;

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/321058ff/runners/spark/src/main/java/org/apache/beam/runners/spark/io/MicrobatchSource.java
----------------------------------------------------------------------
diff --git a/runners/spark/src/main/java/org/apache/beam/runners/spark/io/MicrobatchSource.java
b/runners/spark/src/main/java/org/apache/beam/runners/spark/io/MicrobatchSource.java
index 5656375..f42cca3 100644
--- a/runners/spark/src/main/java/org/apache/beam/runners/spark/io/MicrobatchSource.java
+++ b/runners/spark/src/main/java/org/apache/beam/runners/spark/io/MicrobatchSource.java
@@ -112,11 +112,6 @@ public class MicrobatchSource<T, CheckpointMarkT extends UnboundedSource.Checkpo
   }
 
   @Override
-  public boolean producesSortedKeys(PipelineOptions options) throws Exception {
-    return false;
-  }
-
-  @Override
   public BoundedReader<T> createReader(PipelineOptions options) throws IOException
{
     return createReader(options, null);
   }

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/321058ff/sdks/java/core/src/main/java/org/apache/beam/sdk/io/AvroSource.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/io/AvroSource.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/io/AvroSource.java
index aaf72ac..2ba793a 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/io/AvroSource.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/io/AvroSource.java
@@ -297,11 +297,6 @@ public class AvroSource<T> extends BlockBasedSource<T> {
   }
 
   @Override
-  public boolean producesSortedKeys(PipelineOptions options) throws Exception {
-    return false;
-  }
-
-  @Override
   public AvroCoder<T> getDefaultOutputCoder() {
     if (coder == null) {
       coder = AvroCoder.of(type, internOrParseSchemaString(readSchemaString));

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/321058ff/sdks/java/core/src/main/java/org/apache/beam/sdk/io/BoundedReadFromUnboundedSource.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/io/BoundedReadFromUnboundedSource.java
b/sdks/java/core/src/main/java/org/apache/beam/sdk/io/BoundedReadFromUnboundedSource.java
index 84e3044..f568d86 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/io/BoundedReadFromUnboundedSource.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/io/BoundedReadFromUnboundedSource.java
@@ -194,11 +194,6 @@ public class BoundedReadFromUnboundedSource<T> extends PTransform<PBegin,
PColle
     }
 
     @Override
-    public boolean producesSortedKeys(PipelineOptions options) {
-      return false;
-    }
-
-    @Override
     public Coder<ValueWithRecordId<T>> getDefaultOutputCoder() {
       return ValueWithRecordId.ValueWithRecordIdCoder.of(source.getDefaultOutputCoder());
     }

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/321058ff/sdks/java/core/src/main/java/org/apache/beam/sdk/io/BoundedSource.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/io/BoundedSource.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/io/BoundedSource.java
index b3dd138..8e5145c 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/io/BoundedSource.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/io/BoundedSource.java
@@ -36,8 +36,6 @@ import org.joda.time.Instant;
  * <ul>
  * <li>Splitting into bundles of given size: {@link #splitIntoBundles};
  * <li>Size estimation: {@link #getEstimatedSizeBytes};
- * <li>Telling whether or not this source produces key/value pairs in sorted order:
- * {@link #producesSortedKeys};
  * <li>The accompanying {@link BoundedReader reader} has additional functionality to
enable runners
  * to dynamically adapt based on runtime conditions.
  *     <ul>
@@ -67,12 +65,6 @@ public abstract class BoundedSource<T> extends Source<T> {
   public abstract long getEstimatedSizeBytes(PipelineOptions options) throws Exception;
 
   /**
-   * Whether this source is known to produce key/value pairs sorted by lexicographic order
on
-   * the bytes of the encoded key.
-   */
-  public abstract boolean producesSortedKeys(PipelineOptions options) throws Exception;
-
-  /**
    * Returns a new {@link BoundedReader} that reads from this source.
    */
   public abstract BoundedReader<T> createReader(PipelineOptions options) throws IOException;

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/321058ff/sdks/java/core/src/main/java/org/apache/beam/sdk/io/CompressedSource.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/io/CompressedSource.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/io/CompressedSource.java
index af0c8130..e3bd32e 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/io/CompressedSource.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/io/CompressedSource.java
@@ -376,14 +376,6 @@ public class CompressedSource<T> extends FileBasedSource<T>
{
         this, sourceDelegate.createSingleFileReader(options));
   }
 
-  /**
-   * Returns whether the delegate source produces sorted keys.
-   */
-  @Override
-  public final boolean producesSortedKeys(PipelineOptions options) throws Exception {
-    return sourceDelegate.producesSortedKeys(options);
-  }
-
   @Override
   public void populateDisplayData(DisplayData.Builder builder) {
     // We explicitly do not register base-class data, instead we use the delegate inner source.

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/321058ff/sdks/java/core/src/main/java/org/apache/beam/sdk/io/CountingSource.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/io/CountingSource.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/io/CountingSource.java
index 9752dba..4b9ec66 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/io/CountingSource.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/io/CountingSource.java
@@ -188,11 +188,6 @@ public class CountingSource {
     }
 
     @Override
-    public boolean producesSortedKeys(PipelineOptions options) throws Exception {
-      return true;
-    }
-
-    @Override
     public org.apache.beam.sdk.io.BoundedSource.BoundedReader<Long> createReader(
         PipelineOptions options) throws IOException {
       return new BoundedCountingReader(this);

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/321058ff/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 54e73d5..16b871e 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
@@ -933,11 +933,6 @@ public class TextIO {
     }
 
     @Override
-    public boolean producesSortedKeys(PipelineOptions options) throws Exception {
-      return false;
-    }
-
-    @Override
     public Coder<T> getDefaultOutputCoder() {
       return coder;
     }

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/321058ff/sdks/java/core/src/main/java/org/apache/beam/sdk/io/XmlSource.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/io/XmlSource.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/io/XmlSource.java
index 1458f74..6bf2015 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/io/XmlSource.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/io/XmlSource.java
@@ -201,11 +201,6 @@ public class XmlSource<T> extends FileBasedSource<T> {
   }
 
   @Override
-  public boolean producesSortedKeys(PipelineOptions options) throws Exception {
-    return false;
-  }
-
-  @Override
   public void validate() {
     super.validate();
     checkNotNull(

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/321058ff/sdks/java/core/src/main/java/org/apache/beam/sdk/testing/SourceTestUtils.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/testing/SourceTestUtils.java
b/sdks/java/core/src/main/java/org/apache/beam/sdk/testing/SourceTestUtils.java
index dd62aeb..a2a33f3 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/testing/SourceTestUtils.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/testing/SourceTestUtils.java
@@ -719,11 +719,6 @@ public class SourceTestUtils {
     }
 
     @Override
-    public boolean producesSortedKeys(PipelineOptions options) throws Exception {
-      return boundedSource.producesSortedKeys(options);
-    }
-
-    @Override
     public BoundedReader<T> createReader(PipelineOptions options) throws IOException
{
       return new UnsplittableReader<>(boundedSource, boundedSource.createReader(options));
     }

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/321058ff/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Create.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Create.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Create.java
index a48136f..4446517 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Create.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Create.java
@@ -319,11 +319,6 @@ public class Create<T> {
       }
 
       @Override
-      public boolean producesSortedKeys(PipelineOptions options) throws Exception {
-        return false;
-      }
-
-      @Override
       public BoundedSource.BoundedReader<T> createReader(PipelineOptions options)
           throws IOException {
         return new BytesReader<>(this);

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/321058ff/sdks/java/core/src/test/java/org/apache/beam/sdk/io/CompressedSourceTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/io/CompressedSourceTest.java
b/sdks/java/core/src/test/java/org/apache/beam/sdk/io/CompressedSourceTest.java
index 3871159..1574559 100644
--- a/sdks/java/core/src/test/java/org/apache/beam/sdk/io/CompressedSourceTest.java
+++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/io/CompressedSourceTest.java
@@ -513,11 +513,6 @@ public class CompressedSourceTest {
     }
 
     @Override
-    public boolean producesSortedKeys(PipelineOptions options) throws Exception {
-      return false;
-    }
-
-    @Override
     public Coder<Byte> getDefaultOutputCoder() {
       return SerializableCoder.of(Byte.class);
     }

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/321058ff/sdks/java/core/src/test/java/org/apache/beam/sdk/io/FileBasedSourceTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/io/FileBasedSourceTest.java
b/sdks/java/core/src/test/java/org/apache/beam/sdk/io/FileBasedSourceTest.java
index f709e22..a889305 100644
--- a/sdks/java/core/src/test/java/org/apache/beam/sdk/io/FileBasedSourceTest.java
+++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/io/FileBasedSourceTest.java
@@ -104,11 +104,6 @@ public class FileBasedSourceTest {
     }
 
     @Override
-    public boolean producesSortedKeys(PipelineOptions options) throws Exception {
-      return false;
-    }
-
-    @Override
     public void validate() {}
 
     @Override

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/321058ff/sdks/java/core/src/test/java/org/apache/beam/sdk/io/OffsetBasedSourceTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/io/OffsetBasedSourceTest.java
b/sdks/java/core/src/test/java/org/apache/beam/sdk/io/OffsetBasedSourceTest.java
index 6584e5d..a300a9a 100644
--- a/sdks/java/core/src/test/java/org/apache/beam/sdk/io/OffsetBasedSourceTest.java
+++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/io/OffsetBasedSourceTest.java
@@ -62,11 +62,6 @@ public class OffsetBasedSourceTest {
     }
 
     @Override
-    public boolean producesSortedKeys(PipelineOptions options) {
-      return false;
-    }
-
-    @Override
     public void validate() {}
 
     @Override

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/321058ff/sdks/java/core/src/test/java/org/apache/beam/sdk/io/ReadTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/io/ReadTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/io/ReadTest.java
index 2e90f9a..362139b 100644
--- a/sdks/java/core/src/test/java/org/apache/beam/sdk/io/ReadTest.java
+++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/io/ReadTest.java
@@ -163,11 +163,6 @@ public class ReadTest implements Serializable{
     }
 
     @Override
-    public boolean producesSortedKeys(PipelineOptions options) throws Exception {
-      return false;
-    }
-
-    @Override
     public BoundedReader<String> createReader(PipelineOptions options) throws IOException
{
       return null;
     }

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/321058ff/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/CreateTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/CreateTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/CreateTest.java
index 3bc0a65..a27ba1a 100644
--- a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/CreateTest.java
+++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/CreateTest.java
@@ -23,7 +23,6 @@ import static org.apache.beam.sdk.TestUtils.NO_LINES;
 import static org.apache.beam.sdk.TestUtils.NO_LINES_ARRAY;
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.hasSize;
-import static org.hamcrest.Matchers.is;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertThat;
 
@@ -347,13 +346,6 @@ public class CreateTest {
   }
 
   @Test
-  public void testSourceDoesNotProduceSortedKeys() throws Exception {
-    CreateSource<String> source =
-        CreateSource.fromIterable(ImmutableList.of("spam", "ham", "eggs"), StringUtf8Coder.of());
-    assertThat(source.producesSortedKeys(PipelineOptionsFactory.create()), is(false));
-  }
-
-  @Test
   public void testSourceGetDefaultOutputCoderReturnsConstructorCoder() throws Exception {
     Coder<Integer> coder = VarIntCoder.of();
     CreateSource<Integer> source =

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/321058ff/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIO.java
----------------------------------------------------------------------
diff --git a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIO.java
b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIO.java
index 7bb1e51..de9393a 100644
--- a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIO.java
+++ b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIO.java
@@ -1200,11 +1200,6 @@ public class BigQueryIO {
     protected abstract void cleanupTempResource(BigQueryOptions bqOptions) throws Exception;
 
     @Override
-    public boolean producesSortedKeys(PipelineOptions options) throws Exception {
-      return false;
-    }
-
-    @Override
     public void validate() {
       // Do nothing, validation is done in BigQuery.Read.
     }
@@ -1336,11 +1331,6 @@ public class BigQueryIO {
     }
 
     @Override
-    public boolean producesSortedKeys(PipelineOptions options) throws Exception {
-      return boundedSource.producesSortedKeys(options);
-    }
-
-    @Override
     public BoundedReader<V> createReader(PipelineOptions options) throws IOException
{
       return new TransformingReader(boundedSource.createReader(options));
     }

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/321058ff/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigtable/BigtableIO.java
----------------------------------------------------------------------
diff --git a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigtable/BigtableIO.java
b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigtable/BigtableIO.java
index c27389d..2d6cbba 100644
--- a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigtable/BigtableIO.java
+++ b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigtable/BigtableIO.java
@@ -872,14 +872,6 @@ public class BigtableIO {
       return estimatedSizeBytes;
     }
 
-    /**
-     * Cloud Bigtable returns query results ordered by key.
-     */
-    @Override
-    public boolean producesSortedKeys(PipelineOptions options) throws Exception {
-      return true;
-    }
-
     @Override
     public BoundedReader<Row> createReader(PipelineOptions options) throws IOException
{
       return new BigtableReader(this, serviceFactory.apply(options));

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/321058ff/sdks/java/io/hdfs/src/main/java/org/apache/beam/sdk/io/hdfs/HDFSFileSource.java
----------------------------------------------------------------------
diff --git a/sdks/java/io/hdfs/src/main/java/org/apache/beam/sdk/io/hdfs/HDFSFileSource.java
b/sdks/java/io/hdfs/src/main/java/org/apache/beam/sdk/io/hdfs/HDFSFileSource.java
index 5b0c5b6..61660de 100644
--- a/sdks/java/io/hdfs/src/main/java/org/apache/beam/sdk/io/hdfs/HDFSFileSource.java
+++ b/sdks/java/io/hdfs/src/main/java/org/apache/beam/sdk/io/hdfs/HDFSFileSource.java
@@ -260,11 +260,6 @@ public class HDFSFileSource<K, V> extends BoundedSource<KV<K,
V>> {
     return stat;
   }
 
-  @Override
-  public boolean producesSortedKeys(PipelineOptions options) throws Exception {
-    return false;
-  }
-
   static class HDFSFileReader<K, V> extends BoundedSource.BoundedReader<KV<K,
V>> {
 
     private final BoundedSource<KV<K, V>> source;

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/321058ff/sdks/java/io/mongodb/src/main/java/org/apache/beam/sdk/io/mongodb/MongoDbGridFSIO.java
----------------------------------------------------------------------
diff --git a/sdks/java/io/mongodb/src/main/java/org/apache/beam/sdk/io/mongodb/MongoDbGridFSIO.java
b/sdks/java/io/mongodb/src/main/java/org/apache/beam/sdk/io/mongodb/MongoDbGridFSIO.java
index b65e671..b356dad 100644
--- a/sdks/java/io/mongodb/src/main/java/org/apache/beam/sdk/io/mongodb/MongoDbGridFSIO.java
+++ b/sdks/java/io/mongodb/src/main/java/org/apache/beam/sdk/io/mongodb/MongoDbGridFSIO.java
@@ -429,11 +429,6 @@ public class MongoDbGridFSIO {
       }
 
       @Override
-      public boolean producesSortedKeys(PipelineOptions options) throws Exception {
-        return false;
-      }
-
-      @Override
       public BoundedSource.BoundedReader<ObjectId> createReader(
           PipelineOptions options) throws IOException {
         return new GridFSReader(this, objectIds);

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/321058ff/sdks/java/io/mongodb/src/main/java/org/apache/beam/sdk/io/mongodb/MongoDbIO.java
----------------------------------------------------------------------
diff --git a/sdks/java/io/mongodb/src/main/java/org/apache/beam/sdk/io/mongodb/MongoDbIO.java
b/sdks/java/io/mongodb/src/main/java/org/apache/beam/sdk/io/mongodb/MongoDbIO.java
index de2eb16..bd262e7 100644
--- a/sdks/java/io/mongodb/src/main/java/org/apache/beam/sdk/io/mongodb/MongoDbIO.java
+++ b/sdks/java/io/mongodb/src/main/java/org/apache/beam/sdk/io/mongodb/MongoDbIO.java
@@ -205,11 +205,6 @@ public class MongoDbIO {
     }
 
     @Override
-    public boolean producesSortedKeys(PipelineOptions options) {
-      return false;
-    }
-
-    @Override
     public BoundedReader<Document> createReader(PipelineOptions options) {
       return new BoundedMongoDbReader(this);
     }


Mime
View raw message