jiangxin369 commented on code in PR #22034:
URL: https://github.com/apache/flink/pull/22034#discussion_r1127523251
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/stream/sql/DataStreamJavaITCase.java:
##########
@@ -377,6 +385,102 @@ public void testFromAndToDataStreamEventTime() throws
Exception {
Row.of("c", 1000));
}
+ @Test
+ public void testFromAndToDataStreamBypassWithPojo() throws Exception {
+ env.setParallelism(1);
+ final StreamTableEnvironment tableEnv =
StreamTableEnvironment.create(env);
+
+ final List<Tuple2<Long, String>> tuples =
+ Arrays.asList(Tuple2.of(1L, "a"), Tuple2.of(2L, "b"),
Tuple2.of(3L, "c"));
+
+ final DataStream<Tuple2<Long, String>> dataStream =
+ env.fromCollection(tuples, Types.TUPLE(Types.LONG,
Types.STRING));
+
+ final Table table = tableEnv.fromDataStream(dataStream);
+
+ final DataStream<Tuple2<Long, String>> convertedDataStream =
+ tableEnv.toDataStream(table,
DataTypes.of(dataStream.getType()));
+
+ assertEquals(dataStream, convertedDataStream);
+ testResult(convertedDataStream, tuples.toArray(new Tuple2[0]));
+
+ final Table tableWithPK =
+ tableEnv.fromDataStream(
+ dataStream,
+ Schema.newBuilder()
+ .column("f0", BIGINT().notNull())
+ .column("f1", STRING())
+ .primaryKey("f0")
+ .build());
+ final DataStream<Tuple2<Long, String>> convertedDataStreamWithPK =
+ tableEnv.toDataStream(tableWithPK,
DataTypes.of(dataStream.getType()));
+
+ assertNotEquals(dataStream, convertedDataStreamWithPK);
+ testResult(convertedDataStreamWithPK, tuples.toArray(new Tuple2[0]));
+ }
+
+ @Test
+ public void testFromAndToDataStreamBypassWithRow() throws Exception {
+ env.setParallelism(1);
+ final StreamTableEnvironment tableEnv =
StreamTableEnvironment.create(env);
+
+ final SourceFunction<Row> rowGenerator =
+ new SourceFunction<Row>() {
+ @Override
+ public final void run(SourceContext<Row> ctx) throws
Exception {
+ Row row = new Row(2);
+ row.setField(0, 1L);
+ row.setField(1, "a");
+ ctx.collect(row);
+ }
+
+ @Override
+ public void cancel() {}
+ };
+
+ final RowTypeInfo typeInfo =
+ new RowTypeInfo(new TypeInformation[] {Types.LONG,
Types.STRING});
+
+ // test datastream of rows with non-default name
+ DataStream<Row> dataStream = env.addSource(rowGenerator, typeInfo);
Review Comment:
Similar to the above, `env.fromCollection(...)` generates row data with
RowSerializer and it would construct the row with field name so that the rows
of datastream can be accessed by non-default name anyway. But we want to test
the case that the code change works even if the datastream is composed of
positioned rows without field names.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
