apache · novatechflow · Sep 2, 2026 · Aug 31, 2026
diff --git a/pom.xml b/pom.xml
@@ -95,7 +95,9 @@
 
         <scala.version>2.12.17</scala.version>
         <scala.mayor.version>2.12</scala.mayor.version>
-        <spark.version>3.4.4</spark.version>
+        <spark.version>3.5.7</spark.version>
+        <antlr.version>4.9.3</antlr.version>
+        <parquet.version>1.13.1</parquet.version>
         <sedona.version>1.6.1</sedona.version>
         <flink.version>1.20.0</flink.version>
         <calcite.version>1.39.0</calcite.version>
@@ -392,7 +394,7 @@
             <properties>
                 <scala.version>2.12.17</scala.version>
                 <scala.mayor.version>2.12</scala.mayor.version>
-                <spark.version>3.4.4</spark.version>
+                <spark.version>3.5.7</spark.version>
             </properties>
             <build>
                 <plugins>
@@ -498,7 +500,7 @@
                 </file>
             </activation>
             <properties>
-                <antlr.version>4.13.1</antlr.version>
+                <antlr.version>4.9.3</antlr.version>
             </properties>
             <build>
                 <pluginManagement>
@@ -733,6 +735,56 @@
                 <artifactId>commons-cli</artifactId>
                 <version>1.3.1</version>
             </dependency>
+            <dependency>
+                <groupId>org.antlr</groupId>
+                <artifactId>antlr4-runtime</artifactId>
+                <version>${antlr.version}</version>
+            </dependency>
+            <dependency>
+                <groupId>org.apache.spark</groupId>
+                <artifactId>spark-core_${scala.mayor.version}</artifactId>
+                <version>${spark.version}</version>
+            </dependency>
+            <dependency>
+                <groupId>org.apache.spark</groupId>
+                <artifactId>spark-graphx_${scala.mayor.version}</artifactId>
+                <version>${spark.version}</version>
+            </dependency>
+            <dependency>
+                <groupId>org.apache.spark</groupId>
+                <artifactId>spark-sql_${scala.mayor.version}</artifactId>
+                <version>${spark.version}</version>
+            </dependency>
+            <dependency>
+                <groupId>org.apache.spark</groupId>
+                <artifactId>spark-mllib_${scala.mayor.version}</artifactId>
+                <version>${spark.version}</version>
+            </dependency>
+            <dependency>
+                <groupId>org.apache.parquet</groupId>
+                <artifactId>parquet-hadoop</artifactId>
+                <version>${parquet.version}</version>
+            </dependency>
+            <dependency>
+                <groupId>org.apache.parquet</groupId>
+                <artifactId>parquet-column</artifactId>
+                <version>${parquet.version}</version>
+            </dependency>
+            <dependency>
+                <groupId>org.apache.parquet</groupId>
+                <artifactId>parquet-avro</artifactId>
+                <version>${parquet.version}</version>
+            </dependency>
+            <dependency>
+                <groupId>org.apache.parquet</groupId>
+                <artifactId>parquet-common</artifactId>
+                <version>${parquet.version}</version>
+            </dependency>
+            <dependency>
+                <groupId>org.apache.parquet</groupId>
+                <artifactId>parquet-encoding</artifactId>
+                <version>${parquet.version}</version>
+            </dependency>
         </dependencies>
     </dependencyManagement>
 

diff --git a/wayang-commons/wayang-basic/pom.xml b/wayang-commons/wayang-basic/pom.xml
@@ -87,7 +87,7 @@
         <dependency>
             <groupId>org.apache.parquet</groupId>
             <artifactId>parquet-hadoop</artifactId>
-            <version>1.12.3</version>
+            <version>${parquet.version}</version>
         </dependency>
         <dependency>
             <groupId>org.apache.commons</groupId>

diff --git a/wayang-commons/wayang-core/pom.xml b/wayang-commons/wayang-core/pom.xml
@@ -83,7 +83,7 @@
         <dependency>
             <groupId>org.antlr</groupId>
             <artifactId>antlr4-runtime</artifactId>
-            <version>4.13.1</version>
+            <version>${antlr.version}</version>
         </dependency>
         <dependency>
             <groupId>org.apache.logging.log4j</groupId>

diff --git a/wayang-platforms/wayang-java/pom.xml b/wayang-platforms/wayang-java/pom.xml
@@ -52,12 +52,12 @@
         <dependency>
             <groupId>org.apache.parquet</groupId>
             <artifactId>parquet-avro</artifactId>
-            <version>1.15.2</version>
+            <version>${parquet.version}</version>
         </dependency>
         <dependency>
             <groupId>org.apache.parquet</groupId>
             <artifactId>parquet-hadoop</artifactId>
-            <version>1.15.2</version>
+            <version>${parquet.version}</version>
         </dependency>
         <dependency>
             <groupId>org.apache.avro</groupId>

diff --git a/...yang-jdbc-template/src/main/java/org/apache/wayang/jdbc/execution/DatabaseDescriptor.java b/...yang-jdbc-template/src/main/java/org/apache/wayang/jdbc/execution/DatabaseDescriptor.java
@@ -46,6 +46,22 @@ public DatabaseDescriptor(String jdbcUrl, String user, String password, String j
         this.jdbcDriverClassName = jdbcDriverClassName;
     }
 
+    public String getJdbcUrl() {
+        return this.jdbcUrl;
+    }
+
+    public String getUser() {
+        return this.user;
+    }
+
+    public String getPassword() {
+        return this.password;
+    }
+
+    public String getJdbcDriverClassName() {
+        return this.jdbcDriverClassName;
+    }
+
     /**
      * Creates a {@link Connection} to the database described by this instance.
      *

diff --git a/...wayang-jdbc-template/src/main/java/org/apache/wayang/jdbc/operators/SqlToRddOperator.java b/...wayang-jdbc-template/src/main/java/org/apache/wayang/jdbc/operators/SqlToRddOperator.java
@@ -19,26 +19,32 @@
 package org.apache.wayang.jdbc.operators;
 
 import org.apache.spark.api.java.JavaRDD;
+import org.apache.spark.sql.DataFrameReader;
+import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.Row;
 import org.apache.wayang.basic.data.Record;
 import org.apache.wayang.core.optimizer.OptimizationContext;
+import org.apache.wayang.core.optimizer.costs.LoadProfileEstimators;
 import org.apache.wayang.core.plan.wayangplan.UnaryToUnaryOperator;
 import org.apache.wayang.core.platform.ChannelDescriptor;
 import org.apache.wayang.core.platform.ChannelInstance;
 import org.apache.wayang.core.platform.lineage.ExecutionLineageNode;
 import org.apache.wayang.core.types.DataSetType;
 import org.apache.wayang.core.util.JsonSerializable;
+import org.apache.wayang.core.util.ReflectionUtils;
 import org.apache.wayang.core.util.Tuple;
 import org.apache.wayang.core.util.json.WayangJsonObj;
 import org.apache.wayang.jdbc.channels.SqlQueryChannel;
+import org.apache.wayang.jdbc.execution.DatabaseDescriptor;
 import org.apache.wayang.jdbc.platform.JdbcPlatformTemplate;
 import org.apache.wayang.spark.channels.RddChannel;
 import org.apache.wayang.spark.execution.SparkExecutor;
 import org.apache.wayang.spark.operators.SparkExecutionOperator;
 
-import java.sql.Connection;
-import java.util.*;
-import java.util.stream.Collectors;
-import java.util.stream.StreamSupport;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
 
 public class SqlToRddOperator extends UnaryToUnaryOperator<Record, Record> implements SparkExecutionOperator, JsonSerializable {
 
@@ -79,37 +85,121 @@ public Tuple<Collection<ExecutionLineageNode>, Collection<ChannelInstance>> eval
         final RddChannel.Instance output = (RddChannel.Instance) outputs[0];
 
         JdbcPlatformTemplate producerPlatform = (JdbcPlatformTemplate) input.getChannel().getProducer().getPlatform();
-        final Connection connection = producerPlatform
-                .createDatabaseDescriptor(executor.getConfiguration())
-                .createJdbcConnection();
-
-        Iterator<Record> resultSetIterator = new SqlToStreamOperator.ResultSetIterator(connection, input.getSqlQuery());
-        Iterable<Record> resultSetIterable = () -> resultSetIterator;
-
-        // Convert the ResultSet to a JavaRDD.
-        JavaRDD<Record> resultSetRDD = executor.sc.parallelize(
-                StreamSupport.stream(resultSetIterable.spliterator(), false).collect(Collectors.toList()),
-                executor.getNumDefaultPartitions()
-        );
+        DatabaseDescriptor databaseDescriptor = producerPlatform.createDatabaseDescriptor(executor.getConfiguration());
+
+        String sqlQuery = cleanQuery(input.getSqlQuery());
+        String dbtable = isTableName(sqlQuery) ? sqlQuery : "(" + sqlQuery + ") as wayang_subquery";
+
+        DataFrameReader reader = executor.ss.read()
+                .format("jdbc")
+                .option("url", databaseDescriptor.getJdbcUrl())
+                .option("dbtable", dbtable)
+                .option("driver", databaseDescriptor.getJdbcDriverClassName());
+
+        if (databaseDescriptor.getUser() != null) {
+            reader.option("user", databaseDescriptor.getUser());
+        }
+        if (databaseDescriptor.getPassword() != null) {
+            reader.option("password", databaseDescriptor.getPassword());
+        }
+
+        // Apply optional partition properties if configured
+        String partitionColumn = executor.getConfiguration().getStringProperty(
+                String.format("wayang.%s.jdbc.partitionColumn", producerPlatform.getPlatformId()), null);
+        if (partitionColumn != null) {
+            String lowerBound = executor.getConfiguration().getStringProperty(
+                    String.format("wayang.%s.jdbc.lowerBound", producerPlatform.getPlatformId()), null);
+            String upperBound = executor.getConfiguration().getStringProperty(
+                    String.format("wayang.%s.jdbc.upperBound", producerPlatform.getPlatformId()), null);
+            if (lowerBound == null || upperBound == null) {
+                throw new IllegalArgumentException(
+                        "JDBC partitioning requires lowerBound and upperBound when partitionColumn is set.");
+            }
+            int numPartitions = executor.getConfiguration().getOptionalIntProperty(
+                    String.format("wayang.%s.jdbc.numPartitions", producerPlatform.getPlatformId()))
+                    .orElse(executor.getNumDefaultPartitions());
+            reader.option("partitionColumn", partitionColumn)
+                    .option("lowerBound", lowerBound)
+                    .option("upperBound", upperBound)
+                    .option("numPartitions", String.valueOf(numPartitions));
+        }
+
+        // Apply optional fetchsize if configured
+        String fetchSize = executor.getConfiguration().getStringProperty(
+                String.format("wayang.%s.jdbc.fetchsize", producerPlatform.getPlatformId()), null);
+        if (fetchSize != null) {
+            reader.option("fetchsize", fetchSize);
+        }
+
+        Dataset<Row> df = reader.load();
+
+        // Convert the distributed DataFrame to JavaRDD<Record> lazily on executors
+        JavaRDD<Record> resultSetRDD = df.toJavaRDD().map(SqlToRddOperator::rowToRecord);
 
         output.accept(resultSetRDD, executor);
 
-        // TODO: Add load profile estimators
         ExecutionLineageNode queryLineageNode = new ExecutionLineageNode(operatorContext);
+        queryLineageNode.add(LoadProfileEstimators.createFromSpecification(
+                String.format("wayang.%s.sqltordd.load.query", this.jdbcPlatform.getPlatformId()),
+                executor.getConfiguration()
+        ));
         queryLineageNode.addPredecessor(input.getLineage());
         ExecutionLineageNode outputLineageNode = new ExecutionLineageNode(operatorContext);
+        outputLineageNode.add(LoadProfileEstimators.createFromSpecification(
+                String.format("wayang.%s.sqltordd.load.output", this.jdbcPlatform.getPlatformId()),
+                executor.getConfiguration()
+        ));
         output.getLineage().addPredecessor(outputLineageNode);
 
         return queryLineageNode.collectAndMark();
     }
 
+    public static Record rowToRecord(Row row) {
+        int length = row.size();
+        Object[] fields = new Object[length];
+        for (int i = 0; i < length; i++) {
+            fields[i] = row.get(i);
+        }
+        return new Record(fields);
+    }
+
+    private static String cleanQuery(String query) {
+        if (query == null) {
+            return "";
+        }
+        String trimmed = query.trim();
+        while (trimmed.endsWith(";")) {
+            trimmed = trimmed.substring(0, trimmed.length() - 1).trim();
+        }
+        return trimmed;
+    }
+
+    private static boolean isTableName(String query) {
+        return !query.contains(" ") && !query.contains("\t") && !query.contains("\n") && !query.contains("(");
+    }
+
     @Override
     public boolean containsAction() {
         return false;
     }
 
+    @Override
+    public Collection<String> getLoadProfileEstimatorConfigurationKeys() {
+        return Arrays.asList(
+                String.format("wayang.%s.sqltordd.load.query", this.jdbcPlatform.getPlatformId()),
+                String.format("wayang.%s.sqltordd.load.output", this.jdbcPlatform.getPlatformId())
+        );
+    }
+
     @Override
     public WayangJsonObj toJson() {
-        return null;
+        return new WayangJsonObj().put("platform", this.jdbcPlatform.getClass().getCanonicalName());
+    }
+
+    @SuppressWarnings("unused")
+    public static SqlToRddOperator fromJson(WayangJsonObj wayangJsonObj) {
+        final String platformClassName = wayangJsonObj.getString("platform");
+        JdbcPlatformTemplate jdbcPlatform = ReflectionUtils.evaluate(platformClassName + ".getInstance()");
+        return new SqlToRddOperator(jdbcPlatform);
     }
 }
diff --git a/...ng-jdbc-template/src/test/java/org/apache/wayang/jdbc/operators/SqlToRddOperatorTest.java b/...ng-jdbc-template/src/test/java/org/apache/wayang/jdbc/operators/SqlToRddOperatorTest.java
@@ -165,4 +165,25 @@ void testWithEmptyHsqldb() throws SQLException {
         assertTrue(output.isEmpty());
     }
 
+    @Test
+    void testRowToRecord() {
+        org.apache.spark.sql.Row row = org.apache.spark.sql.RowFactory.create(1, "test", 42.0);
+        Record record = SqlToRddOperator.rowToRecord(row);
+        assertEquals(3, record.size());
+        assertEquals(1, record.getField(0));
+        assertEquals("test", record.getField(1));
+        assertEquals(42.0, record.getField(2));
+    }
+
+    @Test
+    void testJsonSerialization() {
+        SqlToRddOperator operator = new SqlToRddOperator(HsqldbPlatform.getInstance());
+        org.apache.wayang.core.util.json.WayangJsonObj json = operator.toJson();
+        assertEquals(HsqldbPlatform.class.getCanonicalName(), json.getString("platform"));
+
+        SqlToRddOperator deserialized = SqlToRddOperator.fromJson(json);
+        assertEquals(operator.getInputType(), deserialized.getInputType());
+        assertEquals(operator.getOutputType(), deserialized.getOutputType());
+    }
+
 }
diff --git a/wayang-platforms/wayang-spark/pom.xml b/wayang-platforms/wayang-spark/pom.xml
@@ -77,8 +77,8 @@
         </dependency>
         <dependency>
           <groupId>org.apache.spark</groupId>
-          <artifactId>spark-core_2.12</artifactId>
-          <version>3.5.7</version>
+          <artifactId>spark-core_${scala.mayor.version}</artifactId>
+          <version>${spark.version}</version>
           <exclusions>
             <exclusion>
               <groupId>org.xerial.snappy</groupId>
@@ -88,18 +88,18 @@
         </dependency>
         <dependency>
           <groupId>org.apache.spark</groupId>
-          <artifactId>spark-graphx_2.12</artifactId>
-          <version>3.4.4</version>
+          <artifactId>spark-graphx_${scala.mayor.version}</artifactId>
+          <version>${spark.version}</version>
         </dependency>
         <dependency>
             <groupId>org.apache.spark</groupId>
-            <artifactId>spark-sql_2.12</artifactId>
-            <version>3.4.4</version>
+            <artifactId>spark-sql_${scala.mayor.version}</artifactId>
+            <version>${spark.version}</version>
         </dependency>
         <dependency>
           <groupId>org.apache.spark</groupId>
-          <artifactId>spark-mllib_2.12</artifactId>
-          <version>3.4.4</version>
+          <artifactId>spark-mllib_${scala.mayor.version}</artifactId>
+          <version>${spark.version}</version>
         </dependency>
         <!--Error of ArrayIndexOutOfBoundsException-->
         <dependency>
@@ -117,7 +117,7 @@
         <dependency>
             <groupId>org.antlr</groupId>
             <artifactId>antlr4-runtime</artifactId>
-            <version>4.8</version>
+            <version>${antlr.version}</version>
         </dependency>
 
         <dependency>