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/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,54 @@ 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"
+                        "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
+                            "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();
                 }
             }
         };
     }
-}
-
+}

wayang-platforms/wayang-generic-jdbc/src/test/java/org/apache/wayang/genericjdbc/operators/GenericJdbcJoinOperatorTest.java

@@ -0,0 +1,110 @@
+package org.apache.wayang.genericjdbc.operators;
+/*
+ * 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.
+ */
+import org.apache.wayang.basic.data.Record;
+import org.apache.wayang.basic.data.Tuple2;
+import org.apache.wayang.basic.operators.JoinOperator;
+import org.apache.wayang.basic.operators.LocalCallbackSink;
+import org.apache.wayang.core.api.Configuration;
+import org.apache.wayang.core.api.WayangContext;
+import org.apache.wayang.core.function.TransformationDescriptor;
+import org.apache.wayang.core.plan.wayangplan.WayangPlan;
+import org.apache.wayang.genericjdbc.GenericJdbc;
+import org.apache.wayang.genericjdbc.platform.GenericJdbcPlatform;
+import org.apache.wayang.java.platform.JavaPlatform;
+import org.junit.jupiter.api.Test;
+
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.Statement;
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+class GenericJdbcJoinOperatorTest {
+
+    @Test
+    void testJoinExecution() throws Exception {
+
+        String url = "jdbc:hsqldb:mem:wayang_test_db;DB_CLOSE_DELAY=-1";
+
+        Class.forName("org.hsqldb.jdbcDriver");
+
+        try (Connection conn = DriverManager.getConnection(url, "SA", "")) {
+            Statement stmt = conn.createStatement();
+
+            stmt.execute("CREATE TABLE T1 (A INT, VAL1 VARCHAR(20));");
+            stmt.execute("INSERT INTO T1 VALUES (1, 'Apache');");
+
+            stmt.execute("CREATE TABLE T2 (A INT, VAL2 INT);");
+            stmt.execute("INSERT INTO T2 VALUES (1, 2026);");
+        }
+
+        Configuration config = new Configuration();
+        config.setProperty("wayang.genericjdbc.jdbc.url", url);
+        config.setProperty("wayang.genericjdbc.jdbc.user", "SA");
+        config.setProperty("wayang.genericjdbc.jdbc.password", "");
+        config.setProperty("wayang.genericjdbc.jdbc.driverName", "org.hsqldb.jdbcDriver");
+
+        GenericJdbcTableSource source1 = new GenericJdbcTableSource("T1", "A", "VAL1");
+        GenericJdbcTableSource source2 = new GenericJdbcTableSource("T2", "A", "VAL2");
+
+        TransformationDescriptor<Record, Integer> keyExtractor0 =
+                new TransformationDescriptor<>(
+                        r -> (Integer) r.getField(0),
+                        Record.class,
+                        Integer.class
+                ).withSqlImplementation("T1", "A");
+
+        TransformationDescriptor<Record, Integer> keyExtractor1 =
+                new TransformationDescriptor<>(
+                        r -> (Integer) r.getField(0),
+                        Record.class,
+                        Integer.class
+                ).withSqlImplementation("T2", "A");
+
+        JoinOperator<Record, Record, Integer> join =
+                new JoinOperator<>(keyExtractor0, keyExtractor1);
+
+        List<Tuple2<Record, Record>> results = new ArrayList<>();
+
+        @SuppressWarnings("unchecked")
+        LocalCallbackSink<Tuple2<Record, Record>> sink =
+                LocalCallbackSink.createCollectingSink(
+                        results,
+                        (Class<Tuple2<Record, Record>>) (Class<?>) Tuple2.class
+                );
+
+        source1.addTargetPlatform(GenericJdbcPlatform.getInstance());
+        source2.addTargetPlatform(GenericJdbcPlatform.getInstance());
+        join.addTargetPlatform(JavaPlatform.getInstance());
+        sink.addTargetPlatform(JavaPlatform.getInstance());
+
+        source1.connectTo(0, join, 0);
+        source2.connectTo(0, join, 1);
+        join.connectTo(0, sink, 0);
+
+        WayangContext ctx = new WayangContext(config)
+                .with(GenericJdbc.plugin())
+                .with(org.apache.wayang.java.Java.basicPlugin());
+
+        ctx.execute(new WayangPlan(sink));
+
+        assertEquals(1, results.size());
+    }
+}

wayang-platforms/wayang-generic-jdbc/src/test/resources/wayang.properties

@@ -0,0 +1,25 @@
+#
+# 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.
+#
+wayang.T1.jdbc.url=jdbc:hsqldb:mem:wayang_test_db;DB_CLOSE_DELAY=-1
+wayang.T1.jdbc.user=SA
+wayang.T1.jdbc.password=
+wayang.T1.jdbc.driverName=org.hsqldb.jdbcDriver
+
+wayang.T2.jdbc.url=jdbc:hsqldb:mem:wayang_test_db;DB_CLOSE_DELAY=-1
+wayang.T2.jdbc.user=SA
+wayang.T2.jdbc.password=
+wayang.T2.jdbc.driverName=org.hsqldb.jdbcDriver