You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@doris.apache.org by mo...@apache.org on 2022/01/18 05:14:09 UTC
[incubator-doris] branch master updated: [improvement](Load) Cancel the load job ASAP when encounter unqualified data (#6319)
This is an automated email from the ASF dual-hosted git repository.
morningman 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 5fc0a9f [improvement](Load) Cancel the load job ASAP when encounter unqualified data (#6319)
5fc0a9f is described below
commit 5fc0a9f40dd0350ff21f20c332e3d3d29d37a227
Author: Mingyu Chen <mo...@gmail.com>
AuthorDate: Tue Jan 18 13:13:55 2022 +0800
[improvement](Load) Cancel the load job ASAP when encounter unqualified data (#6319)
This PR mainly changes:
1. Help to Cancel the load job ASAP when encounter unqualified data.
Solution is described in #6318 .
Also replace some std::stringstream with fmt::memory_buffer to avoid performance issues.
2. fix a NPE bug when create user with empty host
3. fix compile warning after rebasing the master(vectorization)
---
be/src/exec/CMakeLists.txt | 3 +-
be/src/exec/base_scanner.cpp | 66 ++--
be/src/exec/base_scanner.h | 8 +-
be/src/exec/broker_scanner.cpp | 82 +++--
be/src/exec/broker_scanner.h | 7 +-
be/src/exec/csv_scan_node.cpp | 15 +-
be/src/exec/csv_scan_node.h | 1 +
be/src/exec/csv_scanner.h | 1 +
be/src/exec/data_sink.cpp | 15 +-
be/src/exec/exec_node.cpp | 10 -
be/src/exec/json_scanner.cpp | 191 ++++++----
be/src/exec/json_scanner.h | 13 +-
be/src/exec/olap_rewrite_node.cpp | 245 -------------
be/src/exec/olap_rewrite_node.h | 70 ----
be/src/exec/orc_scanner.cpp | 10 +-
be/src/exec/orc_scanner.h | 1 -
be/src/exec/parquet_scanner.cpp | 10 +-
be/src/exec/parquet_scanner.h | 1 -
be/src/exec/tablet_sink.cpp | 125 ++++---
be/src/exec/tablet_sink.h | 6 +-
be/src/http/action/download_action.h | 1 -
be/src/http/action/stream_load.cpp | 8 +-
be/src/runtime/CMakeLists.txt | 1 -
be/src/runtime/data_spliter.cpp | 325 -----------------
be/src/runtime/data_spliter.h | 108 ------
be/src/runtime/etl_job_mgr.cpp | 1 -
be/src/runtime/plan_fragment_executor.cpp | 6 +-
be/src/runtime/runtime_state.cpp | 41 ++-
be/src/runtime/runtime_state.h | 10 +-
be/src/vec/CMakeLists.txt | 2 +-
be/src/vec/sink/vdata_stream_sender.cpp | 5 +-
.../vec/sink/{vtabet_sink.cpp => vtablet_sink.cpp} | 21 +-
be/test/exec/tablet_sink_test.cpp | 1 +
be/test/vec/runtime/vdata_stream_test.cpp | 1 -
.../org/apache/doris/analysis/UserIdentity.java | 17 +-
.../main/java/org/apache/doris/load/ExportJob.java | 2 +-
.../apache/doris/load/loadv2/BrokerLoadJob.java | 10 +-
.../load/loadv2/BrokerLoadingTaskAttachment.java | 2 +-
.../org/apache/doris/load/loadv2/BulkLoadJob.java | 4 +
.../java/org/apache/doris/load/loadv2/LoadJob.java | 6 +-
.../apache/doris/load/loadv2/LoadLoadingTask.java | 13 +-
.../load/routineload/KafkaRoutineLoadJob.java | 6 +
.../doris/load/update/UpdateStmtExecutor.java | 2 +-
.../org/apache/doris/planner/DataPartition.java | 13 -
.../org/apache/doris/planner/DataSplitSink.java | 401 ---------------------
.../org/apache/doris/planner/OlapTableSink.java | 3 +-
.../apache/doris/planner/StreamLoadPlanner.java | 8 +-
.../main/java/org/apache/doris/qe/Coordinator.java | 9 +-
.../java/org/apache/doris/qe/StmtExecutor.java | 1 +
.../java/org/apache/doris/task/LoadTaskInfo.java | 2 +
.../java/org/apache/doris/task/StreamLoadTask.java | 11 +-
.../doris/alter/MaterializedViewHandlerTest.java | 28 +-
.../java/org/apache/doris/catalog/RecoverTest.java | 17 +-
.../doris/load/loadv2/BrokerLoadJobTest.java | 14 +-
gensrc/thrift/DataSinks.thrift | 2 +-
gensrc/thrift/FrontendService.thrift | 1 +
gensrc/thrift/PaloInternalService.thrift | 3 +
gensrc/thrift/PlanNodes.thrift | 4 +-
58 files changed, 474 insertions(+), 1516 deletions(-)
diff --git a/be/src/exec/CMakeLists.txt b/be/src/exec/CMakeLists.txt
index f3856ec..30ad7eb 100644
--- a/be/src/exec/CMakeLists.txt
+++ b/be/src/exec/CMakeLists.txt
@@ -32,6 +32,8 @@ set(EXEC_FILES
base_scanner.cpp
broker_scanner.cpp
cross_join_node.cpp
+ csv_scan_node.cpp
+ csv_scanner.cpp
data_sink.cpp
decompressor.cpp
empty_set_node.cpp
@@ -48,7 +50,6 @@ set(EXEC_FILES
text_converter.cpp
topn_node.cpp
sort_exec_exprs.cpp
- olap_rewrite_node.cpp
olap_scan_node.cpp
olap_scanner.cpp
olap_common.cpp
diff --git a/be/src/exec/base_scanner.cpp b/be/src/exec/base_scanner.cpp
index 14bad74..f9fb5e3 100644
--- a/be/src/exec/base_scanner.cpp
+++ b/be/src/exec/base_scanner.cpp
@@ -17,6 +17,8 @@
#include "base_scanner.h"
+#include <fmt/format.h>
+
#include "common/logging.h"
#include "common/utils.h"
#include "exec/exec_node.h"
@@ -53,7 +55,9 @@ BaseScanner::BaseScanner(RuntimeState* state, RuntimeProfile* profile,
_profile(profile),
_rows_read_counter(nullptr),
_read_timer(nullptr),
- _materialize_timer(nullptr) {
+ _materialize_timer(nullptr),
+ _success(nullptr),
+ _scanner_eof(false) {
}
Status BaseScanner::open() {
@@ -151,11 +155,12 @@ Status BaseScanner::init_expr_ctxes() {
return Status::OK();
}
-bool BaseScanner::fill_dest_tuple(Tuple* dest_tuple, MemPool* mem_pool) {
+Status BaseScanner::fill_dest_tuple(Tuple* dest_tuple, MemPool* mem_pool) {
// filter src tuple by preceding filter first
if (!ExecNode::eval_conjuncts(&_pre_filter_ctxs[0], _pre_filter_ctxs.size(), _src_tuple_row)) {
_counter->num_rows_unselected++;
- return false;
+ _success = false;
+ return Status::OK();
}
// convert and fill dest tuple
@@ -172,41 +177,47 @@ bool BaseScanner::fill_dest_tuple(Tuple* dest_tuple, MemPool* mem_pool) {
// Only when the expr return value is null, we will check the error message.
std::string expr_error = ctx->get_error_msg();
if (!expr_error.empty()) {
- _state->append_error_msg_to_file(_src_tuple_row->to_string(*(_row_desc.get())),
- expr_error);
+ RETURN_IF_ERROR(_state->append_error_msg_to_file([&]() -> std::string { return _src_tuple_row->to_string(*(_row_desc.get())); },
+ [&]() -> std::string { return expr_error; }, &_scanner_eof));
_counter->num_rows_filtered++;
// The ctx is reused, so must clear the error state and message.
ctx->clear_error_msg();
- return false;
+ _success = false;
+ return Status::OK();
}
// If _strict_mode is false, _src_slot_descs_order_by_dest size could be zero
if (_strict_mode && (_src_slot_descs_order_by_dest[dest_index] != nullptr) &&
!_src_tuple->is_null(
_src_slot_descs_order_by_dest[dest_index]->null_indicator_offset())) {
- //Type of the slot is must be Varchar in _src_tuple.
- StringValue* raw_value = _src_tuple->get_string_slot(
- _src_slot_descs_order_by_dest[dest_index]->tuple_offset());
- std::string raw_string;
- if (raw_value != nullptr) { //is not null then get raw value
- raw_string = raw_value->to_string();
- }
- std::stringstream error_msg;
- error_msg << "column(" << slot_desc->col_name() << ") value is incorrect "
- << "while strict mode is " << std::boolalpha << _strict_mode
- << ", src value is " << raw_string;
- _state->append_error_msg_to_file(_src_tuple_row->to_string(*(_row_desc.get())),
- error_msg.str());
+
+ RETURN_IF_ERROR(_state->append_error_msg_to_file([&]() -> std::string { return _src_tuple_row->to_string(*(_row_desc.get())); },
+ [&]() -> std::string {
+ // Type of the slot is must be Varchar in _src_tuple.
+ StringValue* raw_value = _src_tuple->get_string_slot(
+ _src_slot_descs_order_by_dest[dest_index]->tuple_offset());
+ std::string raw_string;
+ if (raw_value != nullptr) { //is not null then get raw value
+ raw_string = raw_value->to_string();
+ }
+ fmt::memory_buffer error_msg;
+ fmt::format_to(error_msg, "column({}) value is incorrect while strict mode is {}, src value is {}",
+ slot_desc->col_name(), _strict_mode, raw_string);
+ return error_msg.data();
+ }, &_scanner_eof));
_counter->num_rows_filtered++;
- return false;
+ _success = false;
+ return Status::OK();
}
if (!slot_desc->is_nullable()) {
- std::stringstream error_msg;
- error_msg << "column(" << slot_desc->col_name() << ") value is null "
- << "while columns is not nullable";
- _state->append_error_msg_to_file(_src_tuple_row->to_string(*(_row_desc.get())),
- error_msg.str());
+ RETURN_IF_ERROR(_state->append_error_msg_to_file([&]() -> std::string { return _src_tuple_row->to_string(*(_row_desc.get())); },
+ [&]() -> std::string {
+ fmt::memory_buffer error_msg;
+ fmt::format_to(error_msg, "column({}) values is null while columns is not nullable", slot_desc->col_name());
+ return error_msg.data();
+ }, &_scanner_eof));
_counter->num_rows_filtered++;
- return false;
+ _success = false;
+ return Status::OK();
}
dest_tuple->set_null(slot_desc->null_indicator_offset());
continue;
@@ -218,7 +229,8 @@ bool BaseScanner::fill_dest_tuple(Tuple* dest_tuple, MemPool* mem_pool) {
RawValue::write(value, slot, slot_desc->type(), mem_pool);
continue;
}
- return true;
+ _success = true;
+ return Status::OK();
}
void BaseScanner::fill_slots_of_columns_from_path(
diff --git a/be/src/exec/base_scanner.h b/be/src/exec/base_scanner.h
index 9520eab..a745e93 100644
--- a/be/src/exec/base_scanner.h
+++ b/be/src/exec/base_scanner.h
@@ -57,8 +57,8 @@ public:
virtual Status get_next(Tuple* tuple, MemPool* tuple_pool, bool* eof, bool *fill_tuple) = 0;
// Close this scanner
- virtual void close();
- bool fill_dest_tuple(Tuple* dest_tuple, MemPool* mem_pool);
+ virtual void close() = 0;
+ Status fill_dest_tuple(Tuple* dest_tuple, MemPool* mem_pool);
void fill_slots_of_columns_from_path(int start,
const std::vector<std::string>& columns_from_path);
@@ -102,6 +102,10 @@ protected:
RuntimeProfile::Counter* _rows_read_counter;
RuntimeProfile::Counter* _read_timer;
RuntimeProfile::Counter* _materialize_timer;
+
+ // Used to record whether a row of data is successfully read.
+ bool _success = false;
+ bool _scanner_eof = false;
};
} /* namespace doris */
diff --git a/be/src/exec/broker_scanner.cpp b/be/src/exec/broker_scanner.cpp
index 5791b9a..ab96e6a 100644
--- a/be/src/exec/broker_scanner.cpp
+++ b/be/src/exec/broker_scanner.cpp
@@ -17,6 +17,7 @@
#include "exec/broker_scanner.h"
+#include <fmt/format.h>
#include <iostream>
#include <sstream>
@@ -57,7 +58,6 @@ BrokerScanner::BrokerScanner(RuntimeState* state, RuntimeProfile* profile,
_cur_decompressor(nullptr),
_next_range(0),
_cur_line_reader_eof(false),
- _scanner_eof(false),
_skip_next_line(false) {
if (params.__isset.column_separator_length && params.column_separator_length > 1) {
_value_separator = params.column_separator_str;
@@ -114,7 +114,8 @@ Status BrokerScanner::get_next(Tuple* tuple, MemPool* tuple_pool, bool* eof, boo
{
COUNTER_UPDATE(_rows_read_counter, 1);
SCOPED_TIMER(_materialize_timer);
- if (convert_one_row(Slice(ptr, size), tuple, tuple_pool)) {
+ RETURN_IF_ERROR(_convert_one_row(Slice(ptr, size), tuple, tuple_pool));
+ if (_success) {
free_expr_local_allocations();
*fill_tuple = true;
} else {
@@ -460,23 +461,29 @@ bool is_null(const Slice& slice) {
}
// Convert one row to this tuple
-bool BrokerScanner::convert_one_row(const Slice& line, Tuple* tuple, MemPool* tuple_pool) {
- if (!line_to_src_tuple(line)) {
- return false;
+Status BrokerScanner::_convert_one_row(const Slice& line, Tuple* tuple, MemPool* tuple_pool) {
+ RETURN_IF_ERROR(_line_to_src_tuple(line));
+ if (!_success) {
+ // If not success, which means we met an invalid row, return.
+ return Status::OK();
}
return fill_dest_tuple(tuple, tuple_pool);
}
// Convert one row to this tuple
-bool BrokerScanner::line_to_src_tuple(const Slice& line) {
- if (_file_format_type != TFileFormatType::FORMAT_PROTO &&
- !validate_utf8(line.data, line.size)) {
- std::stringstream error_msg;
- error_msg << "data is not encoded by UTF-8";
- _state->append_error_msg_to_file("Unable to display", error_msg.str());
+Status BrokerScanner::_line_to_src_tuple(const Slice& line) {
+ bool is_proto_format = _file_format_type == TFileFormatType::FORMAT_PROTO;
+ if (!is_proto_format && !validate_utf8(line.data, line.size)) {
+ RETURN_IF_ERROR(_state->append_error_msg_to_file([]() -> std::string { return "Unable to display"; },
+ []() -> std::string {
+ fmt::memory_buffer error_msg;
+ fmt::format_to(error_msg, "{}", "Unable to display");
+ return error_msg.data();
+ }, &_scanner_eof));
_counter->num_rows_filtered++;
- return false;
+ _success = false;
+ return Status::OK();
}
split_line(line);
@@ -490,7 +497,7 @@ bool BrokerScanner::line_to_src_tuple(const Slice& line) {
const std::vector<std::string>& columns_from_path = range.columns_from_path;
// read data by column defination, resize _split_values to _src_solt_size
if (read_by_column_def) {
- // fill solts by NULL
+ // fill slots by NULL
while (_split_values.size() + columns_from_path.size() < _src_slot_descs.size()) {
_split_values.emplace_back(_split_values.back().get_data(), 0);
}
@@ -500,33 +507,31 @@ bool BrokerScanner::line_to_src_tuple(const Slice& line) {
}
} else {
if (_split_values.size() + columns_from_path.size() < _src_slot_descs.size()) {
- std::stringstream error_msg;
- error_msg << "actual column number is less than schema column number. "
- << "actual number: " << _split_values.size() << " column separator: ["
- << _value_separator << "], "
- << "line delimiter: [" << _line_delimiter << "], "
- << "schema number: " << _src_slot_descs.size() << "; ";
- if (_file_format_type == TFileFormatType::FORMAT_PROTO) {
- _state->append_error_msg_to_file("", error_msg.str());
- } else {
- _state->append_error_msg_to_file(std::string(line.data, line.size), error_msg.str());
- }
+ RETURN_IF_ERROR(_state->append_error_msg_to_file(
+ [&]() -> std::string { return is_proto_format ? "" : std::string(line.data, line.size); },
+ [&]() -> std::string {
+ fmt::memory_buffer error_msg;
+ fmt::format_to(error_msg, "{}", "actual column number is less than schema column number.");
+ fmt::format_to(error_msg, "actual number: {}, column separator: [{}], ", _split_values.size(), _value_separator);
+ fmt::format_to(error_msg, "line delimiter: [{}], schema number: {}; ", _line_delimiter, _src_slot_descs.size());
+ return error_msg.data();
+ }, &_scanner_eof));
_counter->num_rows_filtered++;
- return false;
+ _success = false;
+ return Status::OK();
} else if (_split_values.size() + columns_from_path.size() > _src_slot_descs.size()) {
- std::stringstream error_msg;
- error_msg << "actual column number is more than schema column number. "
- << "actual number: " << _split_values.size() << " column separator: ["
- << _value_separator << "], "
- << "line delimiter: [" << _line_delimiter << "], "
- << "schema number: " << _src_slot_descs.size() << "; ";
- if (_file_format_type == TFileFormatType::FORMAT_PROTO) {
- _state->append_error_msg_to_file("", error_msg.str());
- } else {
- _state->append_error_msg_to_file(std::string(line.data, line.size), error_msg.str());
- }
+ RETURN_IF_ERROR(_state->append_error_msg_to_file(
+ [&]() -> std::string { return is_proto_format ? "" : std::string(line.data, line.size); },
+ [&]() -> std::string {
+ fmt::memory_buffer error_msg;
+ fmt::format_to(error_msg, "{}", "actual column number is more than schema column number.");
+ fmt::format_to(error_msg, "actual number: {}, column separator: [{}], ", _split_values.size(), _value_separator);
+ fmt::format_to(error_msg, "line delimiter: [{}], schema number: {}; ", _line_delimiter, _src_slot_descs.size());
+ return error_msg.data();
+ }, &_scanner_eof));
_counter->num_rows_filtered++;
- return false;
+ _success = false;
+ return Status::OK();
}
}
@@ -548,7 +553,8 @@ bool BrokerScanner::line_to_src_tuple(const Slice& line) {
fill_slots_of_columns_from_path(range.num_of_columns_from_file, columns_from_path);
}
- return true;
+ _success = true;
+ return Status::OK();
}
} // namespace doris
diff --git a/be/src/exec/broker_scanner.h b/be/src/exec/broker_scanner.h
index 8debcb1..ca66cd0 100644
--- a/be/src/exec/broker_scanner.h
+++ b/be/src/exec/broker_scanner.h
@@ -87,10 +87,9 @@ private:
// Convert one row to one tuple
// 'ptr' and 'len' is csv text line
// output is tuple
- bool convert_one_row(const Slice& line, Tuple* tuple, MemPool* tuple_pool);
+ Status _convert_one_row(const Slice& line, Tuple* tuple, MemPool* tuple_pool);
- Status line_to_src_tuple();
- bool line_to_src_tuple(const Slice& line);
+ Status _line_to_src_tuple(const Slice& line);
private:
const std::vector<TBrokerRangeDesc>& _ranges;
@@ -111,8 +110,6 @@ private:
int _next_range;
bool _cur_line_reader_eof;
- bool _scanner_eof;
-
// When we fetch range doesn't start from 0,
// we will read to one ahead, and skip the first line
bool _skip_next_line;
diff --git a/be/src/exec/csv_scan_node.cpp b/be/src/exec/csv_scan_node.cpp
index f106400..e005b81 100644
--- a/be/src/exec/csv_scan_node.cpp
+++ b/be/src/exec/csv_scan_node.cpp
@@ -330,8 +330,7 @@ Status CsvScanNode::close(RuntimeState* state) {
if (state->num_rows_load_success() == 0) {
std::stringstream error_msg;
error_msg << "Read zero normal line file. ";
- state->append_error_msg_to_file("", error_msg.str(), true);
-
+ LOG(INFO) << error_msg.str();
return Status::InternalError(error_msg.str());
}
@@ -341,7 +340,7 @@ Status CsvScanNode::close(RuntimeState* state) {
std::stringstream summary_msg;
summary_msg << "error line: " << _num_rows_load_filtered
<< "; normal line: " << state->num_rows_load_success();
- state->append_error_msg_to_file("", summary_msg.str(), true);
+ LOG(INFO) << summary_msg.str();
}
return Status::OK();
@@ -526,13 +525,13 @@ bool CsvScanNode::split_check_fill(const std::string& line, RuntimeState* state)
error_msg << "actual column number is less than schema column number. "
<< "actual number: " << fields.size() << " ,"
<< "schema number: " << _columns.size() << "; ";
- _runtime_state->append_error_msg_to_file(line, error_msg.str());
+ LOG(INFO) << error_msg.str();
return false;
} else if (_hll_column_num == 0 && fields.size() > _columns.size()) {
error_msg << "actual column number is more than schema column number. "
<< "actual number: " << fields.size() << " ,"
<< "schema number: " << _columns.size() << "; ";
- _runtime_state->append_error_msg_to_file(line, error_msg.str());
+ LOG(INFO) << error_msg.str();
return false;
}
@@ -557,7 +556,7 @@ bool CsvScanNode::split_check_fill(const std::string& line, RuntimeState* state)
fields[i].length(), slot, state, &error_msg);
if (flag == false) {
- _runtime_state->append_error_msg_to_file(line, error_msg.str());
+ LOG(INFO) << error_msg.str();;
return false;
}
}
@@ -582,7 +581,7 @@ bool CsvScanNode::split_check_fill(const std::string& line, RuntimeState* state)
_default_values[i].length(), slot, state, &error_msg);
if (flag == false) {
- _runtime_state->append_error_msg_to_file(line, error_msg.str());
+ LOG(INFO) << error_msg.str();
return false;
}
}
@@ -599,7 +598,7 @@ bool CsvScanNode::split_check_fill(const std::string& line, RuntimeState* state)
bool flag = check_and_write_text_slot(column_name, column_type, column_string.c_str(),
column_string.length(), slot, state, &error_msg);
if (flag == false) {
- _runtime_state->append_error_msg_to_file(line, error_msg.str());
+ LOG(INFO) << error_msg.str();
return false;
}
}
diff --git a/be/src/exec/csv_scan_node.h b/be/src/exec/csv_scan_node.h
index fe92247..36df97f 100644
--- a/be/src/exec/csv_scan_node.h
+++ b/be/src/exec/csv_scan_node.h
@@ -35,6 +35,7 @@ class RuntimeState;
class MemPool;
class Status;
+// Now, CsvScanNode and CsvScanner are only for unit test
class CsvScanNode : public ScanNode {
public:
CsvScanNode(ObjectPool* pool, const TPlanNode& tnode, const DescriptorTbl& descs);
diff --git a/be/src/exec/csv_scanner.h b/be/src/exec/csv_scanner.h
index 3de2fef..98259c3 100644
--- a/be/src/exec/csv_scanner.h
+++ b/be/src/exec/csv_scanner.h
@@ -26,6 +26,7 @@
namespace doris {
+// Now, CsvScanNode and CsvScanner are only for unit test
class CsvScanner {
public:
CsvScanner(const std::vector<std::string>& csv_file_paths);
diff --git a/be/src/exec/data_sink.cpp b/be/src/exec/data_sink.cpp
index cf06f09..22ec175 100644
--- a/be/src/exec/data_sink.cpp
+++ b/be/src/exec/data_sink.cpp
@@ -26,7 +26,6 @@
#include "exec/tablet_sink.h"
#include "exprs/expr.h"
#include "gen_cpp/PaloInternalService_types.h"
-#include "runtime/data_spliter.h"
#include "runtime/data_stream_sender.h"
#include "runtime/export_sink.h"
#include "runtime/memory_scratch_sink.h"
@@ -132,17 +131,6 @@ Status DataSink::create_data_sink(ObjectPool* pool, const TDataSink& thrift_sink
sink->reset(odbc_tbl_sink);
break;
}
- case TDataSinkType::DATA_SPLIT_SINK: {
- if (!thrift_sink.__isset.split_sink) {
- return Status::InternalError("Missing data split buffer sink.");
- }
-
- // TODO: figure out good buffer size based on size of output row
- std::unique_ptr<DataSpliter> data_spliter(new DataSpliter(row_desc));
- RETURN_IF_ERROR(DataSpliter::from_thrift(pool, thrift_sink.split_sink, data_spliter.get()));
- sink->reset(data_spliter.release());
- break;
- }
case TDataSinkType::EXPORT_SINK: {
if (!thrift_sink.__isset.export_sink) {
@@ -157,7 +145,8 @@ Status DataSink::create_data_sink(ObjectPool* pool, const TDataSink& thrift_sink
Status status;
DCHECK(thrift_sink.__isset.olap_table_sink);
if (is_vec) {
- sink->reset(new stream_load::VOlapTableSink(pool, row_desc, output_exprs, &status));
+ // sink->reset(new stream_load::VOlapTableSink(pool, row_desc, output_exprs, &status));
+ return Status::NotSupported("VOlapTableSink is not supported yet");
} else {
sink->reset(new stream_load::OlapTableSink(pool, row_desc, output_exprs, &status));
}
diff --git a/be/src/exec/exec_node.cpp b/be/src/exec/exec_node.cpp
index 97c3259..4501d42 100644
--- a/be/src/exec/exec_node.cpp
+++ b/be/src/exec/exec_node.cpp
@@ -29,7 +29,6 @@
#include "exec/assert_num_rows_node.h"
#include "exec/broker_scan_node.h"
#include "exec/cross_join_node.h"
-#include "exec/csv_scan_node.h"
#include "exec/empty_set_node.h"
#include "exec/es_http_scan_node.h"
#include "exec/es_scan_node.h"
@@ -41,7 +40,6 @@
#include "exec/merge_node.h"
#include "exec/mysql_scan_node.h"
#include "exec/odbc_scan_node.h"
-#include "exec/olap_rewrite_node.h"
#include "exec/olap_scan_node.h"
#include "exec/partitioned_aggregation_node.h"
#include "exec/repeat_node.h"
@@ -407,10 +405,6 @@ Status ExecNode::create_node(RuntimeState* state, ObjectPool* pool, const TPlanN
VLOG_CRITICAL << "tnode:\n" << apache::thrift::ThriftDebugString(tnode);
switch (tnode.node_type) {
- case TPlanNodeType::CSV_SCAN_NODE:
- *node = pool->add(new CsvScanNode(pool, tnode, descs));
- return Status::OK();
-
case TPlanNodeType::MYSQL_SCAN_NODE:
#ifdef DORIS_WITH_MYSQL
if (state->enable_vectorized_exec()) {
@@ -513,10 +507,6 @@ Status ExecNode::create_node(RuntimeState* state, ObjectPool* pool, const TPlanN
}
return Status::OK();
- case TPlanNodeType::OLAP_REWRITE_NODE:
- *node = pool->add(new OlapRewriteNode(pool, tnode, descs));
- return Status::OK();
-
case TPlanNodeType::SORT_NODE:
if (state->enable_vectorized_exec()) {
*node = pool->add(new vectorized::VSortNode(pool, tnode, descs));
diff --git a/be/src/exec/json_scanner.cpp b/be/src/exec/json_scanner.cpp
index 134d646..eaff43f 100644
--- a/be/src/exec/json_scanner.cpp
+++ b/be/src/exec/json_scanner.cpp
@@ -18,6 +18,7 @@
#include "exec/json_scanner.h"
#include <algorithm>
+#include <fmt/format.h>
#include "env/env.h"
#include "exec/broker_reader.h"
@@ -47,7 +48,6 @@ JsonScanner::JsonScanner(RuntimeState* state, RuntimeProfile* profile,
_cur_json_reader(nullptr),
_next_range(0),
_cur_reader_eof(false),
- _scanner_eof(false),
_read_json_by_line(false) {
if (params.__isset.line_delimiter_length && params.line_delimiter_length > 1) {
_line_delimiter = params.line_delimiter_str;
@@ -96,11 +96,8 @@ Status JsonScanner::get_next(Tuple* tuple, MemPool* tuple_pool, bool* eof, bool
}
COUNTER_UPDATE(_rows_read_counter, 1);
SCOPED_TIMER(_materialize_timer);
- if (fill_dest_tuple(tuple, tuple_pool)) {
- *fill_tuple = true;
- } else {
- *fill_tuple = false;
- }
+ RETURN_IF_ERROR(fill_dest_tuple(tuple, tuple_pool));
+ *fill_tuple = _success;
break; // break always
}
if (_scanner_eof) {
@@ -239,10 +236,10 @@ Status JsonScanner::open_json_reader() {
if (_read_json_by_line) {
_cur_json_reader =
new JsonReader(_state, _counter, _profile, strip_outer_array, num_as_string,
- fuzzy_parse, nullptr, _cur_line_reader);
+ fuzzy_parse, &_scanner_eof, nullptr, _cur_line_reader);
} else {
_cur_json_reader = new JsonReader(_state, _counter, _profile, strip_outer_array, num_as_string,
- fuzzy_parse, _cur_file_reader);
+ fuzzy_parse, &_scanner_eof, _cur_file_reader);
}
RETURN_IF_ERROR(_cur_json_reader->init(jsonpath, json_root));
@@ -286,6 +283,7 @@ rapidjson::Value::ConstValueIterator JsonDataInternal::get_next() {
////// class JsonReader
JsonReader::JsonReader(RuntimeState* state, ScannerCounter* counter, RuntimeProfile* profile,
bool strip_outer_array, bool num_as_string,bool fuzzy_parse,
+ bool* scanner_eof,
FileReader* file_reader, LineReader* line_reader)
: _handle_json_callback(nullptr),
_next_line(0),
@@ -299,7 +297,8 @@ JsonReader::JsonReader(RuntimeState* state, ScannerCounter* counter, RuntimeProf
_strip_outer_array(strip_outer_array),
_num_as_string(num_as_string),
_fuzzy_parse(fuzzy_parse),
- _json_doc(nullptr) {
+ _json_doc(nullptr),
+ _scanner_eof(scanner_eof) {
_bytes_read_counter = ADD_COUNTER(_profile, "BytesRead", TUnit::BYTES);
_read_timer = ADD_TIMER(_profile, "ReadTime");
_file_read_timer = ADD_TIMER(_profile, "FileReadTime");
@@ -362,7 +361,7 @@ void JsonReader::_close() {
_closed = true;
}
-// read one json string from line read or file read and parse it to json doc.
+// read one json string from line reader or file reader and parse it to json doc.
// return Status::DataQualityError() if data has quality error.
// return other error if encounter other problemes.
// return Status::OK() if parse succeed or reach EOF.
@@ -403,14 +402,20 @@ Status JsonReader::_parse_json_doc(size_t* size, bool* eof) {
}
if (has_parse_error) {
- std::stringstream str_error;
- str_error << "Parse json data for JsonDoc failed. code = "
- << _origin_json_doc.GetParseError() << ", error-info:"
- << rapidjson::GetParseError_En(_origin_json_doc.GetParseError());
- _state->append_error_msg_to_file(std::string((char*)json_str, *size),
- str_error.str());
+ fmt::memory_buffer error_msg;
+ fmt::format_to(error_msg, "Parse json data for JsonDoc failed. code: {}, error info: {}",
+ _origin_json_doc.GetParseError(), rapidjson::GetParseError_En(_origin_json_doc.GetParseError()));
+ RETURN_IF_ERROR(_state->append_error_msg_to_file([&]() -> std::string { return std::string((char*)json_str, *size); },
+ [&]() -> std::string { return error_msg.data(); }, _scanner_eof));
_counter->num_rows_filtered++;
- return Status::DataQualityError(str_error.str());
+ if (*_scanner_eof) {
+ // Case A: if _scanner_eof is set to true in "append_error_msg_to_file", which means
+ // we meet enough invalid rows and the scanner should be stopped.
+ // So we set eof to true and return OK, the caller will stop the process as we meet the end of file.
+ *eof = true;
+ return Status::OK();
+ }
+ return Status::DataQualityError(error_msg.data());
}
// set json root
@@ -418,30 +423,48 @@ Status JsonReader::_parse_json_doc(size_t* size, bool* eof) {
_json_doc = JsonFunctions::get_json_object_from_parsed_json(
_parsed_json_root, &_origin_json_doc, _origin_json_doc.GetAllocator());
if (_json_doc == nullptr) {
- std::stringstream str_error;
- str_error << "JSON Root not found.";
- _state->append_error_msg_to_file(_print_json_value(_origin_json_doc), str_error.str());
+ fmt::memory_buffer error_msg;
+ fmt::format_to(error_msg, "{}", "JSON Root not found.");
+ RETURN_IF_ERROR(_state->append_error_msg_to_file([&]() -> std::string { return _print_json_value(_origin_json_doc); },
+ [&]() -> std::string { return error_msg.data(); }, _scanner_eof));
_counter->num_rows_filtered++;
- return Status::DataQualityError(str_error.str());
+ if (*_scanner_eof) {
+ // Same as Case A
+ *eof = true;
+ return Status::OK();
+ }
+ return Status::DataQualityError(error_msg.data());
}
} else {
_json_doc = &_origin_json_doc;
}
if (_json_doc->IsArray() && !_strip_outer_array) {
- std::stringstream str_error;
- str_error << "JSON data is array-object, `strip_outer_array` must be TRUE.";
- _state->append_error_msg_to_file(_print_json_value(_origin_json_doc), str_error.str());
+ fmt::memory_buffer error_msg;
+ fmt::format_to(error_msg, "{}", "JSON data is array-object, `strip_outer_array` must be TRUE.");
+ RETURN_IF_ERROR(_state->append_error_msg_to_file([&]() -> std::string { return _print_json_value(_origin_json_doc); },
+ [&]() -> std::string { return error_msg.data(); }, _scanner_eof));
_counter->num_rows_filtered++;
- return Status::DataQualityError(str_error.str());
+ if (*_scanner_eof) {
+ // Same as Case A
+ *eof = true;
+ return Status::OK();
+ }
+ return Status::DataQualityError(error_msg.data());
}
if (!_json_doc->IsArray() && _strip_outer_array) {
- std::stringstream str_error;
- str_error << "JSON data is not an array-object, `strip_outer_array` must be FALSE.";
- _state->append_error_msg_to_file(_print_json_value(_origin_json_doc), str_error.str());
+ fmt::memory_buffer error_msg;
+ fmt::format_to(error_msg, "{}", "JSON data is not an array-object, `strip_outer_array` must be FALSE.");
+ RETURN_IF_ERROR(_state->append_error_msg_to_file([&]() -> std::string { return _print_json_value(_origin_json_doc); },
+ [&]() -> std::string { return error_msg.data(); }, _scanner_eof));
_counter->num_rows_filtered++;
- return Status::DataQualityError(str_error.str());
+ if (*_scanner_eof) {
+ // Same as Case A
+ *eof = true;
+ return Status::OK();
+ }
+ return Status::DataQualityError(error_msg.data());
}
return Status::OK();
@@ -473,7 +496,7 @@ void JsonReader::_fill_slot(Tuple* tuple, SlotDescriptor* slot_desc, MemPool* me
str_slot->len = len;
}
-void JsonReader::_write_data_to_tuple(rapidjson::Value::ConstValueIterator value,
+Status JsonReader::_write_data_to_tuple(rapidjson::Value::ConstValueIterator value,
SlotDescriptor* desc, Tuple* tuple, MemPool* tuple_pool,
bool* valid) {
const char* str_value = nullptr;
@@ -512,13 +535,15 @@ void JsonReader::_write_data_to_tuple(rapidjson::Value::ConstValueIterator value
if (desc->is_nullable()) {
tuple->set_null(desc->null_indicator_offset());
} else {
- std::stringstream str_error;
- str_error << "Json value is null, but the column `" << desc->col_name()
- << "` is not nullable.";
- _state->append_error_msg_to_file(_print_json_value(*value), str_error.str());
+ RETURN_IF_ERROR(_state->append_error_msg_to_file([&]() -> std::string { return _print_json_value(*value); },
+ [&]() -> std::string {
+ fmt::memory_buffer error_msg;
+ fmt::format_to(error_msg, "Json value is null, but the column `{}` is not nullable.", desc->col_name());
+ return error_msg.data();
+ }, _scanner_eof));
_counter->num_rows_filtered++;
*valid = false;
- return;
+ return Status::OK();
}
break;
default:
@@ -528,20 +553,24 @@ void JsonReader::_write_data_to_tuple(rapidjson::Value::ConstValueIterator value
break;
}
*valid = true;
+ return Status::OK();
}
// for simple format json
-void JsonReader::_set_tuple_value(rapidjson::Value& objectValue, Tuple* tuple,
+// set valid to true and return OK if succeed.
+// set valid to false and return OK if we met an invalid row.
+// return other status if encounter other problmes.
+Status JsonReader::_set_tuple_value(rapidjson::Value& objectValue, Tuple* tuple,
const std::vector<SlotDescriptor*>& slot_descs,
MemPool* tuple_pool, bool* valid) {
if (!objectValue.IsObject()) {
// Here we expect the incoming `objectValue` to be a Json Object, such as {"key" : "value"},
// not other type of Json format.
- _state->append_error_msg_to_file(_print_json_value(objectValue),
- "Expect json object value");
+ RETURN_IF_ERROR(_state->append_error_msg_to_file([&]() -> std::string { return _print_json_value(objectValue); },
+ [&]() -> std::string { return "Expect json object value"; }, _scanner_eof));
_counter->num_rows_filtered++;
*valid = false; // current row is invalid
- return;
+ return Status::OK();
}
int nullcount = 0;
@@ -558,19 +587,21 @@ void JsonReader::_set_tuple_value(rapidjson::Value& objectValue, Tuple* tuple,
}
if (it != objectValue.MemberEnd()) {
const rapidjson::Value& value = it->value;
- _write_data_to_tuple(&value, v, tuple, tuple_pool, valid);
+ RETURN_IF_ERROR(_write_data_to_tuple(&value, v, tuple, tuple_pool, valid));
if (!(*valid)) {
- return;
+ return Status::OK();
}
} else { // not found
if (v->is_nullable()) {
tuple->set_null(v->null_indicator_offset());
nullcount++;
} else {
- std::stringstream str_error;
- str_error << "The column `" << v->col_name()
- << "` is not nullable, but it's not found in jsondata.";
- _state->append_error_msg_to_file(_print_json_value(objectValue), str_error.str());
+ RETURN_IF_ERROR( _state->append_error_msg_to_file([&]() -> std::string { return _print_json_value(objectValue); },
+ [&]() -> std::string {
+ fmt::memory_buffer error_msg;
+ fmt::format_to(error_msg, "The column `{}` is not nullable, but it's not found in jsondata.", v->col_name());
+ return error_msg.data();
+ }, _scanner_eof));
_counter->num_rows_filtered++;
*valid = false; // current row is invalid
break;
@@ -579,14 +610,14 @@ void JsonReader::_set_tuple_value(rapidjson::Value& objectValue, Tuple* tuple,
}
if (nullcount == slot_descs.size()) {
- _state->append_error_msg_to_file(_print_json_value(objectValue),
- "All fields is null, this is a invalid row.");
+ RETURN_IF_ERROR(_state->append_error_msg_to_file([&]() -> std::string { return _print_json_value(objectValue); },
+ [&]() -> std::string { return "All fields is null, this is a invalid row."; }, _scanner_eof));
_counter->num_rows_filtered++;
*valid = false;
- return;
+ return Status::OK();
}
*valid = true;
- return;
+ return Status::OK();
}
/**
@@ -618,11 +649,13 @@ Status JsonReader::_handle_simple_json(Tuple* tuple, const std::vector<SlotDescr
_total_lines = _json_doc->Size();
if (_total_lines == 0) {
// may be passing an empty json, such as "[]"
- std::stringstream str_error;
- str_error << "Empty json line";
- _state->append_error_msg_to_file(_print_json_value(*_json_doc),
- str_error.str());
+ RETURN_IF_ERROR(_state->append_error_msg_to_file([&]() -> std::string { return _print_json_value(*_json_doc); },
+ [&]() -> std::string { return "Empty json line"; }, _scanner_eof));
_counter->num_rows_filtered++;
+ if (*_scanner_eof) {
+ *is_empty_row = true;
+ return Status::OK();
+ }
continue;
}
objectValue = &(*_json_doc)[0];
@@ -646,12 +679,18 @@ Status JsonReader::_handle_simple_json(Tuple* tuple, const std::vector<SlotDescr
if (_json_doc->IsArray()) { // handle case 1
rapidjson::Value& objectValue = (*_json_doc)[_next_line]; // json object
- _set_tuple_value(objectValue, tuple, slot_descs, tuple_pool, &valid);
+ RETURN_IF_ERROR(_set_tuple_value(objectValue, tuple, slot_descs, tuple_pool, &valid));
} else { // handle case 2
- _set_tuple_value(*_json_doc, tuple, slot_descs, tuple_pool, &valid);
+ RETURN_IF_ERROR(_set_tuple_value(*_json_doc, tuple, slot_descs, tuple_pool, &valid));
}
_next_line++;
if (!valid) {
+ if (*_scanner_eof) {
+ // When _scanner_eof is true and valid is false, it means that we have encountered
+ // unqualified data and decided to stop the scan.
+ *is_empty_row = true;
+ return Status::OK();
+ }
continue;
}
*is_empty_row = false;
@@ -660,13 +699,12 @@ Status JsonReader::_handle_simple_json(Tuple* tuple, const std::vector<SlotDescr
return Status::OK();
}
-bool JsonReader::_write_values_by_jsonpath(rapidjson::Value& objectValue, MemPool* tuple_pool,
- Tuple* tuple,
- const std::vector<SlotDescriptor*>& slot_descs) {
+Status JsonReader::_write_values_by_jsonpath(rapidjson::Value& objectValue, MemPool* tuple_pool,
+ Tuple* tuple,
+ const std::vector<SlotDescriptor*>& slot_descs,
+ bool* valid) {
int nullcount = 0;
- bool valid = true;
size_t column_num = slot_descs.size();
-
for (size_t i = 0; i < column_num; i++) {
rapidjson::Value* json_values = nullptr;
bool wrap_explicitly = false;
@@ -681,12 +719,14 @@ bool JsonReader::_write_values_by_jsonpath(rapidjson::Value& objectValue, MemPoo
tuple->set_null(slot_descs[i]->null_indicator_offset());
nullcount++;
} else {
- std::stringstream str_error;
- str_error << "The column `" << slot_descs[i]->col_name()
- << "` is not nullable, but it's not found in jsondata.";
- _state->append_error_msg_to_file(_print_json_value(objectValue), str_error.str());
+ RETURN_IF_ERROR(_state->append_error_msg_to_file([&]() -> std::string { return _print_json_value(objectValue); },
+ [&]() -> std::string {
+ fmt::memory_buffer error_msg;
+ fmt::format_to(error_msg, "The column `{}` is not nullable, but it's not found in jsondata.", slot_descs[i]->col_name());
+ return error_msg.data();
+ }, _scanner_eof));
_counter->num_rows_filtered++;
- valid = false; // current row is invalid
+ *valid = false; // current row is invalid
break;
}
} else {
@@ -697,20 +737,19 @@ bool JsonReader::_write_values_by_jsonpath(rapidjson::Value& objectValue, MemPoo
// if json_values' size > 1, it means we just match an array, not a wrapped one, so no need to unwrap.
json_values = &((*json_values)[0]);
}
- _write_data_to_tuple(json_values, slot_descs[i], tuple, tuple_pool, &valid);
- if (!valid) {
+ RETURN_IF_ERROR(_write_data_to_tuple(json_values, slot_descs[i], tuple, tuple_pool, valid));
+ if (!(*valid)) {
break;
}
}
}
if (nullcount == column_num) {
- _state->append_error_msg_to_file(
- _print_json_value(objectValue),
- "All fields is null or not matched, this is a invalid row.");
+ RETURN_IF_ERROR(_state->append_error_msg_to_file([&]() -> std::string { return _print_json_value(objectValue); },
+ [&]() -> std::string { return "All fields is null or not matched, this is a invalid row."; }, _scanner_eof));
_counter->num_rows_filtered++;
- valid = false;
+ *valid = false;
}
- return valid;
+ return Status::OK();
}
/**
@@ -737,7 +776,9 @@ Status JsonReader::_handle_nested_complex_json(Tuple* tuple,
*is_empty_row = false;
break; // read a valid row
}
- if (!_write_values_by_jsonpath(*_json_doc, tuple_pool, tuple, slot_descs)) {
+ bool valid = true;
+ RETURN_IF_ERROR(_write_values_by_jsonpath(*_json_doc, tuple_pool, tuple, slot_descs, &valid));
+ if (!valid) {
// there is only one line in this case, so if it return false, just set is_empty_row true
// so that the caller will continue reading next line.
*is_empty_row = true;
@@ -780,7 +821,9 @@ Status JsonReader::_handle_flat_array_complex_json(Tuple* tuple,
}
}
rapidjson::Value& objectValue = (*_json_doc)[_next_line++];
- if (!_write_values_by_jsonpath(objectValue, tuple_pool, tuple, slot_descs)) {
+ bool valid = true;
+ RETURN_IF_ERROR(_write_values_by_jsonpath(objectValue, tuple_pool, tuple, slot_descs, &valid));
+ if (!valid) {
continue; // process next line
}
*is_empty_row = false;
diff --git a/be/src/exec/json_scanner.h b/be/src/exec/json_scanner.h
index beefe2c..1a489a5 100644
--- a/be/src/exec/json_scanner.h
+++ b/be/src/exec/json_scanner.h
@@ -92,7 +92,6 @@ private:
JsonReader* _cur_json_reader;
int _next_range;
bool _cur_reader_eof;
- bool _scanner_eof;
bool _read_json_by_line;
// When we fetch range doesn't start from 0,
@@ -124,6 +123,7 @@ class JsonReader {
public:
JsonReader(RuntimeState* state, ScannerCounter* counter, RuntimeProfile* profile,
bool strip_outer_array, bool num_as_string,bool fuzzy_parse,
+ bool* scanner_eof,
FileReader* file_reader = nullptr, LineReader* line_reader = nullptr);
~JsonReader();
@@ -148,13 +148,13 @@ private:
void _fill_slot(Tuple* tuple, SlotDescriptor* slot_desc, MemPool* mem_pool,
const uint8_t* value, int32_t len);
Status _parse_json_doc(size_t* size, bool* eof);
- void _set_tuple_value(rapidjson::Value& objectValue, Tuple* tuple,
+ Status _set_tuple_value(rapidjson::Value& objectValue, Tuple* tuple,
const std::vector<SlotDescriptor*>& slot_descs, MemPool* tuple_pool,
bool* valid);
- void _write_data_to_tuple(rapidjson::Value::ConstValueIterator value, SlotDescriptor* desc,
+ Status _write_data_to_tuple(rapidjson::Value::ConstValueIterator value, SlotDescriptor* desc,
Tuple* tuple, MemPool* tuple_pool, bool* valid);
- bool _write_values_by_jsonpath(rapidjson::Value& objectValue, MemPool* tuple_pool, Tuple* tuple,
- const std::vector<SlotDescriptor*>& slot_descs);
+ Status _write_values_by_jsonpath(rapidjson::Value& objectValue, MemPool* tuple_pool, Tuple* tuple,
+ const std::vector<SlotDescriptor*>& slot_descs, bool* valid);
std::string _print_json_value(const rapidjson::Value& value);
std::string _print_jsonpath(const std::vector<JsonPath>& path);
@@ -184,6 +184,9 @@ private:
rapidjson::Document _origin_json_doc; // origin json document object from parsed json string
rapidjson::Value* _json_doc; // _json_doc equals _final_json_doc iff not set `json_root`
std::unordered_map<std::string, int> _name_map;
+
+ // point to the _scanner_eof of JsonScanner
+ bool* _scanner_eof;
};
} // namespace doris
diff --git a/be/src/exec/olap_rewrite_node.cpp b/be/src/exec/olap_rewrite_node.cpp
deleted file mode 100644
index 449f53a..0000000
--- a/be/src/exec/olap_rewrite_node.cpp
+++ /dev/null
@@ -1,245 +0,0 @@
-// Licensed to the Apache Software Foundation (ASF) under one
-// or more contributor license agreements. See the NOTICE file
-// distributed with this work for additional information
-// regarding copyright ownership. The ASF licenses this file
-// to you under the Apache License, Version 2.0 (the
-// "License"); you may not use this file except in compliance
-// with the License. You may obtain a copy of the License at
-//
-// http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing,
-// software distributed under the License is distributed on an
-// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
-// KIND, either express or implied. See the License for the
-// specific language governing permissions and limitations
-// under the License.
-
-#include "exec/olap_rewrite_node.h"
-
-#include <sstream>
-
-#include "exprs/expr.h"
-#include "exprs/expr_context.h"
-#include "runtime/descriptors.h"
-#include "runtime/raw_value.h"
-#include "runtime/row_batch.h"
-#include "runtime/runtime_state.h"
-#include "runtime/tuple.h"
-
-namespace doris {
-
-OlapRewriteNode::OlapRewriteNode(ObjectPool* pool, const TPlanNode& tnode,
- const DescriptorTbl& descs)
- : ExecNode(pool, tnode, descs),
- _child_row_batch(nullptr),
- _child_row_idx(0),
- _child_eos(false) {}
-
-Status OlapRewriteNode::init(const TPlanNode& tnode, RuntimeState* state) {
- RETURN_IF_ERROR(ExecNode::init(tnode, state));
- DCHECK(tnode.__isset.olap_rewrite_node);
- // create columns
- RETURN_IF_ERROR(Expr::create_expr_trees(_pool, tnode.olap_rewrite_node.columns, &_columns));
- _column_types = tnode.olap_rewrite_node.column_types;
- _output_tuple_id = tnode.olap_rewrite_node.output_tuple_id;
- return Status::OK();
-}
-
-Status OlapRewriteNode::prepare(RuntimeState* state) {
- RETURN_IF_ERROR(ExecNode::prepare(state));
- RETURN_IF_ERROR(Expr::prepare(_columns, state, child(0)->row_desc(), expr_mem_tracker()));
- _output_tuple_desc = state->desc_tbl().get_tuple_descriptor(_output_tuple_id);
- // _child_row_batch.reset(new RowBatch(child(0)->row_desc(), state->batch_size(), mem_tracker()));
- _child_row_batch.reset(new RowBatch(child(0)->row_desc(), state->batch_size(),
- state->fragment_mem_tracker().get()));
-
- _max_decimalv2_val.resize(_column_types.size());
- for (int i = 0; i < _column_types.size(); ++i) {
- if (_column_types[i].type == TPrimitiveType::DECIMALV2) {
- _max_decimalv2_val[i].to_max_decimal(_column_types[i].precision,
- _column_types[i].scale);
- }
- }
- return Status::OK();
-}
-
-Status OlapRewriteNode::open(RuntimeState* state) {
- RETURN_IF_ERROR(ExecNode::open(state));
- RETURN_IF_ERROR(Expr::open(_columns, state));
- RETURN_IF_ERROR(child(0)->open(state));
- return Status::OK();
-}
-
-Status OlapRewriteNode::get_next(RuntimeState* state, RowBatch* row_batch, bool* eos) {
- RETURN_IF_CANCELLED(state);
- SCOPED_TIMER(_runtime_profile->total_time_counter());
-
- if (reached_limit() || (_child_row_idx == _child_row_batch->num_rows() && _child_eos)) {
- // we're already done or we exhausted the last child batch and there won't be any
- // new ones
- *eos = true;
- return Status::OK();
- }
-
- // start (or continue) consuming row batches from child
- while (true) {
- if (_child_row_idx == _child_row_batch->num_rows()) {
- // fetch next batch
- RETURN_IF_CANCELLED(state);
- row_batch->tuple_data_pool()->acquire_data(_child_row_batch->tuple_data_pool(), false);
- _child_row_batch->reset();
- RETURN_IF_ERROR(child(0)->get_next(state, _child_row_batch.get(), &_child_eos));
- _child_row_idx = 0;
- }
-
- if (copy_rows(state, row_batch)) {
- *eos = reached_limit() ||
- (_child_row_idx == _child_row_batch->num_rows() && _child_eos);
- return Status::OK();
- }
-
- if (_child_eos) {
- // finished w/ last child row batch, and child eos is true
- *eos = true;
- return Status::OK();
- }
- }
-
- return Status::OK();
-}
-
-bool OlapRewriteNode::copy_one_row(TupleRow* src_row, Tuple* tuple, MemPool* pool,
- std::stringstream* ss) {
- memset(tuple, 0, _output_tuple_desc->num_null_bytes());
- // check if valid
- for (int i = 0; i < _columns.size(); ++i) {
- void* src_value = _columns[i]->get_value(src_row);
- SlotDescriptor* slot_desc = _output_tuple_desc->slots()[i];
- // support null for insert into statement
- if (!slot_desc->is_nullable()) {
- if (src_value == nullptr) {
- //column in target table satisfy not null constraint
- (*ss) << "column(" << slot_desc->col_name() << ")'s value is null";
- return false;
- }
- } else {
- if (src_value == nullptr) {
- tuple->set_null(slot_desc->null_indicator_offset());
- continue;
- } else {
- tuple->set_not_null(slot_desc->null_indicator_offset());
- }
- }
-
- const TColumnType& column_type = _column_types[i];
- switch (column_type.type) {
- case TPrimitiveType::CHAR:
- case TPrimitiveType::VARCHAR:
- case TPrimitiveType::STRING: {
- // Fixed length string
- StringValue* str_val = (StringValue*)src_value;
- if (str_val->len > column_type.len) {
- (*ss) << "the length of input is too long than schema. "
- << "column_name: " << slot_desc->col_name() << "; "
- << "input_str: [" << std::string(str_val->ptr, str_val->len) << "] "
- << "schema length: " << column_type.len << "; "
- << "actual length: " << str_val->len << "; ";
- return false;
- } StringValue* dst_val = (StringValue*)tuple->get_slot(slot_desc->tuple_offset());
- if (column_type.type == TPrimitiveType::CHAR) {
- dst_val->ptr = (char*)pool->allocate(column_type.len);
- memcpy(dst_val->ptr, str_val->ptr, str_val->len);
- memset(dst_val->ptr + str_val->len, 0, column_type.len - str_val->len);
- dst_val->len = column_type.len;
- } else {
- dst_val->ptr = (char*)pool->allocate(column_type.len);
- memcpy(dst_val->ptr, str_val->ptr, str_val->len);
- dst_val->len = str_val->len;
- } break;
- } case TPrimitiveType::DECIMALV2: {
- DecimalV2Value* dec_val = (DecimalV2Value*)src_value;
- DecimalV2Value* dst_val = (DecimalV2Value*)tuple->get_slot(slot_desc->tuple_offset());
- if (dec_val->greater_than_scale(column_type.scale)) {
- int code = dec_val->round(dst_val, column_type.scale, HALF_UP);
- if (code != E_DEC_OK) {
- (*ss) << "round one decimal failed.value=" << dec_val->to_string();
- return false;
- }
- } else {
- *reinterpret_cast<PackedInt128*>(dst_val) =
- *reinterpret_cast<const PackedInt128*>(dec_val);
- }
- if (*dst_val > _max_decimalv2_val[i]) {
- dst_val->to_max_decimal(column_type.precision, column_type.scale);
- }
- break;
- }
- default: {
- void* dst_val = (void*)tuple->get_slot(slot_desc->tuple_offset());
- RawValue::write(src_value, dst_val, slot_desc->type(), pool);
- break;
- }
- }
- }
- return true;
-}
-
-bool OlapRewriteNode::copy_rows(RuntimeState* state, RowBatch* output_batch) {
- Tuple* tuple = nullptr;
- MemPool* pool = output_batch->tuple_data_pool();
- int64_t num_rows_invalid = 0;
- for (; _child_row_idx < _child_row_batch->num_rows(); ++_child_row_idx) {
- // Add a new row to output_batch
- int dst_row_idx = output_batch->add_row();
-
- if (dst_row_idx == RowBatch::INVALID_ROW_INDEX) {
- return true;
- }
-
- if (tuple == nullptr) {
- tuple = Tuple::create(_output_tuple_desc->byte_size(), pool);
- }
- TupleRow* src_row = _child_row_batch->get_row(_child_row_idx);
-
- std::stringstream ss;
- if (copy_one_row(src_row, tuple, pool, &ss)) {
- TupleRow* dst_row = output_batch->get_row(dst_row_idx);
- dst_row->set_tuple(0, tuple);
- tuple = nullptr;
- output_batch->commit_last_row();
- ++_num_rows_returned;
- COUNTER_SET(_rows_returned_counter, _num_rows_returned);
-
- if (reached_limit()) {
- return true;
- }
- } else {
- num_rows_invalid++;
- state->append_error_msg_to_file("", ss.str());
- }
- }
- if (num_rows_invalid > 0) {
- state->update_num_rows_load_filtered(num_rows_invalid);
- }
-
- if (VLOG_ROW_IS_ON) {
- for (int i = 0; i < output_batch->num_rows(); ++i) {
- TupleRow* row = output_batch->get_row(i);
- VLOG_ROW << "OlapRewriteNode input row: " << row->to_string(row_desc());
- }
- }
-
- return output_batch->is_full() || output_batch->at_resource_limit();
-}
-
-Status OlapRewriteNode::close(RuntimeState* state) {
- if (is_closed()) {
- return Status::OK();
- }
- _child_row_batch.reset();
- // RETURN_IF_ERROR(child(0)->close(state));
- Expr::close(_columns, state);
- return ExecNode::close(state);
-}
-} // namespace doris
diff --git a/be/src/exec/olap_rewrite_node.h b/be/src/exec/olap_rewrite_node.h
deleted file mode 100644
index 3f9ac74..0000000
--- a/be/src/exec/olap_rewrite_node.h
+++ /dev/null
@@ -1,70 +0,0 @@
-// 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.
-
-#ifndef DORIS_BE_SRC_QUERY_EXEC_OLAP_REWRITE_NODE_H
-#define DORIS_BE_SRC_QUERY_EXEC_OLAP_REWRITE_NODE_H
-
-#include <memory>
-
-#include "exec/exec_node.h"
-#include "runtime/mem_pool.h"
-
-namespace doris {
-
-class Tuple;
-class TupleRow;
-
-// OlapRewriteNode used to filter row that not suite for Olap Table
-class OlapRewriteNode : public ExecNode {
-public:
- OlapRewriteNode(ObjectPool* pool, const TPlanNode& tnode, const DescriptorTbl& descs);
-
- virtual Status init(const TPlanNode& tnode, RuntimeState* state = nullptr);
- virtual ~OlapRewriteNode() {}
- virtual Status prepare(RuntimeState* state);
- virtual Status open(RuntimeState* state);
- virtual Status get_next(RuntimeState* state, RowBatch* row_batch, bool* eos);
- virtual Status close(RuntimeState* state);
-
-private:
- // Copy rows from _child_row_batch for which _conjuncts evaluate to true to
- // output_batch, up to _limit.
- // Return true if limit was hit or output_batch should be returned, otherwise false.
- bool copy_rows(RuntimeState* state, RowBatch* output_batch);
- bool copy_one_row(TupleRow* src_row, Tuple* tuple, MemPool* pool, std::stringstream* ss);
-
- // current row batch of child
- std::unique_ptr<RowBatch> _child_row_batch;
-
- // index of current row in _child_row_batch
- int _child_row_idx;
-
- // true if last get_next() call on child signalled eos
- bool _child_eos;
-
- std::vector<ExprContext*> _columns;
- std::vector<TColumnType> _column_types;
-
- TupleId _output_tuple_id;
- TupleDescriptor* _output_tuple_desc;
-
- std::vector<DecimalV2Value> _max_decimalv2_val;
-};
-
-} // namespace doris
-
-#endif
diff --git a/be/src/exec/orc_scanner.cpp b/be/src/exec/orc_scanner.cpp
index 0a666b8..25031c3 100644
--- a/be/src/exec/orc_scanner.cpp
+++ b/be/src/exec/orc_scanner.cpp
@@ -128,7 +128,6 @@ ORCScanner::ORCScanner(RuntimeState* state, RuntimeProfile* profile,
// _splittable(params.splittable),
_next_range(0),
_cur_file_eof(true),
- _scanner_eof(false),
_total_groups(0),
_current_group(0),
_rows_of_group(0),
@@ -358,12 +357,9 @@ Status ORCScanner::get_next(Tuple* tuple, MemPool* tuple_pool, bool* eof, bool*
}
COUNTER_UPDATE(_rows_read_counter, 1);
SCOPED_TIMER(_materialize_timer);
- if (fill_dest_tuple(tuple, tuple_pool)) {
- *fill_tuple = true;
- } else {
- *fill_tuple = false;
- }
- break; //break always
+ RETURN_IF_ERROR(fill_dest_tuple(tuple, tuple_pool));
+ *fill_tuple = _success;
+ break;
}
return Status::OK();
} catch (orc::ParseError& e) {
diff --git a/be/src/exec/orc_scanner.h b/be/src/exec/orc_scanner.h
index f860041..86b73b6 100644
--- a/be/src/exec/orc_scanner.h
+++ b/be/src/exec/orc_scanner.h
@@ -55,7 +55,6 @@ private:
// Reader
int _next_range;
bool _cur_file_eof;
- bool _scanner_eof;
// orc file reader object
orc::ReaderOptions _options;
diff --git a/be/src/exec/parquet_scanner.cpp b/be/src/exec/parquet_scanner.cpp
index e6f0f2c..267397f 100644
--- a/be/src/exec/parquet_scanner.cpp
+++ b/be/src/exec/parquet_scanner.cpp
@@ -58,8 +58,7 @@ ParquetScanner::ParquetScanner(RuntimeState* state, RuntimeProfile* profile,
// _splittable(params.splittable),
_cur_file_reader(nullptr),
_next_range(0),
- _cur_file_eof(false),
- _scanner_eof(false) {}
+ _cur_file_eof(false) {}
ParquetScanner::~ParquetScanner() {
close();
@@ -92,11 +91,8 @@ Status ParquetScanner::get_next(Tuple* tuple, MemPool* tuple_pool, bool* eof, bo
COUNTER_UPDATE(_rows_read_counter, 1);
SCOPED_TIMER(_materialize_timer);
- if (fill_dest_tuple(tuple, tuple_pool)) {
- *fill_tuple = true;
- } else {
- *fill_tuple = false;
- }
+ RETURN_IF_ERROR(fill_dest_tuple(tuple, tuple_pool));
+ *fill_tuple = _success;
break; // break always
}
if (_scanner_eof) {
diff --git a/be/src/exec/parquet_scanner.h b/be/src/exec/parquet_scanner.h
index 63c2b90..de53fb1 100644
--- a/be/src/exec/parquet_scanner.h
+++ b/be/src/exec/parquet_scanner.h
@@ -79,7 +79,6 @@ private:
ParquetReaderWrap* _cur_file_reader;
int _next_range;
bool _cur_file_eof; // is read over?
- bool _scanner_eof;
// used to hold current StreamLoadPipe
std::shared_ptr<StreamLoadPipe> _stream_load_pipe;
diff --git a/be/src/exec/tablet_sink.cpp b/be/src/exec/tablet_sink.cpp
index 7acde28..41f3570 100644
--- a/be/src/exec/tablet_sink.cpp
+++ b/be/src/exec/tablet_sink.cpp
@@ -17,11 +17,14 @@
#include "exec/tablet_sink.h"
+#include <fmt/format.h>
#include <sstream>
+#include <string>
#include "exprs/expr.h"
#include "exprs/expr_context.h"
#include "olap/hll.h"
+#include "olap/olap_define.h"
#include "runtime/exec_env.h"
#include "runtime/row_batch.h"
#include "runtime/runtime_state.h"
@@ -797,34 +800,44 @@ Status OlapTableSink::send(RuntimeState* state, RowBatch* input_batch) {
if (!_output_expr_ctxs.empty()) {
SCOPED_RAW_TIMER(&_convert_batch_ns);
_output_batch->reset();
- _convert_batch(state, input_batch, _output_batch.get());
+ RETURN_IF_ERROR(_convert_batch(state, input_batch, _output_batch.get()));
batch = _output_batch.get();
}
- int num_invalid_rows = 0;
- {
+
+ int filtered_rows = 0;
+ if (_need_validate_data) {
SCOPED_RAW_TIMER(&_validate_data_ns);
_filter_bitmap.Reset(batch->num_rows());
- num_invalid_rows = _validate_data(state, batch, &_filter_bitmap);
- _number_filtered_rows += num_invalid_rows;
+ bool stop_processing = false;
+ RETURN_IF_ERROR(_validate_data(state, batch, &_filter_bitmap, &filtered_rows, &stop_processing));
+ _number_filtered_rows += filtered_rows;
+ if (stop_processing) {
+ // should be returned after updating "_number_filtered_rows", to make sure that load job can be cancelled
+ // because of "data unqualified"
+ return Status::EndOfFile("Encountered unqualified data, stop processing");
+ }
}
+
SCOPED_RAW_TIMER(&_send_data_ns);
+ bool stop_processing = false;
for (int i = 0; i < batch->num_rows(); ++i) {
Tuple* tuple = batch->get_row(i)->get_tuple(0);
- if (num_invalid_rows > 0 && _filter_bitmap.Get(i)) {
+ if (filtered_rows > 0 && _filter_bitmap.Get(i)) {
continue;
}
const OlapTablePartition* partition = nullptr;
uint32_t dist_hash = 0;
if (!_partition->find_tablet(tuple, &partition, &dist_hash)) {
- std::stringstream ss;
- ss << "no partition for this tuple. tuple="
- << Tuple::to_string(tuple, *_output_tuple_desc);
-#if BE_TEST
- LOG(INFO) << ss.str();
-#else
- state->append_error_msg_to_file("", ss.str());
-#endif
+ RETURN_IF_ERROR(state->append_error_msg_to_file([]() -> std::string { return ""; },
+ [&]() -> std::string {
+ fmt::memory_buffer buf;
+ fmt::format_to(buf, "no partition for this tuple. tuple={}", Tuple::to_string(tuple, *_output_tuple_desc));
+ return buf.data();
+ }, &stop_processing));
_number_filtered_rows++;
+ if (stop_processing) {
+ return Status::EndOfFile("Encountered unqualified data, stop processing");
+ }
continue;
}
_partition_ids.emplace(partition->id);
@@ -946,10 +959,11 @@ Status OlapTableSink::close(RuntimeState* state, Status close_status) {
return status;
}
-void OlapTableSink::_convert_batch(RuntimeState* state, RowBatch* input_batch,
- RowBatch* output_batch) {
+Status OlapTableSink::_convert_batch(RuntimeState* state, RowBatch* input_batch,
+ RowBatch* output_batch) {
DCHECK_GE(output_batch->capacity(), input_batch->num_rows());
int commit_rows = 0;
+ bool stop_processing = false;
for (int i = 0; i < input_batch->num_rows(); ++i) {
auto src_row = input_batch->get_row(i);
Tuple* dst_tuple =
@@ -964,7 +978,8 @@ void OlapTableSink::_convert_batch(RuntimeState* state, RowBatch* input_batch,
// Only when the expr return value is null, we will check the error message.
std::string expr_error = _output_expr_ctxs[j]->get_error_msg();
if (!expr_error.empty()) {
- state->append_error_msg_to_file(slot_desc->col_name(), expr_error);
+ RETURN_IF_ERROR(state->append_error_msg_to_file([&]() -> std::string { return slot_desc->col_name(); },
+ [&]() -> std::string { return expr_error; }, &stop_processing));
_number_filtered_rows++;
ignore_this_row = true;
// The ctx is reused, so must clear the error state and message.
@@ -972,13 +987,12 @@ void OlapTableSink::_convert_batch(RuntimeState* state, RowBatch* input_batch,
break;
}
if (!slot_desc->is_nullable()) {
- std::stringstream ss;
- ss << "null value for not null column, column=" << slot_desc->col_name();
-#if BE_TEST
- LOG(INFO) << ss.str();
-#else
- state->append_error_msg_to_file("", ss.str());
-#endif
+ RETURN_IF_ERROR(state->append_error_msg_to_file([]() -> std::string { return ""; },
+ [&]() -> std::string {
+ fmt::memory_buffer buf;
+ fmt::format_to(buf, "null value for not null column, column={}", slot_desc->col_name());
+ return buf.data();
+ }, &stop_processing));
_number_filtered_rows++;
ignore_this_row = true;
break;
@@ -991,28 +1005,33 @@ void OlapTableSink::_convert_batch(RuntimeState* state, RowBatch* input_batch,
}
void* slot = dst_tuple->get_slot(slot_desc->tuple_offset());
RawValue::write(src_val, slot, slot_desc->type(), _output_batch->tuple_data_pool());
- }
+ } // end for output expr
if (!ignore_this_row) {
output_batch->get_row(commit_rows)->set_tuple(0, dst_tuple);
commit_rows++;
}
+
+ if (stop_processing) {
+ return Status::EndOfFile("Encountered unqualified data, stop processing");
+ }
}
output_batch->commit_rows(commit_rows);
+ return Status::OK();
}
-int OlapTableSink::_validate_data(RuntimeState* state, RowBatch* batch, Bitmap* filter_bitmap) {
- int filtered_rows = 0;
+Status OlapTableSink::_validate_data(RuntimeState* state, RowBatch* batch, Bitmap* filter_bitmap, int* filtered_rows,
+ bool* stop_processing) {
for (int row_no = 0; row_no < batch->num_rows(); ++row_no) {
Tuple* tuple = batch->get_row(row_no)->get_tuple(0);
bool row_valid = true;
- std::stringstream ss; // error message
+ fmt::memory_buffer error_msg; // error message
for (int i = 0; row_valid && i < _output_tuple_desc->slots().size(); ++i) {
SlotDescriptor* desc = _output_tuple_desc->slots()[i];
if (desc->is_nullable() && tuple->is_null(desc->null_indicator_offset())) {
if (desc->type().type == TYPE_OBJECT) {
- ss << "null is not allowed for bitmap column, column_name: "
- << desc->col_name();
+ fmt::format_to(error_msg, "null is not allowed for bitmap column, column_name: {}; ",
+ desc->col_name());
row_valid = false;
}
continue;
@@ -1024,11 +1043,11 @@ int OlapTableSink::_validate_data(RuntimeState* state, RowBatch* batch, Bitmap*
// Fixed length string
StringValue* str_val = (StringValue*)slot;
if (str_val->len > desc->type().len) {
- ss << "the length of input is too long than schema. "
- << "column_name: " << desc->col_name() << "; "
- << "input_str: [" << std::string(str_val->ptr, str_val->len) << "] "
- << "schema length: " << desc->type().len << "; "
- << "actual length: " << str_val->len << "; ";
+ fmt::format_to(error_msg, "{}", "the length of input is too long than schema. ");
+ fmt::format_to(error_msg, "column_name: {}; ", desc->col_name());
+ fmt::format_to(error_msg, "input str: [{}] ", std::string(str_val->ptr, str_val->len));
+ fmt::format_to(error_msg, "schema length: {}; ", desc->type().len);
+ fmt::format_to(error_msg, "actual length: {}; ", str_val->len);
row_valid = false;
continue;
}
@@ -1045,13 +1064,12 @@ int OlapTableSink::_validate_data(RuntimeState* state, RowBatch* batch, Bitmap*
}
case TYPE_STRING: {
StringValue* str_val = (StringValue*)slot;
- if (str_val->len > desc->type().MAX_STRING_LENGTH) {
- ss << "the length of input is too long than schema. "
- << "column_name: " << desc->col_name() << "; "
- << "first 128 bytes of input_str: [" << std::string(str_val->ptr, 128)
- << "] "
- << "schema length: " << desc->type().MAX_STRING_LENGTH << "; "
- << "actual length: " << str_val->len << "; ";
+ if (str_val->len > OLAP_STRING_MAX_LENGTH) {
+ fmt::format_to(error_msg, "{}", "the length of input is too long than schema. ");
+ fmt::format_to(error_msg, "column_name: {}; ", desc->col_name());
+ fmt::format_to(error_msg, "first 128 bytes of input str: [{}] ", std::string(str_val->ptr, 128));
+ fmt::format_to(error_msg, "schema length: {}; ", OLAP_STRING_MAX_LENGTH);
+ fmt::format_to(error_msg, "actual length: {}; ", str_val->len);
row_valid = false;
continue;
}
@@ -1063,16 +1081,15 @@ int OlapTableSink::_validate_data(RuntimeState* state, RowBatch* batch, Bitmap*
int code = dec_val.round(&dec_val, desc->type().scale, HALF_UP);
reinterpret_cast<PackedInt128*>(slot)->value = dec_val.value();
if (code != E_DEC_OK) {
- ss << "round one decimal failed.value=" << dec_val.to_string();
+ fmt::format_to(error_msg, "round one decimal failed.value={}; ", dec_val.to_string());
row_valid = false;
continue;
}
}
if (dec_val > _max_decimalv2_val[i] || dec_val < _min_decimalv2_val[i]) {
- ss << "decimal value is not valid for definition, column=" << desc->col_name()
- << ", value=" << dec_val.to_string()
- << ", precision=" << desc->type().precision
- << ", scale=" << desc->type().scale;
+ fmt::format_to(error_msg, "decimal value is not valid for definition, column={}", desc->col_name());
+ fmt::format_to(error_msg, ", value={}", dec_val.to_string());
+ fmt::format_to(error_msg, ", precision={}, scale={}; ", desc->type().precision, desc->type().scale);
row_valid = false;
continue;
}
@@ -1081,8 +1098,7 @@ int OlapTableSink::_validate_data(RuntimeState* state, RowBatch* batch, Bitmap*
case TYPE_HLL: {
Slice* hll_val = (Slice*)slot;
if (!HyperLogLog::is_valid(*hll_val)) {
- ss << "Content of HLL type column is invalid"
- << "column_name: " << desc->col_name() << "; ";
+ fmt::format_to(error_msg, "Content of HLL type column is invalid. column name: {}; ", desc->col_name());
row_valid = false;
continue;
}
@@ -1094,16 +1110,13 @@ int OlapTableSink::_validate_data(RuntimeState* state, RowBatch* batch, Bitmap*
}
if (!row_valid) {
- filtered_rows++;
+ (*filtered_rows)++;
filter_bitmap->Set(row_no, true);
-#if BE_TEST
- LOG(INFO) << ss.str();
-#else
- state->append_error_msg_to_file("", ss.str());
-#endif
+ RETURN_IF_ERROR(state->append_error_msg_to_file([]() -> std::string { return ""; },
+ [&]() -> std::string { return error_msg.data(); }, stop_processing));
}
}
- return filtered_rows;
+ return Status::OK();
}
void OlapTableSink::_send_batch_process() {
diff --git a/be/src/exec/tablet_sink.h b/be/src/exec/tablet_sink.h
index 6f403b1..df40493 100644
--- a/be/src/exec/tablet_sink.h
+++ b/be/src/exec/tablet_sink.h
@@ -342,12 +342,14 @@ public:
private:
// convert input batch to output batch which will be loaded into OLAP table.
// this is only used in insert statement.
- void _convert_batch(RuntimeState* state, RowBatch* input_batch, RowBatch* output_batch);
+ Status _convert_batch(RuntimeState* state, RowBatch* input_batch, RowBatch* output_batch);
// make input data valid for OLAP table
// return number of invalid/filtered rows.
// invalid row number is set in Bitmap
- int _validate_data(RuntimeState* state, RowBatch* batch, Bitmap* filter_bitmap);
+ // set stop_processing is we want to stop the whole process now.
+ Status _validate_data(RuntimeState* state, RowBatch* batch, Bitmap* filter_bitmap, int* filtered_rows,
+ bool* stop_processing);
// the consumer func of sending pending batches in every NodeChannel.
// use polling & NodeChannel::try_send_and_fetch_status() to achieve nonblocking sending.
diff --git a/be/src/http/action/download_action.h b/be/src/http/action/download_action.h
index a348073..ae988ff 100644
--- a/be/src/http/action/download_action.h
+++ b/be/src/http/action/download_action.h
@@ -18,7 +18,6 @@
#ifndef DORIS_BE_SRC_HTTP_DOWNLOAD_ACTION_H
#define DORIS_BE_SRC_HTTP_DOWNLOAD_ACTION_H
-#include "exec/csv_scanner.h"
#include "exec/scan_node.h"
#include "http/http_handler.h"
#include "runtime/descriptors.h"
diff --git a/be/src/http/action/stream_load.cpp b/be/src/http/action/stream_load.cpp
index 369ce58..19bcff2 100644
--- a/be/src/http/action/stream_load.cpp
+++ b/be/src/http/action/stream_load.cpp
@@ -522,13 +522,15 @@ Status StreamLoadAction::_process_put(HttpRequest* http_req, StreamLoadContext*
if (!http_req->header(HTTP_DELETE_CONDITION).empty()) {
request.__set_delete_condition(http_req->header(HTTP_DELETE_CONDITION));
}
- // plan this load
- TNetworkAddress master_addr = _exec_env->master_info()->network_address;
-#ifndef BE_TEST
+
if (!http_req->header(HTTP_MAX_FILTER_RATIO).empty()) {
ctx->max_filter_ratio = strtod(http_req->header(HTTP_MAX_FILTER_RATIO).c_str(), nullptr);
+ request.__set_max_filter_ratio(ctx->max_filter_ratio);
}
+#ifndef BE_TEST
+ // plan this load
+ TNetworkAddress master_addr = _exec_env->master_info()->network_address;
int64_t stream_load_put_start_time = MonotonicNanos();
RETURN_IF_ERROR(ThriftRpcHelper::rpc<FrontendServiceClient>(
master_addr.hostname, master_addr.port,
diff --git a/be/src/runtime/CMakeLists.txt b/be/src/runtime/CMakeLists.txt
index d426559..686207d 100644
--- a/be/src/runtime/CMakeLists.txt
+++ b/be/src/runtime/CMakeLists.txt
@@ -59,7 +59,6 @@ set(RUNTIME_FILES
qsorter.cpp
fragment_mgr.cpp
dpp_sink_internal.cpp
- data_spliter.cpp
dpp_sink.cpp
etl_job_mgr.cpp
load_path_mgr.cpp
diff --git a/be/src/runtime/data_spliter.cpp b/be/src/runtime/data_spliter.cpp
deleted file mode 100644
index 4194af1..0000000
--- a/be/src/runtime/data_spliter.cpp
+++ /dev/null
@@ -1,325 +0,0 @@
-// Licensed to the Apache Software Foundation (ASF) under one
-// or more contributor license agreements. See the NOTICE file
-// distributed with this work for additional information
-// regarding copyright ownership. The ASF licenses this file
-// to you under the Apache License, Version 2.0 (the
-// "License"); you may not use this file except in compliance
-// with the License. You may obtain a copy of the License at
-//
-// http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing,
-// software distributed under the License is distributed on an
-// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
-// KIND, either express or implied. See the License for the
-// specific language governing permissions and limitations
-// under the License.
-
-#include "runtime/data_spliter.h"
-
-#include <thrift/protocol/TDebugProtocol.h>
-
-#include <sstream>
-
-#include "common/object_pool.h"
-#include "exprs/expr.h"
-#include "exprs/expr_context.h"
-#include "gen_cpp/DataSinks_types.h"
-#include "runtime/dpp_sink.h"
-#include "runtime/load_path_mgr.h"
-#include "runtime/mem_tracker.h"
-#include "runtime/raw_value.h"
-#include "runtime/row_batch.h"
-#include "runtime/runtime_state.h"
-#include "runtime/tuple_row.h"
-#include "service/backend_options.h"
-#include "util/file_utils.h"
-#include "util/runtime_profile.h"
-#include "util/uid_util.h"
-
-namespace doris {
-
-DataSpliter::DataSpliter(const RowDescriptor& row_desc)
- : _obj_pool(new ObjectPool()), _row_desc(row_desc) {
- _name = "DataSpliter";
-}
-
-DataSpliter::~DataSpliter() {}
-
-// We use the PartitionRange to compare here. It should not be a member function of PartitionInfo
-// class because there are some other member in it.
-static bool compare_part_use_range(const PartitionInfo* v1, const PartitionInfo* v2) {
- return v1->range() < v2->range();
-}
-
-Status DataSpliter::from_thrift(ObjectPool* pool, const TDataSplitSink& t_sink,
- DataSpliter* spliter) {
- VLOG_ROW << "TDataSplitSink: " << apache::thrift::ThriftDebugString(t_sink);
-
- // Partition Exprs
- RETURN_IF_ERROR(
- Expr::create_expr_trees(pool, t_sink.partition_exprs, &spliter->_partition_expr_ctxs));
- // Partition infos
- int num_parts = t_sink.partition_infos.size();
- if (num_parts == 0) {
- return Status::InternalError("Empty partition info.");
- }
- for (int i = 0; i < num_parts; ++i) {
- PartitionInfo* info = pool->add(new PartitionInfo());
- RETURN_IF_ERROR(PartitionInfo::from_thrift(pool, t_sink.partition_infos[i], info));
- spliter->_partition_infos.push_back(info);
- }
-
- // partitions should be in ascending order
- std::sort(spliter->_partition_infos.begin(), spliter->_partition_infos.end(),
- compare_part_use_range);
-
- // schema infos
- for (auto& iter : t_sink.rollup_schemas) {
- RollupSchema* schema = pool->add(new RollupSchema());
- RETURN_IF_ERROR(RollupSchema::from_thrift(pool, iter.second, schema));
- spliter->_rollup_map[iter.first] = schema;
- }
-
- return Status::OK();
-}
-
-Status DataSpliter::prepare(RuntimeState* state) {
- std::stringstream title;
- title << "DataSplitSink (dst_fragment_instance_id=" << print_id(state->fragment_instance_id())
- << ")";
- RETURN_IF_ERROR(DataSink::prepare(state));
- RETURN_IF_ERROR(Expr::prepare(_partition_expr_ctxs, state, _row_desc, _expr_mem_tracker));
- for (auto& iter : _rollup_map) {
- RETURN_IF_ERROR(iter.second->prepare(state, _row_desc, _expr_mem_tracker));
- }
- _profile = state->obj_pool()->add(new RuntimeProfile(title.str()));
- for (auto iter : _partition_infos) {
- RETURN_IF_ERROR(iter->prepare(state, _row_desc, _expr_mem_tracker));
- }
- return Status::OK();
-}
-
-Status DataSpliter::open(RuntimeState* state) {
- RETURN_IF_ERROR(Expr::open(_partition_expr_ctxs, state));
-
- for (auto& iter : _rollup_map) {
- RETURN_IF_ERROR(iter.second->open(state));
- }
-
- RETURN_IF_ERROR(state->create_load_dir());
-
- for (auto iter : _partition_infos) {
- RETURN_IF_ERROR(iter->open(state));
-
- DppSink* dpp_sink = _obj_pool->add(new DppSink(_row_desc, _rollup_map));
- _dpp_sink_vec.push_back(dpp_sink);
-
- RETURN_IF_ERROR(dpp_sink->init(state));
- _profile->add_child(dpp_sink->profile(), true, nullptr);
- }
-
- _split_timer = ADD_TIMER(_profile, "process batch");
- _finish_timer = ADD_TIMER(_profile, "sort time");
-
- return Status::OK();
-}
-
-int DataSpliter::binary_find_partition(const PartRangeKey& key) const {
- int low = 0;
- int high = _partition_infos.size() - 1;
-
- VLOG_ROW << "range key: " << key.debug_string() << std::endl;
- while (low <= high) {
- int mid = low + (high - low) / 2;
- int cmp = _partition_infos[mid]->range().compare_key(key);
- if (cmp == 0) {
- return mid;
- } else if (cmp < 0) { // current < partition[mid]
- low = mid + 1;
- } else {
- high = mid - 1;
- }
- }
-
- return -1;
-}
-
-Status DataSpliter::process_partition(RuntimeState* state, TupleRow* row, PartitionInfo** info,
- int32_t* part_index) {
- if (_partition_expr_ctxs.size() == 0) {
- *part_index = 0;
- *info = _partition_infos[0];
- return Status::OK();
- } else {
- // use binary search to get the right partition.
- ExprContext* ctx = _partition_expr_ctxs[0];
- void* partition_val = ctx->get_value(row);
- // construct a PartRangeKey
- PartRangeKey tmpPartKey;
- if (nullptr != partition_val) {
- RETURN_IF_ERROR(
- PartRangeKey::from_value(ctx->root()->type().type, partition_val, &tmpPartKey));
- } else {
- tmpPartKey = PartRangeKey::neg_infinite();
- }
-
- *part_index = binary_find_partition(tmpPartKey);
- if (*part_index < 0) {
- std::stringstream error_log;
- error_log << "there is no corresponding partition for this key: ";
- ctx->print_value(row, &error_log);
- state->update_num_rows_load_filtered(1);
- return Status::InternalError(error_log.str());
- }
- *info = _partition_infos[*part_index];
- }
- return Status::OK();
-}
-
-Status DataSpliter::process_distribute(RuntimeState* state, TupleRow* row,
- const PartitionInfo* part, uint32_t* mod) {
- uint32_t hash_val = 0;
-
- for (auto& ctx : part->distributed_expr_ctxs()) {
- void* partition_val = ctx->get_value(row);
- if (partition_val != nullptr) {
- hash_val = RawValue::zlib_crc32(partition_val, ctx->root()->type(), hash_val);
- } else {
- //nullptr is treat as 0 when hash
- static const int INT_VALUE = 0;
- static const TypeDescriptor INT_TYPE(TYPE_INT);
- hash_val = RawValue::zlib_crc32(&INT_VALUE, INT_TYPE, hash_val);
- }
- }
-
- *mod = hash_val % part->distributed_bucket();
-
- return Status::OK();
-}
-
-Status DataSpliter::send_row(RuntimeState* state, const TabletDesc& desc, TupleRow* row,
- DppSink* dpp_sink) {
- RowBatch* batch = nullptr;
- auto batch_iter = _batch_map.find(desc);
- if (batch_iter == _batch_map.end()) {
- batch = _obj_pool->add(
- new RowBatch(_row_desc, state->batch_size(), _expr_mem_tracker.get()));
- _batch_map[desc] = batch;
- } else {
- batch = batch_iter->second;
- }
-
- // Add this row to this batch
- int idx = batch->add_row();
- // Just deep copy this row
- row->deep_copy(batch->get_row(idx), _row_desc.tuple_descriptors(), batch->tuple_data_pool(),
- false);
- batch->commit_last_row();
-
- // If this batch is full send this to dpp_sink
- if (batch->is_full()) {
- RETURN_IF_ERROR(dpp_sink->add_batch(_obj_pool.get(), state, desc, batch));
- batch->reset();
- }
- return Status::OK();
-}
-
-Status DataSpliter::process_one_row(RuntimeState* state, TupleRow* row) {
- TabletDesc desc;
- int32_t part_index = 0;
-
- // process partition
- PartitionInfo* part = nullptr;
- Status status = process_partition(state, row, &part, &part_index);
- // TODO(lingbin): adjust 'process_partition' function's return value. It is a little inelegant
- // to return another OK when pri-status is not OK.
- // If find no partition, this row should be omitted.
- if (!status.ok()) {
- state->set_error_row_number(state->get_error_row_number() + 1);
- state->set_normal_row_number(state->get_normal_row_number() - 1);
-
- state->append_error_msg_to_file(row->to_string(_row_desc), status.get_error_msg());
- return Status::OK();
- }
-
- desc.partition_id = part->id();
-
- // process distribute
- RETURN_IF_ERROR(process_distribute(state, row, part, &desc.bucket_id));
-
- // construct dpp_sink map
- _sink_map[desc] = _dpp_sink_vec[part_index];
-
- // process distribute
- RETURN_IF_ERROR(send_row(state, desc, row, _dpp_sink_vec[part_index]));
-
- return Status::OK();
-}
-
-Status DataSpliter::send(RuntimeState* state, RowBatch* batch) {
- SCOPED_TIMER(_split_timer);
- int num_rows = batch->num_rows();
- for (int i = 0; i < num_rows; ++i) {
- RETURN_IF_ERROR(process_one_row(state, batch->get_row(i)));
- }
- return Status::OK();
-}
-
-Status DataSpliter::close(RuntimeState* state, Status close_status) {
- bool is_ok = true;
- Status err_status;
- if (_closed) {
- return Status::OK();
- }
- if (close_status.ok()) {
- SCOPED_TIMER(_finish_timer);
- // Flush data have not been sent
- for (const auto& iter : _batch_map) {
- if (iter.second->num_rows() > 0) {
- DppSink* dpp_sink = _sink_map[iter.first];
- Status status =
- dpp_sink->add_batch(_obj_pool.get(), state, iter.first, iter.second);
- if (UNLIKELY(is_ok && !status.ok())) {
- LOG(WARNING) << "add_batch error"
- << " err_msg=" << status.get_error_msg();
- is_ok = false;
- err_status = status;
- }
- iter.second->clear();
- }
- }
- } else {
- for (const auto& iter : _batch_map) {
- iter.second->clear();
- }
- }
- // finish sink
- for (const auto& iter : _dpp_sink_vec) {
- Status status = iter->finish(state);
- if (UNLIKELY(is_ok && !status.ok())) {
- LOG(WARNING) << "finish dpp_sink error"
- << " err_msg=" << status.get_error_msg()
- << " backend=" << BackendOptions::get_localhost();
- is_ok = false;
- err_status = status;
- }
- }
- Expr::close(_partition_expr_ctxs, state);
- for (auto& iter : _rollup_map) {
- iter.second->close(state);
- }
- for (auto iter : _partition_infos) {
- iter->close(state);
- }
-
- _expr_mem_tracker.reset();
- _closed = true;
- if (is_ok) {
- return Status::OK();
- } else {
- return err_status;
- }
-}
-
-} // namespace doris
diff --git a/be/src/runtime/data_spliter.h b/be/src/runtime/data_spliter.h
deleted file mode 100644
index 1e26b03..0000000
--- a/be/src/runtime/data_spliter.h
+++ /dev/null
@@ -1,108 +0,0 @@
-// 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.
-
-#ifndef DORIS_BE_RUNTIME_DATA_SPLITER_H
-#define DORIS_BE_RUNTIME_DATA_SPLITER_H
-
-#include <string>
-#include <unordered_map>
-#include <vector>
-
-#include "common/status.h"
-#include "exec/data_sink.h"
-#include "runtime/dpp_sink_internal.h"
-#include "util/runtime_profile.h"
-
-namespace doris {
-
-class RowDescriptor;
-class RuntimeState;
-class RowBatch;
-class ObjectPool;
-class TDataSplitSink;
-class Expr;
-class ExprContext;
-class PartitionInfo;
-class RollupSchema;
-class TupleRow;
-class DppSink;
-class MemTracker;
-
-// DataSpliter used to split data input to groups of data
-// according to partition information, distributed information
-// and rollup information.
-// TODO(zc): think about this to make it more reusable
-class DataSpliter : public DataSink {
-public:
- // Construct from thrift struct which is generated by FE.
- DataSpliter(const RowDescriptor& row_desc);
-
- virtual ~DataSpliter();
-
- virtual Status prepare(RuntimeState* state);
-
- virtual Status open(RuntimeState* state);
-
- virtual Status send(RuntimeState* state, RowBatch* batch);
-
- virtual Status close(RuntimeState* state, Status close_status);
-
- // Returns the runtime profile for the sink.
- virtual RuntimeProfile* profile() { return _profile; }
-
- static Status from_thrift(ObjectPool* pool, const TDataSplitSink& t_sink, DataSpliter* spliter);
-
-private:
- int binary_find_partition(const PartRangeKey& key) const;
- Status process_partition(RuntimeState* state, TupleRow* row, PartitionInfo** info,
- int32_t* part_index);
- Status process_distribute(RuntimeState* state, TupleRow* row, const PartitionInfo* part,
- uint32_t* mod);
- Status send_row(RuntimeState* state, const TabletDesc& desc, TupleRow* row, DppSink* dpp_sink);
- Status process_one_row(RuntimeState* state, TupleRow* row);
-
- std::unique_ptr<ObjectPool> _obj_pool;
- const RowDescriptor& _row_desc;
-
- // Information used to partition data
- // partition exprs
- std::vector<ExprContext*> _partition_expr_ctxs;
-
- // map from range value to partition_id
- // sorted in ascending order by range for binary search
- std::vector<PartitionInfo*> _partition_infos;
-
- // Information of rollup
- // from name to rollup information.
- std::map<std::string, RollupSchema*> _rollup_map;
-
- std::unordered_map<TabletDesc, RowBatch*> _batch_map;
- std::unordered_map<TabletDesc, DppSink*> _sink_map;
-
- std::vector<DppSink*> _dpp_sink_vec;
-
- // Allocated from _pool
- RuntimeProfile* _profile;
-
- // Wall time senders spend waiting for the recv buffer to have capacity.
- RuntimeProfile::Counter* _split_timer;
- RuntimeProfile::Counter* _finish_timer;
-};
-
-} // namespace doris
-
-#endif
diff --git a/be/src/runtime/etl_job_mgr.cpp b/be/src/runtime/etl_job_mgr.cpp
index 95b5539..ce4029b 100644
--- a/be/src/runtime/etl_job_mgr.cpp
+++ b/be/src/runtime/etl_job_mgr.cpp
@@ -26,7 +26,6 @@
#include "gen_cpp/Status_types.h"
#include "gen_cpp/Types_types.h"
#include "runtime/client_cache.h"
-#include "runtime/data_spliter.h"
#include "runtime/exec_env.h"
#include "runtime/fragment_mgr.h"
#include "runtime/plan_fragment_executor.h"
diff --git a/be/src/runtime/plan_fragment_executor.cpp b/be/src/runtime/plan_fragment_executor.cpp
index b878e17..1e5ad0c 100644
--- a/be/src/runtime/plan_fragment_executor.cpp
+++ b/be/src/runtime/plan_fragment_executor.cpp
@@ -386,7 +386,11 @@ Status PlanFragmentExecutor::open_internal() {
if (_collect_query_statistics_with_every_batch) {
_collect_query_statistics();
}
- RETURN_IF_ERROR(_sink->send(runtime_state(), batch));
+ const Status& st = _sink->send(runtime_state(), batch);
+ if (st.is_end_of_file()) {
+ break;
+ }
+ RETURN_IF_ERROR(st);
}
// Close the sink *before* stopping the report thread. Close may
diff --git a/be/src/runtime/runtime_state.cpp b/be/src/runtime/runtime_state.cpp
index 662d4b5..e506189 100644
--- a/be/src/runtime/runtime_state.cpp
+++ b/be/src/runtime/runtime_state.cpp
@@ -18,6 +18,7 @@
#include "runtime/runtime_state.h"
#include <boost/algorithm/string/join.hpp>
+#include <fmt/format.h>
#include <iostream>
#include <sstream>
#include <string>
@@ -180,6 +181,10 @@ Status RuntimeState::init(const TUniqueId& fragment_instance_id, const TQueryOpt
}
TimezoneUtils::find_cctz_time_zone(_timezone, _timezone_obj);
+ if (query_globals.__isset.load_zero_tolerance) {
+ _load_zero_tolerance = query_globals.load_zero_tolerance;
+ }
+
_exec_env = exec_env;
if (_query_options.max_errors <= 0) {
@@ -408,15 +413,16 @@ Status RuntimeState::create_error_log_file() {
LOG(WARNING) << error_msg.str();
return Status::InternalError(error_msg.str());
}
- VLOG_ROW << "create error log file: " << _error_log_file_path;
+ VLOG_FILE << "create error log file: " << _error_log_file_path;
return Status::OK();
}
-void RuntimeState::append_error_msg_to_file(const std::string& line, const std::string& error_msg,
- bool is_summary) {
+Status RuntimeState::append_error_msg_to_file(std::function<std::string()> line, std::function<std::string()> error_msg,
+ bool* stop_processing, bool is_summary) {
+ *stop_processing = false;
if (_query_options.query_type != TQueryType::LOAD) {
- return;
+ return Status::OK();
}
// If file havn't been opened, open it here
if (_error_log_file == nullptr) {
@@ -428,36 +434,37 @@ void RuntimeState::append_error_msg_to_file(const std::string& line, const std::
delete _error_log_file;
_error_log_file = nullptr;
}
- return;
+ return status;
}
}
// if num of printed error row exceeds the limit, and this is not a summary message,
- // return
+ // if _load_zero_tolerance, return Error to stop the load process immediately.
if (_num_print_error_rows.fetch_add(1, std::memory_order_relaxed) > MAX_ERROR_NUM &&
!is_summary) {
- return;
+ if (_load_zero_tolerance) {
+ *stop_processing = true;
+ }
+ return Status::OK();
}
- std::stringstream out;
+ fmt::memory_buffer out;
if (is_summary) {
- out << "Summary: ";
- out << error_msg;
+ fmt::format_to(out, "Summary: {}", error_msg());
} else {
if (_error_row_number < MAX_ERROR_NUM) {
// Note: export reason first in case src line too long and be truncated.
- out << "Reason: " << error_msg;
- out << ". src line: [" << line << "]; ";
+ fmt::format_to(out, "Reason: {}. src line [{}]; ", error_msg(), line());
} else if (_error_row_number == MAX_ERROR_NUM) {
- out << "TOO MUCH ERROR! already reach " << MAX_ERROR_NUM << "."
- << " no more show next error.";
+ fmt::format_to(out, "TOO MUCH ERROR! already reach {}. show no more next error.", MAX_ERROR_NUM);
}
}
- if (!out.str().empty()) {
- (*_error_log_file) << out.str() << std::endl;
- export_load_error(out.str());
+ if (out.size() > 0) {
+ (*_error_log_file) << out.data() << std::endl;
+ export_load_error(out.data());
}
+ return Status::OK();
}
const int64_t HUB_MAX_ERROR_NUM = 10;
diff --git a/be/src/runtime/runtime_state.h b/be/src/runtime/runtime_state.h
index 75d4dfb..8797076 100644
--- a/be/src/runtime/runtime_state.h
+++ b/be/src/runtime/runtime_state.h
@@ -287,9 +287,11 @@ public:
const std::string get_error_log_file_path() const { return _error_log_file_path; }
+ // append error msg and error line to file when loading data.
// is_summary is true, means we are going to write the summary line
- void append_error_msg_to_file(const std::string& line, const std::string& error_msg,
- bool is_summary = false);
+ // If we need to stop the processing, set stop_processing to true
+ Status append_error_msg_to_file(std::function<std::string()> line, std::function<std::string()> error_msg,
+ bool* stop_processing, bool is_summary = false);
int64_t num_bytes_load_total() { return _num_bytes_load_total.load(); }
@@ -528,6 +530,10 @@ private:
AtomicInt32 _initial_reservation_refcnt;
QueryFragmentsCtx* _query_ctx;
+
+ // true if max_filter_ratio is 0
+ bool _load_zero_tolerance = false;
+
// prohibit copies
RuntimeState(const RuntimeState&);
};
diff --git a/be/src/vec/CMakeLists.txt b/be/src/vec/CMakeLists.txt
index f737391..09e86e2 100644
--- a/be/src/vec/CMakeLists.txt
+++ b/be/src/vec/CMakeLists.txt
@@ -147,7 +147,7 @@ set(VEC_FILES
sink/mysql_result_writer.cpp
sink/result_sink.cpp
sink/vdata_stream_sender.cpp
- sink/vtabet_sink.cpp
+ sink/vtablet_sink.cpp
runtime/vdatetime_value.cpp
runtime/vdata_stream_recvr.cpp
runtime/vdata_stream_mgr.cpp
diff --git a/be/src/vec/sink/vdata_stream_sender.cpp b/be/src/vec/sink/vdata_stream_sender.cpp
index d953c9d..8afa66b 100644
--- a/be/src/vec/sink/vdata_stream_sender.cpp
+++ b/be/src/vec/sink/vdata_stream_sender.cpp
@@ -17,7 +17,6 @@
#include "vec/sink/vdata_stream_sender.h"
-#include <boost/bind.hpp>
#include <fmt/format.h>
#include <fmt/ranges.h>
@@ -363,8 +362,8 @@ Status VDataStreamSender::prepare(RuntimeState* state) {
_serialize_batch_timer = ADD_TIMER(profile(), "SerializeBatchTime");
_overall_throughput = profile()->add_derived_counter(
"OverallThroughput", TUnit::BYTES_PER_SECOND,
- boost::bind<int64_t>(&RuntimeProfile::units_per_second, _bytes_sent_counter,
- profile()->total_time_counter()),
+ std::bind<int64_t>(&RuntimeProfile::units_per_second, _bytes_sent_counter,
+ profile()->total_time_counter()),
"");
_local_bytes_send_counter = ADD_COUNTER(profile(), "LocalBytesSent", TUnit::BYTES);
for (int i = 0; i < _channels.size(); ++i) {
diff --git a/be/src/vec/sink/vtabet_sink.cpp b/be/src/vec/sink/vtablet_sink.cpp
similarity index 95%
rename from be/src/vec/sink/vtabet_sink.cpp
rename to be/src/vec/sink/vtablet_sink.cpp
index 3b09cae..8b0f100 100644
--- a/be/src/vec/sink/vtabet_sink.cpp
+++ b/be/src/vec/sink/vtablet_sink.cpp
@@ -85,6 +85,9 @@ Status VOlapTableSink::send(RuntimeState* state, vectorized::Block* input_block)
BlockRow block_row;
SCOPED_RAW_TIMER(&_send_data_ns);
+ // TODO(cmy): vtablet_sink does not implement this "stop_processing" logic.
+ // This is just for passing compilation.
+ bool stop_processing = false;
for (int i = 0; i < num_rows; ++i) {
if (num_invalid_rows > 0 && _filter_vec[i] != 0) {
continue;
@@ -93,14 +96,12 @@ Status VOlapTableSink::send(RuntimeState* state, vectorized::Block* input_block)
uint32_t dist_hash = 0;
block_row = {&block, i};
if (!_vpartition->find_tablet(&block_row, &partition, &dist_hash)) {
- std::stringstream ss;
- ss << "no partition for this tuple. tuple="
- << block.dump_data(i, 1);
-#if BE_TEST
- LOG(INFO) << ss.str();
-#else
- state->append_error_msg_to_file("", ss.str());
-#endif
+ RETURN_IF_ERROR(state->append_error_msg_to_file([]() -> std::string { return ""; },
+ [&]() -> std::string {
+ fmt::memory_buffer buf;
+ fmt::format_to(buf, "no partition for this tuple. tuple=[]");
+ return buf.data();
+ }, &stop_processing));
_number_filtered_rows++;
continue;
}
@@ -123,6 +124,9 @@ Status VOlapTableSink::close(RuntimeState* state, Status exec_status) {
int VOlapTableSink::_validate_data(doris::RuntimeState* state, doris::vectorized::Block* block,
bool* filter_map) {
+ // TODO(cmy): implement it
+ return 0;
+#if 0
const auto num_rows = block->rows();
// set all row is valid
memset(filter_map, 0, num_rows * sizeof(bool));
@@ -269,6 +273,7 @@ int VOlapTableSink::_validate_data(doris::RuntimeState* state, doris::vectorized
filter_row += filter_map[i];
}
return filter_row;
+#endif
}
} // namespace stream_load
diff --git a/be/test/exec/tablet_sink_test.cpp b/be/test/exec/tablet_sink_test.cpp
index 6296f93..44c5fbd 100644
--- a/be/test/exec/tablet_sink_test.cpp
+++ b/be/test/exec/tablet_sink_test.cpp
@@ -31,6 +31,7 @@
#include "runtime/runtime_state.h"
#include "runtime/stream_load/load_stream_mgr.h"
#include "runtime/thread_resource_mgr.h"
+#include "runtime/types.h"
#include "runtime/tuple_row.h"
#include "service/brpc.h"
#include "util/brpc_stub_cache.h"
diff --git a/be/test/vec/runtime/vdata_stream_test.cpp b/be/test/vec/runtime/vdata_stream_test.cpp
index 24bb34c..cc4d429 100644
--- a/be/test/vec/runtime/vdata_stream_test.cpp
+++ b/be/test/vec/runtime/vdata_stream_test.cpp
@@ -15,7 +15,6 @@
// specific language governing permissions and limitations
// under the License.
-#include "bthread/id.h"
#include "common/object_pool.h"
#include "gen_cpp/internal_service.pb.h"
#include "google/protobuf/descriptor.h"
diff --git a/fe/fe-core/src/main/java/org/apache/doris/analysis/UserIdentity.java b/fe/fe-core/src/main/java/org/apache/doris/analysis/UserIdentity.java
index 481979a..538df18 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/analysis/UserIdentity.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/analysis/UserIdentity.java
@@ -63,14 +63,14 @@ public class UserIdentity implements Writable {
}
public UserIdentity(String user, String host) {
- this.user = Strings.emptyToNull(user);
- this.host = Strings.emptyToNull(host);
+ this.user = Strings.nullToEmpty(user);
+ this.host = Strings.nullToEmpty(host);
this.isDomain = false;
}
public UserIdentity(String user, String host, boolean isDomain) {
- this.user = Strings.emptyToNull(user);
- this.host = Strings.emptyToNull(host);
+ this.user = Strings.nullToEmpty(user);
+ this.host = Strings.nullToEmpty(host);
this.isDomain = isDomain;
}
@@ -122,6 +122,14 @@ public class UserIdentity implements Writable {
user = ClusterNamespace.getFullName(clusterName, user);
}
+ if (Strings.isNullOrEmpty(host)) {
+ if (!isDomain) {
+ host = "%";
+ } else {
+ throw new AnalysisException("Domain is empty");
+ }
+ }
+
// reuse createMysqlPattern to validate host pattern
PatternMatcher.createMysqlPattern(host, CaseSensibility.HOST.getCaseSensibility());
isAnalyzed = true;
@@ -230,3 +238,4 @@ public class UserIdentity implements Writable {
isAnalyzed = true;
}
}
+
diff --git a/fe/fe-core/src/main/java/org/apache/doris/load/ExportJob.java b/fe/fe-core/src/main/java/org/apache/doris/load/ExportJob.java
index 617832a..48818f1 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/load/ExportJob.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/load/ExportJob.java
@@ -452,7 +452,7 @@ public class ExportJob implements Writable {
TUniqueId queryId = new TUniqueId(uuid.getMostSignificantBits() + i, uuid.getLeastSignificantBits());
Coordinator coord = new Coordinator(
id, queryId, desc, Lists.newArrayList(fragment), Lists.newArrayList(scanNode),
- TimeUtils.DEFAULT_TIME_ZONE);
+ TimeUtils.DEFAULT_TIME_ZONE, true);
coord.setExecMemoryLimit(getExecMemLimit());
this.coordList.add(coord);
}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/load/loadv2/BrokerLoadJob.java b/fe/fe-core/src/main/java/org/apache/doris/load/loadv2/BrokerLoadJob.java
index 96cf48d..c0290f1 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/load/loadv2/BrokerLoadJob.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/load/loadv2/BrokerLoadJob.java
@@ -51,12 +51,12 @@ import org.apache.doris.transaction.TransactionState;
import org.apache.doris.transaction.TransactionState.TxnCoordinator;
import org.apache.doris.transaction.TransactionState.TxnSourceType;
-import com.google.common.base.Joiner;
-import com.google.common.collect.Lists;
-
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
+import com.google.common.base.Joiner;
+import com.google.common.collect.Lists;
+
import java.util.List;
import java.util.Map;
import java.util.UUID;
@@ -199,7 +199,9 @@ public class BrokerLoadJob extends BulkLoadJob {
LoadLoadingTask task = new LoadLoadingTask(db, table, brokerDesc,
brokerFileGroups, getDeadlineMs(), getExecMemLimit(),
isStrictMode(), transactionId, this, getTimeZone(), getTimeout(),
- getLoadParallelism(), getSendBatchParallelism(), enableProfile ? jobProfile : null);
+ getLoadParallelism(), getSendBatchParallelism(),
+ getMaxFilterRatio() <= 0, enableProfile ? jobProfile : null);
+
UUID uuid = UUID.randomUUID();
TUniqueId loadId = new TUniqueId(uuid.getMostSignificantBits(), uuid.getLeastSignificantBits());
task.init(loadId, attachment.getFileStatusByTable(aggKey),
diff --git a/fe/fe-core/src/main/java/org/apache/doris/load/loadv2/BrokerLoadingTaskAttachment.java b/fe/fe-core/src/main/java/org/apache/doris/load/loadv2/BrokerLoadingTaskAttachment.java
index cde0ed0..0a6acd9 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/load/loadv2/BrokerLoadingTaskAttachment.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/load/loadv2/BrokerLoadingTaskAttachment.java
@@ -37,7 +37,7 @@ public class BrokerLoadingTaskAttachment extends TaskAttachment {
}
public String getCounter(String key) {
- return counters.get(key);
+ return counters.getOrDefault(key, "0");
}
public String getTrackingUrl() {
diff --git a/fe/fe-core/src/main/java/org/apache/doris/load/loadv2/BulkLoadJob.java b/fe/fe-core/src/main/java/org/apache/doris/load/loadv2/BulkLoadJob.java
index e825ce0..7113363 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/load/loadv2/BulkLoadJob.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/load/loadv2/BulkLoadJob.java
@@ -47,6 +47,10 @@ import org.apache.doris.qe.SqlModeHelper;
import org.apache.doris.transaction.TabletCommitInfo;
import org.apache.doris.transaction.TransactionState;
+import org.apache.commons.lang.StringUtils;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+
import com.google.common.base.Strings;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
diff --git a/fe/fe-core/src/main/java/org/apache/doris/load/loadv2/LoadJob.java b/fe/fe-core/src/main/java/org/apache/doris/load/loadv2/LoadJob.java
index d8edfc6..5992bed 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/load/loadv2/LoadJob.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/load/loadv2/LoadJob.java
@@ -59,6 +59,9 @@ import org.apache.doris.transaction.BeginTransactionException;
import org.apache.doris.transaction.TransactionException;
import org.apache.doris.transaction.TransactionState;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+
import com.google.common.base.Joiner;
import com.google.common.collect.HashBasedTable;
import com.google.common.collect.Lists;
@@ -68,9 +71,6 @@ import com.google.common.collect.Table;
import com.google.gson.Gson;
import com.google.gson.annotations.SerializedName;
-import org.apache.logging.log4j.LogManager;
-import org.apache.logging.log4j.Logger;
-
import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
diff --git a/fe/fe-core/src/main/java/org/apache/doris/load/loadv2/LoadLoadingTask.java b/fe/fe-core/src/main/java/org/apache/doris/load/loadv2/LoadLoadingTask.java
index 301da84..dc6c371 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/load/loadv2/LoadLoadingTask.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/load/loadv2/LoadLoadingTask.java
@@ -65,6 +65,7 @@ public class LoadLoadingTask extends LoadTask {
private final long timeoutS;
private final int loadParallelism;
private final int sendBatchParallelism;
+ private final boolean loadZeroTolerance;
private LoadingTaskPlanner planner;
@@ -75,7 +76,8 @@ public class LoadLoadingTask extends LoadTask {
BrokerDesc brokerDesc, List<BrokerFileGroup> fileGroups,
long jobDeadlineMs, long execMemLimit, boolean strictMode,
long txnId, LoadTaskCallback callback, String timezone,
- long timeoutS, int loadParallelism, int sendBatchParallelism, RuntimeProfile profile) {
+ long timeoutS, int loadParallelism, int sendBatchParallelism,
+ boolean loadZeroTolerance, RuntimeProfile profile) {
super(callback, TaskType.LOADING);
this.db = db;
this.table = table;
@@ -91,6 +93,7 @@ public class LoadLoadingTask extends LoadTask {
this.timeoutS = timeoutS;
this.loadParallelism = loadParallelism;
this.sendBatchParallelism = sendBatchParallelism;
+ this.loadZeroTolerance = loadZeroTolerance;
this.jobProfile = profile;
}
@@ -118,7 +121,7 @@ public class LoadLoadingTask extends LoadTask {
private void executeOnce() throws Exception {
// New one query id,
Coordinator curCoordinator = new Coordinator(callback.getCallbackId(), loadId, planner.getDescTable(),
- planner.getFragments(), planner.getScanNodes(), planner.getTimezone());
+ planner.getFragments(), planner.getScanNodes(), planner.getTimezone(), loadZeroTolerance);
curCoordinator.setQueryType(TQueryType.LOAD);
curCoordinator.setExecMemoryLimit(execMemLimit);
/*
@@ -157,9 +160,9 @@ public class LoadLoadingTask extends LoadTask {
Status status = curCoordinator.getExecStatus();
if (status.ok()) {
attachment = new BrokerLoadingTaskAttachment(signature,
- curCoordinator.getLoadCounters(),
- curCoordinator.getTrackingUrl(),
- TabletCommitInfo.fromThrift(curCoordinator.getCommitInfos()));
+ curCoordinator.getLoadCounters(),
+ curCoordinator.getTrackingUrl(),
+ TabletCommitInfo.fromThrift(curCoordinator.getCommitInfos()));
// Create profile of this task and add to the job profile.
createProfile(curCoordinator);
} else {
diff --git a/fe/fe-core/src/main/java/org/apache/doris/load/routineload/KafkaRoutineLoadJob.java b/fe/fe-core/src/main/java/org/apache/doris/load/routineload/KafkaRoutineLoadJob.java
index a24dd29..7d4ad46 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/load/routineload/KafkaRoutineLoadJob.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/load/routineload/KafkaRoutineLoadJob.java
@@ -685,4 +685,10 @@ public class KafkaRoutineLoadJob extends RoutineLoadJob {
Gson gson = new Gson();
return gson.toJson(partitionIdToOffsetLag);
}
+
+ @Override
+ public double getMaxFilterRatio() {
+ // for kafka routine load, the max filter ratio is always 1, because it use max error num instead of this.
+ return 1.0;
+ }
}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/load/update/UpdateStmtExecutor.java b/fe/fe-core/src/main/java/org/apache/doris/load/update/UpdateStmtExecutor.java
index e5cc97b..e1ec792 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/load/update/UpdateStmtExecutor.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/load/update/UpdateStmtExecutor.java
@@ -139,7 +139,7 @@ public class UpdateStmtExecutor {
private void executePlan() throws Exception {
LOG.info("begin execute update stmt, query id:{}", DebugUtil.printId(queryId));
coordinator = new Coordinator(Catalog.getCurrentCatalog().getNextId(), queryId, analyzer.getDescTbl(),
- updatePlanner.getFragments(), updatePlanner.getScanNodes(), TimeUtils.DEFAULT_TIME_ZONE);
+ updatePlanner.getFragments(), updatePlanner.getScanNodes(), TimeUtils.DEFAULT_TIME_ZONE, false);
coordinator.setQueryType(TQueryType.LOAD);
QeProcessorImpl.INSTANCE.registerQuery(queryId, coordinator);
analyzer.getContext().getExecutor().setCoord(coordinator);
diff --git a/fe/fe-core/src/main/java/org/apache/doris/planner/DataPartition.java b/fe/fe-core/src/main/java/org/apache/doris/planner/DataPartition.java
index fabb260..3c80016 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/planner/DataPartition.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/planner/DataPartition.java
@@ -55,8 +55,6 @@ public class DataPartition {
// for hash partition: exprs used to compute hash value
private ImmutableList<Expr> partitionExprs = ImmutableList.of();
- private List<DataSplitSink.EtlRangePartitionInfo> partitions;
-
public DataPartition(TPartitionType type, List<Expr> exprs) {
Preconditions.checkNotNull(exprs);
Preconditions.checkState(!exprs.isEmpty());
@@ -67,14 +65,6 @@ public class DataPartition {
this.partitionExprs = ImmutableList.copyOf(exprs);
}
- public DataPartition(
- List<Expr> partitionExprs,
- List<DataSplitSink.EtlRangePartitionInfo> partitions) {
- this.type = TPartitionType.RANGE_PARTITIONED;
- this.partitionExprs = ImmutableList.copyOf(partitionExprs);
- this.partitions = partitions;
- }
-
public void substitute(ExprSubstitutionMap smap, Analyzer analyzer) throws AnalysisException {
List<Expr> list = Expr.trySubstituteList(partitionExprs, smap, analyzer, false);
partitionExprs = ImmutableList.copyOf(list);
@@ -112,9 +102,6 @@ public class DataPartition {
if (partitionExprs != null) {
result.setPartitionExprs(Expr.treesToThrift(partitionExprs));
}
- if (partitions != null) {
- result.setPartitionInfos(DataSplitSink.EtlRangePartitionInfo.listToThrift(partitions));
- }
return result;
}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/planner/DataSplitSink.java b/fe/fe-core/src/main/java/org/apache/doris/planner/DataSplitSink.java
deleted file mode 100644
index 217f53d..0000000
--- a/fe/fe-core/src/main/java/org/apache/doris/planner/DataSplitSink.java
+++ /dev/null
@@ -1,401 +0,0 @@
-// 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.
-
-package org.apache.doris.planner;
-
-import org.apache.doris.analysis.Expr;
-import org.apache.doris.analysis.SlotDescriptor;
-import org.apache.doris.analysis.SlotRef;
-import org.apache.doris.analysis.TupleDescriptor;
-import org.apache.doris.catalog.AggregateType;
-import org.apache.doris.catalog.Column;
-import org.apache.doris.catalog.DistributionInfo;
-import org.apache.doris.catalog.KeysType;
-import org.apache.doris.catalog.OlapTable;
-import org.apache.doris.catalog.Partition;
-import org.apache.doris.catalog.PartitionInfo;
-import org.apache.doris.catalog.PartitionKey;
-import org.apache.doris.catalog.PartitionType;
-import org.apache.doris.catalog.PrimitiveType;
-import org.apache.doris.catalog.RangePartitionInfo;
-import org.apache.doris.catalog.Type;
-import org.apache.doris.common.AnalysisException;
-import org.apache.doris.thrift.TAggregationType;
-import org.apache.doris.thrift.TDataSink;
-import org.apache.doris.thrift.TDataSinkType;
-import org.apache.doris.thrift.TDataSplitSink;
-import org.apache.doris.thrift.TExplainLevel;
-import org.apache.doris.thrift.TPartitionKey;
-import org.apache.doris.thrift.TPartitionRange;
-import org.apache.doris.thrift.TRangePartition;
-import org.apache.doris.thrift.TRollupSchema;
-
-import com.google.common.base.Function;
-import com.google.common.base.Joiner;
-import com.google.common.base.Preconditions;
-import com.google.common.collect.Lists;
-import com.google.common.collect.Maps;
-import com.google.common.collect.Range;
-
-import org.apache.logging.log4j.LogManager;
-import org.apache.logging.log4j.Logger;
-
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-
-// This class used to split data read from file to batch
-@Deprecated
-public class DataSplitSink extends DataSink {
- private static final Logger LOG = LogManager.getLogger(DataSplitSink.class);
-
- private final OlapTable targetTable;
-
- private List<Expr> partitionExprs;
- private List<EtlRangePartitionInfo> partitions;
- private Map<String, RollupSchema> rollups;
-
- public DataSplitSink(OlapTable targetTable, TupleDescriptor desc) throws AnalysisException {
- this.targetTable = targetTable;
- Map<String, Expr> slotByCol = Maps.newTreeMap(String.CASE_INSENSITIVE_ORDER);
- for (int i = 0; i < desc.getSlots().size(); ++i) {
- Column col = targetTable.getBaseSchema().get(i);
- SlotDescriptor slotDesc = desc.getSlots().get(i);
- SlotRef slotRef = new SlotRef(slotDesc);
- if (slotDesc.getType().getPrimitiveType() == PrimitiveType.CHAR) {
- slotRef.setType(Type.CHAR);
- }
- slotByCol.put(col.getName(), slotRef);
- }
- rollups = RollupSchema.createRollup(targetTable, slotByCol);
-
- partitionExprs = Lists.newArrayList();
- partitions = EtlRangePartitionInfo.createParts(targetTable, slotByCol, null, partitionExprs);
- }
-
- @Override
- public String getExplainString(String prefix, TExplainLevel explainLevel) {
- StringBuilder strBuilder = new StringBuilder();
- strBuilder.append(prefix + "DPP SINK\n");
- strBuilder.append(prefix + " TABLE NAME: " + targetTable.getName() + "\n");
- strBuilder.append(prefix + " PARTITIONS: " + Joiner.on(", ").join(
- Lists.transform(partitions, new Function<EtlRangePartitionInfo, Long>() {
- @Override
- public Long apply(EtlRangePartitionInfo etlRangePartitionInfo) {
- return etlRangePartitionInfo.getId();
- }
- })) + "\n");
- return strBuilder.toString();
- }
-
- @Override
- protected TDataSink toThrift() {
- TDataSplitSink tSplitSink = new TDataSplitSink(Expr.treesToThrift(partitionExprs),
- EtlRangePartitionInfo.listToThrift(partitions),
- RollupSchema.mapToThrift(rollups));
-
- TDataSink tDataSink = new TDataSink(TDataSinkType.DATA_SPLIT_SINK);
- tDataSink.setSplitSink(tSplitSink);
- return tDataSink;
- }
-
- @Override
- public PlanNodeId getExchNodeId() {
- return null;
- }
-
- @Override
- public DataPartition getOutputPartition() {
- return new DataPartition(partitionExprs, partitions);
- }
-
- // Class used to describe rollup schema
- public static class RollupSchema {
- private final KeysType keysType;
- private final List<Expr> keys;
- private final List<Expr> values;
- private final List<AggregateType> valueOps;
-
- public RollupSchema(KeysType keysType, List<Expr> keys, List<Expr> values, List<AggregateType> valueOps) {
- this.keysType = keysType;
- this.keys = keys;
- this.values = values;
- this.valueOps = valueOps;
- }
-
- public TRollupSchema toThrift() {
- TRollupSchema tRollupSchema = new TRollupSchema();
-
- tRollupSchema.setKeysType(keysType.name());
- tRollupSchema.setKeys(Expr.treesToThrift(keys));
- tRollupSchema.setValues(Expr.treesToThrift(values));
- tRollupSchema.setValueOps(Lists.transform(valueOps,
- new Function<AggregateType, TAggregationType>() {
- @Override
- public TAggregationType apply(AggregateType aggregateType) {
- return aggregateType.toThrift();
- }
- })
- );
-
-
- return tRollupSchema;
- }
-
- public static Map<String, TRollupSchema> mapToThrift(Map<String, RollupSchema> rollups) {
- return Maps.transformValues(rollups, new Function<RollupSchema, TRollupSchema>() {
- @Override
- public TRollupSchema apply(RollupSchema rollupSchema) {
- return rollupSchema.toThrift();
- }
- });
- }
-
- // Create one rollup schema
- private static RollupSchema createRollupSchema(
- KeysType keysType, List<Column> columns, Map<String, Expr> exprByCol) {
- List<Expr> keys = Lists.newArrayList();
- List<Expr> values = Lists.newArrayList();
- List<AggregateType> valueOps = Lists.newArrayList();
-
- for (Column column : columns) {
- Expr expr = exprByCol.get(column.getName());
- // NOTE: use CHAR instead of Varchar, because two types is different in storage
- if (column.getDataType() == PrimitiveType.CHAR) {
- expr.setType(Type.CHAR);
- }
- if (column.isKey()) {
- // Key column
- keys.add(expr);
- } else {
- values.add(expr);
- valueOps.add(column.getAggregationType());
- }
- }
-
- return new RollupSchema(keysType, keys, values, valueOps);
- }
-
- public static Map<String, RollupSchema> createRollup(OlapTable tbl, Map<String, Expr> exprByCol) {
- Map<String, RollupSchema> rollups = Maps.newHashMap();
- // create rollup schema
- // add schema hash to job load info
- for (Map.Entry<Long, Integer> entry : tbl.getIndexIdToSchemaHash().entrySet()) {
- long indexId = entry.getKey();
- rollups.put(String.valueOf(indexId),
- createRollupSchema(tbl.getKeysType(), tbl.getSchemaByIndexId(indexId), exprByCol));
- }
- return rollups;
- }
- }
-
- // Partition key used to split data into different partitions.
- public static class EtlPartitionKey {
- public static final EtlPartitionKey POS_INFINITE = new EtlPartitionKey((short) 1);
- public static final EtlPartitionKey NEG_INFINITE = new EtlPartitionKey((short) -1);
-
- // when sign is less than 0 means this key is negative infinite
- // when sign is greater than 0 means this key is positive infinite
- // when sign is 0 means this key represent by item of 'type' and 'key'
- private final short sign;
- private final PrimitiveType type;
- // Memory content in c language for build-in type.
- // and utf-8 string for others.
- private final String key;
-
- private EtlPartitionKey(short sign) {
- this.sign = sign;
- this.type = null;
- this.key = null;
- }
-
- public EtlPartitionKey(PrimitiveType type, String key) {
- this.sign = 0;
- this.type = type;
- this.key = key;
- }
-
- public static EtlPartitionKey getPosInfinite() {
- return POS_INFINITE;
- }
-
- public static EtlPartitionKey getNegInfinite() {
- return NEG_INFINITE;
- }
-
- public TPartitionKey toThrift() {
- TPartitionKey tKey = new TPartitionKey(sign);
- if (sign == 0) {
- tKey.setType(type.toThrift());
- tKey.setKey(key);
- }
-
- return tKey;
- }
- }
-
- // One range of a partition
- public static class EtlPartitionRange {
- private final EtlPartitionKey startKey;
- private final EtlPartitionKey endKey;
- private final boolean includeStartKey;
- private final boolean includeEndKey;
-
- public static EtlPartitionRange closed(EtlPartitionKey startKey, EtlPartitionKey endKey) {
- return new EtlPartitionRange(startKey, true, endKey, true);
- }
-
- public static EtlPartitionRange closedOpen(EtlPartitionKey startKey, EtlPartitionKey endKey) {
- return new EtlPartitionRange(startKey, true, endKey, false);
- }
-
- public static EtlPartitionRange all() {
- return closed(EtlPartitionKey.getNegInfinite(), EtlPartitionKey.getPosInfinite());
- }
-
- private EtlPartitionRange(EtlPartitionKey startKey, boolean includeStartKey,
- EtlPartitionKey endKey, boolean includeEndKey) {
- this.startKey = startKey;
- this.includeStartKey = includeStartKey;
- this.endKey = endKey;
- this.includeEndKey = includeEndKey;
- }
-
- public TPartitionRange toThrift() {
- return new TPartitionRange(startKey.toThrift(), endKey.toThrift(),
- includeStartKey, includeEndKey);
- }
- }
-
- // This partition under table
- public static class EtlRangePartitionInfo {
- private long id;
- private EtlPartitionRange range;
- private List<Expr> distributeExprs;
- private int distributeBuckets;
-
- public EtlRangePartitionInfo(long id, EtlPartitionRange range, List<Expr> distributeExprs,
- int distributeBuckets) {
- this.id = id;
- this.range = range;
- this.distributeExprs = distributeExprs;
- this.distributeBuckets = distributeBuckets;
- }
-
- public long getId() {
- return id;
- }
-
- public TRangePartition toThrift(boolean containDist) {
- TRangePartition tRangePartition = new TRangePartition(id, range.toThrift());
- if (containDist) {
- tRangePartition.setDistributedExprs(Expr.treesToThrift(distributeExprs));
- tRangePartition.setDistributeBucket(distributeBuckets);
- }
- return tRangePartition;
- }
-
- public static List<TRangePartition> listToThrift(List<EtlRangePartitionInfo> partitionInfos) {
- return Lists.transform(partitionInfos, new Function<EtlRangePartitionInfo, TRangePartition>() {
- @Override
- public TRangePartition apply(EtlRangePartitionInfo partitionInfo) {
- return partitionInfo.toThrift(true);
- }
- });
- }
-
- public static List<TRangePartition> listToNonDistThrift(List<EtlRangePartitionInfo> partitionInfos) {
- return Lists.transform(partitionInfos, new Function<EtlRangePartitionInfo, TRangePartition>() {
- @Override
- public TRangePartition apply(EtlRangePartitionInfo partitionInfo) {
- return partitionInfo.toThrift(false);
- }
- });
- }
-
- private static EtlPartitionRange createPartitionRange(
- RangePartitionInfo rangePartitionInfo, long partitionId) throws AnalysisException {
- EtlPartitionRange partitionRange = null;
- // only one partition column
- Column partitionColumn = rangePartitionInfo.getPartitionColumns().get(0);
- PrimitiveType columnType = partitionColumn.getDataType();
-
- Range<PartitionKey> range = rangePartitionInfo.getItem(partitionId).getItems();
- // start key
- EtlPartitionKey startPartitionKey = null;
- PartitionKey startKey = range.lowerEndpoint();
- String startKeyStr = startKey.getKeys().get(0).getStringValue();
- if (startKey.isMinValue()) {
- startPartitionKey = EtlPartitionKey.NEG_INFINITE;
- } else {
- startPartitionKey = new EtlPartitionKey(columnType, startKeyStr);
- }
-
- // end key
- EtlPartitionKey endPartitionKey = null;
- PartitionKey endKey = range.upperEndpoint();
- String endKeyStr = endKey.getKeys().get(0).getStringValue();
- if (endKey.isMaxValue()) {
- endPartitionKey = EtlPartitionKey.POS_INFINITE;
- } else {
- endPartitionKey = new EtlPartitionKey(columnType, endKeyStr);
- }
-
- // create partition range
- partitionRange = EtlPartitionRange.closedOpen(startPartitionKey, endPartitionKey);
- return partitionRange;
- }
-
- public static List<EtlRangePartitionInfo> createParts(
- OlapTable tbl, Map<String, Expr> exprByCol, Set<Long> targetPartitionIds,
- List<Expr> partitionExprs) throws AnalysisException {
- List<EtlRangePartitionInfo> parts = Lists.newArrayList();
- PartitionInfo partInfo = tbl.getPartitionInfo();
- if (partInfo.getType() == PartitionType.RANGE) {
- RangePartitionInfo rangePartitionInfo = (RangePartitionInfo) partInfo;
- for (Column col : rangePartitionInfo.getPartitionColumns()) {
- partitionExprs.add(exprByCol.get(col.getName()));
- }
- for (Partition part : tbl.getAllPartitions()) {
- if (targetPartitionIds != null && !targetPartitionIds.contains(part.getId())) {
- continue;
- }
- DistributionInfo distInfo = part.getDistributionInfo();
- parts.add(new EtlRangePartitionInfo(
- part.getId(),
- createPartitionRange(rangePartitionInfo, part.getId()),
- DistributionInfo.toDistExpr(tbl, distInfo, exprByCol),
- distInfo.getBucketNum()));
- }
- } else if (partInfo.getType() == PartitionType.UNPARTITIONED) {
- Preconditions.checkState(tbl.getPartitions().size() == 1, tbl.getId());
- for (Partition part : tbl.getPartitions()) {
- DistributionInfo distInfo = part.getDistributionInfo();
- parts.add(new EtlRangePartitionInfo(
- part.getId(),
- EtlPartitionRange.all(),
- DistributionInfo.toDistExpr(tbl, distInfo, exprByCol),
- distInfo.getBucketNum()));
- }
- } else {
- throw new AnalysisException("Unsupported partition type: " + partInfo.getType());
- }
- return parts;
- }
-
- }
-}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/planner/OlapTableSink.java b/fe/fe-core/src/main/java/org/apache/doris/planner/OlapTableSink.java
index 1a9a930..edc7627 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/planner/OlapTableSink.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/planner/OlapTableSink.java
@@ -100,8 +100,7 @@ public class OlapTableSink extends DataSink {
tSink.setDbId(dbId);
tSink.setLoadChannelTimeoutS(loadChannelTimeoutS);
tSink.setSendBatchParallelism(sendBatchParallelism);
- tDataSink = new TDataSink(TDataSinkType.DATA_SPLIT_SINK);
- tDataSink.setType(TDataSinkType.OLAP_TABLE_SINK);
+ tDataSink = new TDataSink(TDataSinkType.OLAP_TABLE_SINK);
tDataSink.setOlapTableSink(tSink);
if (partitionIds == null) {
diff --git a/fe/fe-core/src/main/java/org/apache/doris/planner/StreamLoadPlanner.java b/fe/fe-core/src/main/java/org/apache/doris/planner/StreamLoadPlanner.java
index 5bd4cf6..c2a2e71 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/planner/StreamLoadPlanner.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/planner/StreamLoadPlanner.java
@@ -53,12 +53,12 @@ import org.apache.doris.thrift.TScanRangeLocations;
import org.apache.doris.thrift.TScanRangeParams;
import org.apache.doris.thrift.TUniqueId;
-import com.google.common.collect.Lists;
-import com.google.common.collect.Maps;
-
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
+
import java.text.DateFormat;
import java.text.SimpleDateFormat;
import java.util.Date;
@@ -195,6 +195,8 @@ public class StreamLoadPlanner {
queryGlobals.setNowString(DATE_FORMAT.format(new Date()));
queryGlobals.setTimestampMs(new Date().getTime());
queryGlobals.setTimeZone(taskInfo.getTimezone());
+ queryGlobals.setLoadZeroTolerance(taskInfo.getMaxFilterRatio() <= 0.0);
+
params.setQueryGlobals(queryGlobals);
// set load error hub if exist
diff --git a/fe/fe-core/src/main/java/org/apache/doris/qe/Coordinator.java b/fe/fe-core/src/main/java/org/apache/doris/qe/Coordinator.java
index c30e021..87d9c2b 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/qe/Coordinator.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/qe/Coordinator.java
@@ -233,6 +233,7 @@ public class Coordinator {
this.queryGlobals.setNowString(DATE_FORMAT.format(new Date()));
this.queryGlobals.setTimestampMs(new Date().getTime());
+ this.queryGlobals.setLoadZeroTolerance(false);
if (context.getSessionVariable().getTimeZone().equals("CST")) {
this.queryGlobals.setTimeZone(TimeUtils.DEFAULT_TIME_ZONE);
} else {
@@ -249,7 +250,8 @@ public class Coordinator {
// Used for broker load task/export task/update coordinator
public Coordinator(Long jobId, TUniqueId queryId, DescriptorTable descTable,
- List<PlanFragment> fragments, List<ScanNode> scanNodes, String timezone) {
+ List<PlanFragment> fragments, List<ScanNode> scanNodes, String timezone,
+ boolean loadZeroTolerance) {
this.isBlockQuery = true;
this.jobId = jobId;
this.queryId = queryId;
@@ -260,6 +262,7 @@ public class Coordinator {
this.queryGlobals.setNowString(DATE_FORMAT.format(new Date()));
this.queryGlobals.setTimestampMs(new Date().getTime());
this.queryGlobals.setTimeZone(timezone);
+ this.queryGlobals.setLoadZeroTolerance(loadZeroTolerance);
this.tResourceInfo = new TResourceInfo("", "");
this.needReport = true;
this.nextInstanceId = new TUniqueId();
@@ -335,6 +338,10 @@ public class Coordinator {
this.queryOptions.setQueryTimeout(timeout);
}
+ public void setLoadZeroTolerance(boolean loadZeroTolerance) {
+ this.queryGlobals.setLoadZeroTolerance(loadZeroTolerance);
+ }
+
public void clearExportStatus() {
lock.lock();
try {
diff --git a/fe/fe-core/src/main/java/org/apache/doris/qe/StmtExecutor.java b/fe/fe-core/src/main/java/org/apache/doris/qe/StmtExecutor.java
index 6fc0320..975ad36 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/qe/StmtExecutor.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/qe/StmtExecutor.java
@@ -1241,6 +1241,7 @@ public class StmtExecutor implements ProfileWriter {
try {
coord = new Coordinator(context, analyzer, planner);
+ coord.setLoadZeroTolerance(context.getSessionVariable().getEnableInsertStrict());
coord.setQueryType(TQueryType.LOAD);
QeProcessorImpl.INSTANCE.registerQuery(context.queryId(), coord);
diff --git a/fe/fe-core/src/main/java/org/apache/doris/task/LoadTaskInfo.java b/fe/fe-core/src/main/java/org/apache/doris/task/LoadTaskInfo.java
index 6deae19..e5a39f3 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/task/LoadTaskInfo.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/task/LoadTaskInfo.java
@@ -50,6 +50,8 @@ public interface LoadTaskInfo {
public boolean isReadJsonByLine();
public String getPath();
+ public double getMaxFilterRatio();
+
public ImportColumnDescs getColumnExprDescs();
public boolean isStrictMode();
diff --git a/fe/fe-core/src/main/java/org/apache/doris/task/StreamLoadTask.java b/fe/fe-core/src/main/java/org/apache/doris/task/StreamLoadTask.java
index b47b91a..08b7f80 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/task/StreamLoadTask.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/task/StreamLoadTask.java
@@ -17,11 +17,11 @@
package org.apache.doris.task;
-import org.apache.doris.analysis.Separator;
import org.apache.doris.analysis.Expr;
import org.apache.doris.analysis.ImportColumnsStmt;
import org.apache.doris.analysis.ImportWhereStmt;
import org.apache.doris.analysis.PartitionNames;
+import org.apache.doris.analysis.Separator;
import org.apache.doris.analysis.SqlParser;
import org.apache.doris.analysis.SqlScanner;
import org.apache.doris.common.AnalysisException;
@@ -74,6 +74,7 @@ public class StreamLoadTask implements LoadTaskInfo {
private Expr deleteCondition;
private String sequenceCol;
private int sendBatchParallelism = 1;
+ private double maxFilterRatio = 0.0;
public StreamLoadTask(TUniqueId id, long txnId, TFileType fileType, TFileFormatType formatType) {
this.id = id;
@@ -297,6 +298,9 @@ public class StreamLoadTask implements LoadTaskInfo {
if (request.isSetSendBatchParallelism()) {
sendBatchParallelism = request.getSendBatchParallelism();
}
+ if (request.isSetMaxFilterRatio()) {
+ maxFilterRatio = request.getMaxFilterRatio();
+ }
}
// used for stream load
@@ -367,4 +371,9 @@ public class StreamLoadTask implements LoadTaskInfo {
public long getMemLimit() {
return execMemLimit;
}
+
+ @Override
+ public double getMaxFilterRatio() {
+ return maxFilterRatio;
+ }
}
diff --git a/fe/fe-core/src/test/java/org/apache/doris/alter/MaterializedViewHandlerTest.java b/fe/fe-core/src/test/java/org/apache/doris/alter/MaterializedViewHandlerTest.java
index ee43f46..e33f840 100644
--- a/fe/fe-core/src/test/java/org/apache/doris/alter/MaterializedViewHandlerTest.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/alter/MaterializedViewHandlerTest.java
@@ -30,11 +30,13 @@ import org.apache.doris.catalog.Type;
import org.apache.doris.common.jmockit.Deencapsulation;
import com.google.common.collect.Lists;
+import com.google.common.collect.Sets;
import org.junit.Assert;
import org.junit.Test;
import java.util.List;
+import java.util.Set;
import mockit.Expectations;
import mockit.Injectable;
@@ -55,7 +57,7 @@ public class MaterializedViewHandlerTest {
MaterializedViewHandler materializedViewHandler = new MaterializedViewHandler();
try {
Deencapsulation.invoke(materializedViewHandler, "processCreateMaterializedView", createMaterializedViewStmt,
- db, olapTable);
+ db, olapTable);
Assert.fail();
} catch (Exception e) {
System.out.print(e.getMessage());
@@ -80,7 +82,7 @@ public class MaterializedViewHandlerTest {
MaterializedViewHandler materializedViewHandler = new MaterializedViewHandler();
try {
Deencapsulation.invoke(materializedViewHandler, "processCreateMaterializedView", createMaterializedViewStmt,
- db, olapTable);
+ db, olapTable);
Assert.fail();
} catch (Exception e) {
System.out.print(e.getMessage());
@@ -107,7 +109,7 @@ public class MaterializedViewHandlerTest {
MaterializedViewHandler materializedViewHandler = new MaterializedViewHandler();
try {
Deencapsulation.invoke(materializedViewHandler, "processCreateMaterializedView",
- createMaterializedViewStmt, db, olapTable);
+ createMaterializedViewStmt, db, olapTable);
Assert.fail();
} catch (Exception e) {
System.out.print(e.getMessage());
@@ -135,7 +137,7 @@ public class MaterializedViewHandlerTest {
olapTable.getPartitions();
result = Lists.newArrayList(partition);
partition.getIndex(baseIndexId);
- result = materializedIndex;
+ result = materializedIndex;
materializedIndex.getState();
result = MaterializedIndex.IndexState.SHADOW;
}
@@ -143,7 +145,7 @@ public class MaterializedViewHandlerTest {
MaterializedViewHandler materializedViewHandler = new MaterializedViewHandler();
try {
Deencapsulation.invoke(materializedViewHandler, "processCreateMaterializedView",
- createMaterializedViewStmt, db, olapTable);
+ createMaterializedViewStmt, db, olapTable);
Assert.fail();
} catch (Exception e) {
System.out.print(e.getMessage());
@@ -165,7 +167,7 @@ public class MaterializedViewHandlerTest {
MaterializedViewHandler materializedViewHandler = new MaterializedViewHandler();
try {
Deencapsulation.invoke(materializedViewHandler, "checkAndPrepareMaterializedView",
- createMaterializedViewStmt, olapTable);
+ createMaterializedViewStmt, olapTable);
Assert.fail();
} catch (Exception e) {
System.out.print(e.getMessage());
@@ -200,7 +202,7 @@ public class MaterializedViewHandlerTest {
MaterializedViewHandler materializedViewHandler = new MaterializedViewHandler();
try {
Deencapsulation.invoke(materializedViewHandler, "checkAndPrepareMaterializedView",
- createMaterializedViewStmt, olapTable);
+ createMaterializedViewStmt, olapTable);
Assert.fail();
} catch (Exception e) {
System.out.print(e.getMessage());
@@ -223,7 +225,7 @@ public class MaterializedViewHandlerTest {
MaterializedViewHandler materializedViewHandler = new MaterializedViewHandler();
try {
Deencapsulation.invoke(materializedViewHandler, "checkAndPrepareMaterializedView",
- createMaterializedViewStmt, olapTable);
+ createMaterializedViewStmt, olapTable);
Assert.fail();
} catch (Exception e) {
System.out.print(e.getMessage());
@@ -256,8 +258,8 @@ public class MaterializedViewHandlerTest {
MaterializedViewHandler materializedViewHandler = new MaterializedViewHandler();
try {
List<Column> mvColumns = Deencapsulation.invoke(materializedViewHandler,
- "checkAndPrepareMaterializedView",
- createMaterializedViewStmt, olapTable);
+ "checkAndPrepareMaterializedView",
+ createMaterializedViewStmt, olapTable);
Assert.assertEquals(1, mvColumns.size());
Column newMVColumn = mvColumns.get(0);
Assert.assertEquals(columnName1, newMVColumn.getName());
@@ -280,7 +282,7 @@ public class MaterializedViewHandlerTest {
MVColumnItem mvColumnItem = new MVColumnItem(columnName1, Type.VARCHAR);
mvColumnItem.setIsKey(false);
mvColumnItem.setAggregationType(AggregateType.SUM, false);
- List<String> partitionColumnNames = Lists.newArrayList();
+ Set<String> partitionColumnNames = Sets.newHashSet();
partitionColumnNames.add(columnName1);
new Expectations() {
{
@@ -299,7 +301,7 @@ public class MaterializedViewHandlerTest {
MaterializedViewHandler materializedViewHandler = new MaterializedViewHandler();
try {
Deencapsulation.invoke(materializedViewHandler, "checkAndPrepareMaterializedView",
- createMaterializedViewStmt, olapTable);
+ createMaterializedViewStmt, olapTable);
Assert.fail();
} catch (Exception e) {
System.out.print(e.getMessage());
@@ -309,7 +311,7 @@ public class MaterializedViewHandlerTest {
@Test
public void testCheckDropMaterializedView(@Injectable OlapTable olapTable, @Injectable Partition partition,
- @Injectable MaterializedIndex materializedIndex) {
+ @Injectable MaterializedIndex materializedIndex) {
String mvName = "mv_1";
new Expectations() {
{
diff --git a/fe/fe-core/src/test/java/org/apache/doris/catalog/RecoverTest.java b/fe/fe-core/src/test/java/org/apache/doris/catalog/RecoverTest.java
index 68408e5..bf5230e 100644
--- a/fe/fe-core/src/test/java/org/apache/doris/catalog/RecoverTest.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/catalog/RecoverTest.java
@@ -53,7 +53,6 @@ public class RecoverTest {
connectContext = UtFrameUtils.createDefaultCtx();
}
-
@AfterClass
public static void tearDown() {
File file = new File(runningDir);
@@ -61,7 +60,7 @@ public class RecoverTest {
}
private static void createDb(String db) throws Exception {
- CreateDbStmt createDbStmt = (CreateDbStmt)UtFrameUtils.parseAndAnalyzeStmt("create database " + db, connectContext);
+ CreateDbStmt createDbStmt = (CreateDbStmt) UtFrameUtils.parseAndAnalyzeStmt("create database " + db, connectContext);
Catalog.getCurrentCatalog().createDb(createDbStmt);
}
@@ -71,7 +70,7 @@ public class RecoverTest {
}
private static void dropDb(String db) throws Exception {
- DropDbStmt dropDbStmt = (DropDbStmt)UtFrameUtils.parseAndAnalyzeStmt("drop database " + db, connectContext);
+ DropDbStmt dropDbStmt = (DropDbStmt) UtFrameUtils.parseAndAnalyzeStmt("drop database " + db, connectContext);
Catalog.getCurrentCatalog().dropDb(dropDbStmt);
}
@@ -159,15 +158,15 @@ public class RecoverTest {
Assert.assertTrue(checkDbExist("test"));
Assert.assertTrue(checkTableExist("test", "table1"));
- dropTable("test","table1");
+ dropTable("test", "table1");
Assert.assertTrue(checkDbExist("test"));
Assert.assertFalse(checkTableExist("test", "table1"));
- recoverTable("test","table1");
+ recoverTable("test", "table1");
Assert.assertTrue(checkDbExist("test"));
Assert.assertTrue(checkTableExist("test", "table1"));
- dropTable("test","table1");
+ dropTable("test", "table1");
Assert.assertTrue(checkDbExist("test"));
Assert.assertFalse(checkTableExist("test", "table1"));
@@ -201,17 +200,17 @@ public class RecoverTest {
Assert.assertTrue(checkTableExist("test", "table1"));
try {
- recoverTable("test","table1");
+ recoverTable("test", "table1");
Assert.fail("should not recover succeed");
} catch (DdlException e) {
e.printStackTrace();
}
Assert.assertTrue(checkPartitionExist("test", "table1", "p1"));
- dropPartition("test","table1", "p1");
+ dropPartition("test", "table1", "p1");
Assert.assertFalse(checkPartitionExist("test", "table1", "p1"));
- recoverPartition("test","table1", "p1");
+ recoverPartition("test", "table1", "p1");
Assert.assertTrue(checkPartitionExist("test", "table1", "p1"));
}
diff --git a/fe/fe-core/src/test/java/org/apache/doris/load/loadv2/BrokerLoadJobTest.java b/fe/fe-core/src/test/java/org/apache/doris/load/loadv2/BrokerLoadJobTest.java
index 753df1a..ae96752 100644
--- a/fe/fe-core/src/test/java/org/apache/doris/load/loadv2/BrokerLoadJobTest.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/load/loadv2/BrokerLoadJobTest.java
@@ -17,12 +17,6 @@
package org.apache.doris.load.loadv2;
-import mockit.Expectations;
-import mockit.Injectable;
-import mockit.Mock;
-import mockit.MockUp;
-import mockit.Mocked;
-
import org.apache.doris.analysis.Analyzer;
import org.apache.doris.analysis.BrokerDesc;
import org.apache.doris.analysis.DataDescription;
@@ -71,6 +65,12 @@ import java.util.Optional;
import java.util.Set;
import java.util.UUID;
+import mockit.Expectations;
+import mockit.Injectable;
+import mockit.Mock;
+import mockit.MockUp;
+import mockit.Mocked;
+
public class BrokerLoadJobTest {
@BeforeClass
@@ -363,7 +363,7 @@ public class BrokerLoadJobTest {
RuntimeProfile jobProfile = new RuntimeProfile("test");
LoadLoadingTask task = new LoadLoadingTask(database, olapTable,brokerDesc, fileGroups,
100, 100, false, 100, callback, "",
- 100, 1, 1, jobProfile);
+ 100, 1, 1, true, jobProfile);
try {
UserIdentity userInfo = new UserIdentity("root", "localhost");
userInfo.setIsAnalyzed();
diff --git a/gensrc/thrift/DataSinks.thrift b/gensrc/thrift/DataSinks.thrift
index d41d77e..7320fa9 100644
--- a/gensrc/thrift/DataSinks.thrift
+++ b/gensrc/thrift/DataSinks.thrift
@@ -27,7 +27,7 @@ include "PlanNodes.thrift"
enum TDataSinkType {
DATA_STREAM_SINK,
RESULT_SINK,
- DATA_SPLIT_SINK,
+ DATA_SPLIT_SINK, // deprecated
MYSQL_TABLE_SINK,
EXPORT_SINK,
OLAP_TABLE_SINK,
diff --git a/gensrc/thrift/FrontendService.thrift b/gensrc/thrift/FrontendService.thrift
index 013ac0d..1a04b69 100644
--- a/gensrc/thrift/FrontendService.thrift
+++ b/gensrc/thrift/FrontendService.thrift
@@ -594,6 +594,7 @@ struct TStreamLoadPutRequest {
33: optional bool read_json_by_line
34: optional string auth_code_uuid
35: optional i32 send_batch_parallelism
+ 36: optional double max_filter_ratio
}
struct TStreamLoadPutResult {
diff --git a/gensrc/thrift/PaloInternalService.thrift b/gensrc/thrift/PaloInternalService.thrift
index 664dab0..713487f 100644
--- a/gensrc/thrift/PaloInternalService.thrift
+++ b/gensrc/thrift/PaloInternalService.thrift
@@ -249,6 +249,9 @@ struct TQueryGlobals {
// time_zone is the timezone this query used.
// If this value is set, BE will ignore now_string
3: optional string time_zone
+
+ // Set to true if in a load plan, the max_filter_ratio is 0.0
+ 4: optional bool load_zero_tolerance = false;
}
diff --git a/gensrc/thrift/PlanNodes.thrift b/gensrc/thrift/PlanNodes.thrift
index 2f28b32..9d143d0 100644
--- a/gensrc/thrift/PlanNodes.thrift
+++ b/gensrc/thrift/PlanNodes.thrift
@@ -26,7 +26,7 @@ include "Partitions.thrift"
enum TPlanNodeType {
OLAP_SCAN_NODE,
MYSQL_SCAN_NODE,
- CSV_SCAN_NODE,
+ CSV_SCAN_NODE, // deprecated
SCHEMA_SCAN_NODE,
HASH_JOIN_NODE,
MERGE_JOIN_NODE,
@@ -39,7 +39,7 @@ enum TPlanNodeType {
CROSS_JOIN_NODE,
META_SCAN_NODE,
ANALYTIC_EVAL_NODE,
- OLAP_REWRITE_NODE,
+ OLAP_REWRITE_NODE, // deprecated
KUDU_SCAN_NODE, // Deprecated
BROKER_SCAN_NODE,
EMPTY_SET_NODE,
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org