Add testInsertSelectFromHudi case

Author: HamaWhiteGG

lineage-flink1.16.x/pom.xml

@@ -15,6 +15,7 @@
         <hadoop.version>3.2.2</hadoop.version>
         <hive.version>3.1.2</hive.version>
         <mysql.cdc.version>2.2.1</mysql.cdc.version>
+        <hudi.version>0.13.1</hudi.version>
         <paimon.version>0.4-SNAPSHOT</paimon.version>
     </properties>
 
@@ -84,6 +85,12 @@
             <version>${flink.version}</version>
         </dependency>
 
+        <dependency>
+            <groupId>org.apache.hudi</groupId>
+            <artifactId>hudi-flink1.16-bundle</artifactId>
+            <version>${hudi.version}</version>
+        </dependency>
+
         <dependency>
             <groupId>org.apache.flink</groupId>
             <artifactId>flink-json</artifactId>

lineage-flink1.16.x/src/test/java/com/hw/lineage/flink/basic/AbstractBasicTest.java

@@ -169,10 +169,7 @@ protected void createTableOfDimMysqlCompany() {
     }
 
     /**
-     * Create Hudi sink table dwd_hudi_user.
-     * <p>
-     * Since flink1.16 does not support hudi, temporarily change the connector to print,
-     * but the table name remains unchanged
+     * Create Hudi sink table dwd_hudi_users
      */
     protected void createTableOfDwdHudiUsers() {
         context.execute("DROP TABLE IF EXISTS dwd_hudi_users");
@@ -185,7 +182,27 @@ protected void createTableOfDwdHudiUsers() {
                 "       ts                  TIMESTAMP(3)                    ," +
                 "        `partition`        VARCHAR(20)                      " +
                 ") PARTITIONED BY (`partition`) WITH ( " +
-                "       'connector' = 'print'                                " +
+                "       'connector' = 'hudi'                                    ," +
+                "       'table.type' = 'COPY_ON_WRITE'                          ," +
+                "       'path' = '/hudi/users'                                  ," +
+                "       'read.streaming.enabled' = 'true'                       ," +
+                "       'read.streaming.check-interval' = '1'                    " +
+                ")");
+    }
+
+    /**
+     * Create Hudi sink table dws_users_cnt
+     */
+    protected void createTableOfDwsHudiUsersCnt() {
+        context.execute("DROP TABLE IF EXISTS dws_users_cnt");
+
+        context.execute("CREATE TABLE IF NOT EXISTS  dws_users_cnt ( " +
+                "       id                  BIGINT PRIMARY KEY NOT ENFORCED ," +
+                "       name_cnt            BIGINT                          ," +
+                "       company_name_cnt    BIGINT                           " +
+                ") WITH ( " +
+                "       'connector' = 'hudi'                                    ," +
+                "       'table.type' = 'MERGE_ON_READ'                           " +
                 ")");
     }
 

lineage-flink1.16.x/src/test/java/com/hw/lineage/flink/lookup/join/LookupJoinTest.java

@@ -39,6 +39,9 @@ public void createTable() {
 
         // create hudi sink table dwd_hudi_users
         createTableOfDwdHudiUsers();
+
+        // Create Hudi sink table dws_users_cnt
+        createTableOfDwsHudiUsersCnt();
     }
 
     /**
@@ -79,4 +82,27 @@ public void testInsertSelectTwoLookupJoin() {
         analyzeLineage(sql, expectedArray);
     }
 
+    /**
+     * insert-select-from-hudi
+     */
+    @Test
+    public void testInsertSelectFromHudi() {
+        String sql = "INSERT into dws_users_cnt " +
+                "SELECT " +
+                "       id," +
+                "       COUNT(DISTINCT name), " +
+                "       COUNT(DISTINCT company_name) " +
+                "FROM" +
+                "       dwd_hudi_users " +
+                "GROUP BY " +
+                "       id";
+
+        String[][] expectedArray = {
+                {"dwd_hudi_users", "id", "dws_users_cnt", "id"},
+                {"dwd_hudi_users", "name", "dws_users_cnt", "name_cnt", "COUNT(DISTINCT name)"},
+                {"dwd_hudi_users", "company_name", "dws_users_cnt", "company_name_cnt", "COUNT(DISTINCT company_name)"}
+        };
+
+        analyzeLineage(sql, expectedArray);
+    }
 }