From e72bab3b615d47c86e1d782959cd908d9a4b5570 Mon Sep 17 00:00:00 2001 From: zhangstar333 Date: Fri, 17 Jul 2026 15:46:50 +0800 Subject: [PATCH 1/4] [bug](iceberg) use split file format for iceberg scan --- .../doris/datasource/FileQueryScanNode.java | 7 +++-- .../datasource/iceberg/IcebergUtils.java | 26 +++---------------- .../iceberg/source/IcebergScanNode.java | 24 ++++++++++------- .../iceberg/source/IcebergSplit.java | 3 +++ .../datasource/iceberg/IcebergUtilsTest.java | 21 +++++++++++++++ .../iceberg/source/IcebergScanNodeTest.java | 22 ++++++++++++++++ 6 files changed, 68 insertions(+), 35 deletions(-) diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/FileQueryScanNode.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/FileQueryScanNode.java index 55da4f30ad7043..82867528fdd24c 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/datasource/FileQueryScanNode.java +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/FileQueryScanNode.java @@ -476,9 +476,12 @@ 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()); setScanParams(rangeDesc, fileSplit); + // Connector-specific scan params may provide the actual format for this split. + // Otherwise, fall back to the scan-level default initialized once in createScanRangeLocations(). + if (!rangeDesc.isSetFormatType()) { + rangeDesc.setFormatType(params.getFormatType()); + } rangeDesc.setFileCacheAdmission(admissionResult); curLocations.getScanRange().getExtScanRange().getFileScanRange().addToRanges(rangeDesc); diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergUtils.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergUtils.java index 12b5dce11395fa..b6f1d22046d76e 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergUtils.java +++ b/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 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 properties) { + private static String resolveFileFormatName(Map 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 files = icebergTable.newScan().planFiles()) { - java.util.Iterator 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; } diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/source/IcebergScanNode.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/source/IcebergScanNode.java index 70ac1d26b108fe..e9dcba6050365a 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/source/IcebergScanNode.java +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/source/IcebergScanNode.java @@ -325,6 +325,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 +496,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 +956,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 +1379,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 diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/source/IcebergSplit.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/source/IcebergSplit.java index 59b4f483f019e3..eeeff694b8ebce 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/source/IcebergSplit.java +++ b/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; diff --git a/fe/fe-core/src/test/java/org/apache/doris/datasource/iceberg/IcebergUtilsTest.java b/fe/fe-core/src/test/java/org/apache/doris/datasource/iceberg/IcebergUtilsTest.java index ee34b4bee50b0c..6e26972aa281ba 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/datasource/iceberg/IcebergUtilsTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/datasource/iceberg/IcebergUtilsTest.java @@ -78,6 +78,27 @@ import java.util.UUID; public class IcebergUtilsTest { + @Test + public void testGetFileFormatUsesPropertiesWithoutPlanningDataFiles() { + Table table = Mockito.mock(Table.class); + Mockito.when(table.properties()).thenReturn(Collections.emptyMap()); + + 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(); diff --git a/fe/fe-core/src/test/java/org/apache/doris/datasource/iceberg/source/IcebergScanNodeTest.java b/fe/fe-core/src/test/java/org/apache/doris/datasource/iceberg/source/IcebergScanNodeTest.java index 2a331f57462152..a312345cc4def2 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/datasource/iceberg/source/IcebergScanNodeTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/datasource/iceberg/source/IcebergScanNodeTest.java @@ -27,6 +27,7 @@ import org.apache.doris.qe.SessionVariable; import org.apache.doris.system.Backend; import org.apache.doris.thrift.TFileRangeDesc; +import org.apache.doris.thrift.TFileFormatType; import org.apache.doris.thrift.TIcebergDeleteFileDesc; import org.apache.iceberg.DataFile; @@ -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))); From 1db5d6235d014e916e6ce0b5e4ab6a2dde6cc6b7 Mon Sep 17 00:00:00 2001 From: zhangstar333 Date: Fri, 17 Jul 2026 16:23:46 +0800 Subject: [PATCH 2/4] update case --- .../iceberg/run31.sql | 32 +++++++++++ .../iceberg/source/IcebergScanNode.java | 1 - .../iceberg/source/IcebergScanNodeTest.java | 2 +- .../test_iceberg_mixed_file_formats.out | 12 ++++ .../test_iceberg_mixed_file_formats.groovy | 56 +++++++++++++++++++ 5 files changed, 101 insertions(+), 2 deletions(-) create mode 100644 docker/thirdparties/docker-compose/iceberg/scripts/create_preinstalled_scripts/iceberg/run31.sql create mode 100644 regression-test/data/external_table_p0/iceberg/test_iceberg_mixed_file_formats.out create mode 100644 regression-test/suites/external_table_p0/iceberg/test_iceberg_mixed_file_formats.groovy diff --git a/docker/thirdparties/docker-compose/iceberg/scripts/create_preinstalled_scripts/iceberg/run31.sql b/docker/thirdparties/docker-compose/iceberg/scripts/create_preinstalled_scripts/iceberg/run31.sql new file mode 100644 index 00000000000000..2cbb4677e32f1f --- /dev/null +++ b/docker/thirdparties/docker-compose/iceberg/scripts/create_preinstalled_scripts/iceberg/run31.sql @@ -0,0 +1,32 @@ +-- Reproducer for an Iceberg table whose active snapshot contains both +-- Parquet and ORC data files. Run this file once in the Spark Iceberg container. +-- It deliberately does not drop the table, so rerunning requires a new table name. + +CREATE DATABASE IF NOT EXISTS demo.test_db; +USE demo.test_db; + +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'); diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/source/IcebergScanNode.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/source/IcebergScanNode.java index e9dcba6050365a..a13d59a0e30900 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/source/IcebergScanNode.java +++ b/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; diff --git a/fe/fe-core/src/test/java/org/apache/doris/datasource/iceberg/source/IcebergScanNodeTest.java b/fe/fe-core/src/test/java/org/apache/doris/datasource/iceberg/source/IcebergScanNodeTest.java index a312345cc4def2..ec72503f7a674f 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/datasource/iceberg/source/IcebergScanNodeTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/datasource/iceberg/source/IcebergScanNodeTest.java @@ -26,8 +26,8 @@ import org.apache.doris.planner.ScanContext; import org.apache.doris.qe.SessionVariable; import org.apache.doris.system.Backend; -import org.apache.doris.thrift.TFileRangeDesc; import org.apache.doris.thrift.TFileFormatType; +import org.apache.doris.thrift.TFileRangeDesc; import org.apache.doris.thrift.TIcebergDeleteFileDesc; import org.apache.iceberg.DataFile; diff --git a/regression-test/data/external_table_p0/iceberg/test_iceberg_mixed_file_formats.out b/regression-test/data/external_table_p0/iceberg/test_iceberg_mixed_file_formats.out new file mode 100644 index 00000000000000..42771b3c4e3891 --- /dev/null +++ b/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 diff --git a/regression-test/suites/external_table_p0/iceberg/test_iceberg_mixed_file_formats.groovy b/regression-test/suites/external_table_p0/iceberg/test_iceberg_mixed_file_formats.groovy new file mode 100644 index 00000000000000..ebcb2affa29b79 --- /dev/null +++ b/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 + """ +} From 1ea1e69750b9f02a622196bd1f1dc431e0c48ef7 Mon Sep 17 00:00:00 2001 From: zhangstar333 Date: Fri, 17 Jul 2026 16:27:55 +0800 Subject: [PATCH 3/4] update --- .../org/apache/doris/datasource/FileQueryScanNode.java | 8 +++----- 1 file changed, 3 insertions(+), 5 deletions(-) diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/FileQueryScanNode.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/FileQueryScanNode.java index 82867528fdd24c..a0de7fdbe2ebc3 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/datasource/FileQueryScanNode.java +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/FileQueryScanNode.java @@ -476,12 +476,10 @@ private TScanRangeLocations splitToScanRange( pathPartitionKeys, partitionValuesFromPath.getIsNull()); TFileCompressType fileCompressType = getFileCompressType(fileSplit); rangeDesc.setCompressType(fileCompressType); + // 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); - // Connector-specific scan params may provide the actual format for this split. - // Otherwise, fall back to the scan-level default initialized once in createScanRangeLocations(). - if (!rangeDesc.isSetFormatType()) { - rangeDesc.setFormatType(params.getFormatType()); - } rangeDesc.setFileCacheAdmission(admissionResult); curLocations.getScanRange().getExtScanRange().getFileScanRange().addToRanges(rangeDesc); From 2eda0b5de745e1bc32e3ec4383abca99bb5c5b4b Mon Sep 17 00:00:00 2001 From: zhangstar333 Date: Fri, 17 Jul 2026 17:54:56 +0800 Subject: [PATCH 4/4] update --- .../scripts/create_preinstalled_scripts/iceberg/run31.sql | 7 ++++--- .../apache/doris/datasource/iceberg/IcebergUtilsTest.java | 1 + 2 files changed, 5 insertions(+), 3 deletions(-) diff --git a/docker/thirdparties/docker-compose/iceberg/scripts/create_preinstalled_scripts/iceberg/run31.sql b/docker/thirdparties/docker-compose/iceberg/scripts/create_preinstalled_scripts/iceberg/run31.sql index 2cbb4677e32f1f..b1cf79240bb657 100644 --- a/docker/thirdparties/docker-compose/iceberg/scripts/create_preinstalled_scripts/iceberg/run31.sql +++ b/docker/thirdparties/docker-compose/iceberg/scripts/create_preinstalled_scripts/iceberg/run31.sql @@ -1,10 +1,11 @@ --- Reproducer for an Iceberg table whose active snapshot contains both --- Parquet and ORC data files. Run this file once in the Spark Iceberg container. --- It deliberately does not drop the table, so rerunning requires a new table name. +-- 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 diff --git a/fe/fe-core/src/test/java/org/apache/doris/datasource/iceberg/IcebergUtilsTest.java b/fe/fe-core/src/test/java/org/apache/doris/datasource/iceberg/IcebergUtilsTest.java index 6e26972aa281ba..349e1e0bb82ab1 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/datasource/iceberg/IcebergUtilsTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/datasource/iceberg/IcebergUtilsTest.java @@ -82,6 +82,7 @@ public class IcebergUtilsTest { 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()