This is an automated email from the ASF dual-hosted git repository.
Gabriel39 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new 25bdeb83e76 [fix](iceberg) Honor disabled write metrics (#65782)
25bdeb83e76 is described below
commit 25bdeb83e769eb162f9ecdcc03135894dbb00ffd
Author: Gabriel <[email protected]>
AuthorDate: Tue Jul 21 16:07:56 2026 +0800
[fix](iceberg) Honor disabled write metrics (#65782)
## Summary
- honor Iceberg table metrics configuration when Doris creates data-file
metadata
- omit all column metrics whose effective Iceberg metrics mode is `none`
- omit bounds for `counts` and safely truncate string/binary bounds for
`truncate(N)`
- build the table metrics policy once per commit batch instead of once
per output file
- return unknown Iceberg column statistics when required file metrics
are absent
- preserve unknown statistics when closing an Iceberg file scan fails
- add regression coverage for the write-to-statistics path
## Root cause
Doris collected column statistics in the backend and copied every
statistics map into the Iceberg `DataFile` manifest in the frontend. The
conversion never consulted the table's `MetricsConfig`, so metadata for
disabled columns was persisted even though the physical file statistics
were available only as an implementation detail. The initial filtering
also treated every non-`none` mode like `full`, retaining bounds for
`counts` and failing to safely truncate string/binary bounds for
`truncate(N)`.
After disabled metrics were correctly omitted, the downstream Iceberg
statistics reader still assumed `columnSizes` and `nullValueCounts`
always contained every column. It could therefore throw while loading
statistics for a table using metrics mode `none`. A scan-close failure
could also override an in-flight unknown-statistics return and expose
partial or fabricated accumulators. The reader now treats missing
required metrics and scan-close failures as unknown.
## Testing
- `IcebergWriterHelperTest` and `StatisticsUtilTest` (21 tests passed)
- verified the new review regression tests fail before their production
fixes and pass after them
- FE CheckStyle validation (0 violations)
- `git diff --check`
## Links
- Jira: http://39.106.86.136:8090/browse/DORIS-27023
- TeamCity reproduction:
http://172.20.48.17:8111/buildConfiguration/Doris_Doris_x64_Master_Trino_Case/201908
---
be/src/exec/sink/viceberg_merge_sink.cpp | 3 +
.../writer/iceberg/viceberg_partition_writer.cpp | 10 +-
.../writer/iceberg/viceberg_partition_writer.h | 3 +
be/src/format/transformer/vorc_transformer.cpp | 11 +-
.../iceberg/iceberg_partition_writer_test.cpp | 107 +++++++++
.../format/transformer/vorc_transformer_test.cpp | 107 +++++++++
.../doris/datasource/iceberg/IcebergUtils.java | 24 +-
.../iceberg/helper/IcebergWriterHelper.java | 117 +++++++++-
.../org/apache/doris/planner/IcebergMergeSink.java | 1 +
.../org/apache/doris/planner/IcebergTableSink.java | 1 +
.../doris/statistics/util/StatisticsUtil.java | 14 +-
.../iceberg/helper/IcebergWriterHelperTest.java | 251 +++++++++++++++++++++
.../apache/doris/planner/IcebergMergeSinkTest.java | 70 +++++-
.../doris/statistics/util/StatisticsUtilTest.java | 70 ++++++
gensrc/thrift/DataSinks.thrift | 4 +
15 files changed, 780 insertions(+), 13 deletions(-)
diff --git a/be/src/exec/sink/viceberg_merge_sink.cpp
b/be/src/exec/sink/viceberg_merge_sink.cpp
index 7d6c84cf068..e9f4d6bf5c6 100644
--- a/be/src/exec/sink/viceberg_merge_sink.cpp
+++ b/be/src/exec/sink/viceberg_merge_sink.cpp
@@ -254,6 +254,9 @@ Status VIcebergMergeSink::_build_inner_sinks() {
if (merge_sink.__isset.broker_addresses) {
table_sink.__set_broker_addresses(merge_sink.broker_addresses);
}
+ if (merge_sink.__isset.collect_column_stats) {
+ table_sink.__set_collect_column_stats(merge_sink.collect_column_stats);
+ }
_table_sink.__set_type(TDataSinkType::ICEBERG_TABLE_SINK);
_table_sink.__set_iceberg_table_sink(table_sink);
diff --git a/be/src/exec/sink/writer/iceberg/viceberg_partition_writer.cpp
b/be/src/exec/sink/writer/iceberg/viceberg_partition_writer.cpp
index 5f28c41c24d..5958165ad03 100644
--- a/be/src/exec/sink/writer/iceberg/viceberg_partition_writer.cpp
+++ b/be/src/exec/sink/writer/iceberg/viceberg_partition_writer.cpp
@@ -47,7 +47,11 @@ VIcebergPartitionWriter::VIcebergPartitionWriter(
_file_name_index(file_name_index),
_file_format_type(file_format_type),
_compress_type(compress_type),
- _hadoop_conf(hadoop_conf) {}
+ _hadoop_conf(hadoop_conf) {
+ if (t_sink.iceberg_table_sink.__isset.collect_column_stats) {
+ _collect_column_stats = t_sink.iceberg_table_sink.collect_column_stats;
+ }
+}
Status VIcebergPartitionWriter::open(RuntimeState* state, RuntimeProfile*
profile,
const RowDescriptor* row_desc) {
@@ -156,6 +160,10 @@ Status
VIcebergPartitionWriter::_build_iceberg_commit_data(TIcebergCommitData* c
commit_data->__set_file_size(_file_format_transformer->written_len());
commit_data->__set_file_content(TFileContent::DATA);
commit_data->__set_partition_values(_partition_values);
+ // ORC collection reopens the file, so honor the FE policy before any
footer work.
+ if (!_collect_column_stats) {
+ return Status::OK();
+ }
if (_file_format_type == TFileFormatType::FORMAT_PARQUET) {
TIcebergColumnStats column_stats;
RETURN_IF_ERROR(static_cast<VParquetTransformer*>(_file_format_transformer.get())
diff --git a/be/src/exec/sink/writer/iceberg/viceberg_partition_writer.h
b/be/src/exec/sink/writer/iceberg/viceberg_partition_writer.h
index 97b4dd3efdf..b0839a82ed3 100644
--- a/be/src/exec/sink/writer/iceberg/viceberg_partition_writer.h
+++ b/be/src/exec/sink/writer/iceberg/viceberg_partition_writer.h
@@ -67,6 +67,8 @@ public:
inline size_t written_len() const override { return
_file_format_transformer->written_len(); }
private:
+ friend class VIcebergPartitionWriterTest;
+
std::string _get_target_file_name();
Status _build_iceberg_commit_data(TIcebergCommitData* commit_data);
@@ -91,6 +93,7 @@ private:
TFileFormatType::type _file_format_type;
TFileCompressType::type _compress_type;
const std::map<std::string, std::string>& _hadoop_conf;
+ bool _collect_column_stats = true;
std::shared_ptr<io::FileSystem> _fs = nullptr;
diff --git a/be/src/format/transformer/vorc_transformer.cpp
b/be/src/format/transformer/vorc_transformer.cpp
index 7dfa9fb4c64..7ddab79dc1c 100644
--- a/be/src/format/transformer/vorc_transformer.cpp
+++ b/be/src/format/transformer/vorc_transformer.cpp
@@ -385,12 +385,19 @@ Status
VOrcTransformer::collect_file_statistics_after_close(TIcebergColumnStats*
const iceberg::StructType& root_struct =
_iceberg_schema->root_struct();
const auto& nested_fields = root_struct.fields();
+ const orc::Type& orc_root_type = reader->getType();
for (uint32_t i = 0; i < nested_fields.size(); i++) {
- uint32_t orc_col_id = i + 1; // skip root struct
- if (orc_col_id >= file_stats->getNumberOfColumns()) {
+ if (i >= orc_root_type.getSubtypeCount()) {
+ continue;
+ }
+ // ORC IDs are depth-first, so top-level fields after a complex
field are not i + 1.
+ const uint64_t raw_orc_col_id =
orc_root_type.getSubtype(i)->getColumnId();
+ if (raw_orc_col_id >= file_stats->getNumberOfColumns()) {
continue;
}
+ // The uint32_t column-count check above makes narrowing to the
ORC API width safe.
+ const uint32_t orc_col_id = static_cast<uint32_t>(raw_orc_col_id);
const orc::ColumnStatistics* col_stats =
file_stats->getColumnStatistics(orc_col_id);
if (col_stats == nullptr) {
continue;
diff --git a/be/test/exec/sink/writer/iceberg/iceberg_partition_writer_test.cpp
b/be/test/exec/sink/writer/iceberg/iceberg_partition_writer_test.cpp
new file mode 100644
index 00000000000..d453177cf25
--- /dev/null
+++ b/be/test/exec/sink/writer/iceberg/iceberg_partition_writer_test.cpp
@@ -0,0 +1,107 @@
+// 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.
+
+#include <gtest/gtest.h>
+
+#include <optional>
+
+#include "exec/sink/writer/iceberg/viceberg_partition_writer.h"
+
+namespace doris {
+
+namespace {
+
+class FakeFileFormatTransformer final : public VFileFormatTransformer {
+public:
+ explicit FakeFileFormatTransformer(const VExprContextSPtrs& output_exprs)
+ : VFileFormatTransformer(nullptr, output_exprs, false) {}
+
+ Status open() override { return Status::OK(); }
+ Status write(const Block&) override { return Status::OK(); }
+ Status close() override { return Status::OK(); }
+ int64_t written_len() override { return 64; }
+};
+
+TDataSink make_table_sink(std::optional<bool> collect_column_stats) {
+ TIcebergTableSink iceberg_sink;
+ if (collect_column_stats.has_value()) {
+ iceberg_sink.__set_collect_column_stats(*collect_column_stats);
+ }
+ TDataSink sink;
+ sink.__set_type(TDataSinkType::ICEBERG_TABLE_SINK);
+ sink.__set_iceberg_table_sink(iceberg_sink);
+ return sink;
+}
+
+} // namespace
+
+class VIcebergPartitionWriterTest : public testing::Test {
+protected:
+ static std::unique_ptr<VIcebergPartitionWriter> make_writer(
+ const TDataSink& sink, const VExprContextSPtrs& output_exprs,
+ const iceberg::Schema& schema, const std::string* schema_json,
+ const std::map<std::string, std::string>& hadoop_conf) {
+ IPartitionWriterBase::WriteInfo write_info;
+ write_info.file_type = TFileType::FILE_LOCAL;
+ return std::make_unique<VIcebergPartitionWriter>(
+ sink, std::vector<std::string> {}, output_exprs, schema,
schema_json,
+ std::vector<std::string> {}, std::move(write_info), "data", 0,
+ TFileFormatType::FORMAT_ORC, TFileCompressType::ZLIB,
hadoop_conf);
+ }
+
+ static void install_fake_transformer(VIcebergPartitionWriter* writer,
+ const VExprContextSPtrs&
output_exprs) {
+ writer->_file_format_transformer =
+ std::make_unique<FakeFileFormatTransformer>(output_exprs);
+ }
+
+ static Status build_commit_data(VIcebergPartitionWriter* writer,
+ TIcebergCommitData* commit_data) {
+ return writer->_build_iceberg_commit_data(commit_data);
+ }
+
+ static bool collect_column_stats(const VIcebergPartitionWriter& writer) {
+ return writer._collect_column_stats;
+ }
+};
+
+TEST_F(VIcebergPartitionWriterTest,
OrcSkipsFooterCollectionWhenMetricsAreDisabled) {
+ VExprContextSPtrs output_exprs;
+ iceberg::Schema schema(std::vector<iceberg::NestedField> {});
+ std::string schema_json;
+ std::map<std::string, std::string> hadoop_conf;
+ auto writer =
+ make_writer(make_table_sink(false), output_exprs, schema,
&schema_json, hadoop_conf);
+ install_fake_transformer(writer.get(), output_exprs);
+
+ TIcebergCommitData commit_data;
+ ASSERT_TRUE(build_commit_data(writer.get(), &commit_data).ok());
+ EXPECT_FALSE(commit_data.__isset.column_stats);
+}
+
+TEST_F(VIcebergPartitionWriterTest,
MissingPolicyKeepsCollectionEnabledForRollingUpgrade) {
+ VExprContextSPtrs output_exprs;
+ iceberg::Schema schema(std::vector<iceberg::NestedField> {});
+ std::string schema_json;
+ std::map<std::string, std::string> hadoop_conf;
+ auto writer = make_writer(make_table_sink(std::nullopt), output_exprs,
schema, &schema_json,
+ hadoop_conf);
+
+ EXPECT_TRUE(collect_column_stats(*writer));
+}
+
+} // namespace doris
diff --git a/be/test/format/transformer/vorc_transformer_test.cpp
b/be/test/format/transformer/vorc_transformer_test.cpp
new file mode 100644
index 00000000000..4ea14766356
--- /dev/null
+++ b/be/test/format/transformer/vorc_transformer_test.cpp
@@ -0,0 +1,107 @@
+// 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.
+
+#include "format/transformer/vorc_transformer.h"
+
+#include <gtest/gtest.h>
+
+#include "core/block/block.h"
+#include "core/column/column_string.h"
+#include "core/column/column_struct.h"
+#include "core/column/column_vector.h"
+#include "core/data_type/data_type_number.h"
+#include "core/data_type/data_type_string.h"
+#include "core/data_type/data_type_struct.h"
+#include "format/table/iceberg/schema_parser.h"
+#include "io/fs/local_file_system.h"
+#include "runtime/runtime_state.h"
+#include "testutil/mock/mock_slot_ref.h"
+#include "util/uid_util.h"
+
+namespace doris {
+
+class VOrcTransformerTest : public testing::Test {
+protected:
+ void SetUp() override {
+ _file_path = "./vorc_transformer_" + UniqueId::gen_uid().to_string() +
".orc";
+ _fs = io::global_local_filesystem();
+ }
+
+ void TearDown() override {
static_cast<void>(_fs->delete_file(_file_path)); }
+
+ std::string _file_path;
+ std::shared_ptr<io::FileSystem> _fs;
+};
+
+TEST_F(VOrcTransformerTest, CollectsBoundsForTopLevelFieldAfterStruct) {
+ auto int_type = std::make_shared<DataTypeInt32>();
+ auto struct_type = std::make_shared<DataTypeStruct>(DataTypes {int_type},
Strings {"a"});
+ auto string_type = std::make_shared<DataTypeString>();
+ VExprContextSPtrs output_exprs =
+ MockSlotRef::create_mock_contexts(DataTypes {struct_type,
string_type});
+
+ const std::string schema_json = R"({
+ "type": "struct",
+ "fields": [
+ {
+ "id": 1,
+ "name": "s",
+ "required": true,
+ "type": {
+ "type": "struct",
+ "fields": [
+ {"id": 2, "name": "a", "required": true, "type": "int"}
+ ]
+ }
+ },
+ {"id": 3, "name": "b", "required": true, "type": "string"}
+ ]
+ })";
+ std::unique_ptr<iceberg::Schema> schema =
iceberg::SchemaParser::from_json(schema_json);
+
+ io::FileWriterPtr file_writer;
+ ASSERT_TRUE(_fs->create_file(_file_path, &file_writer).ok());
+ RuntimeState state;
+ VOrcTransformer transformer(&state, file_writer.get(), output_exprs, "",
{"s", "b"}, false,
+ TFileCompressType::PLAIN, schema.get(), _fs);
+ ASSERT_TRUE(transformer.open().ok());
+
+ auto nested_column = ColumnInt32::create();
+ nested_column->insert_value(-1);
+ Columns struct_columns;
+ struct_columns.emplace_back(std::move(nested_column));
+ auto struct_column = ColumnStruct::create(std::move(struct_columns));
+ auto string_column = ColumnString::create();
+ string_column->insert_data("hello", 5);
+
+ Block block;
+ block.insert(ColumnWithTypeAndName(std::move(struct_column), struct_type,
"s"));
+ block.insert(ColumnWithTypeAndName(std::move(string_column), string_type,
"b"));
+ ASSERT_TRUE(transformer.write(block).ok());
+ ASSERT_TRUE(transformer.close().ok());
+
+ TIcebergColumnStats stats;
+ ASSERT_TRUE(transformer.collect_file_statistics_after_close(&stats).ok());
+ ASSERT_TRUE(stats.__isset.lower_bounds);
+ ASSERT_TRUE(stats.__isset.upper_bounds);
+ ASSERT_EQ(1, stats.lower_bounds.count(3));
+ ASSERT_EQ(1, stats.upper_bounds.count(3));
+ EXPECT_EQ("hello", stats.lower_bounds.at(3));
+ EXPECT_EQ("hello", stats.upper_bounds.at(3));
+}
+
+} // namespace doris
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 b6f1d22046d..c9705b8492a 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
@@ -79,14 +79,17 @@ import com.google.common.collect.Sets;
import com.google.gson.reflect.TypeToken;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.exception.ExceptionUtils;
-import org.apache.iceberg.BaseTable;
import org.apache.iceberg.CatalogProperties;
import org.apache.iceberg.FileFormat;
import org.apache.iceberg.FileScanTask;
+import org.apache.iceberg.HasTableOperations;
import org.apache.iceberg.ManifestFile;
import org.apache.iceberg.MetadataColumns;
import org.apache.iceberg.MetadataTableType;
import org.apache.iceberg.MetadataTableUtils;
+import org.apache.iceberg.MetricsConfig;
+import org.apache.iceberg.MetricsModes;
+import org.apache.iceberg.MetricsUtil;
import org.apache.iceberg.PartitionData;
import org.apache.iceberg.PartitionField;
import org.apache.iceberg.PartitionSpec;
@@ -1961,10 +1964,25 @@ public class IcebergUtils {
MetadataColumns.ROW_ID,
MetadataColumns.LAST_UPDATED_SEQUENCE_NUMBER));
}
+ public static boolean shouldCollectColumnStats(Table table, Schema
writerSchema) {
+ MetricsConfig metricsConfig = MetricsConfig.forTable(table);
+ if (getFileFormat(table) == FileFormat.ORC) {
+ // Match the footer collectors: ORC reports top-level collection
counts, while Parquet reports leaf fields.
+ return writerSchema.columns().stream()
+ .anyMatch(field -> MetricsUtil.metricsMode(writerSchema,
metricsConfig, field.fieldId())
+ != MetricsModes.None.get());
+ }
+ return TypeUtil.indexById(writerSchema.asStruct()).values().stream()
+ .filter(field -> field.type().isPrimitiveType())
+ .anyMatch(field -> MetricsUtil.metricsMode(writerSchema,
metricsConfig, field.fieldId())
+ != MetricsModes.None.get());
+ }
+
public static int getFormatVersion(Table table) {
int formatVersion = 2; // default format version : 2
- if (table instanceof BaseTable) {
- formatVersion = ((BaseTable)
table).operations().current().formatVersion();
+ if (table instanceof HasTableOperations) {
+ // TransactionTable exposes the real format version through
operations, not table properties.
+ formatVersion = ((HasTableOperations)
table).operations().current().formatVersion();
} else if (table != null && table.properties() != null) {
String version =
table.properties().get(TableProperties.FORMAT_VERSION);
if (version != null) {
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/helper/IcebergWriterHelper.java
b/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/helper/IcebergWriterHelper.java
index b67a5911b64..54a791e7e18 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/helper/IcebergWriterHelper.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/helper/IcebergWriterHelper.java
@@ -30,12 +30,21 @@ import org.apache.iceberg.DeleteFile;
import org.apache.iceberg.FileFormat;
import org.apache.iceberg.FileMetadata;
import org.apache.iceberg.Metrics;
+import org.apache.iceberg.MetricsConfig;
+import org.apache.iceberg.MetricsModes;
+import org.apache.iceberg.MetricsUtil;
import org.apache.iceberg.PartitionData;
import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Schema;
import org.apache.iceberg.SortOrder;
import org.apache.iceberg.Table;
import org.apache.iceberg.io.WriteResult;
+import org.apache.iceberg.types.Conversions;
+import org.apache.iceberg.types.Type;
+import org.apache.iceberg.types.TypeUtil;
import org.apache.iceberg.types.Types;
+import org.apache.iceberg.util.BinaryUtil;
+import org.apache.iceberg.util.UnicodeUtil;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
@@ -61,6 +70,12 @@ public class IcebergWriterHelper {
// Get table specification information
PartitionSpec spec = table.spec();
FileFormat fileFormat = IcebergUtils.getFileFormat(table);
+ MetricsConfig metricsConfig = MetricsConfig.forTable(table);
+ Schema schema = table.schema();
+ if (IcebergUtils.getFormatVersion(table) >=
IcebergUtils.ICEBERG_ROW_LINEAGE_MIN_VERSION) {
+ // Rewrite and merge writers emit v3 lineage columns that are
absent from the table schema.
+ schema = IcebergUtils.appendRowLineageFieldsForV3(schema);
+ }
for (TIcebergCommitData commitData : commitDataList) {
//get the files path
@@ -70,7 +85,7 @@ public class IcebergWriterHelper {
long fileSize = commitData.getFileSize();
long recordCount = commitData.getRowCount();
CommonStatistics stat = new CommonStatistics(recordCount,
DEFAULT_FILE_COUNT, fileSize);
- Metrics metrics = buildDataFileMetrics(table, fileFormat,
commitData);
+ Metrics metrics = buildDataFileMetrics(commitData, schema,
metricsConfig, fileFormat);
Optional<PartitionData> partitionData = Optional.empty();
//get and check partitionValues when table is partitionedTable
if (spec.isPartitioned()) {
@@ -153,7 +168,9 @@ public class IcebergWriterHelper {
return partitionData;
}
- private static Metrics buildDataFileMetrics(Table table, FileFormat
fileFormat, TIcebergCommitData commitData) {
+ private static Metrics buildDataFileMetrics(
+ TIcebergCommitData commitData, Schema schema, MetricsConfig
metricsConfig, FileFormat fileFormat) {
+ Map<Integer, Integer> fieldParents =
TypeUtil.indexParents(schema.asStruct());
Map<Integer, Long> columnSizes = new HashMap<>();
Map<Integer, Long> valueCounts = new HashMap<>();
Map<Integer, Long> nullValueCounts = new HashMap<>();
@@ -178,8 +195,100 @@ public class IcebergWriterHelper {
}
}
- return new Metrics(commitData.getRowCount(), columnSizes, valueCounts,
- nullValueCounts, null, lowerBounds, upperBounds);
+ // Physical file stats may contain every column, but manifest metrics
must honor the table's metadata policy.
+ return new Metrics(commitData.getRowCount(),
+ filterDisabledMetrics(columnSizes, schema, metricsConfig),
+ filterLogicalMetrics(valueCounts, schema, metricsConfig,
fieldParents),
+ filterLogicalMetrics(nullValueCounts, schema, metricsConfig,
fieldParents),
+ null,
+ filterBounds(lowerBounds, schema, metricsConfig, fieldParents,
fileFormat, true),
+ filterBounds(upperBounds, schema, metricsConfig, fieldParents,
fileFormat, false));
+ }
+
+ private static <T> Map<Integer, T> filterDisabledMetrics(
+ Map<Integer, T> metrics, Schema schema, MetricsConfig
metricsConfig) {
+ Map<Integer, T> filteredMetrics = new HashMap<>();
+ metrics.forEach((fieldId, value) -> {
+ if (MetricsUtil.metricsMode(schema, metricsConfig, fieldId) !=
MetricsModes.None.get()) {
+ filteredMetrics.put(fieldId, value);
+ }
+ });
+ return filteredMetrics;
+ }
+
+ private static <T> Map<Integer, T> filterLogicalMetrics(
+ Map<Integer, T> metrics, Schema schema, MetricsConfig
metricsConfig,
+ Map<Integer, Integer> fieldParents) {
+ Map<Integer, T> filteredMetrics = new HashMap<>();
+ metrics.forEach((fieldId, value) -> {
+ // Definition-level values below list/map do not represent logical
element counts.
+ if (!isInRepeatedField(fieldId, schema, fieldParents)
+ && MetricsUtil.metricsMode(schema, metricsConfig, fieldId)
!= MetricsModes.None.get()) {
+ filteredMetrics.put(fieldId, value);
+ }
+ });
+ return filteredMetrics;
+ }
+
+ private static Map<Integer, ByteBuffer> filterBounds(
+ Map<Integer, ByteBuffer> bounds, Schema schema, MetricsConfig
metricsConfig,
+ Map<Integer, Integer> fieldParents, FileFormat fileFormat, boolean
lowerBound) {
+ Map<Integer, ByteBuffer> filteredBounds = new HashMap<>();
+ bounds.forEach((fieldId, value) -> {
+ if (isInRepeatedField(fieldId, schema, fieldParents)) {
+ return;
+ }
+ MetricsModes.MetricsMode mode = MetricsUtil.metricsMode(schema,
metricsConfig, fieldId);
+ if (mode == MetricsModes.None.get() || mode ==
MetricsModes.Counts.get()) {
+ return;
+ }
+
+ ByteBuffer filteredValue = value;
+ if (mode instanceof MetricsModes.Truncate) {
+ Type type = schema.findType(fieldId);
+ int length = ((MetricsModes.Truncate) mode).length();
+ // Truncated upper bounds must round up so file pruning cannot
exclude matching values.
+ filteredValue = truncateBound(type, value, length, fileFormat,
lowerBound);
+ }
+ if (filteredValue != null) {
+ filteredBounds.put(fieldId, filteredValue);
+ }
+ });
+ return filteredBounds;
+ }
+
+ private static boolean isInRepeatedField(
+ int fieldId, Schema schema, Map<Integer, Integer> fieldParents) {
+ Integer parentId = fieldId;
+ while ((parentId = fieldParents.get(parentId)) != null) {
+ Types.NestedField parent = schema.findField(parentId);
+ if (parent != null && !parent.type().isStructType()) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ private static ByteBuffer truncateBound(
+ Type type, ByteBuffer value, int length, FileFormat fileFormat,
boolean lowerBound) {
+ switch (type.typeId()) {
+ case STRING:
+ String stringValue = Conversions.fromByteBuffer(type,
value).toString();
+ String truncatedString = lowerBound
+ ? UnicodeUtil.truncateStringMin(stringValue, length)
+ : UnicodeUtil.truncateStringMax(stringValue, length);
+ // ORC keeps the full maximum when no safe truncated successor
exists.
+ if (!lowerBound && truncatedString == null && fileFormat ==
FileFormat.ORC) {
+ return value;
+ }
+ return truncatedString == null ? null :
Conversions.toByteBuffer(type, truncatedString);
+ case BINARY:
+ return lowerBound
+ ? BinaryUtil.truncateBinaryMin(value, length)
+ : BinaryUtil.truncateBinaryMax(value, length);
+ default:
+ return value;
+ }
}
/**
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/planner/IcebergMergeSink.java
b/fe/fe-core/src/main/java/org/apache/doris/planner/IcebergMergeSink.java
index 4af4ba17e18..9775f81c172 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/planner/IcebergMergeSink.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/planner/IcebergMergeSink.java
@@ -131,6 +131,7 @@ public class IcebergMergeSink extends
BaseExternalTableDataSink {
}
tSink.setFormatVersion(formatVersion);
tSink.setSchemaJson(SchemaParser.toJson(schema));
+
tSink.setCollectColumnStats(IcebergUtils.shouldCollectColumnStats(icebergTable,
schema));
// partition spec
if (icebergTable.spec().isPartitioned()) {
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/planner/IcebergTableSink.java
b/fe/fe-core/src/main/java/org/apache/doris/planner/IcebergTableSink.java
index 0f3b1bb24d2..b7d3da47cb4 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/planner/IcebergTableSink.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/planner/IcebergTableSink.java
@@ -135,6 +135,7 @@ public class IcebergTableSink extends
BaseExternalTableDataSink {
schema = IcebergUtils.appendRowLineageFieldsForV3(schema);
}
tSink.setSchemaJson(SchemaParser.toJson(schema));
+
tSink.setCollectColumnStats(IcebergUtils.shouldCollectColumnStats(icebergTable,
schema));
// partition spec
if (icebergTable.spec().isPartitioned()) {
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/statistics/util/StatisticsUtil.java
b/fe/fe-core/src/main/java/org/apache/doris/statistics/util/StatisticsUtil.java
index 10b139b73f0..d6838246999 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/statistics/util/StatisticsUtil.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/statistics/util/StatisticsUtil.java
@@ -605,12 +605,22 @@ public class StatisticsUtil {
try (CloseableIterable<FileScanTask> fileScanTasks =
tableScan.planFiles()) {
for (FileScanTask task : fileScanTasks) {
int colId = getColId(task.spec(), colName);
- totalDataSize += task.file().columnSizes().get(colId);
+ Map<Integer, Long> columnSizes = task.file().columnSizes();
+ Map<Integer, Long> nullValueCounts =
task.file().nullValueCounts();
+ Long columnSize = columnSizes == null ? null :
columnSizes.get(colId);
+ Long nullValueCount = nullValueCounts == null ? null :
nullValueCounts.get(colId);
+ // Iceberg can omit maps or entries for mode=none; partial
aggregation would fabricate zero stats.
+ if (columnSize == null || nullValueCount == null) {
+ return Optional.empty();
+ }
+ totalDataSize += columnSize;
totalDataCount += task.file().recordCount();
- totalNumNull += task.file().nullValueCounts().get(colId);
+ totalNumNull += nullValueCount;
}
} catch (IOException e) {
LOG.warn("Error to close FileScanTask.", e);
+ // A failed close can cancel an in-flight empty return, so
accumulated stats are not reliable.
+ return Optional.empty();
}
ColumnStatisticBuilder columnStatisticBuilder = new
ColumnStatisticBuilder(totalDataCount);
columnStatisticBuilder.setMaxValue(Double.POSITIVE_INFINITY);
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/datasource/iceberg/helper/IcebergWriterHelperTest.java
b/fe/fe-core/src/test/java/org/apache/doris/datasource/iceberg/helper/IcebergWriterHelperTest.java
index 77d318518f3..39fc15ddbe1 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/datasource/iceberg/helper/IcebergWriterHelperTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/datasource/iceberg/helper/IcebergWriterHelperTest.java
@@ -18,19 +18,34 @@
package org.apache.doris.datasource.iceberg.helper;
import org.apache.doris.thrift.TFileContent;
+import org.apache.doris.thrift.TIcebergColumnStats;
import org.apache.doris.thrift.TIcebergCommitData;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.iceberg.DataFile;
import org.apache.iceberg.DeleteFile;
import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.MetadataColumns;
import org.apache.iceberg.PartitionSpec;
import org.apache.iceberg.Schema;
+import org.apache.iceberg.SortOrder;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.TableProperties;
+import org.apache.iceberg.hadoop.HadoopTables;
+import org.apache.iceberg.io.WriteResult;
+import org.apache.iceberg.types.Conversions;
import org.apache.iceberg.types.Types;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+import org.mockito.Mockito;
+import java.nio.ByteBuffer;
+import java.nio.file.Path;
import java.util.ArrayList;
import java.util.List;
+import java.util.Map;
/**
* Test for IcebergWriterHelper DeleteFile conversion
@@ -58,6 +73,242 @@ public class IcebergWriterHelperTest {
}
+ @Test
+ public void testConvertToWriterResultRespectsNoneMetricsMode() {
+ Table table = Mockito.mock(Table.class);
+ Mockito.when(table.schema()).thenReturn(schema);
+ Mockito.when(table.spec()).thenReturn(unpartitionedSpec);
+ Mockito.when(table.sortOrder()).thenReturn(SortOrder.unsorted());
+ Mockito.when(table.properties()).thenReturn(Map.of(
+ TableProperties.DEFAULT_FILE_FORMAT, "parquet",
+ TableProperties.DEFAULT_WRITE_METRICS_MODE, "none"));
+
+ TIcebergColumnStats columnStats = new TIcebergColumnStats();
+ columnStats.setColumnSizes(Map.of(2, 128L));
+ columnStats.setValueCounts(Map.of(2, 10L));
+ columnStats.setNullValueCounts(Map.of(2, 0L));
+ columnStats.setLowerBounds(Map.of(2, ByteBuffer.wrap(new byte[]
{0x01})));
+ columnStats.setUpperBounds(Map.of(2, ByteBuffer.wrap(new byte[]
{0x02})));
+
+ TIcebergCommitData commitData = new TIcebergCommitData();
+ commitData.setFilePath("/path/to/data.parquet");
+ commitData.setRowCount(10);
+ commitData.setFileSize(1024);
+ commitData.setColumnStats(columnStats);
+
+ WriteResult result = IcebergWriterHelper.convertToWriterResult(table,
List.of(commitData));
+ DataFile dataFile = result.dataFiles()[0];
+
+ Assertions.assertTrue(dataFile.columnSizes() == null ||
dataFile.columnSizes().isEmpty());
+ Assertions.assertTrue(dataFile.valueCounts() == null ||
dataFile.valueCounts().isEmpty());
+ Assertions.assertTrue(dataFile.nullValueCounts() == null ||
dataFile.nullValueCounts().isEmpty());
+ Assertions.assertTrue(dataFile.lowerBounds() == null ||
dataFile.lowerBounds().isEmpty());
+ Assertions.assertTrue(dataFile.upperBounds() == null ||
dataFile.upperBounds().isEmpty());
+ }
+
+ @Test
+ public void testConvertToWriterResultCountsModeOmitsBounds() {
+ Table table = Mockito.mock(Table.class);
+ Mockito.when(table.schema()).thenReturn(schema);
+ Mockito.when(table.spec()).thenReturn(unpartitionedSpec);
+ Mockito.when(table.sortOrder()).thenReturn(SortOrder.unsorted());
+ Mockito.when(table.properties()).thenReturn(Map.of(
+ TableProperties.DEFAULT_FILE_FORMAT, "parquet",
+ TableProperties.DEFAULT_WRITE_METRICS_MODE, "counts"));
+
+ TIcebergColumnStats columnStats = new TIcebergColumnStats();
+ columnStats.setColumnSizes(Map.of(2, 128L));
+ columnStats.setValueCounts(Map.of(2, 10L));
+ columnStats.setNullValueCounts(Map.of(2, 0L));
+ columnStats.setLowerBounds(Map.of(
+ 2, Conversions.toByteBuffer(Types.StringType.get(),
"abcdefgh")));
+ columnStats.setUpperBounds(Map.of(
+ 2, Conversions.toByteBuffer(Types.StringType.get(),
"ijklmnop")));
+
+ TIcebergCommitData commitData = new TIcebergCommitData();
+ commitData.setFilePath("/path/to/data.parquet");
+ commitData.setRowCount(10);
+ commitData.setFileSize(1024);
+ commitData.setColumnStats(columnStats);
+
+ DataFile dataFile = IcebergWriterHelper.convertToWriterResult(table,
List.of(commitData)).dataFiles()[0];
+
+ Assertions.assertEquals(Map.of(2, 128L), dataFile.columnSizes());
+ Assertions.assertEquals(Map.of(2, 10L), dataFile.valueCounts());
+ Assertions.assertEquals(Map.of(2, 0L), dataFile.nullValueCounts());
+ Assertions.assertTrue(dataFile.lowerBounds() == null ||
dataFile.lowerBounds().isEmpty());
+ Assertions.assertTrue(dataFile.upperBounds() == null ||
dataFile.upperBounds().isEmpty());
+ }
+
+ @Test
+ public void testConvertToWriterResultTruncatesStringAndBinaryBounds() {
+ Schema boundsSchema = new Schema(
+ Types.NestedField.optional(1, "text", Types.StringType.get()),
+ Types.NestedField.optional(2, "payload",
Types.BinaryType.get()));
+ Table table = Mockito.mock(Table.class);
+ Mockito.when(table.schema()).thenReturn(boundsSchema);
+ Mockito.when(table.spec()).thenReturn(unpartitionedSpec);
+ Mockito.when(table.sortOrder()).thenReturn(SortOrder.unsorted());
+ Mockito.when(table.properties()).thenReturn(Map.of(
+ TableProperties.DEFAULT_FILE_FORMAT, "parquet",
+ TableProperties.DEFAULT_WRITE_METRICS_MODE, "truncate(3)"));
+
+ TIcebergColumnStats columnStats = new TIcebergColumnStats();
+ columnStats.setLowerBounds(Map.of(
+ 1, Conversions.toByteBuffer(Types.StringType.get(), "abcdef"),
+ 2, ByteBuffer.wrap(new byte[] {1, 2, 3, 4})));
+ columnStats.setUpperBounds(Map.of(
+ 1, Conversions.toByteBuffer(Types.StringType.get(), "uvwxyz"),
+ 2, ByteBuffer.wrap(new byte[] {1, 2, 3, 4})));
+
+ TIcebergCommitData commitData = new TIcebergCommitData();
+ commitData.setFilePath("/path/to/data.parquet");
+ commitData.setRowCount(10);
+ commitData.setFileSize(1024);
+ commitData.setColumnStats(columnStats);
+
+ DataFile dataFile = IcebergWriterHelper.convertToWriterResult(table,
List.of(commitData)).dataFiles()[0];
+
+ Assertions.assertEquals("abc", Conversions.fromByteBuffer(
+ Types.StringType.get(),
dataFile.lowerBounds().get(1)).toString());
+ Assertions.assertEquals("uvx", Conversions.fromByteBuffer(
+ Types.StringType.get(),
dataFile.upperBounds().get(1)).toString());
+ Assertions.assertEquals(ByteBuffer.wrap(new byte[] {1, 2, 3}),
dataFile.lowerBounds().get(2));
+ Assertions.assertEquals(ByteBuffer.wrap(new byte[] {1, 2, 4}),
dataFile.upperBounds().get(2));
+ }
+
+ @Test
+ public void
testConvertToWriterResultPreservesOrcUpperBoundWithoutTruncatedSuccessor() {
+ Schema boundsSchema = new Schema(
+ Types.NestedField.optional(1, "text", Types.StringType.get()));
+ Table table = Mockito.mock(Table.class);
+ Mockito.when(table.schema()).thenReturn(boundsSchema);
+ Mockito.when(table.spec()).thenReturn(unpartitionedSpec);
+ Mockito.when(table.sortOrder()).thenReturn(SortOrder.unsorted());
+ Mockito.when(table.properties()).thenReturn(Map.of(
+ TableProperties.DEFAULT_FILE_FORMAT, "orc",
+ TableProperties.DEFAULT_WRITE_METRICS_MODE, "truncate(1)"));
+
+ String maxWithoutSuccessor = new
String(Character.toChars(Character.MAX_CODE_POINT)) + "tail";
+ TIcebergColumnStats columnStats = new TIcebergColumnStats();
+ columnStats.setUpperBounds(Map.of(
+ 1, Conversions.toByteBuffer(Types.StringType.get(),
maxWithoutSuccessor)));
+
+ TIcebergCommitData commitData = new TIcebergCommitData();
+ commitData.setFilePath("/path/to/data.orc");
+ commitData.setRowCount(1);
+ commitData.setFileSize(128);
+ commitData.setColumnStats(columnStats);
+
+ DataFile dataFile = IcebergWriterHelper.convertToWriterResult(table,
List.of(commitData)).dataFiles()[0];
+
+ Assertions.assertNotNull(dataFile.upperBounds());
+ Assertions.assertEquals(maxWithoutSuccessor,
Conversions.fromByteBuffer(
+ Types.StringType.get(),
dataFile.upperBounds().get(1)).toString());
+ }
+
+ @Test
+ public void testConvertToWriterResultBuildsMetricsPolicyOncePerBatch() {
+ Table table = Mockito.mock(Table.class);
+ Mockito.when(table.schema()).thenReturn(schema);
+ Mockito.when(table.spec()).thenReturn(unpartitionedSpec);
+ Mockito.when(table.sortOrder()).thenReturn(SortOrder.unsorted());
+ Mockito.when(table.properties()).thenReturn(Map.of(
+ TableProperties.DEFAULT_FILE_FORMAT, "parquet",
+ TableProperties.DEFAULT_WRITE_METRICS_MODE, "none"));
+
+ TIcebergCommitData firstCommit = new TIcebergCommitData();
+ firstCommit.setFilePath("/path/to/first.parquet");
+ firstCommit.setRowCount(10);
+ firstCommit.setFileSize(1024);
+
+ TIcebergCommitData secondCommit = new TIcebergCommitData();
+ secondCommit.setFilePath("/path/to/second.parquet");
+ secondCommit.setRowCount(20);
+ secondCommit.setFileSize(2048);
+
+ IcebergWriterHelper.convertToWriterResult(table, List.of(firstCommit,
secondCommit));
+
+ // One schema lookup is made by Iceberg's policy builder and one is
captured for all files in the batch.
+ Mockito.verify(table, Mockito.times(2)).schema();
+ }
+
+ @Test
+ public void
testConvertToWriterResultHandlesV3TransactionTableLineageMetrics(@TempDir Path
tempDir) {
+ HadoopTables tables = new HadoopTables(new Configuration());
+ Table baseTable = tables.create(schema, unpartitionedSpec,
SortOrder.unsorted(), Map.of(
+ TableProperties.FORMAT_VERSION, "3",
+ TableProperties.DEFAULT_FILE_FORMAT, "parquet",
+ TableProperties.DEFAULT_WRITE_METRICS_MODE, "truncate(16)"),
+ tempDir.resolve("table").toUri().toString());
+ Table transactionTable = baseTable.newTransaction().table();
+
+ int rowId = MetadataColumns.ROW_ID.fieldId();
+ int sequenceNumberId =
MetadataColumns.LAST_UPDATED_SEQUENCE_NUMBER.fieldId();
+ ByteBuffer rowIdBound =
Conversions.toByteBuffer(MetadataColumns.ROW_ID.type(), 7L);
+ ByteBuffer sequenceNumberBound = Conversions.toByteBuffer(
+ MetadataColumns.LAST_UPDATED_SEQUENCE_NUMBER.type(), 3L);
+ TIcebergColumnStats columnStats = new TIcebergColumnStats();
+ columnStats.setLowerBounds(Map.of(rowId, rowIdBound, sequenceNumberId,
sequenceNumberBound));
+ columnStats.setUpperBounds(Map.of(rowId, rowIdBound, sequenceNumberId,
sequenceNumberBound));
+
+ TIcebergCommitData commitData = new TIcebergCommitData();
+ commitData.setFilePath("/path/to/v3-data.parquet");
+ commitData.setRowCount(1);
+ commitData.setFileSize(128);
+ commitData.setColumnStats(columnStats);
+
+ DataFile dataFile = Assertions.assertDoesNotThrow(
+ () -> IcebergWriterHelper.convertToWriterResult(
+ transactionTable, List.of(commitData)).dataFiles()[0]);
+ Assertions.assertEquals(rowIdBound, dataFile.lowerBounds().get(rowId));
+ Assertions.assertEquals(rowIdBound, dataFile.upperBounds().get(rowId));
+ Assertions.assertEquals(sequenceNumberBound,
dataFile.lowerBounds().get(sequenceNumberId));
+ Assertions.assertEquals(sequenceNumberBound,
dataFile.upperBounds().get(sequenceNumberId));
+ }
+
+ @Test
+ public void
testConvertToWriterResultSuppressesLogicalMetricsBelowRepeatedFields() {
+ Schema repeatedSchema = new Schema(
+ Types.NestedField.optional(1, "items",
+ Types.ListType.ofOptional(2, Types.IntegerType.get())),
+ Types.NestedField.optional(3, "attributes",
+ Types.MapType.ofOptional(4, 5, Types.StringType.get(),
Types.StringType.get())),
+ Types.NestedField.optional(6, "top_level",
Types.IntegerType.get()));
+ Table table = Mockito.mock(Table.class);
+ Mockito.when(table.schema()).thenReturn(repeatedSchema);
+ Mockito.when(table.spec()).thenReturn(unpartitionedSpec);
+ Mockito.when(table.sortOrder()).thenReturn(SortOrder.unsorted());
+ Mockito.when(table.properties()).thenReturn(Map.of(
+ TableProperties.DEFAULT_FILE_FORMAT, "parquet",
+ TableProperties.DEFAULT_WRITE_METRICS_MODE, "full"));
+
+ TIcebergColumnStats columnStats = new TIcebergColumnStats();
+ columnStats.setColumnSizes(Map.of(2, 20L, 4, 40L, 5, 50L, 6, 60L));
+ columnStats.setValueCounts(Map.of(2, 2L, 4, 4L, 5, 5L, 6, 6L));
+ columnStats.setNullValueCounts(Map.of(2, 0L, 4, 0L, 5, 0L, 6, 0L));
+ columnStats.setLowerBounds(Map.of(
+ 2, Conversions.toByteBuffer(Types.IntegerType.get(), 2),
+ 4, Conversions.toByteBuffer(Types.StringType.get(), "key"),
+ 5, Conversions.toByteBuffer(Types.StringType.get(), "value"),
+ 6, Conversions.toByteBuffer(Types.IntegerType.get(), 6)));
+ columnStats.setUpperBounds(columnStats.getLowerBounds());
+
+ TIcebergCommitData commitData = new TIcebergCommitData();
+ commitData.setFilePath("/path/to/repeated.parquet");
+ commitData.setRowCount(6);
+ commitData.setFileSize(1024);
+ commitData.setColumnStats(columnStats);
+
+ DataFile dataFile = IcebergWriterHelper.convertToWriterResult(table,
List.of(commitData)).dataFiles()[0];
+
+ Assertions.assertEquals(Map.of(2, 20L, 4, 40L, 5, 50L, 6, 60L),
dataFile.columnSizes());
+ Assertions.assertEquals(Map.of(6, 6L), dataFile.valueCounts());
+ Assertions.assertEquals(Map.of(6, 0L), dataFile.nullValueCounts());
+ Assertions.assertEquals(Map.of(6,
columnStats.getLowerBounds().get(6)), dataFile.lowerBounds());
+ Assertions.assertEquals(Map.of(6,
columnStats.getUpperBounds().get(6)), dataFile.upperBounds());
+ }
+
@Test
public void testConvertToDeleteFiles_EmptyList() {
List<TIcebergCommitData> commitDataList = new ArrayList<>();
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/planner/IcebergMergeSinkTest.java
b/fe/fe-core/src/test/java/org/apache/doris/planner/IcebergMergeSinkTest.java
index 23dcb4403ce..df9b2ae80c2 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/planner/IcebergMergeSinkTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/planner/IcebergMergeSinkTest.java
@@ -74,6 +74,63 @@ public class IcebergMergeSinkTest {
IcebergUtils.ICEBERG_LAST_UPDATED_SEQUENCE_NUMBER_COL));
}
+ @Test
+ public void testBindDataSinkDisablesColumnStatsWhenAllMetricsAreNone()
throws Exception {
+ IcebergMergeSink sink = new
IcebergMergeSink(mockIcebergExternalTable(2, Map.of(
+ TableProperties.DEFAULT_WRITE_METRICS_MODE, "none")), new
DeleteCommandContext());
+
+ sink.bindDataSink(Optional.empty());
+
+ TIcebergMergeSink thriftSink = sink.tDataSink.getIcebergMergeSink();
+ Assertions.assertTrue(thriftSink.isSetCollectColumnStats());
+ Assertions.assertFalse(thriftSink.isCollectColumnStats());
+ }
+
+ @Test
+ public void testBindDataSinkKeepsColumnStatsForMetricsOverride() throws
Exception {
+ IcebergMergeSink sink = new
IcebergMergeSink(mockIcebergExternalTable(2, Map.of(
+ TableProperties.DEFAULT_WRITE_METRICS_MODE, "none",
+ TableProperties.METRICS_MODE_COLUMN_CONF_PREFIX + "id",
"counts")),
+ new DeleteCommandContext());
+
+ sink.bindDataSink(Optional.empty());
+
+ TIcebergMergeSink thriftSink = sink.tDataSink.getIcebergMergeSink();
+ Assertions.assertTrue(thriftSink.isSetCollectColumnStats());
+ Assertions.assertTrue(thriftSink.isCollectColumnStats());
+ }
+
+ @Test
+ public void testBindDataSinkKeepsColumnStatsForV3LineageFields() throws
Exception {
+ IcebergMergeSink sink = new
IcebergMergeSink(mockIcebergExternalTable(3, Map.of(
+ TableProperties.DEFAULT_WRITE_METRICS_MODE, "counts",
+ TableProperties.METRICS_MODE_COLUMN_CONF_PREFIX + "id",
"none")),
+ new DeleteCommandContext());
+
+ sink.bindDataSink(Optional.empty());
+
+ TIcebergMergeSink thriftSink = sink.tDataSink.getIcebergMergeSink();
+ Assertions.assertTrue(thriftSink.isSetCollectColumnStats());
+ Assertions.assertTrue(thriftSink.isCollectColumnStats());
+ }
+
+ @Test
+ public void testBindDataSinkKeepsColumnStatsForOrcTopLevelComplexField()
throws Exception {
+ Schema schema = new Schema(Types.NestedField.optional(1, "items",
+ Types.ListType.ofOptional(2, Types.IntegerType.get())));
+ IcebergMergeSink sink = new
IcebergMergeSink(mockIcebergExternalTable(2, schema, Map.of(
+ TableProperties.DEFAULT_FILE_FORMAT, "orc",
+ TableProperties.DEFAULT_WRITE_METRICS_MODE, "none",
+ TableProperties.METRICS_MODE_COLUMN_CONF_PREFIX + "items",
"counts")),
+ new DeleteCommandContext());
+
+ sink.bindDataSink(Optional.empty());
+
+ TIcebergMergeSink thriftSink = sink.tDataSink.getIcebergMergeSink();
+ Assertions.assertTrue(thriftSink.isSetCollectColumnStats());
+ Assertions.assertTrue(thriftSink.isCollectColumnStats());
+ }
+
private static TIcebergRewritableDeleteFileSet buildDeleteFileSet() {
TIcebergDeleteFileDesc deleteFileDesc = new TIcebergDeleteFileDesc();
deleteFileDesc.setPath("file:///tmp/delete.puffin");
@@ -84,13 +141,24 @@ public class IcebergMergeSinkTest {
}
private static IcebergExternalTable mockIcebergExternalTable(int
formatVersion) {
- Schema schema = new Schema(Types.NestedField.required(1, "id",
Types.IntegerType.get()));
+ return mockIcebergExternalTable(formatVersion, Collections.emptyMap());
+ }
+
+ private static IcebergExternalTable mockIcebergExternalTable(
+ int formatVersion, Map<String, String> metricsProperties) {
+ return mockIcebergExternalTable(formatVersion,
+ new Schema(Types.NestedField.required(1, "id",
Types.IntegerType.get())), metricsProperties);
+ }
+
+ private static IcebergExternalTable mockIcebergExternalTable(
+ int formatVersion, Schema schema, Map<String, String>
metricsProperties) {
PartitionSpec spec = PartitionSpec.unpartitioned();
Map<String, String> properties = new HashMap<>();
properties.put(TableProperties.FORMAT_VERSION,
String.valueOf(formatVersion));
properties.put(TableProperties.DEFAULT_FILE_FORMAT, "parquet");
properties.put(TableProperties.PARQUET_COMPRESSION, "snappy");
properties.put(TableProperties.WRITE_DATA_LOCATION,
"file:///tmp/iceberg_tbl/data");
+ properties.putAll(metricsProperties);
Table icebergTable = Mockito.mock(Table.class);
Mockito.when(icebergTable.properties()).thenReturn(properties);
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/statistics/util/StatisticsUtilTest.java
b/fe/fe-core/src/test/java/org/apache/doris/statistics/util/StatisticsUtilTest.java
index 8bccbd2929f..af9bd9d2ac3 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/statistics/util/StatisticsUtilTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/statistics/util/StatisticsUtilTest.java
@@ -44,21 +44,34 @@ import
org.apache.doris.datasource.iceberg.IcebergExternalCatalog;
import org.apache.doris.datasource.iceberg.IcebergExternalDatabase;
import org.apache.doris.datasource.iceberg.IcebergExternalTable;
import org.apache.doris.datasource.iceberg.IcebergHadoopExternalCatalog;
+import org.apache.doris.datasource.iceberg.helper.IcebergWriterHelper;
import org.apache.doris.nereids.trees.expressions.literal.Literal;
import org.apache.doris.qe.SessionVariable;
import org.apache.doris.rpc.RpcException;
import org.apache.doris.statistics.AnalysisManager;
import org.apache.doris.statistics.ColStatsMeta;
import org.apache.doris.statistics.TableStatsMeta;
+import org.apache.doris.thrift.TIcebergColumnStats;
+import org.apache.doris.thrift.TIcebergCommitData;
import org.apache.doris.thrift.TStorageType;
import com.google.common.collect.Maps;
import org.apache.iceberg.CatalogProperties;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.FileScanTask;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.SortOrder;
+import org.apache.iceberg.TableProperties;
+import org.apache.iceberg.TableScan;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.types.Types;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
import org.mockito.MockedStatic;
import org.mockito.Mockito;
+import java.io.IOException;
import java.time.LocalTime;
import java.time.format.DateTimeFormatter;
import java.util.ArrayList;
@@ -67,6 +80,63 @@ import java.util.List;
import java.util.Map;
class StatisticsUtilTest {
+ @Test
+ void testGetIcebergColumnStatsReturnsEmptyForDisabledMetrics() {
+ Schema schema = new Schema(Types.NestedField.optional(1, "id",
Types.IntegerType.get()));
+ PartitionSpec spec = PartitionSpec.builderFor(schema).build();
+ org.apache.iceberg.Table table =
Mockito.mock(org.apache.iceberg.Table.class);
+ Mockito.when(table.schema()).thenReturn(schema);
+ Mockito.when(table.spec()).thenReturn(spec);
+ Mockito.when(table.sortOrder()).thenReturn(SortOrder.unsorted());
+ Mockito.when(table.properties()).thenReturn(Map.of(
+ TableProperties.DEFAULT_FILE_FORMAT, "parquet",
+ TableProperties.DEFAULT_WRITE_METRICS_MODE, "none"));
+
+ TIcebergColumnStats columnStats = new TIcebergColumnStats();
+ columnStats.setColumnSizes(Map.of(1, 128L));
+ columnStats.setValueCounts(Map.of(1, 10L));
+ columnStats.setNullValueCounts(Map.of(1, 0L));
+ TIcebergCommitData commitData = new TIcebergCommitData();
+ commitData.setFilePath("/path/to/data.parquet");
+ commitData.setRowCount(10);
+ commitData.setFileSize(1024);
+ commitData.setColumnStats(columnStats);
+ DataFile dataFile = IcebergWriterHelper.convertToWriterResult(table,
List.of(commitData)).dataFiles()[0];
+
+ TableScan tableScan = Mockito.mock(TableScan.class);
+ FileScanTask fileScanTask = Mockito.mock(FileScanTask.class);
+ Mockito.when(table.newScan()).thenReturn(tableScan);
+ Mockito.when(tableScan.includeColumnStats()).thenReturn(tableScan);
+ Mockito.when(tableScan.planFiles())
+
.thenReturn(CloseableIterable.withNoopClose(List.of(fileScanTask)));
+ Mockito.when(fileScanTask.spec()).thenReturn(spec);
+ Mockito.when(fileScanTask.file()).thenReturn(dataFile);
+
+ Assertions.assertTrue(StatisticsUtil.getIcebergColumnStats("id",
table).isEmpty());
+ }
+
+ @Test
+ void testGetIcebergColumnStatsReturnsEmptyWhenCloseFails() {
+ Schema schema = new Schema(Types.NestedField.optional(1, "id",
Types.IntegerType.get()));
+ PartitionSpec spec = PartitionSpec.builderFor(schema).build();
+ org.apache.iceberg.Table table =
Mockito.mock(org.apache.iceberg.Table.class);
+ TableScan tableScan = Mockito.mock(TableScan.class);
+ FileScanTask fileScanTask = Mockito.mock(FileScanTask.class);
+ DataFile dataFile = Mockito.mock(DataFile.class);
+ Mockito.when(table.newScan()).thenReturn(tableScan);
+ Mockito.when(tableScan.includeColumnStats()).thenReturn(tableScan);
+
Mockito.when(tableScan.planFiles()).thenReturn(CloseableIterable.combine(
+ List.of(fileScanTask), () -> {
+ throw new IOException("close failed");
+ }));
+ Mockito.when(fileScanTask.spec()).thenReturn(spec);
+ Mockito.when(fileScanTask.file()).thenReturn(dataFile);
+ Mockito.when(dataFile.columnSizes()).thenReturn(Map.of());
+ Mockito.when(dataFile.nullValueCounts()).thenReturn(Map.of());
+
+ Assertions.assertTrue(StatisticsUtil.getIcebergColumnStats("id",
table).isEmpty());
+ }
+
@Test
void testConvertToDouble() {
try {
diff --git a/gensrc/thrift/DataSinks.thrift b/gensrc/thrift/DataSinks.thrift
index 3a45208ea9b..18e152f447d 100644
--- a/gensrc/thrift/DataSinks.thrift
+++ b/gensrc/thrift/DataSinks.thrift
@@ -487,6 +487,8 @@ struct TIcebergTableSink {
15: optional map<string, string> static_partition_values;
16: optional PlanNodes.TSortInfo sort_info;
17: optional TIcebergWriteType write_type = TIcebergWriteType.INSERT;
+ // Unset keeps collection enabled for rolling upgrades with older FEs.
+ 18: optional bool collect_column_stats;
}
struct TIcebergRewritableDeleteFileSet {
@@ -533,6 +535,8 @@ struct TIcebergMergeSink {
11: optional map<string, string> hadoop_config
12: optional Types.TFileType file_type
13: optional list<Types.TNetworkAddress> broker_addresses;
+ // Unset keeps collection enabled for rolling upgrades with older FEs.
+ 14: optional bool collect_column_stats;
// delete side (position delete only)
20: optional TFileContent delete_type
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
