import org.eigenbase.rex.RexNode; //导入依赖的package包/类
private ColumnRowType buildColumnRowType() {
    List<TblColRef> columns = new ArrayList<TblColRef>();
    List<Set<TblColRef>> sourceColumns = new ArrayList<Set<TblColRef>>();
    OLAPRel olapChild = (OLAPRel) getChild();
    ColumnRowType inputColumnRowType = olapChild.getColumnRowType();
    for (int i = 0; i < this.rewriteProjects.size(); i++) {
        RexNode rex = this.rewriteProjects.get(i);
        RelDataTypeField columnField = this.rowType.getFieldList().get(i);
        String fieldName = columnField.getName();
        Set<TblColRef> sourceCollector = new HashSet<TblColRef>();
        TblColRef column = translateRexNode(rex, inputColumnRowType, fieldName, sourceCollector);
        columns.add(column);
        sourceColumns.add(sourceCollector);
    }
    return new ColumnRowType(columns, sourceColumns);
}
 

import org.eigenbase.rex.RexNode; //导入依赖的package包/类
private TblColRef translateRexNode(RexNode rexNode, ColumnRowType inputColumnRowType, String fieldName, Set<TblColRef> sourceCollector) {
    TblColRef column = null;
    if (rexNode instanceof RexInputRef) {
        RexInputRef inputRef = (RexInputRef) rexNode;
        column = translateRexInputRef(inputRef, inputColumnRowType, fieldName, sourceCollector);
    } else if (rexNode instanceof RexLiteral) {
        RexLiteral literal = (RexLiteral) rexNode;
        column = translateRexLiteral(literal);
    } else if (rexNode instanceof RexCall) {
        RexCall call = (RexCall) rexNode;
        column = translateRexCall(call, inputColumnRowType, fieldName, sourceCollector);
    } else {
        throw new IllegalStateException("Unsupport RexNode " + rexNode);
    }
    return column;
}
 

import org.eigenbase.rex.RexNode; //导入依赖的package包/类
private void translateJoinColumn(RexCall condition, Map<TblColRef, TblColRef> joinColumns) {
    SqlKind kind = condition.getOperator().getKind();
    if (kind == SqlKind.AND) {
        for (RexNode operand : condition.getOperands()) {
            RexCall subCond = (RexCall) operand;
            translateJoinColumn(subCond, joinColumns);
        }
    } else if (kind == SqlKind.EQUALS) {
        List<RexNode> operands = condition.getOperands();
        RexInputRef op0 = (RexInputRef) operands.get(0);
        TblColRef col0 = columnRowType.getColumnByIndex(op0.getIndex());
        RexInputRef op1 = (RexInputRef) operands.get(1);
        TblColRef col1 = columnRowType.getColumnByIndex(op1.getIndex());
        joinColumns.put(col0, col1);
    }
}
 

import org.eigenbase.rex.RexNode; //导入依赖的package包/类
public static void getExprVarsFromRexCall(RexNode cond, List<ExprVar> result) {
	if(cond instanceof RexCall) {
		RexCall call = (RexCall) cond;
		for(int i = 0; i < call.getOperands().size(); i++) {
			getExprVarsFromRexCall(call.getOperands().get(i), result);
		}
		return;
	}
	// assignable variable
	if(cond instanceof RexInputRef) {
		RexInputRef ref = (RexInputRef) cond;
		for (ExprVar expr : result) {
			if(expr.varName.equals(ref.getName())) {
				// variable already created;
				return;
			}
		}
		ExprVar e = new ExprVar();
		e.varName = ref.getName();
		e.positionInRecord = ref.getIndex();
		e.type = getTypeClass(ref.getType());
		result.add(e);
		return;
	}
}
 

import org.eigenbase.rex.RexNode; //导入依赖的package包/类
@Override
public Operator getStratosphereOperator() {
	Operator inputOp = StratosphereRelUtils.openSingleInputOperator(getInputs());
	final RexNode cond = getCondition();

	Filter f = new Filter();
	f.setCondition(cond);
	f.setRexBuilder(getCluster().getRexBuilder());
	f.prepareShipping(getInput(0).getRowType());

       //final RexExecutorImpl executor = new RexExecutorImpl(null);


	Operator filter = MapOperator.builder(new StratosphereSqlFilterMapOperator(f) )
								.input(inputOp)
								.name(condition.toString())
								.build();
	return filter;
}
 

import org.eigenbase.rex.RexNode; //导入依赖的package包/类
/**
 * Create the list of fields.
 * @param rowType
 */
public void prepareShipping(RelDataType rowType) {
	Preconditions.checkNotNull(condition);
	Preconditions.checkNotNull(rexBuilder);

	StratosphereRexUtils.GetInputRefVisitor replaceInputRefsByExternalInputRefsVisitor = new StratosphereRexUtils.GetInputRefVisitor();
	condition.accept(replaceInputRefsByExternalInputRefsVisitor);

	final ImmutableList<RexNode> localExps = ImmutableList.of(condition);

	fields = new HashSet<StratosphereRexUtils.ProjectionFieldProperties>();
	int pos = 0;
	for(Pair<Integer, RelDataType> rexInput : replaceInputRefsByExternalInputRefsVisitor.getInputPosAndType() ) {
		StratosphereRexUtils.ProjectionFieldProperties field = new StratosphereRexUtils.ProjectionFieldProperties();
		field.fieldIndex = pos++;
		field.positionInInput = rexInput.getKey();
		field.inFieldType = StratosphereRelUtils.getTypeClass(rexInput.getValue());
		field.name = condition.toString();
		fields.add(field);
	}
	final RexExecutorImpl executor = new RexExecutorImpl(null);
	RexExecutable executable = executor.getExecutable(rexBuilder, localExps, rowType);
	System.err.println("Code: "+executable.getSource());
	this.source = executable.getSource();
}
 

import org.eigenbase.rex.RexNode; //导入依赖的package包/类
private int[] getProjectFields(List<RexNode> exps) {
    final int[] fields = new int[exps.size()];
    for (int i = 0; i < exps.size(); i++) {
        final RexNode exp = exps.get(i);
        if (exp instanceof RexInputRef) {
            fields[i] = ((RexInputRef) exp).getIndex();
        } else {
            return null; // not a simple projection
        }
    }
    return fields;
}
 

import org.eigenbase.rex.RexNode; //导入依赖的package包/类
private int[] getProjectFields(List<RexNode> exps) {
  final int[] fields = new int[exps.size()];
  for (int i = 0; i < exps.size(); i++) {
    final RexNode exp = exps.get(i);
    if (exp instanceof RexInputRef) {
      fields[i] = ((RexInputRef) exp).getIndex();
    } else {
      return null; // not a simple projection
    }
  }
  return fields;
}
 

import org.eigenbase.rex.RexNode; //导入依赖的package包/类
public OLAPProjectRel(RelOptCluster cluster, RelTraitSet traitSet, RelNode child, List<RexNode> exps, RelDataType rowType, int flags) {
    super(cluster, traitSet, child, exps, rowType, flags);
    Preconditions.checkArgument(getConvention() == OLAPRel.CONVENTION);
    Preconditions.checkArgument(child.getConvention() == OLAPRel.CONVENTION);
    this.rewriteProjects = exps;
    this.hasJoin = false;
    this.afterJoin = false;
    this.rowType = getRowType();
}
 

import org.eigenbase.rex.RexNode; //导入依赖的package包/类
public OLAPLimitRel(RelOptCluster cluster, RelTraitSet traitSet, RelNode child, RexNode offset, RexNode fetch) {
    super(cluster, traitSet, child);
    Preconditions.checkArgument(getConvention() == CONVENTION);
    Preconditions.checkArgument(getConvention() == child.getConvention());
    this.localOffset = offset;
    this.localFetch = fetch;
}
 

import org.eigenbase.rex.RexNode; //导入依赖的package包/类
public OLAPJoinRel(RelOptCluster cluster, RelTraitSet traits, RelNode left, RelNode right, //
        RexNode condition, ImmutableIntList leftKeys, ImmutableIntList rightKeys, //
        JoinRelType joinType, Set<String> variablesStopped) throws InvalidRelException {
    super(cluster, traits, left, right, condition, leftKeys, rightKeys, joinType, variablesStopped);
    Preconditions.checkArgument(getConvention() == OLAPRel.CONVENTION);
    this.rowType = getRowType();
    this.isTopJoin = false;
}
 

import org.eigenbase.rex.RexNode; //导入依赖的package包/类
@Override
public EnumerableJoinRel copy(RelTraitSet traitSet, RexNode conditionExpr, RelNode left, RelNode right, //
        JoinRelType joinType, boolean semiJoinDone) {

    final JoinInfo joinInfo = JoinInfo.of(left, right, condition);
    assert joinInfo.isEqui();
    try {
        return new OLAPJoinRel(getCluster(), traitSet, left, right, condition, joinInfo.leftKeys, joinInfo.rightKeys, joinType, variablesStopped);
    } catch (InvalidRelException e) {
        // Semantic error not possible. Must be a bug. Convert to
        // internal error.
        throw new AssertionError(e);
    }
}
 

import org.eigenbase.rex.RexNode; //导入依赖的package包/类
public Class<? extends Value>[] getFields() {
	Class<? extends Value>[] fields = new Class[this.exps.size()];
	Iterator<RexNode> it = exps.iterator();
	int i = 0;
	while(it.hasNext()) {
		RexInputRef inputRef = (RexInputRef) it.next();
		fields[i++] = StratosphereRelUtils.getTypeClass(inputRef.getType());
	}
	return fields;
}
 

import org.eigenbase.rex.RexNode; //导入依赖的package包/类
public static String convertRexCallToJexlExpr(RexNode c) {
	StringBuffer sb = new StringBuffer();
	if(c instanceof RexCall) {
		RexCall call = (RexCall) c;
		sb.append("(");
		for(int i = 0; i < call.getOperands().size(); i++) {
			sb.append( convertRexCallToJexlExpr(call.getOperands().get(i)));
			if(i+1 <call.getOperands().size()) {
				switch(c.getKind()) {
					case AND:
						sb.append(" && ");
						break;
					case OR:
						sb.append(" || ");
						break;
					case EQUALS:
						sb.append(" == ");
						break;
					case LESS_THAN:
						sb.append(" < ");
						break;
					default:
						throw new RuntimeException("Unknown kind "+c.getKind());
				}
			}
		}
		sb.append(")");
	}
	// assignable variable
	if(c instanceof RexInputRef) {
		RexInputRef ref = (RexInputRef) c;
		return ref.getName();
	}
	if(c instanceof RexLiteral) {
		RexLiteral lit = (RexLiteral) c;
		return lit.getValue().toString();
	}
	return sb.toString();
}
 

import org.eigenbase.rex.RexNode; //导入依赖的package包/类
public StratosphereSqlJoin(RelOptCluster cluster, RelTraitSet traits,
		RelNode left, RelNode right, RexNode condition,
		JoinRelType joinType, Set<String> variablesStopped) {
	super(cluster, traits, left, right, condition, joinType, variablesStopped);

	Preconditions.checkArgument(getConvention() == CONVENTION);
	Preconditions.checkArgument(joinType == JoinRelType.INNER, "Only inner joins are supported at the moment");
}
 

import org.eigenbase.rex.RexNode; //导入依赖的package包/类
@Override
public JoinRelBase copy(RelTraitSet traitSet, RexNode conditionExpr,
		RelNode left, RelNode right, JoinRelType joinType) {
	System.err.println("StratoJoin.copy()");
	return new StratosphereSqlJoin(getCluster(), traitSet, left,
			right, conditionExpr, joinType, getVariablesStopped());
}
 

import org.eigenbase.rex.RexNode; //导入依赖的package包/类
public EnumerableSort(RelOptCluster cluster, RelTraitSet traits, RelNode input, RexNode offset, RexNode fetch, DataTable<?> accumuloAccessor) {
  super(cluster, traits, input);

  this.offset = offset;
  this.fetch = fetch;
  this.accumuloAccessor = accumuloAccessor;
}
 

import org.eigenbase.rex.RexNode; //导入依赖的package包/类
@Override
public String apply(RexNode exp) {

  if (exp instanceof RexInputRef) {
    return ((RexInputRef) exp).getName();
  } else
    return exp.toString();
}
 

import org.eigenbase.rex.RexNode; //导入依赖的package包/类
@Override
public ChildVisitor visitCall(RexCall call) {
  final SqlSyntax syntax = call.getOperator().getSyntax();
  switch (syntax) {
    case Binary:
      return visitBinary(call);
    case Function:
      buf.append(call.getOperator().getName().toLowerCase()).append("(");
      for (Ord<RexNode> operand : Ord.zip(call.getOperands())) {
        buf.append(operand.i > 0 ? ", " : "");
        operand.e.accept(this);
      }
      return null;
    case Special:
      switch (call.getKind()) {
        case Cast:
          // Ignore casts. Drill is type-less.
          return call.getOperands().get(0).accept(this);
        default:
          break;
      }
      if (call.getOperator() == SqlStdOperatorTable.itemOp) {
        final RexNode left = call.getOperands().get(0);
        final int length = buf.length();
        left.accept(this);
        if (buf.length() > length) {
          // check before generating empty LHS if inputName is null
          buf.append('.');
        }
        // return buf.append(field);
        return null;
      }
      // fall through
    default:
      throw new AssertionError("todo: implement syntax " + syntax + "(" + call + ")");
  }
}
 

import org.eigenbase.rex.RexNode; //导入依赖的package包/类
public OLAPFilterRel(RelOptCluster cluster, RelTraitSet traits, RelNode child, RexNode condition) {
    super(cluster, traits, child, condition);
    Preconditions.checkArgument(getConvention() == CONVENTION);
    Preconditions.checkArgument(getConvention() == child.getConvention());
    this.rowType = getRowType();
}