Use Spark distributed JDBC DataFrame reader in SqlToRddOperator

[AliRana30]

Description

Closes #761

Currently, SqlToRddOperator executes JDBC queries and materializes the entire ResultSet into a java.util.List in the Spark driver JVM memory before calling sc.parallelize(...). On large analytical engines (such as Trino, Presto, BigQuery, and PostgreSQL), this causes out-of-memory bottlenecks on the driver and prevents Spark from leveraging distributed JDBC execution.

This PR refactors SqlToRddOperator to use Spark's distributed DataFrameReader JDBC interface (executor.ss.read().format("jdbc")):

• Reads JDBC data directly into a distributed Dataset<Row>.
• Converts Dataset<Row> to JavaRDD<Record> lazily across Spark executors using a static serializable helper method (rowToRecord).
• Sanitizes incoming SQL queries and wraps subqueries in derived tables (query) as wayang_subquery.
• Exposes connection parameter getters on DatabaseDescriptor (getJdbcUrl(), getUser(), getPassword(), getJdbcDriverClassName()).
• Adds support for optional JDBC partitioning configurations (partitionColumn, lowerBound, upperBound, numPartitions) and fetchsize.
• Implements toJson() and fromJson() serialization and load profile estimator keys.

Type of Change

• Bug fix (non-breaking change which fixes an issue)
• New feature (non-breaking change which adds functionality)
• Breaking change (fix or feature that would cause existing functionality to not work as expected)
• Documentation update

How Has This Been Tested?

• Added testRowToRecord and testJsonSerialization unit tests to SqlToRddOperatorTest.
• Verified query formatting and schema conversion compatibility with HSQLDB platform tests.

[AliRana30]

@zkaoudi
PR is up, needed your review.

[Copilot]

Pull request overview

Refactors the JDBC-template Spark execution path to avoid collecting JDBC ResultSets on the Spark driver by switching SqlToRddOperator to Spark’s distributed JDBC DataFrameReader, improving scalability for large query results.

Changes:

• Replaces driver-side ResultSet materialization + sc.parallelize(...) with distributed spark.read().format("jdbc") and lazy Row -> Record conversion.
• Adds configurable JDBC read options (credentials, optional partitioning, optional fetch size) and load profile estimator keys for SqlToRddOperator.
• Implements JSON serialization/deserialization for SqlToRddOperator and adds unit tests for rowToRecord and JSON round-trip basics.

Reviewed changes

Copilot reviewed 3 out of 3 changed files in this pull request and generated 2 comments.

File	Description
wayang-platforms/wayang-jdbc-template/src/main/java/org/apache/wayang/jdbc/operators/SqlToRddOperator.java	Moves JDBC reads to Spark’s distributed reader, adds query cleanup, optional read tuning, load profiling keys, and JSON (de)serialization.
wayang-platforms/wayang-jdbc-template/src/main/java/org/apache/wayang/jdbc/execution/DatabaseDescriptor.java	Exposes JDBC connection parameter getters needed by Spark’s JDBC reader.
wayang-platforms/wayang-jdbc-template/src/test/java/org/apache/wayang/jdbc/operators/SqlToRddOperatorTest.java	Adds tests for rowToRecord behavior and basic JSON serialization/deserialization.

---

💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.

[novatechflow]

Using Spark SQL at DataFrameReader.load() does break the current dependency set.

Tests:

• Both JDBC integration tests fail at SqlToRddOperator.java:134 with Could not deserialize ATN with version 3 (expected 4).

Resolved Spark dependencies are mixed:

• spark-core: 3.5.7
• spark-sql, Catalyst, GraphX, MLlib: 3.4.4
• Wayang resolves antlr4-runtime:4.13.1.
• Spark 3.5.7 declares ANTLR 4.9.3.
• wayang-spark/pom.xml also declares ANTLR 4.8.

The POM needs one ${spark.version} for all Spark artifacts. The ANTLR runtime must also be aligned or isolated. Updating Spark alone does not fix the failure because Spark 3.5.7 still requires ANTLR 4.9.3.

Please fix the dependency alignment and rerun SqlToRddOperatorTest.

[AliRana30]

@novatechflow

Updated the PR with the following:

• Refactored SqlToRddOperator to use Spark's distributed DataFrameReader.jdbc and added validation for partition parameters.
• Aligned all Spark modules to ${spark.version} (3.5.7) and antlr4-runtime to 4.9.3 to resolve the Catalyst ANTLR deserialization conflict.
• Rebased and consolidated all changes into a single clean commit.

[novatechflow]

The Spark and ANTLR mismatch is fixed, but always validate the full impact of the dependency upgrade with a full test run:

=> The Spark 3.5.7 upgrade introduces a Parquet runtime regression.

Parquet dependencies are mixed:

• parquet-hadoop: 1.12.3
• parquet-column: 1.13.1
• parquet-avro: 1.15.2

Spark 3.5.7 references CompressionCodecName.LZ4_RAW, but parquet-hadoop:1.12.3 does not provide that field.

Test evidence:

Full test run in that PR produced 3 errors:

• SparkParquetSinkTest.writesDatasetToParquet: NoSuchFieldError: LZ4_RAW
• SparkParquetSinkTest.writesRddToParquet: NoClassDefFoundError after ParquetOptions initialization failed.
• SparkParquetSourceDatasetOutputTest.producesDatasetChannel: same initialization failure.

Aligning the Parquet dependencies with Spark 3.5.7 should fix that, and rerun the Spark reactor afterwards ;)

[AliRana30]

@novatechflow understood, i will just fix these deprecations.

[AliRana30]

@novatechflow , I just aligned the Parquet dependencies to 1.13.1 across wayang-basic and wayang-java via root dependencyManagement.
This resolves the Spark 3.5.7 CompressionCodecName.LZ4_RAW compatibility regression and fixes all SparkParquetSinkTest/SparkParquetSourceDatasetOutputTest test failures.

[novatechflow]

+1 - looks good, but also when we merge, we may introduce incompatibility issues with users relying specifically on Spark 3.4.4 or Parquet 1.15.2.

"release note needed" as flag added.

[novatechflow]

thank you @AliRana30 for this contribution.

[AliRana30]

+1 - looks good, but also when we merge, we may introduce incompatibility issues with users relying specifically on Spark 3.4.4 or Parquet 1.15.2.

"release note needed" as flag added.

@novatechflow whenver we will face any issue in the future regarding the incompability just mention this PR and I will continue working on it.