[#4101] feat(core): Support PostgreSQL storage backend (#4611)

### What changes were proposed in this pull request?

- Support PostgreSQL JDBC backend.

### Why are the changes needed?

User need.

Fix: #4101 

### Does this PR introduce _any_ user-facing change?

N/A

### How was this patch tested?

Existing ITs

---------

Co-authored-by: Jerry Shao <[email protected]>

.github/workflows/backend-integration-test.yml

@@ -61,12 +61,15 @@ jobs:
         # Integration test for AMD64 architecture
         architecture: [linux/amd64]
         java-version: [ 8, 11, 17 ]
+        backend: [ h2, mysql, postgresql ]
         test-mode: [ embedded, deploy ]
-        include:
+        exclude:
           - test-mode: 'embedded'
-            backend: 'h2'
-          - test-mode: 'deploy'
             backend: 'mysql'
+          - test-mode: 'embedded'
+            backend: 'postgresql'
+          - test-mode: 'deploy'
+            backend: 'h2'
 
     env:
       PLATFORM: ${{ matrix.architecture }}
@@ -132,11 +135,14 @@ jobs:
         architecture: [ linux/amd64 ]
         java-version: [ 17 ]
         test-mode: [ embedded, deploy ]
-        include:
+        backend: [ h2, mysql, postgresql ]
+        exclude:
           - test-mode: 'embedded'
-            backend: 'h2'
-          - test-mode: 'deploy'
             backend: 'mysql'
+          - test-mode: 'embedded'
+            backend: 'postgresql'
+          - test-mode: 'deploy'
+            backend: 'h2'
 
     env:
       PLATFORM: ${{ matrix.architecture }}

catalogs/catalog-hadoop/build.gradle.kts

@@ -79,6 +79,7 @@ dependencies {
   testImplementation(libs.mockito.core)
   testImplementation(libs.mockito.inline)
   testImplementation(libs.mysql.driver)
+  testImplementation(libs.postgresql.driver)
   testImplementation(libs.junit.jupiter.api)
   testImplementation(libs.junit.jupiter.params)
   testImplementation(libs.testcontainers)

catalogs/catalog-hive/build.gradle.kts

@@ -132,6 +132,7 @@ dependencies {
   testImplementation(libs.junit.jupiter.api)
   testImplementation(libs.mockito.core)
   testImplementation(libs.mysql.driver)
+  testImplementation(libs.postgresql.driver)
 
   testImplementation("org.apache.spark:spark-hive_$scalaVersion:$sparkVersion") {
     exclude("org.apache.hadoop")

catalogs/catalog-jdbc-doris/build.gradle.kts

@@ -52,6 +52,7 @@ dependencies {
   testImplementation(libs.junit.jupiter.api)
   testImplementation(libs.junit.jupiter.params)
   testImplementation(libs.mysql.driver)
+  testImplementation(libs.postgresql.driver)
   testImplementation(libs.testcontainers)
   testImplementation(libs.testcontainers.mysql)
 

catalogs/catalog-jdbc-mysql/build.gradle.kts

@@ -55,6 +55,7 @@ dependencies {
   testImplementation(libs.junit.jupiter.api)
   testImplementation(libs.junit.jupiter.params)
   testImplementation(libs.mysql.driver)
+  testImplementation(libs.postgresql.driver)
   testImplementation(libs.testcontainers)
   testImplementation(libs.testcontainers.mysql)
 

catalogs/catalog-kafka/build.gradle.kts

@@ -51,6 +51,7 @@ dependencies {
   testImplementation(libs.mockito.core)
   testImplementation(libs.mockito.inline)
   testImplementation(libs.mysql.driver)
+  testImplementation(libs.postgresql.driver)
   testImplementation(libs.testcontainers)
   testImplementation(libs.testcontainers.mysql)
 

catalogs/catalog-lakehouse-paimon/build.gradle.kts

@@ -78,6 +78,7 @@ dependencies {
   testImplementation(libs.slf4j.api)
   testImplementation(libs.junit.jupiter.api)
   testImplementation(libs.mysql.driver)
+  testImplementation(libs.postgresql.driver)
   testImplementation(libs.bundles.log4j)
   testImplementation(libs.junit.jupiter.params)
   testImplementation(libs.testcontainers)

core/src/main/java/org/apache/gravitino/storage/relational/converters/PostgreSQLExceptionConverter.java

@@ -0,0 +1,47 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *  http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.gravitino.storage.relational.converters;
+
+import java.io.IOException;
+import java.sql.SQLException;
+import org.apache.gravitino.Entity;
+import org.apache.gravitino.EntityAlreadyExistsException;
+
+/**
+ * Exception converter to Apache Gravitino exception for PostgreSQL. The definition of error codes
+ * can be found in the document: <a
+ * href="https://www.postgresql.org/docs/8.4/errcodes-appendix.html">error code of PostgreSQL</a>
+ */
+public class PostgreSQLExceptionConverter implements SQLExceptionConverter {
+  private static final int DUPLICATED_ENTRY_ERROR_CODE = 23505;
+
+  @Override
+  @SuppressWarnings("FormatStringAnnotation")
+  public void toGravitinoException(SQLException sqlException, Entity.EntityType type, String name)
+      throws IOException {
+    int errorCode = Integer.valueOf(sqlException.getSQLState());
+    switch (errorCode) {
+      case DUPLICATED_ENTRY_ERROR_CODE:
+        throw new EntityAlreadyExistsException(
+            sqlException, "The %s entity: %s already exists.", type.name(), name);
+      default:
+        throw new IOException(sqlException);
+    }
+  }
+}

core/src/main/java/org/apache/gravitino/storage/relational/converters/SQLExceptionConverterFactory.java

@@ -40,6 +40,8 @@ public static synchronized void initConverter(Config config) {
           converter = new MySQLExceptionConverter();
         } else if (jdbcType.equalsIgnoreCase("h2")) {
           converter = new H2ExceptionConverter();
+        } else if (jdbcType.equalsIgnoreCase("postgresql")) {
+          converter = new PostgreSQLExceptionConverter();
         } else {
           throw new IllegalArgumentException(String.format("Unsupported jdbc type: %s", jdbcType));
         }

core/src/main/java/org/apache/gravitino/storage/relational/mapper/CatalogMetaSQLProviderFactory.java

@@ -22,6 +22,7 @@
 import com.google.common.collect.ImmutableMap;
 import java.util.Map;
 import org.apache.gravitino.storage.relational.JDBCBackend.JDBCBackendType;
+import org.apache.gravitino.storage.relational.mapper.postgresql.CatalogMetaPostgreSQLProvider;
 import org.apache.gravitino.storage.relational.po.CatalogPO;
 import org.apache.gravitino.storage.relational.session.SqlSessionFactoryHelper;
 import org.apache.ibatis.annotations.Param;
@@ -32,7 +33,8 @@ public class CatalogMetaSQLProviderFactory {
       METALAKE_META_SQL_PROVIDER_MAP =
           ImmutableMap.of(
               JDBCBackendType.MYSQL, new CatalogMetaMySQLProvider(),
-              JDBCBackendType.H2, new CatalogMetaH2Provider());
+              JDBCBackendType.H2, new CatalogMetaH2Provider(),
+              JDBCBackendType.POSTGRESQL, new CatalogMetaPostgreSQLProvider());
 
   public static CatalogMetaBaseSQLProvider getProvider() {
     String databaseId =

core/src/main/java/org/apache/gravitino/storage/relational/mapper/FilesetMetaSQLProviderFactory.java

@@ -22,6 +22,7 @@
 import com.google.common.collect.ImmutableMap;
 import java.util.Map;
 import org.apache.gravitino.storage.relational.JDBCBackend.JDBCBackendType;
+import org.apache.gravitino.storage.relational.mapper.postgresql.FilesetMetaPostgreSQLProvider;
 import org.apache.gravitino.storage.relational.po.FilesetPO;
 import org.apache.gravitino.storage.relational.session.SqlSessionFactoryHelper;
 import org.apache.ibatis.annotations.Param;
@@ -31,7 +32,8 @@ public class FilesetMetaSQLProviderFactory {
       METALAKE_META_SQL_PROVIDER_MAP =
           ImmutableMap.of(
               JDBCBackendType.MYSQL, new FilesetMetaMySQLProvider(),
-              JDBCBackendType.H2, new FilesetMetaH2Provider());
+              JDBCBackendType.H2, new FilesetMetaH2Provider(),
+              JDBCBackendType.POSTGRESQL, new FilesetMetaPostgreSQLProvider());
 
   public static FilesetMetaBaseSQLProvider getProvider() {
     String databaseId =

core/src/main/java/org/apache/gravitino/storage/relational/mapper/FilesetVersionSQLProviderFactory.java

@@ -21,6 +21,7 @@
 import com.google.common.collect.ImmutableMap;
 import java.util.Map;
 import org.apache.gravitino.storage.relational.JDBCBackend.JDBCBackendType;
+import org.apache.gravitino.storage.relational.mapper.postgresql.FilesetVersionPostgreSQLProvider;
 import org.apache.gravitino.storage.relational.po.FilesetVersionPO;
 import org.apache.gravitino.storage.relational.session.SqlSessionFactoryHelper;
 import org.apache.ibatis.annotations.Param;
@@ -30,7 +31,8 @@ public class FilesetVersionSQLProviderFactory {
       METALAKE_META_SQL_PROVIDER_MAP =
           ImmutableMap.of(
               JDBCBackendType.MYSQL, new FilesetVersionMySQLProvider(),
-              JDBCBackendType.H2, new FilesetVersionH2Provider());
+              JDBCBackendType.H2, new FilesetVersionH2Provider(),
+              JDBCBackendType.POSTGRESQL, new FilesetVersionPostgreSQLProvider());
 
   public static FilesetVersionBaseSQLProvider getProvider() {
     String databaseId =

core/src/main/java/org/apache/gravitino/storage/relational/mapper/GroupMetaSQLProviderFactory.java

@@ -21,6 +21,7 @@
 import com.google.common.collect.ImmutableMap;
 import java.util.Map;
 import org.apache.gravitino.storage.relational.JDBCBackend.JDBCBackendType;
+import org.apache.gravitino.storage.relational.mapper.postgresql.GroupMetaPostgreSQLProvider;
 import org.apache.gravitino.storage.relational.po.GroupPO;
 import org.apache.gravitino.storage.relational.session.SqlSessionFactoryHelper;
 import org.apache.ibatis.annotations.Param;
@@ -30,7 +31,8 @@ public class GroupMetaSQLProviderFactory {
       METALAKE_META_SQL_PROVIDER_MAP =
           ImmutableMap.of(
               JDBCBackendType.MYSQL, new GroupMetaMySQLProvider(),
-              JDBCBackendType.H2, new GroupMetaH2Provider());
+              JDBCBackendType.H2, new GroupMetaH2Provider(),
+              JDBCBackendType.POSTGRESQL, new GroupMetaPostgreSQLProvider());
 
   public static GroupMetaBaseSQLProvider getProvider() {
     String databaseId =

core/src/main/java/org/apache/gravitino/storage/relational/mapper/GroupRoleRelSQLProviderFactory.java

@@ -22,6 +22,7 @@
 import java.util.List;
 import java.util.Map;
 import org.apache.gravitino.storage.relational.JDBCBackend.JDBCBackendType;
+import org.apache.gravitino.storage.relational.mapper.postgresql.GroupRoleRelPostgreSQLProvider;
 import org.apache.gravitino.storage.relational.po.GroupRoleRelPO;
 import org.apache.gravitino.storage.relational.session.SqlSessionFactoryHelper;
 import org.apache.ibatis.annotations.Param;
@@ -32,7 +33,8 @@ public class GroupRoleRelSQLProviderFactory {
       METALAKE_META_SQL_PROVIDER_MAP =
           ImmutableMap.of(
               JDBCBackendType.MYSQL, new GroupRoleRelMySQLProvider(),
-              JDBCBackendType.H2, new GroupRoleRelH2Provider());
+              JDBCBackendType.H2, new GroupRoleRelH2Provider(),
+              JDBCBackendType.POSTGRESQL, new GroupRoleRelPostgreSQLProvider());
 
   public static GroupRoleRelBaseSQLProvider getProvider() {
     String databaseId =

core/src/main/java/org/apache/gravitino/storage/relational/mapper/MetalakeMetaSQLProviderFactory.java

@@ -22,6 +22,7 @@
 import com.google.common.collect.ImmutableMap;
 import java.util.Map;
 import org.apache.gravitino.storage.relational.JDBCBackend.JDBCBackendType;
+import org.apache.gravitino.storage.relational.mapper.postgresql.MetalakeMetaPostgreSQLProvider;
 import org.apache.gravitino.storage.relational.po.MetalakePO;
 import org.apache.gravitino.storage.relational.session.SqlSessionFactoryHelper;
 import org.apache.ibatis.annotations.Param;
@@ -32,7 +33,8 @@ public class MetalakeMetaSQLProviderFactory {
       METALAKE_META_SQL_PROVIDER_MAP =
           ImmutableMap.of(
               JDBCBackendType.MYSQL, new MetalakeMetaMySQLProvider(),
-              JDBCBackendType.H2, new MetalakeMetaH2Provider());
+              JDBCBackendType.H2, new MetalakeMetaH2Provider(),
+              JDBCBackendType.POSTGRESQL, new MetalakeMetaPostgreSQLProvider());
 
   public static MetalakeMetaBaseSQLProvider getProvider() {
     String databaseId =

core/src/main/java/org/apache/gravitino/storage/relational/mapper/OwnerMetaSQLProviderFactory.java

@@ -21,6 +21,7 @@
 import com.google.common.collect.ImmutableMap;
 import java.util.Map;
 import org.apache.gravitino.storage.relational.JDBCBackend.JDBCBackendType;
+import org.apache.gravitino.storage.relational.mapper.postgresql.OwnerMetaPostgreSQLProvider;
 import org.apache.gravitino.storage.relational.po.OwnerRelPO;
 import org.apache.gravitino.storage.relational.session.SqlSessionFactoryHelper;
 import org.apache.ibatis.annotations.Param;
@@ -31,7 +32,8 @@ public class OwnerMetaSQLProviderFactory {
       METALAKE_META_SQL_PROVIDER_MAP =
           ImmutableMap.of(
               JDBCBackendType.MYSQL, new OwnerMetaMySQLProvider(),
-              JDBCBackendType.H2, new OwnerMetaH2Provider());
+              JDBCBackendType.H2, new OwnerMetaH2Provider(),
+              JDBCBackendType.POSTGRESQL, new OwnerMetaPostgreSQLProvider());
 
   public static OwnerMetaBaseSQLProvider getProvider() {
     String databaseId =

core/src/main/java/org/apache/gravitino/storage/relational/mapper/RoleMetaSQLProviderFactory.java

@@ -21,6 +21,7 @@
 import com.google.common.collect.ImmutableMap;
 import java.util.Map;
 import org.apache.gravitino.storage.relational.JDBCBackend.JDBCBackendType;
+import org.apache.gravitino.storage.relational.mapper.postgresql.RoleMetaPostgreSQLProvider;
 import org.apache.gravitino.storage.relational.po.RolePO;
 import org.apache.gravitino.storage.relational.session.SqlSessionFactoryHelper;
 import org.apache.ibatis.annotations.Param;
@@ -30,7 +31,8 @@ public class RoleMetaSQLProviderFactory {
       METALAKE_META_SQL_PROVIDER_MAP =
           ImmutableMap.of(
               JDBCBackendType.MYSQL, new RoleMetaMySQLProvider(),
-              JDBCBackendType.H2, new RoleMetaH2Provider());
+              JDBCBackendType.H2, new RoleMetaH2Provider(),
+              JDBCBackendType.POSTGRESQL, new RoleMetaPostgreSQLProvider());
 
   public static RoleMetaBaseSQLProvider getProvider() {
     String databaseId =

core/src/main/java/org/apache/gravitino/storage/relational/mapper/SchemaMetaSQLProviderFactory.java

@@ -21,6 +21,7 @@
 import com.google.common.collect.ImmutableMap;
 import java.util.Map;
 import org.apache.gravitino.storage.relational.JDBCBackend.JDBCBackendType;
+import org.apache.gravitino.storage.relational.mapper.postgresql.SchemaMetaPostgreSQLProvider;
 import org.apache.gravitino.storage.relational.po.SchemaPO;
 import org.apache.gravitino.storage.relational.session.SqlSessionFactoryHelper;
 import org.apache.ibatis.annotations.Param;
@@ -30,7 +31,8 @@ public class SchemaMetaSQLProviderFactory {
       METALAKE_META_SQL_PROVIDER_MAP =
           ImmutableMap.of(
               JDBCBackendType.MYSQL, new SchemaMetaMySQLProvider(),
-              JDBCBackendType.H2, new SchemaMetaH2Provider());
+              JDBCBackendType.H2, new SchemaMetaH2Provider(),
+              JDBCBackendType.POSTGRESQL, new SchemaMetaPostgreSQLProvider());
 
   public static SchemaMetaBaseSQLProvider getProvider() {
     String databaseId =

core/src/main/java/org/apache/gravitino/storage/relational/mapper/SecurableObjectSQLProviderFactory.java

@@ -22,6 +22,7 @@
 import java.util.List;
 import java.util.Map;
 import org.apache.gravitino.storage.relational.JDBCBackend.JDBCBackendType;
+import org.apache.gravitino.storage.relational.mapper.postgresql.SecurableObjectPostgreSQLProvider;
 import org.apache.gravitino.storage.relational.po.SecurableObjectPO;
 import org.apache.gravitino.storage.relational.session.SqlSessionFactoryHelper;
 import org.apache.ibatis.annotations.Param;
@@ -32,7 +33,8 @@ public class SecurableObjectSQLProviderFactory {
       METALAKE_META_SQL_PROVIDER_MAP =
           ImmutableMap.of(
               JDBCBackendType.MYSQL, new SecurableObjectMySQLProvider(),
-              JDBCBackendType.H2, new SecurableObjectH2Provider());
+              JDBCBackendType.H2, new SecurableObjectH2Provider(),
+              JDBCBackendType.POSTGRESQL, new SecurableObjectPostgreSQLProvider());
 
   public static SecurableObjectBaseSQLProvider getProvider() {
     String databaseId =

core/src/main/java/org/apache/gravitino/storage/relational/mapper/TableMetaSQLProviderFactory.java

@@ -21,6 +21,7 @@
 import com.google.common.collect.ImmutableMap;
 import java.util.Map;
 import org.apache.gravitino.storage.relational.JDBCBackend.JDBCBackendType;
+import org.apache.gravitino.storage.relational.mapper.postgresql.TableMetaPostgreSQLProvider;
 import org.apache.gravitino.storage.relational.po.TablePO;
 import org.apache.gravitino.storage.relational.session.SqlSessionFactoryHelper;
 import org.apache.ibatis.annotations.Param;
@@ -31,7 +32,8 @@ public class TableMetaSQLProviderFactory {
       METALAKE_META_SQL_PROVIDER_MAP =
           ImmutableMap.of(
               JDBCBackendType.MYSQL, new TableMetaMySQLProvider(),
-              JDBCBackendType.H2, new TableMetaH2Provider());
+              JDBCBackendType.H2, new TableMetaH2Provider(),
+              JDBCBackendType.POSTGRESQL, new TableMetaPostgreSQLProvider());
 
   public static TableMetaBaseSQLProvider getProvider() {
     String databaseId =

core/src/main/java/org/apache/gravitino/storage/relational/mapper/TagMetaSQLProviderFactory.java

@@ -22,6 +22,7 @@
 import java.util.List;
 import java.util.Map;
 import org.apache.gravitino.storage.relational.JDBCBackend.JDBCBackendType;
+import org.apache.gravitino.storage.relational.mapper.postgresql.TagMetaPostgreSQLProvider;
 import org.apache.gravitino.storage.relational.po.TagPO;
 import org.apache.gravitino.storage.relational.session.SqlSessionFactoryHelper;
 import org.apache.ibatis.annotations.Param;
@@ -31,7 +32,8 @@ public class TagMetaSQLProviderFactory {
   private static final Map<JDBCBackendType, TagMetaBaseSQLProvider> METALAKE_META_SQL_PROVIDER_MAP =
       ImmutableMap.of(
           JDBCBackendType.MYSQL, new TagMetaMySQLProvider(),
-          JDBCBackendType.H2, new TagMetaH2Provider());
+          JDBCBackendType.H2, new TagMetaH2Provider(),
+          JDBCBackendType.POSTGRESQL, new TagMetaPostgreSQLProvider());
 
   public static TagMetaBaseSQLProvider getProvider() {
     String databaseId =

core/src/main/java/org/apache/gravitino/storage/relational/mapper/TagMetadataObjectRelSQLProviderFactory.java

@@ -22,6 +22,7 @@
 import java.util.List;
 import java.util.Map;
 import org.apache.gravitino.storage.relational.JDBCBackend.JDBCBackendType;
+import org.apache.gravitino.storage.relational.mapper.postgresql.TagMetadataObjectRelPostgreSQLProvider;
 import org.apache.gravitino.storage.relational.po.TagMetadataObjectRelPO;
 import org.apache.gravitino.storage.relational.session.SqlSessionFactoryHelper;
 import org.apache.ibatis.annotations.Param;
@@ -32,7 +33,8 @@ public class TagMetadataObjectRelSQLProviderFactory {
       METALAKE_META_SQL_PROVIDER_MAP =
           ImmutableMap.of(
               JDBCBackendType.MYSQL, new TagMetadataObjectRelMySQLProvider(),
-              JDBCBackendType.H2, new TagMetadataObjectRelH2Provider());
+              JDBCBackendType.H2, new TagMetadataObjectRelH2Provider(),
+              JDBCBackendType.POSTGRESQL, new TagMetadataObjectRelPostgreSQLProvider());
 
   public static TagMetadataObjectRelBaseSQLProvider getProvider() {
     String databaseId =

core/src/main/java/org/apache/gravitino/storage/relational/mapper/TopicMetaSQLProviderFactory.java

@@ -22,6 +22,7 @@
 import com.google.common.collect.ImmutableMap;
 import java.util.Map;
 import org.apache.gravitino.storage.relational.JDBCBackend.JDBCBackendType;
+import org.apache.gravitino.storage.relational.mapper.postgresql.TopicMetaPostgreSQLProvider;
 import org.apache.gravitino.storage.relational.po.TopicPO;
 import org.apache.gravitino.storage.relational.session.SqlSessionFactoryHelper;
 import org.apache.ibatis.annotations.Param;
@@ -32,7 +33,8 @@ public class TopicMetaSQLProviderFactory {
       METALAKE_META_SQL_PROVIDER_MAP =
           ImmutableMap.of(
               JDBCBackendType.MYSQL, new TopicMetaMySQLProvider(),
-              JDBCBackendType.H2, new TopicMetaH2Provider());
+              JDBCBackendType.H2, new TopicMetaH2Provider(),
+              JDBCBackendType.POSTGRESQL, new TopicMetaPostgreSQLProvider());
 
   public static TopicMetaBaseSQLProvider getProvider() {
     String databaseId =