wuchong commented on a change in pull request #13141:
URL: https://github.com/apache/flink/pull/13141#discussion_r470424738
##########
File path:
flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/runtime/stream/table/TableSourceITCase.scala
##########
@@ -81,6 +81,51 @@ class TableSourceITCase extends AbstractTestBase {
// info.
}
+ @Test
+ def testStreamScanParallelism(): Unit = {
+ val env = StreamExecutionEnvironment.getExecutionEnvironment
+ val settings = EnvironmentSettings.newInstance().useOldPlanner().build()
+ val tEnv = StreamTableEnvironment.create(env, settings)
+
+ // set environment parallelism to 4
+ env.setParallelism(4)
+
+ // test DataStreamScan
+ val table = env.fromElements[String]()
+ .setParallelism(1)
+ .toTable(tEnv, 'a)
+
+ tEnv.registerTable("MyTable1", table)
+ val parallelism = tEnv.scan("MyTable1")
+ .toAppendStream[String]
+ .getParallelism
+
+ assertEquals(1, parallelism)
+
+ // test StreamTableSourceScan
+ val tableSource = new StreamTableSource[String]() {
+
+ override def getDataStream(execEnv: JExecEnv): DataStream[String] = {
+ execEnv.fromElements[String]("blabla")
+ }
+
+ override def getTableSchema: TableSchema = {
+ new TableSchema(Array("a"), Array(Types.STRING()))
+ }
+
+ override def getReturnType: TypeInformation[String] = {
+ Types.STRING()
+ }
+ }
+
+
tEnv.asInstanceOf[TableEnvironmentInternal].registerTableSourceInternal("MyTable2",
tableSource)
+ val parallelism2 = tEnv.scan("MyTable2")
+ .toAppendStream[String]
+ .getParallelism
+
+ assertEquals(1, parallelism2)
Review comment:
1. use new methods instead of depreacted methods
2. use DDL to register a table (StreamTableSourceScan path).
`registerTableSourceInternal` is a hack way.
3. We may need to support `parallelism` property for the COLLECTION source
(`TestCollectionTableFactory`).
```suggestion
tEnv.createTemporaryView("MyTable1", table)
val parallelism = tEnv.from("MyTable1")
.toAppendStream[String]
.parallelism
assertEquals(1, parallelism)
// test StreamTableSourceScan
val createTableStmt =
"""
|CREATE TEMPORARY TABLE MyTable2 (
| str varchar
|) with (
| 'connector' = 'COLLECTION',
| 'is-bounded' = 'false',
| 'parallelism' = '1'
|)
""".stripMargin
tEnv.executeSql(createTableStmt)
val parallelism2 = tEnv.from("MyTable2")
.toAppendStream[String]
.parallelism
assertEquals(1, parallelism2)
```
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
[email protected]
