From ea950dd2669d9242f973cadfec2e824a1542c50f Mon Sep 17 00:00:00 2001 From: zhangstar333 Date: Sat, 18 Jul 2026 19:34:11 +0800 Subject: [PATCH] [bug](iceberg) use split file format for iceberg scan (#65760) Problem Summary: A table maybe contain mixed parquet and orc files. before used the table-level format for every split files, it's should use from data file. and when table properties were absent, maybe trigger full planFiles() call for every splits, this cause mush times. --- .../iceberg/run31.sql | 33 +++++++++++ .../doris/datasource/FileQueryScanNode.java | 5 +- .../datasource/iceberg/IcebergUtils.java | 26 +-------- .../iceberg/source/IcebergScanNode.java | 26 +++++---- .../iceberg/source/IcebergSplit.java | 3 + .../datasource/iceberg/IcebergUtilsTest.java | 23 ++++++++ .../iceberg/source/IcebergScanNodeTest.java | 25 +++++++++ .../test_iceberg_mixed_file_formats.out | 12 ++++ .../test_iceberg_mixed_file_formats.groovy | 56 +++++++++++++++++++ 9 files changed, 173 insertions(+), 36 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..b1cf79240bb657 --- /dev/null +++ b/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'); 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 3037eb5e497323..8fc003bd520562 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 @@ -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); 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 222809ca6a41df..cc8970d3d62566 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 @@ -1059,7 +1059,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; @@ -1071,7 +1071,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); @@ -1080,27 +1080,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 3456a0fdf8a97a..07423573423a6a 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; @@ -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 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 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 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 0d08f53b3e8525..339887e5c30fb6 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 @@ -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 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, 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 3f4828efb51a7a..3330685b85a4b7 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 @@ -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 { 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 e16ee803dcaef0..8b42f50774cf54 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 @@ -19,11 +19,16 @@ 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; @@ -31,6 +36,7 @@ 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()); + } } 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..0dfa10f0fb1662 --- /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 + """ +}