Add unit tests for be_refactor

be/src/olap/storage_engine.cpp

@@ -202,7 +202,9 @@ void StorageEngine::_update_storage_medium_type_count() {
     }
 
     _available_storage_medium_type_count = available_storage_medium_types.size();
-    StorageEngine::instance()->tablet_manager()->update_storage_medium_type_count(_available_storage_medium_type_count);
+    if (_tablet_manager != nullptr) {
+        _tablet_manager->update_storage_medium_type_count(_available_storage_medium_type_count);
+    }
 }
 
 
@@ -327,7 +329,7 @@ void StorageEngine::start_disk_stat_monitor() {
         it.second->health_check();
     }
     _update_storage_medium_type_count();
-    _delete_tables_on_unused_root_path();
+    _delete_tablets_on_unused_root_path();
 
     // if drop tables
     // notify disk_state_worker_thread and tablet_worker_thread until they received
@@ -422,19 +424,23 @@ DataDir* StorageEngine::get_store(const std::string& path) {
     return it->second;
 }
 
-void StorageEngine::_delete_tables_on_unused_root_path() {
+void StorageEngine::_delete_tablets_on_unused_root_path() {
     vector<TabletInfo> tablet_info_vec;
     uint32_t unused_root_path_num = 0;
     uint32_t total_root_path_num = 0;
 
     std::lock_guard<std::mutex> l(_store_lock);
+    if (_store_map.size() == 0) {
+        return;
+    }
 
     for (auto& it : _store_map) {
         total_root_path_num++;
         if (it.second->is_used()) {
             continue;
         }
         it.second->clear_tablets(&tablet_info_vec);
+        ++unused_root_path_num;
     }
 
     if (_used_disk_not_enough(unused_root_path_num, total_root_path_num)) {
@@ -473,7 +479,12 @@ OLAPStatus StorageEngine::clear() {
     delete FileHandler::get_fd_cache();
     FileHandler::set_fd_cache(nullptr);
     SAFE_DELETE(_index_stream_lru_cache);
-
+    std::lock_guard<std::mutex> l(_store_lock);
+    for (auto& store_pair : _store_map) {
+        delete store_pair.second;
+        store_pair.second = nullptr;
+    }
+    _store_map.clear();
     return OLAP_SUCCESS;
 }
 

be/src/olap/storage_engine.h

@@ -211,7 +211,7 @@ class StorageEngine {
             const std::string& root_path,
             std::string* unused_flag_file);
 
-    void _delete_tables_on_unused_root_path();
+    void _delete_tablets_on_unused_root_path();
 
     void _update_storage_medium_type_count();
 

be/test/olap/CMakeLists.txt

@@ -44,3 +44,5 @@ ADD_BE_TEST(serialize_test)
 ADD_BE_TEST(olap_meta_test)
 ADD_BE_TEST(tablet_meta_manager_test)
 ADD_BE_TEST(rowset/rowset_meta_manager_test)
+ADD_BE_TEST(rowset/rowset_meta_test)
+ADD_BE_TEST(rowset/alpha_rowset_test)

be/test/olap/delta_writer_test.cpp

@@ -280,7 +280,7 @@ class TestDeltaWriter : public ::testing::Test {
         //ASSERT_EQ(OLAP_SUCCESS, remove_all_dir(config::storage_root_path));
     }
 };
-/*
+
 TEST_F(TestDeltaWriter, open) {
     TCreateTabletReq request;
     create_tablet_request(10003, 270068375, &request);
@@ -311,7 +311,7 @@ TEST_F(TestDeltaWriter, open) {
     res = k_engine->tablet_manager()->drop_tablet(tablet_id, schema_hash);
     ASSERT_EQ(OLAP_SUCCESS, res);
 }
-*/
+
 TEST_F(TestDeltaWriter, write) {
     TCreateTabletReq request;
     create_tablet_request(10004, 270068376, &request);

be/test/olap/rowset/alpha_rowset_test.cpp

@@ -0,0 +1,266 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include <string>
+#include <sstream>
+#include <fstream>
+
+#include "gtest/gtest.h"
+#include "gmock/gmock.h"
+#include "boost/filesystem.hpp"
+#include "json2pb/json_to_pb.h"
+#include "util/logging.h"
+#include "olap/olap_meta.h"
+#include "olap/rowset/rowset_writer_context.h"
+#include "olap/rowset/rowset_reader_context.h"
+#include "olap/rowset/alpha_rowset.h"
+#include "olap/rowset/alpha_rowset_writer.h"
+#include "olap/rowset/alpha_rowset_reader.h"
+#include "olap/data_dir.h"
+#include "olap/storage_engine.h"
+
+#ifndef BE_TEST
+#define BE_TEST
+#endif
+
+using ::testing::_;
+using ::testing::Return;
+using ::testing::SetArgPointee;
+using std::string;
+
+namespace doris {
+
+static const uint32_t MAX_PATH_LEN = 1024;
+
+StorageEngine* k_engine = nullptr;
+
+void set_up() {
+    config::path_gc_check = false;
+    char buffer[MAX_PATH_LEN];
+    getcwd(buffer, MAX_PATH_LEN);
+    config::storage_root_path = std::string(buffer) + "/data_test";
+    remove_all_dir(config::storage_root_path);
+    OLAPStatus res = create_dir(config::storage_root_path);
+    ASSERT_EQ(OLAP_SUCCESS, res);
+    std::vector<StorePath> paths;
+    paths.emplace_back(config::storage_root_path, -1);
+    std::string data_path = config::storage_root_path + "/data";
+    res = create_dir(data_path);
+    ASSERT_EQ(OLAP_SUCCESS, res);
+    std::string shard_path = data_path + "/0";
+    res = create_dir(shard_path);
+    ASSERT_EQ(OLAP_SUCCESS, res);
+    std::string tablet_path = shard_path + "/12345";
+    res = create_dir(tablet_path);
+    ASSERT_EQ(OLAP_SUCCESS, res);
+    std::string schema_hash_path = tablet_path + "/1111";
+    res = create_dir(schema_hash_path);
+    ASSERT_EQ(OLAP_SUCCESS, res);
+
+    doris::EngineOptions options;
+    options.store_paths = paths;
+    doris::StorageEngine::open(options, &k_engine);
+}
+
+void tear_down() {
+    delete k_engine;
+    k_engine = nullptr;
+    remove_all_dir(config::storage_root_path);
+    remove_all_dir(std::string(getenv("DORIS_HOME")) + UNUSED_PREFIX);
+}
+
+void create_rowset_writer_context(TabletSchema* tablet_schema, DataDir* data_dir,
+        RowsetWriterContext* rowset_writer_context) {
+    rowset_writer_context->rowset_id = 10000;
+    rowset_writer_context->tablet_id = 12345;
+    rowset_writer_context->tablet_schema_hash = 1111;
+    rowset_writer_context->partition_id = 10;
+    rowset_writer_context->rowset_type = ALPHA_ROWSET;
+    rowset_writer_context->rowset_path_prefix = config::storage_root_path + "/data/0/12345/1111";
+    rowset_writer_context->rowset_state = VISIBLE;
+    rowset_writer_context->data_dir = data_dir;
+    rowset_writer_context->tablet_schema = tablet_schema;
+    rowset_writer_context->version.first = 0;
+    rowset_writer_context->version.second = 1;
+    rowset_writer_context->version_hash = 110;
+}
+
+void create_rowset_reader_context(TabletSchema* tablet_schema, const std::vector<uint32_t>* return_columns,
+        const DeleteHandler* delete_handler, std::vector<ColumnPredicate*>* predicates,
+        std::set<uint32_t>* load_bf_columns, Conditions* conditions, RowsetReaderContext* rowset_reader_context) {
+    rowset_reader_context->reader_type = READER_ALTER_TABLE;
+    rowset_reader_context->tablet_schema = tablet_schema;
+    rowset_reader_context->preaggregation = false;
+    rowset_reader_context->return_columns = return_columns;
+    rowset_reader_context->delete_handler = delete_handler;
+    rowset_reader_context->is_using_cache = false;
+    rowset_reader_context->lower_bound_keys = nullptr;
+    rowset_reader_context->is_lower_keys_included = nullptr;
+    rowset_reader_context->upper_bound_keys = nullptr;
+    rowset_reader_context->is_upper_keys_included = nullptr;
+    rowset_reader_context->predicates = predicates;
+    rowset_reader_context->load_bf_columns = load_bf_columns;
+    rowset_reader_context->conditions = conditions;
+    rowset_reader_context->lru_cache = k_engine->index_stream_lru_cache();
+}
+
+void create_tablet_schema(KeysType keys_type, TabletSchema* tablet_schema) {
+    TabletSchemaPB tablet_schema_pb;
+    tablet_schema_pb.set_keys_type(keys_type);
+    tablet_schema_pb.set_num_short_key_columns(2);
+    tablet_schema_pb.set_num_rows_per_row_block(1024);
+    tablet_schema_pb.set_compress_kind(COMPRESS_NONE);
+    tablet_schema_pb.set_next_column_unique_id(4);
+
+    ColumnPB* column_1 = tablet_schema_pb.add_column();
+    column_1->set_unique_id(1);
+    column_1->set_name("k1");
+    column_1->set_type("INT");
+    column_1->set_is_key(true);
+    column_1->set_is_nullable(true);
+    column_1->set_is_bf_column(false);
+
+    ColumnPB* column_2 = tablet_schema_pb.add_column();
+    column_2->set_unique_id(2);
+    column_2->set_name("k2");
+    column_2->set_type("VARCHAR");
+    column_2->set_length(20);
+    column_2->set_is_key(true);
+    column_2->set_is_nullable(true);
+    column_2->set_is_bf_column(false);
+
+    ColumnPB* column_3 = tablet_schema_pb.add_column();
+    column_3->set_unique_id(3);
+    column_3->set_name("v1");
+    column_3->set_type("INT");
+    column_3->set_is_key(false);
+    column_3->set_is_nullable(false);
+    column_3->set_is_bf_column(false);
+    column_3->set_aggregation("SUM");
+
+    tablet_schema->init_from_pb(tablet_schema_pb);
+}
+
+class AlphaRowsetTest : public testing::Test {
+public:
+    virtual void SetUp() {
+        set_up();
+        _data_dir = k_engine->get_store(config::storage_root_path);
+        ASSERT_TRUE(_data_dir != nullptr);
+        _alpha_rowset_writer = new(std::nothrow) AlphaRowsetWriter();
+        _mem_tracker.reset(new MemTracker(-1));
+        _mem_pool.reset(new MemPool(_mem_tracker.get()));
+    }
+
+    virtual void TearDown() {
+        delete _alpha_rowset_writer;
+        _alpha_rowset_writer = nullptr;
+        tear_down();
+    }
+
+private:
+    DataDir* _data_dir;
+    AlphaRowsetWriter* _alpha_rowset_writer;
+    std::unique_ptr<MemTracker> _mem_tracker;
+    std::unique_ptr<MemPool> _mem_pool;
+};
+/*
+TEST_F(AlphaRowsetTest, TestAlphaRowsetWriter) {
+    TabletSchema tablet_schema;
+    create_tablet_schema(AGG_KEYS, &tablet_schema);
+    RowsetWriterContext rowset_writer_context;
+    create_rowset_writer_context(&tablet_schema, _data_dir, &rowset_writer_context);
+    _alpha_rowset_writer->init(rowset_writer_context);
+    RowCursor row;
+    OLAPStatus res = row.init(tablet_schema);
+    ASSERT_EQ(OLAP_SUCCESS, res);
+
+    int32_t field_0 = 10;
+    row.set_field_content(0, reinterpret_cast<char*>(&field_0), _mem_pool.get());
+    Slice field_1("well");
+    row.set_field_content(1, reinterpret_cast<char*>(&field_1), _mem_pool.get());
+    int32_t field_2 = 100;
+    row.set_field_content(2, reinterpret_cast<char*>(&field_2), _mem_pool.get());
+    _alpha_rowset_writer->add_row(&row);
+    _alpha_rowset_writer->flush();
+    RowsetSharedPtr alpha_rowset = _alpha_rowset_writer->build();
+    ASSERT_TRUE(alpha_rowset != nullptr);
+    ASSERT_EQ(10000, alpha_rowset->rowset_id());
+    ASSERT_EQ(1, alpha_rowset->num_rows());
+}
+*/
+TEST_F(AlphaRowsetTest, TestAlphaRowsetReader) {
+    TabletSchema tablet_schema;
+    create_tablet_schema(AGG_KEYS, &tablet_schema);
+    RowsetWriterContext rowset_writer_context;
+    create_rowset_writer_context(&tablet_schema, _data_dir, &rowset_writer_context);
+    _alpha_rowset_writer->init(rowset_writer_context);
+    RowCursor row;
+    OLAPStatus res = row.init(tablet_schema);
+    ASSERT_EQ(OLAP_SUCCESS, res);
+
+    int32_t field_0 = 10;
+    row.set_field_content(0, reinterpret_cast<char*>(&field_0), _mem_pool.get());
+    Slice field_1("well");
+    row.set_field_content(1, reinterpret_cast<char*>(&field_1), _mem_pool.get());
+    int32_t field_2 = 100;
+    row.set_field_content(2, reinterpret_cast<char*>(&field_2), _mem_pool.get());
+    _alpha_rowset_writer->add_row(&row);
+    _alpha_rowset_writer->flush();
+    RowsetSharedPtr alpha_rowset = _alpha_rowset_writer->build();
+    ASSERT_TRUE(alpha_rowset != nullptr);
+    ASSERT_EQ(10000, alpha_rowset->rowset_id());
+    ASSERT_EQ(1, alpha_rowset->num_rows());
+    RowsetReaderSharedPtr rowset_reader = alpha_rowset->create_reader();
+    ASSERT_TRUE(rowset_reader != nullptr);
+    std::vector<uint32_t> return_columns;
+    for (int i = 0;  i < tablet_schema.num_columns(); ++i) {
+        return_columns.push_back(i);
+    }
+    DeleteHandler delete_handler;
+    DelPredicateArray predicate_array;
+    res = delete_handler.init(tablet_schema, predicate_array, 4);
+    ASSERT_EQ(OLAP_SUCCESS, res);
+    RowsetReaderContext rowset_reader_context;
+
+    std::set<uint32_t> load_bf_columns;
+    std::vector<ColumnPredicate*> predicates;
+    Conditions conditions;
+    create_rowset_reader_context(&tablet_schema, &return_columns, &delete_handler,
+            &predicates, &load_bf_columns, &conditions, &rowset_reader_context);
+    res = rowset_reader->init(&rowset_reader_context);
+    ASSERT_EQ(OLAP_SUCCESS, res);
+    RowBlock* row_block = nullptr;
+    res = rowset_reader->next_block(&row_block);
+    ASSERT_EQ(OLAP_SUCCESS, res);
+    ASSERT_EQ(1, row_block->remaining());
+}
+
+}  // namespace doris
+
+int main(int argc, char **argv) {
+    std::string conffile = std::string(getenv("DORIS_HOME")) + "/conf/be.conf";
+    if (!doris::config::init(conffile.c_str(), false)) {
+        fprintf(stderr, "error read config file. \n");
+        return -1;
+    }
+    doris::init_glog("be-test");
+    ::testing::InitGoogleTest(&argc, argv);
+    int ret = RUN_ALL_TESTS();
+    google::protobuf::ShutdownProtobufLibrary();
+    return ret;
+}