/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.drill.exec.planner.logical;


import org.apache.calcite.plan.RelOptPlanner;
import org.apache.calcite.plan.RelOptRule;
import org.apache.calcite.plan.RelOptRuleCall;
import org.apache.calcite.plan.RelOptRuleOperand;
import org.apache.calcite.plan.RelOptUtil;
import org.apache.calcite.plan.hep.HepRelVertex;
import org.apache.calcite.plan.volcano.RelSubset;
import org.apache.calcite.rel.RelNode;
import org.apache.calcite.rel.core.JoinRelType;
import org.apache.calcite.rel.metadata.RelMetadataQuery;
import org.apache.calcite.rex.RexCall;
import org.apache.calcite.rex.RexInputRef;
import org.apache.calcite.rex.RexNode;
import org.apache.calcite.sql.SqlKind;
import org.apache.calcite.util.Pair;
import org.apache.drill.exec.planner.logical.RowKeyJoinCallContext.RowKey;
import org.apache.drill.exec.physical.base.DbGroupScan;
import org.apache.drill.exec.planner.index.rules.MatchFunction;
import org.apache.drill.exec.planner.physical.PrelUtil;

import com.google.common.collect.ImmutableList;

import java.util.ArrayList;
import java.util.List;

/**
 * This rule implements the run-time filter pushdown via the rowkey join for queries with row-key filters. Row-key
 * filters are filters on primary-keys which appears in database groupscans {@link DbGroupScan}.
 *
 * Consider the following query:
 * SELECT L.LINEITEM_ID FROM LINEITEM L WHERE L._ID IN (SELECT O.LID FROM ORDERS O WHERE O.ORDER_DATE > '2019-01-01');
 * With this rule the logical plan on the left would transform to the logical plan on the right:
 * Project                                                Project
 *   Join (L._ID = O.LID)                                   RowKeyJoin (L._ID = O.LID)
 *     LineItem L                                ====>>       Lineitem L
 *     Filter (ORDER_DATE > '2019-01-01')                     Filter (ORDER_DATE > '2019-01-01')
 *       Orders O                                               Orders O
 *
 * During physical planning, the plan on the left would end up with e.g. HashJoin whereas the transformed plan would
 * have a RowKeyJoin along with a Restricted GroupScan instead.
 * Project                                                Project
 *   HashJoin (L._ID = O.LID)                               RowKeyJoin (L._ID = O.LID)
 *     Scan (LineItem L)                                      RestrictedScan (Lineitem L)
 *     Filter (ORDER_DATE > '2019-01-01')                     Filter (ORDER_DATE > '2019-01-01')
 *       Scan (Orders O)                                        Scan (Orders O)
 *
 * The row-key join pushes the `row-keys` for rows satisfying the filter into the Lineitem restricted groupscan. So
 * we only fetch these rowkeys instead of fetching all rows into the Hash Join.
 */
public class DrillPushRowKeyJoinToScanRule extends RelOptRule {
  static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(DrillPushRowKeyJoinToScanRule.class);
  final public MatchFunction match;

  private DrillPushRowKeyJoinToScanRule(RelOptRuleOperand operand, String description, MatchFunction match) {
    super(operand, description);
    this.match = match;
  }

  @Override
  public boolean matches(RelOptRuleCall call) {
    return match.match(call);
  }

  @Override
  public void onMatch(RelOptRuleCall call) {
    doOnMatch((RowKeyJoinCallContext) match.onMatch(call));
  }

  public static DrillPushRowKeyJoinToScanRule JOIN = new DrillPushRowKeyJoinToScanRule(
      RelOptHelper.any(DrillJoin.class), "DrillPushRowKeyJoinToScanRule_Join", new MatchRelJ());

  public static class MatchRelJ implements MatchFunction<RowKeyJoinCallContext> {
    /*
     * Returns the rels matching the specified sequence relSequence. The match is executed
     * beginning from startingRel. An example of such a sequence is Join->Filter->Project->Scan
     */
    private List<RelNode> findRelSequence(Class[] relSequence, RelNode startingRel) {
      List<RelNode> matchingRels = new ArrayList<>();
      findRelSequenceInternal(relSequence, 0, startingRel, matchingRels);
      return matchingRels;
    }
    /*
     * Recursively match until the sequence is satisfied. Otherwise return. Recurse down intermediate nodes
     * such as RelSubset/HepRelVertex.
     */
    private void findRelSequenceInternal(Class[] classes, int idx, RelNode rel, List<RelNode> matchingRels) {
      if (rel instanceof HepRelVertex) {
        findRelSequenceInternal(classes, idx, ((HepRelVertex) rel).getCurrentRel(), matchingRels);
      } else if (rel instanceof RelSubset) {
        if (((RelSubset) rel).getBest() != null) {
          findRelSequenceInternal(classes, idx, ((RelSubset) rel).getBest(), matchingRels);
        } else {
          findRelSequenceInternal(classes, idx, ((RelSubset) rel).getOriginal(), matchingRels);
        }
      } else if (classes[idx].isInstance(rel)) {
        matchingRels.add(rel);
        if (idx + 1 < classes.length && rel.getInputs().size() > 0) {
          findRelSequenceInternal(classes, idx + 1, rel.getInput(0), matchingRels);
        }
      } else {
        if (logger.isDebugEnabled()) {
          String sequence, matchingSequence;
          StringBuffer sb = new StringBuffer();
          for (int i = 0; i < classes.length; i++) {
            if (i == classes.length - 1) {
              sb.append(classes[i].getCanonicalName().toString());
            } else {
              sb.append(classes[i].getCanonicalName().toString() + "->");
            }
          }
          sequence = sb.toString();
          sb.delete(0, sb.length());
          for (int i = 0; i < matchingRels.size(); i++) {
            if (i == matchingRels.size() - 1) {
              sb.append(matchingRels.get(i).getClass().getCanonicalName().toString());
            } else {
              sb.append(matchingRels.get(i).getClass().getCanonicalName().toString() + "->");
            }
          }
          matchingSequence = sb.toString();
          logger.debug("FindRelSequence: ABORT: Unexpected Rel={}, After={}, CurSeq={}",
              rel.getClass().getCanonicalName().toString(), matchingSequence, sequence);
        }
        matchingRels.clear();
      }
    }

    /*
     * Generate the rowkeyjoin call context. This context is useful when generating the transformed
     * plan nodes. It tries to identify some RelNode sequences e.g. Filter-Project-Scan and generates
     * the context based on the identified sequence.
     */
    private RowKeyJoinCallContext generateContext(RelOptRuleCall call, DrillJoin joinRel,
      RelNode joinChildRel, RowKey rowKeyLoc, int rowKeyPos, boolean swapInputs) {
      List<RelNode> matchingRels;
      // Sequence of rels (PFPS, FPS, PS, FS, S) matched for this rule
      Class[] PFPS = new Class[] {DrillProjectRel.class, DrillFilterRel.class, DrillProjectRel.class, DrillScanRel.class};
      Class[] FPS = new Class[] {DrillFilterRel.class, DrillProjectRel.class, DrillScanRel.class};
      Class[] PS = new Class[] {DrillProjectRel.class, DrillScanRel.class};
      Class[] FS = new Class[] {DrillFilterRel.class, DrillScanRel.class};
      Class[] S = new Class[] {DrillScanRel.class};
      logger.debug("GenerateContext(): Primary-key: Side={}, RowTypePos={}, SwapInputs={}",
          rowKeyLoc.name(), rowKeyPos, swapInputs);
      matchingRels = findRelSequence(PFPS, joinChildRel);
      if (matchingRels.size() > 0) {
        logger.debug("Matched rel sequence : Project->Filter->Project->Scan");
        return new RowKeyJoinCallContext(call, rowKeyLoc, rowKeyPos, swapInputs, joinRel,
            (DrillProjectRel) matchingRels.get(0), (DrillFilterRel) matchingRels.get(1),
            (DrillProjectRel) matchingRels.get(2), (DrillScanRel) matchingRels.get(3));
      }
      matchingRels = findRelSequence(FPS, joinChildRel);
      if (matchingRels.size() > 0) {
        logger.debug("Matched rel sequence : Filter->Project->Scan");
        return new RowKeyJoinCallContext(call, rowKeyLoc, rowKeyPos, swapInputs, joinRel,
            null, (DrillFilterRel) matchingRels.get(0), (DrillProjectRel) matchingRels.get(1),
            (DrillScanRel) matchingRels.get(2));
      }
      matchingRels = findRelSequence(PS, joinChildRel);
      if (matchingRels.size() > 0) {
        logger.debug("Matched rel sequence : Project->Scan");
        return new RowKeyJoinCallContext(call, rowKeyLoc, rowKeyPos, swapInputs, joinRel, null,
            null, (DrillProjectRel) matchingRels.get(0), (DrillScanRel) matchingRels.get(1));
      }
      matchingRels = findRelSequence(FS, joinChildRel);
      if (matchingRels.size() > 0) {
        logger.debug("Matched rel sequence : Filter->Scan");
        return new RowKeyJoinCallContext(call, rowKeyLoc, rowKeyPos, swapInputs, joinRel, null,
            (DrillFilterRel) matchingRels.get(0), null, (DrillScanRel) matchingRels.get(1));
      }
      matchingRels = findRelSequence(S, joinChildRel);
      if (matchingRels.size() > 0) {
        logger.debug("Matched rel sequence : Scan");
        return new RowKeyJoinCallContext(call, rowKeyLoc, rowKeyPos, swapInputs, joinRel, null, null,
            null, (DrillScanRel) matchingRels.get(0));
      }
      logger.debug("Matched rel sequence : None");
      return new RowKeyJoinCallContext(call, RowKey.NONE, -1, false, null, null, null, null, null);
    }

    @Override
    public boolean match(RelOptRuleCall call) {
      DrillJoin joinRel = call.rel(0);
      //Perform validity checks
      logger.debug("DrillPushRowKeyJoinToScanRule begin()");
      return canPushRowKeyJoinToScan(joinRel, call.getPlanner()).left;
    }

    @Override
    public RowKeyJoinCallContext onMatch(RelOptRuleCall call) {
      DrillJoin joinRel = call.rel(0);
      /*
       * Find which side of the join (left/right) has the primary-key column. Then find which sequence of rels
       * is present on that side of the join. We will need this sequence to correctly transform the left
       * side of the join.
       */
      Pair<Boolean, Pair<RowKey, Integer>> res = canPushRowKeyJoinToScan(joinRel, call.getPlanner());
      if (res.left) {
        if (res.right.left == RowKey.LEFT) {
          return generateContext(call, joinRel, joinRel.getLeft(), res.right.left, res.right.right, false);
        } else if (res.right.left == RowKey.RIGHT) {
          // If the primary-key column is present on the right, swapping of inputs is required. Find out if possible!
          if (canSwapJoinInputs(joinRel, res.right.left)) {
            return generateContext(call, joinRel, joinRel.getRight(), res.right.left, res.right.right, true);
          }
        } else if (res.right.left == RowKey.BOTH) {
          // Create row key join without swapping inputs, since either side of the join is eligible.
          return generateContext(call, joinRel, joinRel.getLeft(), res.right.left, res.right.right, false);
        }
      }
      return new RowKeyJoinCallContext(call, RowKey.NONE, -1, false, null, null, null, null, null);
    }
  }

  /* Assumption : Only the non-rowkey side needs to be checked. The row-key side does not have
   * any blocking operators for the transformation to work
   */
  private static boolean canSwapJoinInputs(DrillJoin joinRel, RowKey rowKeyLocation) {
    // We cannot swap the join inputs if the join is a semi-join. We determine it indirectly, by
    // checking for the presence of a aggregating Aggregate Rel (computes aggregates e.g. sum).
    if (rowKeyLocation == RowKey.LEFT
        || rowKeyLocation == RowKey.BOTH) {
      return canSwapJoinInputsInternal(joinRel.getRight());
    } else if (rowKeyLocation == RowKey.RIGHT) {
      // If the rowkey occurs on the right side, don't swap since it can potentially cause
      // wrong results unless we make additional changes to fix-up column ordinals for the
      // join condition as well as the parent/ancestors of the Join.

      // return canSwapJoinInputsInternal(joinRel.getLeft());
      return false;
    }
    return false;
  }

  /* Recurse down to find an aggregate (DrillAggRel). For semi-joins Calcite adds an aggregate
   * without any agg expressions.
   */
  private static boolean canSwapJoinInputsInternal(RelNode rel) {
    if (rel instanceof DrillAggregateRel &&
        ((DrillAggregateRel) rel).getAggCallList().size() > 0) {
      return false;
    } else if (rel instanceof HepRelVertex) {
      return canSwapJoinInputsInternal(((HepRelVertex) rel).getCurrentRel());
    } else if (rel instanceof RelSubset) {
      if (((RelSubset) rel).getBest() != null) {
        return canSwapJoinInputsInternal(((RelSubset) rel).getBest());
      } else {
        return canSwapJoinInputsInternal(((RelSubset) rel).getOriginal());
      }
    } else {
      for (RelNode child : rel.getInputs()) {
        if (!canSwapJoinInputsInternal(child)) {
          return false;
        }
      }
    }
    return true;
  }

  /*
   * Returns whether the join condition can be pushed (via rowkeyjoin mechanism). It returns true/false alongwith
   * whether the rowkey is present on the left/right side of the join and its 0-based index in the projection of that
   * side.
   */
  private static Pair<Boolean, Pair<RowKey, Integer>> canPushRowKeyJoinToScan(DrillJoin joinRel, RelOptPlanner planner) {
    RowKey rowKeyLoc = RowKey.NONE;
    logger.debug("canPushRowKeyJoinToScan(): Check: Rel={}", joinRel);

    if (joinRel instanceof RowKeyJoinRel) {
      logger.debug("SKIP: Join is a RowKeyJoin");
      return Pair.of(false, Pair.of(rowKeyLoc, -1));
    }

    if (joinRel.getJoinType() != JoinRelType.INNER) {
      logger.debug("SKIP: JoinType={} - NOT an INNER join", joinRel.getJoinType());
      return Pair.of(false, Pair.of(rowKeyLoc, -1));
    }

    // Single column equality condition
    if (joinRel.getCondition().getKind() != SqlKind.EQUALS
        || joinRel.getLeftKeys().size() != 1
        || joinRel.getRightKeys().size() != 1) {
      logger.debug("SKIP: #LeftKeys={}, #RightKeys={} - NOT single predicate join condition",
          joinRel.getLeftKeys().size(), joinRel.getRightKeys().size());
      return Pair.of(false, Pair.of(rowKeyLoc, -1));
    }

    // Join condition is of type primary-key = Col
    boolean hasLeftRowKeyCol = false;
    boolean hasRightRowKeyCol = false;
    int leftRowKeyPos = -1;
    int rightRowKeyPos = -1;
    if (joinRel.getCondition() instanceof RexCall) {
      for (RexNode op : ((RexCall) joinRel.getCondition()).getOperands()) {
        // Only support rowkey column (no expressions involving rowkey column)
        if (op instanceof RexInputRef) {
          //Check the left/right sides of the join to find the primary-key column
          int pos = ((RexInputRef)op).getIndex();
          if (pos < joinRel.getLeft().getRowType().getFieldList().size()) {
            if (isRowKeyColumn(((RexInputRef) op).getIndex(), joinRel.getLeft())) {
              logger.debug("FOUND Primary-key: Side=LEFT, RowType={}", joinRel.getLeft().getRowType());
              hasLeftRowKeyCol = true;
              leftRowKeyPos = pos;
              break;
            }
          } else {
            if (isRowKeyColumn(pos - joinRel.getLeft().getRowType().getFieldList().size(), joinRel.getRight())) {
              logger.debug("FOUND Primary-key: Side=RIGHT, RowType={}", joinRel.getRight().getRowType());
              hasRightRowKeyCol = true;
              rightRowKeyPos = pos;
              break;
            }
          }
        }
      }
    }
    if (!hasLeftRowKeyCol && !hasRightRowKeyCol) {
      logger.debug("SKIP: Primary-key = column condition NOT found");
      return Pair.of(false, Pair.of(rowKeyLoc, -1));
    }
    /* Get the scan rel on left/right side of the join (at least one of them should be non-null for us
     * to proceed). This would be the side with the primary-key column and would be later transformed to restricted
     * group scan.
     */
    RelNode leftScan = getValidJoinInput(joinRel.getLeft());
    RelNode rightScan = getValidJoinInput(joinRel.getRight());

    if (leftScan == null && rightScan == null) {
      logger.debug("SKIP: Blocking operators between join and scans");
      return Pair.of(false, Pair.of(rowKeyLoc, -1));
    }
    // Only valid if the side with the primary-key column doesn't not have any blocking operations e.g. aggregates
    if (leftScan != null && hasLeftRowKeyCol) {
      rowKeyLoc = RowKey.LEFT;
    }
    if (rightScan != null && hasRightRowKeyCol) {
      if (rowKeyLoc == RowKey.LEFT) {
        rowKeyLoc = RowKey.BOTH;
      } else {
        rowKeyLoc = RowKey.RIGHT;
      }
    }
    // Heuristic : only generate such plans if selectivity less than RKJ conversion selectivity threshold.
    // Rowkey join plans do random scans, hence are expensive. Since this transformation takes place in
    // the HEP planner, it is not costed. Hence, the heuristic to potentially prevent an expensive plan!
    RelMetadataQuery mq = RelMetadataQuery.instance();
    double ncSel = PrelUtil.getPlannerSettings(planner).getRowKeyJoinConversionSelThreshold();
    double sel;
    if (rowKeyLoc == RowKey.NONE) {
      return Pair.of(false, Pair.of(rowKeyLoc, -1));
    } else if (rowKeyLoc == RowKey.LEFT) {
      sel = computeSelectivity(joinRel.getRight().estimateRowCount(mq), leftScan.estimateRowCount(mq));
      if (sel > ncSel) {
        logger.debug("SKIP: SEL= {}/{} = {}\\%, THRESHOLD={}\\%",
            joinRel.getRight().estimateRowCount(mq), leftScan.estimateRowCount(mq), sel*100.0, ncSel*100.0);
        return Pair.of(false, Pair.of(rowKeyLoc, -1));
      }
    } else {
      sel = computeSelectivity(joinRel.getLeft().estimateRowCount(mq), rightScan.estimateRowCount(mq));
      if (sel > ncSel) {
        logger.debug("SKIP: SEL= {}/{} = {}\\%, THRESHOLD={}\\%",
            joinRel.getLeft().estimateRowCount(mq), rightScan.estimateRowCount(mq), sel*100.0, ncSel*100.0);
        return Pair.of(false, Pair.of(rowKeyLoc, -1));
      }
    }
    int rowKeyPos = rowKeyLoc == RowKey.RIGHT ? rightRowKeyPos : leftRowKeyPos;
    logger.info("FOUND Primary-key: Side={}, RowTypePos={}, Sel={}, Threshold={}",
        rowKeyLoc.name(), rowKeyPos, sel, ncSel);
    return Pair.of(true, Pair.of(rowKeyLoc, rowKeyPos));
  }

  /*
   * Computes the selectivity given the number of rows selected from the total rows
   */
  private static double computeSelectivity(double selectRows, double totalRows) {
    if (totalRows <= 0) {
      return 1.0;
    }
    return Math.min(1.0, Math.max(0.0, selectRows/totalRows));
  }

  /* Finds the scan rel underlying the given rel. No blocking operators should
   * be present in between. Currently, the rowkeyjoin operator cannot send rowkeys
   * across major fragment boundaries. The presence of blocking operators can
   * lead to creation of a fragment boundary, hence the limitation. Once, we can
   * send rowkeys across fragment boundaries, we can remove this restriction.
   */
  public static RelNode getValidJoinInput(RelNode rel) {
    if (rel instanceof DrillScanRel) {
      return rel;
    } else if (rel instanceof DrillProjectRel
        || rel instanceof DrillFilterRel
        || rel instanceof DrillLimitRel) {
      for (RelNode child : rel.getInputs()) {
        RelNode tgt = getValidJoinInput(child);
        if (tgt != null) {
          return tgt;
        }
      }
    } else if (rel instanceof HepRelVertex) {
      return getValidJoinInput(((HepRelVertex) rel).getCurrentRel());
    } else if (rel instanceof RelSubset) {
      if (((RelSubset) rel).getBest() != null) {
        return getValidJoinInput(((RelSubset) rel).getBest());
      } else {
        return getValidJoinInput(((RelSubset) rel).getOriginal());
      }
    }
    return null;
  }

  /* Finds whether the given column reference is for the rowkey col(also known as primary-key col).
   * We need to recurse down the operators looking at their references down to the scan
   * to figure out whether the reference is a rowkey col. Projections can rearrange the
   * incoming columns. We also need to handle HepRelVertex/RelSubset while handling the rels.
   */
  private static boolean isRowKeyColumn(int index, RelNode rel) {
    RelNode curRel = rel;
    int curIndex = index;
    while (curRel != null && !(curRel instanceof DrillScanRel)) {
      logger.debug("IsRowKeyColumn: Rel={}, RowTypePos={}, RowType={}", curRel.toString(), curIndex,
          curRel.getRowType().toString());
      if (curRel instanceof HepRelVertex) {
        curRel = ((HepRelVertex) curRel).getCurrentRel();
      } else if (curRel instanceof RelSubset) {
        if (((RelSubset) curRel).getBest() != null) {
          curRel = ((RelSubset) curRel).getBest();
        } else {
          curRel = ((RelSubset) curRel).getOriginal();
        }
      } else {
        RelNode child = null;
        // For multi-input parent rels, found out the 0-based index in the child rel,
        // before recursing down that child rel.
        for (RelNode input : curRel.getInputs()) {
          if (input.getRowType().getFieldList().size() <= curIndex) {
            curIndex -= input.getRowType().getFieldList().size();
          } else {
            child = input;
            break;
          }
        }
        curRel = child;
      }
      // If no exprs present in projection the column index remains the same in the child.
      // Otherwise, the column index is the `RexInputRef` index.
      if (curRel instanceof DrillProjectRel) {
        DrillProjectRel projectRel = (DrillProjectRel) curRel;
        List<RexNode> childExprs = projectRel.getProjects();
        if (childExprs != null && childExprs.size() > 0) {
          if (childExprs.get(curIndex) instanceof RexInputRef) {
            curIndex = ((RexInputRef) childExprs.get(curIndex)).getIndex();
          } else {
            // Currently do not support expressions on rowkey col. So if an expr is present,
            // return false
            logger.debug("IsRowKeyColumn: ABORT: Primary-key EXPR$={}", childExprs.get(curIndex).toString());
            return false;
          }
        }
      }
    }
    logger.debug("IsRowKeyColumn:Primary-key Col={} ",
        curRel != null ? curRel.getRowType().getFieldNames().get(curIndex) : "??");
    // Get the primary-key col name from the scan and match with the column being referenced.
    if (curRel != null && curRel instanceof DrillScanRel) {
      if (((DrillScanRel) curRel).getGroupScan() instanceof DbGroupScan) {
        DbGroupScan dbGroupScan = (DbGroupScan) ((DrillScanRel) curRel).getGroupScan();
        String rowKeyName = dbGroupScan.getRowKeyName();
        DbGroupScan restrictedGroupScan = dbGroupScan.getRestrictedScan(((DrillScanRel)curRel).getColumns());
        // Also verify this scan supports restricted groupscans(random seeks)
        if (restrictedGroupScan != null &&
            curRel.getRowType().getFieldNames().get(curIndex).equalsIgnoreCase(rowKeyName)) {
          logger.debug("IsRowKeyColumn: FOUND: Rel={}, RowTypePos={}, RowType={}",
              curRel.toString(), curIndex, curRel.getRowType().toString());
          return true;
        }
      }
    }
    logger.debug("IsRowKeyColumn: NOT FOUND");
    return false;
  }

  protected void doOnMatch(RowKeyJoinCallContext rkjCallContext) {
    if (rkjCallContext.getRowKeyLocation() != RowKey.NONE) {
      doOnMatch(rkjCallContext.getCall(), rkjCallContext.getRowKeyPosition(), rkjCallContext.mustSwapInputs(),
          rkjCallContext.getJoinRel(), rkjCallContext.getUpperProjectRel(), rkjCallContext.getFilterRel(),
          rkjCallContext.getLowerProjectRel(), rkjCallContext.getScanRel());
    }
  }

  private void doOnMatch(RelOptRuleCall call, int rowKeyPosition, boolean swapInputs, DrillJoin joinRel,
      DrillProjectRel upperProjectRel, DrillFilterRel filterRel, DrillProjectRel lowerProjectRel, DrillScanRel scanRel) {
    // Swap the inputs, when necessary (i.e. when the primary-key col is on the right-side of the join)
    logger.debug("Transforming: Swapping of join inputs is required!");
    RelNode right = swapInputs ? joinRel.getLeft() : joinRel.getRight();
    // The join condition is primary-key = COL similarly to PK-FK relationship in relational DBs
    // where primary-key is PK and COL is FK
    List<Integer> leftJoinKeys = ImmutableList.of(rowKeyPosition);
    List<Integer> rightJoinKeys = swapInputs ? joinRel.getLeftKeys() : joinRel.getRightKeys();
    // Create restricted group scan for scanRel and reconstruct the left side of the join.
    DbGroupScan restrictedGroupScan = ((DbGroupScan)scanRel.getGroupScan()).getRestrictedScan(
        scanRel.getColumns());
    RelNode leftRel =  new DrillScanRel(scanRel.getCluster(), scanRel.getTraitSet(), scanRel.getTable(),
        restrictedGroupScan, scanRel.getRowType(), scanRel.getColumns(), scanRel.partitionFilterPushdown());
    // Transform the project/filter rels if present
    if (lowerProjectRel != null) {
      leftRel = lowerProjectRel.copy(lowerProjectRel.getTraitSet(), ImmutableList.of(leftRel));
    }
    if (filterRel != null) {
      leftRel = filterRel.copy(filterRel.getTraitSet(), leftRel, filterRel.getCondition());
    }
    if (upperProjectRel != null) {
      leftRel = upperProjectRel.copy(upperProjectRel.getTraitSet(), ImmutableList.of(leftRel));
    }
    // Create the equi-join condition for the rowkey join
    RexNode joinCondition =
        RelOptUtil.createEquiJoinCondition(leftRel, leftJoinKeys,
            right, rightJoinKeys, joinRel.getCluster().getRexBuilder());
    logger.debug("Transforming: LeftKeys={}, LeftRowType={}, RightKeys={}, RightRowType={}",
        leftJoinKeys, leftRel.getRowType(), rightJoinKeys, right.getRowType());
    RowKeyJoinRel rowKeyJoin = new RowKeyJoinRel(joinRel.getCluster(), joinRel.getTraitSet(), leftRel, right,
        joinCondition, joinRel.getJoinType(), joinRel instanceof DrillSemiJoinRel);
    logger.info("Transforming: SUCCESS: Register runtime filter pushdown plan (rowkeyjoin)");
    call.transformTo(rowKeyJoin);
  }
}
