Add Join operator support for Generic JDBC platform

wayang-platforms/wayang-generic-jdbc/pom.xml

@@ -74,6 +74,13 @@
             <version>1.1.2-SNAPSHOT</version>
             <scope>compile</scope>
         </dependency>
+
+        <dependency>
+            <groupId>org.hsqldb</groupId>
+            <artifactId>hsqldb</artifactId>
+            <version>2.7.2</version>
+            <scope>test</scope>
+        </dependency>
     </dependencies>
 
 

wayang-platforms/wayang-generic-jdbc/src/main/java/org/apache/wayang/genericjdbc/execution/GenericJdbcExecutor.java

@@ -140,14 +140,20 @@ public void execute(ExecutionStage stage, OptimizationContext optimizationContex
      * @return the said follow-up {@link ExecutionTask} or {@code null} if none
      */
     private ExecutionTask findGenericJdbcExecutionOperatorTaskInStage(ExecutionTask task, ExecutionStage stage) {
-        assert task.getNumOuputChannels() == 1;
-        final Channel outputChannel = task.getOutputChannel(0);
-        final ExecutionTask consumer = WayangCollections.getSingle(outputChannel.getConsumers());
-        return consumer.getStage() == stage && consumer.getOperator() instanceof GenericJdbcExecutionOperator ?
-                consumer :
-                null;
+    assert task.getNumOuputChannels() == 1;
+    final Channel outputChannel = task.getOutputChannel(0);
+
+    if (outputChannel.getConsumers().isEmpty()) {
+        return null;
     }
 
+    final ExecutionTask consumer = WayangCollections.getSingle(outputChannel.getConsumers());
+
+    return consumer.getStage() == stage && consumer.getOperator() instanceof GenericJdbcExecutionOperator
+            ? consumer
+            : null;
+}
+
     /**
      * Instantiates the outbound {@link SqlQueryChannel} of an {@link ExecutionTask}.
      *

wayang-platforms/wayang-generic-jdbc/src/main/java/org/apache/wayang/genericjdbc/mapping/JoinMapping.java

@@ -0,0 +1,70 @@
+/*
+ * 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.genericjdbc.mapping;
+
+import org.apache.wayang.basic.data.Record;
+import org.apache.wayang.basic.operators.JoinOperator;
+import org.apache.wayang.core.mapping.Mapping;
+import org.apache.wayang.core.mapping.OperatorPattern;
+import org.apache.wayang.core.mapping.PlanTransformation;
+import org.apache.wayang.core.mapping.ReplacementSubplanFactory;
+import org.apache.wayang.core.mapping.SubplanPattern;
+import org.apache.wayang.core.types.DataSetType;
+import org.apache.wayang.genericjdbc.operators.GenericJdbcJoinOperator;
+import org.apache.wayang.genericjdbc.platform.GenericJdbcPlatform;
+
+import java.util.Collection;
+import java.util.Collections;
+
+/**
+ * Mapping from {@link JoinOperator} to {@link GenericJdbcJoinOperator}.
+ */
+@SuppressWarnings("unchecked")
+public class JoinMapping implements Mapping {
+
+    @Override
+    public Collection<PlanTransformation> getTransformations() {
+        return Collections.singleton(new PlanTransformation(
+                this.createSubplanPattern(),
+                this.createReplacementSubplanFactory(),
+                GenericJdbcPlatform.getInstance()
+        ));
+    }
+
+    private SubplanPattern createSubplanPattern() {
+
+        final OperatorPattern<JoinOperator<Record, Record, ?>> operatorPattern =
+                new OperatorPattern<>(
+                        "join",
+                        new JoinOperator<>(null, null,
+                                DataSetType.createDefault(Record.class),
+                                DataSetType.createDefault(Record.class)),
+                        false
+                );
+
+        return SubplanPattern.createSingleton(operatorPattern);
+    }
+
+    private ReplacementSubplanFactory createReplacementSubplanFactory() {
+        return new ReplacementSubplanFactory.OfSingleOperators<JoinOperator>(
+                (matchedOperator, epoch) ->
+                        new GenericJdbcJoinOperator<>(matchedOperator).at(epoch)
+        );
+    }
+}

wayang-platforms/wayang-generic-jdbc/src/main/java/org/apache/wayang/genericjdbc/mapping/Mappings.java

@@ -30,7 +30,8 @@ public class Mappings {
 
     public static final Collection<Mapping> ALL = Arrays.asList(
             new FilterMapping(),
-            new ProjectionMapping()
+            new ProjectionMapping(),
+            new JoinMapping()
     );
 
 }

wayang-platforms/wayang-generic-jdbc/src/main/java/org/apache/wayang/genericjdbc/operators/GenericJdbcJoinOperator.java

@@ -0,0 +1,47 @@
+/*
+ * 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.genericjdbc.operators;
+
+import org.apache.wayang.basic.data.Record;
+import org.apache.wayang.basic.operators.JoinOperator;
+import org.apache.wayang.core.function.TransformationDescriptor;
+import org.apache.wayang.jdbc.operators.JdbcJoinOperator;
+
+/**
+ * Generic JDBC implementation of the {@link JoinOperator}.
+ */
+public class GenericJdbcJoinOperator<KeyType>
+        extends JdbcJoinOperator<KeyType>
+        implements GenericJdbcExecutionOperator {
+
+    public GenericJdbcJoinOperator(
+            TransformationDescriptor<Record, KeyType> keyDescriptor0,
+            TransformationDescriptor<Record, KeyType> keyDescriptor1) {
+        super(keyDescriptor0, keyDescriptor1);
+    }
+
+    public GenericJdbcJoinOperator(JoinOperator<Record, Record, KeyType> that) {
+        super(that);
+    }
+
+    @Override
+    protected GenericJdbcJoinOperator<KeyType> createCopy() {
+        return new GenericJdbcJoinOperator<>(this);
+    }
+}

wayang-platforms/wayang-generic-jdbc/src/main/java/org/apache/wayang/genericjdbc/operators/GenericJdbcTableSource.java

@@ -33,28 +33,29 @@
 import java.sql.SQLException;
 import java.util.List;
 
-
 public class GenericJdbcTableSource extends JdbcTableSource implements GenericJdbcExecutionOperator {
 
+    /**
+     * Name of the JDBC configuration to use.
+     */
+    public String jdbcName;
+
     /**
      * Creates a new instance.
      *
-     * @see TableSource#TableSource(String, String...)
-     * @param jdbcName on which table resides
-     *
-     *
+     * @param tableName   the table to read from
+     * @param jdbcName    the JDBC configuration name
+     * @param columnNames the columns to read
      */
-
-    public String jdbcName;
-    public GenericJdbcTableSource(String jdbcName, String tableName, String... columnNames) {
+    public GenericJdbcTableSource(String tableName, String... columnNames) {
         super(tableName, columnNames);
-        this.jdbcName = jdbcName;
+        this.jdbcName = "genericjdbc";
     }
 
     /**
      * Copies an instance (exclusive of broadcasts).
      *
-     * @param that that should be copied
+     * @param that the instance that should be copied
      */
     public GenericJdbcTableSource(GenericJdbcTableSource that) {
         super(that);
@@ -66,42 +67,45 @@ public List<ChannelDescriptor> getSupportedInputChannels(int index) {
         throw new UnsupportedOperationException("This operator has no input channels.");
     }
 
+    @Override
     public CardinalityEstimator getCardinalityEstimator(int outputIndex) {
         assert outputIndex == 0;
         return new CardinalityEstimator() {
             @Override
             public CardinalityEstimate estimate(OptimizationContext optimizationContext, CardinalityEstimate... inputEstimates) {
-                // see Job for StopWatch measurements
+
                 final TimeMeasurement timeMeasurement = optimizationContext.getJob().getStopWatch().start(
                         "Optimization", "Cardinality&Load Estimation", "Push Estimation", "Estimate source cardinalities"
                 );
 
-                // Establish a DB connection.
                 try (Connection connection = GenericJdbcPlatform.getInstance()
-                        .createDatabaseDescriptor(optimizationContext.getConfiguration(),jdbcName)
+                        .createDatabaseDescriptor(optimizationContext.getConfiguration(), jdbcName)
                         .createJdbcConnection()) {
 
-                    // Query the table cardinality.
-                    final String sql = String.format("SELECT count(*) FROM %s;", GenericJdbcTableSource.this.getTableName());
+                    final String sql = String.format(
+                            "SELECT count(*) FROM %s;", GenericJdbcTableSource.this.getTableName()
+                    );
+
                     final ResultSet resultSet = connection.createStatement().executeQuery(sql);
+
                     if (!resultSet.next()) {
                         throw new SQLException("No query result for \"" + sql + "\".");
                     }
                     long cardinality = resultSet.getLong(1);
                     return new CardinalityEstimate(cardinality, cardinality, 1d);
 
                 } catch (Exception e) {
+
                     LogManager.getLogger(this.getClass()).error(
                             "Could not estimate cardinality for {}.", GenericJdbcTableSource.this, e
                     );
 
-                    // If we could not load the cardinality, let's use a very conservative estimate.
                     return new CardinalityEstimate(10, 10000000, 0.9);
+
                 } finally {
                     timeMeasurement.stop();
                 }
             }
         };
     }
-}
-
+}