This is an automated email from the ASF dual-hosted git repository.
dataroaring pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new c6063ed92f [Revert](lazy open) revert lazy open and add case (#21821)
c6063ed92f is described below
commit c6063ed92f0b95198b9c15c8e7c63fa02fcf3352
Author: HHoflittlefish777 <[email protected]>
AuthorDate: Tue Jul 18 19:41:33 2023 +0800
[Revert](lazy open) revert lazy open and add case (#21821)
---
be/src/runtime/load_channel.cpp | 31 -----
be/src/runtime/load_channel.h | 2 -
be/src/runtime/load_channel_mgr.cpp | 16 ---
be/src/runtime/load_channel_mgr.h | 2 -
be/src/runtime/tablets_channel.cpp | 132 +------------------
be/src/runtime/tablets_channel.h | 6 -
be/src/service/internal_service.cpp | 23 ----
be/src/service/internal_service.h | 4 -
be/src/vec/sink/vtablet_sink.cpp | 143 ---------------------
be/src/vec/sink/vtablet_sink.h | 12 --
be/test/vec/exec/vtablet_sink_test.cpp | 8 --
docs/en/docs/admin-manual/config/be-config.md | 6 -
docs/zh-CN/docs/admin-manual/config/be-config.md | 6 -
gensrc/proto/internal_service.proto | 11 --
.../suites/version_p0/version_p0.groovy | 45 +++++++
15 files changed, 48 insertions(+), 399 deletions(-)
diff --git a/be/src/runtime/load_channel.cpp b/be/src/runtime/load_channel.cpp
index 5067e2f19a..29ebb63725 100644
--- a/be/src/runtime/load_channel.cpp
+++ b/be/src/runtime/load_channel.cpp
@@ -90,37 +90,6 @@ Status LoadChannel::open(const PTabletWriterOpenRequest&
params) {
return Status::OK();
}
-Status LoadChannel::open_partition(const OpenPartitionRequest& params) {
- int64_t index_id = params.index_id();
-
- // check finish
- {
- std::lock_guard<std::mutex> l(_lock);
- auto it = _finished_channel_ids.find(index_id);
- if (it != _finished_channel_ids.end()) {
- return Status::OK();
- }
- }
- std::shared_ptr<TabletsChannel> channel;
- {
- std::lock_guard<std::mutex> l(_lock);
- auto it = _tablets_channels.find(index_id);
- if (it != _tablets_channels.end()) {
- channel = it->second;
- } else {
- fmt::memory_buffer buf;
- for (auto tablet : params.tablets()) {
- fmt::format_to(buf, "tablet id:{}", tablet.tablet_id());
- }
- LOG(WARNING) << "should be opened partition index id=" <<
params.index_id()
- << "tablet ids=" << fmt::to_string(buf);
- return Status::InternalError("Partition should be opened");
- }
- }
- RETURN_IF_ERROR(channel->open_all_writers_for_partition(params));
- return Status::OK();
-}
-
Status LoadChannel::_get_tablets_channel(std::shared_ptr<TabletsChannel>&
channel,
bool& is_finished, const int64_t
index_id) {
std::lock_guard<std::mutex> l(_lock);
diff --git a/be/src/runtime/load_channel.h b/be/src/runtime/load_channel.h
index 5648cb0ae1..3b5b4b2d58 100644
--- a/be/src/runtime/load_channel.h
+++ b/be/src/runtime/load_channel.h
@@ -60,8 +60,6 @@ public:
// open a new load channel if not exist
Status open(const PTabletWriterOpenRequest& request);
- Status open_partition(const OpenPartitionRequest& params);
-
// this batch must belong to a index in one transaction
Status add_batch(const PTabletWriterAddBlockRequest& request,
PTabletWriterAddBlockResult* response);
diff --git a/be/src/runtime/load_channel_mgr.cpp
b/be/src/runtime/load_channel_mgr.cpp
index 3d9dbc6f19..6862b99b2f 100644
--- a/be/src/runtime/load_channel_mgr.cpp
+++ b/be/src/runtime/load_channel_mgr.cpp
@@ -140,22 +140,6 @@ Status LoadChannelMgr::open(const
PTabletWriterOpenRequest& params) {
return Status::OK();
}
-Status LoadChannelMgr::open_partition(const OpenPartitionRequest& params) {
- UniqueId load_id(params.id());
- std::shared_ptr<LoadChannel> channel;
- {
- std::lock_guard<std::mutex> l(_lock);
- auto it = _load_channels.find(load_id);
- if (it != _load_channels.end()) {
- channel = it->second;
- } else {
- return Status::InternalError("unknown load id, load id=" +
load_id.to_string());
- }
- }
- RETURN_IF_ERROR(channel->open_partition(params));
- return Status::OK();
-}
-
static void dummy_deleter(const CacheKey& key, void* value) {}
Status LoadChannelMgr::_get_load_channel(std::shared_ptr<LoadChannel>&
channel, bool& is_eof,
diff --git a/be/src/runtime/load_channel_mgr.h
b/be/src/runtime/load_channel_mgr.h
index 663a395500..cad611b75b 100644
--- a/be/src/runtime/load_channel_mgr.h
+++ b/be/src/runtime/load_channel_mgr.h
@@ -54,8 +54,6 @@ public:
// open a new load channel if not exist
Status open(const PTabletWriterOpenRequest& request);
- Status open_partition(const OpenPartitionRequest& params);
-
Status add_batch(const PTabletWriterAddBlockRequest& request,
PTabletWriterAddBlockResult* response);
diff --git a/be/src/runtime/tablets_channel.cpp
b/be/src/runtime/tablets_channel.cpp
index 3844ee2d0b..545ee00e21 100644
--- a/be/src/runtime/tablets_channel.cpp
+++ b/be/src/runtime/tablets_channel.cpp
@@ -106,11 +106,8 @@ Status TabletsChannel::open(const
PTabletWriterOpenRequest& request) {
_next_seqs.resize(_num_remaining_senders, 0);
_closed_senders.Reset(_num_remaining_senders);
- if (!config::enable_lazy_open_partition) {
- RETURN_IF_ERROR(_open_all_writers(request));
- } else {
- _build_partition_tablets_relation(request);
- }
+ RETURN_IF_ERROR(_open_all_writers(request));
+
_state = kOpened;
return Status::OK();
}
@@ -300,7 +297,6 @@ void TabletsChannel::get_active_memtable_mem_consumption(
}
}
-// Old logic,used for opening all writers of all partitions.
Status TabletsChannel::_open_all_writers(const PTabletWriterOpenRequest&
request) {
std::vector<SlotDescriptor*>* index_slots = nullptr;
int32_t schema_hash = 0;
@@ -347,115 +343,6 @@ Status TabletsChannel::_open_all_writers(const
PTabletWriterOpenRequest& request
return Status::OK();
}
-// Due to the use of non blocking operations,
-// the open partition rpc has not yet arrived when the actual write request
arrives,
-// it is a logic to avoid delta writer not open.
-template <typename TabletWriterAddRequest>
-Status TabletsChannel::_open_all_writers_for_partition(const int64_t&
tablet_id,
- const
TabletWriterAddRequest& request) {
- std::vector<SlotDescriptor*>* index_slots = nullptr;
- int32_t schema_hash = 0;
- for (auto& index : _schema->indexes()) {
- if (index->index_id == _index_id) {
- index_slots = &index->slots;
- schema_hash = index->schema_hash;
- break;
- }
- }
- if (index_slots == nullptr) {
- return Status::InternalError("unknown index id, key={}",
_key.to_string());
- }
- int64_t partition_id = _tablet_partition_map[tablet_id];
- DCHECK(partition_id != 0);
- auto& tablets = _partition_tablets_map[partition_id];
- DCHECK(tablets.size() > 0);
- VLOG_DEBUG << fmt::format(
- "write tablet={}, open writers for all tablets in partition={},
total writers num={}",
- tablet_id, partition_id, tablets.size());
- for (auto& tablet : tablets) {
- WriteRequest wrequest;
- wrequest.index_id = request.index_id();
- wrequest.tablet_id = tablet;
- wrequest.schema_hash = schema_hash;
- wrequest.txn_id = _txn_id;
- wrequest.partition_id = partition_id;
- wrequest.load_id = request.id();
- wrequest.tuple_desc = _tuple_desc;
- wrequest.slots = index_slots;
- wrequest.is_high_priority = _is_high_priority;
- wrequest.table_schema_param = _schema;
-
- {
- std::lock_guard<SpinLock> l(_tablet_writers_lock);
-
- if (_tablet_writers.find(tablet) == _tablet_writers.end()) {
- DeltaWriter* writer = nullptr;
- auto st = DeltaWriter::open(&wrequest, &writer, _profile,
_load_id);
- if (!st.ok()) {
- auto err_msg = fmt::format(
- "open delta writer failed, tablet_id={}"
- ", txn_id={}, partition_id={}, err={}",
- tablet, _txn_id, partition_id, st.to_string());
- LOG(WARNING) << err_msg;
- return Status::InternalError(err_msg);
- }
- _tablet_writers.emplace(tablet, writer);
- _s_tablet_writer_count += tablets.size();
- }
- }
- }
- return Status::OK();
-}
-
-// The method will called by open partition rpc.
-Status TabletsChannel::open_all_writers_for_partition(const
OpenPartitionRequest& request) {
- std::vector<SlotDescriptor*>* index_slots = nullptr;
- int32_t schema_hash = 0;
- for (auto& index : _schema->indexes()) {
- if (index->index_id == _index_id) {
- index_slots = &index->slots;
- schema_hash = index->schema_hash;
- break;
- }
- }
- if (index_slots == nullptr) {
- return Status::InternalError("unknown index id, key={}",
_key.to_string());
- }
- for (auto& tablet : request.tablets()) {
- WriteRequest wrequest;
- wrequest.index_id = request.index_id();
- wrequest.tablet_id = tablet.tablet_id();
- wrequest.schema_hash = schema_hash;
- wrequest.txn_id = _txn_id;
- wrequest.partition_id = tablet.partition_id();
- wrequest.load_id = request.id();
- wrequest.tuple_desc = _tuple_desc;
- wrequest.slots = index_slots;
- wrequest.is_high_priority = _is_high_priority;
- wrequest.table_schema_param = _schema;
-
- {
- std::lock_guard<SpinLock> l(_tablet_writers_lock);
-
- if (_tablet_writers.find(tablet.tablet_id()) ==
_tablet_writers.end()) {
- DeltaWriter* writer = nullptr;
- auto st = DeltaWriter::open(&wrequest, &writer, _profile,
_load_id);
- if (!st.ok()) {
- auto err_msg = fmt::format(
- "open delta writer failed, tablet_id={}"
- ", txn_id={}, partition_id={}, err={}",
- tablet.tablet_id(), _txn_id,
tablet.partition_id(), st.to_string());
- LOG(WARNING) << err_msg;
- return Status::InternalError(err_msg);
- }
- _tablet_writers.emplace(tablet.tablet_id(), writer);
- _s_tablet_writer_count += _tablet_writers.size();
- }
- }
- }
- return Status::OK();
-}
-
Status TabletsChannel::cancel() {
std::lock_guard<std::mutex> l(_lock);
if (_state == kFinished) {
@@ -525,25 +412,12 @@ Status TabletsChannel::add_batch(const
PTabletWriterAddBlockRequest& request,
std::function<Status(DeltaWriter * writer)>
write_func) {
google::protobuf::RepeatedPtrField<PTabletError>* tablet_errors =
response->mutable_tablet_errors();
- bool open_partition_flag = false;
{
std::lock_guard<SpinLock> l(_tablet_writers_lock);
auto tablet_writer_it = _tablet_writers.find(tablet_id);
if (tablet_writer_it == _tablet_writers.end()) {
- if (!config::enable_lazy_open_partition) {
- return Status::InternalError("unknown tablet to append
data, tablet={}",
- tablet_id);
- } else {
- open_partition_flag = true;
- }
+ return Status::InternalError("unknown tablet to append data,
tablet={}");
}
- }
- if (open_partition_flag) {
- RETURN_IF_ERROR(_open_all_writers_for_partition(tablet_id,
request));
- }
- {
- std::lock_guard<SpinLock> l(_tablet_writers_lock);
- auto tablet_writer_it = _tablet_writers.find(tablet_id);
Status st = write_func(tablet_writer_it->second);
if (!st.ok()) {
auto err_msg =
diff --git a/be/src/runtime/tablets_channel.h b/be/src/runtime/tablets_channel.h
index 4e523a1acf..f728e8ab9b 100644
--- a/be/src/runtime/tablets_channel.h
+++ b/be/src/runtime/tablets_channel.h
@@ -90,9 +90,6 @@ public:
Status open(const PTabletWriterOpenRequest& request);
- // Open specific partition all writers
- Status open_all_writers_for_partition(const OpenPartitionRequest& request);
-
// no-op when this channel has been closed or cancelled
Status add_batch(const PTabletWriterAddBlockRequest& request,
PTabletWriterAddBlockResult* response);
@@ -127,9 +124,6 @@ private:
template <typename Request>
Status _get_current_seq(int64_t& cur_seq, const Request& request);
- template <typename TabletWriterAddRequest>
- Status _open_all_writers_for_partition(const int64_t& tablet_id,
- const TabletWriterAddRequest&
request);
// open all writer
Status _open_all_writers(const PTabletWriterOpenRequest& request);
diff --git a/be/src/service/internal_service.cpp
b/be/src/service/internal_service.cpp
index 745de1bb71..c38416a593 100644
--- a/be/src/service/internal_service.cpp
+++ b/be/src/service/internal_service.cpp
@@ -250,29 +250,6 @@ void
PInternalServiceImpl::tablet_writer_open(google::protobuf::RpcController* c
}
}
-void PInternalServiceImpl::open_partition(google::protobuf::RpcController*
controller,
- const OpenPartitionRequest* request,
- OpenPartitionResult* response,
- google::protobuf::Closure* done) {
- bool ret = _light_work_pool.try_offer([this, request, response, done]() {
- VLOG_RPC << "partition open"
- << ", index_id=" << request->index_id();
- brpc::ClosureGuard closure_guard(done);
- auto st = _exec_env->load_channel_mgr()->open_partition(*request);
- if (!st.ok()) {
- LOG(WARNING) << "load channel open failed, message=" << st
- << ", index_ids=" << request->index_id();
- }
- st.to_protobuf(response->mutable_status());
- });
- if (!ret) {
- LOG(WARNING) << "fail to offer request to the work pool";
- brpc::ClosureGuard closure_guard(done);
- response->mutable_status()->set_status_code(TStatusCode::CANCELLED);
- response->mutable_status()->add_error_msgs("fail to offer request to
the work pool");
- }
-}
-
void PInternalServiceImpl::exec_plan_fragment(google::protobuf::RpcController*
controller,
const PExecPlanFragmentRequest*
request,
PExecPlanFragmentResult*
response,
diff --git a/be/src/service/internal_service.h
b/be/src/service/internal_service.h
index d6e825497e..0d568e2b84 100644
--- a/be/src/service/internal_service.h
+++ b/be/src/service/internal_service.h
@@ -86,10 +86,6 @@ public:
PTabletWriterOpenResult* response,
google::protobuf::Closure* done) override;
- void open_partition(google::protobuf::RpcController* controller,
- const OpenPartitionRequest* request,
OpenPartitionResult* response,
- google::protobuf::Closure* done) override;
-
void tablet_writer_add_block(google::protobuf::RpcController* controller,
const PTabletWriterAddBlockRequest* request,
PTabletWriterAddBlockResult* response,
diff --git a/be/src/vec/sink/vtablet_sink.cpp b/be/src/vec/sink/vtablet_sink.cpp
index afaebcb9ca..519597943a 100644
--- a/be/src/vec/sink/vtablet_sink.cpp
+++ b/be/src/vec/sink/vtablet_sink.cpp
@@ -97,51 +97,6 @@ class TExpr;
namespace stream_load {
-class OpenPartitionClosure : public google::protobuf::Closure {
-public:
- OpenPartitionClosure(VNodeChannel* vnode_channel, IndexChannel*
index_channel,
- int64_t partition_id)
- : vnode_channel(vnode_channel),
- index_channel(index_channel),
- partition_id(partition_id) {};
-
- ~OpenPartitionClosure() override = default;
-
- void Run() override {
- SCOPED_TRACK_MEMORY_TO_UNKNOWN();
- DCHECK(packet_in_flight);
- auto open_partition_failed = [this]() {
- std::stringstream ss;
- ss << "failed to open partition, error=" <<
berror(this->cntl.ErrorCode())
- << ", error_text=" << this->cntl.ErrorText();
- LOG(WARNING) << ss.str() << " " << vnode_channel->channel_info();
- vnode_channel->cancel("Open partition error");
- index_channel->mark_as_failed(vnode_channel->node_id(),
vnode_channel->host(),
- fmt::format("{}, open failed, err:
{}",
-
vnode_channel->channel_info(), ss.str()),
- -1);
- };
- if (cntl.Failed()) {
- open_partition_failed();
- } else {
- Status status(Status::create(result.status()));
- if (!status.ok()) {
- open_partition_failed();
- }
- }
- packet_in_flight = false;
- }
-
- void join() { brpc::Join(cntl.call_id()); }
-
- brpc::Controller cntl;
- OpenPartitionResult result;
- VNodeChannel* vnode_channel;
- IndexChannel* index_channel;
- int64_t partition_id;
- std::atomic<bool> packet_in_flight {false};
-};
-
IndexChannel::~IndexChannel() {}
Status IndexChannel::init(RuntimeState* state, const
std::vector<TTabletWithPartition>& tablets) {
@@ -526,54 +481,6 @@ Status VNodeChannel::open_wait() {
return status;
}
-void VNodeChannel::open_partition(int64_t partition_id) {
- MonotonicStopWatch lazy_open_timeout_watch;
- lazy_open_timeout_watch.start();
- SCOPED_CONSUME_MEM_TRACKER(_node_channel_tracker.get());
- OpenPartitionRequest request;
- auto load_id = std::make_shared<PUniqueId>(_parent->_load_id);
- request.set_allocated_id(load_id.get());
- request.set_index_id(_index_channel->_index_id);
- for (auto& tablet : _all_tablets) {
- if (partition_id == tablet.partition_id) {
- auto ptablet = request.add_tablets();
- ptablet->set_partition_id(tablet.partition_id);
- ptablet->set_tablet_id(tablet.tablet_id);
- }
- }
-
- auto open_partition_closure =
- std::make_unique<OpenPartitionClosure>(this, _index_channel,
partition_id);
-
- int remain_ms = _rpc_timeout_ms - lazy_open_timeout_watch.elapsed_time();
- if (UNLIKELY(remain_ms < config::min_load_rpc_timeout_ms)) {
- remain_ms = config::min_load_rpc_timeout_ms;
- }
- open_partition_closure->cntl.set_timeout_ms(remain_ms);
- open_partition_closure->packet_in_flight = true;
- _stub->open_partition(&open_partition_closure.get()->cntl, &request,
- &open_partition_closure.get()->result,
open_partition_closure.get());
-
- _open_partition_closures.insert(std::move(open_partition_closure));
-
- request.release_id();
-}
-
-void VNodeChannel::open_partition_wait() {
- for (auto& open_partition_closure : _open_partition_closures) {
- open_partition_closure->join();
- }
-}
-
-bool VNodeChannel::open_partition_finished() const {
- for (auto& open_partition_closure : _open_partition_closures) {
- if (open_partition_closure->packet_in_flight) {
- return false;
- }
- }
- return true;
-}
-
Status VNodeChannel::add_block(vectorized::Block* block, const Payload*
payload, bool is_append) {
SCOPED_CONSUME_MEM_TRACKER(_node_channel_tracker.get());
if (payload->second.empty()) {
@@ -1183,25 +1090,6 @@ Status VOlapTableSink::open(RuntimeState* state) {
return Status::OK();
}
-void VOlapTableSink::_open_partition(const VOlapTablePartition* partition) {
- const auto& id = partition->id;
- auto it = _opened_partitions.find(id);
- if (it == _opened_partitions.end()) {
- _opened_partitions.insert(id);
- fmt::memory_buffer buf;
- for (int j = 0; j < partition->indexes.size(); ++j) {
- fmt::format_to(buf, "index id:{}", partition->indexes[j].index_id);
- for (const auto& tid : partition->indexes[j].tablets) {
- auto it = _channels[j]->_channels_by_tablet.find(tid);
- for (const auto& channel : it->second) {
- channel->open_partition(partition->id);
- }
- }
- }
- VLOG_DEBUG << "list of lazy open index id = " << fmt::to_string(buf);
- }
-}
-
void VOlapTableSink::_send_batch_process() {
SCOPED_TIMER(_non_blocking_send_timer);
SCOPED_ATTACH_TASK(_state);
@@ -1273,9 +1161,6 @@ Status
VOlapTableSink::_single_partition_generate(RuntimeState* state, vectorize
if (is_continue) {
continue;
}
- if (config::enable_lazy_open_partition) {
- _open_partition(partition);
- }
break;
}
for (int j = 0; j < partition->indexes.size(); ++j) {
@@ -1360,11 +1245,6 @@ Status VOlapTableSink::send(RuntimeState* state,
vectorized::Block* input_block,
}
// each row
_generate_row_distribution_payload(channel_to_payload, partition,
tablet_index, i, 1);
- // open partition
- if (config::enable_lazy_open_partition) {
- // aysnc open operation,don't block send operation
- _open_partition(partition);
- }
}
}
_row_distribution_watch.stop();
@@ -1445,21 +1325,6 @@ void VOlapTableSink::try_close(RuntimeState* state,
Status exec_status) {
return;
}
- if (config::enable_lazy_open_partition && !_open_partition_done) {
- // open_partition_finished must be before mark_close
- bool open_partition_done = true;
- for (const auto& index_channel : _channels) {
- index_channel->for_each_node_channel(
- [&open_partition_done](const
std::shared_ptr<VNodeChannel>& ch) {
- open_partition_done &= ch->open_partition_finished();
- });
- }
- if (!open_partition_done) {
- return;
- }
- _open_partition_done = true;
- }
-
SCOPED_TIMER(_close_timer);
Status status = exec_status;
if (status.ok()) {
@@ -1524,14 +1389,6 @@ Status VOlapTableSink::close(RuntimeState* state, Status
exec_status) {
SCOPED_TIMER(_close_timer);
SCOPED_TIMER(_profile->total_time_counter());
- if (config::enable_lazy_open_partition) {
- for (const auto& index_channel : _channels) {
- index_channel->for_each_node_channel(
- [](const std::shared_ptr<VNodeChannel>& ch) {
ch->open_partition_wait(); });
- }
- _open_partition_done = true;
- }
-
try_close(state, exec_status);
// If _close_status is not ok, all nodes have been canceled in try_close.
if (_close_status.ok()) {
diff --git a/be/src/vec/sink/vtablet_sink.h b/be/src/vec/sink/vtablet_sink.h
index ea6b857b8b..38c9e7c325 100644
--- a/be/src/vec/sink/vtablet_sink.h
+++ b/be/src/vec/sink/vtablet_sink.h
@@ -83,7 +83,6 @@ namespace stream_load {
class OlapTableBlockConvertor;
class OlapTabletFinder;
-class OpenPartitionClosure;
// The counter of add_batch rpc of a single node
struct AddBatchCounter {
@@ -227,16 +226,10 @@ public:
void open();
- void open_partition(int64_t partition_id);
-
Status init(RuntimeState* state);
Status open_wait();
- void open_partition_wait();
-
- bool open_partition_finished() const;
-
Status add_block(vectorized::Block* block, const Payload* payload, bool
is_append = false);
int try_send_and_fetch_status(RuntimeState* state,
@@ -353,7 +346,6 @@ protected:
std::shared_ptr<PBackendService_Stub> _stub = nullptr;
RefCountClosure<PTabletWriterOpenResult>* _open_closure = nullptr;
- std::unordered_set<std::unique_ptr<OpenPartitionClosure>>
_open_partition_closures;
std::vector<TTabletWithPartition> _all_tablets;
// map from tablet_id to node_id where slave replicas locate in
@@ -518,8 +510,6 @@ private:
ChannelDistributionPayload&
channel_to_payload,
size_t num_rows, bool has_filtered_rows);
- void _open_partition(const VOlapTablePartition* partition);
-
Status _cancel_channel_and_check_intolerable_failure(Status status, const
std::string& err_msg,
const
std::shared_ptr<IndexChannel> ich,
const
std::shared_ptr<VNodeChannel> nch);
@@ -617,8 +607,6 @@ private:
vectorized::VExprContextSPtrs _output_vexpr_ctxs;
RuntimeState* _state = nullptr;
-
- std::unordered_set<int64_t> _opened_partitions;
};
} // namespace stream_load
diff --git a/be/test/vec/exec/vtablet_sink_test.cpp
b/be/test/vec/exec/vtablet_sink_test.cpp
index aa995e671a..15ee6b79a2 100644
--- a/be/test/vec/exec/vtablet_sink_test.cpp
+++ b/be/test/vec/exec/vtablet_sink_test.cpp
@@ -302,14 +302,6 @@ public:
status.to_protobuf(response->mutable_status());
}
- void open_partition(google::protobuf::RpcController* controller,
- const OpenPartitionRequest* request,
OpenPartitionResult* response,
- google::protobuf::Closure* done) override {
- brpc::ClosureGuard done_guard(done);
- Status status;
- status.to_protobuf(response->mutable_status());
- }
-
void tablet_writer_add_block(google::protobuf::RpcController* controller,
const PTabletWriterAddBlockRequest* request,
PTabletWriterAddBlockResult* response,
diff --git a/docs/en/docs/admin-manual/config/be-config.md
b/docs/en/docs/admin-manual/config/be-config.md
index bd6bc961d7..6edaa7bb86 100644
--- a/docs/en/docs/admin-manual/config/be-config.md
+++ b/docs/en/docs/admin-manual/config/be-config.md
@@ -1275,12 +1275,6 @@ BaseCompaction:546859:
* Description: Used to ignore brpc error '[E1011]The server is overcrowded'
when writing data.
* Default value: false
-#### `enable_lazy_open_partition`
-
-* Type: bool
-* Description: When importing, most partitions may not need to be written, and
lazy opening can be used to only open the partitions that need to be
written.When there is mixed deployment in the upgraded version, it needs to be
set to false.
-* Default value: false
-
#### `streaming_load_rpc_max_alive_time_sec`
* Description: The lifetime of TabletsChannel. If the channel does not receive
any data at this time, the channel will be deleted.
diff --git a/docs/zh-CN/docs/admin-manual/config/be-config.md
b/docs/zh-CN/docs/admin-manual/config/be-config.md
index bc4a2cbd1f..d41b2e4cbf 100644
--- a/docs/zh-CN/docs/admin-manual/config/be-config.md
+++ b/docs/zh-CN/docs/admin-manual/config/be-config.md
@@ -1291,12 +1291,6 @@ BaseCompaction:546859:
- 当遇到'[E1011]The server is
overcrowded'的错误时，可以调整配置项`brpc_socket_max_unwritten_bytes`，但这个配置项不能动态调整。所以可通过设置此项为`true`来临时避免写失败。注意，此配置项只影响写流程，其他的rpc请求依旧会检查是否overcrowded。
* 默认值：false
-#### `enable_lazy_open_partition`
-
-* 类型：bool
-*
描述：导入时大部分partition可能都不需要写入，可以使用延迟打开的方式只打开需要写入的partition。升级版本出现混合部署的时候，需要设置为false。
-* 默认值：false
-
#### `streaming_load_rpc_max_alive_time_sec`
* 描述：TabletsChannel 的存活时间。如果此时通道没有收到任何数据， 通道将被删除。
diff --git a/gensrc/proto/internal_service.proto
b/gensrc/proto/internal_service.proto
index 172e5ad4ef..148efec1a2 100644
--- a/gensrc/proto/internal_service.proto
+++ b/gensrc/proto/internal_service.proto
@@ -93,16 +93,6 @@ message PTabletWriterOpenResult {
required PStatus status = 1;
};
-message OpenPartitionRequest {
- required PUniqueId id = 1;
- required int64 index_id = 2;
- repeated PTabletWithPartition tablets = 3;
-}
-
-message OpenPartitionResult {
- required PStatus status = 1;
-}
-
// add batch to tablet writer
message PTabletWriterAddBatchRequest {
required PUniqueId id = 1;
@@ -689,7 +679,6 @@ service PBackendService {
rpc cancel_plan_fragment(PCancelPlanFragmentRequest) returns
(PCancelPlanFragmentResult);
rpc fetch_data(PFetchDataRequest) returns (PFetchDataResult);
rpc tablet_writer_open(PTabletWriterOpenRequest) returns
(PTabletWriterOpenResult);
- rpc open_partition(OpenPartitionRequest) returns (OpenPartitionResult);
rpc tablet_writer_add_block(PTabletWriterAddBlockRequest) returns
(PTabletWriterAddBlockResult);
rpc tablet_writer_add_block_by_http(PEmptyRequest) returns
(PTabletWriterAddBlockResult);
rpc tablet_writer_cancel(PTabletWriterCancelRequest) returns
(PTabletWriterCancelResult);
diff --git a/regression-test/suites/version_p0/version_p0.groovy
b/regression-test/suites/version_p0/version_p0.groovy
new file mode 100644
index 0000000000..b72f231eb6
--- /dev/null
+++ b/regression-test/suites/version_p0/version_p0.groovy
@@ -0,0 +1,45 @@
+// 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.
+
+suite("version_p0", "p0") {
+ def tableName1 = "test_partition_version"
+ sql """ DROP TABLE IF EXISTS ${tableName1} """
+ sql """
+ CREATE TABLE ${tableName1} (
+ `TIME_STAMP` datev2 NOT NULL COMMENT '采集日期'
+ ) ENGINE=OLAP
+ DUPLICATE KEY(`TIME_STAMP`)
+ COMMENT 'OLAP'
+ PARTITION BY RANGE(`TIME_STAMP`)
+ (
+ PARTITION p20221220 VALUES [('2022-12-20'), ('2022-12-21')),
+ PARTITION p20221221 VALUES [('2022-12-21'), ('2022-12-22'))
+ )
+ DISTRIBUTED BY HASH(`TIME_STAMP`) BUCKETS 10
+ PROPERTIES
+ (
+ "replication_allocation" = "tag.location.default: 1"
+ );
+ """
+
+ sql """ insert into ${tableName1} values ('2022-12-20'); """
+ def res = sql """ show partitions from ${tableName1}; """
+
+ // check written partition version is higher
+ assertEquals(res[0][2].toString(), "2")
+ assertEquals(res[1][2].toString(), "1")
+}
\ No newline at end of file
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
