[BEAM-3901] Euphoria: Initial import from github

runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/PTransformTranslation.java

@@ -273,6 +273,12 @@ public static String urnForTransform(PTransform<?, ?> transform) {
     return urn;
   }
 
+  /** Returns the URN for the transform if it is known, otherwise {@code null}. */
+  @Nullable
+  public static String urnForTransformOrNull(RunnerApi.PTransform transform) {
+    return transform.getSpec() == null ? null : transform.getSpec().getUrn();
+  }
+
   /**
    * A bi-directional translator between a Java-based {@link PTransform} and a protobuf payload for
    * that transform.

runners/direct-java/src/main/java/org/apache/beam/runners/direct/Clock.java

@@ -17,12 +17,15 @@
  */
 package org.apache.beam.runners.direct;
 
+import org.apache.beam.sdk.annotations.Internal;
 import org.joda.time.Instant;
 
 /**
  * Access to the current time.
  */
-interface Clock {
+@Internal
+@FunctionalInterface
+public interface Clock {
   /**
    * Returns the current time as an {@link Instant}.
    */

runners/direct-java/src/main/java/org/apache/beam/runners/direct/WatermarkManager.java

@@ -57,6 +57,7 @@
 import org.apache.beam.runners.local.Bundle;
 import org.apache.beam.runners.local.StructuralKey;
 import org.apache.beam.sdk.Pipeline;
+import org.apache.beam.sdk.annotations.Internal;
 import org.apache.beam.sdk.runners.AppliedPTransform;
 import org.apache.beam.sdk.state.TimeDomain;
 import org.apache.beam.sdk.transforms.PTransform;
@@ -126,7 +127,8 @@
  * Watermark_PCollection = Watermark_Out_ProducingPTransform
  * </pre>
  */
-class WatermarkManager<ExecutableT, CollectionT> {
+@Internal
+public class WatermarkManager<ExecutableT, CollectionT> {
   // The number of updates to apply in #tryApplyPendingUpdates
   private static final int MAX_INCREMENTAL_UPDATES = 10;
 
@@ -785,8 +787,9 @@ private static Map<StructuralKey<?>, List<TimerData>> extractFiredTimers(
    * @param clock the clock to use to determine processing time
    * @param graph the graph representing this pipeline
    */
-  public static WatermarkManager<AppliedPTransform<?, ?, ?>, ? super PCollection<?>> create(
-      Clock clock, DirectGraph graph) {
+  public static <ExecutableT, CollectionT>
+      WatermarkManager<ExecutableT, ? super CollectionT> create(
+          Clock clock, ExecutableGraph<ExecutableT, ? super CollectionT> graph) {
     return new WatermarkManager<>(clock, graph);
   }
 
@@ -1033,7 +1036,7 @@ private void updatePending(
    * Refresh the watermarks contained within this {@link WatermarkManager}, causing all
    * watermarks to be advanced as far as possible.
    */
-  synchronized void refreshAll() {
+  public synchronized void refreshAll() {
     refreshLock.lock();
     try {
       applyAllPendingUpdates();
@@ -1443,17 +1446,17 @@ StructuralKey<?> getKey() {
     }
 
     @VisibleForTesting
-    Iterable<? extends TimerData> getCompletedTimers() {
+    public Iterable<? extends TimerData> getCompletedTimers() {
       return completedTimers;
     }
 
     @VisibleForTesting
-    Iterable<? extends TimerData> getSetTimers() {
+    public Iterable<? extends TimerData> getSetTimers() {
       return setTimers;
     }
 
     @VisibleForTesting
-    Iterable<? extends TimerData> getDeletedTimers() {
+    public Iterable<? extends TimerData> getDeletedTimers() {
       return deletedTimers;
     }
 

runners/direct-java/src/main/java/org/apache/beam/runners/direct/portable/BundleFactory.java

@@ -17,7 +17,7 @@
  */
 package org.apache.beam.runners.direct.portable;
 
-import org.apache.beam.runners.direct.portable.DirectGroupByKey.DirectGroupByKeyOnly;
+import org.apache.beam.runners.core.construction.graph.PipelineNode.PCollectionNode;
 import org.apache.beam.runners.local.StructuralKey;
 import org.apache.beam.sdk.transforms.PTransform;
 import org.apache.beam.sdk.values.PCollection;
@@ -38,13 +38,12 @@ interface BundleFactory {
    * Create an {@link UncommittedBundle} from the specified input. Elements added to the bundle
    * belong to the {@code output} {@link PCollection}.
    */
-  <T> UncommittedBundle<T> createBundle(PCollection<T> output);
+  <T> UncommittedBundle<T> createBundle(PCollectionNode output);
 
   /**
    * Create an {@link UncommittedBundle} with the specified keys at the specified step. For use by
-   * {@link DirectGroupByKeyOnly} {@link PTransform PTransforms}. Elements added to the bundle
+   * {@code DirectGroupByKeyOnly} {@link PTransform PTransforms}. Elements added to the bundle
    * belong to the {@code output} {@link PCollection}.
    */
-  <K, T> UncommittedBundle<T> createKeyedBundle(
-      StructuralKey<K> key, PCollection<T> output);
+  <K, T> UncommittedBundle<T> createKeyedBundle(StructuralKey<K> key, PCollectionNode output);
 }

runners/direct-java/src/main/java/org/apache/beam/runners/direct/portable/CommittedBundle.java

@@ -19,6 +19,7 @@
 package org.apache.beam.runners.direct.portable;
 
 import javax.annotation.Nullable;
+import org.apache.beam.runners.core.construction.graph.PipelineNode.PCollectionNode;
 import org.apache.beam.runners.local.Bundle;
 import org.apache.beam.runners.local.StructuralKey;
 import org.apache.beam.sdk.transforms.PTransform;
@@ -33,12 +34,12 @@
  * a part of at a later point.
  * @param <T> the type of elements contained within this bundle
  */
-interface CommittedBundle<T> extends Bundle<T, PCollection<T>> {
+interface CommittedBundle<T> extends Bundle<T, PCollectionNode> {
   /**
    * Returns the PCollection that the elements of this bundle belong to.
    */
   @Nullable
-  PCollection<T> getPCollection();
+  PCollectionNode getPCollection();
 
   /**
    * Returns the key that was output in the most recent {@code GroupByKey} in the

runners/direct-java/src/main/java/org/apache/beam/runners/direct/portable/CommittedResult.java

@@ -21,7 +21,7 @@
 import com.google.auto.value.AutoValue;
 import com.google.common.base.Optional;
 import java.util.Set;
-import org.apache.beam.sdk.runners.AppliedPTransform;
+import org.apache.beam.runners.core.construction.graph.PipelineNode.PTransformNode;
 import org.apache.beam.sdk.transforms.View.CreatePCollectionView;
 
 /**
@@ -30,7 +30,7 @@
 @AutoValue
 abstract class CommittedResult<ExecutableT> {
   /**
-   * Returns the {@link AppliedPTransform} that produced this result.
+   * Returns the {@link PTransformNode} that produced this result.
    */
   public abstract ExecutableT getExecutable();
 
@@ -40,7 +40,6 @@ abstract class CommittedResult<ExecutableT> {
    * input elements, and absent otherwise.
    */
   public abstract Optional<? extends CommittedBundle<?>> getUnprocessedInputs();
-
   /**
    * Returns the outputs produced by the transform.
    */
@@ -55,7 +54,7 @@ abstract class CommittedResult<ExecutableT> {
    */
   public abstract Set<OutputType> getProducedOutputTypes();
 
-  public static CommittedResult<AppliedPTransform<?, ?, ?>> create(
+  public static CommittedResult<PTransformNode> create(
       TransformResult<?> original,
       Optional<? extends CommittedBundle<?>> unprocessedElements,
       Iterable<? extends CommittedBundle<?>> outputs,

runners/direct-java/src/main/java/org/apache/beam/runners/direct/portable/CompletionCallback.java

@@ -17,7 +17,7 @@
  */
 package org.apache.beam.runners.direct.portable;
 
-import org.apache.beam.sdk.runners.AppliedPTransform;
+import org.apache.beam.runners.core.construction.graph.PipelineNode.PTransformNode;
 
 /**
  * A callback for completing a bundle of input.
@@ -34,7 +34,7 @@ CommittedResult handleResult(
    *
    * <p>This occurs when a Source has no splits that can currently produce outputs.
    */
-  void handleEmpty(AppliedPTransform<?, ?, ?> transform);
+  void handleEmpty(PTransformNode transform);
 
   /**
    * Handle a result that terminated abnormally due to the provided {@link Exception}.

runners/direct-java/src/main/java/org/apache/beam/runners/direct/portable/DirectExecutionContext.java

@@ -22,8 +22,9 @@
 import java.util.Map;
 import org.apache.beam.runners.core.StepContext;
 import org.apache.beam.runners.core.TimerInternals;
-import org.apache.beam.runners.direct.portable.WatermarkManager.TimerUpdate;
-import org.apache.beam.runners.direct.portable.WatermarkManager.TransformWatermarks;
+import org.apache.beam.runners.direct.Clock;
+import org.apache.beam.runners.direct.WatermarkManager.TimerUpdate;
+import org.apache.beam.runners.direct.WatermarkManager.TransformWatermarks;
 import org.apache.beam.runners.local.StructuralKey;
 
 /**

runners/direct-java/src/main/java/org/apache/beam/runners/direct/portable/DirectTimerInternals.java

@@ -20,9 +20,10 @@
 import javax.annotation.Nullable;
 import org.apache.beam.runners.core.StateNamespace;
 import org.apache.beam.runners.core.TimerInternals;
-import org.apache.beam.runners.direct.portable.WatermarkManager.TimerUpdate;
-import org.apache.beam.runners.direct.portable.WatermarkManager.TimerUpdate.TimerUpdateBuilder;
-import org.apache.beam.runners.direct.portable.WatermarkManager.TransformWatermarks;
+import org.apache.beam.runners.direct.Clock;
+import org.apache.beam.runners.direct.WatermarkManager.TimerUpdate;
+import org.apache.beam.runners.direct.WatermarkManager.TimerUpdate.TimerUpdateBuilder;
+import org.apache.beam.runners.direct.WatermarkManager.TransformWatermarks;
 import org.apache.beam.sdk.state.TimeDomain;
 import org.joda.time.Instant;
 

runners/direct-java/src/main/java/org/apache/beam/runners/direct/portable/DirectTransformExecutor.java

@@ -20,10 +20,10 @@
 import com.google.common.annotations.VisibleForTesting;
 import java.io.Closeable;
 import java.util.concurrent.Callable;
+import org.apache.beam.runners.core.construction.graph.PipelineNode.PTransformNode;
 import org.apache.beam.runners.core.metrics.MetricUpdates;
 import org.apache.beam.runners.core.metrics.MetricsContainerImpl;
 import org.apache.beam.sdk.metrics.MetricsEnvironment;
-import org.apache.beam.sdk.runners.AppliedPTransform;
 import org.apache.beam.sdk.util.WindowedValue;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -48,7 +48,7 @@ static class Factory implements TransformExecutorFactory {
     @Override
     public TransformExecutor create(
         CommittedBundle<?> bundle,
-        AppliedPTransform<?, ?, ?> transform,
+        PTransformNode transform,
         CompletionCallback onComplete,
         TransformExecutorService executorService) {
       return new DirectTransformExecutor<>(
@@ -59,7 +59,7 @@ public TransformExecutor create(
   private final TransformEvaluatorRegistry evaluatorRegistry;
 
   /** The transform that will be evaluated. */
-  private final AppliedPTransform<?, ?, ?> transform;
+  private final PTransformNode transform;
   /** The inputs this {@link DirectTransformExecutor} will deliver to the transform. */
   private final CommittedBundle<T> inputBundle;
 
@@ -72,7 +72,7 @@ public TransformExecutor create(
       EvaluationContext context,
       TransformEvaluatorRegistry factory,
       CommittedBundle<T> inputBundle,
-      AppliedPTransform<?, ?, ?> transform,
+      PTransformNode transform,
       CompletionCallback completionCallback,
       TransformExecutorService transformEvaluationState) {
     this.evaluatorRegistry = factory;
@@ -88,7 +88,7 @@ public TransformExecutor create(
 
   @Override
   public void run() {
-    MetricsContainerImpl metricsContainer = new MetricsContainerImpl(transform.getFullName());
+    MetricsContainerImpl metricsContainer = new MetricsContainerImpl(transform.getId());
     try (Closeable metricsScope = MetricsEnvironment.scopedMetricsContainer(metricsContainer)) {
       TransformEvaluator<T> evaluator =
           evaluatorRegistry.forApplication(transform, inputBundle);