1

docs/en/docs/lakehouse/multi-catalog/iceberg.md

@@ -61,7 +61,7 @@ Use the Iceberg API to access metadata, and support services such as Hadoop File
 CREATE CATALOG iceberg_hadoop PROPERTIES (
     'type'='iceberg',
     'iceberg.catalog.type' = 'hadoop',
-    'warehouse' = 'hdfs://your-host/dir/key'
+    'warehouse' = 'hdfs://your-host:8020/dir/key'
 );
 ```
 

docs/zh-CN/docs/lakehouse/multi-catalog/iceberg.md

@@ -61,7 +61,7 @@ CREATE CATALOG iceberg PROPERTIES (
 CREATE CATALOG iceberg_hadoop PROPERTIES (
     'type'='iceberg',
     'iceberg.catalog.type' = 'hadoop',
-    'warehouse' = 'hdfs://your-host/dir/key'
+    'warehouse' = 'hdfs://your-host:8020/dir/key'
 );
 ```
 

fe/fe-core/src/main/java/org/apache/doris/catalog/HdfsResource.java

@@ -54,6 +54,7 @@ public class HdfsResource extends Resource {
     public static String HADOOP_SOCKET_PATH = "dfs.domain.socket.path";
     public static String DSF_NAMESERVICES = "dfs.nameservices";
     public static final String HDFS_PREFIX = "hdfs:";
+    public static final String HDFS_FILE_PREFIX = "hdfs://";
 
     @SerializedName(value = "properties")
     private Map<String, String> properties;

fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergHadoopExternalCatalog.java

@@ -17,6 +17,7 @@
 
 package org.apache.doris.datasource.iceberg;
 
+import org.apache.doris.catalog.HdfsResource;
 import org.apache.doris.datasource.CatalogProperty;
 import org.apache.doris.datasource.property.PropertyConverter;
 
@@ -34,7 +35,16 @@ public IcebergHadoopExternalCatalog(long catalogId, String name, String resource
                                         String comment) {
         super(catalogId, name, comment);
         props = PropertyConverter.convertToMetaProperties(props);
+        String warehouse = props.get(CatalogProperties.WAREHOUSE_LOCATION);
+        Preconditions.checkArgument(StringUtils.isNotEmpty(warehouse),
+                "Cannot initialize Iceberg HadoopCatalog because 'warehouse' must not be null or empty");
+        String nameService = StringUtils.substringBetween(warehouse, HdfsResource.HDFS_FILE_PREFIX, "/");
+        if (StringUtils.isEmpty(nameService)) {
+            throw new IllegalArgumentException("Unrecognized 'warehouse' location format"
+                    + " because name service is required.");
+        }
         catalogProperty = new CatalogProperty(resource, props);
+        catalogProperty.addProperty(HdfsResource.HADOOP_FS_NAME, HdfsResource.HDFS_FILE_PREFIX + nameService);
     }
 
     @Override
@@ -45,8 +55,6 @@ protected void initLocalObjectsImpl() {
         // initialize hive catalog
         Map<String, String> catalogProperties = new HashMap<>();
         String warehouse = catalogProperty.getProperties().get(CatalogProperties.WAREHOUSE_LOCATION);
-        Preconditions.checkArgument(StringUtils.isNotEmpty(warehouse),
-                    "Cannot initialize Iceberg HadoopCatalog because 'warehouse' must not be null or empty");
         catalogProperties.put(CatalogProperties.WAREHOUSE_LOCATION, warehouse);
         hadoopCatalog.initialize(icebergCatalogType, catalogProperties);
         catalog = hadoopCatalog;

fe/fe-core/src/main/java/org/apache/doris/persist/gson/GsonUtils.java

@@ -68,6 +68,7 @@
 import org.apache.doris.datasource.iceberg.IcebergExternalCatalog;
 import org.apache.doris.datasource.iceberg.IcebergGlueExternalCatalog;
 import org.apache.doris.datasource.iceberg.IcebergHMSExternalCatalog;
+import org.apache.doris.datasource.iceberg.IcebergHadoopExternalCatalog;
 import org.apache.doris.datasource.iceberg.IcebergRestExternalCatalog;
 import org.apache.doris.datasource.jdbc.JdbcExternalCatalog;
 import org.apache.doris.datasource.paimon.PaimonExternalCatalog;
@@ -205,6 +206,7 @@ public class GsonUtils {
             .registerSubtype(IcebergGlueExternalCatalog.class, IcebergGlueExternalCatalog.class.getSimpleName())
             .registerSubtype(IcebergRestExternalCatalog.class, IcebergRestExternalCatalog.class.getSimpleName())
             .registerSubtype(IcebergDLFExternalCatalog.class, IcebergDLFExternalCatalog.class.getSimpleName())
+            .registerSubtype(IcebergHadoopExternalCatalog.class, IcebergHadoopExternalCatalog.class.getSimpleName())
             .registerSubtype(PaimonExternalCatalog.class, PaimonExternalCatalog.class.getSimpleName())
             .registerSubtype(PaimonHMSExternalCatalog.class, PaimonHMSExternalCatalog.class.getSimpleName())
             .registerSubtype(MaxComputeExternalCatalog.class, MaxComputeExternalCatalog.class.getSimpleName());

fe/fe-core/src/main/java/org/apache/doris/planner/external/iceberg/IcebergScanNode.java

@@ -21,6 +21,7 @@
 import org.apache.doris.analysis.TableSnapshot;
 import org.apache.doris.analysis.TupleDescriptor;
 import org.apache.doris.catalog.Env;
+import org.apache.doris.catalog.HdfsResource;
 import org.apache.doris.catalog.TableIf;
 import org.apache.doris.catalog.external.ExternalTable;
 import org.apache.doris.catalog.external.HMSExternalTable;
@@ -186,15 +187,15 @@ public List<Split> getSplits() throws UserException {
         // Min split size is DEFAULT_SPLIT_SIZE(128MB).
         long splitSize = Math.max(ConnectContext.get().getSessionVariable().getFileSplitSize(), DEFAULT_SPLIT_SIZE);
         HashSet<String> partitionPathSet = new HashSet<>();
-        String dataPath = icebergTable.location() + icebergTable.properties()
+        String dataPath = normalizeLocation(icebergTable.location()) + icebergTable.properties()
                 .getOrDefault(TableProperties.WRITE_DATA_LOCATION, DEFAULT_DATA_PATH);
         boolean isPartitionedTable = icebergTable.spec().isPartitioned();
 
         CloseableIterable<FileScanTask> fileScanTasks = TableScanUtil.splitFiles(scan.planFiles(), splitSize);
         try (CloseableIterable<CombinedScanTask> combinedScanTasks =
                 TableScanUtil.planTasks(fileScanTasks, splitSize, 1, 0)) {
             combinedScanTasks.forEach(taskGrp -> taskGrp.files().forEach(splitTask -> {
-                String dataFilePath = splitTask.file().path().toString();
+                String dataFilePath = normalizeLocation(splitTask.file().path().toString());
 
                 // Counts the number of partitions read
                 if (isPartitionedTable) {
@@ -297,8 +298,21 @@ public TFileType getLocationType() throws UserException {
 
     @Override
     public TFileType getLocationType(String location) throws UserException {
-        return getTFileType(location).orElseThrow(() ->
-                new DdlException("Unknown file location " + location + " for iceberg table " + icebergTable.name()));
+        final String fLocation = normalizeLocation(location);
+        return getTFileType(fLocation).orElseThrow(() ->
+                new DdlException("Unknown file location " + fLocation + " for iceberg table " + icebergTable.name()));
+    }
+
+    private String normalizeLocation(String location) {
+        Map<String, String> props = source.getCatalog().getProperties();
+        String icebergCatalogType = props.get(IcebergExternalCatalog.ICEBERG_CATALOG_TYPE);
+        if (icebergCatalogType.equalsIgnoreCase("hadoop")) {
+            if (!location.startsWith(HdfsResource.HDFS_PREFIX)) {
+                String fsName = props.get(HdfsResource.HADOOP_FS_NAME);
+                location = fsName + location;
+            }
+        }
+        return location;
     }
 
     @Override

regression-test/data/external_table_p2/iceberg/test_external_catalog_iceberg_hadoop_catalog.out

@@ -0,0 +1,17 @@
+-- This file is automatically generated. You should know what you did if you want to edit this
+-- !q01 --
+2879562
+
+-- !q02 --
+1
+3
+5
+6
+7
+8
+11
+
+-- !q03 --
+1	Customer#000000001	j5JsirBM9P	MOROCCO  0	MOROCCO	AFRICA	25-989-741-2988	BUILDING
+3	Customer#000000003	fkRGN8n	ARGENTINA7	ARGENTINA	AMERICA	11-719-748-3364	AUTOMOBILE
+5	Customer#000000005	hwBtxkoBF qSW4KrI	CANADA   5	CANADA	AMERICA	13-750-942-6364	HOUSEHOLD

regression-test/suites/external_table_p2/iceberg/test_external_catalog_iceberg_hadoop_catalog.groovy

@@ -0,0 +1,43 @@
+// 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.
+
+suite("test_external_catalog_iceberg_hadoop_catalog", "p2,external,iceberg,external_remote,external_remote_iceberg") {
+    String enabled = context.config.otherConfigs.get("enableExternalHiveTest")
+    if (enabled != null && enabled.equalsIgnoreCase("true")) {
+        String iceberg_catalog_name = "test_external_iceberg_catalog_hadoop"
+        String extHiveHmsHost = context.config.otherConfigs.get("extHiveHmsHost")
+        String extHdfsPort = context.config.otherConfigs.get("extHdfsPort")
+        sql """drop catalog if exists ${iceberg_catalog_name};"""
+        sql """
+            create catalog if not exists ${iceberg_catalog_name} properties (
+                'type'='iceberg',
+                'iceberg.catalog.type'='hadoop',
+                'warehouse' = 'hdfs://${extHiveHmsHost}:${extHdfsPort}/usr/hive/warehouse/hadoop_catalog'
+            );
+        """
+
+        sql """switch ${iceberg_catalog_name};"""
+        def q01 = {
+            qt_q01 """ select count(*) from iceberg_hadoop_catalog """
+            qt_q02 """ select c_custkey from iceberg_hadoop_catalog group by c_custkey order by c_custkey limit 7 """
+            qt_q03 """ select * from iceberg_hadoop_catalog order by c_custkey limit 3 """
+        }
+
+        sql """ use `multi_catalog`; """
+        q01()
+    }
+}