feat: add SOURCE streams/tables to query anonymizer

ksqldb-engine/src/main/java/io/confluent/ksql/engine/rewrite/QueryAnonymizer.java

@@ -571,6 +571,11 @@ public String visitCreateStreamAs(final CreateStreamAsContext context) {
     public String visitCreateStream(final CreateStreamContext context) {
       final StringBuilder stringBuilder = new StringBuilder("CREATE ");
 
+      // optional source
+      if (context.SOURCE() != null) {
+        stringBuilder.append("SOURCE ");
+      }
+
       // optional replace
       if (context.OR() != null && context.REPLACE() != null) {
         stringBuilder.append("OR REPLACE ");
@@ -637,6 +642,11 @@ public String visitCreateTableAs(final CreateTableAsContext context) {
     public String visitCreateTable(final CreateTableContext context) {
       final StringBuilder stringBuilder = new StringBuilder("CREATE ");
 
+      // optional source
+      if (context.SOURCE() != null) {
+        stringBuilder.append("SOURCE ");
+      }
+
       // optional replace
       if (context.OR() != null && context.REPLACE() != null) {
         stringBuilder.append("OR REPLACE ");

ksqldb-engine/src/test/java/io/confluent/ksql/engine/rewrite/QueryAnonymizerTest.java

@@ -160,6 +160,15 @@ public void shouldAnonymizeCreateStreamQueryCorrectly() {
     Approvals.verify(output);
   }
 
+  @Test
+  public void shouldAnonymizeCreateSourceStreamQueryCorrectly() {
+    final String output = anon.anonymize(
+        "CREATE SOURCE STREAM my_stream (profileId VARCHAR, latitude DOUBLE, longitude DOUBLE)\n"
+            + "WITH (kafka_topic='locations', value_format='json');");
+
+    Approvals.verify(output);
+  }
+
   @Test
   public void shouldAnonymizeCreateStreamAsQueryCorrectly() {
     final String output = anon.anonymize(
@@ -181,6 +190,15 @@ public void shouldAnonymizeCreateTableCorrectly() {
     Approvals.verify(output);
   }
 
+  @Test
+  public void shouldAnonymizeCreateSourceTableCorrectly() {
+    final String output = anon.anonymize(
+        "CREATE SOURCE TABLE my_table (profileId VARCHAR, latitude DOUBLE, longitude DOUBLE)\n"
+            + "WITH (kafka_topic='locations', value_format='json');");
+
+    Approvals.verify(output);
+  }
+
   @Test
   public void shouldAnonymizeCreateTableAsQueryCorrectly() {
     final String output = anon.anonymize(

ksqldb-engine/src/test/java/io/confluent/ksql/engine/rewrite/QueryAnonymizerTest.shouldAnonymizeCreateSourceStreamQueryCorrectly.approved.txt

@@ -0,0 +1 @@
+CREATE SOURCE STREAM stream1 (column1 VARCHAR, column2 DOUBLE, column3 DOUBLE) WITH (KAFKA_TOPIC='[string]', VALUE_FORMAT='[string]');

ksqldb-engine/src/test/java/io/confluent/ksql/engine/rewrite/QueryAnonymizerTest.shouldAnonymizeCreateSourceTableCorrectly.approved.txt

@@ -0,0 +1 @@
+CREATE SOURCE TABLE table1 (column1 VARCHAR, column2 DOUBLE, column3 DOUBLE) WITH (KAFKA_TOPIC='[string]', VALUE_FORMAT='[string]');