Merge remote-tracking branch 'refs/remotes/upstream/master' into master-paimon-list.version

# Conflicts:
#	fe/fe-core/src/main/java/org/apache/doris/datasource/property/storage/AzureProperties.java

Author: CalvinKirs

be/src/common/config.cpp

@@ -1177,6 +1177,9 @@ DEFINE_Int32(inverted_index_query_cache_shards, "256");
 // inverted index match bitmap cache size
 DEFINE_String(inverted_index_query_cache_limit, "10%");
 
+// condition cache limit
+DEFINE_Int16(condition_cache_limit, "512");
+
 // inverted index
 DEFINE_mDouble(inverted_index_ram_buffer_size, "512");
 // -1 indicates not working.

be/src/common/config.h

@@ -1202,6 +1202,9 @@ DECLARE_Int32(inverted_index_query_cache_shards);
 // inverted index match bitmap cache size
 DECLARE_String(inverted_index_query_cache_limit);
 
+// condition cache limit
+DECLARE_Int16(condition_cache_limit);
+
 // inverted index
 DECLARE_mDouble(inverted_index_ram_buffer_size);
 DECLARE_mInt32(inverted_index_max_buffered_docs);

be/src/io/file_factory.h

@@ -113,6 +113,8 @@ class FileFactory {
             return TFileType::FILE_LOCAL;
         case TStorageBackendType::S3:
             return TFileType::FILE_S3;
+        case TStorageBackendType::AZURE:
+            return TFileType::FILE_S3;
         case TStorageBackendType::BROKER:
             return TFileType::FILE_BROKER;
         case TStorageBackendType::HDFS:

be/src/io/fs/azure_obj_storage_client.cpp

@@ -43,6 +43,7 @@
 #include "common/status.h"
 #include "io/fs/obj_storage_client.h"
 #include "util/bvar_helper.h"
+#include "util/coding.h"
 #include "util/s3_util.h"
 
 using namespace Azure::Storage::Blobs;
@@ -54,8 +55,9 @@ std::string wrap_object_storage_path_msg(const doris::io::ObjectStoragePathOptio
 }
 
 auto base64_encode_part_num(int part_num) {
-    return Aws::Utils::HashingUtils::Base64Encode(
-            {reinterpret_cast<unsigned char*>(&part_num), sizeof(part_num)});
+    uint8_t buf[4];
+    encode_fixed32_le(buf, static_cast<uint32_t>(part_num));
+    return Aws::Utils::HashingUtils::Base64Encode({buf, sizeof(buf)});
 }
 
 template <typename Func>

be/src/olap/iterators.h

@@ -25,6 +25,7 @@
 #include "olap/block_column_predicate.h"
 #include "olap/column_predicate.h"
 #include "olap/olap_common.h"
+#include "olap/row_cursor.h"
 #include "olap/rowset/segment_v2/ann_index/ann_topn_runtime.h"
 #include "olap/rowset/segment_v2/row_ranges.h"
 #include "olap/tablet_schema.h"
@@ -35,7 +36,6 @@
 
 namespace doris {
 
-class RowCursor;
 class Schema;
 class ColumnPredicate;
 
@@ -71,6 +71,22 @@ class StorageReadOptions {
         const RowCursor* upper_key = nullptr;
         // whether `upper_key` is included in the range
         bool include_upper;
+
+        uint64_t get_digest(uint64_t seed) const {
+            if (lower_key != nullptr) {
+                auto key_str = lower_key->to_string();
+                seed = HashUtil::hash64(key_str.c_str(), key_str.size(), seed);
+                seed = HashUtil::hash64(&include_lower, sizeof(include_lower), seed);
+            }
+
+            if (upper_key != nullptr) {
+                auto key_str = upper_key->to_string();
+                seed = HashUtil::hash64(key_str.c_str(), key_str.size(), seed);
+                seed = HashUtil::hash64(&include_upper, sizeof(include_upper), seed);
+            }
+
+            return seed;
+        }
     };
 
     // reader's key ranges, empty if not existed.
@@ -129,6 +145,12 @@ class StorageReadOptions {
 
     std::shared_ptr<vectorized::ScoreRuntime> score_runtime;
     CollectionStatisticsPtr collection_statistics;
+
+    // Cache for sparse column data to avoid redundant reads
+    // col_unique_id -> cached column_ptr
+    std::unordered_map<int32_t, vectorized::ColumnPtr> sparse_column_cache;
+
+    uint64_t condition_cache_digest = 0;
 };
 
 struct CompactionSampleInfo {

be/src/olap/olap_common.h

@@ -407,6 +407,10 @@ struct OlapReaderStatistics {
     int64_t output_index_result_column_timer = 0;
     // number of segment filtered by column stat when creating seg iterator
     int64_t filtered_segment_number = 0;
+    // number of segment with condition cache hit
+    int64_t condition_cache_hit_seg_nums = 0;
+    // number of rows filtered by condition cache hit
+    int64_t condition_cache_filtered_rows = 0;
     // total number of segment
     int64_t total_segment_number = 0;
 

be/src/olap/parallel_scanner_builder.cpp

@@ -34,7 +34,7 @@ using namespace vectorized;
 
 Status ParallelScannerBuilder::build_scanners(std::list<ScannerSPtr>& scanners) {
     RETURN_IF_ERROR(_load());
-    if (_optimize_index_scan_parallelism) {
+    if (_scan_parallelism_by_segment) {
         return _build_scanners_by_segment(scanners);
     } else if (_is_dup_mow_key) {
         // Default strategy for DUP/MOW tables: split by rowids within segments
@@ -87,7 +87,7 @@ Status ParallelScannerBuilder::_build_scanners_by_rowid(std::list<ScannerSPtr>&
                     auto rows_need = _rows_per_scanner - rows_collected;
 
                     // 0.9: try to avoid splitting the segments into excessively small parts.
-                    if (rows_need >= remaining_rows * 0.9) {
+                    if (rows_need >= remaining_rows * 9 / 10) {
                         rows_need = remaining_rows;
                     }
                     DCHECK_LE(rows_need, remaining_rows);
@@ -167,6 +167,8 @@ Status ParallelScannerBuilder::_build_scanners_by_rowid(std::list<ScannerSPtr>&
 // for the involved tablets. It preserves delete predicates and key ranges, and clones
 // RowsetReader per scanner to avoid sharing between scanners.
 Status ParallelScannerBuilder::_build_scanners_by_segment(std::list<ScannerSPtr>& scanners) {
+    DCHECK_GE(_rows_per_scanner, _min_rows_per_scanner);
+
     for (auto&& [tablet, version] : _tablets) {
         DCHECK(_all_read_sources.contains(tablet->tablet_id()));
         auto& entire_read_source = _all_read_sources[tablet->tablet_id()];
@@ -176,8 +178,10 @@ Status ParallelScannerBuilder::_build_scanners_by_segment(std::list<ScannerSPtr>
             ExecEnv::GetInstance()->storage_engine().to_cloud().tablet_hotspot().count(*tablet);
         }
 
-        // For each RowSet split in the read source, split by segment id and build
-        // one scanner per segment. Keep delete predicates shared.
+        // Collect segments into scanners based on rows count instead of one scanner per segment
+        TabletReader::ReadSource partitial_read_source;
+        int64_t rows_collected = 0;
+
         for (auto& rs_split : entire_read_source.rs_splits) {
             auto reader = rs_split.rs_reader;
             auto rowset = reader->rowset();
@@ -188,21 +192,62 @@ Status ParallelScannerBuilder::_build_scanners_by_segment(std::list<ScannerSPtr>
                 continue;
             }
 
-            // Build scanners for [i, i+1) segment range, without row-range slicing.
-            for (int64_t i = 0; i < rowset->num_segments(); ++i) {
-                RowSetSplits split(reader->clone());
-                split.segment_offsets.first = i;
-                split.segment_offsets.second = i + 1;
-                // No row-ranges slicing; scan whole segment i.
-                DCHECK_GE(split.segment_offsets.second, split.segment_offsets.first + 1);
+            int64_t segment_start = 0;
+            auto split = RowSetSplits(reader->clone());
 
-                TabletReader::ReadSource partitial_read_source;
+            for (size_t i = 0; i < segments_rows.size(); ++i) {
+                const size_t rows_of_segment = segments_rows[i];
+
+                // Check if adding this segment would exceed rows_per_scanner
+                // 0.9: try to avoid splitting the segments into excessively small parts.
+                if (rows_collected > 0 && (rows_collected + rows_of_segment > _rows_per_scanner &&
+                                           rows_collected < _rows_per_scanner * 9 / 10)) {
+                    // Create a new scanner with collected segments
+                    split.segment_offsets.first = segment_start;
+                    split.segment_offsets.second =
+                            i; // Range is [segment_start, i), including all segments from segment_start to i-1
+
+                    DCHECK_GT(split.segment_offsets.second, split.segment_offsets.first);
+
+                    partitial_read_source.rs_splits.emplace_back(std::move(split));
+
+                    scanners.emplace_back(
+                            _build_scanner(tablet, version, _key_ranges,
+                                           {std::move(partitial_read_source.rs_splits),
+                                            entire_read_source.delete_predicates}));
+
+                    // Reset for next scanner
+                    partitial_read_source = TabletReader::ReadSource();
+                    split = RowSetSplits(reader->clone());
+                    segment_start = i;
+                    rows_collected = 0;
+                }
+
+                // Add current segment to the current scanner
+                rows_collected += rows_of_segment;
+            }
+
+            // Add remaining segments in this rowset to a scanner
+            if (rows_collected > 0) {
+                split.segment_offsets.first = segment_start;
+                split.segment_offsets.second = segments_rows.size();
+                DCHECK_GT(split.segment_offsets.second, split.segment_offsets.first);
                 partitial_read_source.rs_splits.emplace_back(std::move(split));
+            }
+        }
 
-                scanners.emplace_back(_build_scanner(tablet, version, _key_ranges,
-                                                     {std::move(partitial_read_source.rs_splits),
-                                                      entire_read_source.delete_predicates}));
+        // Add remaining segments across all rowsets to a scanner
+        if (rows_collected > 0) {
+            DCHECK_GT(partitial_read_source.rs_splits.size(), 0);
+#ifndef NDEBUG
+            for (auto& split : partitial_read_source.rs_splits) {
+                DCHECK(split.rs_reader != nullptr);
+                DCHECK_LT(split.segment_offsets.first, split.segment_offsets.second);
             }
+#endif
+            scanners.emplace_back(_build_scanner(tablet, version, _key_ranges,
+                                                 {std::move(partitial_read_source.rs_splits),
+                                                  entire_read_source.delete_predicates}));
         }
     }
 

be/src/olap/parallel_scanner_builder.h

@@ -64,7 +64,7 @@ class ParallelScannerBuilder {
 
     void set_min_rows_per_scanner(int64_t size) { _min_rows_per_scanner = size; }
 
-    void set_optimize_index_scan_parallelism(bool v) { _optimize_index_scan_parallelism = v; }
+    void set_scan_parallelism_by_segment(bool v) { _scan_parallelism_by_segment = v; }
 
     const OlapReaderStatistics* builder_stats() const { return &_builder_stats; }
 
@@ -95,7 +95,7 @@ class ParallelScannerBuilder {
     std::map<RowsetId, std::vector<size_t>> _all_segments_rows;
 
     // Force building one scanner per segment when true.
-    bool _optimize_index_scan_parallelism {false};
+    bool _scan_parallelism_by_segment {false};
 
     std::shared_ptr<RuntimeProfile> _scanner_profile;
     OlapReaderStatistics _builder_stats;

be/src/olap/predicate_creator.h

@@ -17,9 +17,15 @@
 
 #pragma once
 
+#include <fast_float/fast_float.h>
+
 #include <charconv>
+#include <stdexcept>
+#include <string>
 #include <type_traits>
 
+#include "common/exception.h"
+#include "common/status.h"
 #include "exec/olap_utils.h"
 #include "exprs/create_predicate_function.h"
 #include "exprs/hybrid_set.h"
@@ -33,7 +39,10 @@
 #include "runtime/primitive_type.h"
 #include "util/date_func.h"
 #include "util/string_util.h"
+#include "vec/common/string_ref.h"
 #include "vec/data_types/data_type.h"
+#include "vec/functions/cast/cast_parameters.h"
+#include "vec/functions/cast/cast_to_basic_number_common.h"
 
 namespace doris {
 #include "common/compile_check_begin.h"
@@ -65,15 +74,29 @@ class IntegerPredicateCreator : public PredicateCreator<ConditionType> {
 private:
     static CppType convert(const std::string& condition) {
         CppType value = 0;
-        // because std::from_chars can't compile on macOS
-        if constexpr (std::is_same_v<CppType, double>) {
-            value = std::stod(condition, nullptr);
-        } else if constexpr (std::is_same_v<CppType, float>) {
-            value = std::stof(condition, nullptr);
+        if constexpr (std::is_floating_point_v<CppType>) {
+            vectorized::CastParameters params;
+            if (vectorized::CastToFloat::from_string(StringRef {condition.data(), condition.size()},
+                                                     value, params)) {
+                return value;
+            } else {
+                throw Exception(
+                        ErrorCode::INVALID_ARGUMENT,
+                        fmt::format("convert string to number failed, str: {} to float/double",
+                                    condition));
+            }
         } else {
-            std::from_chars(condition.data(), condition.data() + condition.size(), value);
+            auto ret =
+                    std::from_chars(condition.data(), condition.data() + condition.size(), value);
+            if (ret.ptr == condition.data() + condition.size()) {
+                return value;
+            } else {
+                throw Exception(
+                        ErrorCode::INVALID_ARGUMENT,
+                        fmt::format("convert string to number failed, str: {}, error: [{}] {}",
+                                    condition, ret.ec, std::make_error_code(ret.ec).message()));
+            }
         }
-        return value;
     }
 };
 

be/src/olap/rowset/beta_rowset_reader.cpp

@@ -142,6 +142,10 @@ Status BetaRowsetReader::get_segment_iterators(RowsetReaderContext* read_context
             read_columns.push_back(cid);
         }
     }
+    // disable condition cache if you have delete condition
+    _read_context->condition_cache_digest =
+            delete_columns_set.empty() ? _read_context->condition_cache_digest : 0;
+    // create segment iterators
     VLOG_NOTICE << "read columns size: " << read_columns.size();
     _input_schema = std::make_shared<Schema>(_read_context->tablet_schema->columns(), read_columns);
     if (_read_context->predicates != nullptr) {
@@ -216,6 +220,14 @@ Status BetaRowsetReader::get_segment_iterators(RowsetReaderContext* read_context
                 _read_context->runtime_state->query_options().disable_file_cache;
     }
 
+    if (_read_context->condition_cache_digest) {
+        for (const auto& key_range : _read_options.key_ranges) {
+            _read_context->condition_cache_digest =
+                    key_range.get_digest(_read_context->condition_cache_digest);
+        }
+        _read_options.condition_cache_digest = _read_context->condition_cache_digest;
+    }
+
     _read_options.io_ctx.expiration_time =
             read_context->ttl_seconds > 0 && _rowset->rowset_meta()->newest_write_timestamp() > 0
                     ? _rowset->rowset_meta()->newest_write_timestamp() + read_context->ttl_seconds
@@ -267,6 +279,10 @@ Status BetaRowsetReader::get_segment_iterators(RowsetReaderContext* read_context
             DCHECK_EQ(seg_end - seg_start, _segment_row_ranges.size());
             auto local_options = _read_options;
             local_options.row_ranges = _segment_row_ranges[i - seg_start];
+            if (local_options.condition_cache_digest) {
+                local_options.condition_cache_digest =
+                        local_options.row_ranges.get_digest(local_options.condition_cache_digest);
+            }
             iter = std::make_unique<LazyInitSegmentIterator>(_rowset, i, should_use_cache,
                                                              _input_schema, local_options);
         }