BeamIOSourceRel extends TableScan implements BeamRelNode { @Override public double estimateRowCount(RelMetadataQuery mq) { BeamTableStatistics rowCountStatistics = calciteTable.getStatistic(); if (beamTable.isBounded() == PCollection.IsBounded.BOUNDED) { return rowCountStatistics.getRowCount(); } else { return rowCountStatistics.getRate(); } } BeamIOSourceRel(
      RelOptCluster cluster,
      RelTraitSet traitSet,
      RelOptTable table,
      BeamSqlTable beamTable,
      Map<String, String> pipelineOptions,
      BeamCalciteTable calciteTable); BeamPushDownIOSourceRel createPushDownRel(
      RelDataType newType, List<String> usedFields, BeamSqlTableFilter tableFilters); @Override double estimateRowCount(RelMetadataQuery mq); @Override NodeStats estimateNodeStats(RelMetadataQuery mq); @Override PCollection.IsBounded isBounded(); @Override PTransform<PCollectionList<Row>, PCollection<Row>> buildPTransform(); @Override RelOptCost computeSelfCost(RelOptPlanner planner, RelMetadataQuery mq); @Override BeamCostModel beamComputeSelfCost(RelOptPlanner planner, RelMetadataQuery mq); BeamSqlTable getBeamSqlTable(); @Override Map<String, String> getPipelineOptions(); static final double CONSTANT_WINDOW_SIZE; }
@Test public void boundedRowCount() { String sql = "SELECT * FROM ORDER_DETAILS_BOUNDED"; RelNode root = env.parseQuery(sql); while (!(root instanceof BeamIOSourceRel)) { root = root.getInput(0); } Assert.assertEquals(5d, root.estimateRowCount(RelMetadataQuery.instance()), 0.001); }
@Test public void unboundedRowCount() { String sql = "SELECT * FROM ORDER_DETAILS_UNBOUNDED"; RelNode root = env.parseQuery(sql); while (!(root instanceof BeamIOSourceRel)) { root = root.getInput(0); } Assert.assertEquals(2d, root.estimateRowCount(RelMetadataQuery.instance()), 0.001); }