This is an automated email from the ASF dual-hosted git repository. hello-stephen pushed a commit to branch ai/backport-doris-27023-4.1 in repository https://gitbox.apache.org/repos/asf/doris.git
commit 99d3a36a3568dd21e1f52243b9eed3c9ab7463fa Author: Gabriel <[email protected]> AuthorDate: Tue Jul 21 16:07:56 2026 +0800 [fix](iceberg) Honor disabled write metrics (#65782) - 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 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. - `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` - 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 434488266bb..25da8724b28 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) { @@ -155,6 +159,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 3e3949cd924..db2ca74c64a 100644 --- a/be/src/format/transformer/vorc_transformer.cpp +++ b/be/src/format/transformer/vorc_transformer.cpp @@ -386,12 +386,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 638b0474771..2eb2b0d3894 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 @@ -77,14 +77,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; @@ -1882,10 +1885,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 6a0b7c71ffd..c59992082ad 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 @@ -730,12 +730,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 2de5d19e7fb..16640db7ccd 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 @@ -39,6 +39,7 @@ 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.datasource.jdbc.JdbcExternalCatalog; import org.apache.doris.datasource.jdbc.JdbcExternalDatabase; import org.apache.doris.datasource.jdbc.JdbcExternalTable; @@ -49,6 +50,8 @@ import org.apache.doris.statistics.AnalysisManager; import org.apache.doris.statistics.ColStatsMeta; import org.apache.doris.statistics.ResultRow; 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.Lists; @@ -56,10 +59,20 @@ import com.google.common.collect.Maps; import mockit.Mock; import mockit.MockUp; 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.Mockito; +import java.io.IOException; import java.nio.charset.StandardCharsets; import java.time.LocalTime; import java.time.format.DateTimeFormatter; @@ -70,6 +83,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 3640117def8..a0cb46f1166 100644 --- a/gensrc/thrift/DataSinks.thrift +++ b/gensrc/thrift/DataSinks.thrift @@ -486,6 +486,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 { @@ -532,6 +534,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]
