import org.apache.calcite.tools.RelBuilder; //导入依赖的package包/类
@Override
public RelNode visit(final LogicalFilter filter) {
  final RelBuilder relBuilder = newCalciteRelBuilderWithoutContext(filter.getCluster());
  RelNode input = filter.getInput().accept(this);
  relBuilder.push(input);

  RexNode newCondition = filter.getCondition().accept(new RexShuttle() {
    @Override
    public RexNode visitInputRef(RexInputRef inputRef) {
      return relBuilder.field(filter.getRowType().getFieldNames().get(inputRef.getIndex()));
    }
  });

  relBuilder.filter(newCondition);
  return relBuilder.build();
}
 

import org.apache.calcite.tools.RelBuilder; //导入依赖的package包/类
@BeforeClass
public static void prepare() {
  relDataType = TYPE_FACTORY.builder()
      .add("order_id", SqlTypeName.BIGINT)
      .add("site_id", SqlTypeName.INTEGER)
      .add("price", SqlTypeName.DOUBLE)
      .add("order_time", SqlTypeName.BIGINT).build();

  beamRowType = CalciteUtils.toBeamRowType(relDataType);
  record = new BeamRecord(beamRowType
      , 1234567L, 0, 8.9, 1234567L);

  SchemaPlus schema = Frameworks.createRootSchema(true);
  final List<RelTraitDef> traitDefs = new ArrayList<>();
  traitDefs.add(ConventionTraitDef.INSTANCE);
  traitDefs.add(RelCollationTraitDef.INSTANCE);
  FrameworkConfig config = Frameworks.newConfigBuilder()
      .parserConfig(SqlParser.configBuilder().setLex(Lex.MYSQL).build()).defaultSchema(schema)
      .traitDefs(traitDefs).context(Contexts.EMPTY_CONTEXT).ruleSets(BeamRuleSets.getRuleSets())
      .costFactory(null).typeSystem(BeamRelDataTypeSystem.BEAM_REL_DATATYPE_SYSTEM).build();

  relBuilder = RelBuilder.create(config);
}
 

import org.apache.calcite.tools.RelBuilder; //导入依赖的package包/类
/**
 * Evaluates a {@link NumericHaving}.
 *
 * @param having  The NumericHaving filter to be evaluated.
 * @param builder  The RelBuilder used with Calcite to make queries.
 * @param apiToFieldMapper  A function to get the aliased aggregation's name from the metric name.
 *
 * @return the equivalent {@link RexNode} to be used in a sql query.
 */
public RexNode evaluate(NumericHaving having, RelBuilder builder, ApiToFieldMapper apiToFieldMapper) {
    Having.DefaultHavingType havingType = (Having.DefaultHavingType) having.getType();
    SqlOperator operator = null;
    switch (havingType) {
        case EQUAL_TO:
            operator = SqlStdOperatorTable.EQUALS;
            break;
        case LESS_THAN:
            operator = SqlStdOperatorTable.LESS_THAN;
            break;
        case GREATER_THAN:
            operator = SqlStdOperatorTable.GREATER_THAN;
            break;
    }
    return builder.call(
            operator,
            builder.field(apiToFieldMapper.apply(having.getAggregation())),
            builder.literal(having.getValue())
    );
}
 

import org.apache.calcite.tools.RelBuilder; //导入依赖的package包/类
/**
 * Evaluates a {@link MultiClauseHaving} filter by performing it's operation over a list of other havings.
 *
 * @param multiClauseHaving  The MultiClauseHaving filter to be evaluated.
 * @param operator  The operator to be performed over the inner clauses of this having filter.
 * @param builder  The RelBuilder used with Calcite to make queries.
 * @param apiToFieldMapper  A function to get the aliased aggregation's name from the metric name.
 *
 * @return the equivalent {@link RexNode} to be used in a sql query.
 */
public RexNode listEvaluate(
        MultiClauseHaving multiClauseHaving,
        SqlOperator operator,
        RelBuilder builder,
        ApiToFieldMapper apiToFieldMapper
) {
    List<RexNode> rexNodes = multiClauseHaving.getHavings()
            .stream()
            .map(having -> dispatcher.invoke(having, builder, apiToFieldMapper))
            .collect(Collectors.toList());

    return builder.call(
            operator,
            rexNodes
    );
}
 

import org.apache.calcite.tools.RelBuilder; //导入依赖的package包/类
/**
 * Evaluates a complex filter by performing a {@link SqlOperator} over a list of dimensions.
 *
 * @param complexFilter  A complexFilter to be evaluated.
 * @param operator  The sql operator to be applied to a complexFilter's fields.
 * @param builder  The RelBuilder used to build queries with Calcite.
 * @param apiToFieldMapper  A function to get the aliased aggregation's name from the metric name.
 *
 * @return a RexNode containing an equivalent filter to the one given.
 */
private RexNode listEvaluate(
        ComplexFilter complexFilter,
        SqlOperator operator,
        RelBuilder builder,
        ApiToFieldMapper apiToFieldMapper
) {
    List<RexNode> rexNodes = complexFilter.getFields()
            .stream()
            .map(filter -> dispatcher.invoke(filter, builder, apiToFieldMapper))
            .collect(Collectors.toList());

    return builder.call(
            operator,
            rexNodes
    );
}
 

import org.apache.calcite.tools.RelBuilder; //导入依赖的package包/类
/**
 * Returns the RexNode used to filter the druidQuery.
 *
 * @param builder  The RelBuilder created with Calcite.
 * @param druidQuery  The query from which to find filter all the filters for.
 * @param apiToFieldMapper  The mapping from api to physical names.
 * @param timestampColumn  The name of the timestamp column in the database.
 *
 * @return the combined RexNodes that should be filtered on.
 */
protected RexNode getAllWhereFilters(
        RelBuilder builder,
        DruidAggregationQuery<?> druidQuery,
        ApiToFieldMapper apiToFieldMapper,
        String timestampColumn
) {
    RexNode timeFilter = sqlTimeConverter.buildTimeFilters(
            builder,
            druidQuery,
            timestampColumn
    );

    if (druidQuery.getFilter() != null) {
        RexNode druidQueryFilter = filterEvaluator.evaluateFilter(
                druidQuery.getFilter(),
                builder,
                apiToFieldMapper
        );
        return builder.and(timeFilter, druidQueryFilter);
    }

    return timeFilter;
}
 

import org.apache.calcite.tools.RelBuilder; //导入依赖的package包/类
/**
 * Gets the collection of having filters to be applied from the druid query.
 *
 * @param builder  The RelBuilder created with Calcite.
 * @param druidQuery  The query to find the having filter from.
 * @param apiToFieldMapper  The mapping from api to physical name.
 *
 * @return the collection of equivalent filters for calcite.
 */
protected Collection<RexNode> getHavingFilter(
        RelBuilder builder,
        DruidAggregationQuery<?> druidQuery,
        ApiToFieldMapper apiToFieldMapper
) {
    RexNode filter = null;
    if (druidQuery.getQueryType().equals(GROUP_BY)) {
        Having having = ((GroupByQuery) druidQuery).getHaving();

        if (having != null) {
            filter = havingEvaluator.evaluateHaving(having, builder, apiToFieldMapper);
        }
    }

    return Collections.singletonList(filter);
}
 

import org.apache.calcite.tools.RelBuilder; //导入依赖的package包/类
/**
 * Find all druid aggregations and convert them to {@link org.apache.calcite.tools.RelBuilder.AggCall}.
 *
 * @param builder  The RelBuilder created with Calcite.
 * @param druidQuery  The druid query to get the aggregations of.
 * @param apiToFieldMapper  The mapping from api to physical name.
 *
 * @return the list of aggregations.
 */
protected List<RelBuilder.AggCall> getAllQueryAggregations(
        RelBuilder builder,
        DruidAggregationQuery<?> druidQuery,
        ApiToFieldMapper apiToFieldMapper
) {
    return druidQuery.getAggregations()
            .stream()
            .map(aggregation -> druidSqlAggregationConverter.apply(aggregation, apiToFieldMapper))
            .map(optionalSqlAggregation -> optionalSqlAggregation.orElseThrow(() -> {
                String msg = "Couldn't build sql aggregation with " + optionalSqlAggregation;
                LOG.debug(msg);
                return new RuntimeException(msg);
            }))
            .map(sqlAggregation -> builder.aggregateCall(
                    sqlAggregation.getSqlAggFunction(),
                    false,
                    null,
                    sqlAggregation.getSqlAggregationAsName(),
                    builder.field(sqlAggregation.getSqlAggregationFieldName())
            ))
            .collect(Collectors.toList());
}
 

import org.apache.calcite.tools.RelBuilder; //导入依赖的package包/类
/**
 * Collects all the time columns and dimensions to be grouped on.
 *
 * @param builder  The RelBuilder created with Calcite.
 * @param druidQuery  The query to find grouping columns from.
 * @param apiToFieldMapper  The mapping from api to physical name.
 * @param timestampColumn  The name of the timestamp column in the database.
 *
 * @return all columns which should be grouped on.
 */
protected List<RexNode> getAllGroupByColumns(
        RelBuilder builder,
        DruidAggregationQuery<?> druidQuery,
        ApiToFieldMapper apiToFieldMapper,
        String timestampColumn
) {
    List<RexNode> timeFilters = sqlTimeConverter.buildGroupBy(
            builder,
            druidQuery.getGranularity(),
            timestampColumn
    );

    List<RexNode> dimensionFields = getDimensionFields(builder, druidQuery, apiToFieldMapper);

    List<RexNode> allGroupBys = new ArrayList<>(timeFilters.size() + dimensionFields.size());
    allGroupBys.addAll(timeFilters);
    allGroupBys.addAll(dimensionFields);
    return allGroupBys;
}
 

import org.apache.calcite.tools.RelBuilder; //导入依赖的package包/类
@Test
@Ignore("CALCITE-1751")
public void testImplWithMultipleFilters() {
  final SchemaPlus schema = createTestSchema();
  final RelBuilder builder = createRelBuilder(schema);
  final RelNode node = builder.scan("t")
      .filter(
          builder.and(builder.call(GREATER_THAN, builder.field("tc0"), builder.literal("abc")),
              builder.call(EQUALS, builder.field("tc1"), builder.literal("3"))))
      .build();
  final RelNode optimized = optimizeWithVolcano(node);
  assertScriptAndResults("t", getPigScript(optimized, schema),
      "t = LOAD 'target/data.txt"
          + "' USING PigStorage() AS (tc0:chararray, tc1:chararray);\n"
          + "t = FILTER t BY (tc0 > 'abc') AND (tc1 == '3');",
      new String[] { "(c,3)" });
}
 

import org.apache.calcite.tools.RelBuilder; //导入依赖的package包/类
@Test
@Ignore("CALCITE-1751")
public void testImplWithGroupByAndCount() {
  final SchemaPlus schema = createTestSchema();
  final RelBuilder builder = createRelBuilder(schema);
  final RelNode node = builder.scan("t")
      .aggregate(builder.groupKey("tc0"), builder.count(false, "c", builder.field("tc1")))
      .build();
  final RelNode optimized = optimizeWithVolcano(node);
  assertScriptAndResults("t", getPigScript(optimized, schema),
      "t = LOAD 'target/data.txt"
          + "' USING PigStorage() AS (tc0:chararray, tc1:chararray);\n"
          + "t = GROUP t BY (tc0);\n"
          + "t = FOREACH t {\n"
          + "  GENERATE group AS tc0, COUNT(t.tc1) AS c;\n"
          + "};",
      new String[] { "(a,1)", "(b,1)", "(c,1)" });
}
 

import org.apache.calcite.tools.RelBuilder; //导入依赖的package包/类
@Test
public void testImplWithCountWithoutGroupBy() {
  final SchemaPlus schema = createTestSchema();
  final RelBuilder builder = createRelBuilder(schema);
  final RelNode node = builder.scan("t")
      .aggregate(builder.groupKey(), builder.count(false, "c", builder.field("tc0"))).build();
  final RelNode optimized = optimizeWithVolcano(node);
  assertScriptAndResults("t", getPigScript(optimized, schema),
      "t = LOAD 'target/data.txt"
          + "' USING PigStorage() AS (tc0:chararray, tc1:chararray);\n"
          + "t = GROUP t ALL;\n"
          + "t = FOREACH t {\n"
          + "  GENERATE COUNT(t.tc0) AS c;\n"
          + "};",
      new String[] { "(3)" });
}
 

import org.apache.calcite.tools.RelBuilder; //导入依赖的package包/类
@Test public void testGetPredicatesForJoin() throws Exception {
  final FrameworkConfig config = RelBuilderTest.config().build();
  final RelBuilder builder = RelBuilder.create(config);
  RelNode join = builder
      .scan("EMP")
      .scan("DEPT")
      .join(JoinRelType.INNER, builder.call(NONDETERMINISTIC_OP))
      .build();
  RelMetadataQuery mq = RelMetadataQuery.instance();
  assertTrue(mq.getPulledUpPredicates(join).pulledUpPredicates.isEmpty());

  RelNode join1 = builder
      .scan("EMP")
      .scan("DEPT")
      .join(JoinRelType.INNER,
        builder.call(SqlStdOperatorTable.EQUALS,
          builder.field(2, 0, 0),
          builder.field(2, 1, 0)))
      .build();
  assertEquals("=($0, $8)",
      mq.getPulledUpPredicates(join1).pulledUpPredicates.get(0).toString());
}
 

import org.apache.calcite.tools.RelBuilder; //导入依赖的package包/类
@Test public void testDistinct() {
  // Equivalent SQL:
  //   SELECT DISTINCT deptno
  //   FROM emp
  final RelBuilder builder = RelBuilder.create(config().build());
  RelNode root =
      builder.scan("EMP")
          .project(builder.field("DEPTNO"))
          .distinct()
          .build();
  final String expected = "LogicalAggregate(group=[{0}])\n"
      + "  LogicalProject(DEPTNO=[$7])\n"
      + "    LogicalTableScan(table=[[scott, EMP]])\n";
  assertThat(str(root),
      is(expected));
}
 

import org.apache.calcite.tools.RelBuilder; //导入依赖的package包/类
@Test public void testScanFilterOr2() {
  // Equivalent SQL:
  //   SELECT *
  //   FROM emp
  //   WHERE deptno = 20 OR deptno = 20
  // simplifies to
  //   SELECT *
  //   FROM emp
  //   WHERE deptno = 20
  final RelBuilder builder = RelBuilder.create(config().build());
  RelNode root =
      builder.scan("EMP")
          .filter(
              builder.call(SqlStdOperatorTable.OR,
                  builder.call(SqlStdOperatorTable.GREATER_THAN,
                      builder.field("DEPTNO"),
                      builder.literal(20)),
                  builder.call(SqlStdOperatorTable.GREATER_THAN,
                      builder.field("DEPTNO"),
                      builder.literal(20))))
          .build();
  assertThat(str(root),
      is("LogicalFilter(condition=[>($7, 20)])\n"
          + "  LogicalTableScan(table=[[scott, EMP]])\n"));
}
 

import org.apache.calcite.tools.RelBuilder; //导入依赖的package包/类
/** Tests creating Values with some field names and some values null. */
@Test public void testValuesNullable() {
  // Equivalent SQL:
  //   VALUES (null, 1, 'abc'), (false, null, 'longer string')
  final RelBuilder builder = RelBuilder.create(config().build());
  RelNode root =
      builder.values(new String[]{"a", null, "c"},
          null, 1, "abc",
          false, null, "longer string").build();
  final String expected =
      "LogicalValues(tuples=[[{ null, 1, 'abc' }, { false, null, 'longer string' }]])\n";
  assertThat(str(root), is(expected));
  final String expectedType =
      "RecordType(BOOLEAN a, INTEGER expr$1, CHAR(13) CHARACTER SET \"ISO-8859-1\" COLLATE \"ISO-8859-1$en_US$primary\" NOT NULL c) NOT NULL";
  assertThat(root.getRowType().getFullTypeString(), is(expectedType));
}
 

import org.apache.calcite.tools.RelBuilder; //导入依赖的package包/类
@Test public void testCorrelationWithCondition() {
  final RelBuilder builder = RelBuilder.create(config().build());
  final Holder<RexCorrelVariable> v = Holder.of(null);
  RelNode root = builder.scan("EMP")
      .variable(v)
      .scan("DEPT")
      .filter(
          builder.equals(builder.field(0),
              builder.field(v.get(), "DEPTNO")))
      .join(JoinRelType.LEFT,
          builder.equals(builder.field(2, 0, "SAL"),
              builder.literal(1000)),
          ImmutableSet.of(v.get().id))
      .build();
  // Note that the join filter gets pushed to the right-hand input of
  // LogicalCorrelate
  final String expected = ""
      + "LogicalCorrelate(correlation=[$cor0], joinType=[left], requiredColumns=[{7}])\n"
      + "  LogicalTableScan(table=[[scott, EMP]])\n"
      + "  LogicalFilter(condition=[=($cor0.SAL, 1000)])\n"
      + "    LogicalFilter(condition=[=($0, $cor0.DEPTNO)])\n"
      + "      LogicalTableScan(table=[[scott, DEPT]])\n";
  assertThat(str(root), is(expected));
}
 

import org.apache.calcite.tools.RelBuilder; //导入依赖的package包/类
@Test public void testAggregate() {
  // Equivalent SQL:
  //   SELECT COUNT(DISTINCT deptno) AS c
  //   FROM emp
  //   GROUP BY ()
  final RelBuilder builder = RelBuilder.create(config().build());
  RelNode root =
      builder.scan("EMP")
          .aggregate(builder.groupKey(),
              builder.aggregateCall(SqlStdOperatorTable.COUNT, true, false,
                  null, "C", builder.field("DEPTNO")))
          .build();
  assertThat(str(root),
      is("LogicalAggregate(group=[{}], C=[COUNT(DISTINCT $7)])\n"
          + "  LogicalTableScan(table=[[scott, EMP]])\n"));
}
 

import org.apache.calcite.tools.RelBuilder; //导入依赖的package包/类
/** Variation on {@link #testProjectIdentityWithFieldsRename}: don't use a
 * table alias, and make sure the field names propagate through a filter. */
@Test public void testProjectIdentityWithFieldsRenameFilter() {
  final RelBuilder builder = RelBuilder.create(config().build());
  RelNode root =
      builder.scan("DEPT")
          .project(builder.alias(builder.field(0), "a"),
              builder.alias(builder.field(1), "b"),
              builder.alias(builder.field(2), "c"))
          .filter(
              builder.call(SqlStdOperatorTable.EQUALS,
                  builder.field("a"),
                  builder.literal(20)))
          .aggregate(builder.groupKey(0, 1, 2),
              builder.aggregateCall(SqlStdOperatorTable.SUM,
                  false, false, null, null,
                  builder.field(0)))
          .project(builder.field("c"),
              builder.field("a"))
          .build();
  final String expected = ""
      + "LogicalProject(c=[$2], a=[$0])\n"
      + "  LogicalAggregate(group=[{0, 1, 2}], agg#0=[SUM($0)])\n"
      + "    LogicalFilter(condition=[=($0, 20)])\n"
      + "      LogicalTableScan(table=[[scott, DEPT]])\n";
  assertThat(str(root), is(expected));
}
 

import org.apache.calcite.tools.RelBuilder; //导入依赖的package包/类
/** Test case for
 * <a href="https://issues.apache.org/jira/browse/CALCITE-1610">[CALCITE-1610]
 * RelBuilder sort-combining optimization treats aliases incorrectly</a>. */
@Test public void testSortOverProjectSort() {
  final RelBuilder builder = RelBuilder.create(config().build());
  builder.scan("EMP")
      .sort(0)
      .project(builder.field(1))
      // was throwing exception here when attempting to apply to
      // inner sort node
      .limit(0, 1)
      .build();
  RelNode r = builder.scan("EMP")
      .sort(0)
      .project(Lists.newArrayList(builder.field(1)),
          Lists.newArrayList("F1"))
      .limit(0, 1)
      // make sure we can still access the field by alias
      .project(builder.field("F1"))
      .build();
  String expected = "LogicalProject(F1=[$1])\n"
      + "  LogicalSort(sort0=[$0], dir0=[ASC], fetch=[1])\n"
      + "    LogicalTableScan(table=[[scott, EMP]])\n";
  assertThat(str(r), is(expected));
}