This is an automated email from the ASF dual-hosted git repository. suxiaogang223 pushed a commit to branch codex/fix-4.0-per-range-partition-values in repository https://gitbox.apache.org/repos/asf/doris.git
commit dbfaa8455d25b3742e6e158e6c7964df0014cfae Author: suxiaogang <[email protected]> AuthorDate: Fri Aug 7 14:30:06 2026 +0800 [fix](multi-catalog) Handle evolved partition metadata per range --- be/src/vec/exec/scan/file_scanner.cpp | 132 +++++++++++++-------- be/src/vec/exec/scan/file_scanner.h | 13 +- .../doris/datasource/hudi/source/HudiScanNode.java | 10 +- .../doris/datasource/iceberg/IcebergUtils.java | 36 +----- .../datasource/iceberg/source/IcebergScanNode.java | 50 ++++---- .../ExternalFileTableValuedFunction.java | 6 +- .../datasource/hudi/source/HudiScanNodeTest.java | 46 +++++++ .../doris/datasource/iceberg/IcebergUtilsTest.java | 75 ++++++------ .../iceberg/source/IcebergScanNodeTest.java | 45 +++---- .../ExternalFileTableValuedFunctionTest.java | 8 ++ 10 files changed, 239 insertions(+), 182 deletions(-) diff --git a/be/src/vec/exec/scan/file_scanner.cpp b/be/src/vec/exec/scan/file_scanner.cpp index 1a770a002b9..dd62c761fb5 100644 --- a/be/src/vec/exec/scan/file_scanner.cpp +++ b/be/src/vec/exec/scan/file_scanner.cpp @@ -210,8 +210,7 @@ Status FileScanner::init(RuntimeState* state, const VExprContextSPtrs& conjuncts bool FileScanner::_check_partition_prune_expr(const VExprSPtr& expr) { if (expr->is_slot_ref()) { auto* slot_ref = static_cast<VSlotRef*>(expr.get()); - return _partition_slot_index_map.find(slot_ref->slot_id()) != - _partition_slot_index_map.end(); + return _partition_slot_ids.contains(slot_ref->slot_id()); } if (expr->is_literal()) { return true; @@ -361,8 +360,7 @@ Status FileScanner::_open_impl(RuntimeState* state) { if (_first_scan_range) { RETURN_IF_ERROR(_init_expr_ctxes()); if (_state->query_options().enable_runtime_filter_partition_prune && - !_partition_slot_index_map.empty()) { - _init_runtime_filter_partition_prune_ctxs(); + (!_is_load || !_partition_slot_index_map.empty())) { _init_runtime_filter_partition_prune_block(); } } else { @@ -597,11 +595,11 @@ Status FileScanner::_cast_to_input_block(Block* block) { } Status FileScanner::_fill_columns_from_path(size_t rows) { - if (!_fill_partition_from_path) { + if (_partition_col_descs_to_fill.empty()) { return Status::OK(); } DataTypeSerDe::FormatOptions _text_formatOptions; - for (auto& kv : _partition_col_descs) { + for (auto& kv : _partition_col_descs_to_fill) { auto doris_column = _src_block_ptr->get_by_position(_src_block_name_to_idx[kv.first]).column; // _src_block_ptr points to a mutable block created by this class itself, so const_cast can be used here. @@ -895,26 +893,20 @@ Status FileScanner::_get_next_reader() { const TFileRangeDesc& range = _current_range; _current_range_path = range.path; - if (!_partition_slot_descs.empty()) { - // we need get partition columns first for runtime filter partition pruning - RETURN_IF_ERROR(_generate_partition_columns()); + // Partition keys may vary between ranges when a table's partition spec evolves. + RETURN_IF_ERROR(_generate_partition_columns()); - if (_state->query_options().enable_runtime_filter_partition_prune) { - // if enable_runtime_filter_partition_prune is true, we need to check whether this range can be filtered out - // by runtime filter partition prune - if (_push_down_conjuncts.size() < _conjuncts.size()) { - // there are new runtime filters, need to re-init runtime filter partition pruning ctxs - _init_runtime_filter_partition_prune_ctxs(); - } + if (_state->query_options().enable_runtime_filter_partition_prune && + !_partition_slot_ids.empty()) { + // Rebuild the contexts because only columns provided by this range can be used + // for partition pruning. + _init_runtime_filter_partition_prune_ctxs(); - bool can_filter_all = false; - RETURN_IF_ERROR(_process_runtime_filters_partition_prune(can_filter_all)); - if (can_filter_all) { - // this range can be filtered out by runtime filter partition pruning - // so we need to skip this range - COUNTER_UPDATE(_runtime_filter_partition_pruned_range_counter, 1); - continue; - } + bool can_filter_all = false; + RETURN_IF_ERROR(_process_runtime_filters_partition_prune(can_filter_all)); + if (can_filter_all) { + COUNTER_UPDATE(_runtime_filter_partition_pruned_range_counter, 1); + continue; } } @@ -1364,6 +1356,7 @@ Status FileScanner::_init_orc_reader(std::unique_ptr<OrcReader>&& orc_reader, Status FileScanner::_set_fill_or_truncate_columns(bool need_to_get_parsed_schema) { _missing_cols.clear(); _slot_lower_name_to_col_type.clear(); + _partition_col_descs_to_fill.clear(); std::unordered_map<std::string, DataTypePtr> name_to_col_type; RETURN_IF_ERROR(_cur_reader->get_columns(&name_to_col_type, &_missing_cols)); for (const auto& [col_name, col_type] : name_to_col_type) { @@ -1384,23 +1377,28 @@ Status FileScanner::_set_fill_or_truncate_columns(bool need_to_get_parsed_schema _slot_lower_name_to_col_type.emplace(col_name_lower, col_type); } - if (!_fill_partition_from_path && config::enable_iceberg_partition_column_fallback) { - // check if the cols of _partition_col_descs are in _missing_cols - // if so, set _fill_partition_from_path to true and remove the col from _missing_cols - for (const auto& [col_name, col_type] : _partition_col_descs) { - if (_missing_cols.contains(col_name)) { - _fill_partition_from_path = true; + if (_is_load) { + if (_load_fill_partition_from_path) { + _partition_col_descs_to_fill = _partition_col_descs; + } + } else { + for (const auto& [col_name, partition_col_desc] : _partition_col_descs) { + const auto* slot_desc = std::get<1>(partition_col_desc); + if (!_is_file_slot.contains(slot_desc->id())) { + _partition_col_descs_to_fill.emplace(col_name, partition_col_desc); + } else if (config::enable_iceberg_partition_column_fallback && + _missing_cols.contains(col_name)) { + _partition_col_descs_to_fill.emplace(col_name, partition_col_desc); _missing_cols.erase(col_name); } } } RETURN_IF_ERROR(_generate_missing_columns()); - if (_fill_partition_from_path) { - RETURN_IF_ERROR(_cur_reader->set_fill_columns(_partition_col_descs, _missing_col_descs, - _partition_value_is_null)); + if (!_partition_col_descs_to_fill.empty()) { + RETURN_IF_ERROR(_cur_reader->set_fill_columns( + _partition_col_descs_to_fill, _missing_col_descs, _partition_value_is_null)); } else { - // If the partition columns are not from path, we only fill the missing columns. RETURN_IF_ERROR(_cur_reader->set_fill_columns({}, _missing_col_descs)); } if (VLOG_NOTICE_IS_ON && !_missing_cols.empty() && _is_load) { @@ -1534,11 +1532,19 @@ Status FileScanner::read_lines_from_range(const TFileRangeDesc& range, Status FileScanner::_generate_partition_columns() { _partition_col_descs.clear(); _partition_value_is_null.clear(); + _partition_slot_ids.clear(); const TFileRangeDesc& range = _current_range; - if (range.__isset.columns_from_path && !_partition_slot_descs.empty()) { - if (range.__isset.columns_from_path_is_null) { - DORIS_CHECK(range.columns_from_path_is_null.size() == range.columns_from_path.size()); - } + if (!range.__isset.columns_from_path) { + return Status::OK(); + } + if (range.__isset.columns_from_path_is_null && + range.columns_from_path_is_null.size() != range.columns_from_path.size()) { + return Status::InternalError("Partition null marker count {} does not match value count {}", + range.columns_from_path_is_null.size(), + range.columns_from_path.size()); + } + + if (_is_load) { for (const auto& slot_desc : _partition_slot_descs) { if (slot_desc) { auto it = _partition_slot_index_map.find(slot_desc->id()); @@ -1555,12 +1561,48 @@ Status FileScanner::_generate_partition_columns() { const std::string& column_from_path = range.columns_from_path[it->second]; _partition_col_descs.emplace(slot_desc->col_name(), std::make_tuple(column_from_path, slot_desc)); + _partition_slot_ids.emplace(slot_desc->id()); if (range.__isset.columns_from_path_is_null) { _partition_value_is_null.emplace(slot_desc->col_name(), range.columns_from_path_is_null[it->second]); } } } + return Status::OK(); + } + + if (!range.__isset.columns_from_path_keys) { + return Status::OK(); + } + if (range.columns_from_path_keys.size() != range.columns_from_path.size()) { + return Status::InternalError("Partition key count {} does not match value count {}", + range.columns_from_path_keys.size(), + range.columns_from_path.size()); + } + + std::unordered_map<std::string, size_t> partition_name_to_index; + for (size_t i = 0; i < range.columns_from_path_keys.size(); ++i) { + partition_name_to_index.emplace(range.columns_from_path_keys[i], i); + } + for (const auto& slot_info : _params->required_slots) { + auto* slot_desc = _state->desc_tbl().get_slot_descriptor(slot_info.slot_id); + if (slot_desc == nullptr) { + return Status::InternalError("Unknown source slot descriptor, slot_id={}", + slot_info.slot_id); + } + auto index_it = partition_name_to_index.find(slot_desc->col_name()); + if (index_it == partition_name_to_index.end()) { + continue; + } + size_t value_index = index_it->second; + _partition_col_descs.emplace( + slot_desc->col_name(), + std::make_tuple(range.columns_from_path[value_index], slot_desc)); + _partition_slot_ids.emplace(slot_desc->id()); + if (range.__isset.columns_from_path_is_null) { + _partition_value_is_null.emplace(slot_desc->col_name(), + range.columns_from_path_is_null[value_index]); + } } return Status::OK(); } @@ -1594,13 +1636,8 @@ Status FileScanner::_init_expr_ctxes() { full_src_index_map.emplace(slot_desc->id(), index++); } - // For external table query, find the index of column in path. - // Because query doesn't always search for all columns in a table - // and the order of selected columns is random. - // All ranges in _ranges vector should have identical columns_from_path_keys - // because they are all file splits for the same external table. - // So here use the first element of _ranges to fill the partition_name_to_key_index_map - if (_current_range.__isset.columns_from_path_keys) { + // Load tasks do not always read all source columns, and the selected column order may vary. + if (_is_load && _current_range.__isset.columns_from_path_keys) { std::vector<std::string> key_map = _current_range.columns_from_path_keys; if (!key_map.empty()) { for (size_t i = 0; i < key_map.size(); i++) { @@ -1633,9 +1670,8 @@ Status FileScanner::_init_expr_ctxes() { if (slot_info.is_file_slot) { // If there is slot which is both a partition column and a file column, // we should not fill the partition column from path. - _fill_partition_from_path = false; - } else if (!_fill_partition_from_path) { - // This should not happen + _load_fill_partition_from_path = false; + } else if (!_load_fill_partition_from_path) { return Status::InternalError( "Partition column {} is not a file column, but there is already a column " "which is both a partition column and a file column.", diff --git a/be/src/vec/exec/scan/file_scanner.h b/be/src/vec/exec/scan/file_scanner.h index 7bce9f5f200..d5938345ef7 100644 --- a/be/src/vec/exec/scan/file_scanner.h +++ b/be/src/vec/exec/scan/file_scanner.h @@ -129,10 +129,12 @@ protected: // col names from _file_slot_descs std::vector<std::string> _file_col_names; - // Partition source slot descriptors + // Partition source slot descriptors used by load tasks. std::vector<SlotDescriptor*> _partition_slot_descs; - // Partition slot id to index in _partition_slot_descs + // Partition slot id to value index used by load tasks. std::unordered_map<SlotId, int> _partition_slot_index_map; + // Partition slot ids provided by the current query range. + std::unordered_set<SlotId> _partition_slot_ids; // created from param.expr_of_dest_slot // For query, it saves default value expr of all dest columns, or nullptr for NULL. // For load, it saves conversion expr/default value of all dest columns. @@ -187,10 +189,13 @@ protected: std::unique_ptr<io::FileReaderStats> _file_reader_stats; std::unique_ptr<io::IOContext> _io_ctx; - // Whether to fill partition columns from path, default is true. - bool _fill_partition_from_path = true; + // Whether load tasks should fill partition columns from the path. + bool _load_fill_partition_from_path = true; std::unordered_map<std::string, std::tuple<std::string, const SlotDescriptor*>> _partition_col_descs; + // Partition columns that should be filled for the current reader. + std::unordered_map<std::string, std::tuple<std::string, const SlotDescriptor*>> + _partition_col_descs_to_fill; std::unordered_map<std::string, bool> _partition_value_is_null; std::unordered_map<std::string, VExprContextSPtr> _missing_col_descs; diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/hudi/source/HudiScanNode.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/hudi/source/HudiScanNode.java index fd6979aedc1..b3b6eec5ad2 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/datasource/hudi/source/HudiScanNode.java +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/hudi/source/HudiScanNode.java @@ -24,6 +24,7 @@ import org.apache.doris.catalog.Column; import org.apache.doris.catalog.Env; import org.apache.doris.catalog.ListPartitionItem; import org.apache.doris.catalog.PartitionItem; +import org.apache.doris.catalog.PartitionKey; import org.apache.doris.catalog.Type; import org.apache.doris.common.AnalysisException; import org.apache.doris.common.UserException; @@ -355,7 +356,7 @@ public class HudiScanNode extends HiveScanNode { String path = basePath + "/" + key; hivePartitions.add(new HivePartition( nameMapping, false, inputFormat, path, - ((ListPartitionItem) value).getItems().get(0).getPartitionValuesAsStringList(), + getPartitionValues((ListPartitionItem) value), Maps.newHashMap())); } ); @@ -372,6 +373,13 @@ public class HudiScanNode extends HiveScanNode { return Lists.newArrayList(dummyPartition); } + static List<String> getPartitionValues(ListPartitionItem partitionItem) { + PartitionKey partitionKey = partitionItem.getItems().get(0); + return partitionKey.getKeys().stream() + .map(key -> key.isNullLiteral() ? null : key.getStringValue()) + .collect(Collectors.toList()); + } + private List<Split> getIncrementalSplits() { if (canUseNativeReader()) { List<Split> splits = incrementalRelation.collectSplits(); 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 2984393d0e4..5d41e400c0e 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 @@ -131,10 +131,8 @@ import java.time.temporal.ChronoField; import java.time.temporal.TemporalAccessor; import java.util.ArrayList; import java.util.Comparator; -import java.util.LinkedHashSet; import java.util.List; import java.util.Map; -import java.util.Objects; import java.util.Optional; import java.util.Set; import java.util.UUID; @@ -638,35 +636,8 @@ public class IcebergUtils { } } - /** - * Get identity partition columns that exist in all partition specs. - * The file scanner uses partition columns in the first scan range for all ranges, - * so only common identity partition columns can be used for partition pruning. - */ - public static List<String> getCommonIdentityPartitionColumns(Table table) { - LinkedHashSet<Integer> commonSourceIds = new LinkedHashSet<>(); - for (PartitionField field : table.spec().fields()) { - NestedField sourceField = table.schema().findField(field.sourceId()); - if (field.transform().isIdentity() && sourceField != null - && isSupportedPartitionValueType(sourceField.type().typeId())) { - commonSourceIds.add(field.sourceId()); - } - } - for (PartitionSpec spec : table.specs().values()) { - Set<Integer> specIdentitySourceIds = spec.fields().stream() - .filter(field -> field.transform().isIdentity()) - .map(PartitionField::sourceId) - .collect(Collectors.toSet()); - commonSourceIds.retainAll(specIdentitySourceIds); - } - return commonSourceIds.stream() - .map(table.schema()::findColumnName) - .filter(Objects::nonNull) - .collect(Collectors.toList()); - } - public static Map<String, String> getIdentityPartitionInfoMap(PartitionData partitionData, - PartitionSpec partitionSpec, Table table, String timeZone) { + PartitionSpec partitionSpec, Schema querySchema, String timeZone) { Map<String, String> partitionInfoMap = Maps.newLinkedHashMap(); List<NestedField> fields = partitionData.getPartitionType().asNestedType().fields(); List<PartitionField> partitionFields = partitionSpec.fields(); @@ -682,7 +653,7 @@ public class IcebergUtils { if (!isSupportedPartitionValueType(field.type().typeId())) { continue; } - String columnName = table.schema().findColumnName(partitionField.sourceId()); + String columnName = querySchema.findColumnName(partitionField.sourceId()); if (columnName == null) { continue; } @@ -751,7 +722,8 @@ public class IcebergUtils { long timestampMicros = (Long) value; TimestampType timestampType = (TimestampType) type; LocalDateTime timestamp = LocalDateTime.ofEpochSecond( - timestampMicros / 1_000_000, (int) (timestampMicros % 1_000_000) * 1000, + Math.floorDiv(timestampMicros, 1_000_000L), + (int) Math.floorMod(timestampMicros, 1_000_000L) * 1000, ZoneOffset.UTC); // type is timestamptz if timestampType.shouldAdjustToUTC() is true if (timestampType.shouldAdjustToUTC()) { diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/source/IcebergScanNode.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/source/IcebergScanNode.java index 74a535d4107..55dd03c35f5 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/source/IcebergScanNode.java +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/source/IcebergScanNode.java @@ -79,6 +79,7 @@ import org.apache.iceberg.MetadataColumns; import org.apache.iceberg.PartitionData; import org.apache.iceberg.PartitionSpec; import org.apache.iceberg.PartitionSpecParser; +import org.apache.iceberg.Schema; import org.apache.iceberg.SchemaParser; import org.apache.iceberg.Snapshot; import org.apache.iceberg.Table; @@ -121,6 +122,7 @@ public class IcebergScanNode extends FileQueryScanNode { private IcebergSource source; private Table icebergTable; + private Schema querySchema; private List<String> pushdownIcebergPredicates = Lists.newArrayList(); // If tableLevelPushDownCount is true, means we can do count push down opt at table level. // which means all splits have no position/equality delete files, @@ -135,7 +137,6 @@ public class IcebergScanNode extends FileQueryScanNode { // Used to avoid repeatedly calculating partition info map for the same // partition data and spec. private Map<Pair<Integer, PartitionData>, Map<String, String>> partitionMapInfos; - private boolean isPartitionedTable; private int formatVersion; private ExecutionAuthenticator preExecutionAuthenticator; private TableScan icebergTableScan; @@ -203,8 +204,12 @@ public class IcebergScanNode extends FileQueryScanNode { @Override protected void doInitialize() throws UserException { icebergTable = source.getIcebergTable(); + IcebergTableQueryInfo queryInfo = getSpecifiedSnapshot(); + querySchema = queryInfo == null ? icebergTable.schema() + : Preconditions.checkNotNull(icebergTable.schemas().get(queryInfo.getSchemaId()), + "Schema with schemaId %s not found for table %s", + queryInfo.getSchemaId(), icebergTable.name()); partitionMapInfos = new HashMap<>(); - isPartitionedTable = icebergTable.spec().isPartitioned(); formatVersion = ((BaseTable) icebergTable).operations().current().formatVersion(); preExecutionAuthenticator = source.getCatalog().getExecutionAuthenticator(); storagePropertiesMap = VendedCredentialsFactory.getStoragePropertiesMapWithVendedCredentials( @@ -305,36 +310,25 @@ public class IcebergScanNode extends FileQueryScanNode { rangeDesc.setTableFormatParams(tableFormatFileDesc); } - private List<String> getOrderedPathPartitionKeys() { - if (icebergTable == null) { - return Collections.emptyList(); - } - return IcebergUtils.getCommonIdentityPartitionColumns(icebergTable); - } - @VisibleForTesting void setPartitionValues(TFileRangeDesc rangeDesc, Map<String, String> partitionValues) { rangeDesc.unsetColumnsFromPathKeys(); rangeDesc.unsetColumnsFromPath(); rangeDesc.unsetColumnsFromPathIsNull(); - List<String> orderedPartitionKeys = getOrderedPathPartitionKeys(); - if (orderedPartitionKeys.isEmpty()) { + if (partitionValues == null || partitionValues.isEmpty()) { return; } - Preconditions.checkState(partitionValues != null, - "Missing partition values for Iceberg identity-partitioned table"); - - List<String> fromPathValues = new ArrayList<>(orderedPartitionKeys.size()); - List<Boolean> fromPathIsNull = new ArrayList<>(orderedPartitionKeys.size()); - for (String partitionKey : orderedPartitionKeys) { - Preconditions.checkState(partitionValues.containsKey(partitionKey), - "Missing partition value for Iceberg partition key: %s", partitionKey); - String partitionValue = partitionValues.get(partitionKey); + + List<String> fromPathKeys = new ArrayList<>(partitionValues.size()); + List<String> fromPathValues = new ArrayList<>(partitionValues.size()); + List<Boolean> fromPathIsNull = new ArrayList<>(partitionValues.size()); + partitionValues.forEach((partitionKey, partitionValue) -> { + fromPathKeys.add(partitionKey); fromPathValues.add(partitionValue == null ? "" : partitionValue); fromPathIsNull.add(partitionValue == null); - } - rangeDesc.setColumnsFromPathKeys(orderedPartitionKeys); + }); + rangeDesc.setColumnsFromPathKeys(fromPathKeys); rangeDesc.setColumnsFromPath(fromPathValues); rangeDesc.setColumnsFromPathIsNull(fromPathIsNull); } @@ -792,15 +786,15 @@ public class IcebergScanNode extends FileQueryScanNode { } split.setTableFormatType(TableFormatType.ICEBERG); split.setTargetSplitSize(targetSplitSize); - if (isPartitionedTable) { + int specId = fileScanTask.file().specId(); + PartitionSpec partitionSpec = icebergTable.specs().get(specId); + Preconditions.checkNotNull(partitionSpec, "Partition spec with specId %s not found for table %s", + specId, icebergTable.name()); + if (partitionSpec.isPartitioned()) { PartitionData partitionData = (PartitionData) fileScanTask.file().partition(); - int specId = fileScanTask.file().specId(); - PartitionSpec partitionSpec = icebergTable.specs().get(specId); - Preconditions.checkNotNull(partitionSpec, "Partition spec with specId %s not found for table %s", - specId, icebergTable.name()); Map<String, String> partitionInfoMap = partitionMapInfos.computeIfAbsent( Pair.of(specId, partitionData), k -> IcebergUtils.getIdentityPartitionInfoMap( - partitionData, partitionSpec, icebergTable, sessionVariable.getTimeZone())); + partitionData, partitionSpec, querySchema, sessionVariable.getTimeZone())); if (!partitionInfoMap.isEmpty()) { split.setIcebergPartitionValues(partitionInfoMap); } diff --git a/fe/fe-core/src/main/java/org/apache/doris/tablefunction/ExternalFileTableValuedFunction.java b/fe/fe-core/src/main/java/org/apache/doris/tablefunction/ExternalFileTableValuedFunction.java index e08eaa2c825..c71512896a1 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/tablefunction/ExternalFileTableValuedFunction.java +++ b/fe/fe-core/src/main/java/org/apache/doris/tablefunction/ExternalFileTableValuedFunction.java @@ -424,10 +424,14 @@ public abstract class ExternalFileTableValuedFunction extends TableValuedFunctio // HACK(tsy): path columns are all treated as STRING type now, after BE supports reading all columns // types by all format readers from file meta, maybe reading path columns types from BE then. for (String colName : pathPartitionKeys) { - columns.add(new Column(colName, ScalarType.createVarcharType(ScalarType.MAX_VARCHAR_LENGTH), false)); + columns.add(createPathPartitionColumn(colName)); } } + static Column createPathPartitionColumn(String colName) { + return new Column(colName, ScalarType.createVarcharType(ScalarType.MAX_VARCHAR_LENGTH), true); + } + private PFetchTableSchemaRequest getFetchTableStructureRequest() throws TException { // set TFileScanRangeParams TFileScanRangeParams fileScanRangeParams = new TFileScanRangeParams(); diff --git a/fe/fe-core/src/test/java/org/apache/doris/datasource/hudi/source/HudiScanNodeTest.java b/fe/fe-core/src/test/java/org/apache/doris/datasource/hudi/source/HudiScanNodeTest.java new file mode 100644 index 00000000000..6befa82f418 --- /dev/null +++ b/fe/fe-core/src/test/java/org/apache/doris/datasource/hudi/source/HudiScanNodeTest.java @@ -0,0 +1,46 @@ +// 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. + +package org.apache.doris.datasource.hudi.source; + +import org.apache.doris.analysis.PartitionValue; +import org.apache.doris.catalog.ListPartitionItem; +import org.apache.doris.catalog.PartitionKey; +import org.apache.doris.catalog.ScalarType; +import org.apache.doris.catalog.Type; +import org.apache.doris.common.AnalysisException; + +import com.google.common.collect.ImmutableList; +import org.junit.Assert; +import org.junit.Test; + +import java.util.Arrays; +import java.util.List; + +public class HudiScanNodeTest { + @Test + public void testGetPartitionValuesPreservesNullLiteral() throws AnalysisException { + List<PartitionValue> values = Arrays.asList( + new PartitionValue("__HIVE_DEFAULT_PARTITION__", true), + new PartitionValue("NULL")); + List<Type> types = Arrays.asList(ScalarType.STRING, ScalarType.STRING); + PartitionKey key = PartitionKey.createListPartitionKeyWithTypes(values, types, false); + ListPartitionItem item = new ListPartitionItem(ImmutableList.of(key)); + + Assert.assertEquals(Arrays.asList(null, "NULL"), HudiScanNode.getPartitionValues(item)); + } +} diff --git a/fe/fe-core/src/test/java/org/apache/doris/datasource/iceberg/IcebergUtilsTest.java b/fe/fe-core/src/test/java/org/apache/doris/datasource/iceberg/IcebergUtilsTest.java index 69dba111508..6e0f448aa90 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/datasource/iceberg/IcebergUtilsTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/datasource/iceberg/IcebergUtilsTest.java @@ -137,35 +137,6 @@ public class IcebergUtilsTest { Assert.assertEquals("PART", columns.get(1).getName()); } - @Test - public void testGetCommonIdentityPartitionColumnsUsesSafeIntersection() { - Schema schema = new Schema( - Types.NestedField.required(1, "id", Types.IntegerType.get()), - Types.NestedField.required(2, "Dt", Types.StringType.get()), - Types.NestedField.required(3, "ts", Types.TimestampType.withoutZone())); - PartitionSpec oldSpec = PartitionSpec.builderFor(schema) - .withSpecId(1) - .identity("id") - .identity("Dt") - .build(); - PartitionSpec currentSpec = PartitionSpec.builderFor(schema) - .withSpecId(2) - .identity("Dt") - .day("ts") - .build(); - Map<Integer, PartitionSpec> specs = new LinkedHashMap<>(); - specs.put(oldSpec.specId(), oldSpec); - specs.put(currentSpec.specId(), currentSpec); - - Table table = Mockito.mock(Table.class); - Mockito.when(table.schema()).thenReturn(schema); - Mockito.when(table.spec()).thenReturn(currentSpec); - Mockito.when(table.specs()).thenReturn(specs); - - Assert.assertEquals(Arrays.asList("Dt"), - IcebergUtils.getCommonIdentityPartitionColumns(table)); - } - @Test public void testGetIdentityPartitionInfoMapReturnsIdentityColumnsOnly() { Schema schema = new Schema( @@ -179,11 +150,8 @@ public class IcebergUtilsTest { partitionData.set(0, "2025-01-01"); partitionData.set(1, 20000); - Table table = Mockito.mock(Table.class); - Mockito.when(table.schema()).thenReturn(schema); - Map<String, String> partitionInfoMap = IcebergUtils.getIdentityPartitionInfoMap( - partitionData, partitionSpec, table, "UTC"); + partitionData, partitionSpec, schema, "UTC"); Assert.assertEquals(Collections.singletonMap("Dt", "2025-01-01"), partitionInfoMap); } @@ -202,11 +170,8 @@ public class IcebergUtilsTest { partitionData.set(0, floatValue); partitionData.set(1, doubleValue); - Table table = Mockito.mock(Table.class); - Mockito.when(table.schema()).thenReturn(schema); - Map<String, String> partitionInfoMap = IcebergUtils.getIdentityPartitionInfoMap( - partitionData, partitionSpec, table, "UTC"); + partitionData, partitionSpec, schema, "UTC"); String serializedFloat = partitionInfoMap.get("float_partition"); String serializedDouble = partitionInfoMap.get("double_partition"); @@ -218,6 +183,42 @@ public class IcebergUtilsTest { Double.doubleToLongBits(Double.parseDouble(serializedDouble))); } + @Test + public void testGetIdentityPartitionInfoMapUsesQuerySchemaName() { + Schema specSchema = new Schema( + Types.NestedField.required(1, "old_name", Types.StringType.get())); + PartitionSpec partitionSpec = PartitionSpec.builderFor(specSchema) + .identity("old_name") + .build(); + PartitionData partitionData = new PartitionData(partitionSpec.partitionType()); + partitionData.set(0, "value"); + + Map<String, String> partitionInfoMap = IcebergUtils.getIdentityPartitionInfoMap( + partitionData, partitionSpec, specSchema, "UTC"); + + Assert.assertEquals(Collections.singletonMap("old_name", "value"), partitionInfoMap); + } + + @Test + public void testGetIdentityPartitionInfoMapSupportsNegativeTimestampMicros() { + Schema schema = new Schema( + Types.NestedField.required(1, "local_ts", Types.TimestampType.withoutZone()), + Types.NestedField.required(2, "utc_ts", Types.TimestampType.withZone())); + PartitionSpec partitionSpec = PartitionSpec.builderFor(schema) + .identity("local_ts") + .identity("utc_ts") + .build(); + PartitionData partitionData = new PartitionData(partitionSpec.partitionType()); + partitionData.set(0, -1L); + partitionData.set(1, -1L); + + Map<String, String> partitionInfoMap = IcebergUtils.getIdentityPartitionInfoMap( + partitionData, partitionSpec, schema, "Asia/Shanghai"); + + Assert.assertEquals("1969-12-31T23:59:59.999999", partitionInfoMap.get("local_ts")); + Assert.assertEquals("1970-01-01T07:59:59.999999", partitionInfoMap.get("utc_ts")); + } + @Test public void testGetMatchingManifest() { diff --git a/fe/fe-core/src/test/java/org/apache/doris/datasource/iceberg/source/IcebergScanNodeTest.java b/fe/fe-core/src/test/java/org/apache/doris/datasource/iceberg/source/IcebergScanNodeTest.java index b855646d533..fd8e8643344 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/datasource/iceberg/source/IcebergScanNodeTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/datasource/iceberg/source/IcebergScanNodeTest.java @@ -30,21 +30,15 @@ import org.apache.doris.thrift.TFileRangeDesc; import org.apache.iceberg.DataFile; import org.apache.iceberg.FileFormat; import org.apache.iceberg.FileScanTask; -import org.apache.iceberg.PartitionSpec; -import org.apache.iceberg.Schema; -import org.apache.iceberg.Table; -import org.apache.iceberg.types.Types; import org.apache.iceberg.util.ScanTaskUtil; import org.junit.Assert; import org.junit.Test; import org.mockito.Mockito; -import java.lang.reflect.Field; import java.lang.reflect.Method; import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; -import java.util.HashMap; import java.util.LinkedHashMap; import java.util.Map; @@ -106,36 +100,25 @@ public class IcebergScanNodeTest { } @Test - public void testSetPartitionValuesBuildsStableAlignedMetadata() throws Exception { + public void testSetPartitionValuesBuildsPerRangeAlignedMetadata() throws Exception { TestIcebergScanNode node = new TestIcebergScanNode(new SessionVariable()); - Schema schema = new Schema( - Types.NestedField.required(1, "Region", Types.StringType.get()), - Types.NestedField.required(2, "Dt", Types.StringType.get())); - PartitionSpec spec = PartitionSpec.builderFor(schema) - .identity("Region") - .identity("Dt") - .build(); - Map<Integer, PartitionSpec> specs = new LinkedHashMap<>(); - specs.put(spec.specId(), spec); - Table table = Mockito.mock(Table.class); - Mockito.when(table.schema()).thenReturn(schema); - Mockito.when(table.spec()).thenReturn(spec); - Mockito.when(table.specs()).thenReturn(specs); - - Field icebergTable = IcebergScanNode.class.getDeclaredField("icebergTable"); - icebergTable.setAccessible(true); - icebergTable.set(node, table); - Assert.assertTrue(node.getPathPartitionKeys().isEmpty()); - Map<String, String> partitionValues = new HashMap<>(); + Map<String, String> partitionValues = new LinkedHashMap<>(); partitionValues.put("Dt", null); partitionValues.put("Region", "cn"); - TFileRangeDesc rangeDesc = new TFileRangeDesc(); - node.setPartitionValues(rangeDesc, partitionValues); + TFileRangeDesc oldSpecRange = new TFileRangeDesc(); + node.setPartitionValues(oldSpecRange, partitionValues); + + Assert.assertEquals(Arrays.asList("Dt", "Region"), oldSpecRange.getColumnsFromPathKeys()); + Assert.assertEquals(Arrays.asList("", "cn"), oldSpecRange.getColumnsFromPath()); + Assert.assertEquals(Arrays.asList(true, false), oldSpecRange.getColumnsFromPathIsNull()); + + TFileRangeDesc newSpecRange = new TFileRangeDesc(); + node.setPartitionValues(newSpecRange, Collections.singletonMap("Region", "us")); - Assert.assertEquals(Arrays.asList("Region", "Dt"), rangeDesc.getColumnsFromPathKeys()); - Assert.assertEquals(Arrays.asList("cn", ""), rangeDesc.getColumnsFromPath()); - Assert.assertEquals(Arrays.asList(false, true), rangeDesc.getColumnsFromPathIsNull()); + Assert.assertEquals(Collections.singletonList("Region"), newSpecRange.getColumnsFromPathKeys()); + Assert.assertEquals(Collections.singletonList("us"), newSpecRange.getColumnsFromPath()); + Assert.assertEquals(Collections.singletonList(false), newSpecRange.getColumnsFromPathIsNull()); } } diff --git a/fe/fe-core/src/test/java/org/apache/doris/tablefunction/ExternalFileTableValuedFunctionTest.java b/fe/fe-core/src/test/java/org/apache/doris/tablefunction/ExternalFileTableValuedFunctionTest.java index e5b06bd5dd4..e0ead50ed13 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/tablefunction/ExternalFileTableValuedFunctionTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/tablefunction/ExternalFileTableValuedFunctionTest.java @@ -33,6 +33,14 @@ import java.util.List; import java.util.Map; public class ExternalFileTableValuedFunctionTest { + @Test + public void testPathPartitionColumnIsNullable() { + Column column = ExternalFileTableValuedFunction.createPathPartitionColumn("part"); + + Assert.assertTrue(column.isAllowNull()); + Assert.assertEquals(PrimitiveType.VARCHAR, column.getType().getPrimitiveType()); + } + @Test public void testCsvSchemaParse() { Config.enable_date_conversion = true; --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
