Merge 1c42641 into 6807480

apache · Dec 8, 2016 · 8f56e3d · 8f56e3d
2 parents 6807480 + 1c42641
commit 8f56e3d
Show file tree

Hide file tree

Showing 11 changed files with 66 additions and 33 deletions.
diff --git a/...e/CopyOnAccessInMemoryStateInternals.java → ...t/CopyOnAccessInMemoryStateInternals.java b/...e/CopyOnAccessInMemoryStateInternals.java → ...t/CopyOnAccessInMemoryStateInternals.java
@@ -15,7 +15,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.apache.beam.sdk.util.state;
+package org.apache.beam.runners.direct;
 
 import static com.google.common.base.Preconditions.checkState;
 
@@ -32,8 +32,24 @@
 import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
 import org.apache.beam.sdk.transforms.windowing.OutputTimeFn;
 import org.apache.beam.sdk.util.CombineFnUtil;
+import org.apache.beam.sdk.util.state.AccumulatorCombiningState;
+import org.apache.beam.sdk.util.state.BagState;
+import org.apache.beam.sdk.util.state.InMemoryStateInternals.InMemoryBag;
+import org.apache.beam.sdk.util.state.InMemoryStateInternals.InMemoryCombiningValue;
 import org.apache.beam.sdk.util.state.InMemoryStateInternals.InMemoryState;
+import org.apache.beam.sdk.util.state.InMemoryStateInternals.InMemoryStateBinder;
+import org.apache.beam.sdk.util.state.InMemoryStateInternals.InMemoryValue;
+import org.apache.beam.sdk.util.state.InMemoryStateInternals.InMemoryWatermarkHold;
+import org.apache.beam.sdk.util.state.State;
+import org.apache.beam.sdk.util.state.StateContext;
+import org.apache.beam.sdk.util.state.StateContexts;
+import org.apache.beam.sdk.util.state.StateInternals;
+import org.apache.beam.sdk.util.state.StateNamespace;
+import org.apache.beam.sdk.util.state.StateTable;
+import org.apache.beam.sdk.util.state.StateTag;
 import org.apache.beam.sdk.util.state.StateTag.StateBinder;
+import org.apache.beam.sdk.util.state.ValueState;
+import org.apache.beam.sdk.util.state.WatermarkHoldState;
 import org.joda.time.Instant;
 
 /**
@@ -262,11 +278,11 @@ public <W extends BoundedWindow> WatermarkHoldState<W> bindWatermark(
             if (containedInUnderlying(namespace, address)) {
               @SuppressWarnings("unchecked")
               InMemoryState<? extends WatermarkHoldState<W>> existingState =
-                  (InMemoryStateInternals.InMemoryState<? extends WatermarkHoldState<W>>)
+                  (InMemoryState<? extends WatermarkHoldState<W>>)
                   underlying.get().get(namespace, address, c);
               return existingState.copy();
             } else {
-              return new InMemoryStateInternals.InMemoryWatermarkHold<>(
+              return new InMemoryWatermarkHold<>(
                   outputTimeFn);
             }
           }
@@ -277,11 +293,11 @@ public <T> ValueState<T> bindValue(
             if (containedInUnderlying(namespace, address)) {
               @SuppressWarnings("unchecked")
               InMemoryState<? extends ValueState<T>> existingState =
-                  (InMemoryStateInternals.InMemoryState<? extends ValueState<T>>)
+                  (InMemoryState<? extends ValueState<T>>)
                   underlying.get().get(namespace, address, c);
               return existingState.copy();
             } else {
-              return new InMemoryStateInternals.InMemoryValue<>();
+              return new InMemoryValue<>();
             }
           }
 
@@ -294,12 +310,11 @@ public <T> ValueState<T> bindValue(
               @SuppressWarnings("unchecked")
               InMemoryState<? extends AccumulatorCombiningState<InputT, AccumT, OutputT>>
                   existingState = (
-                      InMemoryStateInternals
-                          .InMemoryState<? extends AccumulatorCombiningState<InputT, AccumT,
-                          OutputT>>) underlying.get().get(namespace, address, c);
+                  InMemoryState<? extends AccumulatorCombiningState<InputT, AccumT,
+                                            OutputT>>) underlying.get().get(namespace, address, c);
               return existingState.copy();
             } else {
-              return new InMemoryStateInternals.InMemoryCombiningValue<>(
+              return new InMemoryCombiningValue<>(
                   key, combineFn.asKeyedFn());
             }
           }
@@ -310,11 +325,11 @@ public <T> BagState<T> bindBag(
             if (containedInUnderlying(namespace, address)) {
               @SuppressWarnings("unchecked")
               InMemoryState<? extends BagState<T>> existingState =
-                  (InMemoryStateInternals.InMemoryState<? extends BagState<T>>)
+                  (InMemoryState<? extends BagState<T>>)
                   underlying.get().get(namespace, address, c);
               return existingState.copy();
             } else {
-              return new InMemoryStateInternals.InMemoryBag<>();
+              return new InMemoryBag<>();
             }
           }
 
@@ -328,12 +343,11 @@ public <T> BagState<T> bindBag(
               @SuppressWarnings("unchecked")
               InMemoryState<? extends AccumulatorCombiningState<InputT, AccumT, OutputT>>
                   existingState = (
-                      InMemoryStateInternals
-                          .InMemoryState<? extends AccumulatorCombiningState<InputT, AccumT,
-                          OutputT>>) underlying.get().get(namespace, address, c);
+                  InMemoryState<? extends AccumulatorCombiningState<InputT, AccumT,
+                                            OutputT>>) underlying.get().get(namespace, address, c);
               return existingState.copy();
             } else {
-              return new InMemoryStateInternals.InMemoryCombiningValue<>(key, combineFn);
+              return new InMemoryCombiningValue<>(key, combineFn);
             }
           }
 
@@ -446,7 +460,7 @@ public InMemoryStateBinderFactory(K key) {
 
       @Override
       public StateBinder<K> forNamespace(StateNamespace namespace, StateContext<?> c) {
-        return new InMemoryStateInternals.InMemoryStateBinder<>(key, c);
+        return new InMemoryStateBinder<>(key, c);
       }
     }
   }

diff --git a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/DirectExecutionContext.java b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/DirectExecutionContext.java
@@ -23,7 +23,6 @@
 import org.apache.beam.sdk.util.BaseExecutionContext;
 import org.apache.beam.sdk.util.ExecutionContext;
 import org.apache.beam.sdk.util.TimerInternals;
-import org.apache.beam.sdk.util.state.CopyOnAccessInMemoryStateInternals;
 
 /**
  * Execution Context for the {@link DirectRunner}.

diff --git a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/EvaluationContext.java b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/EvaluationContext.java
@@ -48,7 +48,6 @@
 import org.apache.beam.sdk.util.TimerInternals.TimerData;
 import org.apache.beam.sdk.util.WindowedValue;
 import org.apache.beam.sdk.util.WindowingStrategy;
-import org.apache.beam.sdk.util.state.CopyOnAccessInMemoryStateInternals;
 import org.apache.beam.sdk.values.PCollection;
 import org.apache.beam.sdk.values.PCollection.IsBounded;
 import org.apache.beam.sdk.values.PCollectionView;

diff --git a/...-java/src/main/java/org/apache/beam/runners/direct/GroupAlsoByWindowEvaluatorFactory.java b/...-java/src/main/java/org/apache/beam/runners/direct/GroupAlsoByWindowEvaluatorFactory.java
@@ -50,7 +50,6 @@
 import org.apache.beam.sdk.util.WindowTracing;
 import org.apache.beam.sdk.util.WindowedValue;
 import org.apache.beam.sdk.util.WindowingStrategy;
-import org.apache.beam.sdk.util.state.CopyOnAccessInMemoryStateInternals;
 import org.apache.beam.sdk.values.KV;
 import org.apache.beam.sdk.values.PCollection;
 import org.apache.beam.sdk.values.PCollectionView;

diff --git a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/ParDoEvaluator.java b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/ParDoEvaluator.java
@@ -35,7 +35,6 @@
 import org.apache.beam.sdk.util.UserCodeException;
 import org.apache.beam.sdk.util.WindowedValue;
 import org.apache.beam.sdk.util.WindowingStrategy;
-import org.apache.beam.sdk.util.state.CopyOnAccessInMemoryStateInternals;
 import org.apache.beam.sdk.values.PCollection;
 import org.apache.beam.sdk.values.PCollectionView;
 import org.apache.beam.sdk.values.TupleTag;

diff --git a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/StepTransformResult.java b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/StepTransformResult.java
@@ -30,7 +30,6 @@
 import org.apache.beam.sdk.transforms.AppliedPTransform;
 import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
 import org.apache.beam.sdk.util.WindowedValue;
-import org.apache.beam.sdk.util.state.CopyOnAccessInMemoryStateInternals;
 import org.joda.time.Instant;
 
 /**

diff --git a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/TransformResult.java b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/TransformResult.java
@@ -28,7 +28,6 @@
 import org.apache.beam.sdk.transforms.ParDo;
 import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
 import org.apache.beam.sdk.util.WindowedValue;
-import org.apache.beam.sdk.util.state.CopyOnAccessInMemoryStateInternals;
 import org.joda.time.Instant;
 
 /**

diff --git a/...pyOnAccessInMemoryStateInternalsTest.java → ...pyOnAccessInMemoryStateInternalsTest.java b/...pyOnAccessInMemoryStateInternalsTest.java → ...pyOnAccessInMemoryStateInternalsTest.java
@@ -15,7 +15,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.apache.beam.sdk.util.state;
+package org.apache.beam.runners.direct;
 
 import static org.hamcrest.Matchers.containsInAnyOrder;
 import static org.hamcrest.Matchers.emptyIterable;
@@ -39,6 +39,16 @@
 import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
 import org.apache.beam.sdk.transforms.windowing.OutputTimeFn;
 import org.apache.beam.sdk.transforms.windowing.OutputTimeFns;
+import org.apache.beam.sdk.util.state.AccumulatorCombiningState;
+import org.apache.beam.sdk.util.state.BagState;
+import org.apache.beam.sdk.util.state.CombiningState;
+import org.apache.beam.sdk.util.state.StateNamespace;
+import org.apache.beam.sdk.util.state.StateNamespaceForTest;
+import org.apache.beam.sdk.util.state.StateNamespaces;
+import org.apache.beam.sdk.util.state.StateTag;
+import org.apache.beam.sdk.util.state.StateTags;
+import org.apache.beam.sdk.util.state.ValueState;
+import org.apache.beam.sdk.util.state.WatermarkHoldState;
 import org.joda.time.Instant;
 import org.junit.Rule;
 import org.junit.Test;

diff --git a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/EvaluationContextTest.java b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/EvaluationContextTest.java
@@ -58,7 +58,6 @@
 import org.apache.beam.sdk.util.WindowedValue;
 import org.apache.beam.sdk.util.WindowingStrategy;
 import org.apache.beam.sdk.util.state.BagState;
-import org.apache.beam.sdk.util.state.CopyOnAccessInMemoryStateInternals;
 import org.apache.beam.sdk.util.state.StateNamespaces;
 import org.apache.beam.sdk.util.state.StateTag;
 import org.apache.beam.sdk.util.state.StateTags;

diff --git a/...-java/src/test/java/org/apache/beam/runners/direct/StatefulParDoEvaluatorFactoryTest.java b/...-java/src/test/java/org/apache/beam/runners/direct/StatefulParDoEvaluatorFactoryTest.java
@@ -51,7 +51,6 @@
 import org.apache.beam.sdk.util.ReadyCheckingSideInputReader;
 import org.apache.beam.sdk.util.WindowedValue;
 import org.apache.beam.sdk.util.WindowingStrategy;
-import org.apache.beam.sdk.util.state.CopyOnAccessInMemoryStateInternals;
 import org.apache.beam.sdk.util.state.StateInternals;
 import org.apache.beam.sdk.util.state.StateNamespace;
 import org.apache.beam.sdk.util.state.StateNamespaces;

diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/state/InMemoryStateInternals.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/util/state/InMemoryStateInternals.java
@@ -56,7 +56,11 @@ public K getKey() {
     return key;
   }
 
-  interface InMemoryState<T extends InMemoryState<T>> {
+  /**
+   * Interface common to all in-memory state cells. Includes ability to see whether a cell has been
+   * cleared and the ability to create a clone of the contents.
+   */
+  public interface InMemoryState<T extends InMemoryState<T>> {
     boolean isCleared();
     T copy();
   }
@@ -94,11 +98,11 @@ public <T extends State> T state(
   /**
    * A {@link StateBinder} that returns In Memory {@link State} objects.
    */
-  static class InMemoryStateBinder<K> implements StateBinder<K> {
+  public static class InMemoryStateBinder<K> implements StateBinder<K> {
     private final K key;
     private final StateContext<?> c;
 
-    InMemoryStateBinder(K key, StateContext<?> c) {
+    public InMemoryStateBinder(K key, StateContext<?> c) {
       this.key = key;
       this.c = c;
     }
@@ -150,7 +154,11 @@ public <W extends BoundedWindow> WatermarkHoldState<W> bindWatermark(
     }
   }
 
-  static final class InMemoryValue<T> implements ValueState<T>, InMemoryState<InMemoryValue<T>> {
+  /**
+   * An {@link InMemoryState} implementation of {@link ValueState}.
+   */
+  public static final class InMemoryValue<T>
+      implements ValueState<T>, InMemoryState<InMemoryValue<T>> {
     private boolean isCleared = true;
     private T value = null;
 
@@ -194,7 +202,10 @@ public boolean isCleared() {
     }
   }
 
-  static final class InMemoryWatermarkHold<W extends BoundedWindow>
+  /**
+   * An {@link InMemoryState} implementation of {@link WatermarkHoldState}.
+   */
+  public static final class InMemoryWatermarkHold<W extends BoundedWindow>
       implements WatermarkHoldState<W>, InMemoryState<InMemoryWatermarkHold<W>> {
 
     private final OutputTimeFn<? super W> outputTimeFn;
@@ -267,15 +278,18 @@ public InMemoryWatermarkHold<W> copy() {
     }
   }
 
-  static final class InMemoryCombiningValue<K, InputT, AccumT, OutputT>
+  /**
+   * An {@link InMemoryState} implementation of {@link AccumulatorCombiningState}.
+   */
+  public static final class InMemoryCombiningValue<K, InputT, AccumT, OutputT>
       implements AccumulatorCombiningState<InputT, AccumT, OutputT>,
           InMemoryState<InMemoryCombiningValue<K, InputT, AccumT, OutputT>> {
     private final K key;
     private boolean isCleared = true;
     private final KeyedCombineFn<? super K, InputT, AccumT, OutputT> combineFn;
     private AccumT accum;
 
-    InMemoryCombiningValue(
+    public InMemoryCombiningValue(
         K key, KeyedCombineFn<? super K, InputT, AccumT, OutputT> combineFn) {
       this.key = key;
       this.combineFn = combineFn;
@@ -353,7 +367,10 @@ public InMemoryCombiningValue<K, InputT, AccumT, OutputT> copy() {
     }
   }
 
-  static final class InMemoryBag<T> implements BagState<T>, InMemoryState<InMemoryBag<T>> {
+  /**
+   * An {@link InMemoryState} implementation of {@link BagState}.
+   */
+  public static final class InMemoryBag<T> implements BagState<T>, InMemoryState<InMemoryBag<T>> {
     private List<T> contents = new ArrayList<>();
 
     @Override