import org.apache.calcite.plan.Contexts; //导入依赖的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.plan.Contexts; //导入依赖的package包/类
private static Planner createPlanner() {
  Connection connection;
  SchemaPlus rootSchema;
  try {
    connection = DriverManager.getConnection("jdbc:calcite:");
    CalciteConnection calciteConnection = connection.unwrap(CalciteConnection.class);
    rootSchema = calciteConnection.getRootSchema();
  } catch (SQLException e) {
    throw new SamzaException(e);
  }

  final List<RelTraitDef> traitDefs = new ArrayList<>();

  traitDefs.add(ConventionTraitDef.INSTANCE);
  traitDefs.add(RelCollationTraitDef.INSTANCE);

  FrameworkConfig frameworkConfig = Frameworks.newConfigBuilder()
      .parserConfig(SqlParser.configBuilder().setLex(Lex.JAVA).build())
      .defaultSchema(rootSchema)
      .operatorTable(SqlStdOperatorTable.instance())
      .traitDefs(traitDefs)
      .context(Contexts.EMPTY_CONTEXT)
      .costFactory(null)
      .build();
  return Frameworks.getPlanner(frameworkConfig);
}
 

import org.apache.calcite.plan.Contexts; //导入依赖的package包/类
public BeamQueryPlanner(SchemaPlus schema) {
  String defaultCharsetKey = "saffron.default.charset";
  if (System.getProperty(defaultCharsetKey) == null) {
    System.setProperty(defaultCharsetKey, ConversionUtil.NATIVE_UTF16_CHARSET_NAME);
    System.setProperty("saffron.default.nationalcharset",
      ConversionUtil.NATIVE_UTF16_CHARSET_NAME);
    System.setProperty("saffron.default.collation.name",
      String.format("%s$%s", ConversionUtil.NATIVE_UTF16_CHARSET_NAME, "en_US"));
  }

  final List<RelTraitDef> traitDefs = new ArrayList<>();
  traitDefs.add(ConventionTraitDef.INSTANCE);
  traitDefs.add(RelCollationTraitDef.INSTANCE);

  List<SqlOperatorTable> sqlOperatorTables = new ArrayList<>();
  sqlOperatorTables.add(SqlStdOperatorTable.instance());
  sqlOperatorTables.add(new CalciteCatalogReader(CalciteSchema.from(schema), false,
      Collections.<String>emptyList(), TYPE_FACTORY));

  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)
      .operatorTable(new ChainedSqlOperatorTable(sqlOperatorTables))
      .build();
  this.planner = Frameworks.getPlanner(config);

  for (String t : schema.getTableNames()) {
    sourceTables.put(t, (BaseBeamTable) schema.getTable(t));
  }
}
 

import org.apache.calcite.plan.Contexts; //导入依赖的package包/类
/**
 * @deprecated to be removed before 2.0
 */
@Deprecated
public FlinkAggregateExpandDistinctAggregatesRule(
		Class<? extends LogicalAggregate> clazz,
		boolean useGroupingSets,
		RelFactories.JoinFactory joinFactory) {
	this(clazz, useGroupingSets, RelBuilder.proto(Contexts.of(joinFactory)));
}
 

import org.apache.calcite.plan.Contexts; //导入依赖的package包/类
@Deprecated // to be removed before 2.0
public JoinProjectTransposeRule(RelOptRuleOperand operand,
    String description, boolean includeOuter,
    ProjectFactory projectFactory) {
  this(operand, description, includeOuter,
      RelBuilder.proto(Contexts.of(projectFactory)));
}
 

import org.apache.calcite.plan.Contexts; //导入依赖的package包/类
@Deprecated // to be removed before 2.0
public AggregateExpandDistinctAggregatesRule(
    Class<? extends LogicalAggregate> clazz,
    boolean useGroupingSets,
    RelFactories.JoinFactory joinFactory) {
  this(clazz, useGroupingSets, RelBuilder.proto(Contexts.of(joinFactory)));
}
 

import org.apache.calcite.plan.Contexts; //导入依赖的package包/类
@Deprecated // to be removed before 2.0
public FilterAggregateTransposeRule(
    Class<? extends Filter> filterClass,
    RelFactories.FilterFactory filterFactory,
    Class<? extends Aggregate> aggregateClass) {
  this(filterClass, RelBuilder.proto(Contexts.of(filterFactory)),
      aggregateClass);
}
 

import org.apache.calcite.plan.Contexts; //导入依赖的package包/类
/**
 * Creates a TesterImpl.
 *
 * @param diffRepos Diff repository
 * @param enableDecorrelate Whether to decorrelate
 * @param enableTrim Whether to trim unused fields
 * @param enableExpand Whether to expand sub-queries
 * @param catalogReaderFactory Function to create catalog reader, or null
 * @param clusterFactory Called after a cluster has been created
 */
protected TesterImpl(DiffRepository diffRepos, boolean enableDecorrelate,
    boolean enableTrim, boolean enableExpand,
    boolean enableLateDecorrelate,
    Function<RelDataTypeFactory, Prepare.CatalogReader>
        catalogReaderFactory,
    Function<RelOptCluster, RelOptCluster> clusterFactory) {
  this(diffRepos, enableDecorrelate, enableTrim, enableExpand,
      enableLateDecorrelate,
      catalogReaderFactory,
      clusterFactory,
      SqlToRelConverter.Config.DEFAULT,
      SqlConformanceEnum.DEFAULT,
      Contexts.empty());
}
 

import org.apache.calcite.plan.Contexts; //导入依赖的package包/类
/** Test case for
 * <a href="https://issues.apache.org/jira/browse/CALCITE-931">[CALCITE-931]
 * Wrong collation trait in SortJoinTransposeRule for right joins</a>. */
@Test public void testSortJoinTranspose4() {
  // Create a customized test with RelCollation trait in the test cluster.
  Tester tester = new TesterImpl(getDiffRepos(), true, true, false, false,
      null, null) {
    @Override public RelOptPlanner createPlanner() {
      return new MockRelOptPlanner(Contexts.empty()) {
        @Override public List<RelTraitDef> getRelTraitDefs() {
          return ImmutableList.<RelTraitDef>of(RelCollationTraitDef.INSTANCE);
        }
        @Override public RelTraitSet emptyTraitSet() {
          return RelTraitSet.createEmpty().plus(
              RelCollationTraitDef.INSTANCE.getDefault());
        }
      };
    }
  };

  final HepProgram preProgram = new HepProgramBuilder()
      .addRuleInstance(SortProjectTransposeRule.INSTANCE)
      .build();
  final HepProgram program = new HepProgramBuilder()
      .addRuleInstance(SortJoinTransposeRule.INSTANCE)
      .build();
  final String sql = "select * from sales.emp e right join (\n"
      + "select * from sales.dept d) using (deptno)\n"
      + "order by name";
  checkPlanning(tester, preProgram, new HepPlanner(program), sql);
}
 

import org.apache.calcite.plan.Contexts; //导入依赖的package包/类
/** Test case for
 * <a href="https://issues.apache.org/jira/browse/CALCITE-434">[CALCITE-434]
 * Converting predicates on date dimension columns into date ranges</a>,
 * specifically a rule that converts {@code EXTRACT(YEAR FROM ...) = constant}
 * to a range. */
@Test public void testExtractYearToRange() {
  final String sql = "select *\n"
      + "from sales.emp_b as e\n"
      + "where extract(year from birthdate) = 2014";
  HepProgram program = new HepProgramBuilder()
      .addRuleInstance(DateRangeRules.FILTER_INSTANCE)
      .build();
  final Context context =
      Contexts.of(new CalciteConnectionConfigImpl(new Properties()));
  sql(sql).with(program).withContext(context).check();
}
 

import org.apache.calcite.plan.Contexts; //导入依赖的package包/类
@Test public void testExtractYearMonthToRange() {
  final String sql = "select *\n"
      + "from sales.emp_b as e\n"
      + "where extract(year from birthdate) = 2014"
      + "and extract(month from birthdate) = 4";
  HepProgram program = new HepProgramBuilder()
      .addRuleInstance(DateRangeRules.FILTER_INSTANCE)
      .build();
  final Context context =
      Contexts.of(new CalciteConnectionConfigImpl(new Properties()));
  sql(sql).with(program).withContext(context).check();
}
 

import org.apache.calcite.plan.Contexts; //导入依赖的package包/类
private DrillFilterAggregateTransposeRule() {
  super(Filter.class, RelBuilder.proto(Contexts.of(RelFactories.DEFAULT_FILTER_FACTORY)), Aggregate.class);
}
 

import org.apache.calcite.plan.Contexts; //导入依赖的package包/类
public RelRoot plan(String query) {
  try {
    Connection connection = DriverManager.getConnection("jdbc:calcite:");
    CalciteConnection calciteConnection = connection.unwrap(CalciteConnection.class);
    SchemaPlus rootSchema = calciteConnection.getRootSchema();

    for (SqlSystemStreamConfig ssc : systemStreamConfigBySource.values()) {
      SchemaPlus previousLevelSchema = rootSchema;
      List<String> sourceParts = ssc.getSourceParts();
      RelSchemaProvider relSchemaProvider = relSchemaProviders.get(ssc.getSource());

      for (String sourcePart : sourceParts) {
        if (!sourcePart.equalsIgnoreCase(ssc.getStreamName())) {
          SchemaPlus sourcePartSchema = rootSchema.getSubSchema(sourcePart);
          if (sourcePartSchema == null) {
            sourcePartSchema = previousLevelSchema.add(sourcePart, new AbstractSchema());
          }
          previousLevelSchema = sourcePartSchema;
        } else {
          // If the source part is the streamName, then fetch the schema corresponding to the stream and register.
          RelDataType relationalSchema = relSchemaProvider.getRelationalSchema();
          previousLevelSchema.add(ssc.getStreamName(), createTableFromRelSchema(relationalSchema));
          break;
        }
      }
    }

    List<SamzaSqlScalarFunctionImpl> samzaSqlFunctions = udfMetadata.stream()
        .map(x -> new SamzaSqlScalarFunctionImpl(x.getName(), x.getUdfMethod()))
        .collect(Collectors.toList());

    final List<RelTraitDef> traitDefs = new ArrayList<>();

    traitDefs.add(ConventionTraitDef.INSTANCE);
    traitDefs.add(RelCollationTraitDef.INSTANCE);

    List<SqlOperatorTable> sqlOperatorTables = new ArrayList<>();
    sqlOperatorTables.add(new SamzaSqlOperatorTable());
    sqlOperatorTables.add(new SamzaSqlUdfOperatorTable(samzaSqlFunctions));

    FrameworkConfig frameworkConfig = Frameworks.newConfigBuilder()
        .parserConfig(SqlParser.configBuilder().setLex(Lex.JAVA).build())
        .defaultSchema(rootSchema)
        .operatorTable(new ChainedSqlOperatorTable(sqlOperatorTables))
        .traitDefs(traitDefs)
        .context(Contexts.EMPTY_CONTEXT)
        .costFactory(null)
        .build();
    Planner planner = Frameworks.getPlanner(frameworkConfig);

    SqlNode sql = planner.parse(query);
    SqlNode validatedSql = planner.validate(sql);
    RelRoot relRoot = planner.rel(validatedSql);
    LOG.info("query plan:\n" + sql.toString());
    LOG.info("relational graph:");
    printRelGraph(relRoot.project());
    return relRoot;
  } catch (Exception e) {
    LOG.error("Query planner failed with exception.", e);
    throw new SamzaException(e);
  }
}
 

import org.apache.calcite.plan.Contexts; //导入依赖的package包/类
@Deprecated // to be removed before 2.0
public JoinCommuteRule(Class<? extends Join> clazz,
    ProjectFactory projectFactory) {
  this(clazz, RelBuilder.proto(Contexts.of(projectFactory)), false);
}
 

import org.apache.calcite.plan.Contexts; //导入依赖的package包/类
@Deprecated // to be removed before 2.0
public FilterMergeRule(RelFactories.FilterFactory filterFactory) {
  this(RelBuilder.proto(Contexts.of(filterFactory)));
}
 

import org.apache.calcite.plan.Contexts; //导入依赖的package包/类
@Deprecated // to be removed before 2.0
protected JoinToCorrelateRule(RelFactories.FilterFactory filterFactory) {
  this(RelBuilder.proto(Contexts.of(filterFactory)));
}
 

import org.apache.calcite.plan.Contexts; //导入依赖的package包/类
@Deprecated // to  be removed before 2.0
public FilterSetOpTransposeRule(RelFactories.FilterFactory filterFactory) {
  this(RelBuilder.proto(Contexts.of(filterFactory)));
}
 

import org.apache.calcite.plan.Contexts; //导入依赖的package包/类
@Deprecated // to be removed before 2.0
public JoinPushTransitivePredicatesRule(Class<? extends Join> clazz,
    RelFactories.FilterFactory filterFactory) {
  this(clazz, RelBuilder.proto(Contexts.of(filterFactory)));
}
 

import org.apache.calcite.plan.Contexts; //导入依赖的package包/类
protected RelBuilder(Context context, RelOptCluster cluster,
    RelOptSchema relOptSchema) {
  this.cluster = cluster;
  this.relOptSchema = relOptSchema;
  if (context == null) {
    context = Contexts.EMPTY_CONTEXT;
  }
  this.simplify = Hook.REL_BUILDER_SIMPLIFY.get(true);
  this.aggregateFactory =
      Util.first(context.unwrap(RelFactories.AggregateFactory.class),
          RelFactories.DEFAULT_AGGREGATE_FACTORY);
  this.filterFactory =
      Util.first(context.unwrap(RelFactories.FilterFactory.class),
          RelFactories.DEFAULT_FILTER_FACTORY);
  this.projectFactory =
      Util.first(context.unwrap(RelFactories.ProjectFactory.class),
          RelFactories.DEFAULT_PROJECT_FACTORY);
  this.sortFactory =
      Util.first(context.unwrap(RelFactories.SortFactory.class),
          RelFactories.DEFAULT_SORT_FACTORY);
  this.setOpFactory =
      Util.first(context.unwrap(RelFactories.SetOpFactory.class),
          RelFactories.DEFAULT_SET_OP_FACTORY);
  this.joinFactory =
      Util.first(context.unwrap(RelFactories.JoinFactory.class),
          RelFactories.DEFAULT_JOIN_FACTORY);
  this.semiJoinFactory =
      Util.first(context.unwrap(RelFactories.SemiJoinFactory.class),
          RelFactories.DEFAULT_SEMI_JOIN_FACTORY);
  this.correlateFactory =
      Util.first(context.unwrap(RelFactories.CorrelateFactory.class),
          RelFactories.DEFAULT_CORRELATE_FACTORY);
  this.valuesFactory =
      Util.first(context.unwrap(RelFactories.ValuesFactory.class),
          RelFactories.DEFAULT_VALUES_FACTORY);
  this.scanFactory =
      Util.first(context.unwrap(RelFactories.TableScanFactory.class),
          RelFactories.DEFAULT_TABLE_SCAN_FACTORY);
  this.matchFactory =
      Util.first(context.unwrap(RelFactories.MatchFactory.class),
          RelFactories.DEFAULT_MATCH_FACTORY);
  final RexExecutor executor =
      Util.first(context.unwrap(RexExecutor.class),
          Util.first(cluster.getPlanner().getExecutor(), RexUtil.EXECUTOR));
  final RelOptPredicateList predicates = RelOptPredicateList.EMPTY;
  this.simplifier =
      new RexSimplify(cluster.getRexBuilder(), predicates, false, executor);
  this.simplifierUnknownAsFalse =
      new RexSimplify(cluster.getRexBuilder(), predicates, true, executor);
}
 

import org.apache.calcite.plan.Contexts; //导入依赖的package包/类
/** Creates a {@link RelBuilderFactory} that uses a given set of factories. */
public static RelBuilderFactory proto(Object... factories) {
  return proto(Contexts.of(factories));
}