[SQL / Core] Preserve tuple schema and validate types between SQL and Java operators

wayang-commons/wayang-core/src/main/java/org/apache/wayang/core/optimizer/channels/DefaultChannelConversion.java

@@ -69,6 +69,18 @@ public DefaultChannelConversion(
         );
     }
 
+    public DefaultChannelConversion(
+            ChannelDescriptor sourceChannelDescriptor,
+            ChannelDescriptor targetChannelDescriptor,
+            BiFunction<Channel, Configuration, ExecutionOperator> executionOperatorFactory) {
+        this(
+                sourceChannelDescriptor,
+                targetChannelDescriptor,
+                executionOperatorFactory,
+                "via " + executionOperatorFactory.getClass().getSimpleName()
+        );
+    }
+
     public DefaultChannelConversion(
             ChannelDescriptor sourceChannelDescriptor,
             ChannelDescriptor targetChannelDescriptor,
@@ -89,14 +101,32 @@ public Channel convert(Channel sourceChannel,
         assert executionOperator.getNumInputs() <= 1 && executionOperator.getNumOutputs() <= 1;
         executionOperator.setAuxiliary(true);
 
+        if (sourceChannel != null && sourceChannel.getProducerSlot() != null && executionOperator.getNumInputs() > 0) {
+            org.apache.wayang.core.types.DataSetType<?> sourceChannelType = sourceChannel.getProducerSlot().getType();
+            org.apache.wayang.core.types.DataSetType<?> inputType = executionOperator.getInput(0).getType();
+            if (sourceChannelType != null && inputType != null && !inputType.isSupertypeOf(sourceChannelType)) {
+                try {
+                    java.lang.reflect.Method adaptTypeMethod = executionOperator.getClass().getMethod("adaptType", org.apache.wayang.core.types.DataSetType.class);
+                    adaptTypeMethod.invoke(executionOperator, sourceChannelType);
+                } catch (NoSuchMethodException e) {
+                    throw new IllegalArgumentException(String.format(
+                            "Cannot convert channel %s of type %s with %s of type %s: type mismatch.",
+                            sourceChannel, sourceChannelType, executionOperator, inputType));
+                } catch (Exception e) {
+                    throw new IllegalArgumentException(String.format(
+                            "Cannot convert channel %s of type %s with %s of type %s: type mismatch.",
+                            sourceChannel, sourceChannelType, executionOperator, inputType), e);
+                }
+            }
+        }
+
         // Set up the Channels and the ExecutionTask.
         final ExecutionTask task = new ExecutionTask(executionOperator, 1, 1);
         sourceChannel.addConsumer(task, 0);
         final Channel outputChannel = task.initializeOutputChannel(0, configuration);
         sourceChannel.addSibling(outputChannel);
         setCardinalityAndTimeEstimates(sourceChannel, optimizationContexts, optCardinality, task);
 
-
         return outputChannel;
     }
 

wayang-commons/wayang-core/src/main/java/org/apache/wayang/core/plan/executionplan/Channel.java

@@ -124,6 +124,25 @@ public void addConsumer(ExecutionTask consumer, int inputIndex) {
             assert this.isReusable() || this.consumers.isEmpty() :
                     String.format("Cannot add %s as consumer of non-reusable %s, there is already %s.",
                             consumer, this, this.consumers);
+            if (this.producerSlot != null && consumer.getOperator() != null && inputIndex < consumer.getOperator().getNumInputs()) {
+                InputSlot<?> consumerInput = consumer.getOperator().getInput(inputIndex);
+                if (consumerInput != null && consumerInput.getType() != null && this.producerSlot.getType() != null) {
+                    if (!consumerInput.getType().isSupertypeOf(this.producerSlot.getType())) {
+                        try {
+                            java.lang.reflect.Method adaptTypeMethod = consumer.getOperator().getClass().getMethod("adaptType", DataSetType.class);
+                            adaptTypeMethod.invoke(consumer.getOperator(), this.producerSlot.getType());
+                        } catch (NoSuchMethodException e) {
+                            throw new IllegalArgumentException(String.format(
+                                    "Cannot add consumer %s (input %d type %s) to channel %s with producer type %s: mismatching types.",
+                                    consumer, inputIndex, consumerInput.getType(), this, this.producerSlot.getType()));
+                        } catch (Exception e) {
+                            throw new IllegalArgumentException(String.format(
+                                    "Cannot add consumer %s (input %d type %s) to channel %s with producer type %s: type mismatch.",
+                                    consumer, inputIndex, consumerInput.getType(), this, this.producerSlot.getType()), e);
+                        }
+                    }
+                }
+            }
             this.consumers.add(consumer);
             consumer.setInputChannel(inputIndex, this);
         }

wayang-commons/wayang-core/src/test/java/org/apache/wayang/core/optimizer/channels/ChannelTypeValidationTest.java

@@ -0,0 +1,99 @@
+/*
+ * 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.core.optimizer.channels;
+
+import org.apache.wayang.core.api.Configuration;
+import org.apache.wayang.core.plan.executionplan.Channel;
+import org.apache.wayang.core.plan.executionplan.ExecutionTask;
+import org.apache.wayang.core.platform.ChannelDescriptor;
+import org.apache.wayang.core.test.DummyExecutionOperator;
+import org.apache.wayang.core.test.DummyReusableChannel;
+import org.apache.wayang.core.types.DataSetType;
+import org.junit.jupiter.api.Test;
+
+import java.util.Collections;
+
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+class ChannelTypeValidationTest {
+
+    static class TypedDummyOperator extends DummyExecutionOperator {
+        TypedDummyOperator(DataSetType<?> inputType, DataSetType<?> outputType) {
+            super(inputType != null ? 1 : 0, outputType != null ? 1 : 0, false);
+            if (inputType != null) {
+                this.inputSlots[0] = new org.apache.wayang.core.plan.wayangplan.InputSlot<>("in", this, inputType);
+            }
+            if (outputType != null) {
+                this.outputSlots[0] = new org.apache.wayang.core.plan.wayangplan.OutputSlot<>("out", this, outputType);
+            }
+        }
+    }
+
+    @Test
+    void testChannelAddConsumerThrowsOnIncompatibleTypes() {
+        TypedDummyOperator producerOp = new TypedDummyOperator(null, DataSetType.createDefault(String.class));
+        ExecutionTask producerTask = new ExecutionTask(producerOp);
+        Channel channel = new DummyReusableChannel(DummyReusableChannel.DESCRIPTOR, producerOp.getOutput(0));
+        producerTask.setOutputChannel(0, channel);
+
+        TypedDummyOperator consumerOp = new TypedDummyOperator(DataSetType.createDefault(Integer.class), null);
+        ExecutionTask consumerTask = new ExecutionTask(consumerOp);
+
+        IllegalArgumentException thrown = assertThrows(
+                IllegalArgumentException.class,
+                () -> channel.addConsumer(consumerTask, 0)
+        );
+        assertTrue(thrown.getMessage().contains("mismatching types"));
+    }
+
+    @Test
+    void testChannelAddConsumerSucceedsOnCompatibleTypes() {
+        TypedDummyOperator producerOp = new TypedDummyOperator(null, DataSetType.createDefault(String.class));
+        ExecutionTask producerTask = new ExecutionTask(producerOp);
+        Channel channel = new DummyReusableChannel(DummyReusableChannel.DESCRIPTOR, producerOp.getOutput(0));
+        producerTask.setOutputChannel(0, channel);
+
+        TypedDummyOperator consumerOp = new TypedDummyOperator(DataSetType.createDefault(CharSequence.class), null);
+        ExecutionTask consumerTask = new ExecutionTask(consumerOp);
+
+        assertDoesNotThrow(() -> channel.addConsumer(consumerTask, 0));
+    }
+
+    @Test
+    void testDefaultChannelConversionThrowsOnIncompatibleTypes() {
+        TypedDummyOperator producerOp = new TypedDummyOperator(null, DataSetType.createDefault(String.class));
+        Channel channel = new DummyReusableChannel(DummyReusableChannel.DESCRIPTOR, producerOp.getOutput(0));
+
+        DefaultChannelConversion conversion = new DefaultChannelConversion(
+                DummyReusableChannel.DESCRIPTOR,
+                DummyReusableChannel.DESCRIPTOR,
+                (ch, conf) -> new TypedDummyOperator(
+                        DataSetType.createDefault(Integer.class),
+                        DataSetType.createDefault(Integer.class)
+                )
+        );
+
+        assertThrows(
+                IllegalArgumentException.class,
+                () -> conversion.convert(channel, new Configuration(), Collections.emptyList(), null)
+        );
+    }
+}

wayang-platforms/wayang-bigquery/src/main/java/org/apache/wayang/bigquery/channels/ChannelConversions.java

@@ -18,9 +18,11 @@
 
 package org.apache.wayang.bigquery.channels;
 
+import org.apache.wayang.basic.data.Record;
 import org.apache.wayang.bigquery.platform.BigQueryPlatform;
 import org.apache.wayang.core.optimizer.channels.ChannelConversion;
 import org.apache.wayang.core.optimizer.channels.DefaultChannelConversion;
+import org.apache.wayang.core.types.DataSetType;
 import org.apache.wayang.java.channels.StreamChannel;
 import org.apache.wayang.jdbc.operators.SqlToRddOperator;
 import org.apache.wayang.jdbc.operators.SqlToStreamOperator;
@@ -37,13 +39,23 @@ public class ChannelConversions {
     public static final ChannelConversion SQL_TO_STREAM_CONVERSION = new DefaultChannelConversion(
             BigQueryPlatform.getInstance().getSqlQueryChannelDescriptor(),
             StreamChannel.DESCRIPTOR,
-            () -> new SqlToStreamOperator(BigQueryPlatform.getInstance())
+            (channel, conf) -> new SqlToStreamOperator<>(
+                    BigQueryPlatform.getInstance(),
+                    channel != null && channel.getProducerSlot() != null ?
+                            channel.getProducerSlot().getType() :
+                            DataSetType.createDefault(Record.class)
+            )
     );
 
     public static final ChannelConversion SQL_TO_UNCACHED_RDD_CONVERSION = new DefaultChannelConversion(
             BigQueryPlatform.getInstance().getSqlQueryChannelDescriptor(),
             RddChannel.UNCACHED_DESCRIPTOR,
-            () -> new SqlToRddOperator(BigQueryPlatform.getInstance())
+            (channel, conf) -> new SqlToRddOperator<>(
+                    BigQueryPlatform.getInstance(),
+                    channel != null && channel.getProducerSlot() != null ?
+                            channel.getProducerSlot().getType() :
+                            DataSetType.createDefault(Record.class)
+            )
     );
 
     public static final Collection<ChannelConversion> ALL = Arrays.asList(

wayang-platforms/wayang-generic-jdbc/src/main/java/org/apache/wayang/genericjdbc/channels/GenericChannelConversions.java

@@ -18,8 +18,10 @@
 
 package org.apache.wayang.genericjdbc.channels;
 
+import org.apache.wayang.basic.data.Record;
 import org.apache.wayang.core.optimizer.channels.ChannelConversion;
 import org.apache.wayang.core.optimizer.channels.DefaultChannelConversion;
+import org.apache.wayang.core.types.DataSetType;
 import org.apache.wayang.genericjdbc.operators.GenericSqlToStreamOperator;
 import org.apache.wayang.java.channels.StreamChannel;
 import org.apache.wayang.genericjdbc.platform.GenericJdbcPlatform;
@@ -35,7 +37,12 @@ public class GenericChannelConversions {
     public static final ChannelConversion SQL_TO_STREAM_CONVERSION = new DefaultChannelConversion(
             GenericJdbcPlatform.getInstance().getGenericSqlQueryChannelDescriptor(),
             StreamChannel.DESCRIPTOR,
-            () -> new GenericSqlToStreamOperator(GenericJdbcPlatform.getInstance())
+            (channel, conf) -> new GenericSqlToStreamOperator<>(
+                    GenericJdbcPlatform.getInstance(),
+                    channel != null && channel.getProducerSlot() != null ?
+                            channel.getProducerSlot().getType() :
+                            DataSetType.createDefault(Record.class)
+            )
     );
 
     public static final Collection<ChannelConversion> ALL = Collections.singleton(