Merge branch 'apache:master' into s7

Author: Tech-Circle-48

.github/workflows/build-thirdparty.yml

@@ -142,7 +142,7 @@ jobs:
     name: Build Third Party Libraries (macOS)
     needs: changes
     if: ${{ needs.changes.outputs.thirdparty_changes == 'true' }}
-    runs-on: macos-13
+    runs-on: macos-15
     steps:
       - name: Checkout ${{ github.ref }}
         uses: actions/checkout@v4

be/src/exec/rowid_fetcher.cpp

@@ -38,27 +38,22 @@
 #include <vector>
 
 #include "bthread/countdown_event.h"
-#include "cloud/cloud_storage_engine.h"
-#include "cloud/cloud_tablet.h"
-#include "cloud/cloud_tablet_mgr.h"
-#include "cloud/config.h"
 #include "common/config.h"
 #include "common/consts.h"
 #include "common/exception.h"
 #include "common/signal_handler.h"
 #include "exec/tablet_info.h" // DorisNodesInfo
 #include "olap/olap_common.h"
 #include "olap/rowset/beta_rowset.h"
+#include "olap/rowset/segment_v2/column_reader.h"
 #include "olap/storage_engine.h"
 #include "olap/tablet_fwd.h"
-#include "olap/tablet_manager.h"
 #include "olap/tablet_schema.h"
 #include "olap/utils.h"
 #include "runtime/descriptors.h"
 #include "runtime/exec_env.h"      // ExecEnv
 #include "runtime/fragment_mgr.h"  // FragmentMgr
 #include "runtime/runtime_state.h" // RuntimeState
-#include "runtime/types.h"
 #include "runtime/workload_group/workload_group_manager.h"
 #include "semaphore"
 #include "util/brpc_client_cache.h" // BrpcClientCache
@@ -69,13 +64,11 @@
 #include "vec/common/assert_cast.h"
 #include "vec/common/string_ref.h"
 #include "vec/core/block.h" // Block
-#include "vec/data_types/data_type_factory.hpp"
 #include "vec/data_types/data_type_struct.h"
 #include "vec/data_types/serde/data_type_serde.h"
 #include "vec/exec/format/orc/vorc_reader.h"
 #include "vec/exec/format/parquet/vparquet_reader.h"
 #include "vec/exec/scan/file_scanner.h"
-#include "vec/functions/function_helpers.h"
 #include "vec/jsonb/serialize.h"
 
 namespace doris {

be/src/index-tools/index_tool.cpp

@@ -621,11 +621,16 @@ int main(int argc, char** argv) {
         _CLLDELETE(analyzer);
         _CLLDELETE(char_string_reader);
 
-        auto ret = index_file_writer->close();
+        auto ret = index_file_writer->begin_close();
         if (!ret.ok()) {
             std::cerr << "IndexFileWriter close error:" << ret.msg() << std::endl;
             return -1;
         }
+        ret = index_file_writer->finish_close();
+        if (!ret.ok()) {
+            std::cerr << "IndexFileWriter wait close error:" << ret.msg() << std::endl;
+            return -1;
+        }
     } else if (FLAGS_operation == "show_nested_files_v2") {
         if (FLAGS_idx_file_path == "") {
             std::cout << "no file flag for show " << std::endl;

be/src/olap/base_tablet.cpp

@@ -45,6 +45,7 @@
 #include "olap/rowset/rowset.h"
 #include "olap/rowset/rowset_fwd.h"
 #include "olap/rowset/rowset_reader.h"
+#include "olap/rowset/segment_v2/column_reader.h"
 #include "olap/tablet_fwd.h"
 #include "olap/txn_manager.h"
 #include "service/point_query_executor.h"

be/src/olap/delete_bitmap_calculator.cpp

@@ -22,10 +22,22 @@
 #include "common/cast_set.h"
 #include "common/status.h"
 #include "olap/primary_key_index.h"
+#include "olap/rowset/segment_v2/indexed_column_reader.h"
+#include "olap/tablet_meta.h"
 #include "vec/data_types/data_type_factory.hpp"
 
 namespace doris {
 #include "common/compile_check_begin.h"
+
+MergeIndexDeleteBitmapCalculatorContext::MergeIndexDeleteBitmapCalculatorContext(
+        std::unique_ptr<segment_v2::IndexedColumnIterator> iter, vectorized::DataTypePtr index_type,
+        int32_t segment_id, size_t num_rows, size_t batch_max_size)
+        : _iter(std::move(iter)),
+          _index_type(std::move(index_type)),
+          _num_rows(num_rows),
+          _max_batch_size(batch_max_size),
+          _segment_id(segment_id) {}
+
 Status MergeIndexDeleteBitmapCalculatorContext::get_current_key(Slice& slice) {
     if (_cur_row_id >= _num_rows) {
         return Status::EndOfFile("Reach the end of file");

be/src/olap/delete_bitmap_calculator.h

@@ -27,23 +27,19 @@
 
 #include "common/config.h"
 #include "common/status.h"
-#include "olap/base_tablet.h"
-#include "olap/binlog_config.h"
-#include "olap/data_dir.h"
 #include "olap/key_coder.h"
 #include "olap/olap_common.h"
-#include "olap/rowset/rowset.h"
-#include "olap/rowset/rowset_meta.h"
-#include "olap/rowset/rowset_reader.h"
 #include "olap/rowset/segment_v2/segment.h"
-#include "olap/tablet_meta.h"
-#include "olap/tablet_schema.h"
-#include "olap/version_graph.h"
-#include "util/metrics.h"
-#include "util/once.h"
 #include "util/slice.h"
 
 namespace doris {
+namespace segment_v2 {
+class IndexedColumnIterator;
+class Segment;
+
+} // namespace segment_v2
+
+using SegmentSharedPtr = std::shared_ptr<segment_v2::Segment>;
 
 class MergeIndexDeleteBitmapCalculatorContext {
 public:
@@ -62,12 +58,7 @@ class MergeIndexDeleteBitmapCalculatorContext {
 
     MergeIndexDeleteBitmapCalculatorContext(std::unique_ptr<segment_v2::IndexedColumnIterator> iter,
                                             vectorized::DataTypePtr index_type, int32_t segment_id,
-                                            size_t num_rows, size_t batch_max_size = 1024)
-            : _iter(std::move(iter)),
-              _index_type(index_type),
-              _num_rows(num_rows),
-              _max_batch_size(batch_max_size),
-              _segment_id(segment_id) {}
+                                            size_t num_rows, size_t batch_max_size = 1024);
     Status get_current_key(Slice& slice);
     Status advance();
     Status seek_at_or_after(Slice const& key);

be/src/olap/rowset/beta_rowset_writer.cpp

@@ -206,16 +206,24 @@ Status InvertedIndexFileCollection::add(int seg_id, IndexFileWriterPtr&& index_w
     return Status::OK();
 }
 
-Status InvertedIndexFileCollection::close() {
+Status InvertedIndexFileCollection::begin_close() {
     std::lock_guard lock(_lock);
     for (auto&& [id, writer] : _inverted_index_file_writers) {
-        RETURN_IF_ERROR(writer->close());
+        RETURN_IF_ERROR(writer->begin_close());
         _total_size += writer->get_index_file_total_size();
     }
 
     return Status::OK();
 }
 
+Status InvertedIndexFileCollection::finish_close() {
+    std::lock_guard lock(_lock);
+    for (auto&& [id, writer] : _inverted_index_file_writers) {
+        RETURN_IF_ERROR(writer->finish_close());
+    }
+    return Status::OK();
+}
+
 Result<std::vector<const InvertedIndexFileInfo*>>
 InvertedIndexFileCollection::inverted_index_file_info(int seg_id_offset) {
     std::lock_guard lock(_lock);
@@ -1097,7 +1105,8 @@ Status BetaRowsetWriter::create_segment_writer_for_segcompaction(
     _segcompaction_worker->get_file_writer().reset(file_writer.release());
     if (auto& idx_file_writer = _segcompaction_worker->get_inverted_index_file_writer();
         idx_file_writer != nullptr) {
-        RETURN_IF_ERROR(idx_file_writer->close());
+        RETURN_IF_ERROR(idx_file_writer->begin_close());
+        RETURN_IF_ERROR(idx_file_writer->finish_close());
     }
     _segcompaction_worker->get_inverted_index_file_writer().reset(index_file_writer.release());
     return Status::OK();

be/src/olap/rowset/beta_rowset_writer.h

@@ -91,7 +91,10 @@ class InvertedIndexFileCollection {
 
     // Close all file writers
     // If the inverted index file writer is not closed, an error will be thrown during destruction
-    Status close();
+    Status begin_close();
+
+    // Wait for all inverted index file writers to be closed
+    Status finish_close();
 
     // Get inverted index file info in segment id order.
     // `seg_id_offset` is the offset of the segment id relative to the subscript of `_inverted_index_file_writers`,
@@ -214,9 +217,11 @@ class BaseBetaRowsetWriter : public RowsetWriter {
     // Some index files are written during normal compaction and some files are written during index compaction.
     // After all index writes are completed, call this method to write the final compound index file.
     Status _close_inverted_index_file_writers() {
-        RETURN_NOT_OK_STATUS_WITH_WARN(_idx_files.close(),
+        RETURN_NOT_OK_STATUS_WITH_WARN(_idx_files.begin_close(),
                                        "failed to close index file when build new rowset");
         this->_total_index_size += _idx_files.get_total_index_size();
+        RETURN_NOT_OK_STATUS_WITH_WARN(_idx_files.finish_close(),
+                                       "failed to wait close index file when build new rowset");
         return Status::OK();
     }
 

be/src/olap/rowset/segment_creator.cpp

@@ -111,7 +111,9 @@ Status SegmentFlusher::_internal_parse_variant_columns(vectorized::Block& block)
 }
 
 Status SegmentFlusher::close() {
-    return _seg_files.close();
+    RETURN_IF_ERROR(_seg_files.close());
+    RETURN_IF_ERROR(_idx_files.finish_close());
+    return Status::OK();
 }
 
 Status SegmentFlusher::_add_rows(std::unique_ptr<segment_v2::SegmentWriter>& segment_writer,

be/src/olap/rowset/segment_v2/column_reader.h

@@ -24,17 +24,14 @@
 #include <cstddef> // for size_t
 #include <cstdint> // for uint32_t
 #include <memory>  // for unique_ptr
-#include <mutex>
 #include <string>
 #include <utility>
 #include <vector>
 
 #include "common/config.h"
-#include "common/exception.h"
 #include "common/logging.h"
 #include "common/status.h" // for Status
 #include "io/fs/file_reader_writer_fwd.h"
-#include "io/fs/file_system.h"
 #include "io/io_common.h"
 #include "olap/olap_common.h"
 #include "olap/rowset/segment_v2/common.h"
@@ -50,9 +47,7 @@
 #include "util/once.h"
 #include "vec/columns/column.h"
 #include "vec/columns/column_array.h" // ColumnArray
-#include "vec/columns/subcolumn_tree.h"
 #include "vec/data_types/data_type.h"
-#include "vec/json/path_in_data.h"
 
 namespace doris {
 #include "common/compile_check_begin.h"
@@ -662,7 +657,7 @@ class RowIdColumnIterator : public ColumnIterator {
 
     Status next_batch(size_t* n, vectorized::MutableColumnPtr& dst, bool* has_null) override {
         for (size_t i = 0; i < *n; ++i) {
-            rowid_t row_id = cast_set<uint32_t>(_current_rowid + i);
+            const auto row_id = cast_set<uint32_t>(_current_rowid + i);
             GlobalRowLoacation location(_tablet_id, _rowset_id, _segment_id, row_id);
             dst->insert_data(reinterpret_cast<const char*>(&location), sizeof(GlobalRowLoacation));
         }
@@ -722,14 +717,12 @@ class RowIdColumnIteratorV2 : public ColumnIterator {
 // This iterator is used to read default value column
 class DefaultValueColumnIterator : public ColumnIterator {
 public:
-    DefaultValueColumnIterator(bool has_default_value, const std::string& default_value,
-                               bool is_nullable, TypeInfoPtr type_info, int precision, int scale)
+    DefaultValueColumnIterator(bool has_default_value, std::string default_value, bool is_nullable,
+                               TypeInfoPtr type_info, int precision, int scale)
             : _has_default_value(has_default_value),
-              _default_value(default_value),
+              _default_value(std::move(default_value)),
               _is_nullable(is_nullable),
               _type_info(std::move(type_info)),
-              _is_default_value_null(false),
-              _type_size(0),
               _precision(precision),
               _scale(scale) {}
 
@@ -766,8 +759,8 @@ class DefaultValueColumnIterator : public ColumnIterator {
     std::string _default_value;
     bool _is_nullable;
     TypeInfoPtr _type_info;
-    bool _is_default_value_null;
-    size_t _type_size;
+    bool _is_default_value_null {false};
+    size_t _type_size {0};
     int _precision;
     int _scale;
     std::vector<char> _mem_value;