[GitHub] [flink] twalthr commented on a change in pull request #16793: [FLINK-20897][table-planner] Support batch mode in StreamTableEnvironment

[GitBox <gi...@apache.org>]

twalthr commented on a change in pull request #16793:
URL: https://github.com/apache/flink/pull/16793#discussion_r688288828



##########
File path: flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/stream/sql/DataStreamJavaITCase.java
##########
@@ -501,10 +527,87 @@ public void processElement(
         config.setLocalTimeZone(originalZone);
     }
 
+    @Test
+    public void testComplexUnifiedPipelineBatch() throws Exception {
+        env.setRuntimeMode(RuntimeExecutionMode.BATCH);
+
+        final Table resultTable = getComplexUnifiedPipeline(env);
+
+        testResult(resultTable.execute(), Row.of("Bob", 1L), Row.of("Alice", 1L));
+    }
+
+    @Test
+    public void testComplexUnifiedPipelineStreaming() throws Exception {
+        final Table resultTable = getComplexUnifiedPipeline(env);
+
+        // more rows than in batch mode due to incremental computations
+        testResult(
+                resultTable.execute(),
+                Row.of("Bob", 1L),
+                Row.of("Bob", 2L),
+                Row.of("Bob", 3L),
+                Row.of("Alice", 1L));
+    }
+
     // --------------------------------------------------------------------------------------------
     // Helper methods
     // --------------------------------------------------------------------------------------------
 
+    private Table getComplexUnifiedPipeline(StreamExecutionEnvironment env) {
+
+        final DataStream<String> allowedNamesStream = env.fromElements("Bob", "Alice");
+
+        final StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
+
+        tableEnv.createTemporaryView(
+                "AllowedNamesTable", tableEnv.fromDataStream(allowedNamesStream).as("allowedName"));
+
+        final Table nameCountTable =
+                tableEnv.sqlQuery(
+                        "SELECT name, COUNT(*) AS c "
+                                + "FROM (VALUES ('Bob'), ('Alice'), ('Greg'), ('Bob')) AS NameTable(name) "
+                                + "WHERE name IN (SELECT allowedName FROM AllowedNamesTable)"
+                                + "GROUP BY name");
+
+        final DataStream<Row> nameCountStream = tableEnv.toChangelogStream(nameCountTable);

Review comment:
       This is more of a testing example. It doesn't really serve a purpose. I'm just counting the changes that come out of the Table API to have mixed stateful operators.




-- 
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: issues-unsubscribe@flink.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org