zzwqqq commented on code in PR #5356:
URL: https://github.com/apache/calcite/pull/5356#discussion_r4384956639
##########
core/src/main/java/org/apache/calcite/rel/rules/SubQueryRemoveRule.java:
##########
@@ -1275,17 +1325,508 @@ private static void matchJoin(SubQueryRemoveRule rule,
RelOptRuleCall call) {
rule.apply(subQuery, variablesSet, logic, builder, 1, nFieldsRight,
0);
final RexShuttle shuttle = new ReplaceSubQueryShuttle(e,
RexUtil.shift(target, nFieldsLeft));
- RelNode newRight = builder.build();
- builder.push(join.getLeft());
- builder.push(newRight);
+ final RelNode newRight = builder.build();
+ final RexNode newCond = shuttle.apply(join.getCondition());
- builder.join(join.getJoinType(), shuttle.apply(join.getCondition()));
+ // The auxiliary fields are appended after the right-side fields, so the
indexes
+ // that the remaining sub-queries use do not change (hence start and
offset 0).
+ pushJoin(builder, join, join.getLeft(), newRight, newCond,
remainingVariables, 0, 0);
builder.project(fields(builder, nFields));
}
call.transformTo(builder.build());
}
+ /** Moves the part of the condition of {@code join} that contains
+ * {@code e}, a sub-query that references both inputs, to an INNER join of
+ * the domains of the inputs, as in Section 4.1 of "Improving Unnesting of
+ * Complex Queries", and joins the inputs back to it. {@code join} is
+ * neither INNER, which {@link #matchJoin} flattens, nor LEFT_MARK or ASOF,
+ * which are not supported.
+ *
+ * <p>This call rewrites no sub-query itself: {@code e}, and the other
+ * sub-queries of that part, end up in the condition of the INNER join, from
+ * which later calls remove them one at a time.
+ *
+ * <p>For example, for a LEFT join, converts
+ *
+ * <pre>{@code
+ * L LEFT JOIN R ON P(L.x, R.y) AND REST
+ * }</pre>
+ *
+ * <p>into
+ *
+ * <pre>{@code
+ * L LEFT JOIN (
+ * R JOIN (
+ * (SELECT DISTINCT x FROM L) AS DL
+ * JOIN (SELECT DISTINCT y FROM R) AS DR
+ * ON P(DL.x, DR.y)) AS D
+ * ON R.y IS NOT DISTINCT FROM D.y)
+ * ON L.x IS NOT DISTINCT FROM D.x AND REST
+ * }</pre>
+ *
+ * <p>"P" is the conjuncts of the condition that contain {@code e}, "x" and
+ * "y" are the fields of the inputs that it uses, including through
+ * correlation variables, and "REST" is the other conjuncts. D holds one
+ * row per binding for which P is TRUE, so a pair of rows of the inputs
+ * matches at most one row of D, and does so if and only if P is TRUE for
+ * it. Therefore the pairs that match are those of the original join, with
+ * the same multiplicities, and the rows of the preserved input that match no
+ * row are emitted once, null-extended.
+ *
+ * <p>For a RIGHT join, the inputs exchange their roles, so that the
+ * preserved input is attached last, and attached once. For a FULL join both
+ * inputs are preserved. There the entire condition is P, and R is joined
+ * to D with a LEFT join; a marker that is null in the null-extended rows of
+ * D keeps the rows of R that have no binding from matching the rows of L
+ * whose keys are null. Each remaining row of R with a binding matches at
+ * least one row of L, because the bindings come from L, so no row of R is
+ * emitted null-extended more than once.
+ *
+ * <p>Because the inputs are evaluated more than once, and P is evaluated
+ * once per binding instead of once per pair of rows, bails out if the
+ * inputs or P contain a non-deterministic call such as {@code RAND()}. */
+ private static void rewriteOnDomain(RelOptRuleCall call, Join join,
+ RexSubQuery e) {
+ final JoinRelType joinType = join.getJoinType();
+ switch (joinType) {
+ case LEFT:
+ case RIGHT:
+ case FULL:
+ case SEMI:
+ case ANTI:
+ break;
+ default:
+ return;
+ }
+ final RelBuilder builder = call.builder();
+ final RexBuilder rexBuilder = builder.getRexBuilder();
+ final RelNode left = join.getLeft();
+ final RelNode right = join.getRight();
+ final int nLeft = left.getRowType().getFieldCount();
+ final int nRight = right.getRowType().getFieldCount();
+ final boolean fullJoin = joinType == JoinRelType.FULL;
+
+ // A FULL join has to evaluate the entire condition on the domain: if a
+ // binding satisfied P but no pair of rows with it satisfied REST, the rows
+ // of R with that binding would be emitted null-extended once per binding.
+ final List<RexNode> predicates = new ArrayList<>();
+ final List<RexNode> rest = new ArrayList<>();
+ for (RexNode conjunct : RelOptUtil.conjunctions(join.getCondition())) {
+ if (fullJoin || containsSubQuery(conjunct, e)) {
+ predicates.add(conjunct);
+ } else {
+ rest.add(conjunct);
+ }
+ }
+ RexNode predicate = RexUtil.composeConjunction(rexBuilder, predicates);
+ if (!isDeterministic(predicate) || !isDeterministic(left)
+ || !isDeterministic(right)) {
+ return;
+ }
+
+ // The fields that P uses, "x" and "y".
+ final ImmutableBitSet used =
+ RelOptUtil.InputFinder.analyze(predicate,
join.getVariablesSet()).build();
+ final ImmutableBitSet leftUsed =
used.intersect(ImmutableBitSet.range(nLeft));
+ final ImmutableBitSet rightUsed = used.except(leftUsed).shift(-nLeft);
+ final int nLeftUsed = leftUsed.cardinality();
+ final int nUsed = used.cardinality();
+
+ // The sub-queries of REST stay in the join that attaches the last input,
+ // which therefore keeps defining their correlation variables; the
+ // sub-queries of P must use different ids (see renameSharedVariables).
+ final Set<CorrelationId> predicateVariables =
+ intersect(join.getVariablesSet(),
+ RelOptUtil.getVariablesUsed(subQueries(predicate)));
+ final Set<CorrelationId> restVariables =
+ intersect(join.getVariablesSet(),
+ RelOptUtil.getVariablesUsed(
+ subQueries(RexUtil.composeConjunction(rexBuilder, rest))));
+ final Pair<RexNode, Set<CorrelationId>> unshared =
+ renameSharedVariables(predicate, predicateVariables, restVariables,
builder);
+ predicate = unshared.left;
+ final Set<CorrelationId> domainVariables = unshared.right;
+
+ // DL x DR, then the INNER join on P. The domains must be duplicate-free,
+ // otherwise a pair of rows would match several bindings; "aggregateUnique"
+ // keeps the Aggregate even if metadata claims that the keys are unique.
+ final RelBuilder domainBuilder =
+ builder.transform(c -> c.withAggregateUnique(true));
+ final RelNode leftDomain =
+ domainBuilder.push(left).project(domainBuilder.fields(leftUsed))
+ .distinct().build();
+ final RelNode rightDomain =
+ domainBuilder.push(right).project(domainBuilder.fields(rightUsed))
+ .distinct().build();
+ final Mapping mapping = Mappings.target(used.asList(), nLeft + nRight);
+ RexNode domainCondition =
+ predicate.accept(
+ new RexPermuteInputsShuttle(mapping, leftDomain, rightDomain));
+ final RelDataType domainRowType =
+ SqlValidatorUtil.createJoinType(builder.getTypeFactory(),
+ leftDomain.getRowType(), rightDomain.getRowType(), null,
+ ImmutableList.of());
+ for (CorrelationId id : domainVariables) {
+ domainCondition =
+ domainCondition.accept(
+ new RelOptUtil.RexCorrelVariableMapShuttle(id, domainRowType,
+ mapping, rexBuilder));
+ }
+ // The joins that replace "join" keep its hints.
+ builder.push(
+ LogicalJoin.create(leftDomain, rightDomain, join.getHints(),
Review Comment:
Should we use the configured JoinFactory here to support custom join types?
##########
core/src/main/java/org/apache/calcite/rel/rules/SubQueryRemoveRule.java:
##########
@@ -1275,17 +1325,508 @@ private static void matchJoin(SubQueryRemoveRule rule,
RelOptRuleCall call) {
rule.apply(subQuery, variablesSet, logic, builder, 1, nFieldsRight,
0);
final RexShuttle shuttle = new ReplaceSubQueryShuttle(e,
RexUtil.shift(target, nFieldsLeft));
- RelNode newRight = builder.build();
- builder.push(join.getLeft());
- builder.push(newRight);
+ final RelNode newRight = builder.build();
+ final RexNode newCond = shuttle.apply(join.getCondition());
- builder.join(join.getJoinType(), shuttle.apply(join.getCondition()));
+ // The auxiliary fields are appended after the right-side fields, so the
indexes
+ // that the remaining sub-queries use do not change (hence start and
offset 0).
+ pushJoin(builder, join, join.getLeft(), newRight, newCond,
remainingVariables, 0, 0);
builder.project(fields(builder, nFields));
}
call.transformTo(builder.build());
}
+ /** Moves the part of the condition of {@code join} that contains
+ * {@code e}, a sub-query that references both inputs, to an INNER join of
+ * the domains of the inputs, as in Section 4.1 of "Improving Unnesting of
+ * Complex Queries", and joins the inputs back to it. {@code join} is
+ * neither INNER, which {@link #matchJoin} flattens, nor LEFT_MARK or ASOF,
+ * which are not supported.
+ *
+ * <p>This call rewrites no sub-query itself: {@code e}, and the other
+ * sub-queries of that part, end up in the condition of the INNER join, from
+ * which later calls remove them one at a time.
+ *
+ * <p>For example, for a LEFT join, converts
+ *
+ * <pre>{@code
+ * L LEFT JOIN R ON P(L.x, R.y) AND REST
+ * }</pre>
+ *
+ * <p>into
+ *
+ * <pre>{@code
+ * L LEFT JOIN (
+ * R JOIN (
+ * (SELECT DISTINCT x FROM L) AS DL
+ * JOIN (SELECT DISTINCT y FROM R) AS DR
+ * ON P(DL.x, DR.y)) AS D
+ * ON R.y IS NOT DISTINCT FROM D.y)
+ * ON L.x IS NOT DISTINCT FROM D.x AND REST
+ * }</pre>
+ *
+ * <p>"P" is the conjuncts of the condition that contain {@code e}, "x" and
+ * "y" are the fields of the inputs that it uses, including through
+ * correlation variables, and "REST" is the other conjuncts. D holds one
+ * row per binding for which P is TRUE, so a pair of rows of the inputs
+ * matches at most one row of D, and does so if and only if P is TRUE for
+ * it. Therefore the pairs that match are those of the original join, with
+ * the same multiplicities, and the rows of the preserved input that match no
+ * row are emitted once, null-extended.
+ *
+ * <p>For a RIGHT join, the inputs exchange their roles, so that the
+ * preserved input is attached last, and attached once. For a FULL join both
+ * inputs are preserved. There the entire condition is P, and R is joined
+ * to D with a LEFT join; a marker that is null in the null-extended rows of
+ * D keeps the rows of R that have no binding from matching the rows of L
+ * whose keys are null. Each remaining row of R with a binding matches at
+ * least one row of L, because the bindings come from L, so no row of R is
+ * emitted null-extended more than once.
+ *
+ * <p>Because the inputs are evaluated more than once, and P is evaluated
+ * once per binding instead of once per pair of rows, bails out if the
+ * inputs or P contain a non-deterministic call such as {@code RAND()}. */
+ private static void rewriteOnDomain(RelOptRuleCall call, Join join,
+ RexSubQuery e) {
+ final JoinRelType joinType = join.getJoinType();
+ switch (joinType) {
+ case LEFT:
+ case RIGHT:
+ case FULL:
+ case SEMI:
+ case ANTI:
+ break;
+ default:
+ return;
+ }
+ final RelBuilder builder = call.builder();
+ final RexBuilder rexBuilder = builder.getRexBuilder();
+ final RelNode left = join.getLeft();
+ final RelNode right = join.getRight();
+ final int nLeft = left.getRowType().getFieldCount();
+ final int nRight = right.getRowType().getFieldCount();
+ final boolean fullJoin = joinType == JoinRelType.FULL;
+
+ // A FULL join has to evaluate the entire condition on the domain: if a
+ // binding satisfied P but no pair of rows with it satisfied REST, the rows
+ // of R with that binding would be emitted null-extended once per binding.
+ final List<RexNode> predicates = new ArrayList<>();
+ final List<RexNode> rest = new ArrayList<>();
+ for (RexNode conjunct : RelOptUtil.conjunctions(join.getCondition())) {
+ if (fullJoin || containsSubQuery(conjunct, e)) {
+ predicates.add(conjunct);
+ } else {
+ rest.add(conjunct);
+ }
+ }
+ RexNode predicate = RexUtil.composeConjunction(rexBuilder, predicates);
+ if (!isDeterministic(predicate) || !isDeterministic(left)
+ || !isDeterministic(right)) {
+ return;
+ }
+
+ // The fields that P uses, "x" and "y".
+ final ImmutableBitSet used =
+ RelOptUtil.InputFinder.analyze(predicate,
join.getVariablesSet()).build();
+ final ImmutableBitSet leftUsed =
used.intersect(ImmutableBitSet.range(nLeft));
+ final ImmutableBitSet rightUsed = used.except(leftUsed).shift(-nLeft);
+ final int nLeftUsed = leftUsed.cardinality();
+ final int nUsed = used.cardinality();
+
+ // The sub-queries of REST stay in the join that attaches the last input,
+ // which therefore keeps defining their correlation variables; the
+ // sub-queries of P must use different ids (see renameSharedVariables).
+ final Set<CorrelationId> predicateVariables =
+ intersect(join.getVariablesSet(),
+ RelOptUtil.getVariablesUsed(subQueries(predicate)));
+ final Set<CorrelationId> restVariables =
+ intersect(join.getVariablesSet(),
+ RelOptUtil.getVariablesUsed(
+ subQueries(RexUtil.composeConjunction(rexBuilder, rest))));
+ final Pair<RexNode, Set<CorrelationId>> unshared =
+ renameSharedVariables(predicate, predicateVariables, restVariables,
builder);
+ predicate = unshared.left;
+ final Set<CorrelationId> domainVariables = unshared.right;
+
+ // DL x DR, then the INNER join on P. The domains must be duplicate-free,
+ // otherwise a pair of rows would match several bindings; "aggregateUnique"
+ // keeps the Aggregate even if metadata claims that the keys are unique.
+ final RelBuilder domainBuilder =
+ builder.transform(c -> c.withAggregateUnique(true));
+ final RelNode leftDomain =
+ domainBuilder.push(left).project(domainBuilder.fields(leftUsed))
+ .distinct().build();
+ final RelNode rightDomain =
+ domainBuilder.push(right).project(domainBuilder.fields(rightUsed))
+ .distinct().build();
+ final Mapping mapping = Mappings.target(used.asList(), nLeft + nRight);
+ RexNode domainCondition =
+ predicate.accept(
+ new RexPermuteInputsShuttle(mapping, leftDomain, rightDomain));
+ final RelDataType domainRowType =
+ SqlValidatorUtil.createJoinType(builder.getTypeFactory(),
+ leftDomain.getRowType(), rightDomain.getRowType(), null,
+ ImmutableList.of());
+ for (CorrelationId id : domainVariables) {
+ domainCondition =
+ domainCondition.accept(
+ new RelOptUtil.RexCorrelVariableMapShuttle(id, domainRowType,
+ mapping, rexBuilder));
+ }
+ // The joins that replace "join" keep its hints.
+ builder.push(
+ LogicalJoin.create(leftDomain, rightDomain, join.getHints(),
+ domainCondition, domainVariables, JoinRelType.INNER));
+ if (fullJoin) {
+ final List<RexNode> domainFields = new ArrayList<>(builder.fields());
+ domainFields.add(builder.literal(true));
+ builder.project(domainFields);
+ }
+ final RelNode domain = builder.build();
+
+ final List<RexNode> conditions = new ArrayList<>();
+ if (joinType == JoinRelType.RIGHT) {
+ // (L JOIN D ON L.x = D.x) RIGHT JOIN R ON R.y = D.y AND REST. The
+ // fields of R move past those of D.
+ builder.push(left).push(domain)
+ .join(JoinRelType.INNER,
+ domainKeys(rexBuilder, left, domain, leftUsed, 0, nLeft))
+ .hints(join.getHints());
+ final RelNode first = builder.build();
+ conditions.addAll(
+ domainKeys(rexBuilder, first, right, rightUsed, nLeft + nUsed,
+ nLeft + nLeftUsed));
+ for (RexNode conjunct : rest) {
+ conditions.add(RexUtil.shift(conjunct, nLeft, nUsed));
+ }
+ pushJoin(builder, join, first, right,
+ RexUtil.composeConjunction(rexBuilder, conditions), restVariables,
+ nLeft, nUsed);
+ builder.project(
+ builder.fields(ImmutableBitSet.range(nLeft)
+ .union(ImmutableBitSet.range(nLeft + nUsed, nLeft + nUsed +
nRight))),
+ join.getRowType().getFieldNames());
+ } else {
+ // L JOIN_TYPE (R JOIN D ON R.y = D.y) ON L.x = D.x AND REST. The fields
+ // of D are appended after those of R, so REST does not change.
+ builder.push(right).push(domain)
+ .join(fullJoin ? JoinRelType.LEFT : JoinRelType.INNER,
+ domainKeys(rexBuilder, right, domain, rightUsed, 0,
+ nRight + nLeftUsed))
+ .hints(join.getHints());
+ final RelNode first = builder.build();
+ conditions.addAll(
+ domainKeys(rexBuilder, left, first, leftUsed, 0, nLeft + nRight));
+ if (fullJoin) {
+ conditions.add(joinField(rexBuilder, left, first, nLeft + nRight +
nUsed));
+ }
+ conditions.addAll(rest);
+ pushJoin(builder, join, left, first,
+ RexUtil.composeConjunction(rexBuilder, conditions), restVariables,
+ 0, 0);
+ final List<RexNode> output = new ArrayList<>(fields(builder, nLeft));
+ if (joinType.projectsRight()) {
+ output.addAll(
+ builder.fields(ImmutableBitSet.range(nLeft, nLeft + nRight)));
+ }
+ builder.project(output, join.getRowType().getFieldNames());
+ }
+ call.transformTo(builder.build());
+ }
+
+ /** Returns the conditions of a join of {@code left} and {@code right} that
+ * the fields {@code fields}, shifted by {@code fieldOffset}, have the same
+ * values as the consecutive fields that start at {@code keyOffset}; nulls
+ * are equal to each other. Indexes are those of the row that the condition
+ * sees, the fields of {@code left} followed by those of {@code right}. */
+ private static List<RexNode> domainKeys(RexBuilder rexBuilder, RelNode left,
+ RelNode right, ImmutableBitSet fields, int fieldOffset, int keyOffset) {
+ final List<RexNode> keys = new ArrayList<>();
+ int key = keyOffset;
+ for (int field : fields) {
+ keys.add(
+ RelOptUtil.isDistinctFrom(rexBuilder,
+ joinField(rexBuilder, left, right, fieldOffset + field),
+ joinField(rexBuilder, left, right, key++), true));
+ }
+ return keys;
+ }
+
+ /** Returns a reference to field {@code index} of the row that the condition
+ * of a join of {@code left} and {@code right} sees. */
+ private static RexNode joinField(RexBuilder rexBuilder, RelNode left,
+ RelNode right, int index) {
+ final int nLeft = left.getRowType().getFieldCount();
+ final RelDataType type = index < nLeft
+ ? left.getRowType().getFieldList().get(index).getType()
+ : right.getRowType().getFieldList().get(index - nLeft).getType();
+ return rexBuilder.makeInputRef(type, index);
+ }
+
+ /** Returns whether {@code node} contains {@code subQuery}, not counting
+ * the relational expressions of sub-queries. */
+ private static boolean containsSubQuery(RexNode node, RexSubQuery subQuery) {
+ return subQueries(node).contains(subQuery);
+ }
+
+ /** Returns the sub-queries of {@code node}, including those in the
+ * operands of other sub-queries but not those in their relational
+ * expressions. */
+ private static List<RexSubQuery> subQueries(RexNode node) {
+ final RexUtil.SubQueryCollector collector = new
RexUtil.SubQueryCollector(true);
+ node.accept(collector);
+ return collector.getSubQueries();
+ }
+
+ /** Returns whether {@code node}, including its sub-queries, contains no
+ * non-deterministic call. */
+ private static boolean isDeterministic(RexNode node) {
+ try {
+ node.accept(NonDeterministicFinder.INSTANCE);
+ return true;
+ } catch (Util.FoundOne e) {
+ return false;
+ }
+ }
+
+ /** Returns whether {@code rel}, including its inputs and sub-queries,
+ * contains no non-deterministic call. */
+ private static boolean isDeterministic(RelNode rel) {
+ try {
+ NonDeterministicFinder.INSTANCE.visitRel(rel);
+ return true;
+ } catch (Util.FoundOne e) {
+ return false;
+ }
+ }
+
+ /** Throws {@link Util.FoundOne} if it finds a non-deterministic call in an
+ * expression, in the relational expressions of its sub-queries, or in a
+ * relational expression and its inputs. */
+ private static class NonDeterministicFinder extends RexShuttle {
Review Comment:
Should we also consider non-repeatable Sample inputs? Reading them twice can
produce different samples and incorrect results.
##########
core/src/main/java/org/apache/calcite/rel/rules/SubQueryRemoveRule.java:
##########
@@ -1275,17 +1325,508 @@ private static void matchJoin(SubQueryRemoveRule rule,
RelOptRuleCall call) {
rule.apply(subQuery, variablesSet, logic, builder, 1, nFieldsRight,
0);
final RexShuttle shuttle = new ReplaceSubQueryShuttle(e,
RexUtil.shift(target, nFieldsLeft));
- RelNode newRight = builder.build();
- builder.push(join.getLeft());
- builder.push(newRight);
+ final RelNode newRight = builder.build();
+ final RexNode newCond = shuttle.apply(join.getCondition());
- builder.join(join.getJoinType(), shuttle.apply(join.getCondition()));
+ // The auxiliary fields are appended after the right-side fields, so the
indexes
+ // that the remaining sub-queries use do not change (hence start and
offset 0).
+ pushJoin(builder, join, join.getLeft(), newRight, newCond,
remainingVariables, 0, 0);
builder.project(fields(builder, nFields));
}
call.transformTo(builder.build());
}
+ /** Moves the part of the condition of {@code join} that contains
+ * {@code e}, a sub-query that references both inputs, to an INNER join of
+ * the domains of the inputs, as in Section 4.1 of "Improving Unnesting of
+ * Complex Queries", and joins the inputs back to it. {@code join} is
+ * neither INNER, which {@link #matchJoin} flattens, nor LEFT_MARK or ASOF,
+ * which are not supported.
+ *
+ * <p>This call rewrites no sub-query itself: {@code e}, and the other
+ * sub-queries of that part, end up in the condition of the INNER join, from
+ * which later calls remove them one at a time.
+ *
+ * <p>For example, for a LEFT join, converts
+ *
+ * <pre>{@code
+ * L LEFT JOIN R ON P(L.x, R.y) AND REST
+ * }</pre>
+ *
+ * <p>into
+ *
+ * <pre>{@code
+ * L LEFT JOIN (
+ * R JOIN (
+ * (SELECT DISTINCT x FROM L) AS DL
+ * JOIN (SELECT DISTINCT y FROM R) AS DR
+ * ON P(DL.x, DR.y)) AS D
+ * ON R.y IS NOT DISTINCT FROM D.y)
+ * ON L.x IS NOT DISTINCT FROM D.x AND REST
+ * }</pre>
+ *
+ * <p>"P" is the conjuncts of the condition that contain {@code e}, "x" and
+ * "y" are the fields of the inputs that it uses, including through
+ * correlation variables, and "REST" is the other conjuncts. D holds one
+ * row per binding for which P is TRUE, so a pair of rows of the inputs
+ * matches at most one row of D, and does so if and only if P is TRUE for
+ * it. Therefore the pairs that match are those of the original join, with
+ * the same multiplicities, and the rows of the preserved input that match no
+ * row are emitted once, null-extended.
+ *
+ * <p>For a RIGHT join, the inputs exchange their roles, so that the
+ * preserved input is attached last, and attached once. For a FULL join both
+ * inputs are preserved. There the entire condition is P, and R is joined
+ * to D with a LEFT join; a marker that is null in the null-extended rows of
+ * D keeps the rows of R that have no binding from matching the rows of L
+ * whose keys are null. Each remaining row of R with a binding matches at
+ * least one row of L, because the bindings come from L, so no row of R is
+ * emitted null-extended more than once.
+ *
+ * <p>Because the inputs are evaluated more than once, and P is evaluated
+ * once per binding instead of once per pair of rows, bails out if the
+ * inputs or P contain a non-deterministic call such as {@code RAND()}. */
+ private static void rewriteOnDomain(RelOptRuleCall call, Join join,
+ RexSubQuery e) {
+ final JoinRelType joinType = join.getJoinType();
+ switch (joinType) {
+ case LEFT:
+ case RIGHT:
+ case FULL:
+ case SEMI:
+ case ANTI:
+ break;
+ default:
+ return;
+ }
+ final RelBuilder builder = call.builder();
+ final RexBuilder rexBuilder = builder.getRexBuilder();
+ final RelNode left = join.getLeft();
+ final RelNode right = join.getRight();
+ final int nLeft = left.getRowType().getFieldCount();
+ final int nRight = right.getRowType().getFieldCount();
+ final boolean fullJoin = joinType == JoinRelType.FULL;
+
+ // A FULL join has to evaluate the entire condition on the domain: if a
+ // binding satisfied P but no pair of rows with it satisfied REST, the rows
+ // of R with that binding would be emitted null-extended once per binding.
+ final List<RexNode> predicates = new ArrayList<>();
+ final List<RexNode> rest = new ArrayList<>();
+ for (RexNode conjunct : RelOptUtil.conjunctions(join.getCondition())) {
+ if (fullJoin || containsSubQuery(conjunct, e)) {
+ predicates.add(conjunct);
+ } else {
+ rest.add(conjunct);
+ }
+ }
+ RexNode predicate = RexUtil.composeConjunction(rexBuilder, predicates);
+ if (!isDeterministic(predicate) || !isDeterministic(left)
+ || !isDeterministic(right)) {
+ return;
+ }
+
+ // The fields that P uses, "x" and "y".
+ final ImmutableBitSet used =
+ RelOptUtil.InputFinder.analyze(predicate,
join.getVariablesSet()).build();
+ final ImmutableBitSet leftUsed =
used.intersect(ImmutableBitSet.range(nLeft));
+ final ImmutableBitSet rightUsed = used.except(leftUsed).shift(-nLeft);
+ final int nLeftUsed = leftUsed.cardinality();
+ final int nUsed = used.cardinality();
+
+ // The sub-queries of REST stay in the join that attaches the last input,
+ // which therefore keeps defining their correlation variables; the
+ // sub-queries of P must use different ids (see renameSharedVariables).
+ final Set<CorrelationId> predicateVariables =
+ intersect(join.getVariablesSet(),
+ RelOptUtil.getVariablesUsed(subQueries(predicate)));
+ final Set<CorrelationId> restVariables =
+ intersect(join.getVariablesSet(),
+ RelOptUtil.getVariablesUsed(
+ subQueries(RexUtil.composeConjunction(rexBuilder, rest))));
+ final Pair<RexNode, Set<CorrelationId>> unshared =
+ renameSharedVariables(predicate, predicateVariables, restVariables,
builder);
+ predicate = unshared.left;
+ final Set<CorrelationId> domainVariables = unshared.right;
+
+ // DL x DR, then the INNER join on P. The domains must be duplicate-free,
+ // otherwise a pair of rows would match several bindings; "aggregateUnique"
+ // keeps the Aggregate even if metadata claims that the keys are unique.
+ final RelBuilder domainBuilder =
+ builder.transform(c -> c.withAggregateUnique(true));
+ final RelNode leftDomain =
+ domainBuilder.push(left).project(domainBuilder.fields(leftUsed))
+ .distinct().build();
+ final RelNode rightDomain =
+ domainBuilder.push(right).project(domainBuilder.fields(rightUsed))
+ .distinct().build();
+ final Mapping mapping = Mappings.target(used.asList(), nLeft + nRight);
+ RexNode domainCondition =
+ predicate.accept(
+ new RexPermuteInputsShuttle(mapping, leftDomain, rightDomain));
+ final RelDataType domainRowType =
+ SqlValidatorUtil.createJoinType(builder.getTypeFactory(),
+ leftDomain.getRowType(), rightDomain.getRowType(), null,
+ ImmutableList.of());
+ for (CorrelationId id : domainVariables) {
+ domainCondition =
+ domainCondition.accept(
+ new RelOptUtil.RexCorrelVariableMapShuttle(id, domainRowType,
+ mapping, rexBuilder));
+ }
+ // The joins that replace "join" keep its hints.
+ builder.push(
+ LogicalJoin.create(leftDomain, rightDomain, join.getHints(),
+ domainCondition, domainVariables, JoinRelType.INNER));
+ if (fullJoin) {
+ final List<RexNode> domainFields = new ArrayList<>(builder.fields());
+ domainFields.add(builder.literal(true));
+ builder.project(domainFields);
+ }
+ final RelNode domain = builder.build();
+
+ final List<RexNode> conditions = new ArrayList<>();
+ if (joinType == JoinRelType.RIGHT) {
+ // (L JOIN D ON L.x = D.x) RIGHT JOIN R ON R.y = D.y AND REST. The
+ // fields of R move past those of D.
+ builder.push(left).push(domain)
+ .join(JoinRelType.INNER,
+ domainKeys(rexBuilder, left, domain, leftUsed, 0, nLeft))
+ .hints(join.getHints());
+ final RelNode first = builder.build();
+ conditions.addAll(
+ domainKeys(rexBuilder, first, right, rightUsed, nLeft + nUsed,
+ nLeft + nLeftUsed));
+ for (RexNode conjunct : rest) {
+ conditions.add(RexUtil.shift(conjunct, nLeft, nUsed));
+ }
+ pushJoin(builder, join, first, right,
+ RexUtil.composeConjunction(rexBuilder, conditions), restVariables,
+ nLeft, nUsed);
+ builder.project(
+ builder.fields(ImmutableBitSet.range(nLeft)
+ .union(ImmutableBitSet.range(nLeft + nUsed, nLeft + nUsed +
nRight))),
+ join.getRowType().getFieldNames());
+ } else {
+ // L JOIN_TYPE (R JOIN D ON R.y = D.y) ON L.x = D.x AND REST. The fields
+ // of D are appended after those of R, so REST does not change.
+ builder.push(right).push(domain)
+ .join(fullJoin ? JoinRelType.LEFT : JoinRelType.INNER,
+ domainKeys(rexBuilder, right, domain, rightUsed, 0,
+ nRight + nLeftUsed))
+ .hints(join.getHints());
+ final RelNode first = builder.build();
+ conditions.addAll(
+ domainKeys(rexBuilder, left, first, leftUsed, 0, nLeft + nRight));
+ if (fullJoin) {
+ conditions.add(joinField(rexBuilder, left, first, nLeft + nRight +
nUsed));
+ }
+ conditions.addAll(rest);
+ pushJoin(builder, join, left, first,
+ RexUtil.composeConjunction(rexBuilder, conditions), restVariables,
+ 0, 0);
+ final List<RexNode> output = new ArrayList<>(fields(builder, nLeft));
+ if (joinType.projectsRight()) {
+ output.addAll(
+ builder.fields(ImmutableBitSet.range(nLeft, nLeft + nRight)));
+ }
+ builder.project(output, join.getRowType().getFieldNames());
+ }
+ call.transformTo(builder.build());
+ }
+
+ /** Returns the conditions of a join of {@code left} and {@code right} that
+ * the fields {@code fields}, shifted by {@code fieldOffset}, have the same
+ * values as the consecutive fields that start at {@code keyOffset}; nulls
+ * are equal to each other. Indexes are those of the row that the condition
+ * sees, the fields of {@code left} followed by those of {@code right}. */
+ private static List<RexNode> domainKeys(RexBuilder rexBuilder, RelNode left,
+ RelNode right, ImmutableBitSet fields, int fieldOffset, int keyOffset) {
+ final List<RexNode> keys = new ArrayList<>();
+ int key = keyOffset;
+ for (int field : fields) {
+ keys.add(
+ RelOptUtil.isDistinctFrom(rexBuilder,
+ joinField(rexBuilder, left, right, fieldOffset + field),
+ joinField(rexBuilder, left, right, key++), true));
+ }
+ return keys;
+ }
+
+ /** Returns a reference to field {@code index} of the row that the condition
+ * of a join of {@code left} and {@code right} sees. */
+ private static RexNode joinField(RexBuilder rexBuilder, RelNode left,
+ RelNode right, int index) {
+ final int nLeft = left.getRowType().getFieldCount();
+ final RelDataType type = index < nLeft
+ ? left.getRowType().getFieldList().get(index).getType()
+ : right.getRowType().getFieldList().get(index - nLeft).getType();
+ return rexBuilder.makeInputRef(type, index);
+ }
+
+ /** Returns whether {@code node} contains {@code subQuery}, not counting
+ * the relational expressions of sub-queries. */
+ private static boolean containsSubQuery(RexNode node, RexSubQuery subQuery) {
+ return subQueries(node).contains(subQuery);
+ }
+
+ /** Returns the sub-queries of {@code node}, including those in the
+ * operands of other sub-queries but not those in their relational
+ * expressions. */
+ private static List<RexSubQuery> subQueries(RexNode node) {
+ final RexUtil.SubQueryCollector collector = new
RexUtil.SubQueryCollector(true);
+ node.accept(collector);
+ return collector.getSubQueries();
+ }
+
+ /** Returns whether {@code node}, including its sub-queries, contains no
+ * non-deterministic call. */
+ private static boolean isDeterministic(RexNode node) {
+ try {
+ node.accept(NonDeterministicFinder.INSTANCE);
+ return true;
+ } catch (Util.FoundOne e) {
+ return false;
+ }
+ }
+
+ /** Returns whether {@code rel}, including its inputs and sub-queries,
+ * contains no non-deterministic call. */
+ private static boolean isDeterministic(RelNode rel) {
+ try {
+ NonDeterministicFinder.INSTANCE.visitRel(rel);
+ return true;
+ } catch (Util.FoundOne e) {
+ return false;
+ }
+ }
+
+ /** Throws {@link Util.FoundOne} if it finds a non-deterministic call in an
+ * expression, in the relational expressions of its sub-queries, or in a
+ * relational expression and its inputs. */
+ private static class NonDeterministicFinder extends RexShuttle {
+ static final NonDeterministicFinder INSTANCE = new
NonDeterministicFinder();
+
+ @Override public RexNode visitCall(RexCall call) {
+ if (!call.getOperator().isDeterministic()) {
+ throw Util.FoundOne.NULL;
+ }
+ return super.visitCall(call);
+ }
+
+ @Override public RexNode visitSubQuery(RexSubQuery subQuery) {
+ visitRel(subQuery.rel);
+ return super.visitSubQuery(subQuery);
+ }
+
+ void visitRel(RelNode rel) {
+ // Look through the wrappers of the planners, such as HepRelVertex,
+ // whose inputs are not those of the expression that they wrap.
+ final RelNode stripped = rel.stripped();
+ stripped.accept(this);
+ for (RelNode input : stripped.getInputs()) {
+ visitRel(input);
+ }
+ }
+ }
+
+ /** Returns the intersection of two sets of correlation variables, in the
+ * iteration order of {@code defined}.
+ *
+ * <p>The result ends up in the {@code variablesSet} of a relational
expression
+ * (and therefore in its digest), so a deterministic order matters. */
+ private static Set<CorrelationId> intersect(Set<CorrelationId> defined,
+ Set<CorrelationId> used) {
+ if (defined.isEmpty() || used.isEmpty()) {
+ return ImmutableSet.of();
+ }
+ return defined.stream().filter(used::contains)
+ .collect(ImmutableSet.toImmutableSet());
+ }
+
+ /** Returns the sub-queries of {@code condition}, except {@code e}.
+ *
+ * <p>Sub-queries are compared by value, as {@link ReplaceSubQueryShuttle}
+ * does; if the condition contains several equal sub-queries, they are all
+ * rewritten at once, and therefore none of them is "remaining". */
+ private static List<RexSubQuery> otherSubQueries(RexNode condition,
RexSubQuery e) {
+ final RexUtil.SubQueryCollector collector = new
RexUtil.SubQueryCollector(true);
+ condition.accept(collector);
+ return collector.getSubQueries().stream()
+ .filter(subQuery -> !subQuery.equals(e))
+ .collect(Collectors.toList());
+ }
+
+ /** Gives {@code node} fresh correlation ids for the variables in
+ * {@code variablesSet} that it shares with {@code remainingVariables}, and
+ * returns the rewritten expression along with its new set of variables.
+ *
+ * <p>Visits the complete expression, including operands and relational
+ * expressions of nested sub-queries. Otherwise a sub-query in an operand
+ * could retain an id that its new binder no longer defines. Variables that
+ * are not shared, including those bound by an outer scope, are unchanged.
+ *
+ * <p>Rewriting the sub-query binds its variables to a new {@link Correlate}
+ * or a join of the domains, whereas the remaining sub-queries need the
+ * original join to keep defining the same variables over the joint row
+ * type. For example, in
+ *
+ * <pre>{@code
+ * SELECT e.ename
+ * FROM emp e
+ * JOIN dept d
+ * ON EXISTS (SELECT 1 FROM emp s WHERE s.mgr = e.empno)
+ * AND EXISTS (SELECT 1 FROM emp s WHERE s.ename = d.dname)
+ * }</pre>
+ *
+ * <p>both sub-queries use {@code $cor0}. Rewriting the first one renames its
+ * {@code $cor0} to a fresh id, so that {@code $cor0} is not defined twice:
by
+ * the Correlate over EMP (8 fields) and by the Join over EMP + auxiliary
field
+ * + DEPT (12 fields). */
+ private static Pair<RexNode, Set<CorrelationId>> renameSharedVariables(
+ RexNode node, Set<CorrelationId> variablesSet,
+ Set<CorrelationId> remainingVariables, RelBuilder builder) {
+ final Set<CorrelationId> sharedVariables = intersect(variablesSet,
remainingVariables);
+ if (sharedVariables.isEmpty()) {
+ return Pair.of(node, variablesSet);
+ }
+ final Set<CorrelationId> newVariablesSet = new
LinkedHashSet<>(variablesSet);
+ for (CorrelationId id : sharedVariables) {
+ final CorrelationId newId = builder.getCluster().createCorrel();
+ node =
+ DeduplicateCorrelateVariables.go(builder.getRexBuilder(), newId,
+ ImmutableSet.of(id), node);
+ newVariablesSet.remove(id);
+ newVariablesSet.add(newId);
+ }
+ return Pair.of(node, newVariablesSet);
+ }
+
+ /** Creates the join that replaces {@code join} once one of the sub-queries
in
+ * its condition has been rewritten, and pushes it onto {@code builder}.
+ *
+ * <p>If sub-queries that use the join's correlation variables are left in
+ * {@code condition}, the new join must keep defining those variables,
+ * otherwise they would be treated as uncorrelated and left undefined. Their
+ * field indexes are remapped as {@code start} and {@code offset} describe
(see
+ * {@link #remapCorrelationVariables}), because the rewrite may have added
+ * auxiliary fields to one of the inputs. */
+ private static void pushJoin(RelBuilder builder, Join join, RelNode left,
+ RelNode right, RexNode condition, Set<CorrelationId> remainingVariables,
+ int start, int offset) {
+ if (remainingVariables.isEmpty()) {
+ builder.push(left);
+ builder.push(right);
+ builder.join(join.getJoinType(), condition);
+ return;
+ }
+
+ // The correlation row type comes from createJoinType rather than from the
+ // join's output, because the ON condition sees both inputs even for the
+ // SEMI and ANTI joins that do not output the right one.
+ final RelDataType correlRowType =
+ SqlValidatorUtil.createJoinType(builder.getTypeFactory(),
+ left.getRowType(), right.getRowType(), null,
+ join.getSystemFieldList());
+ final RexNode newCondition =
+ remapCorrelationVariables(condition, remainingVariables, correlRowType,
+ start, offset, builder.getRexBuilder());
+
+ // Not RelBuilder#join: given a non-empty variablesSet it may create a
+ // Correlate (and a Correlate has no room for a condition that references
+ // both inputs). Not Join#copy either, because that would keep the
original,
+ // now stale, variablesSet.
+ builder.push(
+ LogicalJoin.create(left, right, join.getHints(), newCondition,
Review Comment:
Should we use the configured JoinFactory here as well?
--
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]
