import org.eigenbase.util.Pair; //导入依赖的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.util.Pair; //导入依赖的package包/类
public RelDataType getRowType(JavaTypeFactory typeFactory) {
    initialize();
    List<String> names = new ArrayList<String>();
    List<RelDataType> types = new ArrayList<RelDataType>();

    // iterate through FieldDefs, populating names and types
    for (FieldDef f : this.fields) {
        names.add(f.getName());

        WebFieldType fieldType = f.getType();
        RelDataType type;

        if (fieldType == null) {
            type = typeFactory.createJavaType(String.class);
        } else {
            type = fieldType.toType(typeFactory);
        }

        types.add(type);
    }

    if (names.isEmpty()) {
        names.add("line");
        types.add(typeFactory.createJavaType(String.class));
    }

    return typeFactory.createStructType(Pair.zip(names, types));
}
 

import org.eigenbase.util.Pair; //导入依赖的package包/类
public Void visitInputRef(RexInputRef inputRef) {
//			if (true) { // do I need a condition here?
//				inputRef.setExternalRef(true);
//			}
//			System.err.println("Setting input ref " + inputRef + " index "
//					+ inputRef.getIndex());
			inputPosAndType
					.add(Pair.of(inputRef.getIndex(), inputRef.getType()));
			return null;
		}
 

import org.eigenbase.util.Pair; //导入依赖的package包/类
@Override
public LoadingCache<IClass, Pair<RelDataType, List<Function<Integer, Object>>>> load(
		final RelDataTypeFactory typeFactory) throws Exception {
	return CacheBuilder.newBuilder().weakKeys()
			.build(new ClassRowTypeResolver(typeFactory));
}
 

import org.eigenbase.util.Pair; //导入依赖的package包/类
public List<Pair<Integer, RelDataType>> getInputPosAndType() {
	return inputPosAndType;
}
 

import org.eigenbase.util.Pair; //导入依赖的package包/类
@Override
public Operator getStratosphereOperator() {
	System.err.println("Preparing operator "+this.getDigest());

	// get Input
	Operator inputOp = StratosphereRelUtils.openSingleInputOperator(getInputs());
	System.err.println("Input is "+inputOp);

	final RexBuilder rexBuilder = getCluster().getRexBuilder();
	final RexExecutorImpl executor = new RexExecutorImpl(null);

	final List<RexNode> complexExps = new ArrayList<RexNode>();


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

	Set<StratosphereRexUtils.ProjectionFieldProperties> fields = new HashSet<StratosphereRexUtils.ProjectionFieldProperties>();
	int pos = 0;
	int rexpos = 0;
	for(RexNode rex : exps) {
		if(rex.getKind() != SqlKind.INPUT_REF) {
			complexExps.add(rex);
		}
		rex.accept(replaceInputRefsByExternalInputRefsVisitor);
		boolean trivialProjection = rex.getKind() == SqlKind.INPUT_REF;
		for(Pair<Integer, RelDataType> rexInput : replaceInputRefsByExternalInputRefsVisitor.getInputPosAndType() ) {
			StratosphereRexUtils.ProjectionFieldProperties field = new StratosphereRexUtils.ProjectionFieldProperties();
			field.positionInOutput = pos;
			field.fieldIndex = pos;
			field.positionInRex = rexpos;
			field.positionInInput = rexInput.getKey();
			field.inFieldType = StratosphereRelUtils.getTypeClass(rexInput.getValue());
			field.outFieldType = StratosphereRelUtils.getTypeClass( getRowType().getFieldList().get(pos).getType() );
			field.trivialProjection = trivialProjection;
			if(fields.add(field)) {
				System.err.println("adding projection field="+field+" for rex="+rex);
			} else {
				System.err.println("fields already contained "+field+" for rex="+rex);
			}
			field.name = rex.toString();
		}
		pos++;
		if(!trivialProjection) {
			rexpos++;
		}
		replaceInputRefsByExternalInputRefsVisitor.resetInputList();
	}

	RexExecutable executable = executor.getExecutable(rexBuilder, complexExps, getInput(0).getRowType() );
	System.err.println("Code="+executable.getSource());
	// create MapOperator
	MapOperator proj = MapOperator	.builder(new StratosphereSqlProjectionMapOperator(executable.getFunction(),
																					fields, executable.getSource()))
									.input(inputOp)
									.name(buildName())
									.build();
	return proj;
}
 

import org.eigenbase.util.Pair; //导入依赖的package包/类
private void parseRowType(RelDataTypeFactory typeFactory) {
	// parse fields
	List<Entry<String, RelDataType>> optiqFields = new ArrayList<Entry<String,RelDataType>>();
	Schema schema = getSchema();

	for (Schema.Field field : schema.getFields() ) {
		Schema.Type type = field.schema().getType();
		Map.Entry<String, RelDataType> mapPair = null;
		switch (type) {
			case INT :
				mapPair = Pair.of(field.name(),typeFactory.createSqlType(SqlTypeName.INTEGER));
				break;
			case BOOLEAN :
				mapPair = Pair.of(field.name(),typeFactory.createSqlType(SqlTypeName.BOOLEAN));
				break;
			case BYTES :
				mapPair = Pair.of(field.name(),typeFactory.createSqlType(SqlTypeName.TINYINT));
				break;
			case DOUBLE :
				mapPair = Pair.of(field.name(),typeFactory.createSqlType(SqlTypeName.DOUBLE));
				break;
			case FLOAT :
				mapPair = Pair.of(field.name(),typeFactory.createSqlType(SqlTypeName.FLOAT));
				break;
			case LONG :
				mapPair = Pair.of(field.name(),typeFactory.createSqlType(SqlTypeName.BIGINT));
				break;
			case NULL :
				mapPair = Pair.of(field.name(),typeFactory.createSqlType(SqlTypeName.NULL));
				break;
			case RECORD :
			case STRING :
				mapPair = Pair.of(field.name(),typeFactory.createSqlType(SqlTypeName.VARCHAR));
				break;
			case FIXED :
			case UNION :
			case ARRAY :
			case ENUM :
			case MAP :
			default :
				throw new RuntimeException("Complex Avro Data Types are not supported\n");
		} //end switch
		optiqFields.add(mapPair);
	}//end for

	// create row type for optiq
	rowType = typeFactory.createStructType(optiqFields);

	System.err.println("Created type "+rowType);
}