(incubator-paimon) branch master updated: [spark] Auto modify conf for Aliyun EMR DLF in SparkGenericCatalog (#2525)

[lzljs3620320]

This is an automated email from the ASF dual-hosted git repository.

lzljs3620320 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-paimon.git


The following commit(s) were added to refs/heads/master by this push:
     new 2dda81686 [spark] Auto modify conf for Aliyun EMR DLF in 
SparkGenericCatalog (#2525)
2dda81686 is described below

commit 2dda81686d29309e88699fc3b19289c2fef974cb
Author: Zouxxyy <zouxinyu....@alibaba-inc.com>
AuthorDate: Mon Dec 18 17:28:31 2023 +0800

    [spark] Auto modify conf for Aliyun EMR DLF in SparkGenericCatalog (#2525)
---
 .../apache/paimon/spark/SparkGenericCatalog.java   | 37 ++++++++++++++++------
 1 file changed, 27 insertions(+), 10 deletions(-)

diff --git 
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkGenericCatalog.java
 
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkGenericCatalog.java
index dbbf03b36..0f9f53ee7 100644
--- 
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkGenericCatalog.java
+++ 
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkGenericCatalog.java
@@ -219,12 +219,12 @@ public class SparkGenericCatalog<T extends TableCatalog & 
SupportsNamespaces>
 
     @Override
     public final void initialize(String name, CaseInsensitiveStringMap 
options) {
+        Configuration hadoopConf = 
SparkSession.active().sessionState().newHadoopConf();
         if (options.containsKey(METASTORE.key())
                 && options.get(METASTORE.key()).equalsIgnoreCase("hive")) {
             String uri = options.get(CatalogOptions.URI.key());
             if (uri != null) {
-                Configuration conf = 
SparkSession.active().sessionState().newHadoopConf();
-                String envHmsUri = conf.get("hive.metastore.uris", null);
+                String envHmsUri = hadoopConf.get("hive.metastore.uris", null);
                 if (envHmsUri != null) {
                     Preconditions.checkArgument(
                             uri.equals(envHmsUri),
@@ -243,24 +243,41 @@ public class SparkGenericCatalog<T extends TableCatalog & 
SupportsNamespaces>
         this.sparkCatalog = new SparkCatalog();
 
         this.sparkCatalog.initialize(
-                name, autoFillConfigurations(options, 
SparkSession.active().sessionState().conf()));
+                name,
+                autoFillConfigurations(
+                        options, SparkSession.active().sessionState().conf(), 
hadoopConf));
     }
 
     private CaseInsensitiveStringMap autoFillConfigurations(
-            CaseInsensitiveStringMap options, SQLConf conf) {
+            CaseInsensitiveStringMap options, SQLConf sqlConf, Configuration 
hadoopConf) {
         Map<String, String> newOptions = new 
HashMap<>(options.asCaseSensitiveMap());
+        fillAliyunConfigurations(newOptions, hadoopConf);
+        fillCommonConfigurations(newOptions, sqlConf);
+        return new CaseInsensitiveStringMap(newOptions);
+    }
+
+    private void fillAliyunConfigurations(Map<String, String> options, 
Configuration hadoopConf) {
+        if (!options.containsKey(METASTORE.key())) {
+            // In Alibaba Cloud EMR, `hive.metastore.type` has two types: DLF 
or LOCAL, for DLF, we
+            // set `metastore` to dlf, for LOCAL, do nothing.
+            String aliyunEMRHiveMetastoreType = 
hadoopConf.get("hive.metastore.type", null);
+            if ("dlf".equalsIgnoreCase(aliyunEMRHiveMetastoreType)) {
+                options.put(METASTORE.key(), "dlf");
+            }
+        }
+    }
+
+    private void fillCommonConfigurations(Map<String, String> options, SQLConf 
sqlConf) {
         if (!options.containsKey(WAREHOUSE.key())) {
-            String warehouse = conf.warehousePath();
-            newOptions.put(WAREHOUSE.key(), warehouse);
+            String warehouse = sqlConf.warehousePath();
+            options.put(WAREHOUSE.key(), warehouse);
         }
         if (!options.containsKey(METASTORE.key())) {
-            String metastore = 
conf.getConf(StaticSQLConf.CATALOG_IMPLEMENTATION());
+            String metastore = 
sqlConf.getConf(StaticSQLConf.CATALOG_IMPLEMENTATION());
             if (HiveCatalogOptions.IDENTIFIER.equals(metastore)) {
-                newOptions.put(METASTORE.key(), metastore);
+                options.put(METASTORE.key(), metastore);
             }
         }
-
-        return new CaseInsensitiveStringMap(newOptions);
     }
 
     @Override