This is an automated email from the ASF dual-hosted git repository.
eldenmoon 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 4b07b5f5941 [refactor](lazy materialization) Remove legacy two-phase
row ID read path (#66918)
4b07b5f5941 is described below
commit 4b07b5f5941b0c4d93a5f837c45b465d371f0f15
Author: lihangyu <[email protected]>
AuthorDate: Thu Aug 20 15:36:14 2026 +0800
[refactor](lazy materialization) Remove legacy two-phase row ID read path
(#66918)
Related PR: #62917
Problem Summary: The legacy `__DORIS_ROWID_COL__` ResultSink two-phase
fetch path remained after TopN lazy materialization moved to
`__DORIS_GLOBAL_ROWID_COL__` and `multiget_data_v2`. This PR removes the
unused FE and BE execution, rowset-retention, and row-at-a-time storage
read code.
The old protobuf and Thrift definitions retain their original field IDs
and are marked deprecated. Deprecated `multiget_data` calls now return
`NotSupported`.
---
be/src/cloud/cloud_storage_engine.cpp | 8 +-
be/src/common/config.cpp | 1 +
be/src/common/config.h | 1 +
be/src/common/consts.h | 1 -
be/src/exec/operator/materialization_opertor.cpp | 5 +
be/src/exec/operator/result_sink_operator.cpp | 28 +-
be/src/exec/operator/result_sink_operator.h | 5 -
be/src/exec/rowid_fetcher.cpp | 356 ---------------------
be/src/exec/rowid_fetcher.h | 39 ---
be/src/exec/scan/olap_scanner.cpp | 15 -
be/src/service/internal_service.cpp | 23 +-
be/src/service/internal_service.h | 2 -
be/src/storage/olap_server.cpp | 8 +-
be/src/storage/rowset/rowset.h | 10 -
be/src/storage/schema.h | 3 +-
be/src/storage/segment/column_reader.h | 46 ---
be/src/storage/segment/segment_iterator.cpp | 6 -
be/src/storage/storage_engine.cpp | 33 +-
be/src/storage/storage_engine.h | 13 +-
be/src/storage/utils.h | 19 --
.../variant/variant_column_writer_reader_test.cpp | 115 -------
.../main/java/org/apache/doris/catalog/Column.java | 1 -
.../java/org/apache/doris/analysis/SortInfo.java | 9 -
.../org/apache/doris/planner/OlapScanNode.java | 18 --
.../java/org/apache/doris/planner/ResultSink.java | 17 -
.../java/org/apache/doris/planner/SortNode.java | 12 +-
gensrc/proto/internal_service.proto | 36 ++-
gensrc/thrift/DataSinks.thrift | 3 +-
gensrc/thrift/PlanNodes.thrift | 2 +-
29 files changed, 53 insertions(+), 782 deletions(-)
diff --git a/be/src/cloud/cloud_storage_engine.cpp
b/be/src/cloud/cloud_storage_engine.cpp
index a8cebc7449d..c8bd87a828b 100644
--- a/be/src/cloud/cloud_storage_engine.cpp
+++ b/be/src/cloud/cloud_storage_engine.cpp
@@ -387,10 +387,10 @@ Status
CloudStorageEngine::start_bg_threads(std::shared_ptr<WorkloadGroup> wg_sp
LOG(INFO) << "sync tablets thread started";
RETURN_IF_ERROR(Thread::create(
- "CloudStorageEngine", "evict_querying_rowset_thread",
- [this]() { this->_evict_quring_rowset_thread_callback(); },
- &_evict_quering_rowset_thread));
- LOG(INFO) << "evict quering thread started";
+ "CloudStorageEngine", "id_file_map_gc_thread",
+ [this]() { this->_gc_expired_id_file_map_thread_callback(); },
+ &_id_file_map_gc_thread));
+ LOG(INFO) << "id file map gc thread started";
// add calculate tablet delete bitmap task thread pool
RETURN_IF_ERROR(ThreadPoolBuilder("TabletCalDeleteBitmapThreadPool")
diff --git a/be/src/common/config.cpp b/be/src/common/config.cpp
index 5df2edfa20c..9902818d792 100644
--- a/be/src/common/config.cpp
+++ b/be/src/common/config.cpp
@@ -375,6 +375,7 @@ DEFINE_mInt32(tablet_lookup_cache_stale_sweep_time_sec,
"30");
DEFINE_mInt32(point_query_row_cache_stale_sweep_time_sec, "300");
DEFINE_mInt32(disk_stat_monitor_interval, "5");
DEFINE_mInt32(unused_rowset_monitor_interval, "30");
+// Legacy name retained for compatibility; controls GLOBAL_ROWID_COL file-map
GC.
DEFINE_mInt32(quering_rowsets_evict_interval, "30");
DEFINE_String(storage_root_path, "${DORIS_HOME}/storage");
DEFINE_mString(broken_storage_path, "");
diff --git a/be/src/common/config.h b/be/src/common/config.h
index 1f53b060c3e..2b5197a484c 100644
--- a/be/src/common/config.h
+++ b/be/src/common/config.h
@@ -460,6 +460,7 @@ DECLARE_mInt32(tablet_lookup_cache_stale_sweep_time_sec);
DECLARE_mInt32(point_query_row_cache_stale_sweep_time_sec);
DECLARE_mInt32(disk_stat_monitor_interval);
DECLARE_mInt32(unused_rowset_monitor_interval);
+// Legacy name retained for compatibility; controls GLOBAL_ROWID_COL file-map
GC.
DECLARE_mInt32(quering_rowsets_evict_interval);
DECLARE_String(storage_root_path);
DECLARE_mString(broken_storage_path);
diff --git a/be/src/common/consts.h b/be/src/common/consts.h
index 77d2acadf48..3e5d3b4c646 100644
--- a/be/src/common/consts.h
+++ b/be/src/common/consts.h
@@ -24,7 +24,6 @@ namespace BeConsts {
const std::string CSV = "csv";
const std::string CSV_WITH_NAMES = "csv_with_names";
const std::string CSV_WITH_NAMES_AND_TYPES = "csv_with_names_and_types";
-const std::string ROWID_COL = "__DORIS_ROWID_COL__";
const std::string GLOBAL_ROWID_COL = "__DORIS_GLOBAL_ROWID_COL__";
const std::string ROW_STORE_COL = "__DORIS_ROW_STORE_COL__";
const std::string DYNAMIC_COLUMN_NAME = "__DORIS_DYNAMIC_COL__";
diff --git a/be/src/exec/operator/materialization_opertor.cpp
b/be/src/exec/operator/materialization_opertor.cpp
index 5628faceecc..5370dcf330c 100644
--- a/be/src/exec/operator/materialization_opertor.cpp
+++ b/be/src/exec/operator/materialization_opertor.cpp
@@ -36,12 +36,17 @@
#include "runtime/workload_group/workload_group.h"
#include "util/brpc_client_cache.h"
#include "util/brpc_closure.h"
+#include "util/defer_op.h"
#include "util/pretty_printer.h"
namespace doris {
namespace {
+void fetch_callback(bthread::CountdownEvent* counter) {
+ Defer defer([&] { counter->signal(); });
+}
+
constexpr const char* TOPN_LAZY_MAT_PHASE2_PER_BACKEND =
"TopNLazyMaterializationSecondPhasePerBackend";
constexpr const char* TOPN_LAZY_MAT_PHASE2_PER_BACKEND_ROWS_READ =
diff --git a/be/src/exec/operator/result_sink_operator.cpp
b/be/src/exec/operator/result_sink_operator.cpp
index 12611b36aaa..c523542aea2 100644
--- a/be/src/exec/operator/result_sink_operator.cpp
+++ b/be/src/exec/operator/result_sink_operator.cpp
@@ -24,7 +24,6 @@
#include "common/config.h"
#include "exec/operator/operator.h"
-#include "exec/rowid_fetcher.h"
#include "exec/sink/writer/varrow_flight_result_writer.h"
#include "exec/sink/writer/vmysql_result_writer.h"
#include "exprs/vexpr.h"
@@ -40,7 +39,6 @@ Status ResultSinkLocalState::init(RuntimeState* state,
LocalSinkStateInfo& info)
RETURN_IF_ERROR(Base::init(state, info));
SCOPED_TIMER(exec_time_counter());
SCOPED_TIMER(_init_timer);
- _fetch_row_id_timer = ADD_TIMER(custom_profile(), "FetchRowIdTime");
_write_data_timer = ADD_TIMER(custom_profile(), "WriteDataTime");
static const std::string timer_name = "WaitForDependencyTime";
_wait_for_dependency_timer = ADD_TIMER_WITH_LEVEL(custom_profile(),
timer_name, 1);
@@ -106,8 +104,7 @@ ResultSinkOperatorX::ResultSinkOperatorX(int operator_id,
int node_id,
?
config::arrow_flight_result_sink_buffer_size_rows
: RESULT_SINK_BUFFER_SIZE),
_row_desc(row_desc),
- _t_output_expr(t_output_expr),
- _fetch_option(sink.fetch_option) {
+ _t_output_expr(t_output_expr) {
_name = "ResultSink";
}
@@ -138,33 +135,10 @@ Status ResultSinkOperatorX::sink_impl(RuntimeState*
state, Block* block, bool eo
auto& local_state = get_local_state(state);
SCOPED_TIMER(local_state.exec_time_counter());
COUNTER_UPDATE(local_state.rows_input_counter(), (int64_t)block->rows());
- if (_fetch_option.use_two_phase_fetch && block->rows() > 0) {
- SCOPED_TIMER(local_state._fetch_row_id_timer);
- RETURN_IF_ERROR(_second_phase_fetch_data(state, block));
- }
{
SCOPED_TIMER(local_state._write_data_timer);
RETURN_IF_ERROR(local_state._writer->write(state, *block));
}
- if (_fetch_option.use_two_phase_fetch) {
- // Block structure may be changed by calling
_second_phase_fetch_data().
- // So we should clear block in case of unmatched columns
- block->clear();
- }
- return Status::OK();
-}
-
-Status ResultSinkOperatorX::_second_phase_fetch_data(RuntimeState* state,
Block* final_block) {
- auto row_id_col = final_block->get_by_position(final_block->columns() - 1);
- CHECK(row_id_col.name == BeConsts::ROWID_COL);
- auto* tuple_desc = _row_desc.tuple_descriptors()[0];
- FetchOption fetch_option;
- fetch_option.desc = tuple_desc;
- fetch_option.t_fetch_opt = _fetch_option;
- fetch_option.runtime_state = state;
- RowIDFetcher id_fetcher(fetch_option);
- RETURN_IF_ERROR(id_fetcher.init());
- RETURN_IF_ERROR(id_fetcher.fetch(row_id_col.column, final_block));
return Status::OK();
}
diff --git a/be/src/exec/operator/result_sink_operator.h
b/be/src/exec/operator/result_sink_operator.h
index 7ff4b18c9b2..690a99495ca 100644
--- a/be/src/exec/operator/result_sink_operator.h
+++ b/be/src/exec/operator/result_sink_operator.h
@@ -149,7 +149,6 @@ private:
std::shared_ptr<ResultBlockBufferBase> _sender = nullptr;
std::shared_ptr<ResultWriter> _writer = nullptr;
- RuntimeProfile::Counter* _fetch_row_id_timer = nullptr;
RuntimeProfile::Counter* _write_data_timer = nullptr;
};
@@ -164,7 +163,6 @@ public:
private:
friend class ResultSinkLocalState;
- Status _second_phase_fetch_data(RuntimeState* state, Block* final_block);
const TResultSinkType::type _sink_type;
const int _result_sink_buffer_size_rows;
// set file options when sink type is FILE
@@ -177,9 +175,6 @@ private:
const std::vector<TExpr>& _t_output_expr;
VExprContextSPtrs _output_vexpr_ctxs;
- // for fetch data by rowids
- const TFetchOption _fetch_option;
-
std::shared_ptr<ResultBlockBufferBase> _sender = nullptr;
};
diff --git a/be/src/exec/rowid_fetcher.cpp b/be/src/exec/rowid_fetcher.cpp
index e9f0f01bdab..95544b4422e 100644
--- a/be/src/exec/rowid_fetcher.cpp
+++ b/be/src/exec/rowid_fetcher.cpp
@@ -17,8 +17,6 @@
#include "exec/rowid_fetcher.h"
-#include <brpc/callback.h>
-#include <butil/endpoint.h>
#include <fmt/format.h>
#include <gen_cpp/data.pb.h>
#include <gen_cpp/internal_service.pb.h>
@@ -37,19 +35,13 @@
#include <utility>
#include <vector>
-#include "bthread/countdown_event.h"
#include "common/config.h"
-#include "common/consts.h"
#include "common/exception.h"
#include "common/signal_handler.h"
-#include "core/assert_cast.h"
#include "core/block/block.h" // Block
#include "core/column/column.h"
-#include "core/column/column_nullable.h"
-#include "core/column/column_string.h"
#include "core/data_type/data_type_struct.h"
#include "core/data_type_serde/data_type_serde.h"
-#include "core/string_ref.h"
#include "exec/scan/file_scanner.h"
#include "format/orc/vorc_reader.h"
#include "format/parquet/vparquet_reader.h"
@@ -63,13 +55,9 @@
#include "storage/olap_common.h"
#include "storage/rowset/beta_rowset.h"
#include "storage/segment/column_reader.h"
-#include "storage/storage_engine.h"
#include "storage/tablet/tablet_fwd.h"
#include "storage/tablet/tablet_schema.h"
-#include "storage/tablet_info.h" // DorisNodesInfo
#include "storage/utils.h"
-#include "util/brpc_client_cache.h" // BrpcClientCache
-#include "util/defer_op.h"
#include "util/jsonb/serialize.h"
namespace doris {
@@ -91,194 +79,6 @@ void set_topn_lazy_materialization_file_cache_stats(
} // namespace
-Status RowIDFetcher::init() {
- DorisNodesInfo nodes_info;
- nodes_info.setNodes(_fetch_option.t_fetch_opt.nodes_info);
- for (auto [node_id, node_info] : nodes_info.nodes_info()) {
- auto client =
ExecEnv::GetInstance()->brpc_internal_client_cache()->get_client(
- node_info.host, node_info.brpc_port);
- if (!client) {
- LOG(WARNING) << "Get rpc stub failed, host=" << node_info.host
- << ", port=" << node_info.brpc_port;
- return Status::InternalError("RowIDFetcher failed to init rpc
client, host={}, port={}",
- node_info.host, node_info.brpc_port);
- }
- _stubs.push_back(client);
- }
- return Status::OK();
-}
-
-PMultiGetRequest RowIDFetcher::_init_fetch_request(const ColumnString&
row_locs) const {
- PMultiGetRequest mget_req;
- _fetch_option.desc->to_protobuf(mget_req.mutable_desc());
- for (SlotDescriptor* slot : _fetch_option.desc->slots()) {
- // ignore rowid
- if (slot->col_name() == BeConsts::ROWID_COL) {
- continue;
- }
- slot->to_protobuf(mget_req.add_slots());
- }
- for (size_t i = 0; i < row_locs.size(); ++i) {
- PRowLocation row_loc;
- StringRef row_id_rep = row_locs.get_data_at(i);
- // TODO: When transferring data between machines with different byte
orders (endianness),
- // not performing proper handling may lead to issues in parsing and
exchanging the data.
- auto location = reinterpret_cast<const
GlobalRowLoacation*>(row_id_rep.data);
- row_loc.set_tablet_id(location->tablet_id);
- row_loc.set_rowset_id(location->row_location.rowset_id.to_string());
- row_loc.set_segment_id(location->row_location.segment_id);
- row_loc.set_ordinal_id(location->row_location.row_id);
- *mget_req.add_row_locs() = std::move(row_loc);
- }
- // Set column desc
- for (const TColumn& tcolumn : _fetch_option.t_fetch_opt.column_desc) {
- TabletColumn column(tcolumn);
- column.to_schema_pb(mget_req.add_column_desc());
- }
- PUniqueId& query_id = *mget_req.mutable_query_id();
- query_id.set_hi(_fetch_option.runtime_state->query_id().hi);
- query_id.set_lo(_fetch_option.runtime_state->query_id().lo);
-
mget_req.set_be_exec_version(_fetch_option.runtime_state->be_exec_version());
- mget_req.set_fetch_row_store(_fetch_option.t_fetch_opt.fetch_row_store);
- return mget_req;
-}
-
-Status RowIDFetcher::_merge_rpc_results(const PMultiGetRequest& request,
- const std::vector<PMultiGetResponse>&
rsps,
- const std::vector<brpc::Controller>&
cntls,
- Block* output_block,
- std::vector<PRowLocation>* rows_id)
const {
- output_block->clear();
- for (const auto& cntl : cntls) {
- if (cntl.Failed()) {
- LOG(WARNING) << "Failed to fetch meet rpc error:" <<
cntl.ErrorText()
- << ", host:" << cntl.remote_side();
- return Status::InternalError(cntl.ErrorText());
- }
- }
- DataTypeSerDeSPtrs serdes;
- std::unordered_map<uint32_t, uint32_t> col_uid_to_idx;
- std::vector<std::string> default_values;
- default_values.resize(_fetch_option.desc->slots().size());
- auto merge_function = [&](const PMultiGetResponse& resp) {
- Status st(Status::create(resp.status()));
- if (!st.ok()) {
- LOG(WARNING) << "Failed to fetch " << st.to_string();
- return st;
- }
- for (const PRowLocation& row_id : resp.row_locs()) {
- rows_id->push_back(row_id);
- }
- // Merge binary rows
- if (request.fetch_row_store()) {
- CHECK(resp.row_locs().size() == resp.binary_row_data_size());
- if (output_block->is_empty_column()) {
- *output_block = Block(_fetch_option.desc->slots(), 1);
- }
- if (serdes.empty() && col_uid_to_idx.empty()) {
- serdes = create_data_type_serdes(_fetch_option.desc->slots());
- for (int i = 0; i < _fetch_option.desc->slots().size(); ++i) {
-
col_uid_to_idx[_fetch_option.desc->slots()[i]->col_unique_id()] = i;
- default_values[i] =
_fetch_option.desc->slots()[i]->col_default_value();
- }
- }
- auto output_columns_guard = output_block->mutate_columns_scoped();
- MutableColumns& output_columns =
output_columns_guard.mutable_columns();
- for (int i = 0; i < resp.binary_row_data_size(); ++i) {
- RETURN_IF_ERROR(JsonbSerializeUtil::jsonb_to_columns(
- serdes, resp.binary_row_data(i).data(),
resp.binary_row_data(i).size(),
- col_uid_to_idx, output_columns, default_values, {}));
- }
- return Status::OK();
- }
- // Merge partial blocks
- Block partial_block;
- [[maybe_unused]] size_t uncompressed_size = 0;
- [[maybe_unused]] int64_t uncompressed_time = 0;
-
- RETURN_IF_ERROR(
- partial_block.deserialize(resp.block(), &uncompressed_size,
&uncompressed_time));
- if (partial_block.is_empty_column()) {
- return Status::OK();
- }
- CHECK(resp.row_locs().size() == partial_block.rows());
- if (output_block->is_empty_column()) {
- output_block->swap(partial_block);
- } else if (partial_block.columns() != output_block->columns()) {
- return Status::Error<ErrorCode::INTERNAL_ERROR>(
- "Merge block not match, self:[{}], input:[{}], ",
output_block->dump_types(),
- partial_block.dump_types());
- } else {
- for (int i = 0; i < output_block->columns(); ++i) {
- auto column_guard = output_block->mutate_column_scoped(i);
- MutableColumnPtr& column = column_guard.mutable_column();
- column->insert_range_from(
-
*partial_block.get_by_position(i).column->convert_to_full_column_if_const(),
- 0, partial_block.rows());
- }
- }
- return Status::OK();
- };
-
- for (const auto& resp : rsps) {
- RETURN_IF_ERROR(merge_function(resp));
- }
- return Status::OK();
-}
-
-Status RowIDFetcher::fetch(const ColumnPtr& column_row_ids, Block* res_block) {
- CHECK(!_stubs.empty());
- PMultiGetRequest mget_req = _init_fetch_request(
- assert_cast<const
ColumnString&>(*remove_nullable(column_row_ids).get()));
- std::vector<PMultiGetResponse> resps(_stubs.size());
- std::vector<brpc::Controller> cntls(_stubs.size());
- bthread::CountdownEvent counter(cast_set<int>(_stubs.size()));
- for (size_t i = 0; i < _stubs.size(); ++i) {
-
cntls[i].set_timeout_ms(_fetch_option.runtime_state->execution_timeout() *
1000);
- auto callback = brpc::NewCallback(fetch_callback, &counter);
- _stubs[i]->multiget_data(&cntls[i], &mget_req, &resps[i], callback);
- }
- counter.wait();
-
- // Merge
- std::vector<PRowLocation> rows_locs;
- rows_locs.reserve(rows_locs.size());
- RETURN_IF_ERROR(_merge_rpc_results(mget_req, resps, cntls, res_block,
&rows_locs));
- if (rows_locs.size() < column_row_ids->size()) {
- return Status::InternalError("Miss matched return row loc count {},
expected {}, input {}",
- rows_locs.size(), res_block->rows(),
column_row_ids->size());
- }
- // Final sort by row_ids sequence, since row_ids is already sorted if need
- std::map<GlobalRowLoacation, size_t> positions;
- for (size_t i = 0; i < rows_locs.size(); ++i) {
- RowsetId rowset_id;
- rowset_id.init(rows_locs[i].rowset_id());
- GlobalRowLoacation grl(rows_locs[i].tablet_id(), rowset_id,
- cast_set<uint32_t>(rows_locs[i].segment_id()),
- cast_set<uint32_t>(rows_locs[i].ordinal_id()));
- positions[grl] = i;
- };
- // TODO remove this warning code
- if (positions.size() < rows_locs.size()) {
- LOG(WARNING) << "cwntains duplicated row entry";
- }
- IColumn::Permutation permutation;
- permutation.reserve(column_row_ids->size());
- for (size_t i = 0; i < column_row_ids->size(); ++i) {
- auto location =
- reinterpret_cast<const
GlobalRowLoacation*>(column_row_ids->get_data_at(i).data);
- permutation.push_back(positions[*location]);
- }
- for (size_t i = 0; i < res_block->columns(); ++i) {
- res_block->get_by_position(i).column =
- res_block->get_by_position(i).column->permute(permutation,
permutation.size());
- }
- // Check row consistency
- RETURN_IF_CATCH_EXCEPTION(res_block->check_number_of_rows());
- VLOG_DEBUG << "dump block:" << res_block->dump_data(0, 10);
- return Status::OK();
-}
-
struct IteratorKey {
int64_t tablet_id;
RowsetId rowset_id;
@@ -396,162 +196,6 @@ static void scatter_scan_blocks_to_result_block(
}
}
-Status RowIdStorageReader::read_by_rowids(const PMultiGetRequest& request,
- PMultiGetResponse* response) {
- // read from storage engine row id by row id
- OlapReaderStatistics stats;
- Block result_block;
- int64_t acquire_tablet_ms = 0;
- int64_t acquire_rowsets_ms = 0;
- int64_t acquire_segments_ms = 0;
- int64_t lookup_row_data_ms = 0;
-
- // init desc
- std::vector<SlotDescriptor> slots;
- slots.reserve(request.slots().size());
- for (const auto& pslot : request.slots()) {
- slots.push_back(SlotDescriptor(pslot));
- }
-
- // init read schema
- TabletSchema full_read_schema;
- for (const ColumnPB& column_pb : request.column_desc()) {
- full_read_schema.append_column(TabletColumn(column_pb));
- }
-
- std::unordered_map<IteratorKey, IteratorItem, HashOfIteratorKey>
iterator_map;
- // read row by row
- for (int i = 0; i < request.row_locs_size(); ++i) {
- const auto& row_loc = request.row_locs(i);
- MonotonicStopWatch watch;
- watch.start();
- BaseTabletSPtr tablet = scope_timer_run(
- [&]() {
- auto res = ExecEnv::get_tablet(row_loc.tablet_id(),
nullptr, true);
- return !res.has_value() ? nullptr
- :
std::dynamic_pointer_cast<BaseTablet>(res.value());
- },
- &acquire_tablet_ms);
- RowsetId rowset_id;
- rowset_id.init(row_loc.rowset_id());
- if (!tablet) {
- continue;
- }
- // We ensured it's rowset is not released when init Tablet reader
param, rowset->update_delayed_expired_timestamp();
- BetaRowsetSharedPtr rowset =
std::static_pointer_cast<BetaRowset>(scope_timer_run(
- [&]() {
- return
ExecEnv::GetInstance()->storage_engine().get_quering_rowset(rowset_id);
- },
- &acquire_rowsets_ms));
- if (!rowset) {
- LOG(INFO) << "no such rowset " << rowset_id;
- continue;
- }
- size_t row_size = 0;
- Defer _defer([&]() {
- LOG_EVERY_N(INFO, 100)
- << "multiget_data single_row, cost(us):" <<
watch.elapsed_time() / 1000
- << ", row_size:" << row_size;
- *response->add_row_locs() = row_loc;
- });
- // TODO: supoort session variable enable_page_cache and
disable_file_cache if necessary.
- SegmentCacheHandle segment_cache;
- RETURN_IF_ERROR(scope_timer_run(
- [&]() {
- return SegmentLoader::instance()->load_segments(rowset,
&segment_cache, true);
- },
- &acquire_segments_ms));
- // find segment
- auto it = std::find_if(segment_cache.get_segments().cbegin(),
- segment_cache.get_segments().cend(),
- [&row_loc](const segment_v2::SegmentSharedPtr&
seg) {
- return seg->id() == row_loc.segment_id();
- });
- if (it == segment_cache.get_segments().end()) {
- continue;
- }
- segment_v2::SegmentSharedPtr segment = *it;
- GlobalRowLoacation row_location(row_loc.tablet_id(),
rowset->rowset_id(),
-
cast_set<uint32_t>(row_loc.segment_id()),
-
cast_set<uint32_t>(row_loc.ordinal_id()));
- // fetch by row store, more effcient way
- if (request.fetch_row_store()) {
- if (!tablet->tablet_schema()->has_row_store_for_all_columns()) {
- return Status::InternalError("Tablet {} does not have row
store for all columns",
- tablet->tablet_id());
- }
- RowLocation loc(rowset_id, segment->id(),
cast_set<uint32_t>(row_loc.ordinal_id()));
- std::string* value = response->add_binary_row_data();
- RETURN_IF_ERROR(scope_timer_run(
- [&]() { return tablet->lookup_row_data({}, loc, rowset,
stats, *value); },
- &lookup_row_data_ms));
- row_size = value->size();
- continue;
- }
-
- // fetch by column store
- if (result_block.is_empty_column()) {
- result_block = Block(slots, request.row_locs().size());
- }
- VLOG_DEBUG << "Read row location "
- << fmt::format("{}, {}, {}, {}", row_location.tablet_id,
-
row_location.row_location.rowset_id.to_string(),
- row_location.row_location.segment_id,
- row_location.row_location.row_id);
- for (int x = 0; x < slots.size(); ++x) {
- std::vector<segment_v2::rowid_t> row_ids {
- static_cast<segment_v2::rowid_t>(row_loc.ordinal_id())};
- // The scoped mutation republishes a potentially replaced V2
column when the reader
- // transfers ownership of its first assembled batch.
- auto column_guard = result_block.mutate_column_scoped(x);
- MutableColumnPtr& column = column_guard.mutable_column();
- IteratorKey iterator_key {.tablet_id = tablet->tablet_id(),
- .rowset_id = rowset_id,
- .segment_id = row_loc.segment_id(),
- .slot_id = slots[x].id()};
- IteratorItem& iterator_item = iterator_map[iterator_key];
- if (iterator_item.segment == nullptr) {
- // hold the reference
- iterator_map[iterator_key].segment = segment;
- iterator_item.storage_read_options.stats = &stats;
- iterator_item.storage_read_options.io_ctx.reader_type =
ReaderType::READER_QUERY;
- }
- segment = iterator_item.segment;
- set_slot_access_paths(slots[x], full_read_schema,
iterator_item.storage_read_options);
- RETURN_IF_ERROR(segment->seek_and_read_by_rowid(
- full_read_schema, &slots[x], row_ids, column,
- iterator_item.storage_read_options,
iterator_item.iterator));
- }
- }
- // serialize block if not empty
- if (!result_block.is_empty_column()) {
- VLOG_DEBUG << "dump block:" << result_block.dump_data(0, 10)
- << ", be_exec_version:" << request.be_exec_version();
- [[maybe_unused]] size_t compressed_size = 0;
- [[maybe_unused]] size_t uncompressed_size = 0;
- [[maybe_unused]] int64_t compress_time = 0;
- int be_exec_version = request.has_be_exec_version() ?
request.be_exec_version() : 0;
- RETURN_IF_ERROR(result_block.serialize(be_exec_version,
response->mutable_block(),
- &uncompressed_size,
&compressed_size, &compress_time,
-
segment_v2::CompressionTypePB::LZ4));
- }
-
- LOG(INFO) << "Query stats: "
- << fmt::format(
- "query_id:{}, "
- "hit_cached_pages:{}, total_pages_read:{},
compressed_bytes_read:{}, "
- "io_latency:{}ns, "
- "uncompressed_bytes_read:{},"
- "bytes_read:{},"
- "acquire_tablet_ms:{}, acquire_rowsets_ms:{},
acquire_segments_ms:{}, "
- "lookup_row_data_ms:{}",
- print_id(request.query_id()), stats.cached_pages_num,
- stats.total_pages_num, stats.compressed_bytes_read,
stats.io_ns,
- stats.uncompressed_bytes_read, stats.bytes_read,
acquire_tablet_ms,
- acquire_rowsets_ms, acquire_segments_ms,
lookup_row_data_ms);
- return Status::OK();
-}
-
Status RowIdStorageReader::read_by_rowids(const PMultiGetRequestV2& request,
PMultiGetResponseV2* response) {
if (request.request_block_descs_size()) {
diff --git a/be/src/exec/rowid_fetcher.h b/be/src/exec/rowid_fetcher.h
index 795854a0ebc..55bab866ff1 100644
--- a/be/src/exec/rowid_fetcher.h
+++ b/be/src/exec/rowid_fetcher.h
@@ -17,9 +17,6 @@
#pragma once
-#include <brpc/controller.h>
-#include <bthread/countdown_event.h>
-#include <gen_cpp/DataSinks_types.h>
#include <gen_cpp/internal_service.pb.h>
#include <condition_variable>
@@ -33,11 +30,9 @@
#include "core/block/block.h"
#include "core/data_type/data_type.h"
#include "storage/id_manager.h"
-#include "storage/tablet_info.h" // DorisNodesInfo
namespace doris {
-class DorisNodesInfo;
class RuntimeState;
class TupleDescriptor;
namespace io {
@@ -52,41 +47,8 @@ struct IteratorKey;
struct IteratorItem;
struct HashOfIteratorKey;
-inline void fetch_callback(bthread::CountdownEvent* counter) {
- Defer __defer([&] { counter->signal(); });
-}
-
-template <typename T>
-class ColumnStr;
-using ColumnString = ColumnStr<UInt32>;
class MutableBlock;
-// fetch rows by global rowid
-// tablet_id/rowset_name/segment_id/ordinal_id
-
-struct FetchOption {
- TupleDescriptor* desc = nullptr;
- RuntimeState* runtime_state = nullptr;
- TFetchOption t_fetch_opt;
-};
-
-class RowIDFetcher {
-public:
- RowIDFetcher(FetchOption fetch_opt) : _fetch_option(std::move(fetch_opt))
{}
- Status init();
- Status fetch(const ColumnPtr& row_ids, Block* block);
-
-private:
- PMultiGetRequest _init_fetch_request(const ColumnString& row_ids) const;
- Status _merge_rpc_results(const PMultiGetRequest& request,
- const std::vector<PMultiGetResponse>& rsps,
- const std::vector<brpc::Controller>& cntls,
Block* output_block,
- std::vector<PRowLocation>* rows_id) const;
-
- std::vector<std::shared_ptr<PBackendService_Stub>> _stubs;
- FetchOption _fetch_option;
-};
-
struct RowStoreReadStruct {
RowStoreReadStruct(std::string& buffer) : row_store_buffer(buffer) {};
std::string& row_store_buffer;
@@ -114,7 +76,6 @@ public:
static const std::string TopNLazyMaterializationSecondPhaseRowsRead;
static const std::string TopNLazyMaterializationSecondPhaseSegmentsRead;
- static Status read_by_rowids(const PMultiGetRequest& request,
PMultiGetResponse* response);
static Status read_by_rowids(const PMultiGetRequestV2& request,
PMultiGetResponseV2* response);
private:
diff --git a/be/src/exec/scan/olap_scanner.cpp
b/be/src/exec/scan/olap_scanner.cpp
index 91aa9685f0c..fbcfcbb61db 100644
--- a/be/src/exec/scan/olap_scanner.cpp
+++ b/be/src/exec/scan/olap_scanner.cpp
@@ -668,21 +668,6 @@ Status OlapScanner::_init_tablet_reader_params(
}
}
- // If this is a Two-Phase read query, and we need to delay the release of
Rowset
- // by rowset->update_delayed_expired_timestamp().This could expand the
lifespan of Rowset
- if (tablet_schema->field_index(BeConsts::ROWID_COL) >= 0) {
- constexpr static int delayed_s = 60;
- for (auto rs_reader : _tablet_reader_params.rs_splits) {
- uint64_t delayed_expired_timestamp =
- UnixSeconds() +
_tablet_reader_params.runtime_state->execution_timeout() +
- delayed_s;
- rs_reader.rs_reader->rowset()->update_delayed_expired_timestamp(
- delayed_expired_timestamp);
- ExecEnv::GetInstance()->storage_engine().add_quering_rowset(
- rs_reader.rs_reader->rowset());
- }
- }
-
if (tablet_schema->has_global_row_id()) {
auto& id_file_map = _state->get_id_file_map();
for (auto rs_reader : _tablet_reader_params.rs_splits) {
diff --git a/be/src/service/internal_service.cpp
b/be/src/service/internal_service.cpp
index 371b7803c19..9776461a61a 100644
--- a/be/src/service/internal_service.cpp
+++ b/be/src/service/internal_service.cpp
@@ -1985,22 +1985,13 @@ void
PInternalServiceImpl::response_slave_tablet_pull_rowset(
void PInternalService::multiget_data(google::protobuf::RpcController*
controller,
const PMultiGetRequest* request,
PMultiGetResponse* response,
google::protobuf::Closure* done) {
- bool ret = _heavy_work_pool.try_offer([request, response, done]() {
- signal::SignalTaskIdKeeper keeper(request->query_id());
- // multi get data by rowid
- MonotonicStopWatch watch;
- watch.start();
- brpc::ClosureGuard closure_guard(done);
- response->mutable_status()->set_status_code(0);
-
SCOPED_ATTACH_TASK(ExecEnv::GetInstance()->rowid_storage_reader_tracker());
- Status st = RowIdStorageReader::read_by_rowids(*request, response);
- st.to_protobuf(response->mutable_status());
- LOG(INFO) << "multiget_data finished, cost(us):" <<
watch.elapsed_time() / 1000;
- });
- if (!ret) {
- offer_failed(response, done, _heavy_work_pool);
- return;
- }
+ brpc::ClosureGuard closure_guard(done);
+#pragma GCC diagnostic push
+#pragma GCC diagnostic ignored "-Wdeprecated-declarations"
+ // The deprecated response field is retained only to reject legacy callers
explicitly.
+ Status::NotSupported("multiget_data is deprecated; use multiget_data_v2")
+ .to_protobuf(response->mutable_status());
+#pragma GCC diagnostic pop
}
void PInternalService::multiget_data_v2(google::protobuf::RpcController*
controller,
diff --git a/be/src/service/internal_service.h
b/be/src/service/internal_service.h
index 66816a1b245..550f2af8637 100644
--- a/be/src/service/internal_service.h
+++ b/be/src/service/internal_service.h
@@ -309,8 +309,6 @@ public:
google::protobuf::Closure* done) override;
private:
- Status _multi_get(const PMultiGetRequest& request, PMultiGetResponse*
response);
-
void _get_column_ids_by_tablet_ids(google::protobuf::RpcController*
controller,
const PFetchColIdsRequest* request,
PFetchColIdsResponse* response,
diff --git a/be/src/storage/olap_server.cpp b/be/src/storage/olap_server.cpp
index 07f179f6348..78021456ae2 100644
--- a/be/src/storage/olap_server.cpp
+++ b/be/src/storage/olap_server.cpp
@@ -227,10 +227,10 @@ Status
StorageEngine::start_bg_threads(std::shared_ptr<WorkloadGroup> wg_sptr) {
LOG(INFO) << "unused rowset monitor thread started";
RETURN_IF_ERROR(Thread::create(
- "StorageEngine", "evict_querying_rowset_thread",
- [this]() { this->_evict_quring_rowset_thread_callback(); },
- &_evict_quering_rowset_thread));
- LOG(INFO) << "evict quering thread started";
+ "StorageEngine", "id_file_map_gc_thread",
+ [this]() { this->_gc_expired_id_file_map_thread_callback(); },
+ &_id_file_map_gc_thread));
+ LOG(INFO) << "id file map gc thread started";
// start thread for monitoring the snapshot and trash folder
RETURN_IF_ERROR(Thread::create(
diff --git a/be/src/storage/rowset/rowset.h b/be/src/storage/rowset/rowset.h
index b952c44197c..1a1575ed707 100644
--- a/be/src/storage/rowset/rowset.h
+++ b/be/src/storage/rowset/rowset.h
@@ -262,14 +262,6 @@ public:
}
}
- void update_delayed_expired_timestamp(uint64_t delayed_expired_timestamp) {
- if (delayed_expired_timestamp > _delayed_expired_timestamp) {
- _delayed_expired_timestamp = delayed_expired_timestamp;
- }
- }
-
- uint64_t delayed_expired_timestamp() { return _delayed_expired_timestamp; }
-
virtual Status get_segments_key_bounds(std::vector<KeyBoundsPB>*
segments_key_bounds) {
_rowset_meta->get_segments_key_bounds(segments_key_bounds);
return Status::OK();
@@ -376,8 +368,6 @@ protected:
std::atomic<uint64_t> _refs_by_reader;
// rowset state machine
RowsetStateMachine _rowset_state_machine;
- std::atomic<uint64_t> _delayed_expired_timestamp = 0;
-
// <column_uniq_id>, skip index compaction
std::set<int32_t> skip_index_compaction;
diff --git a/be/src/storage/schema.h b/be/src/storage/schema.h
index e35d4c37ac1..fe02682019a 100644
--- a/be/src/storage/schema.h
+++ b/be/src/storage/schema.h
@@ -58,8 +58,7 @@ public:
if (columns[i]->name() == DELETE_SIGN) {
_delete_sign_idx = i;
}
- if (columns[i]->name() == BeConsts::ROWID_COL ||
- columns[i]->name().starts_with(BeConsts::GLOBAL_ROWID_COL)) {
+ if (columns[i]->name().starts_with(BeConsts::GLOBAL_ROWID_COL)) {
_rowid_col_idx = i;
}
if (columns[i]->name() == VERSION_COL) {
diff --git a/be/src/storage/segment/column_reader.h
b/be/src/storage/segment/column_reader.h
index 327c44c4fee..4cadc599d31 100644
--- a/be/src/storage/segment/column_reader.h
+++ b/be/src/storage/segment/column_reader.h
@@ -894,52 +894,6 @@ private:
Status _seek_by_offsets(ordinal_t ord);
};
-class RowIdColumnIterator : public ColumnIterator {
-public:
- RowIdColumnIterator() = delete;
- RowIdColumnIterator(int64_t tid, RowsetId rid, int32_t segid)
- : _tablet_id(tid), _rowset_id(rid), _segment_id(segid) {}
-
- Status seek_to_ordinal(ordinal_t ord_idx) override {
- _current_rowid = cast_set<uint32_t>(ord_idx);
- return Status::OK();
- }
-
- Status next_batch(size_t* n, MutableColumnPtr& dst) {
- bool has_null;
- return next_batch(n, dst, &has_null);
- }
-
- Status next_batch(size_t* n, MutableColumnPtr& dst, bool* has_null)
override {
- for (size_t i = 0; i < *n; ++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));
- }
- _current_rowid += *n;
- return Status::OK();
- }
-
- Status read_by_rowids(const rowid_t* rowids, const size_t count,
- MutableColumnPtr& dst) override {
- for (size_t i = 0; i < count; ++i) {
- rowid_t row_id = rowids[i];
- GlobalRowLoacation location(_tablet_id, _rowset_id, _segment_id,
row_id);
- dst->insert_data(reinterpret_cast<const char*>(&location),
sizeof(GlobalRowLoacation));
- }
- return Status::OK();
- }
-
- ordinal_t get_current_ordinal() const override { return _current_rowid; }
-
-private:
- rowid_t _current_rowid = 0;
- int64_t _tablet_id = 0;
- RowsetId _rowset_id;
- int32_t _segment_id = 0;
-};
-
-// Add new RowIdColumnIteratorV2
class RowIdColumnIteratorV2 : public ColumnIterator {
public:
RowIdColumnIteratorV2(uint8_t version, int64_t backend_id, uint32_t
file_id)
diff --git a/be/src/storage/segment/segment_iterator.cpp
b/be/src/storage/segment/segment_iterator.cpp
index 85bf9bf53fa..c8d768322b8 100644
--- a/be/src/storage/segment/segment_iterator.cpp
+++ b/be/src/storage/segment/segment_iterator.cpp
@@ -1718,12 +1718,6 @@ Status SegmentIterator::_init_return_column_iterators() {
}
for (auto cid : _schema->column_ids()) {
- if (_schema->column(cid)->name() == BeConsts::ROWID_COL) {
- _column_iterators[cid].reset(
- new RowIdColumnIterator(_opts.tablet_id, _opts.rowset_id,
_segment->id()));
- continue;
- }
-
if
(_schema->column(cid)->name().starts_with(BeConsts::GLOBAL_ROWID_COL)) {
auto& id_file_map = _opts.runtime_state->get_id_file_map();
uint32_t file_id =
id_file_map->get_file_mapping_id(std::make_shared<FileMapping>(
diff --git a/be/src/storage/storage_engine.cpp
b/be/src/storage/storage_engine.cpp
index 2093a17e663..dcb81f3f111 100644
--- a/be/src/storage/storage_engine.cpp
+++ b/be/src/storage/storage_engine.cpp
@@ -1521,10 +1521,10 @@ bool
BaseStorageEngine::notify_listener(std::string_view name) {
return found;
}
-void BaseStorageEngine::_evict_quring_rowset_thread_callback() {
+void BaseStorageEngine::_gc_expired_id_file_map_thread_callback() {
int32_t interval = config::quering_rowsets_evict_interval;
do {
- _evict_querying_rowset();
+ _gc_expired_id_file_map();
interval = config::quering_rowsets_evict_interval;
if (interval <= 0) {
LOG(WARNING) << "quering_rowsets_evict_interval config is illegal:
" << interval
@@ -1639,34 +1639,7 @@ void
StorageEngine::get_compaction_status_json(std::string* result) {
_compaction_submit_registry.jsonfy_compaction_status(result);
}
-void BaseStorageEngine::add_quering_rowset(RowsetSharedPtr rs) {
- std::lock_guard<std::mutex> lock(_quering_rowsets_mutex);
- _querying_rowsets.emplace(rs->rowset_id(), rs);
-}
-
-RowsetSharedPtr BaseStorageEngine::get_quering_rowset(RowsetId rs_id) {
- std::lock_guard<std::mutex> lock(_quering_rowsets_mutex);
- auto it = _querying_rowsets.find(rs_id);
- if (it != _querying_rowsets.end()) {
- return it->second;
- }
- return nullptr;
-}
-
-void BaseStorageEngine::_evict_querying_rowset() {
- {
- std::lock_guard<std::mutex> lock(_quering_rowsets_mutex);
- for (auto it = _querying_rowsets.begin(); it !=
_querying_rowsets.end();) {
- uint64_t now = UnixSeconds();
- // We delay the GC time of this rowset since it's maybe still
needed, see #20732
- if (now > it->second->delayed_expired_timestamp()) {
- it = _querying_rowsets.erase(it);
- } else {
- ++it;
- }
- }
- }
-
+void BaseStorageEngine::_gc_expired_id_file_map() {
uint64_t now = UnixSeconds();
ExecEnv::GetInstance()->get_id_manager()->gc_expired_id_file_map(now);
}
diff --git a/be/src/storage/storage_engine.h b/be/src/storage/storage_engine.h
index c503fc6cb31..d029d436624 100644
--- a/be/src/storage/storage_engine.h
+++ b/be/src/storage/storage_engine.h
@@ -151,10 +151,6 @@ public:
return _calc_delete_bitmap_executor_for_load.get();
}
- void add_quering_rowset(RowsetSharedPtr rs);
-
- RowsetSharedPtr get_quering_rowset(RowsetId rs_id);
-
int64_t memory_limitation_bytes_per_thread_for_schema_change() const;
int get_disk_num() { return _disk_num; }
@@ -167,8 +163,8 @@ public:
protected:
void _start_adaptive_thread_controller();
- void _evict_querying_rowset();
- void _evict_quring_rowset_thread_callback();
+ void _gc_expired_id_file_map();
+ void _gc_expired_id_file_map_thread_callback();
bool _should_delay_large_task();
int32_t _effective_cluster_id = -1;
@@ -185,10 +181,7 @@ protected:
std::unique_ptr<CalcDeleteBitmapExecutor>
_calc_delete_bitmap_executor_for_load;
CountDownLatch _stop_background_threads_latch;
- // Hold reference of quering rowsets
- std::mutex _quering_rowsets_mutex;
- std::unordered_map<RowsetId, RowsetSharedPtr> _querying_rowsets;
- std::shared_ptr<Thread> _evict_quering_rowset_thread;
+ std::shared_ptr<Thread> _id_file_map_gc_thread;
int64_t _memory_limitation_bytes_for_schema_change;
diff --git a/be/src/storage/utils.h b/be/src/storage/utils.h
index a0b3557489d..a6ea7a1874a 100644
--- a/be/src/storage/utils.h
+++ b/be/src/storage/utils.h
@@ -223,25 +223,6 @@ struct RowLocation {
using RowLocationSet = std::set<RowLocation>;
using RowLocationPairList = std::list<std::pair<RowLocation, RowLocation>>;
-struct GlobalRowLoacation {
- GlobalRowLoacation(int64_t tid, RowsetId rsid, uint32_t sid, uint32_t rid)
- : tablet_id(tid), row_location(rsid, sid, rid) {}
- int64_t tablet_id;
- RowLocation row_location;
-
- bool operator==(const GlobalRowLoacation& rhs) const {
- return tablet_id == rhs.tablet_id && row_location == rhs.row_location;
- }
-
- bool operator<(const GlobalRowLoacation& rhs) const {
- if (tablet_id != rhs.tablet_id) {
- return tablet_id < rhs.tablet_id;
- } else {
- return row_location < rhs.row_location;
- }
- }
-};
-
struct GlobalRowLoacationV2 {
GlobalRowLoacationV2(uint8_t ver, uint64_t bid, uint32_t fid, uint32_t rid)
: version(ver), backend_id(bid), file_id(fid), row_id(rid) {}
diff --git a/be/test/storage/variant/variant_column_writer_reader_test.cpp
b/be/test/storage/variant/variant_column_writer_reader_test.cpp
index 17e1ad07416..887bd5c4fb2 100644
--- a/be/test/storage/variant/variant_column_writer_reader_test.cpp
+++ b/be/test/storage/variant/variant_column_writer_reader_test.cpp
@@ -44,7 +44,6 @@
#include "core/value/jsonb_value.h"
#include "core/value/variant/variant_batch_builder.h"
#include "core/value/variant/variant_parquet_encoding.h"
-#include "exec/rowid_fetcher.h"
#include "gtest/gtest.h"
#include "runtime/descriptor_helper.h"
#include "runtime/descriptors.h"
@@ -3706,120 +3705,6 @@ TEST_F(VariantColumnWriterReaderTest,
test_segment_rowid_read_by_reader_version)
}
}
-TEST_F(VariantColumnWriterReaderTest,
test_legacy_rowid_storage_reader_extracted_leaf) {
- init_variant_tablet(21002, 1);
- // "hot" consumes the only materialized slot. The sparse "z" path stays
homogeneous, while
- // "mixed" exercises missing -> int -> string through the row-at-a-time
caller.
- const std::vector<std::string> jsons {R"({"hot":0,"z":100})",
- R"({"hot":1,"z":101,"mixed":101})",
-
R"({"hot":2,"z":102,"mixed":"mixed"})", R"({"hot":3})"};
- auto rowset = create_variant_rowset({jsons}, 1, 100);
-
- RuntimeProfile profile("RegisterVariantRowIdTablet");
- auto* tablet_manager = _engine_ref->tablet_manager();
- {
- std::lock_guard<std::shared_mutex> lock(
- tablet_manager->_get_tablets_shard_lock(_tablet->tablet_id()));
- ASSERT_TRUE(tablet_manager
- ->_add_tablet_unlocked(_tablet->tablet_id(),
_tablet,
- /*update_meta=*/false,
/*force=*/false, &profile)
- .ok());
- }
- _engine_ref->add_quering_rowset(rowset);
-
- auto make_slot = [](std::string path, bool use_v2) {
- auto slot = TSlotDescriptorBuilder()
- .type(TYPE_VARIANT)
- .nullable(true)
- .column_name("v1")
- .column_pos(0)
- .build();
- slot.__set_col_unique_id(1);
- slot.__set_column_paths({std::move(path)});
- slot.__set_primitive_type(TPrimitiveType::VARIANT);
- auto& scalar = slot.slotType.types[0].scalar_type;
- scalar.__set_variant_max_subcolumns_count(1);
- scalar.__set_variant_enable_doc_mode(false);
- scalar.__set_variant_is_v2(use_v2);
- return slot;
- };
-
- for (const bool use_v2 : {false, true}) {
- TDescriptorTableBuilder descriptor_builder;
- TTupleDescriptorBuilder tuple_builder;
- tuple_builder.add_slot(make_slot("z", use_v2));
- tuple_builder.add_slot(make_slot("mixed", use_v2));
- tuple_builder.build(&descriptor_builder);
- ObjectPool object_pool;
- DescriptorTbl* descriptor_table = nullptr;
- ASSERT_TRUE(DescriptorTbl::create(&object_pool,
descriptor_builder.desc_tbl(),
- &descriptor_table)
- .ok());
- const auto& slots = descriptor_table->get_tuple_descriptor(0)->slots();
- ASSERT_EQ(slots.size(), 2);
-
- PMultiGetRequest request;
- slots[0]->to_protobuf(request.add_slots());
- slots[1]->to_protobuf(request.add_slots());
- _tablet_schema->column(0).to_schema_pb(request.add_column_desc());
- request.set_fetch_row_store(false);
- request.mutable_query_id()->set_hi(1);
- request.mutable_query_id()->set_lo(use_v2 ? 2 : 1);
- for (uint32_t row_id = 0; row_id < 3; ++row_id) {
- auto* location = request.add_row_locs();
- location->set_tablet_id(_tablet->tablet_id());
- location->set_rowset_id(rowset->rowset_id().to_string());
- location->set_segment_id(0);
- location->set_ordinal_id(row_id);
- }
-
- PMultiGetResponse response;
- auto st = RowIdStorageReader::read_by_rowids(request, &response);
- ASSERT_TRUE(st.ok()) << "use_v2=" << use_v2 << ": " << st.to_string();
- ASSERT_TRUE(response.has_block());
- ASSERT_EQ(response.row_locs_size(), 3);
-
- Block result;
- size_t uncompressed_size = 0;
- int64_t uncompressed_time = 0;
- st = result.deserialize(response.block(), &uncompressed_size,
&uncompressed_time);
- ASSERT_TRUE(st.ok()) << "use_v2=" << use_v2 << ": " << st.to_string();
- ASSERT_EQ(result.columns(), 2);
- ASSERT_EQ(result.rows(), 3);
-
- const auto& result_column = result.get_by_position(0);
- const auto result_type = remove_nullable(result_column.type);
- EXPECT_EQ(typeid_cast<const DataTypeVariantV2*>(result_type.get()) !=
nullptr, use_v2);
- EXPECT_EQ(typeid_cast<const DataTypeVariant*>(result_type.get()) !=
nullptr, !use_v2);
- const auto& nullable = assert_cast<const
ColumnNullable&>(*result_column.column);
- const auto& variant = nullable.get_nested_column();
- for (size_t row = 0; row < 3; ++row) {
- EXPECT_FALSE(nullable.is_null_at(row)) << "use_v2=" << use_v2 <<
", row=" << row;
- EXPECT_EQ(variant_json_at(variant, row), std::to_string(row + 100))
- << "use_v2=" << use_v2 << ", row=" << row;
- }
-
- const auto& mixed_result_column = result.get_by_position(1);
- const auto mixed_type = remove_nullable(mixed_result_column.type);
- EXPECT_EQ(typeid_cast<const DataTypeVariantV2*>(mixed_type.get()) !=
nullptr, use_v2);
- EXPECT_EQ(typeid_cast<const DataTypeVariant*>(mixed_type.get()) !=
nullptr, !use_v2);
- const auto& mixed_nullable =
- assert_cast<const
ColumnNullable&>(*mixed_result_column.column);
- const auto& mixed_variant = mixed_nullable.get_nested_column();
- ASSERT_EQ(mixed_nullable.size(), 3);
- EXPECT_TRUE(mixed_nullable.is_null_at(0)) << "use_v2=" << use_v2;
- EXPECT_FALSE(mixed_nullable.is_null_at(1)) << "use_v2=" << use_v2;
- EXPECT_FALSE(mixed_nullable.is_null_at(2)) << "use_v2=" << use_v2;
- EXPECT_EQ(variant_json_at(mixed_variant, 1), "101") << "use_v2=" <<
use_v2;
- EXPECT_EQ(variant_json_at(mixed_variant, 2), R"("mixed")") <<
"use_v2=" << use_v2;
-
- if (use_v2) {
- EXPECT_TRUE(assert_cast<const
ColumnVariantV2&>(variant).is_typed());
- EXPECT_FALSE(assert_cast<const
ColumnVariantV2&>(mixed_variant).is_typed());
- }
- }
-}
-
TEST_F(VariantColumnWriterReaderTest,
test_speculative_sparse_read_after_statistics_truncation) {
TabletSchemaPB schema_pb;
schema_pb.set_keys_type(KeysType::DUP_KEYS);
diff --git a/fe/fe-catalog/src/main/java/org/apache/doris/catalog/Column.java
b/fe/fe-catalog/src/main/java/org/apache/doris/catalog/Column.java
index 805bc7db19b..91289f09e17 100644
--- a/fe/fe-catalog/src/main/java/org/apache/doris/catalog/Column.java
+++ b/fe/fe-catalog/src/main/java/org/apache/doris/catalog/Column.java
@@ -55,7 +55,6 @@ public class Column implements GsonPostProcessable {
public static final String DELETE_SIGN = "__DORIS_DELETE_SIGN__";
public static final String WHERE_SIGN = "__DORIS_WHERE_SIGN__";
public static final String SEQUENCE_COL = "__DORIS_SEQUENCE_COL__";
- public static final String ROWID_COL = "__DORIS_ROWID_COL__";
public static final String GLOBAL_ROWID_COL = "__DORIS_GLOBAL_ROWID_COL__";
public static final String ROW_STORE_COL = "__DORIS_ROW_STORE_COL__";
public static final String VERSION_COL = "__DORIS_VERSION_COL__";
diff --git a/fe/fe-core/src/main/java/org/apache/doris/analysis/SortInfo.java
b/fe/fe-core/src/main/java/org/apache/doris/analysis/SortInfo.java
index 5cea596a5dc..3854290e1ba 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/analysis/SortInfo.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/analysis/SortInfo.java
@@ -45,7 +45,6 @@ public class SortInfo {
private TupleDescriptor sortTupleDesc;
// Input expressions materialized into sortTupleDesc_. One expr per slot in
// sortTupleDesc_.
- private boolean useTwoPhaseRead = false;
/**
* Used by new optimizer.
@@ -68,7 +67,6 @@ public class SortInfo {
isAscOrder = Lists.newArrayList(other.isAscOrder);
nullsFirstParams = Lists.newArrayList(other.nullsFirstParams);
sortTupleDesc = other.sortTupleDesc;
- useTwoPhaseRead = other.useTwoPhaseRead;
}
public List<Expr> getOrderingExprs() {
@@ -79,10 +77,6 @@ public class SortInfo {
return isAscOrder;
}
- public void setUseTwoPhaseRead() {
- useTwoPhaseRead = true;
- }
-
public TupleDescriptor getSortTupleDescriptor() {
return sortTupleDesc;
}
@@ -114,9 +108,6 @@ public class SortInfo {
ExprToThriftVisitor.treesToThrift(orderingExprs),
isAscOrder,
nullsFirstParams);
- if (useTwoPhaseRead) {
- sortInfo.setUseTwoPhaseRead(true);
- }
return sortInfo;
}
}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/planner/OlapScanNode.java
b/fe/fe-core/src/main/java/org/apache/doris/planner/OlapScanNode.java
index 9ff57d1ef1a..ee7331cc3df 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/planner/OlapScanNode.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/planner/OlapScanNode.java
@@ -54,7 +54,6 @@ import org.apache.doris.catalog.PartitionType;
import org.apache.doris.catalog.RangePartitionItem;
import org.apache.doris.catalog.Replica;
import org.apache.doris.catalog.RowBinlogTableWrapper;
-import org.apache.doris.catalog.ScalarType;
import org.apache.doris.catalog.Tablet;
import org.apache.doris.cloud.catalog.CloudReplica;
import org.apache.doris.cloud.qe.ComputeGroupException;
@@ -81,7 +80,6 @@ import org.apache.doris.resource.BackendSelectionManager;
import org.apache.doris.resource.Tag;
import org.apache.doris.resource.computegroup.ComputeGroup;
import org.apache.doris.system.Backend;
-import org.apache.doris.thrift.TAggregationType;
import org.apache.doris.thrift.TBinlogScanType;
import org.apache.doris.thrift.TColumn;
import org.apache.doris.thrift.TExplainLevel;
@@ -1362,22 +1360,6 @@ public class OlapScanNode extends ScanNode {
columnsDesc.add(ColumnToThrift.toThrift(globalRowIdColumn));
} else {
olapTable.getColumnDesc(selectedIndexId, columnsDesc,
keyColumnNames, keyColumnTypes);
-
- // Add extra row id column
- ArrayList<SlotDescriptor> slots = desc.getSlots();
- Column lastColumn = slots.get(slots.size() - 1).getColumn();
- if (lastColumn != null &&
lastColumn.getName().equalsIgnoreCase(Column.ROWID_COL)) {
- TColumn tColumn = new TColumn();
- tColumn.setColumnName(Column.ROWID_COL);
-
tColumn.setColumnType(ScalarType.createStringType().toColumnTypeThrift());
- tColumn.setAggregationType(TAggregationType.REPLACE);
- tColumn.setIsKey(false);
- tColumn.setIsAllowNull(false);
- // keep compatibility
- tColumn.setVisible(false);
- tColumn.setColUniqueId(Integer.MAX_VALUE);
- columnsDesc.add(tColumn);
- }
}
// Add virtual column to ColumnsDesc so that backend could
diff --git a/fe/fe-core/src/main/java/org/apache/doris/planner/ResultSink.java
b/fe/fe-core/src/main/java/org/apache/doris/planner/ResultSink.java
index 25e72ed7598..2bc65ed3545 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/planner/ResultSink.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/planner/ResultSink.java
@@ -20,7 +20,6 @@ package org.apache.doris.planner;
import org.apache.doris.thrift.TDataSink;
import org.apache.doris.thrift.TDataSinkType;
import org.apache.doris.thrift.TExplainLevel;
-import org.apache.doris.thrift.TFetchOption;
import org.apache.doris.thrift.TResultSink;
import org.apache.doris.thrift.TResultSinkType;
@@ -31,9 +30,6 @@ import org.apache.doris.thrift.TResultSinkType;
*/
public class ResultSink extends DataSink {
private final PlanNodeId exchNodeId;
- // Two phase fetch option
- private TFetchOption fetchOption;
-
private TResultSinkType resultSinkType = TResultSinkType.MYSQL_PROTOCOL;
public ResultSink(PlanNodeId exchNodeId) {
@@ -51,27 +47,14 @@ public class ResultSink extends DataSink {
strBuilder.append(prefix);
strBuilder.append("V");
strBuilder.append("RESULT SINK\n");
- if (fetchOption != null) {
- strBuilder.append(prefix).append(" ").append("OPT TWO PHASE\n");
- if (fetchOption.isFetchRowStore()) {
- strBuilder.append(prefix).append(" ").append("FETCH ROW
STORE\n");
- }
- }
strBuilder.append(prefix).append("
").append(resultSinkType).append("\n");
return strBuilder.toString();
}
- public void setFetchOption(TFetchOption fetchOption) {
- this.fetchOption = fetchOption;
- }
-
@Override
protected TDataSink toThrift() {
TDataSink result = new TDataSink(TDataSinkType.RESULT_SINK);
TResultSink tResultSink = new TResultSink();
- if (fetchOption != null) {
- tResultSink.setFetchOption(fetchOption);
- }
tResultSink.setType(resultSinkType);
result.setResultSink(tResultSink);
return result;
diff --git a/fe/fe-core/src/main/java/org/apache/doris/planner/SortNode.java
b/fe/fe-core/src/main/java/org/apache/doris/planner/SortNode.java
index 716dc1739f3..e57cf29143e 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/planner/SortNode.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/planner/SortNode.java
@@ -52,7 +52,6 @@ public class SortNode extends PlanNode {
private final SortInfo info;
private final boolean useTopN;
private boolean useTopnOpt = false;
- private boolean useTwoPhaseReadOpt;
private boolean hasRuntimePredicate = false;
// If mergeByexchange is set to true, the sort information is pushed to the
@@ -98,7 +97,7 @@ public class SortNode extends PlanNode {
} else {
if (limit <= 0) {
algorithm = TSortAlgorithm.FULL_SORT;
- } else if (hasRuntimePredicate || useTwoPhaseReadOpt) {
+ } else if (hasRuntimePredicate) {
algorithm = TSortAlgorithm.HEAP_SORT;
} else {
if (limit + offset < 50000) {
@@ -143,11 +142,6 @@ public class SortNode extends PlanNode {
this.useTopnOpt = useTopnOpt;
}
- public void setUseTwoPhaseReadOpt(boolean useTwoPhaseReadOpt) {
- this.useTwoPhaseReadOpt = useTwoPhaseReadOpt;
- updateSortAlgorithm();
- }
-
@Override
public String getNodeExplainString(String detailPrefix, TExplainLevel
detailLevel) {
if (detailLevel == TExplainLevel.BRIEF) {
@@ -173,10 +167,6 @@ public class SortNode extends PlanNode {
if (useTopnOpt) {
output.append(detailPrefix + "TOPN filter targets:
").append(topnFilterTargets).append("\n");
}
- if (useTwoPhaseReadOpt) {
- output.append(detailPrefix + "OPT TWO PHASE\n");
- }
-
output.append(detailPrefix + "algorithm: ");
if (algorithm == TSortAlgorithm.HEAP_SORT) {
output.append("heap sort\n");
diff --git a/gensrc/proto/internal_service.proto
b/gensrc/proto/internal_service.proto
index 54c7af9171b..214aea5e6a5 100644
--- a/gensrc/proto/internal_service.proto
+++ b/gensrc/proto/internal_service.proto
@@ -792,35 +792,35 @@ message PJdbcTestConnectionResult {
}
message PRowLocation {
- optional int64 tablet_id = 1;
- optional string rowset_id = 2;
- optional uint64 segment_id = 3;
- optional uint64 ordinal_id = 4;
+ optional int64 tablet_id = 1 [deprecated = true];
+ optional string rowset_id = 2 [deprecated = true];
+ optional uint64 segment_id = 3 [deprecated = true];
+ optional uint64 ordinal_id = 4 [deprecated = true];
}
message PMultiGetRequest {
- repeated PRowLocation row_locs = 1;
- optional PTupleDescriptor desc = 2;
- repeated PSlotDescriptor slots = 3;
+ repeated PRowLocation row_locs = 1 [deprecated = true];
+ optional PTupleDescriptor desc = 2 [deprecated = true];
+ repeated PSlotDescriptor slots = 3 [deprecated = true];
// for compability
- optional int32 be_exec_version = 4;
- optional bool fetch_row_store = 5;
- optional PUniqueId query_id = 6;
- repeated ColumnPB column_desc = 7;
+ optional int32 be_exec_version = 4 [deprecated = true];
+ optional bool fetch_row_store = 5 [deprecated = true];
+ optional PUniqueId query_id = 6 [deprecated = true];
+ repeated ColumnPB column_desc = 7 [deprecated = true];
};
message PMultiGetResponse {
- optional PBlock block = 1;
- optional PStatus status = 2;
+ optional PBlock block = 1 [deprecated = true];
+ optional PStatus status = 2 [deprecated = true];
// more effecient serialization fields for row store
enum RowFormat {
JSONB = 0;
};
- optional RowFormat format = 3;
- repeated bytes binary_row_data = 4;
+ optional RowFormat format = 3 [deprecated = true];
+ repeated bytes binary_row_data = 4 [deprecated = true];
// for sorting rows
- repeated PRowLocation row_locs = 5;
+ repeated PRowLocation row_locs = 5 [deprecated = true];
};
// Eeach block have own schema to read
@@ -1261,7 +1261,9 @@ service PBackendService {
rpc response_slave_tablet_pull_rowset(PTabletWriteSlaveDoneRequest)
returns (PTabletWriteSlaveDoneResult);
rpc outfile_write_success(POutfileWriteSuccessRequest) returns
(POutfileWriteSuccessResult);
rpc fetch_table_schema(PFetchTableSchemaRequest) returns
(PFetchTableSchemaResult);
- rpc multiget_data(PMultiGetRequest) returns (PMultiGetResponse);
+ rpc multiget_data(PMultiGetRequest) returns (PMultiGetResponse) {
+ option deprecated = true;
+ }
rpc multiget_data_v2(PMultiGetRequestV2) returns (PMultiGetResponseV2);
rpc get_file_cache_meta_by_tablet_id(PGetFileCacheMetaRequest) returns
(PGetFileCacheMetaResponse);
rpc warm_up_rowset(PWarmUpRowsetRequest) returns (PWarmUpRowsetResponse);
diff --git a/gensrc/thrift/DataSinks.thrift b/gensrc/thrift/DataSinks.thrift
index 2eb770b383e..1c776b83066 100644
--- a/gensrc/thrift/DataSinks.thrift
+++ b/gensrc/thrift/DataSinks.thrift
@@ -216,7 +216,7 @@ struct TMultiCastDataStreamSink {
2: optional list<list<TPlanFragmentDestination>> destinations;
}
-// [deprecated] two phase read is replaced by topn lazy materialize.
TFetchOption is not used.
+// [deprecated] Two-phase read is replaced by TopN lazy materialization.
struct TFetchOption {
1: optional bool use_two_phase_fetch;
// Nodes in this cluster, used for second phase fetch
@@ -230,6 +230,7 @@ struct TFetchOption {
struct TResultSink {
1: optional TResultSinkType type;
2: optional TResultFileSinkOptions file_options; // deprecated
+ // [deprecated] Two-phase read is replaced by TopN lazy materialization.
3: optional TFetchOption fetch_option;
}
diff --git a/gensrc/thrift/PlanNodes.thrift b/gensrc/thrift/PlanNodes.thrift
index 2532c0cc5cd..b2702dd0ed4 100644
--- a/gensrc/thrift/PlanNodes.thrift
+++ b/gensrc/thrift/PlanNodes.thrift
@@ -996,7 +996,7 @@ struct TSortInfo {
// Expressions evaluated over the input row that materialize the tuple to be
sorted.
// Contains one expr per slot in the materialized tuple.
4: optional list<Exprs.TExpr> sort_tuple_slot_exprs
- // Indicates whether topn query using two phase read
+ // [deprecated] Two-phase read is replaced by TopN lazy materialization.
6: optional bool use_two_phase_read
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
