Copilot commented on code in PR #798:
URL: https://github.com/apache/wayang/pull/798#discussion_r3890663436
##########
wayang-platforms/wayang-jdbc-template/src/main/java/org/apache/wayang/jdbc/operators/SqlToRddOperator.java:
##########
@@ -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";
+
Review Comment:
`cleanQuery` can turn a null/blank query (or a query that is only
semicolons) into an empty string. That will make `dbtable` become empty/invalid
("" or "() as wayang_subquery") and fail later with a less clear Spark/JDBC
error. It’s safer to validate the cleaned query and fail fast with a clear
exception message.
##########
wayang-platforms/wayang-jdbc-template/src/main/java/org/apache/wayang/jdbc/operators/SqlToRddOperator.java:
##########
@@ -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) {
+ reader.option("partitionColumn", partitionColumn);
+ String lowerBound = executor.getConfiguration().getStringProperty(
+ String.format("wayang.%s.jdbc.lowerBound",
producerPlatform.getPlatformId()), null);
+ if (lowerBound != null) {
+ reader.option("lowerBound", lowerBound);
+ }
+ String upperBound = executor.getConfiguration().getStringProperty(
+ String.format("wayang.%s.jdbc.upperBound",
producerPlatform.getPlatformId()), null);
+ if (upperBound != null) {
+ reader.option("upperBound", upperBound);
+ }
+ long numPartitions = executor.getConfiguration().getLongProperty(
+ String.format("wayang.%s.jdbc.numPartitions",
producerPlatform.getPlatformId()),
+ executor.getNumDefaultPartitions());
+ reader.option("numPartitions", String.valueOf(numPartitions));
+ }
Review Comment:
When `partitionColumn` is configured, Spark’s JDBC reader expects
`lowerBound`, `upperBound`, and `numPartitions` to be provided together;
allowing `lowerBound`/`upperBound` to be omitted can cause runtime failures
during `load()`. Consider validating these properties and using an `int` for
`numPartitions` to match Spark’s expectation.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
