
package org.apache.calcite.interpreter;

import org.apache.calcite.rel.core.Join;

import com.google.common.collect.ImmutableList;
import com.google.common.collect.Lists;

import java.util.List;

/**
 * Interpreter node that implements a
 * {@link org.apache.calcite.rel.core.Join}.
 */
public class JoinNode implements Node {
  private final Source leftSource;
  private final Source rightSource;
  private final Sink sink;
  private final Join rel;
  private final Scalar condition;
  private final Context context;

  public JoinNode(Interpreter interpreter, Join rel) {
    this.leftSource = interpreter.source(rel, 0);
    this.rightSource = interpreter.source(rel, 1);
    this.sink = interpreter.sink(rel);
    this.condition = interpreter.compile(ImmutableList.of(rel.getCondition()),
        rel.getInputs());
    this.rel = rel;
    this.context = interpreter.createContext();

  }

  public void run() throws InterruptedException {
    List<Row> rightList = null;
    final int leftCount = rel.getLeft().getRowType().getFieldCount();
    final int rightCount = rel.getRight().getRowType().getFieldCount();
    context.values = new Object[rel.getRowType().getFieldCount()];
    Row left;
    Row right;
    while ((left = leftSource.receive()) != null) {
      System.arraycopy(left.getValues(), 0, context.values, 0, leftCount);
      if (rightList == null) {
        rightList = Lists.newArrayList();
        while ((right = rightSource.receive()) != null) {
          rightList.add(right);
        }
      }
      for (Row right2 : rightList) {
        System.arraycopy(right2.getValues(), 0, context.values, leftCount,
            rightCount);
        final Boolean execute = (Boolean) condition.execute(context);
        if (execute != null && execute) {
          sink.send(Row.asCopy(context.values));
        }
      }
    }
  }
}

// End JoinNode.java
