This is an automated email from the ASF dual-hosted git repository.
zhaoc pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-doris.git
The following commit(s) were added to refs/heads/master by this push:
new 3690f3e Add rowset state (#2691)
3690f3e is described below
commit 3690f3e917c4aea8127d1e02acf42a4174cdc44d
Author: kangpinghuang <[email protected]>
AuthorDate: Fri Jan 10 14:17:57 2020 +0800
Add rowset state (#2691)
1. add rowset state to rowset
2. add close api to rowset to release resources
issue: #2665
---
be/src/olap/compaction.cpp | 1 +
be/src/olap/olap_define.h | 3 +-
be/src/olap/rowset/alpha_rowset.cpp | 7 +-
be/src/olap/rowset/alpha_rowset.h | 6 +-
be/src/olap/rowset/alpha_rowset_reader.cpp | 3 +
be/src/olap/rowset/beta_rowset.cpp | 11 ++-
be/src/olap/rowset/beta_rowset.h | 9 ++-
be/src/olap/rowset/beta_rowset_reader.cpp | 2 +
be/src/olap/rowset/beta_rowset_reader.h | 4 +-
be/src/olap/rowset/rowset.cpp | 25 +++++-
be/src/olap/rowset/rowset.h | 126 ++++++++++++++++++++++++++++-
be/src/olap/storage_engine.cpp | 3 +-
12 files changed, 181 insertions(+), 19 deletions(-)
diff --git a/be/src/olap/compaction.cpp b/be/src/olap/compaction.cpp
index 418b695..dd21b24 100644
--- a/be/src/olap/compaction.cpp
+++ b/be/src/olap/compaction.cpp
@@ -175,6 +175,7 @@ OLAPStatus Compaction::gc_unused_rowsets() {
for (auto& rowset : _input_rowsets) {
storage_engine->add_unused_rowset(rowset);
}
+ _input_rowsets.clear();
return OLAP_SUCCESS;
}
diff --git a/be/src/olap/olap_define.h b/be/src/olap/olap_define.h
index 78d327b..ed4b0fd 100644
--- a/be/src/olap/olap_define.h
+++ b/be/src/olap/olap_define.h
@@ -369,7 +369,8 @@ enum OLAPStatus {
OLAP_ERR_ROWSET_INVALID = -3108,
OLAP_ERR_ROWSET_LOAD_FAILED = -3109,
OLAP_ERR_ROWSET_READER_INIT = -3110,
- OLAP_ERR_ROWSET_READ_FAILED = -3111
+ OLAP_ERR_ROWSET_READ_FAILED = -3111,
+ OLAP_ERR_ROWSET_INVALID_STATE_TRANSITION = -3112
};
enum ColumnFamilyIndex {
diff --git a/be/src/olap/rowset/alpha_rowset.cpp
b/be/src/olap/rowset/alpha_rowset.cpp
index 4f419a3..9e84b3e 100644
--- a/be/src/olap/rowset/alpha_rowset.cpp
+++ b/be/src/olap/rowset/alpha_rowset.cpp
@@ -30,7 +30,7 @@ AlphaRowset::AlphaRowset(const TabletSchema* schema,
: Rowset(schema, std::move(rowset_path), std::move(rowset_meta)) {
}
-OLAPStatus AlphaRowset::do_load_once(bool use_cache) {
+OLAPStatus AlphaRowset::do_load(bool use_cache) {
for (auto& segment_group: _segment_groups) {
// validate segment group
if (segment_group->validate() != OLAP_SUCCESS) {
@@ -52,14 +52,15 @@ OLAPStatus AlphaRowset::do_load_once(bool use_cache) {
}
OLAPStatus AlphaRowset::create_reader(std::shared_ptr<RowsetReader>* result) {
- RETURN_NOT_OK(load());
result->reset(new AlphaRowsetReader(
_schema->num_rows_per_row_block(),
std::static_pointer_cast<AlphaRowset>(shared_from_this())));
return OLAP_SUCCESS;
}
OLAPStatus AlphaRowset::remove() {
- LOG(INFO) << "begin to remove files in rowset " << unique_id();
+ LOG(INFO) << "begin to remove files in rowset " << unique_id()
+ << ", version:" << start_version() << "-" << end_version()
+ << ", tabletid:" << _rowset_meta->tablet_id();
for (auto segment_group : _segment_groups) {
bool ret = segment_group->delete_all_files();
if (!ret) {
diff --git a/be/src/olap/rowset/alpha_rowset.h
b/be/src/olap/rowset/alpha_rowset.h
index 35e44f4..1d4f8a9 100644
--- a/be/src/olap/rowset/alpha_rowset.h
+++ b/be/src/olap/rowset/alpha_rowset.h
@@ -73,11 +73,13 @@ protected:
AlphaRowset(const TabletSchema* schema,
std::string rowset_path,
RowsetMetaSharedPtr rowset_meta);
-
+
// init segment groups
OLAPStatus init() override;
- OLAPStatus do_load_once(bool use_cache) override ;
+ OLAPStatus do_load(bool use_cache) override;
+
+ void do_close() override { }
// add custom logic when rowset is published
void make_visible_extra(Version version, VersionHash version_hash)
override;
diff --git a/be/src/olap/rowset/alpha_rowset_reader.cpp
b/be/src/olap/rowset/alpha_rowset_reader.cpp
index 9460601..278576a 100644
--- a/be/src/olap/rowset/alpha_rowset_reader.cpp
+++ b/be/src/olap/rowset/alpha_rowset_reader.cpp
@@ -29,13 +29,16 @@ AlphaRowsetReader::AlphaRowsetReader(
_alpha_rowset_meta(std::static_pointer_cast<AlphaRowsetMeta>(_rowset->rowset_meta()).get()),
_segment_groups(_rowset->_segment_groups),
_key_range_size(0) {
+ _rowset->aquire();
}
AlphaRowsetReader::~AlphaRowsetReader() {
delete _dst_cursor;
+ _rowset->release();
}
OLAPStatus AlphaRowsetReader::init(RowsetReaderContext* read_context) {
+ RETURN_NOT_OK(_rowset->load());
if (read_context == nullptr) {
return OLAP_ERR_INIT_FAILED;
}
diff --git a/be/src/olap/rowset/beta_rowset.cpp
b/be/src/olap/rowset/beta_rowset.cpp
index 204e581..6207034 100644
--- a/be/src/olap/rowset/beta_rowset.cpp
+++ b/be/src/olap/rowset/beta_rowset.cpp
@@ -45,7 +45,7 @@ OLAPStatus BetaRowset::init() {
}
// `use_cache` is ignored because beta rowset doesn't support fd cache now
-OLAPStatus BetaRowset::do_load_once(bool /*use_cache*/) {
+OLAPStatus BetaRowset::do_load(bool /*use_cache*/) {
// Open all segments under the current rowset
for (int seg_id = 0; seg_id < num_segments(); ++seg_id) {
std::string seg_path = segment_file_path(_rowset_path, rowset_id(),
seg_id);
@@ -62,7 +62,6 @@ OLAPStatus BetaRowset::do_load_once(bool /*use_cache*/) {
}
OLAPStatus BetaRowset::create_reader(RowsetReaderSharedPtr* result) {
- RETURN_NOT_OK(load());
// NOTE: We use std::static_pointer_cast for performance
result->reset(new
BetaRowsetReader(std::static_pointer_cast<BetaRowset>(shared_from_this())));
return OLAP_SUCCESS;
@@ -79,7 +78,9 @@ OLAPStatus BetaRowset::split_range(const RowCursor& start_key,
OLAPStatus BetaRowset::remove() {
// TODO should we close and remove all segment reader first?
- LOG(INFO) << "begin to remove files in rowset " << unique_id();
+ LOG(INFO) << "begin to remove files in rowset " << unique_id()
+ << ", version:" << start_version() << "-" << end_version()
+ << ", tabletid:" << _rowset_meta->tablet_id();
bool success = true;
for (int i = 0; i < num_segments(); ++i) {
std::string path = segment_file_path(_rowset_path, rowset_id(), i);
@@ -99,6 +100,10 @@ OLAPStatus BetaRowset::remove() {
return OLAP_SUCCESS;
}
+void BetaRowset::do_close() {
+ _segments.clear();
+}
+
OLAPStatus BetaRowset::link_files_to(const std::string& dir, RowsetId
new_rowset_id) {
for (int i = 0; i < num_segments(); ++i) {
std::string dst_link_path = segment_file_path(dir, new_rowset_id, i);
diff --git a/be/src/olap/rowset/beta_rowset.h b/be/src/olap/rowset/beta_rowset.h
index 3c6955e..f37d213 100644
--- a/be/src/olap/rowset/beta_rowset.h
+++ b/be/src/olap/rowset/beta_rowset.h
@@ -36,11 +36,11 @@ class BetaRowset : public Rowset {
public:
virtual ~BetaRowset();
+ OLAPStatus create_reader(RowsetReaderSharedPtr* result) override;
+
static std::string segment_file_path(
const std::string& segment_dir, const RowsetId& rowset_id, int
segment_id);
- OLAPStatus create_reader(RowsetReaderSharedPtr* result) override;
-
OLAPStatus split_range(const RowCursor& start_key,
const RowCursor& end_key,
uint64_t request_block_row_count,
@@ -66,9 +66,12 @@ protected:
std::string rowset_path,
RowsetMetaSharedPtr rowset_meta);
+ // init segment groups
OLAPStatus init() override;
- OLAPStatus do_load_once(bool use_cache) override ;
+ OLAPStatus do_load(bool use_cache) override;
+
+ void do_close() override;
private:
friend class BetaRowsetReader;
diff --git a/be/src/olap/rowset/beta_rowset_reader.cpp
b/be/src/olap/rowset/beta_rowset_reader.cpp
index e161c8d..4e52eb0 100644
--- a/be/src/olap/rowset/beta_rowset_reader.cpp
+++ b/be/src/olap/rowset/beta_rowset_reader.cpp
@@ -28,9 +28,11 @@ namespace doris {
BetaRowsetReader::BetaRowsetReader(BetaRowsetSharedPtr rowset)
: _rowset(std::move(rowset)), _stats(&_owned_stats) {
+ _rowset->aquire();
}
OLAPStatus BetaRowsetReader::init(RowsetReaderContext* read_context) {
+ RETURN_NOT_OK(_rowset->load());
_context = read_context;
if (_context->stats != nullptr) {
// schema change/compaction should use owned_stats
diff --git a/be/src/olap/rowset/beta_rowset_reader.h
b/be/src/olap/rowset/beta_rowset_reader.h
index 8afdcb1..ae5b687 100644
--- a/be/src/olap/rowset/beta_rowset_reader.h
+++ b/be/src/olap/rowset/beta_rowset_reader.h
@@ -31,7 +31,9 @@ class BetaRowsetReader : public RowsetReader {
public:
explicit BetaRowsetReader(BetaRowsetSharedPtr rowset);
- ~BetaRowsetReader() override = default;
+ ~BetaRowsetReader() override {
+ _rowset->release();
+ }
OLAPStatus init(RowsetReaderContext* read_context) override;
diff --git a/be/src/olap/rowset/rowset.cpp b/be/src/olap/rowset/rowset.cpp
index 7ca47ee..c30653d 100644
--- a/be/src/olap/rowset/rowset.cpp
+++ b/be/src/olap/rowset/rowset.cpp
@@ -24,7 +24,9 @@ Rowset::Rowset(const TabletSchema *schema,
RowsetMetaSharedPtr rowset_meta)
: _schema(schema),
_rowset_path(std::move(rowset_path)),
- _rowset_meta(std::move(rowset_meta)) {
+ _rowset_meta(std::move(rowset_meta)),
+ _refs_by_reader(0),
+ _rowset_state_machine(RowsetStateMachine()) {
_is_pending = !_rowset_meta->has_version();
if (_is_pending) {
@@ -36,7 +38,26 @@ Rowset::Rowset(const TabletSchema *schema,
}
OLAPStatus Rowset::load(bool use_cache) {
- return _load_once.call([this, use_cache] { return do_load_once(use_cache);
});
+ // if the state is ROWSET_UNLOADING it means close() is called
+ // and the rowset is already loaded, and the resource is not closed yet.
+ if (_rowset_state_machine.rowset_state() == ROWSET_LOADED) {
+ return OLAP_SUCCESS;
+ }
+ {
+ // before lock, if rowset state is ROWSET_UNLOADING, maybe it is doing
do_close in release
+ std::lock_guard<std::mutex> load_lock(_lock);
+ // after lock, if rowset state is ROWSET_UNLOADING, it is ok to return
+ if (_rowset_state_machine.rowset_state() == ROWSET_UNLOADED) {
+ // first do load, then change the state
+ RETURN_NOT_OK(do_load(use_cache));
+ RETURN_NOT_OK(_rowset_state_machine.on_load());
+ }
+ }
+ // load is done
+ LOG(INFO) << "rowset is loaded. rowset version:" << start_version() << "-"
<< end_version()
+ << ", state from ROWSET_UNLOADED to ROWSET_LOADED. tabletid:"
+ << _rowset_meta->tablet_id();
+ return OLAP_SUCCESS;
}
void Rowset::make_visible(Version version, VersionHash version_hash) {
diff --git a/be/src/olap/rowset/rowset.h b/be/src/olap/rowset/rowset.h
index d07c15e..7f0a0ca 100644
--- a/be/src/olap/rowset/rowset.h
+++ b/be/src/olap/rowset/rowset.h
@@ -20,11 +20,11 @@
#include <memory>
#include <vector>
+#include <mutex>
#include "gen_cpp/olap_file.pb.h"
#include "gutil/macros.h"
#include "olap/rowset/rowset_meta.h"
-#include "util/once.h"
namespace doris {
@@ -37,6 +37,61 @@ class RowsetFactory;
class RowsetReader;
class TabletSchema;
+// the rowset state transfer graph:
+// ROWSET_UNLOADED <--|
+// ↓ |
+// ROWSET_LOADED |
+// ↓ |
+// ROWSET_UNLOADING -->|
+enum RowsetState {
+ // state for new created rowset
+ ROWSET_UNLOADED,
+ // state after load() called
+ ROWSET_LOADED,
+ // state for closed() called but owned by some readers
+ ROWSET_UNLOADING
+};
+
+class RowsetStateMachine {
+public:
+ RowsetStateMachine() : _rowset_state(ROWSET_UNLOADED) { }
+
+ OLAPStatus on_load() {
+ switch (_rowset_state) {
+ case ROWSET_UNLOADED:
+ _rowset_state = ROWSET_LOADED;
+ break;
+
+ default:
+ return OLAP_ERR_ROWSET_INVALID_STATE_TRANSITION;
+ }
+ return OLAP_SUCCESS;
+ }
+
+ OLAPStatus on_close(uint64_t refs_by_reader) {
+ switch (_rowset_state) {
+ case ROWSET_LOADED:
+ if (refs_by_reader == 0) {
+ _rowset_state = ROWSET_UNLOADED;
+ } else {
+ _rowset_state = ROWSET_UNLOADING;
+ }
+ break;
+
+ default:
+ return OLAP_ERR_ROWSET_INVALID_STATE_TRANSITION;
+ }
+ return OLAP_SUCCESS;
+ }
+
+ RowsetState rowset_state() {
+ return _rowset_state;
+ }
+
+private:
+ RowsetState _rowset_state;
+};
+
class Rowset : public std::enable_shared_from_this<Rowset> {
public:
virtual ~Rowset() { }
@@ -95,6 +150,37 @@ public:
// TODO should we rename the method to remove_files() to be more specific?
virtual OLAPStatus remove() = 0;
+ // close to clear the resource owned by rowset
+ // including: open files, indexes and so on
+ // NOTICE: can not call this function in multithreads
+ void close() {
+ RowsetState old_state = _rowset_state_machine.rowset_state();
+ if (old_state != ROWSET_LOADED) {
+ return;
+ }
+ OLAPStatus st = OLAP_SUCCESS;
+ {
+ std::lock_guard<std::mutex> close_lock(_lock);
+ uint64_t current_refs = _refs_by_reader;
+ old_state = _rowset_state_machine.rowset_state();
+ if (old_state != ROWSET_LOADED) {
+ return;
+ }
+ if (current_refs == 0) {
+ do_close();
+ }
+ st = _rowset_state_machine.on_close(current_refs);
+ }
+ if (st != OLAP_SUCCESS) {
+ LOG(WARNING) << "state transition failed from:" <<
_rowset_state_machine.rowset_state();
+ return;
+ }
+ LOG(INFO) << "rowset is close. rowset state from:" << old_state
+ << " to " << _rowset_state_machine.rowset_state()
+ << ", version:" << start_version() << "-" << end_version()
+ << ", tabletid:" << _rowset_meta->tablet_id();
+ }
+
// hard link all files in this rowset to `dir` to form a new rowset with
id `new_rowset_id`.
virtual OLAPStatus link_files_to(const std::string& dir, RowsetId
new_rowset_id) = 0;
@@ -127,6 +213,32 @@ public:
return left->end_version() < right->end_version();
}
+ // this function is called by reader to increase reference of rowset
+ void aquire() {
+ ++_refs_by_reader;
+ }
+
+ void release() {
+ // if the refs by reader is 0 and the rowset is closed, should release
the resouce
+ uint64_t current_refs = --_refs_by_reader;
+ if (current_refs == 0 && _rowset_state_machine.rowset_state() ==
ROWSET_UNLOADING) {
+ {
+ std::lock_guard<std::mutex> release_lock(_lock);
+ // rejudge _refs_by_reader because we do not add lock in
create reader
+ if (_refs_by_reader == 0 &&
_rowset_state_machine.rowset_state() == ROWSET_UNLOADING) {
+ // first do close, then change state
+ do_close();
+ _rowset_state_machine.on_close(0);
+ }
+ }
+ if (_rowset_state_machine.rowset_state() == ROWSET_UNLOADED) {
+ LOG(INFO) << "close the rowset. rowset state from
ROWSET_UNLOADING to ROWSET_UNLOADED"
+ << ", version:" << start_version() << "-" <<
end_version()
+ << ", tabletid:" << _rowset_meta->tablet_id();
+ }
+ }
+ }
+
protected:
friend class RowsetFactory;
@@ -140,7 +252,10 @@ protected:
virtual OLAPStatus init() = 0;
// The actual implementation of load(). Guaranteed by to called exactly
once.
- virtual OLAPStatus do_load_once(bool use_cache) = 0;
+ virtual OLAPStatus do_load(bool use_cache) = 0;
+
+ // release resources in this api
+ virtual void do_close() = 0;
// allow subclass to add custom logic when rowset is being published
virtual void make_visible_extra(Version version, VersionHash version_hash)
{}
@@ -152,8 +267,13 @@ protected:
bool _is_pending; // rowset is pending iff it's not in visible state
bool _is_cumulative; // rowset is cumulative iff it's visible and start
version < end version
- DorisCallOnce<OLAPStatus> _load_once;
+ // mutex lock for load/close api because it is costly
+ std::mutex _lock;
bool _need_delete_file = false;
+ // variable to indicate how many rowset readers owned this rowset
+ std::atomic<uint64_t> _refs_by_reader;
+ // rowset state machine
+ RowsetStateMachine _rowset_state_machine;
};
} // namespace doris
diff --git a/be/src/olap/storage_engine.cpp b/be/src/olap/storage_engine.cpp
index 0bac06d..23e9dec 100644
--- a/be/src/olap/storage_engine.cpp
+++ b/be/src/olap/storage_engine.cpp
@@ -748,7 +748,7 @@ void StorageEngine::start_delete_unused_rowset() {
for (auto it = _unused_rowsets.begin(); it != _unused_rowsets.end();) {
if (it->second.use_count() != 1) {
++it;
- } else if (it->second->need_delete_file()){
+ } else if (it->second->need_delete_file()) {
LOG(INFO) << "start to remove rowset:" << it->second->rowset_id()
<< ", version:" << it->second->version().first << "-" <<
it->second->version().second;
OLAPStatus status = it->second->remove();
@@ -769,6 +769,7 @@ void StorageEngine::add_unused_rowset(RowsetSharedPtr
rowset) {
auto it = _unused_rowsets.find(rowset->unique_id());
if (it == _unused_rowsets.end()) {
rowset->set_need_delete_file();
+ rowset->close();
_unused_rowsets[rowset->unique_id()] = rowset;
release_rowset_id(rowset->rowset_id());
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
