[Feature] Support custom catalog name for Iceberg JDBC catalog (#75966)

Signed-off-by: chrisyguo <511955993@qq.com>

Author: Guosmilesmile

docs/en/data_source/catalog/iceberg/iceberg_catalog.md

@@ -433,6 +433,11 @@ The following table describes the parameter you need to configure in `MetastoreP
   - Required: No
   - Description: Whether to create the tables `iceberg_namespace_properties` and `iceberg_tables` for storing metadata in the database specified by `iceberg.catalog.uri`. The default value is `false`. Specify `true` if these two tables have not yet been created in the database specified by `iceberg.catalog.uri`.
 
+- `iceberg.catalog.jdbc.catalog-name`
+  - Required: No
+  - Description: Explicitly specifies the value of the `catalog_name` column in the Iceberg JDBC metadata tables (`iceberg_tables` and `iceberg_namespace_properties`). When not set, the name of the StarRocks-side catalog (the `<catalog_name>` provided after `CREATE EXTERNAL CATALOG`) is used, which preserves backward compatibility.
+  - When you need to attach StarRocks to an Iceberg JDBC catalog that has already been created by other engines (such as Spark or Flink) and share the same metadata, you must set this parameter to the catalog name used by the other engine. Otherwise the underlying `WHERE catalog_name = ?` query cannot match the existing metadata, and StarRocks will not be able to see any databases or tables.
+
 The following example creates an Iceberg catalog named `iceberg_jdbc` and uses JDBC as metastore:
 
 ```SQL
@@ -452,6 +457,25 @@ PROPERTIES
 ```
 If using MySQL or other custom JDBC drivers, the corresponding JAR files need to be placed in the `fe/lib` and `be/lib/jni-packages` directories.
 
+The following example shows how to attach StarRocks to an Iceberg JDBC catalog that has already been created by another engine (for example Spark) whose underlying `catalog_name` is `"spark_ice"`. In this case, the StarRocks-side catalog name `sr_side_view` may differ from the underlying `catalog_name`, as long as you explicitly specify `iceberg.catalog.jdbc.catalog-name`:
+
+```SQL
+CREATE EXTERNAL CATALOG sr_side_view
+PROPERTIES
+(
+    "type" = "iceberg",
+    "iceberg.catalog.type" = "jdbc",
+    "iceberg.catalog.warehouse" = "s3://my_bucket/warehouse_location",
+    "iceberg.catalog.uri" = "jdbc:mysql://ip:port/db_name",
+    "iceberg.catalog.jdbc.user" = "username",
+    "iceberg.catalog.jdbc.password" = "password",
+    "iceberg.catalog.jdbc.catalog-name" = "spark_ice",
+    "aws.s3.endpoint" = "<s3_endpoint>",
+    "aws.s3.access_key" = "<iam_user_access_key>",
+    "aws.s3.secret_key" = "<iam_user_secret_key>"
+);
+```
+
 </TabItem>
 
 </Tabs>

docs/ja/data_source/catalog/iceberg/iceberg_catalog.md

@@ -432,6 +432,11 @@ SHOW DATABASES FROM r2;
   - 必須：いいえ
   - 説明：`iceberg.catalog.uri` で指定されたデータベースにメタデータを格納するためのテーブル `iceberg_namespace_properties` および `iceberg_tables` を作成するかどうか。デフォルト値は `false` です。`iceberg.catalog.uri` で指定されたデータベースにこれらの 2 つのテーブルがまだ作成されていない場合は `true` を指定してください。
 
+- `iceberg.catalog.jdbc.catalog-name`
+  - 必須：いいえ
+  - 説明：Iceberg JDBC のメタデータテーブル（`iceberg_tables` および `iceberg_namespace_properties`）の `catalog_name` 列の値を明示的に指定します。未設定の場合は、StarRocks 側のカタログ名（`CREATE EXTERNAL CATALOG` の後に指定した `<catalog_name>`）が使用され、後方互換性が維持されます。
+  - StarRocks を、他のエンジン（例：Spark、Flink）で既に作成された Iceberg JDBC Catalog に接続し、同じメタデータを共有したい場合には、このパラメーターに相手側で使用しているカタログ名を設定する必要があります。設定しない場合、内部で発行される `WHERE catalog_name = ?` クエリが既存のメタデータにマッチせず、StarRocks 側から一切のデータベースやテーブルが見えなくなります。
+
 次の例は、Iceberg catalog `iceberg_jdbc` を作成し、メタストアとして JDBC を使用します。
 
 ```SQL
@@ -452,6 +457,25 @@ PROPERTIES
 
 MySQL やその他のカスタム JDBC ドライバを使用する場合、対応する JAR ファイルを `fe/lib` ディレクトリおよび `be/lib/jni-packages` ディレクトリに配置する必要があります。
 
+次の例は、他のエンジン（例：Spark）で既に作成され、内部の `catalog_name` が `"spark_ice"` になっている Iceberg JDBC Catalog に StarRocks を接続する方法を示しています。この場合、StarRocks 側のカタログ名 `sr_side_view` は内部の `catalog_name` と異なっていても構いません。`iceberg.catalog.jdbc.catalog-name` を明示的に指定するだけで接続できます。
+
+```SQL
+CREATE EXTERNAL CATALOG sr_side_view
+PROPERTIES
+(
+    "type" = "iceberg",
+    "iceberg.catalog.type" = "jdbc",
+    "iceberg.catalog.warehouse" = "s3://my_bucket/warehouse_location",
+    "iceberg.catalog.uri" = "jdbc:mysql://ip:port/db_name",
+    "iceberg.catalog.jdbc.user" = "username",
+    "iceberg.catalog.jdbc.password" = "password",
+    "iceberg.catalog.jdbc.catalog-name" = "spark_ice",
+    "aws.s3.endpoint" = "<s3_endpoint>",
+    "aws.s3.access_key" = "<iam_user_access_key>",
+    "aws.s3.secret_key" = "<iam_user_secret_key>"
+);
+```
+
 </TabItem>
 
 </Tabs>

docs/zh/data_source/catalog/iceberg/iceberg_catalog.md

@@ -472,6 +472,14 @@ SHOW DATABASES FROM r2;
 
 描述：是否在 `iceberg.catalog.uri` 指定的数据库中创建用于存储元数据的表 `iceberg_namespace_properties` 和 `iceberg_tables`，默认值为`false`。当`iceberg.catalog.uri` 指定的数据库中尚未创建上述两张表时需要指定为`true`。
 
+##### iceberg.catalog.jdbc.catalog-name
+
+必需：否
+
+描述：显式指定 Iceberg JDBC 底层元数据表（`iceberg_tables` 和 `iceberg_namespace_properties`）中 `catalog_name` 字段的值。未设置时，默认使用 StarRocks 侧的 Catalog 名（即 `CREATE EXTERNAL CATALOG` 后的 `<catalog_name>`），保持向后兼容。
+
+当您需要在 StarRocks 中接入由其他引擎（例如 Spark、Flink）已经建立的 Iceberg JDBC Catalog、并复用同一份元数据时，需要将该参数设置为对方使用的 catalog name，否则底层查询 SQL `WHERE catalog_name = ?` 将无法命中已有元数据，导致 StarRocks 侧看不到任何库表。
+
 以下示例创建了一个名为 `iceberg_jdbc` 的 Iceberg catalog，并使用 JDBC 作为元存储：
 
 ```SQL
@@ -489,7 +497,27 @@ PROPERTIES
     "aws.s3.secret_key" = "<iam_user_secret_key>"
 );
 ```
-若使用MySQL或其他自定义的JDBC驱动程序，需要将相应的JAR包放置于 `fe/lib` 和 `be/lib/jni-packages` 目录下。
+
+若使用 MySQL 或其他自定义的 JDBC 驱动程序，需要将相应的 JAR 包放置于 `fe/lib` 和 `be/lib/jni-packages` 目录下。
+
+以下示例演示了如何接入由其他引擎（例如 Spark）已经建立、底层 `catalog_name = "spark_ice"` 的 Iceberg JDBC Catalog。此时 StarRocks 侧 Catalog 名 `sr_side_view` 可以与底层 `catalog_name` 不同，只需通过 `iceberg.catalog.jdbc.catalog-name` 显式指定即可：
+
+```SQL
+CREATE EXTERNAL CATALOG sr_side_view
+PROPERTIES
+(
+    "type" = "iceberg",
+    "iceberg.catalog.type" = "jdbc",
+    "iceberg.catalog.warehouse" = "s3://my_bucket/warehouse_location",
+    "iceberg.catalog.uri" = "jdbc:mysql://ip:port/db_name",
+    "iceberg.catalog.jdbc.user" = "username",
+    "iceberg.catalog.jdbc.password" = "password",
+    "iceberg.catalog.jdbc.catalog-name" = "spark_ice",
+    "aws.s3.endpoint" = "<s3_endpoint>",
+    "aws.s3.access_key" = "<iam_user_access_key>",
+    "aws.s3.secret_key" = "<iam_user_secret_key>"
+);
+```
 
 </TabItem>
 

fe/fe-core/src/main/java/com/starrocks/connector/iceberg/IcebergCatalogProperties.java

@@ -33,6 +33,7 @@ public class IcebergCatalogProperties {
     public static final String HIVE_METASTORE_TIMEOUT = "hive.metastore.timeout";
     public static final String ICEBERG_CUSTOM_PROPERTIES_PREFIX = "iceberg.catalog.";
     public static final String ICEBERG_JDBC_PASSWORD = "jdbc.password";
+    public static final String ICEBERG_JDBC_CATALOG_NAME = "jdbc.catalog-name";
     public static final String ENABLE_ICEBERG_METADATA_CACHE = "enable_iceberg_metadata_cache";
     public static final String ENABLE_ICEBERG_TABLE_CACHE = "enable_iceberg_table_cache";
     public static final String ICEBERG_META_CACHE_TTL = "iceberg_meta_cache_ttl_sec"; // implicit for user

fe/fe-core/src/main/java/com/starrocks/connector/iceberg/jdbc/IcebergJdbcCatalog.java

@@ -52,6 +52,7 @@
 
 import static com.starrocks.connector.ConnectorTableId.CONNECTOR_ID_GENERATOR;
 import static com.starrocks.connector.iceberg.IcebergCatalogProperties.ICEBERG_CUSTOM_PROPERTIES_PREFIX;
+import static com.starrocks.connector.iceberg.IcebergCatalogProperties.ICEBERG_JDBC_CATALOG_NAME;
 import static com.starrocks.connector.iceberg.IcebergMetadata.LOCATION_PROPERTY;
 import static org.apache.iceberg.CatalogProperties.WAREHOUSE_LOCATION;
 
@@ -93,8 +94,11 @@ public IcebergJdbcCatalog(String name, Configuration conf, Map<String, String> p
                     ICEBERG_CUSTOM_PROPERTIES_PREFIX + WAREHOUSE_LOCATION));
         }
 
-        this.delegate = (JdbcCatalog) CatalogUtil.loadCatalog(JdbcCatalog.class.getName(), name, copiedProperties, conf);
+        String jdbcCatalogName = copiedProperties.getOrDefault(ICEBERG_JDBC_CATALOG_NAME, name);
+        copiedProperties.remove(ICEBERG_JDBC_CATALOG_NAME);
 
+        this.delegate = (JdbcCatalog) CatalogUtil.loadCatalog(
+                JdbcCatalog.class.getName(), jdbcCatalogName, copiedProperties, conf);
     }
 
 

fe/fe-core/src/test/java/com/starrocks/connector/iceberg/IcebergJdbcCatalogTest.java

@@ -43,6 +43,7 @@
 import java.util.Arrays;
 import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
 
 import static com.starrocks.connector.iceberg.IcebergCatalogProperties.ICEBERG_CUSTOM_PROPERTIES_PREFIX;
 import static com.starrocks.connector.iceberg.IcebergMetadata.LOCATION_PROPERTY;
@@ -284,4 +285,62 @@ public void testDeleteUncommittedDataFiles(@Mocked JdbcCatalog jdbcCatalog, @Moc
                 Arrays.asList("/tmp/iceberg/db/table/part-00000", "/tmp/iceberg/db/table/part-00001"));
     }
 
+    @Test
+    public void testJdbcCatalogNameDefaultToStarRocksCatalogName(@Mocked JdbcCatalog jdbcCatalog) {
+        final String[] capturedName = new String[1];
+        final Map<String, String>[] capturedProps = new Map[1];
+
+        new Expectations() {
+            {
+                jdbcCatalog.initialize(anyString, (Map<String, String>) any);
+                result = new mockit.Delegate<Void>() {
+                    @SuppressWarnings("unused")
+                    void initialize(String name, Map<String, String> props) {
+                        capturedName[0] = name;
+                        capturedProps[0] = props;
+                    }
+                };
+                minTimes = 0;
+            }
+        };
+
+        new IcebergJdbcCatalog("sr_catalog", new Configuration(),
+                ImmutableMap.of("iceberg.catalog.warehouse", LOCATION,
+                        "iceberg.catalog.uri", URI));
+
+        assertEquals("sr_catalog", capturedName[0]);
+        Assertions.assertNotNull(capturedProps[0]);
+        Assertions.assertFalse(capturedProps[0].containsKey("jdbc.catalog-name"));
+    }
+
+    @Test
+    public void testJdbcCatalogNameOverride(@Mocked JdbcCatalog jdbcCatalog) {
+        final String[] capturedName = new String[1];
+        final Map<String, String>[] capturedProps = new Map[1];
+
+        new Expectations() {
+            {
+                jdbcCatalog.initialize(anyString, (Map<String, String>) any);
+                result = new mockit.Delegate<Void>() {
+                    @SuppressWarnings("unused")
+                    void initialize(String name, java.util.Map<String, String> props) {
+                        capturedName[0] = name;
+                        capturedProps[0] = props;
+                    }
+                };
+                minTimes = 0;
+            }
+        };
+
+        new IcebergJdbcCatalog("sr_side_view", new Configuration(),
+                ImmutableMap.of(
+                        "iceberg.catalog.warehouse", LOCATION,
+                        "iceberg.catalog.uri", URI,
+                        "iceberg.catalog.jdbc.catalog-name", "spark_ice"));
+
+        assertEquals("spark_ice", capturedName[0]);
+        Assertions.assertNotNull(capturedProps[0]);
+        Assertions.assertFalse(capturedProps[0].containsKey("jdbc.catalog-name"));
+    }
+
 }