[Fix-3006] Fix CDCSOURCE does not supports delete event after optimiz…

…e code (#3012) Co-authored-by: wenmo <32723967+wenmo@users.noreply.github.com>
DataLinkDC · Jan 17, 2024 · adf013f · adf013f
1 parent 3e50168
commit adf013f
Show file tree

Hide file tree

Showing 2 changed files with 4 additions and 2 deletions.
diff --git a/dinky-cdc/dinky-cdc-core/src/main/java/org/dinky/cdc/sql/SQLSinkBuilder.java b/dinky-cdc/dinky-cdc-core/src/main/java/org/dinky/cdc/sql/SQLSinkBuilder.java
@@ -71,7 +71,8 @@ private String addSourceTableView(
             CustomTableEnvironment customTableEnvironment, DataStream<Row> rowDataDataStream, Table table) {
         // 上游表名称
         String viewName = "VIEW_" + table.getSchemaTableNameWithUnderline();
-        customTableEnvironment.createTemporaryView(viewName, rowDataDataStream);
+        customTableEnvironment.createTemporaryView(
+                viewName, customTableEnvironment.fromChangelogStream(rowDataDataStream));
         logger.info("Create {} temporaryView successful...", viewName);
         return viewName;
     }

diff --git a/dinky-cdc/dinky-cdc-core/src/main/java/org/dinky/cdc/sql/catalog/SQLCatalogSinkBuilder.java b/dinky-cdc/dinky-cdc-core/src/main/java/org/dinky/cdc/sql/catalog/SQLCatalogSinkBuilder.java
@@ -70,7 +70,8 @@ public void addTableSink(
         String sinkTableName = catalogName + "." + sinkSchemaName + "." + tableName;
         String viewName = "VIEW_" + table.getSchemaTableNameWithUnderline();
 
-        customTableEnvironment.createTemporaryView(viewName, rowDataDataStream);
+        customTableEnvironment.createTemporaryView(
+                viewName, customTableEnvironment.fromChangelogStream(rowDataDataStream));
         logger.info("Create {} temporaryView successful...", viewName);
 
         createInsertOperations(customTableEnvironment, table, viewName, sinkTableName);