diff --git a/be/src/vec/exec/scan/file_scanner.cpp b/be/src/vec/exec/scan/file_scanner.cpp index 01144f1e55d42e..efa1a666374dcd 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(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; @@ -247,12 +246,6 @@ Status FileScanner::_process_runtime_filters_partition_prune(bool& can_filter_al if (_runtime_filter_partition_prune_ctxs.empty() || _partition_col_descs.empty()) { return Status::OK(); } - if (_partition_col_descs.size() != _partition_slot_descs.size()) { - // This range only carries values for a subset of the partition slots (e.g. a file - // written under an older Iceberg partition spec). The prune conjuncts may reference - // unbound slots whose block columns would stay empty, so skip pruning this range. - return Status::OK(); - } size_t partition_value_column_size = 1; // 1. Get partition key values to string columns. @@ -367,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 { @@ -603,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. @@ -901,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; } } @@ -1370,6 +1356,7 @@ Status FileScanner::_init_orc_reader(std::unique_ptr&& 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 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) { @@ -1390,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) { @@ -1540,68 +1532,76 @@ 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) { return Status::OK(); } - 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_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 && range.__isset.columns_from_path_keys) { - // Ranges of one scanner may carry different columns_from_path_keys. E.g. after - // Iceberg partition evolution, a file written under an older partition spec only - // carries the identity partition values of that spec (possibly none), while - // _partition_slot_descs was built from the first range's keys. So bind values by - // this range's own key list. A partition slot absent from this range's keys is - // read from the data file instead (for such tables all partition columns are - // also file slots, see IcebergScanNode.getPathPartitionKeys). - if (range.columns_from_path.size() != range.columns_from_path_keys.size()) { - return Status::InternalError( - "columns_from_path size {} does not match columns_from_path_keys size {}", - range.columns_from_path.size(), range.columns_from_path_keys.size()); - } - std::unordered_map name_to_value_index; - for (size_t i = 0; i < range.columns_from_path_keys.size(); ++i) { - name_to_value_index.emplace(range.columns_from_path_keys[i], i); - } + + if (_is_load) { for (const auto& slot_desc : _partition_slot_descs) { - if (!slot_desc) { - continue; - } - auto it = name_to_value_index.find(slot_desc->col_name()); - if (it == name_to_value_index.end()) { - continue; - } - _partition_col_descs.emplace( - slot_desc->col_name(), - std::make_tuple(range.columns_from_path[it->second], slot_desc)); - 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]); + if (slot_desc) { + auto it = _partition_slot_index_map.find(slot_desc->id()); + if (it == std::end(_partition_slot_index_map)) { + return Status::InternalError("Unknown source slot descriptor, slot_id={}", + slot_desc->id()); + } + if (it->second < 0 || + static_cast(it->second) >= range.columns_from_path.size()) { + return Status::InternalError( + "Invalid partition value index {}, value count {} for column {}", + it->second, range.columns_from_path.size(), slot_desc->col_name()); + } + 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(); } - for (const auto& slot_desc : _partition_slot_descs) { - if (slot_desc) { - auto it = _partition_slot_index_map.find(slot_desc->id()); - if (it == std::end(_partition_slot_index_map)) { - return Status::InternalError("Unknown source slot descriptor, slot_id={}", - slot_desc->id()); - } - if (it->second < 0 || - static_cast(it->second) >= range.columns_from_path.size()) { - return Status::InternalError( - "Invalid partition value index {}, value count {} for column {}", - it->second, range.columns_from_path.size(), slot_desc->col_name()); - } - 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)); - 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]); - } + + 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 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(); @@ -1636,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 key_map = _current_range.columns_from_path_keys; if (!key_map.empty()) { for (size_t i = 0; i < key_map.size(); i++) { @@ -1675,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 4ca982e0bea137..b39ba1c8fefbc8 100644 --- a/be/src/vec/exec/scan/file_scanner.h +++ b/be/src/vec/exec/scan/file_scanner.h @@ -131,10 +131,12 @@ class FileScanner : public Scanner { // col names from _file_slot_descs std::vector _file_col_names; - // Partition source slot descriptors + // Partition source slot descriptors used by load tasks. std::vector _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 _partition_slot_index_map; + // Partition slot ids provided by the current query range. + std::unordered_set _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. @@ -189,10 +191,13 @@ class FileScanner : public Scanner { std::unique_ptr _file_reader_stats; std::unique_ptr _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> _partition_col_descs; + // Partition columns that should be filled for the current reader. + std::unordered_map> + _partition_col_descs_to_fill; std::unordered_map _partition_value_is_null; std::unordered_map _missing_col_descs; diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/FileGroupInfo.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/FileGroupInfo.java index d81ba7daa8419c..9efeda06dbc83e 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/datasource/FileGroupInfo.java +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/FileGroupInfo.java @@ -26,7 +26,6 @@ import org.apache.doris.common.Config; import org.apache.doris.common.Pair; import org.apache.doris.common.UserException; -import org.apache.doris.common.util.BrokerUtil; import org.apache.doris.common.util.Util; import org.apache.doris.load.BrokerFileGroup; import org.apache.doris.planner.FileLoadScanNode; @@ -266,11 +265,12 @@ public void createScanRangeLocationsUnsplittable(FileLoadScanNode.ParamCreateCon Util.getOrInferCompressType(context.fileGroup.getFileFormatProperties().getCompressionType(), fileStatus.path); context.params.setCompressType(compressType); - List columnsFromPath = BrokerUtil.parseColumnsFromPath(fileStatus.path, - context.fileGroup.getColumnNamesFromPath()); + FilePartitionUtils.ParsedColumnsFromPath columnsFromPath = + FilePartitionUtils.parseColumnsFromPathWithNullInfo(fileStatus.path, + context.fileGroup.getColumnNamesFromPath(), true, false); List columnsFromPathKeys = context.fileGroup.getColumnNamesFromPath(); - TFileRangeDesc rangeDesc = createFileRangeDesc(0, fileStatus, fileStatus.size, columnsFromPath, - columnsFromPathKeys); + TFileRangeDesc rangeDesc = createFileRangeDesc(0, fileStatus, fileStatus.size, + columnsFromPath.getValues(), columnsFromPathKeys, columnsFromPath.getIsNull()); locations.getScanRange().getExtScanRange().getFileScanRange().addToRanges(rangeDesc); } scanRangeLocations.add(locations); @@ -312,15 +312,16 @@ public void createScanRangeLocationsSplittable(FileLoadScanNode.ParamCreateConte Util.getOrInferCompressType(context.fileGroup.getFileFormatProperties().getCompressionType(), fileStatus.path); context.params.setCompressType(compressType); - List columnsFromPath = BrokerUtil.parseColumnsFromPath(fileStatus.path, - context.fileGroup.getColumnNamesFromPath()); + FilePartitionUtils.ParsedColumnsFromPath columnsFromPath = + FilePartitionUtils.parseColumnsFromPathWithNullInfo(fileStatus.path, + context.fileGroup.getColumnNamesFromPath(), true, false); List columnsFromPathKeys = context.fileGroup.getColumnNamesFromPath(); // Assign scan range locations only for broker load. // stream load has only one file, and no need to set multi scan ranges. if (tmpBytes > bytesPerInstance && jobType != JobType.STREAM_LOAD) { long rangeBytes = bytesPerInstance - curInstanceBytes; TFileRangeDesc rangeDesc = createFileRangeDesc(curFileOffset, fileStatus, rangeBytes, - columnsFromPath, columnsFromPathKeys); + columnsFromPath.getValues(), columnsFromPathKeys, columnsFromPath.getIsNull()); curLocations.getScanRange().getExtScanRange().getFileScanRange().addToRanges(rangeDesc); curFileOffset += rangeBytes; @@ -329,8 +330,8 @@ public void createScanRangeLocationsSplittable(FileLoadScanNode.ParamCreateConte curLocations = newLocations(context.params, brokerDesc, backendPolicy); curInstanceBytes = 0; } else { - TFileRangeDesc rangeDesc = createFileRangeDesc(curFileOffset, fileStatus, leftBytes, columnsFromPath, - columnsFromPathKeys); + TFileRangeDesc rangeDesc = createFileRangeDesc(curFileOffset, fileStatus, leftBytes, + columnsFromPath.getValues(), columnsFromPathKeys, columnsFromPath.getIsNull()); curLocations.getScanRange().getExtScanRange().getFileScanRange().addToRanges(rangeDesc); curFileOffset = 0; curInstanceBytes += leftBytes; @@ -401,7 +402,7 @@ private TFileFormatType formatType(String fileFormat, String path) throws UserEx } private TFileRangeDesc createFileRangeDesc(long curFileOffset, TBrokerFileStatus fileStatus, long rangeBytes, - List columnsFromPath, List columnsFromPathKeys) { + List columnsFromPath, List columnsFromPathKeys, List columnsFromPathIsNull) { TFileRangeDesc rangeDesc = new TFileRangeDesc(); if (jobType == JobType.BULK_LOAD) { rangeDesc.setPath(fileStatus.path); @@ -410,6 +411,7 @@ private TFileRangeDesc createFileRangeDesc(long curFileOffset, TBrokerFileStatus rangeDesc.setFileSize(fileStatus.size); rangeDesc.setColumnsFromPath(columnsFromPath); rangeDesc.setColumnsFromPathKeys(columnsFromPathKeys); + rangeDesc.setColumnsFromPathIsNull(columnsFromPathIsNull); if (getFileType() == TFileType.FILE_HDFS) { URI fileUri = new Path(fileStatus.path).toUri(); rangeDesc.setFsName(fileUri.getScheme() + "://" + fileUri.getAuthority()); diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/FilePartitionUtils.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/FilePartitionUtils.java new file mode 100644 index 00000000000000..0a50b73d8d899e --- /dev/null +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/FilePartitionUtils.java @@ -0,0 +1,148 @@ +// 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; + +import org.apache.doris.common.UserException; +import org.apache.doris.datasource.hive.HiveMetaStoreCache; + +import com.google.common.collect.Lists; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.Locale; + +/** + * Utility methods for parsing partition column values from Hive-style file paths. + */ +public final class FilePartitionUtils { + + private FilePartitionUtils() {} + + public static final class ParsedColumnsFromPath { + private final List values; + private final List isNull; + + private ParsedColumnsFromPath(List values, List isNull) { + this.values = values; + this.isNull = isNull; + } + + public List getValues() { + return values; + } + + public List getIsNull() { + return isNull; + } + } + + public static List parseColumnsFromPath(String filePath, List columnsFromPath) + throws UserException { + return parseColumnsFromPathWithNullInfo(filePath, columnsFromPath, true, false).getValues(); + } + + public static List parseColumnsFromPath( + String filePath, + List columnsFromPath, + boolean caseSensitive, + boolean isACID) + throws UserException { + return parseColumnsFromPathWithNullInfo(filePath, columnsFromPath, caseSensitive, isACID) + .getValues(); + } + + public static ParsedColumnsFromPath parseColumnsFromPathWithNullInfo( + String filePath, + List columnsFromPath, + boolean caseSensitive, + boolean isACID) + throws UserException { + if (columnsFromPath == null || columnsFromPath.isEmpty()) { + return new ParsedColumnsFromPath(Collections.emptyList(), Collections.emptyList()); + } + int pathCount = isACID ? 3 : 2; + List expectedColumns = columnsFromPath; + if (!caseSensitive) { + expectedColumns = new ArrayList<>(columnsFromPath.size()); + for (String path : columnsFromPath) { + expectedColumns.add(path.toLowerCase(Locale.ROOT)); + } + } + String[] strings = filePath.split("/"); + if (strings.length < 2) { + throw new UserException("Fail to parse columnsFromPath, expected: " + + expectedColumns + ", filePath: " + filePath); + } + String[] columns = new String[expectedColumns.size()]; + Boolean[] columnValueIsNull = new Boolean[expectedColumns.size()]; + int size = 0; + boolean skipOnce = true; + for (int i = strings.length - pathCount; i >= 0; i--) { + String str = strings[i]; + if (str != null && str.isEmpty()) { + continue; + } + if (str == null || !str.contains("=")) { + if (!isACID && skipOnce) { + skipOnce = false; + continue; + } + throw new UserException("Fail to parse columnsFromPath, expected: " + + expectedColumns + ", filePath: " + filePath); + } + skipOnce = false; + String[] pair = str.split("=", 2); + if (pair.length != 2) { + throw new UserException("Fail to parse columnsFromPath, expected: " + + expectedColumns + ", filePath: " + filePath); + } + String parsedColumnName = caseSensitive ? pair[0] : pair[0].toLowerCase(Locale.ROOT); + int index = expectedColumns.indexOf(parsedColumnName); + if (index == -1) { + continue; + } + boolean isNull = HiveMetaStoreCache.HIVE_DEFAULT_PARTITION.equals(pair[1]); + columns[index] = isNull ? "" : pair[1]; + columnValueIsNull[index] = isNull; + size++; + if (size >= expectedColumns.size()) { + break; + } + } + if (size != expectedColumns.size()) { + throw new UserException("Fail to parse columnsFromPath, expected: " + + expectedColumns + ", filePath: " + filePath); + } + return new ParsedColumnsFromPath(Lists.newArrayList(columns), Lists.newArrayList(columnValueIsNull)); + } + + public static ParsedColumnsFromPath normalizeColumnsFromPath(List columnsFromPath) { + if (columnsFromPath == null || columnsFromPath.isEmpty()) { + return new ParsedColumnsFromPath(Collections.emptyList(), Collections.emptyList()); + } + List values = new ArrayList<>(columnsFromPath.size()); + List isNull = new ArrayList<>(columnsFromPath.size()); + for (String value : columnsFromPath) { + boolean nullValue = value == null || HiveMetaStoreCache.HIVE_DEFAULT_PARTITION.equals(value); + values.add(nullValue ? "" : value); + isNull.add(nullValue); + } + return new ParsedColumnsFromPath(values, isNull); + } +} diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/FileQueryScanNode.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/FileQueryScanNode.java index 8fc003bd520562..29bfcf6a2b9e87 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/datasource/FileQueryScanNode.java +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/FileQueryScanNode.java @@ -31,7 +31,6 @@ import org.apache.doris.common.AnalysisException; import org.apache.doris.common.NotImplementedException; import org.apache.doris.common.UserException; -import org.apache.doris.common.util.BrokerUtil; import org.apache.doris.common.util.Util; import org.apache.doris.datasource.hive.source.HiveSplit; import org.apache.doris.planner.PlanNodeId; @@ -410,11 +409,14 @@ private TScanRangeLocations splitToScanRange( HiveSplit hiveSplit = (HiveSplit) fileSplit; isACID = hiveSplit.isACID(); } - List partitionValuesFromPath = fileSplit.getPartitionValues() == null - ? BrokerUtil.parseColumnsFromPath(fileSplit.getPathString(), pathPartitionKeys, - false, isACID) : fileSplit.getPartitionValues(); + FilePartitionUtils.ParsedColumnsFromPath partitionValuesFromPath = + fileSplit.getPartitionValues() == null + ? FilePartitionUtils.parseColumnsFromPathWithNullInfo( + fileSplit.getPathString(), pathPartitionKeys, false, isACID) + : FilePartitionUtils.normalizeColumnsFromPath(fileSplit.getPartitionValues()); - TFileRangeDesc rangeDesc = createFileRangeDesc(fileSplit, partitionValuesFromPath, pathPartitionKeys); + TFileRangeDesc rangeDesc = createFileRangeDesc(fileSplit, partitionValuesFromPath.getValues(), + pathPartitionKeys, partitionValuesFromPath.getIsNull()); TFileCompressType fileCompressType = getFileCompressType(fileSplit); rangeDesc.setCompressType(fileCompressType); // Seed connector-specific setup with the scan-level default. A connector may then @@ -491,7 +493,8 @@ private TScanRangeLocations newLocations() { } private TFileRangeDesc createFileRangeDesc(FileSplit fileSplit, List columnsFromPath, - List columnsFromPathKeys) { + List columnsFromPathKeys, + List columnsFromPathIsNull) { TFileRangeDesc rangeDesc = new TFileRangeDesc(); rangeDesc.setStartOffset(fileSplit.getStart()); rangeDesc.setSize(fileSplit.getLength()); @@ -501,6 +504,7 @@ private TFileRangeDesc createFileRangeDesc(FileSplit fileSplit, List col if (!columnsFromPathKeys.isEmpty()) { rangeDesc.setColumnsFromPath(columnsFromPath); rangeDesc.setColumnsFromPathKeys(columnsFromPathKeys); + rangeDesc.setColumnsFromPathIsNull(columnsFromPathIsNull); } rangeDesc.setFileType(fileSplit.getLocationType()); diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/hive/HiveMetaStoreCache.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/hive/HiveMetaStoreCache.java index 8b07f3e875a5ce..957a2f903e956f 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/datasource/hive/HiveMetaStoreCache.java +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/hive/HiveMetaStoreCache.java @@ -29,7 +29,6 @@ import org.apache.doris.common.AnalysisException; import org.apache.doris.common.CacheFactory; import org.apache.doris.common.Config; -import org.apache.doris.common.FeConstants; import org.apache.doris.common.UserException; import org.apache.doris.common.security.authentication.AuthenticationConfig; import org.apache.doris.common.security.authentication.HadoopAuthenticator; @@ -393,10 +392,10 @@ private FileCacheValue loadFiles(FileCacheKey key, DirectoryLister directoryList try { FileCacheValue result = getFileCache(finalLocation, key.inputFormat, key.getPartitionValues(), directoryLister, table); - // Replace default hive partition with a null_string. + // Replace default hive partition with null to distinguish it from a literal "\N". for (int i = 0; i < result.getValuesSize(); i++) { if (HIVE_DEFAULT_PARTITION.equals(result.getPartitionValues().get(i))) { - result.getPartitionValues().set(i, FeConstants.null_string); + result.getPartitionValues().set(i, null); } } 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 0b63d4d539af43..b3b6eec5ad282d 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.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; @@ -31,6 +32,7 @@ import org.apache.doris.common.util.LocationPath; import org.apache.doris.datasource.ExternalTable; import org.apache.doris.datasource.ExternalUtil; +import org.apache.doris.datasource.FilePartitionUtils; import org.apache.doris.datasource.NameMapping; import org.apache.doris.datasource.TableFormatType; import org.apache.doris.datasource.hive.HivePartition; @@ -324,8 +326,11 @@ private void setHudiParams(TFileRangeDesc rangeDesc, HudiSplit hudiSplit) { formPathKeys.add(entry.getKey()); formPathValues.add(entry.getValue()); } + FilePartitionUtils.ParsedColumnsFromPath parsedColumnsFromPath = + FilePartitionUtils.normalizeColumnsFromPath(formPathValues); rangeDesc.setColumnsFromPathKeys(formPathKeys); - rangeDesc.setColumnsFromPath(formPathValues); + rangeDesc.setColumnsFromPath(parsedColumnsFromPath.getValues()); + rangeDesc.setColumnsFromPathIsNull(parsedColumnsFromPath.getIsNull()); } rangeDesc.setTableFormatParams(tableFormatFileDesc); } @@ -351,7 +356,7 @@ private List getPrunedPartitions(HoodieTableMetaClient metaClient String path = basePath + "/" + key; hivePartitions.add(new HivePartition( nameMapping, false, inputFormat, path, - ((ListPartitionItem) value).getItems().get(0).getPartitionValuesAsStringList(), + getPartitionValues((ListPartitionItem) value), Maps.newHashMap())); } ); @@ -368,6 +373,13 @@ private List getPrunedPartitions(HoodieTableMetaClient metaClient return Lists.newArrayList(dummyPartition); } + static List 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 getIncrementalSplits() { if (canUseNativeReader()) { List 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 cc8970d3d62566..5d41e400c0e88a 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,7 +131,6 @@ import java.time.temporal.TemporalAccessor; import java.util.ArrayList; import java.util.Comparator; -import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.Optional; @@ -637,27 +636,10 @@ public static Type icebergTypeToDorisType(org.apache.iceberg.types.Type type, bo } } - /** - * Get partition info map for identity partitions only, considering partition - * evolution. - * For non-identity partitions (e.g., day, bucket, truncate), returns null to - * skip - * dynamic partition pruning. - * - * @param partitionData The partition data from the file - * @param partitionSpec The partition spec corresponding to the file's specId - * (required) - * @param timeZone The time zone for timestamp serialization - * @return Map of partition field name to partition value string, or null if - * there are non-identity partitions - */ - public static Map getPartitionInfoMap(PartitionData partitionData, PartitionSpec partitionSpec, - String timeZone) { - Map partitionInfoMap = new HashMap<>(); + public static Map getIdentityPartitionInfoMap(PartitionData partitionData, + PartitionSpec partitionSpec, Schema querySchema, String timeZone) { + Map partitionInfoMap = Maps.newLinkedHashMap(); List fields = partitionData.getPartitionType().asNestedType().fields(); - - // Check if all partition fields are identity transform - // If any field is not identity, return null to skip dynamic partition pruning List partitionFields = partitionSpec.fields(); Preconditions.checkArgument(fields.size() == partitionFields.size(), "PartitionData fields size does not match PartitionSpec fields size"); @@ -665,32 +647,33 @@ public static Map getPartitionInfoMap(PartitionData partitionDat for (int i = 0; i < fields.size(); i++) { NestedField field = fields.get(i); PartitionField partitionField = partitionFields.get(i); - - // Only process identity transform partitions - // For other transforms (day, bucket, truncate, etc.), skip dynamic partition - // pruning if (!partitionField.transform().isIdentity()) { - if (LOG.isDebugEnabled()) { - LOG.debug( - "Skip dynamic partition pruning for non-identity partition field: {} with transform: {}", - field.name(), partitionField.transform().toString()); - } - return null; + continue; + } + if (!isSupportedPartitionValueType(field.type().typeId())) { + continue; + } + String columnName = querySchema.findColumnName(partitionField.sourceId()); + if (columnName == null) { + continue; } Object value = partitionData.get(i); try { String partitionString = serializePartitionValue(field.type(), value, timeZone); - partitionInfoMap.put(field.name(), partitionString); + partitionInfoMap.put(columnName, partitionString); } catch (UnsupportedOperationException e) { LOG.warn("Failed to serialize Iceberg table partition value for field {}: {}", field.name(), e.getMessage()); - return null; } } return partitionInfoMap; } + private static boolean isSupportedPartitionValueType(TypeID typeId) { + return typeId != TypeID.BINARY && typeId != TypeID.FIXED; + } + private static String serializePartitionValue(org.apache.iceberg.types.Type type, Object value, String timeZone) { switch (type.typeId()) { case BOOLEAN: @@ -703,6 +686,16 @@ private static String serializePartitionValue(org.apache.iceberg.types.Type type return null; } return value.toString(); + case FLOAT: + if (value == null) { + return null; + } + return Float.toString((Float) value); + case DOUBLE: + if (value == null) { + return null; + } + return Double.toString((Double) value); // case binary, fixed should not supported, because if return string with utf8, // the data maybe be corrupted case DATE: @@ -729,7 +722,8 @@ private static String serializePartitionValue(org.apache.iceberg.types.Type type 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 07423573423a6a..55dd03c35f587f 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/source/IcebergScanNode.java +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/source/IcebergScanNode.java @@ -25,6 +25,7 @@ import org.apache.doris.catalog.Column; import org.apache.doris.catalog.Env; import org.apache.doris.catalog.TableIf; +import org.apache.doris.common.Pair; import org.apache.doris.common.UserException; import org.apache.doris.common.profile.SummaryProfile; import org.apache.doris.common.security.authentication.ExecutionAuthenticator; @@ -78,6 +79,7 @@ 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; @@ -120,6 +122,7 @@ public class IcebergScanNode extends FileQueryScanNode { private IcebergSource source; private Table icebergTable; + private Schema querySchema; private List 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, @@ -131,9 +134,9 @@ public class IcebergScanNode extends FileQueryScanNode { private long countFromSnapshot; private static final long COUNT_WITH_PARALLEL_SPLITS = 10000; private long targetSplitSize = 0; - // This is used to avoid repeatedly calculating partition info map for the same partition data. - private Map> partitionMapInfos; - private boolean isPartitionedTable; + // Used to avoid repeatedly calculating partition info map for the same + // partition data and spec. + private Map, Map> partitionMapInfos; private int formatVersion; private ExecutionAuthenticator preExecutionAuthenticator; private TableScan icebergTableScan; @@ -201,8 +204,12 @@ public IcebergScanNode(PlanNodeId id, TupleDescriptor desc, boolean needCheckCol @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( @@ -299,23 +306,33 @@ private void setIcebergParams(TFileRangeDesc rangeDesc, IcebergSplit icebergSpli } } tableFormatFileDesc.setIcebergParams(fileDesc); - Map partitionValues = icebergSplit.getIcebergPartitionValues(); - if (partitionValues != null) { - List fromPathKeys = new ArrayList<>(); - List fromPathValues = new ArrayList<>(); - List fromPathIsNull = new ArrayList<>(); - for (Map.Entry entry : partitionValues.entrySet()) { - fromPathKeys.add(entry.getKey()); - fromPathValues.add(entry.getValue() != null ? entry.getValue() : ""); - fromPathIsNull.add(entry.getValue() == null); - } - rangeDesc.setColumnsFromPathKeys(fromPathKeys); - rangeDesc.setColumnsFromPath(fromPathValues); - rangeDesc.setColumnsFromPathIsNull(fromPathIsNull); - } + setPartitionValues(rangeDesc, icebergSplit.getIcebergPartitionValues()); rangeDesc.setTableFormatParams(tableFormatFileDesc); } + @VisibleForTesting + void setPartitionValues(TFileRangeDesc rangeDesc, Map partitionValues) { + rangeDesc.unsetColumnsFromPathKeys(); + rangeDesc.unsetColumnsFromPath(); + rangeDesc.unsetColumnsFromPathIsNull(); + + if (partitionValues == null || partitionValues.isEmpty()) { + return; + } + + List fromPathKeys = new ArrayList<>(partitionValues.size()); + List fromPathValues = new ArrayList<>(partitionValues.size()); + List fromPathIsNull = new ArrayList<>(partitionValues.size()); + partitionValues.forEach((partitionKey, partitionValue) -> { + fromPathKeys.add(partitionKey); + fromPathValues.add(partitionValue == null ? "" : partitionValue); + fromPathIsNull.add(partitionValue == null); + }); + rangeDesc.setColumnsFromPathKeys(fromPathKeys); + rangeDesc.setColumnsFromPath(fromPathValues); + rangeDesc.setColumnsFromPathIsNull(fromPathIsNull); + } + @Override protected List getDeleteFiles(TFileRangeDesc rangeDesc) { List deleteFiles = new ArrayList<>(); @@ -769,27 +786,17 @@ private Split createIcebergSplit(FileScanTask fileScanTask) { } 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(); - if (sessionVariable.isEnableRuntimeFilterPartitionPrune()) { - // Get specId and corresponding PartitionSpec to handle partition evolution - 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 partitionInfoMap = partitionMapInfos.computeIfAbsent( - partitionData, k -> { - return IcebergUtils.getPartitionInfoMap(partitionData, partitionSpec, - sessionVariable.getTimeZone()); - }); - // Only set partition values if all partitions are identity transform - // For non-identity partitions, getPartitionInfoMap returns null to skip dynamic partition pruning - if (partitionInfoMap != null) { - split.setIcebergPartitionValues(partitionInfoMap); - } - } else { - partitionMapInfos.put(partitionData, null); + Map partitionInfoMap = partitionMapInfos.computeIfAbsent( + Pair.of(specId, partitionData), k -> IcebergUtils.getIdentityPartitionInfoMap( + partitionData, partitionSpec, querySchema, sessionVariable.getTimeZone())); + if (!partitionInfoMap.isEmpty()) { + split.setIcebergPartitionValues(partitionInfoMap); } } return split; @@ -947,20 +954,9 @@ public TFileFormatType getFileFormatType() throws UserException { @Override public List getPathPartitionKeys() throws UserException { - // return icebergTable.spec().fields().stream().map(PartitionField::name).map(String::toLowerCase) - // .collect(Collectors.toList()); - /**First, iceberg partition columns are based on existing fields, which will be stored in the actual data file. - * Second, iceberg partition columns support Partition transforms. In this case, the path partition key is not - * equal to the column name of the partition column, so remove this code and get all the columns you want to - * read from the file. - * Related code: - * be/src/vec/exec/scan/vfile_scanner.cpp: - * VFileScanner::_init_expr_ctxes() - * if (slot_info.is_file_slot) { - * xxxx - * } - */ - return new ArrayList<>(); + // Iceberg identity partition columns are still file columns. Per-range partition values are + // only used for partition pruning and as a fallback for files written before partition evolution. + return Collections.emptyList(); } private void recordManifestCacheAccess(boolean cacheHit) { diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/load/NereidsFileGroupInfo.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/load/NereidsFileGroupInfo.java index b0862277a633ed..b9a2da955a363d 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/nereids/load/NereidsFileGroupInfo.java +++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/load/NereidsFileGroupInfo.java @@ -26,10 +26,10 @@ import org.apache.doris.common.Config; import org.apache.doris.common.Pair; import org.apache.doris.common.UserException; -import org.apache.doris.common.util.BrokerUtil; import org.apache.doris.common.util.Util; import org.apache.doris.datasource.FederationBackendPolicy; import org.apache.doris.datasource.FileGroupInfo; +import org.apache.doris.datasource.FilePartitionUtils; import org.apache.doris.system.Backend; import org.apache.doris.thrift.TBrokerFileStatus; import org.apache.doris.thrift.TExternalScanRange; @@ -279,11 +279,12 @@ public void createScanRangeLocationsUnsplittable(NereidsParamCreateContext conte context.fileGroup.getFileFormatProperties().getCompressionType(), fileStatus.path); context.params.setCompressType(compressType); - List columnsFromPath = BrokerUtil.parseColumnsFromPath(fileStatus.path, - context.fileGroup.getColumnNamesFromPath()); + FilePartitionUtils.ParsedColumnsFromPath columnsFromPath = + FilePartitionUtils.parseColumnsFromPathWithNullInfo(fileStatus.path, + context.fileGroup.getColumnNamesFromPath(), true, false); List columnsFromPathKeys = context.fileGroup.getColumnNamesFromPath(); - TFileRangeDesc rangeDesc = createFileRangeDesc(0, fileStatus, fileStatus.size, columnsFromPath, - columnsFromPathKeys); + TFileRangeDesc rangeDesc = createFileRangeDesc(0, fileStatus, fileStatus.size, + columnsFromPath.getValues(), columnsFromPathKeys, columnsFromPath.getIsNull()); locations.getScanRange().getExtScanRange().getFileScanRange().addToRanges(rangeDesc); } scanRangeLocations.add(locations); @@ -331,15 +332,16 @@ public void createScanRangeLocationsSplittable(NereidsParamCreateContext context context.fileGroup.getFileFormatProperties().getCompressionType(), fileStatus.path); context.params.setCompressType(compressType); - List columnsFromPath = BrokerUtil.parseColumnsFromPath(fileStatus.path, - context.fileGroup.getColumnNamesFromPath()); + FilePartitionUtils.ParsedColumnsFromPath columnsFromPath = + FilePartitionUtils.parseColumnsFromPathWithNullInfo(fileStatus.path, + context.fileGroup.getColumnNamesFromPath(), true, false); List columnsFromPathKeys = context.fileGroup.getColumnNamesFromPath(); // Assign scan range locations only for broker load. // stream load has only one file, and no need to set multi scan ranges. if (tmpBytes > bytesPerInstance && jobType != FileGroupInfo.JobType.STREAM_LOAD) { long rangeBytes = bytesPerInstance - curInstanceBytes; TFileRangeDesc rangeDesc = createFileRangeDesc(curFileOffset, fileStatus, rangeBytes, - columnsFromPath, columnsFromPathKeys); + columnsFromPath.getValues(), columnsFromPathKeys, columnsFromPath.getIsNull()); curLocations.getScanRange().getExtScanRange().getFileScanRange().addToRanges(rangeDesc); curFileOffset += rangeBytes; @@ -348,8 +350,8 @@ public void createScanRangeLocationsSplittable(NereidsParamCreateContext context curLocations = newLocations(context.params, brokerDesc, backendPolicy); curInstanceBytes = 0; } else { - TFileRangeDesc rangeDesc = createFileRangeDesc(curFileOffset, fileStatus, leftBytes, columnsFromPath, - columnsFromPathKeys); + TFileRangeDesc rangeDesc = createFileRangeDesc(curFileOffset, fileStatus, leftBytes, + columnsFromPath.getValues(), columnsFromPathKeys, columnsFromPath.getIsNull()); curLocations.getScanRange().getExtScanRange().getFileScanRange().addToRanges(rangeDesc); curFileOffset = 0; curInstanceBytes += leftBytes; @@ -420,7 +422,7 @@ private TFileFormatType formatType(String fileFormat, String path) throws UserEx } private TFileRangeDesc createFileRangeDesc(long curFileOffset, TBrokerFileStatus fileStatus, long rangeBytes, - List columnsFromPath, List columnsFromPathKeys) { + List columnsFromPath, List columnsFromPathKeys, List columnsFromPathIsNull) { TFileRangeDesc rangeDesc = new TFileRangeDesc(); if (jobType == FileGroupInfo.JobType.BULK_LOAD) { rangeDesc.setPath(fileStatus.path); @@ -429,6 +431,7 @@ private TFileRangeDesc createFileRangeDesc(long curFileOffset, TBrokerFileStatus rangeDesc.setFileSize(fileStatus.size); rangeDesc.setColumnsFromPath(columnsFromPath); rangeDesc.setColumnsFromPathKeys(columnsFromPathKeys); + rangeDesc.setColumnsFromPathIsNull(columnsFromPathIsNull); if (getFileType() == TFileType.FILE_HDFS) { URI fileUri = new Path(fileStatus.path).toUri(); rangeDesc.setFsName(fileUri.getScheme() + "://" + fileUri.getAuthority()); 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 e08eaa2c8256fb..c71512896a1af1 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 @@ private void fillColumns(InternalService.PFetchTableSchemaResult result) { // 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/FilePartitionUtilsTest.java b/fe/fe-core/src/test/java/org/apache/doris/datasource/FilePartitionUtilsTest.java new file mode 100644 index 00000000000000..544431d96e723e --- /dev/null +++ b/fe/fe-core/src/test/java/org/apache/doris/datasource/FilePartitionUtilsTest.java @@ -0,0 +1,71 @@ +// 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; + +import org.apache.doris.datasource.FilePartitionUtils.ParsedColumnsFromPath; +import org.apache.doris.datasource.hive.HiveMetaStoreCache; + +import org.junit.Assert; +import org.junit.Test; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; + +public class FilePartitionUtilsTest { + + @Test + public void testParseColumnsFromPathPreservesNullMetadataAndInputKeys() throws Exception { + List partitionKeys = new ArrayList<>(Arrays.asList("region", "dt")); + ParsedColumnsFromPath parsed = FilePartitionUtils.parseColumnsFromPathWithNullInfo( + "hdfs://host/table/Region=cn/Dt=" + HiveMetaStoreCache.HIVE_DEFAULT_PARTITION + + "/data.parquet", + partitionKeys, false, false); + + Assert.assertEquals(Arrays.asList("cn", ""), parsed.getValues()); + Assert.assertEquals(Arrays.asList(false, true), parsed.getIsNull()); + Assert.assertEquals(Arrays.asList("region", "dt"), partitionKeys); + } + + @Test + public void testParseColumnsFromPathPreservesLiteralBackslashN() throws Exception { + ParsedColumnsFromPath parsed = FilePartitionUtils.parseColumnsFromPathWithNullInfo( + "hdfs://host/table/p=\\N/data.orc", Arrays.asList("p"), true, false); + + Assert.assertEquals(Arrays.asList("\\N"), parsed.getValues()); + Assert.assertEquals(Arrays.asList(false), parsed.getIsNull()); + } + + @Test + public void testParseAcidColumnsFromPath() throws Exception { + ParsedColumnsFromPath parsed = FilePartitionUtils.parseColumnsFromPathWithNullInfo( + "hdfs://host/table/p=value/delta_1_1/bucket_00000", Arrays.asList("p"), true, true); + + Assert.assertEquals(Arrays.asList("value"), parsed.getValues()); + Assert.assertEquals(Arrays.asList(false), parsed.getIsNull()); + } + + @Test + public void testNormalizeColumnsFromPath() { + ParsedColumnsFromPath parsed = FilePartitionUtils.normalizeColumnsFromPath( + Arrays.asList(null, HiveMetaStoreCache.HIVE_DEFAULT_PARTITION, "", "\\N")); + + Assert.assertEquals(Arrays.asList("", "", "", "\\N"), parsed.getValues()); + Assert.assertEquals(Arrays.asList(true, true, false, false), parsed.getIsNull()); + } +} 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 00000000000000..6befa82f418cf2 --- /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 values = Arrays.asList( + new PartitionValue("__HIVE_DEFAULT_PARTITION__", true), + new PartitionValue("NULL")); + List 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 3330685b85a4b7..81e935fc7d9064 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 @@ -30,6 +30,7 @@ import org.apache.iceberg.ManifestContent; import org.apache.iceberg.ManifestFile; import org.apache.iceberg.ManifestFile.PartitionFieldSummary; +import org.apache.iceberg.PartitionData; import org.apache.iceberg.PartitionSpec; import org.apache.iceberg.Schema; import org.apache.iceberg.Snapshot; @@ -134,6 +135,88 @@ public void testParseSchemaPreservesNonLowercaseColumnNames() { Assert.assertEquals("PART", columns.get(1).getName()); } + @Test + public void testGetIdentityPartitionInfoMapReturnsIdentityColumnsOnly() { + Schema schema = new Schema( + Types.NestedField.required(1, "Dt", Types.StringType.get()), + Types.NestedField.required(2, "ts", Types.TimestampType.withoutZone())); + PartitionSpec partitionSpec = PartitionSpec.builderFor(schema) + .identity("Dt") + .day("ts") + .build(); + PartitionData partitionData = new PartitionData(partitionSpec.partitionType()); + partitionData.set(0, "2025-01-01"); + partitionData.set(1, 20000); + + Map partitionInfoMap = IcebergUtils.getIdentityPartitionInfoMap( + partitionData, partitionSpec, schema, "UTC"); + Assert.assertEquals(Collections.singletonMap("Dt", "2025-01-01"), partitionInfoMap); + } + + @Test + public void testGetIdentityPartitionInfoMapSupportsFloatingPointPartitions() { + Schema schema = new Schema( + Types.NestedField.required(1, "float_partition", Types.FloatType.get()), + Types.NestedField.required(2, "double_partition", Types.DoubleType.get())); + PartitionSpec partitionSpec = PartitionSpec.builderFor(schema) + .identity("float_partition") + .identity("double_partition") + .build(); + float floatValue = Math.nextUp(0.1F); + double doubleValue = Math.nextUp(0.1D); + PartitionData partitionData = new PartitionData(partitionSpec.partitionType()); + partitionData.set(0, floatValue); + partitionData.set(1, doubleValue); + + Map partitionInfoMap = IcebergUtils.getIdentityPartitionInfoMap( + partitionData, partitionSpec, schema, "UTC"); + + String serializedFloat = partitionInfoMap.get("float_partition"); + String serializedDouble = partitionInfoMap.get("double_partition"); + Assert.assertEquals(Float.toString(floatValue), serializedFloat); + Assert.assertEquals(Double.toString(doubleValue), serializedDouble); + Assert.assertEquals(Float.floatToIntBits(floatValue), + Float.floatToIntBits(Float.parseFloat(serializedFloat))); + Assert.assertEquals(Double.doubleToLongBits(doubleValue), + 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 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 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 8b42f50774cf54..fd8e8643344aea 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 @@ -37,7 +37,10 @@ import java.lang.reflect.Method; import java.util.ArrayList; +import java.util.Arrays; import java.util.Collections; +import java.util.LinkedHashMap; +import java.util.Map; public class IcebergScanNodeTest { private static final long MB = 1024L * 1024L; @@ -95,4 +98,27 @@ public void testSetIcebergParamsUsesSplitFileFormat() throws Exception { Assert.assertEquals(TFileFormatType.FORMAT_ORC, rangeDesc.getFormatType()); } + + @Test + public void testSetPartitionValuesBuildsPerRangeAlignedMetadata() throws Exception { + TestIcebergScanNode node = new TestIcebergScanNode(new SessionVariable()); + Assert.assertTrue(node.getPathPartitionKeys().isEmpty()); + + Map partitionValues = new LinkedHashMap<>(); + partitionValues.put("Dt", null); + partitionValues.put("Region", "cn"); + 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(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 e5b06bd5dd4101..e0ead50ed13450 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.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;