[BEAM-2292] PCollection as a Table

[mingmxu]

add BeamPCollectionTable to create table from PCollection

Note that it changes the interface of BaseBeamTable, as a BeamPCollectionTable doesn't start from PBegin, return PCollection<BeamSQLRow> is more clear:

-  public abstract PTransform<? super PBegin, PCollection<BeamSQLRow>> buildIOReader();
+  public abstract PCollection<BeamSQLRow> buildIOReader(Pipeline pipeline);

[mingmxu]

R: @xumingming
CC: @takidau @jbonofre @davorbonaci

[coveralls]

Changes Unknown when pulling 148e0b3 on XuMingmin:BEAM-2292 into ** on apache:DSL_SQL**.

[mingmxu]

retest this please

[coveralls]

Changes Unknown when pulling 6fd2550 on XuMingmin:BEAM-2292 into ** on apache:DSL_SQL**.

[mingmxu]

retest this please

[coveralls]

Changes Unknown when pulling 6fd2550 on XuMingmin:BEAM-2292 into ** on apache:DSL_SQL**.

[mingmxu]

@xumingming Can you take a look here?

[xumingming]

The only comment i have is: the dataType of the table seems inferable from the rows? Quote from BaseBeamTable:

  public BaseBeamTable(RelProtoDataType protoRowType) {
    this.relDataType = protoRowType.apply(BeamQueryPlanner.TYPE_FACTORY); // table dataType
    this.beamSqlRecordType = BeamSQLRecordType.from(relDataType); // rowDataType
  }

[mingmxu]

Do you mean the parameter of

public BeamPCollectionTable(PCollection<BeamSQLRow> upstream, RelProtoDataType protoRowType)

?
PCollection<BeamSQLRow> itself doesn't return value of field BeamSQLRow.dataType, though it can be known when processing records. This schema information is required for Calcite to validate query, I suppose we need to pass it to the construction method.

[xumingming]

can the protoRowType infered from the rows?

[xumingming]

LGTM

[mingmxu]

@jbonofre can you also take a peak here?

[jbonofre]

R: @jbonofre

[jbonofre]

LGTM, I'm merging.

[jbonofre]

Merged on DSL_SQL branch. You can close this pull request. Thanks !

[mingmxu]

Thank you @jbonofre @xumingming