chore: Changes for removing CTAS (#1643)

bbejeck · web-flow · commit 3f041d7c6ddb · 2023-10-19T18:37:14.000-04:00
* changes for removing CTAS

* changes for failing test

* updates for removing CTAS

* increase sleep to wait for topic creation

* add semicolon to end statement

* add await to create table statements

* remove space

* fix expected output with types
diff --git a/_data/harnesses/hopping-windows/flinksql.yml b/_data/harnesses/hopping-windows/flinksql.yml
@@ -55,12 +55,16 @@ dev:
               render:
                 file: tutorials/hopping-windows/flinksql/markup/dev/create-temperature-by-10min-window.adoc
 
+            - file: tutorial-steps/dev/populate-temperature-by-10min-window.sql
+              render:
+                file: tutorials/hopping-windows/flinksql/markup/dev/populate-temperature-by-10min-window.adoc
+
           stdout:
             directory: tutorial-steps/dev/outputs
 
         - name: wait for table execution to complete
           action: sleep
-          ms: 10000
+          ms: 30000
           render:
              skip: true
 
@@ -114,6 +118,11 @@ test:
           render:
             file: tutorials/hopping-windows/flinksql/markup/test/create-resource-create-temperature-by-10min-window.sql.template.adoc
 
+        - action: make_file
+          file: src/test/resources/populate-temperature-by-10min-window.sql
+          render:
+            file: tutorials/hopping-windows/flinksql/markup/test/populate-resource-temperature-by-10min-window.sql.adoc
+
         - action: make_file
           file: src/test/resources/query-temperature-by-10min-window.sql
           render:
diff --git a/_data/harnesses/tumbling-windows/flinksql.yml b/_data/harnesses/tumbling-windows/flinksql.yml
@@ -55,12 +55,16 @@ dev:
               render:
                 file: tutorials/tumbling-windows/flinksql/markup/dev/create-ratings-by-6hr-window.adoc
 
+            - file: tutorial-steps/dev/populate-ratings-by-6hr-window.sql
+              render:
+                file: tutorials/tumbling-windows/flinksql/markup/dev/populate-ratings-by-6hr-window.adoc
+
           stdout:
             directory: tutorial-steps/dev/outputs
 
         - name: wait for table execution to complete
           action: sleep
-          ms: 10000
+          ms: 30000
           render:
              skip: true
 
@@ -114,6 +118,11 @@ test:
           render:
             file: tutorials/tumbling-windows/flinksql/markup/test/create-resource-create-ratings-by-6hr-window.sql.template.adoc
 
+        - action: make_file
+          file: src/test/resources/populate-ratings-by-6hr-window.sql
+          render:
+            file: tutorials/tumbling-windows/flinksql/markup/test/create-resource-populate-ratings-by-6hr-window.sql.adoc
+
         - action: make_file
           file: src/test/resources/query-ratings-by-6hr-window.sql
           render:
diff --git a/_includes/tutorials/hopping-windows/flinksql/code/settings.gradle b/_includes/tutorials/hopping-windows/flinksql/code/settings.gradle
@@ -7,4 +7,4 @@
  * in the user manual at https://docs.gradle.org/6.7.1/userguide/multi_project_builds.html
  */
 
-rootProject.name = 'aggregating-count-flinksql'
+rootProject.name = 'hopping-windows-flinksql'
diff --git a/_includes/tutorials/hopping-windows/flinksql/code/src/test/java/io/confluent/developer/FlinkSqlHoppingWindowTest.java b/_includes/tutorials/hopping-windows/flinksql/code/src/test/java/io/confluent/developer/FlinkSqlHoppingWindowTest.java
@@ -17,7 +17,8 @@ public void testHoppingWindows() throws Exception {
         Optional.of(kafkaPort), Optional.of(schemaRegistryPort))).await();
     streamTableEnv.executeSql(getResourceFileContents("populate-temperature-readings.sql"));
     streamTableEnv.executeSql(getResourceFileContents("create-temperature-by-10min-window.sql.template",
-        Optional.of(kafkaPort), Optional.of(schemaRegistryPort)));
+        Optional.of(kafkaPort), Optional.of(schemaRegistryPort))).await();
+    streamTableEnv.executeSql(getResourceFileContents("populate-temperature-by-10min-window.sql"));
 
     TableResult tableResult = streamTableEnv.executeSql(getResourceFileContents("query-temperature-by-10min-window.sql"));
 
diff --git a/_includes/tutorials/hopping-windows/flinksql/code/src/test/resources/create-temperature-by-10min-window.sql.template b/_includes/tutorials/hopping-windows/flinksql/code/src/test/resources/create-temperature-by-10min-window.sql.template
@@ -1,5 +1,9 @@
-CREATE TABLE temperature_by_10min_window
-WITH (
+CREATE TABLE temperature_by_10min_window (
+    sensor_id INT,
+    avg_temperature DOUBLE,
+    window_start TIMESTAMP(3),
+    window_end TIMESTAMP(3)
+) WITH (
     'connector' = 'kafka',
     'topic' = 'temperature-by-10min-window',
     'properties.bootstrap.servers' = 'localhost:KAFKA_PORT',
@@ -10,10 +14,4 @@ WITH (
     'value.format' = 'avro-confluent',
     'value.avro-confluent.url' = 'http://localhost:SCHEMA_REGISTRY_PORT',
     'value.fields-include' = 'ALL'
-) AS
-    SELECT sensor_id,
-        AVG(temperature) AS avg_temperature,
-        window_start,
-        window_end
-    FROM TABLE(HOP(TABLE temperature_readings, DESCRIPTOR(ts), INTERVAL '5' MINUTES, INTERVAL '10' MINUTES))
-    GROUP BY sensor_id, window_start, window_end;
+);
diff --git a/_includes/tutorials/hopping-windows/flinksql/code/src/test/resources/populate-temperature-by-10min-window.sql b/_includes/tutorials/hopping-windows/flinksql/code/src/test/resources/populate-temperature-by-10min-window.sql
@@ -0,0 +1,7 @@
+INSERT INTO temperature_by_10min_window
+    SELECT sensor_id,
+        AVG(temperature) AS avg_temperature,
+        window_start,
+        window_end
+    FROM TABLE(HOP(TABLE temperature_readings, DESCRIPTOR(ts), INTERVAL '5' MINUTES, INTERVAL '10' MINUTES))
+    GROUP BY sensor_id, window_start, window_end;
diff --git a/_includes/tutorials/hopping-windows/flinksql/code/tutorial-steps/dev/create-temperature-by-10min-window.sql b/_includes/tutorials/hopping-windows/flinksql/code/tutorial-steps/dev/create-temperature-by-10min-window.sql
@@ -1,5 +1,9 @@
-CREATE TABLE temperature_by_10min_window
-WITH (
+CREATE TABLE temperature_by_10min_window (
+    sensor_id INT,
+    avg_temperature DOUBLE,
+    window_start TIMESTAMP(3),
+    window_end TIMESTAMP(3)
+) WITH (
     'connector' = 'kafka',
     'topic' = 'temperature-by-10min-window',
     'properties.bootstrap.servers' = 'broker:9092',
@@ -10,10 +14,4 @@ WITH (
     'value.format' = 'avro-confluent',
     'value.avro-confluent.url' = 'http://schema-registry:8081',
     'value.fields-include' = 'ALL'
-) AS
-    SELECT sensor_id,
-        AVG(temperature) AS avg_temperature,
-        window_start,
-        window_end
-    FROM TABLE(HOP(TABLE temperature_readings, DESCRIPTOR(ts), INTERVAL '5' MINUTES, INTERVAL '10' MINUTES))
-    GROUP BY sensor_id, window_start, window_end;
+); 
diff --git a/_includes/tutorials/hopping-windows/flinksql/code/tutorial-steps/dev/expected-temperature-by-10min-window.log b/_includes/tutorials/hopping-windows/flinksql/code/tutorial-steps/dev/expected-temperature-by-10min-window.log
@@ -1,8 +1,8 @@
-{"sensor_id":{"int":0},"avg_temperature":{"double":55.0},"window_start":1673748600000,"window_end":1673749200000}
-{"sensor_id":{"int":0},"avg_temperature":{"double":52.5},"window_start":1673748900000,"window_end":1673749500000}
-{"sensor_id":{"int":0},"avg_temperature":{"double":47.5},"window_start":1673749200000,"window_end":1673749800000}
-{"sensor_id":{"int":0},"avg_temperature":{"double":42.5},"window_start":1673749500000,"window_end":1673750100000}
-{"sensor_id":{"int":0},"avg_temperature":{"double":42.5},"window_start":1673749800000,"window_end":1673750400000}
-{"sensor_id":{"int":0},"avg_temperature":{"double":47.5},"window_start":1673750100000,"window_end":1673750700000}
-{"sensor_id":{"int":0},"avg_temperature":{"double":52.5},"window_start":1673750400000,"window_end":1673751000000}
+{"sensor_id":{"int":0},"avg_temperature":{"double":55.0},"window_start":{"long":1673748600000},"window_end":{"long":1673749200000}}
+{"sensor_id":{"int":0},"avg_temperature":{"double":52.5},"window_start":{"long":1673748900000},"window_end":{"long":1673749500000}}
+{"sensor_id":{"int":0},"avg_temperature":{"double":47.5},"window_start":{"long":1673749200000},"window_end":{"long":1673749800000}}
+{"sensor_id":{"int":0},"avg_temperature":{"double":42.5},"window_start":{"long":1673749500000},"window_end":{"long":1673750100000}}
+{"sensor_id":{"int":0},"avg_temperature":{"double":42.5},"window_start":{"long":1673749800000},"window_end":{"long":1673750400000}}
+{"sensor_id":{"int":0},"avg_temperature":{"double":47.5},"window_start":{"long":1673750100000},"window_end":{"long":1673750700000}}
+{"sensor_id":{"int":0},"avg_temperature":{"double":52.5},"window_start":{"long":1673750400000},"window_end":{"long":1673751000000}}
 Processed a total of 7 messages
diff --git a/_includes/tutorials/hopping-windows/flinksql/code/tutorial-steps/dev/populate-temperature-by-10min-window.sql b/_includes/tutorials/hopping-windows/flinksql/code/tutorial-steps/dev/populate-temperature-by-10min-window.sql
@@ -0,0 +1,7 @@
+INSERT INTO temperature_by_10min_window
+    SELECT sensor_id,
+        AVG(temperature) AS avg_temperature,
+        window_start,
+        window_end
+    FROM TABLE(HOP(TABLE temperature_readings, DESCRIPTOR(ts), INTERVAL '5' MINUTES, INTERVAL '10' MINUTES))
+    GROUP BY sensor_id, window_start, window_end;
diff --git a/_includes/tutorials/hopping-windows/flinksql/markup/dev/create-temperature-by-10min-window.adoc b/_includes/tutorials/hopping-windows/flinksql/markup/dev/create-temperature-by-10min-window.adoc
@@ -1,4 +1,4 @@
-Since the output of our transient query looks right, the next step is to make it persistent with the following `CREATE TABLE AS SELECT` statement. Go ahead and execute the following statement in your Flink SQL session:
+Since the output of our transient query looks right, the next step is to make it persistent with the following `CREATE TABLE` statement. Go ahead and execute the following statement in your Flink SQL session:
 
 +++++
 <pre class="snippet"><code class="sql">{% include_raw tutorials/hopping-windows/flinksql/code/tutorial-steps/dev/create-temperature-by-10min-window.sql %}</code></pre>
diff --git a/_includes/tutorials/hopping-windows/flinksql/markup/dev/populate-temperature-by-10min-window.adoc b/_includes/tutorials/hopping-windows/flinksql/markup/dev/populate-temperature-by-10min-window.adoc
@@ -0,0 +1,5 @@
+Next execute the following statement in your Flink SQL session to populate the table:
+
++++++
+<pre class="snippet"><code class="sql">{% include_raw tutorials/hopping-windows/flinksql/code/tutorial-steps/dev/populate-temperature-by-10min-window.sql %}</code></pre>
++++++
diff --git a/_includes/tutorials/hopping-windows/flinksql/markup/test/populate-resource-temperature-by-10min-window.sql.adoc b/_includes/tutorials/hopping-windows/flinksql/markup/test/populate-resource-temperature-by-10min-window.sql.adoc
@@ -0,0 +1,5 @@
+Then create the following file at `src/test/resources/populate-temperature-by-10min-window.sql` that inserts data into the table:
+
++++++
+<pre class="snippet"><code class="groovy">{% include_raw tutorials/hopping-windows/flinksql/code/src/test/resources/populate-temperature-by-10min-window.sql %}</code></pre>
++++++
diff --git a/_includes/tutorials/tumbling-windows/flinksql/code/build.gradle b/_includes/tutorials/tumbling-windows/flinksql/code/build.gradle
@@ -6,6 +6,7 @@ buildscript {
 
 plugins {
     id "java"
+    id 'idea'
 }
 
 sourceCompatibility = JavaVersion.VERSION_11
diff --git a/_includes/tutorials/tumbling-windows/flinksql/code/settings.gradle b/_includes/tutorials/tumbling-windows/flinksql/code/settings.gradle
@@ -7,4 +7,4 @@
  * in the user manual at https://docs.gradle.org/6.7.1/userguide/multi_project_builds.html
  */
 
-rootProject.name = 'aggregating-count-flinksql'
+rootProject.name = 'tumbling-windows-flinksql'
diff --git a/_includes/tutorials/tumbling-windows/flinksql/code/src/test/java/io/confluent/developer/FlinkSqlTumblingWindowTest.java b/_includes/tutorials/tumbling-windows/flinksql/code/src/test/java/io/confluent/developer/FlinkSqlTumblingWindowTest.java
@@ -17,7 +17,8 @@ public void testTumblingWindows() throws Exception {
         Optional.of(kafkaPort), Optional.of(schemaRegistryPort))).await();
     streamTableEnv.executeSql(getResourceFileContents("populate-ratings.sql"));
     streamTableEnv.executeSql(getResourceFileContents("create-ratings-by-6hr-window.sql.template",
-        Optional.of(kafkaPort), Optional.of(schemaRegistryPort)));
+        Optional.of(kafkaPort), Optional.of(schemaRegistryPort))).await();
+    streamTableEnv.executeSql(getResourceFileContents("populate-ratings-by-6hr-window.sql"));
 
     TableResult tableResult = streamTableEnv.executeSql(getResourceFileContents("query-ratings-by-6hr-window.sql"));
 
diff --git a/_includes/tutorials/tumbling-windows/flinksql/code/src/test/resources/create-ratings-by-6hr-window.sql.template b/_includes/tutorials/tumbling-windows/flinksql/code/src/test/resources/create-ratings-by-6hr-window.sql.template
@@ -1,5 +1,10 @@
-CREATE TABLE ratings_by_6hr_window
-WITH (
+CREATE TABLE ratings_by_6hr_window (
+   title STRING,
+    rating_count BIGINT,
+    avg_rating DOUBLE,
+    window_start TIMESTAMP(3),
+    window_end TIMESTAMP(3)
+) WITH (
     'connector' = 'kafka',
     'topic' = 'ratings-by-6hr-window',
     'properties.bootstrap.servers' = 'localhost:KAFKA_PORT',
@@ -10,11 +15,4 @@ WITH (
     'value.format' = 'avro-confluent',
     'value.avro-confluent.url' = 'http://localhost:SCHEMA_REGISTRY_PORT',
     'value.fields-include' = 'ALL'
-) AS
-    SELECT title,
-       COUNT(*) AS rating_count,
-       AVG(rating) AS avg_rating,
-       window_start,
-       window_end
-    FROM TABLE(TUMBLE(TABLE ratings, DESCRIPTOR(ts), INTERVAL '6' HOURS))
-    GROUP BY title, window_start, window_end;
+); 
diff --git a/_includes/tutorials/tumbling-windows/flinksql/code/src/test/resources/populate-ratings-by-6hr-window.sql b/_includes/tutorials/tumbling-windows/flinksql/code/src/test/resources/populate-ratings-by-6hr-window.sql
@@ -0,0 +1,8 @@
+INSERT INTO ratings_by_6hr_window
+    SELECT title,
+       COUNT(*) AS rating_count,
+       AVG(rating) AS avg_rating,
+       window_start,
+       window_end
+    FROM TABLE(TUMBLE(TABLE ratings, DESCRIPTOR(ts), INTERVAL '6' HOURS))
+    GROUP BY title, window_start, window_end;
diff --git a/_includes/tutorials/tumbling-windows/flinksql/code/tutorial-steps/dev/create-ratings-by-6hr-window.sql b/_includes/tutorials/tumbling-windows/flinksql/code/tutorial-steps/dev/create-ratings-by-6hr-window.sql
@@ -1,5 +1,10 @@
-CREATE TABLE ratings_by_6hr_window
-WITH (
+CREATE TABLE ratings_by_6hr_window (
+    title STRING,
+    rating_count BIGINT,
+    avg_rating DOUBLE,
+    window_start TIMESTAMP(3),
+    window_end TIMESTAMP(3)
+) WITH (
     'connector' = 'kafka',
     'topic' = 'ratings-by-6hr-window',
     'properties.bootstrap.servers' = 'broker:9092',
@@ -10,11 +15,4 @@ WITH (
     'value.format' = 'avro-confluent',
     'value.avro-confluent.url' = 'http://schema-registry:8081',
     'value.fields-include' = 'ALL'
-) AS
-    SELECT title,
-       COUNT(*) AS rating_count,
-       AVG(rating) AS avg_rating,
-       window_start,
-       window_end
-    FROM TABLE(TUMBLE(TABLE ratings, DESCRIPTOR(ts), INTERVAL '6' HOURS))
-    GROUP BY title, window_start, window_end;
+);
diff --git a/_includes/tutorials/tumbling-windows/flinksql/code/tutorial-steps/dev/expected-ratings-by-6hr-window.log b/_includes/tutorials/tumbling-windows/flinksql/code/tutorial-steps/dev/expected-ratings-by-6hr-window.log
@@ -1,8 +1,8 @@
-{"title":{"string":"Die Hard"},"rating_count":2,"avg_rating":{"double":6.35},"window_start":1688860800000,"window_end":1688882400000}
-{"title":{"string":"The Big Lebowski"},"rating_count":1,"avg_rating":{"double":4.2},"window_start":1688860800000,"window_end":1688882400000}
-{"title":{"string":"Die Hard"},"rating_count":1,"avg_rating":{"double":5.1},"window_start":1688882400000,"window_end":1688904000000}
-{"title":{"string":"Tree of Life"},"rating_count":2,"avg_rating":{"double":5.25},"window_start":1688882400000,"window_end":1688904000000}
-{"title":{"string":"The Big Lebowski"},"rating_count":1,"avg_rating":{"double":9.9},"window_start":1688882400000,"window_end":1688904000000}
-{"title":{"string":"A Walk in the Clouds"},"rating_count":2,"avg_rating":{"double":4.8},"window_start":1688904000000,"window_end":1688925600000}
-{"title":{"string":"Super Mario Bros."},"rating_count":1,"avg_rating":{"double":3.5},"window_start":1688925600000,"window_end":1688947200000}
+{"title":{"string":"Die Hard"},"rating_count":{"long":2},"avg_rating":{"double":6.35},"window_start":{"long":1688860800000},"window_end":{"long":1688882400000}}
+{"title":{"string":"The Big Lebowski"},"rating_count":{"long":1},"avg_rating":{"double":4.2},"window_start":{"long":1688860800000},"window_end":{"long":1688882400000}}
+{"title":{"string":"Die Hard"},"rating_count":{"long":1},"avg_rating":{"double":5.1},"window_start":{"long":1688882400000},"window_end":{"long":1688904000000}}
+{"title":{"string":"Tree of Life"},"rating_count":{"long":2},"avg_rating":{"double":5.25},"window_start":{"long":1688882400000},"window_end":{"long":1688904000000}}
+{"title":{"string":"The Big Lebowski"},"rating_count":{"long":1},"avg_rating":{"double":9.9},"window_start":{"long":1688882400000},"window_end":{"long":1688904000000}}
+{"title":{"string":"A Walk in the Clouds"},"rating_count":{"long":2},"avg_rating":{"double":4.8},"window_start":{"long":1688904000000},"window_end":{"long":1688925600000}}
+{"title":{"string":"Super Mario Bros."},"rating_count":{"long":1},"avg_rating":{"double":3.5},"window_start":{"long":1688925600000},"window_end":{"long":1688947200000}}
 Processed a total of 7 messages
diff --git a/_includes/tutorials/tumbling-windows/flinksql/code/tutorial-steps/dev/populate-ratings-by-6hr-window.sql b/_includes/tutorials/tumbling-windows/flinksql/code/tutorial-steps/dev/populate-ratings-by-6hr-window.sql
@@ -0,0 +1,8 @@
+INSERT INTO ratings_by_6hr_window 
+    SELECT title,
+       COUNT(*) AS rating_count,
+       AVG(rating) AS avg_rating,
+       window_start,
+       window_end
+    FROM TABLE(TUMBLE(TABLE ratings, DESCRIPTOR(ts), INTERVAL '6' HOURS))
+    GROUP BY title, window_start, window_end;
diff --git a/_includes/tutorials/tumbling-windows/flinksql/markup/dev/create-ratings-by-6hr-window.adoc b/_includes/tutorials/tumbling-windows/flinksql/markup/dev/create-ratings-by-6hr-window.adoc
@@ -1,4 +1,4 @@
-Since the output of our transient query looks right, the next step is to make it persistent with the following `CREATE TABLE AS SELECT` statement. Go ahead and execute the following statement in your Flink SQL session:
+Since the output of our transient query looks right, the next step is to make it persistent with the following `CREATE TABLE` statement. Go ahead and execute the following statement in your Flink SQL session:
 
 +++++
 <pre class="snippet"><code class="sql">{% include_raw tutorials/tumbling-windows/flinksql/code/tutorial-steps/dev/create-ratings-by-6hr-window.sql %}</code></pre>
diff --git a/_includes/tutorials/tumbling-windows/flinksql/markup/dev/populate-ratings-by-6hr-window.adoc b/_includes/tutorials/tumbling-windows/flinksql/markup/dev/populate-ratings-by-6hr-window.adoc
@@ -0,0 +1,5 @@
+Now let's insert data into the table with the following statement:
+
++++++
+<pre class="snippet"><code class="sql">{% include_raw tutorials/tumbling-windows/flinksql/code/tutorial-steps/dev/populate-ratings-by-6hr-window.sql %}</code></pre>
++++++
diff --git a/_includes/tutorials/tumbling-windows/flinksql/markup/test/create-resource-populate-ratings-by-6hr-window.sql.adoc b/_includes/tutorials/tumbling-windows/flinksql/markup/test/create-resource-populate-ratings-by-6hr-window.sql.adoc
@@ -0,0 +1,5 @@
+Next create the following file at `src/test/resources/populate-ratings-by-6hr-window.sql` to populate the table.
+
++++++
+<pre class="snippet"><code class="groovy">{% include_raw tutorials/tumbling-windows/flinksql/code/src/test/resources/populate-ratings-by-6hr-window.sql %}</code></pre>
++++++

Original file line number	Diff line number	Diff line change
`@@ -1,4 +1,4 @@`
`1`		-Since the output of our transient query looks right, the next step is to make it persistent with the following `CREATE TABLE AS SELECT` statement. Go ahead and execute the following statement in your Flink SQL session:
	`1`	+Since the output of our transient query looks right, the next step is to make it persistent with the following `CREATE TABLE` statement. Go ahead and execute the following statement in your Flink SQL session:
`2`	`2`
`3`	`3`	`+++++`
`4`	`4`	`<pre class="snippet"><code class="sql">{% include_raw tutorials/hopping-windows/flinksql/code/tutorial-steps/dev/create-temperature-by-10min-window.sql %}</code></pre>`