zkaoudi commented on code in PR #793:
URL: https://github.com/apache/wayang/pull/793#discussion_r4155414253
##########
wayang-platforms/wayang-jdbc-template/src/main/java/org/apache/wayang/jdbc/execution/JdbcExecutor.java:
##########
@@ -154,93 +172,106 @@ protected static Tuple2<String,
SqlQueryChannel.Instance> createSqlQuery(final E
assert allTasks.size() <= 3;
ExecutionTask nextTask =
JdbcExecutor.findJdbcExecutionOperatorTaskInStage(startTask, stage);
while (nextTask != null) {
- // Evaluate the nextTask.
final ExecutionOperator operator = nextTask.getOperator();
if (operator instanceof FilterOperator || operator instanceof
SpatialFilterOperator) {
filterTasks.add((JdbcExecutionOperator) operator);
} else if (operator instanceof JdbcProjectionOperator) {
- assert projectionTask == null; // Allow one projection
operator per stage for now.
+ assert projectionTask == null;
projectionTask = (JdbcProjectionOperator) operator;
} else if (operator instanceof final JdbcGlobalReduceOperator
globalReduce) {
- assert globalReduceTask == null; // Allow one projection
operator per stage for now.
+ assert globalReduceTask == null;
globalReduceTask = globalReduce;
} else if (operator instanceof final JdbcReduceByOperator
reduceBy) {
- assert reduceByTask == null; // Allow one projection operator
per stage for now.
+ assert reduceByTask == null;
reduceByTask = reduceBy;
} else if (operator instanceof final JdbcSortOperator sort) {
- assert sortTask == null; // Allow one projection operator per
stage for now.
+ assert sortTask == null;
sortTask = sort;
- } else if (operator instanceof JoinOperator || (operator
instanceof SpatialJoinOperator)) {
+ } else if (operator instanceof JoinOperator || operator instanceof
SpatialJoinOperator) {
joinTasks.add((JdbcExecutionOperator) operator);
} else {
- throw new WayangException(String.format("Unsupported JDBC
execution task %s", nextTask.toString()));
+ throw new WayangException(String.format("Unsupported JDBC
execution task %s", nextTask));
}
- // Move the tipChannelInstance.
tipChannelInstance =
JdbcExecutor.instantiateOutboundChannel(nextTask, context, tipChannelInstance,
jdbcExecutor);
-
- // Go to the next nextTask.
nextTask =
JdbcExecutor.findJdbcExecutionOperatorTaskInStage(nextTask, stage);
}
- // Create the SQL query.
- final StringBuilder query = createSqlString(jdbcExecutor, tableOp,
filterTasks, projectionTask, globalReduceTask, reduceByTask, sortTask,
joinTasks);
-
+ final StringBuilder query = createSqlString(jdbcExecutor, sourceOp,
filterTasks, projectionTask,
+ globalReduceTask, reduceByTask, sortTask, joinTasks,
context.getConfiguration());
return new Tuple2<>(query.toString(), tipChannelInstance);
}
/**
- * Selects the table source that belongs on the left-hand side of a JDBC
join.
+ * Selects the source that belongs on the left-hand side of a JDBC join.
* Stage start tasks are not ordered, but {@link
JdbcJoinOperator#createSqlClause}
- * assumes its first key descriptor's table is used in the {@code FROM}
clause.
+ * assumes its first key descriptor's source is used in the {@code FROM}
clause.
*/
- private static ExecutionTask selectStartTask(final Collection<?>
startTasks, final ExecutionStage stage) {
+ private static ExecutionTask selectStartTask(final Collection<?>
startTasks,
+ final ExecutionStage stage,
+ final Configuration
configuration) {
if (startTasks.size() == 1) {
return (ExecutionTask) startTasks.iterator().next();
}
for (ExecutionTask task : stage.getAllTasks()) {
if (task.getOperator() instanceof JdbcJoinOperator) {
final JdbcJoinOperator<?> joinOperator = (JdbcJoinOperator<?>)
task.getOperator();
- final String leftTableName =
joinOperator.getKeyDescriptor0().getSqlImplementation().field0;
+ final String leftSourceName =
joinOperator.getKeyDescriptor0().getSqlImplementation().field0;
for (Object startTaskObject : startTasks) {
final ExecutionTask startTask = (ExecutionTask)
startTaskObject;
- if (startTask.getOperator() instanceof JdbcTableSource
- && ((JdbcTableSource)
startTask.getOperator()).getTableName().equals(leftTableName)) {
- return startTask;
+ if (startTask.getOperator() instanceof JdbcSourceOperator)
{
+ final JdbcSourceOperator sourceOperator =
(JdbcSourceOperator) startTask.getOperator();
+ if
(sourceOperator.getSourceName().equals(leftSourceName)
+ ||
sourceOperator.getSourceName(configuration).equals(leftSourceName)) {
+ return startTask;
+ }
}
}
}
}
- throw new WayangException("Could not determine the left table source
for JDBC stage.");
+ throw new WayangException("Could not determine the left source for
JDBC stage.");
+ }
+
+ private static void prepareSourceTasks(final Collection<?> startTasks,
+ final JdbcExecutor jdbcExecutor,
+ final Configuration configuration) {
+ for (Object startTaskObject : startTasks) {
+ final ExecutionTask startTask = (ExecutionTask) startTaskObject;
+ if (startTask.getOperator() instanceof JdbcSourceOperator) {
+ ((JdbcSourceOperator) startTask.getOperator()).prepareSource(
+ jdbcExecutor.connection,
+ jdbcExecutor.functionCompiler,
+ configuration
+ );
+ }
+ }
}
/**
* Handles execution stages that end with a {@link JdbcTableSinkOperator}.
- * Composes a SQL query from the stage's operators and executes it
directly on
Review Comment:
Hi @JunWang222, can you put back the comments? Or there is a specific reason
they have to be removed?
##########
wayang-platforms/wayang-jdbc-template/src/main/java/org/apache/wayang/jdbc/execution/JdbcExecutor.java:
##########
@@ -154,93 +172,106 @@ protected static Tuple2<String,
SqlQueryChannel.Instance> createSqlQuery(final E
assert allTasks.size() <= 3;
ExecutionTask nextTask =
JdbcExecutor.findJdbcExecutionOperatorTaskInStage(startTask, stage);
while (nextTask != null) {
- // Evaluate the nextTask.
final ExecutionOperator operator = nextTask.getOperator();
if (operator instanceof FilterOperator || operator instanceof
SpatialFilterOperator) {
filterTasks.add((JdbcExecutionOperator) operator);
} else if (operator instanceof JdbcProjectionOperator) {
- assert projectionTask == null; // Allow one projection
operator per stage for now.
+ assert projectionTask == null;
projectionTask = (JdbcProjectionOperator) operator;
} else if (operator instanceof final JdbcGlobalReduceOperator
globalReduce) {
- assert globalReduceTask == null; // Allow one projection
operator per stage for now.
+ assert globalReduceTask == null;
globalReduceTask = globalReduce;
} else if (operator instanceof final JdbcReduceByOperator
reduceBy) {
- assert reduceByTask == null; // Allow one projection operator
per stage for now.
Review Comment:
This comment is also useful for future and it's better that it stays.
##########
wayang-platforms/wayang-jdbc-template/src/main/java/org/apache/wayang/jdbc/execution/JdbcExecutor.java:
##########
@@ -158,23 +146,18 @@ public static StringBuilder createSqlString(final
JdbcExecutor jdbcExecutor, fin
}
/**
- * Creates a query channel and the sql statement
- *
- * @param stage
Review Comment:
why are these comments deleted?
--
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]
