[feature-#1926] add paimon dependency (#1943)

Author: dujie

bin/start-chunjun

@@ -33,7 +33,7 @@ else
   fi
 fi
 
-JAR_DIR=$CHUNJUN_HOME/lib/*
+JAR_DIR=$CHUNJUN_HOME/client/*
 CLASS_NAME=com.dtstack.chunjun.client.Launcher
 
 echo "ChunJun starting ..."

bin/submit.sh

@@ -46,9 +46,9 @@ fi
 # 1.In yarn-session case, JAR_DIR can not be found
 # 2.In other cases, JAR_DIR can be found
 if [ $CHUNJUN_DEPLOY_MODE -eq 1 ]; then
-  JAR_DIR=$CHUNJUN_HOME/lib/chunjun-clients.jar:$CHUNJUN_HOME/lib/*
+  JAR_DIR=$CHUNJUN_HOME/client/chunjun-clients.jar:$CHUNJUN_HOME/lib/*
 else
-  JAR_DIR=$CHUNJUN_HOME/../lib/chunjun-clients.jar:$CHUNJUN_HOME/../lib/*
+  JAR_DIR=$CHUNJUN_HOME/../client/chunjun-clients.jar:$CHUNJUN_HOME/../client/*
 fi
 
 CLASS_NAME=com.dtstack.chunjun.client.Launcher

chunjun-catalog/chunjun-catalog-iceberg/pom.xml

@@ -16,16 +16,16 @@
 		<maven.compiler.source>8</maven.compiler.source>
 		<maven.compiler.target>8</maven.compiler.target>
 		<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+		<iceberg.version>1.5.2</iceberg.version>
 	</properties>
 
 	<description>iceberg</description>
 
 	<dependencies>
-		<!-- https://mvnrepository.com/artifact/org.apache.iceberg/iceberg-core -->
 		<dependency>
 			<groupId>org.apache.iceberg</groupId>
-			<artifactId>iceberg-core</artifactId>
-			<version>1.10.0</version>
+			<artifactId>iceberg-flink-runtime-1.16</artifactId>
+			<version>${iceberg.version}</version>
 		</dependency>
 	</dependencies>
 

chunjun-catalog/chunjun-catalog-paimon/pom.xml

@@ -22,6 +22,13 @@
 	<description>paimon</description>
 
 	<dependencies>
+		<dependency>
+			<groupId>log4j</groupId>
+			<artifactId>log4j</artifactId>
+			<version>${log4j.version}</version>
+			<!--<scope>provided</scope>-->
+		</dependency>
+
 		<!-- https://mvnrepository.com/artifact/org.apache.paimon/paimon-flink-1.16 -->
 		<dependency>
 			<groupId>org.apache.paimon</groupId>

chunjun-clients/pom.xml

@@ -44,6 +44,18 @@
 			</exclusions>
 		</dependency>
 
+		<dependency>
+			<groupId>org.apache.flink</groupId>
+			<artifactId>flink-connector-base</artifactId>
+			<version>${flink.version}</version>
+		</dependency>
+
+		<dependency>
+			<groupId>org.apache.flink</groupId>
+			<artifactId>flink-connector-files</artifactId>
+			<version>${flink.version}</version>
+		</dependency>
+
 		<dependency>
 			<groupId>com.dtstack.chunjun</groupId>
 			<artifactId>chunjun-core</artifactId>

chunjun-core/pom.xml

@@ -191,6 +191,18 @@
 			<artifactId>flink-table-planner-loader</artifactId>
 			<version>${flink.version}</version>
 		</dependency>
+
+		<dependency>
+			<groupId>org.apache.flink</groupId>
+			<artifactId>flink-connector-base</artifactId>
+			<version>${flink.version}</version>
+		</dependency>
+
+		<dependency>
+			<groupId>org.apache.flink</groupId>
+			<artifactId>flink-connector-files</artifactId>
+			<version>${flink.version}</version>
+		</dependency>
 		<!-- flink sql -->
 
 		<!--flink与Hadoop mapReduce兼容包-->

chunjun-core/src/main/java/com/dtstack/chunjun/constants/ConstantValue.java

@@ -72,6 +72,8 @@ public class ConstantValue {
 
     public static final String CONNECTOR_DIR_NAME = "connector";
 
+    public static final String CATALOG_DIR_NAME = "catalog";
+
     public static final String FORMAT_DIR_NAME = "formats";
 
     public static final String DIRTY_DATA_DIR_NAME = "dirty-data-collector";

chunjun-core/src/main/java/org/apache/flink/table/factories/FactoryUtil.java

@@ -546,6 +546,12 @@ public static <T extends Factory> T discoverFactory(
                         classLoader,
                         ConstantValue.CONNECTOR_DIR_NAME);
             }
+        } else if (factoryClass.isAssignableFrom(CatalogFactory.class)) {
+            com.dtstack.chunjun.util.FactoryHelper factoryHelper = factoryHelperThreadLocal.get();
+            if (factoryHelper != null) {
+                factoryHelper.registerCachedFile(
+                        unconvertedFactoryIdentifier, classLoader, ConstantValue.CATALOG_DIR_NAME);
+            }
         } else {
             com.dtstack.chunjun.util.FactoryHelper factoryHelper = factoryHelperThreadLocal.get();
             if (factoryHelper != null) {

chunjun-examples/sql/catalog/paimon_stream.sql

@@ -0,0 +1,25 @@
+CREATE
+CATALOG my_catalog WITH (
+    'type'='paimon',
+    'metastore' = 'filesystem',
+    'warehouse'='file:/tmp/paimon'
+);
+
+USE CATALOG my_catalog;
+
+-- create a word count table
+CREATE TABLE if not exists word_count
+(
+    word STRING PRIMARY KEY NOT ENFORCED,
+    cnt  BIGINT
+);
+
+CREATE
+TEMPORARY TABLE sink (
+    word STRING,
+   cnt  BIGINT
+) WITH (
+    'connector' = 'stream-x'
+);
+
+insert into sink  SELECT word, cnt FROM word_count ;

chunjun-examples/sql/catalog/stream_paimon.sql

@@ -0,0 +1,33 @@
+CREATE
+CATALOG my_catalog WITH (
+    'type'='paimon',
+    'metastore' = 'filesystem',
+    'warehouse'='file:/tmp/paimon'
+);
+
+USE CATALOG my_catalog;
+
+-- create a word count table
+CREATE TABLE if not exists word_count
+(
+    word STRING PRIMARY KEY NOT ENFORCED,
+    cnt  BIGINT
+);
+
+
+CREATE
+TEMPORARY TABLE word_table (
+    word STRING
+) WITH (
+    'connector' = 'datagen',
+    'fields.word.length' = '1'
+);
+
+
+
+SET execution.checkpointing.interval=10000;
+
+INSERT INTO word_count
+SELECT word, COUNT(*)
+FROM word_table
+GROUP BY word;