atoomula commented on a change in pull request #1393:
URL: https://github.com/apache/samza/pull/1393#discussion_r446549507
##########
File path:
samza-sql/src/main/java/org/apache/samza/sql/translator/JoinTranslator.java
##########
@@ -193,88 +215,133 @@ private void validateJoinQuery(LogicalJoin join,
JoinInputNode.InputType inputTy
dumpRelPlanForNode(join));
}
- if (joinRelType.compareTo(JoinRelType.LEFT) == 0 && isTablePosOnLeft &&
!isTablePosOnRight) {
+ if (joinRelType.compareTo(JoinRelType.LEFT) == 0 && isTablePosOnLeft) {
throw new SamzaException("Invalid query for outer left join. Left side
of the join should be a 'stream' and "
+ "right side of join should be a 'table'. " +
dumpRelPlanForNode(join));
}
- if (joinRelType.compareTo(JoinRelType.RIGHT) == 0 && isTablePosOnRight &&
!isTablePosOnLeft) {
+ if (joinRelType.compareTo(JoinRelType.RIGHT) == 0 && isTablePosOnRight) {
throw new SamzaException("Invalid query for outer right join. Left side
of the join should be a 'table' and "
+ "right side of join should be a 'stream'. " +
dumpRelPlanForNode(join));
}
- validateJoinCondition(join.getCondition());
- }
-
- private void validateJoinCondition(RexNode operand) {
- if (!(operand instanceof RexCall)) {
- throw new SamzaException("SQL Query is not supported. Join condition
operand " + operand +
- " is of type " + operand.getClass());
- }
-
- RexCall condition = (RexCall) operand;
+ final List<RexNode> conjunctionList = new ArrayList<>();
+ decomposeAndValidateConjunction(join.getCondition(), conjunctionList);
- if (condition.isAlwaysTrue()) {
+ if (conjunctionList.isEmpty()) {
throw new SamzaException("Query results in a cross join, which is not
supported. Please optimize the query."
+ " It is expected that the joins should include JOIN ON operator in
the sql query.");
}
-
- if (condition.getKind() != SqlKind.EQUALS && condition.getKind() !=
SqlKind.AND) {
- throw new SamzaException("Only equi-joins and AND operator is supported
in join condition.");
- }
- }
-
- // Fetch the stream and table key indices corresponding to the fields given
in the join condition by parsing through
- // the condition. Stream and table key indices are populated in streamKeyIds
and tableKeyIds respectively.
- private void populateStreamAndTableKeyIds(List<RexNode> operands, final
LogicalJoin join, boolean isTablePosOnRight,
- List<Integer> streamKeyIds, List<Integer> tableKeyIds) {
-
- // All non-leaf operands in the join condition should be expressions.
- if (operands.get(0) instanceof RexCall) {
- operands.forEach(operand -> {
- validateJoinCondition(operand);
- populateStreamAndTableKeyIds(((RexCall) operand).getOperands(), join,
isTablePosOnRight, streamKeyIds, tableKeyIds);
- });
+ //TODO Not sure why we can not allow literal as part of the join condition
will revisit this in another scope
+ conjunctionList.forEach(rexNode -> rexNode.accept(new RexShuttle(){
+ @Override
+ public RexNode visitLiteral(RexLiteral literal) {
+ throw new SamzaException(
+ "Join Condition can not allow literal " + literal.toString() + "
join node" + join.getDigest());
+ }
+ }));
+ final JoinInputNode.InputType rootTableInput = isTablePosOnRight ?
inputTypeOnRight : inputTypeOnLeft;
+ if (rootTableInput.compareTo(JoinInputNode.InputType.REMOTE_TABLE) != 0) {
+ // it is not a remote table all is good we do not have to validate the
project on key Column
return;
}
- // We are at the leaf of the join condition. Only binary operators are
supported.
- Validate.isTrue(operands.size() == 2);
+ /*
+ For remote Table we need to validate The join Condition and The project
that is above remote table scan.
+ - As of today Filter need to be exactly one equi-join using the __key__
column (see SAMZA-2554)
+ - The Project on the top of the remote table has to contain only simple
input references to any of the column used in the join.
+ */
+
+ // First let's collect the ref of columns used by the join condition.
+ List<RexInputRef> refCollector = new ArrayList<>();
+ join.getCondition().accept(new RexShuttle(){
+ @Override
+ public RexNode visitInputRef(RexInputRef inputRef) {
+ refCollector.add(inputRef);
+ return inputRef;
+ }
+ });
+ // start index of the Remote table within the Join Row
+ final int tableStartIndex = isTablePosOnRight ?
join.getLeft().getRowType().getFieldCount() : 0;
+ // end index of the Remote table withing the Join Row
+ final int tableEndIndex =
+ isTablePosOnRight ? join.getRowType().getFieldCount() :
join.getLeft().getRowType().getFieldCount();
+
+ List<Integer> tableRefsIdx = refCollector.stream()
+ .map(x -> x.getIndex())
+ .filter(x -> tableStartIndex <= x && x < tableEndIndex) // collect all
the refs form table side
+ .map(x -> x - tableStartIndex) // re-adjust the offset
+ .sorted()
+ .collect(Collectors.toList()); // we have a list with all the input
from table side with 0 based index.
+
+ // Validate the Condition must contain a ref to remote table primary key
column.
+
+ if (conjunctionList.size() != 1 || tableRefsIdx.size() != 1) {
+ //TODO We can relax this by allowing another filter to be evaluated post
lookup see SAMZA-2554
+ throw new SamzaException(
+ "Invalid query for join condition must contain exactly one predicate
for remote table on __key__ column "
+ + dumpRelPlanForNode(join));
+ }
- // Only reference operands are supported in row expressions and not
constants.
- // a.key = b.key is supported with a.key and b.key being reference
operands.
- // a.key = "constant" is not yet supported.
- if (!(operands.get(0) instanceof RexInputRef) || !(operands.get(1)
instanceof RexInputRef)) {
- throw new SamzaException("SQL query is not supported. Join condition " +
join.getCondition() + " should have "
- + "reference operands but the types are " +
operands.get(0).getClass() + " and " + operands.get(1).getClass());
+ // Validate the Project, follow each input and ensure that it is a simple
ref with no rexCall in the way.
+ if (!isValidRemoteJoinRef(tableRefsIdx.get(0), isTablePosOnRight ?
join.getRight() : join.getLeft())) {
+ throw new SamzaException("Invalid query for join condition can not have
an expression and must be reference "
+ + SamzaSqlRelMessage.KEY_NAME + " column " +
dumpRelPlanForNode(join));
}
+ }
- // Join condition is commutative, meaning, a.key = b.key is equivalent to
b.key = a.key.
- // Calcite assigns the indices to the fields based on the order a and b
are specified in
- // the sql 'from' clause. Let's put the operand with smaller index in
leftRef and larger
- // index in rightRef so that the order of operands in the join condition
is in the order
- // the stream and table are specified in the 'from' clause.
+ /**
+ * Helper method to check if the join condition can be evaluated by the
remote table.
+ * It does follow single path using the index ref path checking if it is a
simple reference all the way to table scan.
+ * In case any RexCall is encountered will stop an return null as a marker
otherwise will return Column Name.
+ *
+ * @param inputRexIndex rex ref index
+ * @param relNode current Rel Node
+ * @return false if any Relational Expression is encountered on the path,
true if is simple ref to __key__ column.
+ */
+ private static boolean isValidRemoteJoinRef(int inputRexIndex, RelNode
relNode){
+ if (relNode instanceof TableScan) {
+ return
relNode.getRowType().getFieldList().get(inputRexIndex).getName().equals(SamzaSqlRelMessage.KEY_NAME);
+ }
+ // has to be a single rel kind filter/project/table scan
Review comment:
could we add tests that go thru' the below code and fail ? If you plan
to add them as part of your nested struct support, please feel free to drop
this.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
[email protected]
