Add PipelineOptions changes for merging ValidatesRunner tests.

sdks/java/core/src/main/java/org/apache/beam/sdk/options/PipelineOptionsFactory.java

@@ -85,6 +85,7 @@
 import org.apache.beam.model.jobmanagement.v1.JobApi.PipelineOptionDescriptor;
 import org.apache.beam.model.jobmanagement.v1.JobApi.PipelineOptionType;
 import org.apache.beam.sdk.PipelineRunner;
+import org.apache.beam.sdk.annotations.Internal;
 import org.apache.beam.sdk.options.Validation.Required;
 import org.apache.beam.sdk.runners.PipelineRunnerRegistrar;
 import org.apache.beam.sdk.transforms.display.DisplayData;
@@ -575,6 +576,49 @@ public static Set<Class<? extends PipelineOptions>> getRegisteredOptions() {
     return Collections.unmodifiableSet(CACHE.get().registeredOptions);
   }
 
+  /**
+   * <b><i>For internal use only; no backwards-compatibility guarantees.</i></b>
+   *
+   * <p>Jackson serialization attribute (see {@link
+   * com.fasterxml.jackson.databind.ObjectWriter#withAttribute}) which, when set to {@code true},
+   * makes the serialized form of a {@link PipelineOptions} additionally record which of the
+   * serialized values were lazily bound defaults, so that {@link #explicitlySetProperties} on the
+   * deserialized copy gives the same answer as on the original.
+   *
+   * <p>It is off by default: the serialized form handed to runners and workers is unchanged, and a
+   * {@link PipelineOptions} deserialized from JSON without this marker reports every serialized
+   * property as explicitly set.
+   */
+  @Internal
+  public static final String SERIALIZE_DEFAULTS_ATTRIBUTE =
+      ProxyInvocationHandler.SERIALIZE_DEFAULTS_ATTRIBUTE;
+
+  /**
+   * <b><i>For internal use only; no backwards-compatibility guarantees.</i></b>
+   *
+   * <p>Returns the names of the properties of {@code options} that have been explicitly set
+   * (through a setter, from command-line arguments, or from the JSON the object was deserialized
+   * from), as opposed to defaults that were lazily bound the first time a getter was called.
+   *
+   * <p>Unlike the serialized form of {@code options}, which contains every bound value, this allows
+   * callers to compare two options objects by what was actually configured on them without being
+   * confused by which getters happen to have been invoked on each. Note that a serialization round
+   * trip loses this distinction unless the producer set {@link #SERIALIZE_DEFAULTS_ATTRIBUTE}.
+   *
+   * @throws IllegalArgumentException if {@code options} was not created by this factory
+   */
+  @Internal
+  public static Set<String> explicitlySetProperties(PipelineOptions options) {
+    checkArgumentNotNull(options);
+    checkArgument(
+        Proxy.isProxyClass(options.getClass())
+            && Proxy.getInvocationHandler(options) instanceof ProxyInvocationHandler,
+        "PipelineOptions %s was not created by PipelineOptionsFactory",
+        options.getClass().getName());
+    return Collections.unmodifiableSet(
+        ((ProxyInvocationHandler) Proxy.getInvocationHandler(options)).explicitlySetProperties());
+  }
+
   /**
    * Outputs the set of registered options with the PipelineOptionsFactory with a description for
    * each one if available to the output stream. This output is pretty printed and meant to be human

sdks/java/core/src/main/java/org/apache/beam/sdk/options/ProxyInvocationHandler.java

@@ -53,6 +53,7 @@
 import java.util.Set;
 import java.util.SortedMap;
 import java.util.TreeMap;
+import java.util.TreeSet;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ThreadLocalRandom;
 import java.util.concurrent.atomic.AtomicInteger;
@@ -163,6 +164,26 @@ <T extends PipelineOptions> ComputedProperties updated(
 
   private final ImmutableMap<String, JsonNode> jsonOptions;
 
+  /**
+   * Names of the entries in {@link #jsonOptions} that were lazily bound defaults, rather than
+   * explicitly set values, on the instance that was serialized. Only populated when that instance
+   * was serialized with {@link #SERIALIZE_DEFAULTS_ATTRIBUTE}; otherwise empty, in which case every
+   * entry of {@link #jsonOptions} is treated as explicitly set.
+   */
+  private final ImmutableSet<String> jsonDefaults;
+
+  /**
+   * Jackson serialization attribute which, when set to {@code true} on the {@link
+   * com.fasterxml.jackson.databind.ObjectWriter}, makes {@link Serializer} additionally record
+   * which serialized values were lazily bound defaults, so that {@link #explicitlySetProperties()}
+   * on the deserialized instance can tell them apart from explicitly set values. Off by default so
+   * that the serialized form seen by runners and workers is unchanged.
+   */
+  static final String SERIALIZE_DEFAULTS_ATTRIBUTE =
+      "org.apache.beam.sdk.options.ProxyInvocationHandler.serializeDefaults";
+
+  private static final String DEFAULTS_FIELD = "defaults";
+
   ProxyInvocationHandler(Map<String, Object> options) {
     this(bindOptions(options), Maps.newHashMap(), 0);
   }
@@ -178,8 +199,17 @@ private static Map<String, BoundValue> bindOptions(Map<String, Object> inputOpti
 
   private ProxyInvocationHandler(
       Map<String, BoundValue> options, Map<String, JsonNode> jsonOptions, int revision) {
+    this(options, jsonOptions, ImmutableSet.of(), revision);
+  }
+
+  private ProxyInvocationHandler(
+      Map<String, BoundValue> options,
+      Map<String, JsonNode> jsonOptions,
+      Set<String> jsonDefaults,
+      int revision) {
     this.options = new ConcurrentHashMap<>(options);
     this.jsonOptions = ImmutableMap.copyOf(jsonOptions);
+    this.jsonDefaults = ImmutableSet.copyOf(jsonDefaults);
     this.revision = new AtomicInteger(revision);
     this.computedProperties =
         new ComputedProperties(
@@ -249,6 +279,34 @@ public String getOptionName(Method method) {
     return computedProperties.gettersToPropertyNames.get(method.getName());
   }
 
+  /**
+   * Returns the names of the properties that have been explicitly set, either through a setter or
+   * by construction from arguments or JSON, as opposed to defaults that were lazily bound by a
+   * getter.
+   *
+   * <p>Properties present in the JSON this instance was deserialized from are counted as explicitly
+   * set unless that JSON was produced with {@link #SERIALIZE_DEFAULTS_ATTRIBUTE} and marked them as
+   * defaults; without that marker the serialized form does not record whether a value was a default
+   * on the instance that was serialized.
+   */
+  Set<String> explicitlySetProperties() {
+    Set<String> result = new HashSet<>();
+    for (String name : Sets.union(jsonOptions.keySet(), options.keySet())) {
+      if (isExplicitlySet(name)) {
+        result.add(name);
+      }
+    }
+    return result;
+  }
+
+  private boolean isExplicitlySet(String name) {
+    BoundValue bound = options.get(name);
+    if (bound != null && !bound.isDefault()) {
+      return true;
+    }
+    return jsonOptions.containsKey(name) && !jsonDefaults.contains(name);
+  }
+
   private void writeObject(java.io.ObjectOutputStream stream) throws IOException {
     throw new NotSerializableException(
         "PipelineOptions objects are not serializable and should not be embedded into transforms "
@@ -810,6 +868,16 @@ public void serialize(PipelineOptions value, JsonGenerator jgen, SerializerProvi
       jgen.writeObject(serializedDisplayData);
 
       jgen.writeNumberField("revision", handler.revision.get());
+      if (Boolean.TRUE.equals(provider.getAttribute(SERIALIZE_DEFAULTS_ATTRIBUTE))) {
+        Set<String> defaults = new TreeSet<>();
+        for (String name : serializableOptions.keySet()) {
+          if (!handler.isExplicitlySet(name)) {
+            defaults.add(name);
+          }
+        }
+        jgen.writeFieldName(DEFAULTS_FIELD);
+        jgen.writeObject(defaults);
+      }
       jgen.writeEndObject();
     }
 
@@ -909,8 +977,18 @@ public PipelineOptions deserialize(JsonParser jp, DeserializationContext ctxt)
         }
       }
       int revision = objectNode.hasNonNull("revision") ? objectNode.get("revision").asInt() : 0;
+      Set<String> defaults = new HashSet<>();
+      JsonNode defaultsNode = objectNode.get(DEFAULTS_FIELD);
+      if (defaultsNode != null && defaultsNode.isArray()) {
+        for (JsonNode name : defaultsNode) {
+          if (name.isTextual() && fields.containsKey(name.asText())) {
+            defaults.add(name.asText());
+          }
+        }
+      }
       PipelineOptions options =
-          new ProxyInvocationHandler(Maps.newHashMap(), fields, revision).as(PipelineOptions.class);
+          new ProxyInvocationHandler(Maps.newHashMap(), fields, defaults, revision)
+              .as(PipelineOptions.class);
       ValueProvider.RuntimeValueProvider.setRuntimeOptions(options);
       return options;
     }

sdks/java/core/src/main/java/org/apache/beam/sdk/testing/TestPipeline.java

@@ -434,8 +434,21 @@ public PipelineResult run(PipelineOptions options) {
     final PipelineResult pipelineResult;
     try {
       enforcement.get().beforePipelineExecution();
-      PipelineOptions updatedOptions =
-          MAPPER.convertValue(MAPPER.valueToTree(options), PipelineOptions.class);
+      // Round-trip through JSON so that the pipeline runs on a private copy of the options. The
+      // defaults marker keeps PipelineOptionsFactory.explicitlySetProperties meaningful on that
+      // copy; it is only ever set here and does not reach the serialized form of a submitted job.
+      PipelineOptions updatedOptions;
+      try {
+        updatedOptions =
+            MAPPER.readValue(
+                MAPPER
+                    .writer()
+                    .withAttribute(PipelineOptionsFactory.SERIALIZE_DEFAULTS_ATTRIBUTE, true)
+                    .writeValueAsBytes(options),
+                PipelineOptions.class);
+      } catch (IOException e) {
+        throw new IllegalArgumentException(e.getMessage(), e);
+      }
       updatedOptions
           .as(TestValueProviderOptions.class)
           .setProviderRuntimeValues(StaticValueProvider.of(providerRuntimeValues));

sdks/java/core/src/test/java/org/apache/beam/sdk/options/ProxyInvocationHandlerTest.java

@@ -39,6 +39,7 @@
 import com.fasterxml.jackson.core.JsonGenerator;
 import com.fasterxml.jackson.core.JsonParser;
 import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.core.type.TypeReference;
 import com.fasterxml.jackson.databind.DeserializationContext;
 import com.fasterxml.jackson.databind.JsonMappingException;
 import com.fasterxml.jackson.databind.JsonNode;
@@ -1111,6 +1112,76 @@ public void testDisplayDataIncludesExplicitlySetDefaults() {
     assertThat(data, hasDisplayItem("foo", defaultValue));
   }
 
+  @Test
+  public void testExplicitlySetPropertiesIgnoresLazilyBoundDefaults() {
+    HasDefaults options = PipelineOptionsFactory.as(HasDefaults.class);
+    assertThat(PipelineOptionsFactory.explicitlySetProperties(options), not(hasItem("foo")));
+
+    // Reading a default binds it, but does not make it explicitly set.
+    assertEquals("bar", options.getFoo());
+    assertThat(PipelineOptionsFactory.explicitlySetProperties(options), not(hasItem("foo")));
+
+    // Setting it, even to the default value, does.
+    options.setFoo("bar");
+    assertThat(PipelineOptionsFactory.explicitlySetProperties(options), hasItem("foo"));
+  }
+
+  @Test
+  public void testExplicitlySetPropertiesIncludesArgumentsAndJsonOptions() throws Exception {
+    FooOptions options = PipelineOptionsFactory.fromArgs("--foo=baz").as(FooOptions.class);
+    assertThat(PipelineOptionsFactory.explicitlySetProperties(options), hasItem("foo"));
+
+    FooOptions deserialized = serializeDeserialize(FooOptions.class, options);
+    assertThat(PipelineOptionsFactory.explicitlySetProperties(deserialized), hasItem("foo"));
+  }
+
+  @Test
+  public void testSerializationWithoutDefaultsMarkerIsUnchangedAndCountsAllAsExplicit()
+      throws Exception {
+    HasDefaults options = PipelineOptionsFactory.as(HasDefaults.class);
+    assertEquals("bar", options.getFoo()); // binds the default
+
+    JsonNode serialized = MAPPER.readTree(MAPPER.writeValueAsString(options));
+    assertTrue(serialized.has("options"));
+    assertTrue(serialized.has("display_data"));
+    assertTrue(serialized.has("revision"));
+    assertFalse("defaults marker must be opt-in", serialized.has("defaults"));
+
+    HasDefaults deserialized = serializeDeserialize(HasDefaults.class, options);
+    assertThat(PipelineOptionsFactory.explicitlySetProperties(deserialized), hasItem("foo"));
+  }
+
+  @Test
+  public void testSerializationWithDefaultsMarkerPreservesExplicitness() throws Exception {
+    HasDefaults options = PipelineOptionsFactory.fromArgs("--foo=explicit").as(HasDefaults.class);
+    options.as(ObjectPipelineOptions.class).getValue(); // binds a default
+
+    byte[] serialized =
+        MAPPER
+            .writer()
+            .withAttribute(PipelineOptionsFactory.SERIALIZE_DEFAULTS_ATTRIBUTE, true)
+            .writeValueAsBytes(options);
+    JsonNode tree = MAPPER.readTree(serialized);
+    assertTrue(tree.get("options").has("value"));
+    assertThat(
+        MAPPER.convertValue(tree.get("defaults"), new TypeReference<List<String>>() {}),
+        hasItem("value"));
+
+    PipelineOptions deserialized = MAPPER.readValue(serialized, PipelineOptions.class);
+    assertThat(PipelineOptionsFactory.explicitlySetProperties(deserialized), hasItem("foo"));
+    assertThat(PipelineOptionsFactory.explicitlySetProperties(deserialized), not(hasItem("value")));
+    // The marked default is still served from the JSON.
+    assertNull(deserialized.as(ObjectPipelineOptions.class).getValue());
+
+    // Setting the property on the copy makes it explicit.
+    deserialized.as(ObjectPipelineOptions.class).setValue("set");
+    assertThat(PipelineOptionsFactory.explicitlySetProperties(deserialized), hasItem("value"));
+
+    // A further serialization without the attribute drops the marker again.
+    JsonNode reserialized = MAPPER.readTree(MAPPER.writeValueAsString(deserialized));
+    assertFalse(reserialized.has("defaults"));
+  }
+
   @Test
   public void testDisplayDataNullValuesConvertedToEmptyString() throws Exception {
     FooOptions options = PipelineOptionsFactory.as(FooOptions.class);

sdks/java/core/src/test/java/org/apache/beam/sdk/testing/TestPipelineTest.java

@@ -18,10 +18,14 @@
 package org.apache.beam.sdk.testing;
 
 import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.hasItem;
+import static org.hamcrest.Matchers.not;
+import static org.hamcrest.Matchers.sameInstance;
 import static org.hamcrest.Matchers.startsWith;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.fail;
 
 import com.fasterxml.jackson.databind.ObjectMapper;
 import java.io.Serializable;
@@ -33,9 +37,12 @@
 import java.util.UUID;
 import org.apache.beam.sdk.Pipeline;
 import org.apache.beam.sdk.PipelineResult;
+import org.apache.beam.sdk.PipelineRunner;
 import org.apache.beam.sdk.coders.StringUtf8Coder;
 import org.apache.beam.sdk.options.ApplicationNameOptions;
 import org.apache.beam.sdk.options.PipelineOptions;
+import org.apache.beam.sdk.options.PipelineOptionsFactory;
+import org.apache.beam.sdk.options.StreamingOptions;
 import org.apache.beam.sdk.options.ValueProvider;
 import org.apache.beam.sdk.transforms.Create;
 import org.apache.beam.sdk.transforms.MapElements;
@@ -121,6 +128,48 @@ public void testRunWithDummyEnvironmentVariableFails() {
       pipeline.run();
     }
 
+    @Test
+    public void testRunPreservesWhichOptionsWereExplicitlySet() {
+      pipeline.getOptions().setRunner(OptionsCapturingRunner.class);
+      // Reading a default binds it on the options object without making it explicitly set.
+      assertFalse(pipeline.getOptions().as(StreamingOptions.class).isStreaming());
+      pipeline.apply(Create.of(1, 2, 3));
+
+      OptionsCapturingRunner.captured = null;
+      try {
+        pipeline.run();
+        fail("expected the runner to throw");
+      } catch (UnsupportedOperationException expected) {
+        // The runner always throws; what matters is the options it was created with.
+      }
+
+      PipelineOptions seenByRunner = OptionsCapturingRunner.captured;
+      assertNotNull(seenByRunner);
+      assertThat(seenByRunner, not(sameInstance(pipeline.getOptions())));
+      // The copy still serves the bound default ...
+      assertFalse(seenByRunner.as(StreamingOptions.class).isStreaming());
+      // ... but does not report it as explicitly set, while explicitly set options are reported.
+      assertThat(
+          PipelineOptionsFactory.explicitlySetProperties(seenByRunner), not(hasItem("streaming")));
+      assertThat(PipelineOptionsFactory.explicitlySetProperties(seenByRunner), hasItem("runner"));
+    }
+
+    /** A runner that records the options it was constructed from and refuses to run. */
+    public static class OptionsCapturingRunner extends PipelineRunner<PipelineResult> {
+      static volatile @Nullable PipelineOptions captured;
+
+      @SuppressWarnings("unused") // used by reflection
+      public static OptionsCapturingRunner fromOptions(PipelineOptions options) {
+        captured = options;
+        return new OptionsCapturingRunner();
+      }
+
+      @Override
+      public PipelineResult run(Pipeline pipeline) {
+        throw new UnsupportedOperationException("OptionsCapturingRunner does not run pipelines");
+      }
+    }
+
     /** TestMatcher is a matcher designed for testing matcher serialization/deserialization. */
     public static class TestMatcher extends BaseMatcher<PipelineResult>
         implements SerializableMatcher<PipelineResult> {