branch-4.0: [bug](iceberg) use split file format for iceberg scan (#65760)

docker/thirdparties/docker-compose/iceberg/scripts/create_preinstalled_scripts/iceberg/run31.sql

@@ -0,0 +1,33 @@
+-- Bootstrap an Iceberg table whose active snapshot contains both Parquet and ORC data files.
+-- This script is sourced on every Iceberg container start, so keep it repeatable.
+
+CREATE DATABASE IF NOT EXISTS demo.test_db;
+USE demo.test_db;
+
+DROP TABLE IF EXISTS mixed_file_format;
+
+CREATE TABLE mixed_file_format (
+    id INT,
+    source STRING
+)
+USING iceberg
+TBLPROPERTIES (
+    'format-version' = '2',
+    'write.format.default' = 'parquet'
+);
+
+-- The first snapshot's data files are Parquet.
+INSERT INTO mixed_file_format VALUES
+    (1, 'parquet'),
+    (2, 'parquet'),
+    (3, 'parquet');
+
+-- Change only the format for subsequent writes. The Parquet files above remain
+-- referenced by the current snapshot, while this append produces ORC files.
+ALTER TABLE mixed_file_format
+    SET TBLPROPERTIES ('write.format.default' = 'orc');
+
+INSERT INTO mixed_file_format VALUES
+    (4, 'orc'),
+    (5, 'orc'),
+    (6, 'orc');

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

@@ -417,8 +417,9 @@ private TScanRangeLocations splitToScanRange(
         TFileRangeDesc rangeDesc = createFileRangeDesc(fileSplit, partitionValuesFromPath, pathPartitionKeys);
         TFileCompressType fileCompressType = getFileCompressType(fileSplit);
         rangeDesc.setCompressType(fileCompressType);
-        // set file format type, and the type might fall back to native format in setScanParams
-        rangeDesc.setFormatType(getFileFormatType());
+        // Seed connector-specific setup with the scan-level default. A connector may then
+        // override it with the actual format carried by an individual split.
+        rangeDesc.setFormatType(params.getFormatType());
         setScanParams(rangeDesc, fileSplit);
 
         curLocations.getScanRange().getExtScanRange().getFileScanRange().addToRanges(rangeDesc);

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

@@ -1059,7 +1059,7 @@ public static long getIcebergRowCount(ExternalTable tbl) {
 
     public static FileFormat getFileFormat(Table icebergTable) {
         Map<String, String> properties = icebergTable.properties();
-        String fileFormatName = resolveFileFormatName(icebergTable, properties);
+        String fileFormatName = resolveFileFormatName(properties);
         FileFormat fileFormat;
         if (fileFormatName.toLowerCase().contains(ORC_NAME)) {
             fileFormat = FileFormat.ORC;
@@ -1071,7 +1071,7 @@ public static FileFormat getFileFormat(Table icebergTable) {
         return fileFormat;
     }
 
-    private static String resolveFileFormatName(Table icebergTable, Map<String, String> properties) {
+    private static String resolveFileFormatName(Map<String, String> properties) {
         // 1. Check "write-format" (nickname in Flink and Spark)
         if (properties.containsKey(WRITE_FORMAT)) {
             return properties.get(WRITE_FORMAT);
@@ -1080,27 +1080,7 @@ private static String resolveFileFormatName(Table icebergTable, Map<String, Stri
         if (properties.containsKey(TableProperties.DEFAULT_FILE_FORMAT)) {
             return properties.get(TableProperties.DEFAULT_FILE_FORMAT);
         }
-        // 3. Last resort: infer from the actual data files in the current snapshot.
-        //    This handles migrated tables where none of the above properties are set.
-        return inferFileFormatFromDataFiles(icebergTable);
-    }
-
-    private static String inferFileFormatFromDataFiles(Table icebergTable) {
-        if (icebergTable.currentSnapshot() == null) {
-            LOG.info("Iceberg table {} has no snapshot, defaulting to {}", icebergTable.name(), PARQUET_NAME);
-            return PARQUET_NAME;
-        }
-        try (CloseableIterable<FileScanTask> files = icebergTable.newScan().planFiles()) {
-            java.util.Iterator<FileScanTask> it = files.iterator();
-            if (it.hasNext()) {
-                String format = it.next().file().format().name().toLowerCase();
-                LOG.info("Iceberg table {} inferred file format {} from data files", icebergTable.name(), format);
-                return format;
-            }
-        } catch (Exception e) {
-            LOG.warn("Failed to infer file format from data files for table {}, defaulting to {}",
-                    icebergTable.name(), PARQUET_NAME, e);
-        }
+        // Iceberg defaults the write format to Parquet when the table does not declare one.
         return PARQUET_NAME;
     }
 

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

@@ -25,7 +25,6 @@
 import org.apache.doris.catalog.Column;
 import org.apache.doris.catalog.Env;
 import org.apache.doris.catalog.TableIf;
-import org.apache.doris.common.DdlException;
 import org.apache.doris.common.UserException;
 import org.apache.doris.common.profile.SummaryProfile;
 import org.apache.doris.common.security.authentication.ExecutionAuthenticator;
@@ -71,6 +70,7 @@
 import org.apache.iceberg.DeleteFile;
 import org.apache.iceberg.DeleteFileIndex;
 import org.apache.iceberg.FileContent;
+import org.apache.iceberg.FileFormat;
 import org.apache.iceberg.FileScanTask;
 import org.apache.iceberg.ManifestContent;
 import org.apache.iceberg.ManifestFile;
@@ -258,6 +258,8 @@ protected void setScanParams(TFileRangeDesc rangeDesc, Split split) {
     private void setIcebergParams(TFileRangeDesc rangeDesc, IcebergSplit icebergSplit) {
         TTableFormatFileDesc tableFormatFileDesc = new TTableFormatFileDesc();
         tableFormatFileDesc.setTableFormatType(icebergSplit.getTableFormatType().value());
+        // update for every split file format
+        rangeDesc.setFormatType(toTFileFormatType(icebergSplit.getSplitFileFormat()));
         if (tableLevelPushDownCount) {
             tableFormatFileDesc.setTableLevelRowCount(icebergSplit.getTableLevelRowCount());
         } else {
@@ -340,6 +342,15 @@ protected List<String> getDeleteFiles(TFileRangeDesc rangeDesc) {
         return deleteFiles;
     }
 
+    private TFileFormatType toTFileFormatType(FileFormat fileFormat) {
+        if (fileFormat == FileFormat.PARQUET) {
+            return TFileFormatType.FORMAT_PARQUET;
+        } else if (fileFormat == FileFormat.ORC) {
+            return TFileFormatType.FORMAT_ORC;
+        }
+        throw new UnsupportedOperationException("Unsupported Iceberg data file format: " + fileFormat);
+    }
+
     private String getDeleteFileContentType(int content) {
         // Iceberg file type: 0: data, 1: position delete, 2: equality delete, 3: deletion vector
         switch (content) {
@@ -752,6 +763,7 @@ private Split createIcebergSplit(FileScanTask fileScanTask) {
                 storagePropertiesMap,
                 new ArrayList<>(),
                 originalPath);
+        split.setSplitFileFormat(fileScanTask.file().format());
         if (!fileScanTask.deletes().isEmpty()) {
             split.setDeleteFileFilters(getDeleteFileFilters(fileScanTask));
         }
@@ -929,16 +941,8 @@ private List<IcebergDeleteFileFilter> getDeleteFileFilters(FileScanTask spitTask
 
     @Override
     public TFileFormatType getFileFormatType() throws UserException {
-        TFileFormatType type;
-        String icebergFormat = source.getFileFormat();
-        if (icebergFormat.equalsIgnoreCase("parquet")) {
-            type = TFileFormatType.FORMAT_PARQUET;
-        } else if (icebergFormat.equalsIgnoreCase("orc")) {
-            type = TFileFormatType.FORMAT_ORC;
-        } else {
-            throw new DdlException(String.format("Unsupported format name: %s for iceberg table.", icebergFormat));
-        }
-        return type;
+        // for table level file format
+        return toTFileFormatType(IcebergUtils.getFileFormat(icebergTable));
     }
 
     @Override

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

@@ -22,6 +22,7 @@
 import org.apache.doris.datasource.property.storage.StorageProperties;
 
 import lombok.Data;
+import org.apache.iceberg.FileFormat;
 
 import java.util.ArrayList;
 import java.util.List;
@@ -43,6 +44,8 @@ public class IcebergSplit extends FileSplit {
     private long tableLevelRowCount = -1;
     // Partition values are used to do runtime filter partition pruning.
     private Map<String, String> icebergPartitionValues = null;
+    // maybe mixed file format type in one table. so need record it for every split
+    private FileFormat splitFileFormat;
 
     // File path will be changed if the file is modified, so there's no need to get modification time.
     public IcebergSplit(LocationPath file, long start, long length, long fileLength, String[] hosts,

fe/fe-core/src/test/java/org/apache/doris/datasource/iceberg/IcebergUtilsTest.java

@@ -35,6 +35,7 @@
 import org.apache.iceberg.Snapshot;
 import org.apache.iceberg.SnapshotRef;
 import org.apache.iceberg.Table;
+import org.apache.iceberg.TableProperties;
 import org.apache.iceberg.expressions.Expression;
 import org.apache.iceberg.expressions.Expressions;
 import org.apache.iceberg.expressions.UnboundPredicate;
@@ -63,6 +64,28 @@
 import java.util.Optional;
 
 public class IcebergUtilsTest {
+    @Test
+    public void testGetFileFormatUsesPropertiesWithoutPlanningDataFiles() {
+        Table table = Mockito.mock(Table.class);
+        Mockito.when(table.properties()).thenReturn(Collections.emptyMap());
+        Mockito.when(table.currentSnapshot()).thenReturn(Mockito.mock(Snapshot.class));
+
+        Assert.assertEquals(org.apache.iceberg.FileFormat.PARQUET, IcebergUtils.getFileFormat(table));
+        // Do not call newScan planFiles()
+        Mockito.verify(table, Mockito.never()).newScan();
+    }
+
+    @Test
+    public void testGetFileFormatUsesConfiguredTableFormat() {
+        Table table = Mockito.mock(Table.class);
+        Mockito.when(table.properties()).thenReturn(
+                ImmutableMap.of(TableProperties.DEFAULT_FILE_FORMAT, "orc"));
+
+        Assert.assertEquals(org.apache.iceberg.FileFormat.ORC, IcebergUtils.getFileFormat(table));
+        // Do not call newScan planFiles()
+        Mockito.verify(table, Mockito.never()).newScan();
+    }
+
     @Test
     public void testParseTableName() {
         try {

fe/fe-core/src/test/java/org/apache/doris/datasource/iceberg/source/IcebergScanNodeTest.java

@@ -19,18 +19,24 @@
 
 import org.apache.doris.analysis.TupleDescriptor;
 import org.apache.doris.analysis.TupleId;
+import org.apache.doris.common.util.LocationPath;
+import org.apache.doris.datasource.TableFormatType;
 import org.apache.doris.planner.PlanNodeId;
 import org.apache.doris.planner.ScanContext;
 import org.apache.doris.qe.SessionVariable;
+import org.apache.doris.thrift.TFileFormatType;
+import org.apache.doris.thrift.TFileRangeDesc;
 
 import org.apache.iceberg.DataFile;
+import org.apache.iceberg.FileFormat;
 import org.apache.iceberg.FileScanTask;
 import org.apache.iceberg.util.ScanTaskUtil;
 import org.junit.Assert;
 import org.junit.Test;
 import org.mockito.Mockito;
 
 import java.lang.reflect.Method;
+import java.util.ArrayList;
 import java.util.Collections;
 
 public class IcebergScanNodeTest {
@@ -70,4 +76,23 @@ public void testDetermineTargetFileSplitSizeHonorsMaxFileSplitNum() throws Excep
             Assert.assertEquals(100 * MB, target);
         }
     }
+
+    @Test
+    public void testSetIcebergParamsUsesSplitFileFormat() throws Exception {
+        TestIcebergScanNode node = new TestIcebergScanNode(new SessionVariable());
+        String dataPath = "file:///tmp/data-file.orc";
+        IcebergSplit split = new IcebergSplit(LocationPath.of(dataPath), 0, 128, 128, new String[0],
+                2, Collections.emptyMap(), new ArrayList<>(), dataPath);
+        split.setTableFormatType(TableFormatType.ICEBERG);
+        split.setSplitFileFormat(FileFormat.ORC);
+
+        Method method = IcebergScanNode.class.getDeclaredMethod("setIcebergParams",
+                TFileRangeDesc.class, IcebergSplit.class);
+        method.setAccessible(true);
+
+        TFileRangeDesc rangeDesc = new TFileRangeDesc();
+        method.invoke(node, rangeDesc, split);
+
+        Assert.assertEquals(TFileFormatType.FORMAT_ORC, rangeDesc.getFormatType());
+    }
 }

regression-test/data/external_table_p0/iceberg/test_iceberg_mixed_file_formats.out

@@ -0,0 +1,12 @@
+-- This file is automatically generated. You should know what you did if you want to edit this
+-- !mixed_file_formats_files --
+orc	3
+parquet	3
+
+-- !mixed_file_formats_data --
+1	parquet
+2	parquet
+3	parquet
+4	orc
+5	orc
+6	orc

regression-test/suites/external_table_p0/iceberg/test_iceberg_mixed_file_formats.groovy

@@ -0,0 +1,56 @@
+// 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_iceberg_mixed_file_formats", "p0,external,iceberg,external_docker,external_docker_iceberg") {
+    String enabled = context.config.otherConfigs.get("enableIcebergTest")
+    if (enabled == null || !enabled.equalsIgnoreCase("true")) {
+        logger.info("disable iceberg test.")
+        return
+    }
+
+    String catalogName = "test_iceberg_mixed_file_formats"
+    String restPort = context.config.otherConfigs.get("iceberg_rest_uri_port")
+    String minioPort = context.config.otherConfigs.get("iceberg_minio_port")
+    String externalEnvIp = context.config.otherConfigs.get("externalEnvIp")
+
+    sql """drop catalog if exists ${catalogName}"""
+    sql """
+        CREATE CATALOG ${catalogName} PROPERTIES (
+            'type' = 'iceberg',
+            'iceberg.catalog.type' = 'rest',
+            'uri' = 'http://${externalEnvIp}:${restPort}',
+            's3.access_key' = 'admin',
+            's3.secret_key' = 'password',
+            's3.endpoint' = 'http://${externalEnvIp}:${minioPort}',
+            's3.region' = 'us-east-1'
+        )
+    """
+    sql """switch ${catalogName}"""
+    sql """ set parallel_pipeline_task_num = 1; """
+    order_qt_mixed_file_formats_files """
+        SELECT lower(file_format), sum(record_count)
+        FROM test_db.mixed_file_format\$files
+        GROUP BY lower(file_format)
+        ORDER BY 1
+    """
+
+    order_qt_mixed_file_formats_data """
+        SELECT id, source
+        FROM test_db.mixed_file_format
+        ORDER BY id
+    """
+}