apache#51 Do _not_ store storage-provider in states by default

Novotnik, Petr · David Moravek · commit cc5102155ba2 · 2018-08-17T11:58:30.000+02:00
diff --git a/sdks/java/extensions/euphoria/euphoria-core/src/main/java/cz/seznam/euphoria/core/client/operator/Join.java b/sdks/java/extensions/euphoria/euphoria-core/src/main/java/cz/seznam/euphoria/core/client/operator/Join.java
@@ -318,7 +318,7 @@ private class JoinState
 
     @SuppressWarnings("unchecked")
     public JoinState(Context<OUT> context, StorageProvider storageProvider) {
-      super(context, storageProvider);
+      super(context);
       leftElements = storageProvider.getListStorage(LEFT_STATE_DESCR);
       rightElements = storageProvider.getListStorage(RIGHT_STATE_DESCR);
     }
diff --git a/sdks/java/extensions/euphoria/euphoria-core/src/main/java/cz/seznam/euphoria/core/client/operator/ReduceByKey.java b/sdks/java/extensions/euphoria/euphoria-core/src/main/java/cz/seznam/euphoria/core/client/operator/ReduceByKey.java
@@ -310,7 +310,7 @@ public State<E, E> apply(Context<E> ctx, StorageProvider storageProvider) {
     CombiningReduceState(Context<E> context,
                          StorageProvider storageProvider,
                          CombinableReduceFunction<E> reducer) {
-      super(context, storageProvider);
+      super(context);
       this.reducer = Objects.requireNonNull(reducer);
 
       @SuppressWarnings("unchecked")
@@ -373,7 +373,7 @@ static final class Factory<VALUE, OUT>
     NonCombiningReduceState(Context<OUT> context,
                             StorageProvider storageProvider,
                             ReduceFunction<VALUE, OUT> reducer) {
-      super(context, storageProvider);
+      super(context);
       this.reducer = Objects.requireNonNull(reducer);
 
       @SuppressWarnings("unchecked")
diff --git a/sdks/java/extensions/euphoria/euphoria-core/src/main/java/cz/seznam/euphoria/core/client/operator/TopPerKey.java b/sdks/java/extensions/euphoria/euphoria-core/src/main/java/cz/seznam/euphoria/core/client/operator/TopPerKey.java
@@ -57,7 +57,7 @@ private static final class MaxScored<V, C extends Comparable<C>>
 
     @SuppressWarnings("unchecked")
     MaxScored(Context<Pair<V, C>> context, StorageProvider storageProvider) {
-      super(context, storageProvider);
+      super(context);
       curr = (ValueStorage) storageProvider.getValueStorage(MAX_STATE_DESCR);
     }
 
diff --git a/sdks/java/extensions/euphoria/euphoria-core/src/main/java/cz/seznam/euphoria/core/client/operator/state/State.java b/sdks/java/extensions/euphoria/euphoria-core/src/main/java/cz/seznam/euphoria/core/client/operator/state/State.java
@@ -26,8 +26,6 @@ public abstract class State<IN, OUT> implements Closeable {
 
   /** Collector of output of this state. */
   private final Context<OUT> context;
-  /** Provider of state storage. */
-  private final StorageProvider storageProvider;
 
   /**
    * Add element to this state.
@@ -42,22 +40,15 @@ public abstract class State<IN, OUT> implements Closeable {
    */
   public abstract void flush();
 
-  protected State(
-      Context<OUT> context,
-      StorageProvider storageProvider) {
+  protected State(Context<OUT> context) {
     
     this.context = context;
-    this.storageProvider = storageProvider;
   }
 
   public Context<OUT> getContext() {
     return context;
   }
 
-  public StorageProvider getStorageProvider() {
-    return storageProvider;
-  }
-
   /**
    * Closes this state. Invoked after {@link #flush()} and before
    * this state gets disposed.
diff --git a/sdks/java/extensions/euphoria/euphoria-core/src/test/java/cz/seznam/euphoria/core/client/operator/ReduceStateByKeyTest.java b/sdks/java/extensions/euphoria/euphoria-core/src/test/java/cz/seznam/euphoria/core/client/operator/ReduceStateByKeyTest.java
@@ -29,6 +29,7 @@
 import org.junit.Test;
 
 import java.time.Duration;
+import java.util.Iterator;
 
 import static org.junit.Assert.*;
 
@@ -155,7 +156,7 @@ private static class WordCountState extends State<Long, Long> {
     protected WordCountState(
         Context<Long> context,
         StorageProvider storageProvider) {
-      super(context, storageProvider);
+      super(context);
       sum = storageProvider.getValueStorage(ValueStorageDescriptor.of(
           "sum", Long.class, 0L));
     }
@@ -171,17 +172,12 @@ public void flush() {
     }
 
     static WordCountState combine(Iterable<WordCountState> others) {
-      WordCountState state = null;
-      for (WordCountState s : others) {
-        if (state == null) {
-          state = new WordCountState(
-              s.getContext(),
-              s.getStorageProvider());
-        }
-        state.add(s.sum.get());
+      Iterator<WordCountState> iter = others.iterator();
+      WordCountState target = iter.next();
+      while (iter.hasNext()) {
+        target.add(iter.next().sum.get());
       }
-
-      return state;
+      return target;
     }
 
     @Override
diff --git a/sdks/java/extensions/euphoria/euphoria-flink/src/test/java/cz/seznam/euphoria/flink/streaming/RSBKWindowingTest.java b/sdks/java/extensions/euphoria/euphoria-flink/src/test/java/cz/seznam/euphoria/flink/streaming/RSBKWindowingTest.java
@@ -51,7 +51,7 @@ private static class AccState<VALUE> extends State<VALUE, VALUE> {
     AccState(Context<VALUE> context,
              StorageProvider storageProvider)
     {
-      super(context, storageProvider);
+      super(context);
       reducableValues = storageProvider.getListStorage(
           ListStorageDescriptor.of("vals", (Class) Object.class));
     }
diff --git a/sdks/java/extensions/euphoria/euphoria-inmem/src/test/java/cz/seznam/euphoria/inmem/InMemExecutorTest.java b/sdks/java/extensions/euphoria/euphoria-inmem/src/test/java/cz/seznam/euphoria/inmem/InMemExecutorTest.java
@@ -58,6 +58,7 @@
 import java.util.Collections;
 import java.util.Comparator;
 import java.util.HashMap;
+import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
@@ -276,7 +277,7 @@ public static class SortState extends State<Integer, Integer> {
         Context<Integer> c,
         StorageProvider storageProvider) {
       
-      super(c, storageProvider);
+      super(c);
       data = storageProvider.getListStorage(
           ListStorageDescriptor.of("data", Integer.class));
     }
@@ -297,16 +298,12 @@ public void flush() {
     }
 
     static SortState combine(Iterable<SortState> others) {
-      SortState ret = null;
-      for (SortState s : others) {
-        if (ret == null) {
-          ret = new SortState(
-              s.getContext(),
-              s.getStorageProvider());
-        }
-        ret.data.addAll(s.data.get());
+      Iterator<SortState> iter = others.iterator();
+      SortState target = iter.next();
+      while (iter.hasNext()) {
+        target.data.addAll(iter.next().data.get());
       }
-      return ret;
+      return target;
     }
 
     @Override
diff --git a/sdks/java/extensions/euphoria/euphoria-operator-testkit/src/main/java/cz/seznam/euphoria/operator/test/ReduceByKeyTest.java b/sdks/java/extensions/euphoria/euphoria-operator-testkit/src/main/java/cz/seznam/euphoria/operator/test/ReduceByKeyTest.java
@@ -56,6 +56,7 @@
 import java.util.Comparator;
 import java.util.HashMap;
 import java.util.HashSet;
+import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
 import java.util.stream.Collectors;
@@ -539,7 +540,7 @@ static class SumState extends State<Integer, Integer> {
     private final ValueStorage<Integer> sum;
 
     SumState(Context<Integer> context, StorageProvider storageProvider) {
-      super(context, storageProvider);
+      super(context);
       sum = storageProvider.getValueStorage(
           ValueStorageDescriptor.of("sum-state", Integer.class, 0));
     }
@@ -560,12 +561,10 @@ public void close() {
     }
 
     static SumState combine(Iterable<SumState> states) {
-      SumState target = null;
-      for (SumState state : states) {
-        if (target == null) {
-          target = new SumState(state.getContext(), state.getStorageProvider());
-        }
-        target.add(state.sum.get());
+      Iterator<SumState> iter = states.iterator();
+      SumState target = iter.next();
+      while (iter.hasNext()) {
+        target.add(iter.next().sum.get());
       }
       return target;
     }
diff --git a/sdks/java/extensions/euphoria/euphoria-operator-testkit/src/main/java/cz/seznam/euphoria/operator/test/ReduceStateByKeyTest.java b/sdks/java/extensions/euphoria/euphoria-operator-testkit/src/main/java/cz/seznam/euphoria/operator/test/ReduceStateByKeyTest.java
@@ -83,11 +83,8 @@ static class SortState extends State<Integer, Integer> {
 
     final ListStorage<Integer> data;
 
-    SortState(        
-        Context<Integer> c,
-        StorageProvider storageProvider) {
-
-      super(c, storageProvider);
+    SortState(Context<Integer> c, StorageProvider storageProvider) {
+      super(c);
       this.data = storageProvider.getListStorage(
           ListStorageDescriptor.of("data", Integer.class));
     }
@@ -112,16 +109,12 @@ public void close() {
     }
 
     static SortState combine(Iterable<SortState> states) {
-      SortState ret = null;
-      for (SortState state : states) {
-        if (ret == null) {
-          ret = new SortState(
-              state.getContext(),
-              state.getStorageProvider());
-        }
-        ret.data.addAll(state.data.get());
+      Iterator<SortState> iter = states.iterator();
+      SortState target = iter.next();
+      while (iter.hasNext()) {
+        target.data.addAll(iter.next().data.get());
       }
-      return ret;
+      return target;
     }
 
   }
@@ -249,7 +242,7 @@ private static class CountState extends State<String, Long> {
     final ValueStorage<Long> count;
     CountState(Context<Long> context, StorageProvider storageProvider)
     {
-      super(context, storageProvider);
+      super(context);
       this.count = storageProvider.getValueStorage(
           ValueStorageDescriptor.of("count-state", Long.class, 0L));
     }
@@ -351,7 +344,7 @@ private static class AccState<VALUE> extends State<VALUE, VALUE> {
     AccState(Context<VALUE> context,
              StorageProvider storageProvider)
     {
-      super(context, storageProvider);
+      super(context);
       vals = storageProvider.getListStorage(
           ListStorageDescriptor.of("vals", (Class) Object.class));
     }
diff --git a/sdks/java/extensions/euphoria/euphoria-operator-testkit/src/main/java/cz/seznam/euphoria/operator/test/WindowingTest.java b/sdks/java/extensions/euphoria/euphoria-operator-testkit/src/main/java/cz/seznam/euphoria/operator/test/WindowingTest.java
@@ -208,7 +208,7 @@ private static class DistinctState extends State<Object, Object> {
     private final ValueStorage<Object> storage;
 
     public DistinctState(Context<Object> context, StorageProvider storageProvider) {
-      super(context, storageProvider);
+      super(context);
       this.storage = storageProvider.getValueStorage(
               ValueStorageDescriptor.of("element", Object.class, null));
     }

Original file line number	Diff line number	Diff line change
`@@ -318,7 +318,7 @@ private class JoinState`
`318`	`318`
`319`	`319`	`@SuppressWarnings("unchecked")`
`320`	`320`	`public JoinState(Context<OUT> context, StorageProvider storageProvider) {`
`321`		`- super(context, storageProvider);`
	`321`	`+ super(context);`
`322`	`322`	`leftElements = storageProvider.getListStorage(LEFT_STATE_DESCR);`
`323`	`323`	`rightElements = storageProvider.getListStorage(RIGHT_STATE_DESCR);`
`324`	`324`	`}`
Original file line number	Diff line number	Diff line change
`@@ -57,7 +57,7 @@ private static final class MaxScored<V, C extends Comparable<C>>`
`57`	`57`
`58`	`58`	`@SuppressWarnings("unchecked")`
`59`	`59`	`MaxScored(Context<Pair<V, C>> context, StorageProvider storageProvider) {`
`60`		`- super(context, storageProvider);`
	`60`	`+ super(context);`
`61`	`61`	`curr = (ValueStorage) storageProvider.getValueStorage(MAX_STATE_DESCR);`
`62`	`62`	`}`
`63`	`63`
Original file line number	Diff line number	Diff line change
`@@ -51,7 +51,7 @@ private static class AccState<VALUE> extends State<VALUE, VALUE> {`
`51`	`51`	`AccState(Context<VALUE> context,`
`52`	`52`	`StorageProvider storageProvider)`
`53`	`53`	`{`
`54`		`- super(context, storageProvider);`
	`54`	`+ super(context);`
`55`	`55`	`reducableValues = storageProvider.getListStorage(`
`56`	`56`	`ListStorageDescriptor.of("vals", (Class) Object.class));`
`57`	`57`	`}`
Original file line number	Diff line number	Diff line change
`@@ -83,11 +83,8 @@ static class SortState extends State<Integer, Integer> {`
`83`	`83`
`84`	`84`	`final ListStorage<Integer> data;`
`85`	`85`
`86`		`- SortState(`
`87`		`- Context<Integer> c,`
`88`		`- StorageProvider storageProvider) {`
`89`		`-`
`90`		`- super(c, storageProvider);`
	`86`	`+ SortState(Context<Integer> c, StorageProvider storageProvider) {`
	`87`	`+ super(c);`
`91`	`88`	`this.data = storageProvider.getListStorage(`
`92`	`89`	`ListStorageDescriptor.of("data", Integer.class));`
`93`	`90`	`}`
`@@ -112,16 +109,12 @@ public void close() {`
`112`	`109`	`}`
`113`	`110`
`114`	`111`	`static SortState combine(Iterable<SortState> states) {`
`115`		`- SortState ret = null;`
`116`		`- for (SortState state : states) {`
`117`		`- if (ret == null) {`
`118`		`- ret = new SortState(`
`119`		`- state.getContext(),`
`120`		`- state.getStorageProvider());`
`121`		`- }`
`122`		`- ret.data.addAll(state.data.get());`
	`112`	`+ Iterator<SortState> iter = states.iterator();`
	`113`	`+ SortState target = iter.next();`
	`114`	`+ while (iter.hasNext()) {`
	`115`	`+ target.data.addAll(iter.next().data.get());`
`123`	`116`	`}`
`124`		`- return ret;`
	`117`	`+ return target;`
`125`	`118`	`}`
`126`	`119`
`127`	`120`	`}`
`@@ -249,7 +242,7 @@ private static class CountState extends State<String, Long> {`
`249`	`242`	`final ValueStorage<Long> count;`
`250`	`243`	`CountState(Context<Long> context, StorageProvider storageProvider)`
`251`	`244`	`{`
`252`		`- super(context, storageProvider);`
	`245`	`+ super(context);`
`253`	`246`	`this.count = storageProvider.getValueStorage(`
`254`	`247`	`ValueStorageDescriptor.of("count-state", Long.class, 0L));`
`255`	`248`	`}`
`@@ -351,7 +344,7 @@ private static class AccState<VALUE> extends State<VALUE, VALUE> {`
`351`	`344`	`AccState(Context<VALUE> context,`
`352`	`345`	`StorageProvider storageProvider)`
`353`	`346`	`{`
`354`		`- super(context, storageProvider);`
	`347`	`+ super(context);`
`355`	`348`	`vals = storageProvider.getListStorage(`
`356`	`349`	`ListStorageDescriptor.of("vals", (Class) Object.class));`
`357`	`350`	`}`
Original file line number	Diff line number	Diff line change
`@@ -208,7 +208,7 @@ private static class DistinctState extends State<Object, Object> {`
`208`	`208`	`private final ValueStorage<Object> storage;`
`209`	`209`
`210`	`210`	`public DistinctState(Context<Object> context, StorageProvider storageProvider) {`
`211`		`- super(context, storageProvider);`
	`211`	`+ super(context);`
`212`	`212`	`this.storage = storageProvider.getValueStorage(`
`213`	`213`	`ValueStorageDescriptor.of("element", Object.class, null));`
`214`	`214`	`}`