beam-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From bchamb...@apache.org
Subject [1/2] incubator-beam git commit: Validate display data on transform primitives
Date Mon, 16 May 2016 17:30:41 GMT
Repository: incubator-beam
Updated Branches:
  refs/heads/master 9b5cf3247 -> c0efe568e


Validate display data on transform primitives


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

Branch: refs/heads/master
Commit: 6269b8fa06daa4608f5ce5b11054cca58747a229
Parents: 9b5cf32
Author: Scott Wegner <swegner@google.com>
Authored: Mon May 9 13:31:39 2016 -0700
Committer: Scott Wegner <swegner@google.com>
Committed: Mon May 16 09:41:21 2016 -0700

----------------------------------------------------------------------
 runners/google-cloud-dataflow-java/pom.xml      | 10 +++
 .../runners/dataflow/io/DataflowAvroIOTest.java | 69 ++++++++++++++++++++
 .../dataflow/io/DataflowBigQueryIOTest.java     | 29 +++++++-
 .../dataflow/io/DataflowDatastoreIOTest.java    | 63 ++++++++++++++++++
 .../dataflow/io/DataflowPubsubIOTest.java       | 62 ++++++++++++++++++
 .../runners/dataflow/io/DataflowTextIOTest.java | 15 ++++-
 .../DataflowDisplayDataEvaluator.java           |  3 +-
 .../java/org/apache/beam/sdk/io/BigQueryIO.java | 11 ++++
 .../java/org/apache/beam/sdk/io/PubsubIO.java   |  5 ++
 .../org/apache/beam/sdk/io/BigQueryIOTest.java  |  6 +-
 .../beam/sdk/io/CompressedSourceTest.java       |  3 +-
 .../sdk/options/ProxyInvocationHandlerTest.java |  6 +-
 .../sdk/transforms/ApproximateUniqueTest.java   |  3 +-
 .../display/DisplayDataEvaluator.java           | 28 ++++----
 .../display/DisplayDataEvaluatorTest.java       | 24 ++++++-
 .../transforms/display/DisplayDataMatchers.java |  8 +++
 .../display/DisplayDataMatchersTest.java        |  3 +-
 .../sdk/transforms/windowing/WindowTest.java    |  4 +-
 .../sdk/io/gcp/bigtable/BigtableIOTest.java     |  9 +--
 19 files changed, 316 insertions(+), 45 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/6269b8fa/runners/google-cloud-dataflow-java/pom.xml
----------------------------------------------------------------------
diff --git a/runners/google-cloud-dataflow-java/pom.xml b/runners/google-cloud-dataflow-java/pom.xml
index a1d48ce..f7f1d80 100644
--- a/runners/google-cloud-dataflow-java/pom.xml
+++ b/runners/google-cloud-dataflow-java/pom.xml
@@ -288,11 +288,21 @@
     </dependency>
 
     <dependency>
+      <groupId>org.apache.avro</groupId>
+      <artifactId>avro</artifactId>
+    </dependency>
+
+    <dependency>
       <groupId>com.google.api-client</groupId>
       <artifactId>google-api-client</artifactId>
     </dependency>
 
     <dependency>
+      <groupId>com.google.apis</groupId>
+      <artifactId>google-api-services-datastore-protobuf</artifactId>
+    </dependency>
+
+    <dependency>
       <groupId>com.google.oauth-client</groupId>
       <artifactId>google-oauth-client</artifactId>
     </dependency>

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/6269b8fa/runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/io/DataflowAvroIOTest.java
----------------------------------------------------------------------
diff --git a/runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/io/DataflowAvroIOTest.java
b/runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/io/DataflowAvroIOTest.java
new file mode 100644
index 0000000..614affb
--- /dev/null
+++ b/runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/io/DataflowAvroIOTest.java
@@ -0,0 +1,69 @@
+/*
+ * 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.runners.dataflow.io;
+
+import static org.apache.beam.sdk.transforms.display.DisplayDataMatchers.hasDisplayItem;
+
+import static org.hamcrest.Matchers.hasItem;
+import static org.junit.Assert.assertThat;
+
+import org.apache.beam.runners.dataflow.DataflowPipelineRunner;
+import org.apache.beam.runners.dataflow.transforms.DataflowDisplayDataEvaluator;
+import org.apache.beam.sdk.io.AvroIO;
+import org.apache.beam.sdk.transforms.display.DisplayData;
+import org.apache.beam.sdk.transforms.display.DisplayDataEvaluator;
+
+import org.apache.avro.Schema;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+
+import java.util.Set;
+
+/**
+ * {@link DataflowPipelineRunner} specific tests for {@link AvroIO} transforms.
+ */
+@RunWith(JUnit4.class)
+public class DataflowAvroIOTest {
+  @Test
+  public void testPrimitiveWriteDisplayData() {
+    DisplayDataEvaluator evaluator = DataflowDisplayDataEvaluator.create();
+
+    AvroIO.Write.Bound<?> write = AvroIO.Write
+        .to("foo")
+        .withSchema(Schema.create(Schema.Type.STRING))
+        .withoutValidation();
+
+    Set<DisplayData> displayData = evaluator.displayDataForPrimitiveTransforms(write);
+    assertThat("AvroIO.Write should include the file pattern in its primitive transform",
+        displayData, hasItem(hasDisplayItem("fileNamePattern")));
+  }
+
+  @Test
+  public void testPrimitiveReadDisplayData() {
+    DisplayDataEvaluator evaluator = DataflowDisplayDataEvaluator.create();
+
+    AvroIO.Read.Bound<?> read = AvroIO.Read.from("foo.*")
+        .withSchema(Schema.create(Schema.Type.STRING))
+        .withoutValidation();
+
+    Set<DisplayData> displayData = evaluator.displayDataForPrimitiveTransforms(read);
+    assertThat("AvroIO.Read should include the file pattern in its primitive transform",
+        displayData, hasItem(hasDisplayItem("filePattern")));
+  }
+}

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/6269b8fa/runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/io/DataflowBigQueryIOTest.java
----------------------------------------------------------------------
diff --git a/runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/io/DataflowBigQueryIOTest.java
b/runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/io/DataflowBigQueryIOTest.java
index 619da04..2b13b9c 100644
--- a/runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/io/DataflowBigQueryIOTest.java
+++ b/runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/io/DataflowBigQueryIOTest.java
@@ -18,7 +18,6 @@
 package org.apache.beam.runners.dataflow.io;
 
 import static org.apache.beam.sdk.transforms.display.DisplayDataMatchers.hasDisplayItem;
-import static org.apache.beam.sdk.transforms.display.DisplayDataMatchers.hasKey;
 
 import static org.hamcrest.Matchers.hasItem;
 import static org.junit.Assert.assertThat;
@@ -40,6 +39,30 @@ import java.util.Set;
  */
 public class DataflowBigQueryIOTest {
   @Test
+  public void testTableSourcePrimitiveDisplayData() {
+    DisplayDataEvaluator evaluator = DataflowDisplayDataEvaluator.create();
+    BigQueryIO.Read.Bound read = BigQueryIO.Read
+        .from("project:dataset.tableId")
+        .withoutValidation();
+
+    Set<DisplayData> displayData = evaluator.displayDataForPrimitiveTransforms(read);
+    assertThat("BigQueryIO.Read should include the table spec in its primitive display data",
+        displayData, hasItem(hasDisplayItem("table")));
+  }
+
+  @Test
+  public void testQuerySourcePrimitiveDisplayData() {
+    DisplayDataEvaluator evaluator = DataflowDisplayDataEvaluator.create();
+    BigQueryIO.Read.Bound read = BigQueryIO.Read
+        .fromQuery("foobar")
+        .withoutValidation();
+
+    Set<DisplayData> displayData = evaluator.displayDataForPrimitiveTransforms(read);
+    assertThat("BigQueryIO.Read should include the query in its primitive display data",
+        displayData, hasItem(hasDisplayItem("query")));
+  }
+
+  @Test
   public void testBatchSinkPrimitiveDisplayData() {
     DataflowPipelineOptions options = DataflowDisplayDataEvaluator.getDefaultOptions();
     options.setStreaming(false);
@@ -63,9 +86,9 @@ public class DataflowBigQueryIOTest {
 
     Set<DisplayData> displayData = evaluator.displayDataForPrimitiveTransforms(write);
     assertThat("BigQueryIO.Write should include the table spec in its primitive display data",
-        displayData, hasItem(hasDisplayItem(hasKey("tableSpec"))));
+        displayData, hasItem(hasDisplayItem("tableSpec")));
 
     assertThat("BigQueryIO.Write should include the table schema in its primitive display
data",
-        displayData, hasItem(hasDisplayItem(hasKey("schema"))));
+        displayData, hasItem(hasDisplayItem("schema")));
   }
 }

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/6269b8fa/runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/io/DataflowDatastoreIOTest.java
----------------------------------------------------------------------
diff --git a/runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/io/DataflowDatastoreIOTest.java
b/runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/io/DataflowDatastoreIOTest.java
new file mode 100644
index 0000000..42a0b99
--- /dev/null
+++ b/runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/io/DataflowDatastoreIOTest.java
@@ -0,0 +1,63 @@
+/*
+ * 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.runners.dataflow.io;
+
+import static org.apache.beam.sdk.transforms.display.DisplayDataMatchers.hasDisplayItem;
+
+import static org.hamcrest.Matchers.hasItem;
+import static org.junit.Assert.assertThat;
+
+import org.apache.beam.runners.dataflow.transforms.DataflowDisplayDataEvaluator;
+import org.apache.beam.sdk.io.DatastoreIO;
+import org.apache.beam.sdk.transforms.PTransform;
+import org.apache.beam.sdk.transforms.display.DisplayData;
+import org.apache.beam.sdk.transforms.display.DisplayDataEvaluator;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.PInput;
+
+import com.google.api.services.datastore.DatastoreV1;
+
+import org.junit.Test;
+
+import java.util.Set;
+
+/**
+ * Unit tests for Dataflow usage of {@link DatastoreIO} transforms.
+ */
+public class DataflowDatastoreIOTest {
+  @Test
+  public void testSourcePrimitiveDisplayData() {
+    DisplayDataEvaluator evaluator = DataflowDisplayDataEvaluator.create();
+    PTransform<PInput, ?> read = DatastoreIO.readFrom(
+        "myDataset", DatastoreV1.Query.newBuilder().build());
+
+    Set<DisplayData> displayData = evaluator.displayDataForPrimitiveTransforms(read);
+    assertThat("DatastoreIO read should include the dataset in its primitive display data",
+        displayData, hasItem(hasDisplayItem("dataset")));
+  }
+
+  @Test
+  public void testSinkPrimitiveDisplayData() {
+    DisplayDataEvaluator evaluator = DataflowDisplayDataEvaluator.create();
+    PTransform<PCollection<DatastoreV1.Entity>, ?> write = DatastoreIO.writeTo("myDataset");
+
+    Set<DisplayData> displayData = evaluator.displayDataForPrimitiveTransforms(write);
+    assertThat("DatastoreIO write should include the dataset in its primitive display data",
+        displayData, hasItem(hasDisplayItem("dataset")));
+  }
+}

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/6269b8fa/runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/io/DataflowPubsubIOTest.java
----------------------------------------------------------------------
diff --git a/runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/io/DataflowPubsubIOTest.java
b/runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/io/DataflowPubsubIOTest.java
new file mode 100644
index 0000000..4874877
--- /dev/null
+++ b/runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/io/DataflowPubsubIOTest.java
@@ -0,0 +1,62 @@
+/*
+ * 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.runners.dataflow.io;
+
+import static org.apache.beam.sdk.transforms.display.DisplayDataMatchers.hasDisplayItem;
+
+import static org.hamcrest.Matchers.hasItem;
+import static org.junit.Assert.assertThat;
+
+import org.apache.beam.runners.dataflow.DataflowPipelineRunner;
+import org.apache.beam.runners.dataflow.transforms.DataflowDisplayDataEvaluator;
+import org.apache.beam.sdk.io.PubsubIO;
+import org.apache.beam.sdk.transforms.display.DisplayData;
+import org.apache.beam.sdk.transforms.display.DisplayDataEvaluator;
+
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+
+import java.util.Set;
+
+/**
+ * {@link DataflowPipelineRunner} specific tests for {@link PubsubIO} transforms.
+ */
+@RunWith(JUnit4.class)
+public class DataflowPubsubIOTest {
+  @Test
+  public void testPrimitiveWriteDisplayData() {
+    DisplayDataEvaluator evaluator = DataflowDisplayDataEvaluator.create();
+    PubsubIO.Write.Bound<?> write = PubsubIO.Write
+        .topic("projects/project/topics/topic");
+
+    Set<DisplayData> displayData = evaluator.displayDataForPrimitiveTransforms(write);
+    assertThat("PubsubIO.Write should include the topic in its primitive display data",
+        displayData, hasItem(hasDisplayItem("topic")));
+  }
+
+  @Test
+  public void testPrimitiveReadDisplayData() {
+    DisplayDataEvaluator evaluator = DataflowDisplayDataEvaluator.create();
+    PubsubIO.Read.Bound<String> read = PubsubIO.Read.topic("projects/project/topics/topic");
+
+    Set<DisplayData> displayData = evaluator.displayDataForPrimitiveTransforms(read);
+    assertThat("PubsubIO.Read should include the topic in its primitive display data",
+        displayData, hasItem(hasDisplayItem("topic")));
+  }
+}

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/6269b8fa/runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/io/DataflowTextIOTest.java
----------------------------------------------------------------------
diff --git a/runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/io/DataflowTextIOTest.java
b/runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/io/DataflowTextIOTest.java
index 8ff7d0e..0d7c1cb 100644
--- a/runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/io/DataflowTextIOTest.java
+++ b/runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/io/DataflowTextIOTest.java
@@ -128,7 +128,7 @@ public class DataflowTextIOTest {
   }
 
   @Test
-  public void testPrimitiveDisplayData() {
+  public void testPrimitiveWriteDisplayData() {
     DisplayDataEvaluator evaluator = DataflowDisplayDataEvaluator.create();
 
     TextIO.Write.Bound<?> write = TextIO.Write.to("foobar");
@@ -137,4 +137,17 @@ public class DataflowTextIOTest {
     assertThat("TextIO.Write should include the file prefix in its primitive display data",
         displayData, hasItem(hasDisplayItem(hasValue(startsWith("foobar")))));
   }
+
+  @Test
+  public void testPrimitiveReadDisplayData() {
+    DisplayDataEvaluator evaluator = DataflowDisplayDataEvaluator.create();
+
+    TextIO.Read.Bound<String> read = TextIO.Read
+        .from("foobar")
+        .withoutValidation();
+
+    Set<DisplayData> displayData = evaluator.displayDataForPrimitiveTransforms(read);
+    assertThat("TextIO.Read should include the file prefix in its primitive display data",
+        displayData, hasItem(hasDisplayItem(hasValue(startsWith("foobar")))));
+  }
 }

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/6269b8fa/runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/transforms/DataflowDisplayDataEvaluator.java
----------------------------------------------------------------------
diff --git a/runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/transforms/DataflowDisplayDataEvaluator.java
b/runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/transforms/DataflowDisplayDataEvaluator.java
index db42c77..0b865c3 100644
--- a/runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/transforms/DataflowDisplayDataEvaluator.java
+++ b/runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/transforms/DataflowDisplayDataEvaluator.java
@@ -43,6 +43,7 @@ public final class DataflowDisplayDataEvaluator {
   public static DataflowPipelineOptions getDefaultOptions() {
     DataflowPipelineOptions options = PipelineOptionsFactory.as(DataflowPipelineOptions.class);
 
+    options.setRunner(DataflowPipelineRunner.class);
     options.setProject("foobar");
     options.setTempLocation("gs://bucket/tmpLocation");
     options.setFilesToStage(Lists.<String>newArrayList());
@@ -66,6 +67,6 @@ public final class DataflowDisplayDataEvaluator {
    * the {@link DataflowPipelineRunner} with the specified {@code options}.
    */
   public static DisplayDataEvaluator create(DataflowPipelineOptions options) {
-    return DisplayDataEvaluator.forRunner(DataflowPipelineRunner.class, options);
+    return DisplayDataEvaluator.create(options);
   }
 }

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/6269b8fa/sdks/java/core/src/main/java/org/apache/beam/sdk/io/BigQueryIO.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/io/BigQueryIO.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/io/BigQueryIO.java
index a5ef39f..3f22648 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/io/BigQueryIO.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/io/BigQueryIO.java
@@ -802,6 +802,12 @@ public class BigQueryIO {
     protected void cleanupTempResource(BigQueryOptions bqOptions) throws Exception {
       // Do nothing.
     }
+
+    @Override
+    public void populateDisplayData(DisplayData.Builder builder) {
+      super.populateDisplayData(builder);
+      builder.add(DisplayData.item("table", jsonTable));
+    }
   }
 
   /**
@@ -893,6 +899,11 @@ public class BigQueryIO {
       tableService.deleteDataset(tableToRemove.getProjectId(), tableToRemove.getDatasetId());
     }
 
+    @Override
+    public void populateDisplayData(DisplayData.Builder builder) {
+      super.populateDisplayData(builder);
+      builder.add(DisplayData.item("query", query));
+    }
     private synchronized JobStatistics dryRunQueryIfNeeded(BigQueryOptions bqOptions)
         throws InterruptedException, IOException {
       if (dryRunJobStats.get() == null) {

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/6269b8fa/sdks/java/core/src/main/java/org/apache/beam/sdk/io/PubsubIO.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/io/PubsubIO.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/io/PubsubIO.java
index fa867c2..6a14477 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/io/PubsubIO.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/io/PubsubIO.java
@@ -1030,6 +1030,11 @@ public class PubsubIO {
           checkState(n == output.size());
           output.clear();
         }
+
+        @Override
+        public void populateDisplayData(DisplayData.Builder builder) {
+          Bound.this.populateDisplayData(builder);
+        }
       }
     }
 

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/6269b8fa/sdks/java/core/src/test/java/org/apache/beam/sdk/io/BigQueryIOTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/io/BigQueryIOTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/io/BigQueryIOTest.java
index d00914c..6849018 100644
--- a/sdks/java/core/src/test/java/org/apache/beam/sdk/io/BigQueryIOTest.java
+++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/io/BigQueryIOTest.java
@@ -20,7 +20,7 @@ package org.apache.beam.sdk.io;
 import static org.apache.beam.sdk.io.BigQueryIO.fromJsonString;
 import static org.apache.beam.sdk.io.BigQueryIO.toJsonString;
 import static org.apache.beam.sdk.transforms.display.DisplayDataMatchers.hasDisplayItem;
-import static org.apache.beam.sdk.transforms.display.DisplayDataMatchers.hasKey;
+
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertThat;
@@ -710,8 +710,8 @@ public class BigQueryIOTest implements Serializable {
 
     DisplayData displayData = DisplayData.from(write);
 
-    assertThat(displayData, hasDisplayItem(hasKey("table")));
-    assertThat(displayData, hasDisplayItem(hasKey("schema")));
+    assertThat(displayData, hasDisplayItem("table"));
+    assertThat(displayData, hasDisplayItem("schema"));
     assertThat(displayData,
         hasDisplayItem("createDisposition", CreateDisposition.CREATE_IF_NEEDED.toString()));
     assertThat(displayData,

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/6269b8fa/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 76e547f..542e734 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
@@ -18,7 +18,6 @@
 package org.apache.beam.sdk.io;
 
 import static org.apache.beam.sdk.transforms.display.DisplayDataMatchers.hasDisplayItem;
-import static org.apache.beam.sdk.transforms.display.DisplayDataMatchers.hasKey;
 import static org.apache.beam.sdk.transforms.display.DisplayDataMatchers.includesDisplayDataFrom;
 
 import static org.hamcrest.Matchers.containsString;
@@ -353,7 +352,7 @@ public class CompressedSourceTest {
     DisplayData compressedSourceDisplayData = DisplayData.from(compressedSource);
     DisplayData gzipDisplayData = DisplayData.from(gzipSource);
 
-    assertThat(compressedSourceDisplayData, hasDisplayItem(hasKey("compressionMode")));
+    assertThat(compressedSourceDisplayData, hasDisplayItem("compressionMode"));
     assertThat(gzipDisplayData, hasDisplayItem("compressionMode", CompressionMode.GZIP.toString()));
     assertThat(compressedSourceDisplayData, hasDisplayItem("source", inputSource.getClass()));
     assertThat(compressedSourceDisplayData, includesDisplayDataFrom(inputSource));

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/6269b8fa/sdks/java/core/src/test/java/org/apache/beam/sdk/options/ProxyInvocationHandlerTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/options/ProxyInvocationHandlerTest.java
b/sdks/java/core/src/test/java/org/apache/beam/sdk/options/ProxyInvocationHandlerTest.java
index 228a6ba..6fc9700 100644
--- a/sdks/java/core/src/test/java/org/apache/beam/sdk/options/ProxyInvocationHandlerTest.java
+++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/options/ProxyInvocationHandlerTest.java
@@ -834,7 +834,7 @@ public class ProxyInvocationHandlerTest {
     PipelineOptions options = PipelineOptionsFactory.as(HasDefaults.class);
     DisplayData data = DisplayData.from(options);
 
-    assertThat(data, not(hasDisplayItem(hasKey("foo"))));
+    assertThat(data, not(hasDisplayItem("foo")));
   }
 
   interface HasDefaults extends PipelineOptions {
@@ -849,7 +849,7 @@ public class ProxyInvocationHandlerTest {
     assertEquals("bar", options.getFoo());
 
     DisplayData data = DisplayData.from(options);
-    assertThat(data, not(hasDisplayItem(hasKey("foo"))));
+    assertThat(data, not(hasDisplayItem("foo")));
   }
 
   @Test
@@ -958,6 +958,6 @@ public class ProxyInvocationHandlerTest {
     options.setValue("foobar");
 
     DisplayData data = DisplayData.from(options);
-    assertThat(data, not(hasDisplayItem(hasKey("value"))));
+    assertThat(data, not(hasDisplayItem("value")));
   }
 }

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/6269b8fa/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/ApproximateUniqueTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/ApproximateUniqueTest.java
b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/ApproximateUniqueTest.java
index c94c9f1..4f00ed4 100644
--- a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/ApproximateUniqueTest.java
+++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/ApproximateUniqueTest.java
@@ -18,7 +18,6 @@
 package org.apache.beam.sdk.transforms;
 
 import static org.apache.beam.sdk.transforms.display.DisplayDataMatchers.hasDisplayItem;
-import static org.apache.beam.sdk.transforms.display.DisplayDataMatchers.hasKey;
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertTrue;
@@ -305,6 +304,6 @@ public class ApproximateUniqueTest implements Serializable {
     DisplayData maxErrorDisplayData = DisplayData.from(specifiedMaxError);
     assertThat(maxErrorDisplayData, hasDisplayItem("maximumEstimationError", 0.1234));
     assertThat("calculated sampleSize should be included", maxErrorDisplayData,
-        hasDisplayItem(hasKey("sampleSize")));
+        hasDisplayItem("sampleSize"));
   }
 }

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/6269b8fa/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/display/DisplayDataEvaluator.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/display/DisplayDataEvaluator.java
b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/display/DisplayDataEvaluator.java
index 6759003..a17e06f 100644
--- a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/display/DisplayDataEvaluator.java
+++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/display/DisplayDataEvaluator.java
@@ -20,9 +20,9 @@ package org.apache.beam.sdk.transforms.display;
 import org.apache.beam.sdk.Pipeline;
 import org.apache.beam.sdk.coders.Coder;
 import org.apache.beam.sdk.options.PipelineOptions;
-import org.apache.beam.sdk.options.PipelineOptionsFactory;
 import org.apache.beam.sdk.runners.PipelineRunner;
 import org.apache.beam.sdk.runners.TransformTreeNode;
+import org.apache.beam.sdk.testing.TestPipeline;
 import org.apache.beam.sdk.transforms.Create;
 import org.apache.beam.sdk.transforms.PTransform;
 import org.apache.beam.sdk.values.PCollection;
@@ -40,25 +40,21 @@ public class DisplayDataEvaluator {
   private final PipelineOptions options;
 
   /**
-   * Create a new {@link DisplayDataEvaluator} using the specified {@link PipelineRunner}
and
-   * default {@link PipelineOptions}.
+   * Create a new {@link DisplayDataEvaluator} using {@link TestPipeline#testingPipelineOptions()}.
    */
-  public static DisplayDataEvaluator forRunner(Class<? extends PipelineRunner<?>>
pipelineRunner) {
-    return forRunner(pipelineRunner, PipelineOptionsFactory.create());
+  public static DisplayDataEvaluator create() {
+    return create(TestPipeline.testingPipelineOptions());
   }
 
   /**
-   * Create a new {@link DisplayDataEvaluator} using the specified {@link PipelineRunner}
and
-   * {@link PipelineOptions}.
+   * Create a new {@link DisplayDataEvaluator} using the specified {@link PipelineOptions}.
    */
-  public static DisplayDataEvaluator forRunner(
-      Class<? extends PipelineRunner<?>> pipelineRunner, PipelineOptions pipelineOptions)
{
-    return new DisplayDataEvaluator(pipelineRunner, pipelineOptions);
+  public static DisplayDataEvaluator create(PipelineOptions pipelineOptions) {
+    return new DisplayDataEvaluator(pipelineOptions);
   }
 
-  private DisplayDataEvaluator(Class<? extends PipelineRunner<?>> runner, PipelineOptions
options) {
+  private DisplayDataEvaluator(PipelineOptions options) {
     this.options = options;
-    this.options.setRunner(runner);
   }
 
   /**
@@ -109,7 +105,7 @@ public class DisplayDataEvaluator {
   extends Pipeline.PipelineVisitor.Defaults {
     private final PTransform root;
     private final Set<DisplayData> displayData;
-    private boolean shouldRecord = false;
+    private boolean inCompositeRoot = false;
 
     PrimitiveDisplayDataPTransformVisitor(PTransform root) {
       this.root = root;
@@ -123,7 +119,7 @@ public class DisplayDataEvaluator {
     @Override
     public CompositeBehavior enterCompositeTransform(TransformTreeNode node) {
       if (Objects.equals(root, node.getTransform())) {
-        shouldRecord = true;
+        inCompositeRoot = true;
       }
 
       return CompositeBehavior.ENTER_TRANSFORM;
@@ -132,13 +128,13 @@ public class DisplayDataEvaluator {
     @Override
     public void leaveCompositeTransform(TransformTreeNode node) {
       if (Objects.equals(root, node.getTransform())) {
-        shouldRecord = false;
+        inCompositeRoot = false;
       }
     }
 
     @Override
     public void visitPrimitiveTransform(TransformTreeNode node) {
-      if (shouldRecord) {
+      if (inCompositeRoot || Objects.equals(root, node.getTransform())) {
         displayData.add(DisplayData.from(node.getTransform()));
       }
     }

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/6269b8fa/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/display/DisplayDataEvaluatorTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/display/DisplayDataEvaluatorTest.java
b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/display/DisplayDataEvaluatorTest.java
index f24c133..7b1dc79 100644
--- a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/display/DisplayDataEvaluatorTest.java
+++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/display/DisplayDataEvaluatorTest.java
@@ -23,7 +23,6 @@ import static org.hamcrest.Matchers.hasItem;
 import static org.hamcrest.Matchers.not;
 import static org.junit.Assert.assertThat;
 
-import org.apache.beam.sdk.runners.DirectPipelineRunner;
 import org.apache.beam.sdk.transforms.DoFn;
 import org.apache.beam.sdk.transforms.PTransform;
 import org.apache.beam.sdk.transforms.ParDo;
@@ -41,7 +40,7 @@ import java.util.Set;
 public class DisplayDataEvaluatorTest implements Serializable {
 
   @Test
-  public void testDisplayDataForPrimitiveTransforms() {
+  public void testCompositeTransform() {
     PTransform<? super PCollection<String>, ? super POutput> myTransform =
         new PTransform<PCollection<String>, POutput> () {
           @Override
@@ -65,10 +64,29 @@ public class DisplayDataEvaluatorTest implements Serializable {
           }
         };
 
-    DisplayDataEvaluator evaluator = DisplayDataEvaluator.forRunner(DirectPipelineRunner.class);
+    DisplayDataEvaluator evaluator = DisplayDataEvaluator.create();
     Set<DisplayData> displayData = evaluator.displayDataForPrimitiveTransforms(myTransform);
 
     assertThat(displayData, not(hasItem(hasDisplayItem("compositeKey", "compositeValue"))));
     assertThat(displayData, hasItem(hasDisplayItem("primitiveKey", "primitiveValue")));
   }
+
+  @Test
+  public void testPrimitiveTransform() {
+    PTransform<? super PCollection<Integer>, ? super PCollection<Integer>>
myTransform = ParDo.of(
+        new DoFn<Integer, Integer>() {
+      @Override
+      public void processElement(ProcessContext c) throws Exception {}
+
+      @Override
+      public void populateDisplayData(DisplayData.Builder builder) {
+        builder.add(DisplayData.item("foo", "bar"));
+      }
+    });
+
+    DisplayDataEvaluator evaluator = DisplayDataEvaluator.create();
+    Set<DisplayData> displayData = evaluator.displayDataForPrimitiveTransforms(myTransform);
+
+    assertThat(displayData, hasItem(hasDisplayItem("foo")));
+  }
 }

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/6269b8fa/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/display/DisplayDataMatchers.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/display/DisplayDataMatchers.java
b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/display/DisplayDataMatchers.java
index e3721b8..4207624 100644
--- a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/display/DisplayDataMatchers.java
+++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/display/DisplayDataMatchers.java
@@ -57,6 +57,14 @@ public class DisplayDataMatchers {
   }
 
   /**
+   * Creates a matcher that matches if the examined {@link DisplayData} contains an item
with the
+   * specified key.
+   */
+  public static Matcher<DisplayData> hasDisplayItem(String key) {
+    return hasDisplayItem(hasKey(key));
+  }
+
+  /**
    * Create a matcher that matches if the examined {@link DisplayData} contains an item with
the
    * specified key and String value.
    */

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/6269b8fa/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/display/DisplayDataMatchersTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/display/DisplayDataMatchersTest.java
b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/display/DisplayDataMatchersTest.java
index f848c5e..f9f2911 100644
--- a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/display/DisplayDataMatchersTest.java
+++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/display/DisplayDataMatchersTest.java
@@ -18,7 +18,6 @@
 package org.apache.beam.sdk.transforms.display;
 
 import static org.apache.beam.sdk.transforms.display.DisplayDataMatchers.hasDisplayItem;
-import static org.apache.beam.sdk.transforms.display.DisplayDataMatchers.hasKey;
 import static org.apache.beam.sdk.transforms.display.DisplayDataMatchers.hasNamespace;
 import static org.apache.beam.sdk.transforms.display.DisplayDataMatchers.hasType;
 import static org.apache.beam.sdk.transforms.display.DisplayDataMatchers.hasValue;
@@ -67,7 +66,7 @@ public class DisplayDataMatchersTest {
 
   @Test
   public void testHasKey() {
-    Matcher<DisplayData> matcher = hasDisplayItem(hasKey("foo"));
+    Matcher<DisplayData> matcher = hasDisplayItem("foo");
 
     assertFalse(matcher.matches(createDisplayDataWithItem("fooz", "bar")));
 

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/6269b8fa/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/windowing/WindowTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/windowing/WindowTest.java
b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/windowing/WindowTest.java
index cd5eb2d..c858f32 100644
--- a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/windowing/WindowTest.java
+++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/windowing/WindowTest.java
@@ -292,7 +292,7 @@ public class WindowTest implements Serializable {
         .withAllowedLateness(Duration.millis(BoundedWindow.TIMESTAMP_MAX_VALUE.getMillis()));
 
     DisplayData data = DisplayData.from(window);
-    assertThat(data, not(hasDisplayItem(hasKey("trigger"))));
-    assertThat(data, not(hasDisplayItem(hasKey("allowedLateness"))));
+    assertThat(data, not(hasDisplayItem("trigger")));
+    assertThat(data, not(hasDisplayItem("allowedLateness")));
   }
 }

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/6269b8fa/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigtable/BigtableIOTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigtable/BigtableIOTest.java
b/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigtable/BigtableIOTest.java
index 403ad9d..357ab44 100644
--- a/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigtable/BigtableIOTest.java
+++ b/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigtable/BigtableIOTest.java
@@ -20,10 +20,8 @@ package org.apache.beam.sdk.io.gcp.bigtable;
 import static org.apache.beam.sdk.testing.SourceTestUtils.assertSourcesEqualReferenceSource;
 import static org.apache.beam.sdk.testing.SourceTestUtils.assertSplitAtFractionExhaustive;
 import static org.apache.beam.sdk.testing.SourceTestUtils.assertSplitAtFractionFails;
-import static org.apache.beam.sdk.testing.SourceTestUtils
-    .assertSplitAtFractionSucceedsAndConsistent;
+import static org.apache.beam.sdk.testing.SourceTestUtils.assertSplitAtFractionSucceedsAndConsistent;
 import static org.apache.beam.sdk.transforms.display.DisplayDataMatchers.hasDisplayItem;
-import static org.apache.beam.sdk.transforms.display.DisplayDataMatchers.hasKey;
 
 import static com.google.common.base.Preconditions.checkArgument;
 import static com.google.common.base.Verify.verifyNotNull;
@@ -428,7 +426,7 @@ public class BigtableIOTest {
     assertThat(displayData, hasDisplayItem("rowFilter", rowFilter.toString()));
 
     // BigtableIO adds user-agent to options; assert only on key and not value.
-    assertThat(displayData, hasDisplayItem(hasKey("bigtableOptions")));
+    assertThat(displayData, hasDisplayItem("bigtableOptions"));
   }
 
   /** Tests that a record gets written to the service and messages are logged. */
@@ -494,10 +492,7 @@ public class BigtableIOTest {
         .withBigtableOptions(BIGTABLE_OPTIONS);
 
     DisplayData displayData = DisplayData.from(write);
-
     assertThat(displayData, hasDisplayItem("tableId", "fooTable"));
-    // BigtableIO adds user-agent to options; assert only on key and not value.
-    assertThat(displayData, hasDisplayItem(hasKey("bigtableOptions")));
   }
 
   ////////////////////////////////////////////////////////////////////////////////////////////



Mime
View raw message