Fix flink distributed

pom.xml

@@ -1058,7 +1058,7 @@
 
                         <!--RAT files-->
                         <exclude>**/apache-rat-0.13/**</exclude>
-                      
+
                         <!--Javadoc files-->
                         <exclude>**/docs/**</exclude>
 

wayang-commons/wayang-core/src/main/java/org/apache/wayang/core/plan/wayangplan/Operator.java

@@ -26,6 +26,8 @@
 import org.apache.wayang.core.optimizer.cardinality.DefaultCardinalityPusher;
 import org.apache.wayang.core.platform.Platform;
 
+import java.io.Serializable;
+
 import java.lang.reflect.Field;
 import java.util.Arrays;
 import java.util.Collection;
@@ -57,7 +59,7 @@
  * provided <i>before</i> the regular data.</li>
  * </ol>
  */
-public interface Operator {
+public interface Operator extends Serializable {
 
     /**
      * @return the number of {@link InputSlot}s of this instance; inclusive of broadcast {@link InputSlot}s
@@ -463,6 +465,13 @@ default boolean isElementary() {
         return true;
     }
 
+    /**
+     * @return whether this is a conversion operator
+     */
+    default boolean isConversion() {
+        return false;
+    }
+
     /**
      * This method is part of the visitor pattern and calls the appropriate visit method on {@code visitor}.
      */
@@ -641,5 +650,15 @@ default Collection<String> getEstimationContextProperties() {
         return properties;
     }
 
+    /*
+     * Collects all in and outputs of this operator instance and connects
+     * the operator given as parameter to them, effectively rendering this
+     * instance useless
+     *
+     * @param operator the operator to replace this one with
+     */
+    default void replaceWith(Operator operator) {
+        InputSlot.stealConnections(this, operator);
+        OutputSlot.stealConnections(this, operator);
+    }
 }
-

wayang-platforms/wayang-flink/pom.xml

@@ -49,6 +49,11 @@
             <artifactId>wayang-basic</artifactId>
             <version>1.0.1-SNAPSHOT</version>
         </dependency>
+        <dependency>
+            <groupId>org.apache.wayang</groupId>
+            <artifactId>wayang-api-sql</artifactId>
+            <version>1.0.1-SNAPSHOT</version>
+        </dependency>
         <dependency>
             <groupId>org.apache.wayang</groupId>
             <artifactId>wayang-java</artifactId>
@@ -62,6 +67,11 @@
             <groupId>org.apache.hadoop</groupId>
             <artifactId>hadoop-hdfs</artifactId>
         </dependency>
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-core</artifactId>
+            <version>${flink.version}</version>
+        </dependency>
         <!-- depencies of flink -->
         <dependency>
             <groupId>org.apache.flink</groupId>
@@ -111,5 +121,15 @@
             <artifactId>commons-math3</artifactId>
             <version>3.4.1</version>
         </dependency>
+        <dependency>
+            <groupId>com.esotericsoftware</groupId>
+            <artifactId>kryo</artifactId>
+            <version>5.6.2</version>
+        </dependency>
+        <dependency>
+            <groupId>io.altoo</groupId>
+            <artifactId>akka-kryo-serialization_2.12</artifactId>
+            <version>2.5.2</version>
+        </dependency>
     </dependencies>
 </project>

wayang-platforms/wayang-flink/src/main/java/org/apache/wayang/flink/compiler/FunctionCompiler.java

@@ -137,7 +137,6 @@ public <I, O> MapPartitionFunction<I, O> compile(MapPartitionsDescriptor<I, O> d
         return new MapPartitionFunction<I, O>() {
             @Override
             public void mapPartition(Iterable<I> iterable, Collector<O> collector) throws Exception {
-                System.out.println(collector.getClass());
                 Iterable<O> out = function.apply(iterable);
                 for(O element: out){
                     collector.collect(element);

wayang-platforms/wayang-flink/src/main/java/org/apache/wayang/flink/compiler/KeySelectorFunction.java

@@ -23,13 +23,12 @@
 import org.apache.flink.api.java.typeutils.ResultTypeQueryable;
 import org.apache.wayang.core.function.TransformationDescriptor;
 
-import java.io.Serializable;
 import java.util.function.Function;
 
 /**
- * Wrapper for {@Link KeySelector}
+ * Wrapper for {@link KeySelector}
  */
-public class KeySelectorFunction<T, K> implements KeySelector<T, K>, ResultTypeQueryable<K>, Serializable {
+public class KeySelectorFunction<T, K> implements KeySelector<T, K>, ResultTypeQueryable<K> {
 
     public Function<T, K> impl;
 

wayang-platforms/wayang-flink/src/main/java/org/apache/wayang/flink/execution/FlinkExecutor.java

@@ -35,6 +35,7 @@
 import org.apache.wayang.flink.compiler.FunctionCompiler;
 import org.apache.wayang.flink.operators.FlinkExecutionOperator;
 import org.apache.wayang.flink.platform.FlinkPlatform;
+import com.esotericsoftware.kryo.serializers.DefaultSerializers;
 
 import java.util.Arrays;
 import java.util.Collection;
@@ -76,7 +77,7 @@ public FlinkExecutor(FlinkPlatform flinkPlatform, Job job) {
         this.platform = flinkPlatform;
         this.flinkContextReference = this.platform.getFlinkContext(job);
         this.fee = this.flinkContextReference.get();
-        this.numDefaultPartitions = (int)this.getConfiguration().getLongProperty("wayang.flink.paralelism");
+        this.numDefaultPartitions = (int) this.getConfiguration().getLongProperty("wayang.flink.parallelism");
         this.fee.setParallelism(this.numDefaultPartitions);
         this.flinkContextReference.noteObtainedReference();
     }
@@ -133,7 +134,7 @@ protected Tuple<List<ChannelInstance>, PartialExecution> execute(
             }else {
                 try {
                     //TODO validate the execute in different contexts
-                    //this.fee.execute();
+                    this.fee.execute();
                 } catch (Exception e) {
                     throw new WayangException(e);
                 }

wayang-platforms/wayang-flink/src/main/java/org/apache/wayang/flink/operators/CollectionSplittableIterator.java

@@ -0,0 +1,74 @@
+/*
+ * 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.wayang.flink.operators;
+
+import org.apache.flink.util.SplittableIterator;
+import java.util.Iterator;
+import java.util.List;
+import java.io.Serializable;
+
+public class CollectionSplittableIterator<T> extends SplittableIterator<T> implements Serializable {
+    private int numSplits;
+    private int head;
+    private final List<T> collection;
+
+    public CollectionSplittableIterator(List<T> collection, int numSplits) {
+        this.collection = collection;
+        this.numSplits = numSplits;
+    }
+
+    @Override
+    public Iterator<T>[] split(int numSplits) {
+        // Split the collection into chunks
+        int chunkSize = (int) Math.ceil((double) numElements() / this.numSplits);
+        @SuppressWarnings("unchecked")
+        Iterator<T>[] splits = new Iterator[this.numSplits];
+
+
+        for (int i = 0; i < this.numSplits; i++) {
+            int fromIndex = i * chunkSize;
+            int toIndex = Math.min(fromIndex + chunkSize, numElements());
+            splits[i] = new CollectionSplittableIterator<>(this.collection.subList(fromIndex, toIndex), 1);
+        }
+
+        return splits;
+    }
+
+    @Override
+    public boolean hasNext() {
+        return this.head < this.collection.size() - 1;
+    }
+
+    @Override
+    public T next() {
+        T next = this.collection.get(this.head);
+        this.head++;
+
+        return next;
+    }
+
+    @Override
+    public int getMaximumNumberOfSplits() {
+        return this.numSplits;
+    }
+
+    private int numElements() {
+        return this.collection.size();
+    }
+}
+

wayang-platforms/wayang-flink/src/main/java/org/apache/wayang/flink/operators/FlinkCartesianOperator.java

@@ -81,7 +81,9 @@ public Tuple<Collection<ExecutionLineageNode>, Collection<ChannelInstance>> eval
                 (dataInput0, dataInput1) -> {
                     return new Tuple2<>(dataInput0, dataInput1);
                 }
-        ).returns(ReflectionUtils.specify(Tuple2.class));
+        )
+        .setParallelism(flinkExecutor.fee.getParallelism())
+        .returns(ReflectionUtils.specify(Tuple2.class));
 
         output.accept(datasetOutput, flinkExecutor);
 

wayang-platforms/wayang-flink/src/main/java/org/apache/wayang/flink/operators/FlinkCoGroupOperator.java

@@ -127,7 +127,9 @@ public void coGroup (
                     iterable1.forEach(list1::add);
                     collector.collect( new Tuple2<>(list0, list1));
                 }
-        }).returns(ReflectionUtils.specify(Tuple2.class));
+        })
+        .setParallelism(flinkExecutor.fee.getParallelism())
+        .returns(ReflectionUtils.specify(Tuple2.class));
 
         output.accept(datasetOutput, flinkExecutor);
 
@@ -159,5 +161,5 @@ public List<ChannelDescriptor> getSupportedOutputChannels(int index) {
     public boolean containsAction() {
         return false;
     }
-    
+
 }

wayang-platforms/wayang-flink/src/main/java/org/apache/wayang/flink/operators/FlinkCollectionSink.java

@@ -20,6 +20,8 @@
 
 import org.apache.commons.lang3.Validate;
 import org.apache.flink.api.java.DataSet;
+import org.apache.flink.util.Collector;
+import org.apache.flink.api.common.functions.MapPartitionFunction;
 import org.apache.wayang.core.api.Configuration;
 import org.apache.wayang.core.optimizer.OptimizationContext;
 import org.apache.wayang.core.optimizer.cardinality.CardinalityEstimator;
@@ -34,10 +36,13 @@
 import org.apache.wayang.flink.channels.DataSetChannel;
 import org.apache.wayang.flink.execution.FlinkExecutor;
 import org.apache.wayang.java.channels.CollectionChannel;
+import com.esotericsoftware.kryo.Serializer;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
 
 import java.util.Collection;
 import java.util.Collections;
 import java.util.List;
+import java.util.ArrayList;
 import java.util.Optional;
 
 
@@ -60,8 +65,15 @@ public Tuple<Collection<ExecutionLineageNode>, Collection<ChannelInstance>> eval
         final CollectionChannel.Instance output = (CollectionChannel.Instance) outputs[0];
 
         final DataSet<Type> dataSetInput = input.provideDataSet();
+        TypeInformation<Type> type = dataSetInput.getType();
 
-        output.accept(dataSetInput.filter(a -> true).setParallelism(1).collect());
+        if (type.getTypeClass().getName().contains("scala.Tuple")) {
+            flinkExecutor.fee.getConfig().registerTypeWithKryoSerializer(type.getTypeClass(), ScalaTupleSerializer.class);
+        }
+
+        output.accept(dataSetInput
+                .collect()
+        );
 
         return ExecutionOperator.modelLazyExecution(inputs, outputs, operatorContext);
 
@@ -95,4 +107,8 @@ public Optional<CardinalityEstimator> createCardinalityEstimator(
     public String getLoadProfileEstimatorConfigurationKey() {
         return "wayang.flink.collect.load";
     }
+
+    @Override public boolean isConversion() {
+        return true;
+    }
 }

wayang-platforms/wayang-flink/src/main/java/org/apache/wayang/flink/operators/FlinkCollectionSource.java

@@ -19,6 +19,14 @@
 package org.apache.wayang.flink.operators;
 
 import org.apache.flink.api.java.DataSet;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.common.typeinfo.TypeHint;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.java.io.CollectionInputFormat;
+import io.altoo.akka.serialization.kryo.KryoSerializer;
+import org.apache.flink.util.SplittableIterator;
+import org.apache.flink.api.java.typeutils.TypeExtractor;
 import org.apache.wayang.basic.operators.CollectionSource;
 import org.apache.wayang.core.optimizer.OptimizationContext;
 import org.apache.wayang.core.plan.wayangplan.ExecutionOperator;
@@ -30,10 +38,14 @@
 import org.apache.wayang.flink.channels.DataSetChannel;
 import org.apache.wayang.flink.execution.FlinkExecutor;
 import org.apache.wayang.java.channels.CollectionChannel;
+import com.esotericsoftware.kryo.serializers.DefaultSerializers;
 
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.List;
+import java.util.ArrayList;
+import java.util.stream.Collectors;
+import java.util.Collections;
 
 /**
  * This is execution operator implements the {@link CollectionSource}.
@@ -63,8 +75,6 @@ public Tuple<Collection<ExecutionLineageNode>, Collection<ChannelInstance>> eval
             ChannelInstance[] outputs,
             FlinkExecutor flinkExecutor,
             OptimizationContext.OperatorContext operatorContext) {
-        assert inputs.length == 0;
-        assert outputs.length == 1;
 
         final Collection<Type> collection;
         if (this.collection != null) {
@@ -73,8 +83,38 @@ public Tuple<Collection<ExecutionLineageNode>, Collection<ChannelInstance>> eval
             collection = ((CollectionChannel.Instance)inputs[0]).provideCollection();
         }
 
-        final DataSet<Type> datasetOutput = flinkExecutor.fee.fromCollection(collection);
-        ((DataSetChannel.Instance) outputs[0]).accept(datasetOutput, flinkExecutor);
+        SplittableIterator<Type> iterator = new CollectionSplittableIterator<Type>(new ArrayList(collection), flinkExecutor.fee.getParallelism());
+        if (collection.iterator().hasNext()) {
+            Type firstValue = collection.iterator().next();
+
+            TypeInformation<Type> type = TypeExtractor.getForObject(firstValue);
+
+            flinkExecutor.fee.getConfig().registerTypeWithKryoSerializer(ExecutionLineageNode.class, DefaultSerializers.ClassSerializer.class);
+            flinkExecutor.fee.getConfig().registerTypeWithKryoSerializer(ChannelInstance.class, DefaultSerializers.ClassSerializer.class);
+            flinkExecutor.fee.getConfig().registerTypeWithKryoSerializer(CollectionSplittableIterator.class, DefaultSerializers.ClassSerializer.class);
+
+            if (firstValue.getClass().getName().contains("scala.Tuple")) {
+                flinkExecutor.fee.getConfig().registerTypeWithKryoSerializer(firstValue.getClass(), ScalaTupleSerializer.class);
+            }
+
+
+            final DataSet<Type> datasetOutput = flinkExecutor.fee.fromCollection(collection.parallelStream().collect(Collectors.toList()))
+                .setParallelism(flinkExecutor.fee.getParallelism());
+
+            ((DataSetChannel.Instance) outputs[0]).accept(datasetOutput, flinkExecutor);
+
+        } else {
+            //Extremely hacky for empty lists - lord forgive me
+            //This needs to be done as .fromCollection throws exceptions
+            //on an empty list, but in Wayang we can operate on empty results.
+            final DataSet<int[]> datasetOutput = flinkExecutor.fee
+                .fromCollection(List.of(new int[]{0}))
+                .setParallelism(flinkExecutor.fee.getParallelism());
+
+            ((DataSetChannel.Instance) outputs[0]).accept(
+                datasetOutput.filter(x -> false)
+                ,flinkExecutor);
+        }
 
         return ExecutionOperator.modelLazyExecution(inputs, outputs, operatorContext);
     }
@@ -107,4 +147,8 @@ public List<ChannelDescriptor> getSupportedOutputChannels(int index) {
         return Arrays.asList(DataSetChannel.DESCRIPTOR, DataSetChannel.DESCRIPTOR_MANY);
     }
 
+    @Override public boolean isConversion() {
+        return true;
+    }
+
 }

wayang-platforms/wayang-flink/src/main/java/org/apache/wayang/flink/operators/FlinkDistinctOperator.java

@@ -77,7 +77,8 @@ public Tuple<Collection<ExecutionLineageNode>, Collection<ChannelInstance>> eval
 
         final DataSet<Type> dataSetInput = input.provideDataSet();
 
-        final DataSet<Type> dataSetOutput = dataSetInput.distinct(new KeySelectorDistinct<Type>());
+        final DataSet<Type> dataSetOutput = dataSetInput.distinct(new KeySelectorDistinct<Type>())
+            .setParallelism(flinkExecutor.fee.getParallelism());
 
         output.accept(dataSetOutput, flinkExecutor);