[bug](iceberg) use split file format for iceberg scan

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

@@ -476,8 +476,9 @@ private TScanRangeLocations splitToScanRange(
                 pathPartitionKeys, partitionValuesFromPath.getIsNull());
         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);
         rangeDesc.setFileCacheAdmission(admissionResult);
 

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

@@ -1293,7 +1293,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;
@@ -1305,7 +1305,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);
@@ -1314,27 +1314,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;
@@ -325,6 +324,8 @@ private void setIcebergParams(TFileRangeDesc rangeDesc, IcebergSplit icebergSpli
             rangeDesc.unsetColumnsFromPathIsNull();
             return;
         }
+        // update for every split file format
+        rangeDesc.setFormatType(toTFileFormatType(icebergSplit.getSplitFileFormat()));
         if (tableLevelPushDownCount) {
             tableFormatFileDesc.setTableLevelRowCount(icebergSplit.getTableLevelRowCount());
         } else {
@@ -494,6 +495,15 @@ private void setDeleteFileFormat(TIcebergDeleteFileDesc deleteFileDesc, FileForm
         }
     }
 
+    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) {
@@ -945,6 +955,7 @@ private Split createIcebergSplit(FileScanTask fileScanTask) {
                 storagePropertiesMap,
                 new ArrayList<>(),
                 originalPath);
+        split.setSplitFileFormat(dataFile.format());
         if (formatVersion >= 3) {
             // -1 means that this table was just upgraded from v2 to v3.
             // _row_id and _last_updated_sequence_number column is NULL.
@@ -1367,16 +1378,8 @@ public TFileFormatType getFileFormatType() throws UserException {
         if (isSystemTable) {
             return TFileFormatType.FORMAT_JNI;
         }
-        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

@@ -24,6 +24,7 @@
 
 import lombok.Data;
 import org.apache.iceberg.DeleteFile;
+import org.apache.iceberg.FileFormat;
 
 import java.util.ArrayList;
 import java.util.Collections;
@@ -53,6 +54,8 @@ public class IcebergSplit extends FileSplit {
     private Long firstRowId = null;
     private Long lastUpdatedSequenceNumber = null;
     private String serializedSplit;
+    // maybe mixed file format type in one table. so need record it for every split
+    private FileFormat splitFileFormat;
     private boolean positionDeleteSystemTableSplit = false;
     private TFileFormatType positionDeleteFileFormat;
     private int positionDeleteContent;

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

@@ -78,6 +78,28 @@
 import java.util.UUID;
 
 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 testGetIcebergViewUsesSessionCatalogWithDelegatedCredential() {
         ConnectContext context = new ConnectContext();

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

@@ -26,6 +26,7 @@
 import org.apache.doris.planner.ScanContext;
 import org.apache.doris.qe.SessionVariable;
 import org.apache.doris.system.Backend;
+import org.apache.doris.thrift.TFileFormatType;
 import org.apache.doris.thrift.TFileRangeDesc;
 import org.apache.doris.thrift.TIcebergDeleteFileDesc;
 
@@ -180,6 +181,7 @@ public void testSetIcebergParamsKeepsDeletionVectorOffsetAsLong() throws Excepti
         IcebergSplit split = new IcebergSplit(LocationPath.of(dataPath), 0, 128, 128, new String[0],
                 3, Collections.emptyMap(), new ArrayList<>(), dataPath);
         split.setTableFormatType(TableFormatType.ICEBERG);
+        split.setSplitFileFormat(FileFormat.PARQUET);
         split.setFirstRowId(10L);
         split.setLastUpdatedSequenceNumber(20L);
         split.setDeleteFileFilters(Collections.emptyList(), Collections.singletonList(
@@ -201,6 +203,25 @@ public void testSetIcebergParamsKeepsDeletionVectorOffsetAsLong() throws Excepti
         Assert.assertEquals((long) Integer.MAX_VALUE + 7L, deleteFileDesc.getContentSizeInBytes());
     }
 
+    @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());
+    }
+
     @Test
     public void testSetIcebergParamsPropagatesPositionDeleteFileFormat() throws Exception {
         SessionVariable sv = new SessionVariable();
@@ -215,6 +236,7 @@ public void testSetIcebergParamsPropagatesPositionDeleteFileFormat() throws Exce
         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.PARQUET);
         split.setDeleteFileFilters(Collections.emptyList(), Collections.singletonList(
                 new IcebergDeleteFileFilter.PositionDelete(deletePath, -1L, -1L, 256L,
                         org.apache.iceberg.FileFormat.ORC)));

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
+    """
+}