[AMORO-1966] Add PostgreSQL as system database (apache#1967)

* add postgresSQL * expose db type for SqlSessionFactoryProvider * add postgres docs * review * review * review
shendanfengg · Sep 15, 2023 · afc4e6f · afc4e6f
1 parent a0d9c59
commit afc4e6f
Show file tree

Hide file tree

Showing 8 changed files with 413 additions and 7 deletions.
diff --git a/ams/server/pom.xml b/ams/server/pom.xml
@@ -40,6 +40,7 @@
         <junit5.version>5.9.1</junit5.version>
         <mybatis.version>3.5.6</mybatis.version>
         <mysql-jdbc.version>8.0.30</mysql-jdbc.version>
+        <postgres-jdbc.version>42.5.1</postgres-jdbc.version>
         <derby-jdbc.version>10.13.1.1</derby-jdbc.version>
         <commons-dbcp2.version>2.9.0</commons-dbcp2.version>
         <snakeyaml.version>1.30</snakeyaml.version>
@@ -198,6 +199,12 @@
             <artifactId>mysql-connector-java</artifactId>
         </dependency>
 
+        <dependency>
+            <groupId>org.postgresql</groupId>
+            <artifactId>postgresql</artifactId>
+            <version>${postgres-jdbc.version}</version>
+        </dependency>
+
         <dependency>
             <groupId>org.apache.derby</groupId>
             <artifactId>derby</artifactId>

diff --git a/ams/server/src/main/java/com/netease/arctic/server/ArcticManagementConf.java b/ams/server/src/main/java/com/netease/arctic/server/ArcticManagementConf.java
@@ -272,5 +272,5 @@ public class ArcticManagementConf {
 
   public static final String DB_TYPE_DERBY = "derby";
   public static final String DB_TYPE_MYSQL = "mysql";
-
+  public static final String DB_TYPE_POSTGRES = "postgres";
 }
diff --git a/...server/src/main/java/com/netease/arctic/server/persistence/SqlSessionFactoryProvider.java b/...server/src/main/java/com/netease/arctic/server/persistence/SqlSessionFactoryProvider.java
@@ -29,6 +29,7 @@
 import com.netease.arctic.server.persistence.mapper.TableMetaMapper;
 import com.netease.arctic.server.utils.Configurations;
 import org.apache.commons.dbcp2.BasicDataSource;
+import org.apache.commons.lang3.StringUtils;
 import org.apache.commons.pool2.impl.BaseObjectPoolConfig;
 import org.apache.ibatis.jdbc.ScriptRunner;
 import org.apache.ibatis.mapping.Environment;
@@ -54,20 +55,24 @@ public class SqlSessionFactoryProvider {
 
   private static final String DERBY_INIT_SQL_SCRIPT = "derby/ams-derby-init.sql";
   private static final String MYSQL_INIT_SQL_SCRIPT = "mysql/ams-mysql-init.sql";
+  private static final String POSTGRES_INIT_SQL_SCRIPT = "postgres/ams-postgres-init.sql";
 
   private static final SqlSessionFactoryProvider INSTANCE = new SqlSessionFactoryProvider();
 
   public static SqlSessionFactoryProvider getInstance() {
     return INSTANCE;
   }
 
+  private static String dbType;
+
   private volatile SqlSessionFactory sqlSessionFactory;
 
   public void init(Configurations config) {
     BasicDataSource dataSource = new BasicDataSource();
     dataSource.setUrl(config.getString(ArcticManagementConf.DB_CONNECTION_URL));
     dataSource.setDriverClassName(config.getString(ArcticManagementConf.DB_DRIVER_CLASS_NAME));
-    if (ArcticManagementConf.DB_TYPE_MYSQL.equals(config.getString(ArcticManagementConf.DB_TYPE))) {
+    dbType = config.getString(ArcticManagementConf.DB_TYPE);
+    if (ArcticManagementConf.DB_TYPE_MYSQL.equals(dbType) || ArcticManagementConf.DB_TYPE_POSTGRES.equals(dbType)) {
       dataSource.setUsername(config.getString(ArcticManagementConf.DB_USER_NAME));
       dataSource.setPassword(config.getString(ArcticManagementConf.DB_PASSWORD));
     }
@@ -127,8 +132,12 @@ private void createTablesIfNeed(Configurations config) {
         query = String.format(
             "SELECT 1 FROM information_schema.tables WHERE table_schema = '%s' AND table_name = '%s'",
             connection.getCatalog(), "CATALOG_METADATA");
+      } else if (ArcticManagementConf.DB_TYPE_POSTGRES.equals(dbTypeConfig)) {
+        query = String.format(
+            "SELECT 1 FROM information_schema.tables WHERE table_schema = %s AND table_name = '%s'",
+            "current_schema()", "catalog_metadata");
       }
-      try (ResultSet rs = statement.executeQuery(query);) {
+      try (ResultSet rs = statement.executeQuery(query)) {
         if (!rs.next()) {
           ScriptRunner runner = new ScriptRunner(connection);
           runner.runScript(new InputStreamReader(Files.newInputStream(
@@ -147,6 +156,8 @@ private String getInitSqlScriptPath(String type) {
       scriptPath = MYSQL_INIT_SQL_SCRIPT;
     } else if (type.equals(ArcticManagementConf.DB_TYPE_DERBY)) {
       scriptPath = DERBY_INIT_SQL_SCRIPT;
+    } else if (type.equals(ArcticManagementConf.DB_TYPE_POSTGRES)) {
+      scriptPath = POSTGRES_INIT_SQL_SCRIPT;
     }
     URL scriptUrl = ClassLoader.getSystemResource(scriptPath);
     if (scriptUrl == null) {
@@ -161,4 +172,11 @@ public SqlSessionFactory get() {
         "Persistent configuration is not initialized yet.");
     return sqlSessionFactory;
   }
+
+  public static String getDbType() {
+    Preconditions.checkState(
+        StringUtils.isNotBlank(dbType),
+        "Persistent configuration is not initialized yet.");
+    return dbType;
+  }
 }
diff --git a/...rc/main/java/com/netease/arctic/server/persistence/converter/Object2ByteArrayConvert.java b/...rc/main/java/com/netease/arctic/server/persistence/converter/Object2ByteArrayConvert.java
@@ -1,5 +1,7 @@
 package com.netease.arctic.server.persistence.converter;
 
+import com.netease.arctic.server.ArcticManagementConf;
+import com.netease.arctic.server.persistence.SqlSessionFactoryProvider;
 import com.netease.arctic.server.utils.CompressUtil;
 import com.netease.arctic.utils.SerializationUtil;
 import org.apache.ibatis.type.JdbcType;
@@ -17,7 +19,11 @@ public class Object2ByteArrayConvert<T> implements TypeHandler<T> {
   @Override
   public void setParameter(PreparedStatement ps, int i, T parameter, JdbcType jdbcType) throws SQLException {
     if (parameter == null) {
-      ps.setNull(i, Types.BLOB);
+      if (SqlSessionFactoryProvider.getDbType().equals(ArcticManagementConf.DB_TYPE_POSTGRES)) {
+        ps.setNull(i, Types.BINARY);
+      } else {
+        ps.setNull(i, Types.BLOB);
+      }
       return;
     }