Skip to content

Commit fa8ed2b

Browse files
carlvinhust2012carlvinhust2012
authored
[fix](array-type) fix the invalid format load for stream load (#12424)
this pr is used to fix the invalid format load for stream load. before the change , we will get the error when we load the invalid array format. the origin file to load : 1 [1, 2, 3] 2 [4, 5, 6] 3 \N 4 [7, \N, 8] 5 10, 11, 12 [hugo@xafj-palo]$ sh curl_cmd.sh { "TxnId": 11035, "Label": "11c9f111-188e-4616-9a50-aec8b7814513", "TwoPhaseCommit": "false", "Status": "Fail", "Message": "Array does not start with '[' character, found '1'", "NumberTotalRows": 0, "NumberLoadedRows": 0, "NumberFilteredRows": 0, "NumberUnselectedRows": 0, "LoadBytes": 55, "LoadTimeMs": 7, "BeginTxnTimeMs": 0, "StreamLoadPutTimeMs": 2, "ReadDataTimeMs": 0, "WriteDataTimeMs": 3, "CommitAndPublishTimeMs": 0 } 3. after this change, we will get success and the error url which report the error line. [hugo@xafj-palo]$ sh curl_cmd.sh { "TxnId": 11046, "Label": "249808ee-55f4-4c08-b671-b3d82689d614", "TwoPhaseCommit": "false", "Status": "Success", "Message": "OK", "NumberTotalRows": 5, "NumberLoadedRows": 4, "NumberFilteredRows": 1, "NumberUnselectedRows": 0, "LoadBytes": 55, "LoadTimeMs": 39, "BeginTxnTimeMs": 0, "StreamLoadPutTimeMs": 2, "ReadDataTimeMs": 0, "WriteDataTimeMs": 19, "CommitAndPublishTimeMs": 16, "ErrorURL": "http://10.81.85.89:8502/api/_load_error_log?file=__shard_3/error_log_insert_stmt_8d4130f0c18aeb0a-ad7ffd4233c41893_8d4130f0c18aeb0a_ad7ffd4233c41893" } the sql select result: MySQL [example_db]> select * from array_test06; +------+--------------+ | k1 | k2 | +------+--------------+ | 1 | [1, 2, 3] | | 2 | [4, 5, 6] | | 3 | NULL | | 4 | [7, NULL, 8] | +------+--------------+ 4 rows in set (0.019 sec) the url page show us: "Reason: Invalid format for array column(k2). src line [10, 11, 12]; " Issue Number: #7570
1 parent 65cff8d commit fa8ed2b

7 files changed

Lines changed: 53 additions & 10 deletions

File tree

‎be/src/exec/base_scanner.cpp‎

Lines changed: 34 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -487,4 +487,38 @@ void BaseScanner::_fill_columns_from_path() {
487487
}
488488
}
489489

490+
bool BaseScanner::is_null(const Slice& slice) {
491+
return slice.size == 2 && slice.data[0] == '\\' && slice.data[1] == 'N';
492+
}
493+
494+
bool BaseScanner::is_array(const Slice& slice) {
495+
return slice.size > 1 && slice.data[0] == '[' && slice.data[slice.size - 1] == ']';
496+
}
497+
498+
bool BaseScanner::check_array_format(std::vector<Slice>& split_values) {
499+
// if not the array format, filter this line and return error url
500+
auto dest_slot_descs = _dest_tuple_desc->slots();
501+
for (int j = 0; j < split_values.size() && j < dest_slot_descs.size(); ++j) {
502+
auto dest_slot_desc = dest_slot_descs[j];
503+
if (!dest_slot_desc->is_materialized()) {
504+
continue;
505+
}
506+
const Slice& value = split_values[j];
507+
if (dest_slot_desc->type().is_array_type() && !is_null(value) && !is_array(value)) {
508+
RETURN_IF_ERROR(_state->append_error_msg_to_file(
509+
[&]() -> std::string { return std::string(value.data, value.size); },
510+
[&]() -> std::string {
511+
fmt::memory_buffer err_msg;
512+
fmt::format_to(err_msg, "Invalid format for array column({})",
513+
dest_slot_desc->col_name());
514+
return fmt::to_string(err_msg);
515+
},
516+
&_scanner_eof));
517+
_counter->num_rows_filtered++;
518+
return false;
519+
}
520+
}
521+
return true;
522+
}
523+
490524
} // namespace doris

‎be/src/exec/base_scanner.h‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -92,6 +92,10 @@ class BaseScanner {
9292
Status _fill_dest_block(vectorized::Block* dest_block, bool* eof);
9393
virtual Status _init_src_block();
9494

95+
bool is_null(const Slice& slice);
96+
bool is_array(const Slice& slice);
97+
bool check_array_format(std::vector<Slice>& split_values);
98+
9599
RuntimeState* _state;
96100
const TBrokerScanRangeParams& _params;
97101

‎be/src/exec/broker_scanner.cpp‎

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -389,10 +389,6 @@ bool BrokerScanner::check_decimal_input(const Slice& slice, int precision, int s
389389
return true;
390390
}
391391

392-
bool is_null(const Slice& slice) {
393-
return slice.size == 2 && slice.data[0] == '\\' && slice.data[1] == 'N';
394-
}
395-
396392
// Convert one row to this tuple
397393
Status BrokerScanner::_convert_one_row(const Slice& line, Tuple* tuple, MemPool* tuple_pool,
398394
bool* fill_tuple) {
@@ -494,6 +490,10 @@ Status BrokerScanner::_line_to_src_tuple(const Slice& line) {
494490
return Status::OK();
495491
}
496492

493+
if (!check_array_format(_split_values)) {
494+
return Status::OK();
495+
}
496+
497497
for (int i = 0; i < _split_values.size(); ++i) {
498498
auto slot_desc = _src_slot_descs[i];
499499
const Slice& value = _split_values[i];

‎be/src/runtime/types.h‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -185,6 +185,8 @@ struct TypeDescriptor {
185185

186186
bool is_collection_type() const { return type == TYPE_ARRAY || type == TYPE_MAP; }
187187

188+
bool is_array_type() const { return type == TYPE_ARRAY; }
189+
188190
/// Returns the byte size of this type. Returns 0 for variable length types.
189191
int get_byte_size() const { return ::doris::get_byte_size(type); }
190192

‎be/src/vec/exec/vbroker_scanner.cpp‎

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -30,10 +30,6 @@
3030

3131
namespace doris::vectorized {
3232

33-
bool is_null(const Slice& slice) {
34-
return slice.size == 2 && slice.data[0] == '\\' && slice.data[1] == 'N';
35-
}
36-
3733
VBrokerScanner::VBrokerScanner(RuntimeState* state, RuntimeProfile* profile,
3834
const TBrokerScanRangeParams& params,
3935
const std::vector<TBrokerRangeDesc>& ranges,
@@ -95,6 +91,10 @@ Status VBrokerScanner::_fill_dest_columns(const Slice& line,
9591
return Status::OK();
9692
}
9793

94+
if (!check_array_format(_split_values)) {
95+
return Status::OK();
96+
}
97+
9898
int idx = 0;
9999
for (int i = 0; i < _split_values.size(); ++i) {
100100
int dest_index = idx++;

‎regression-test/data/load_p0/broker_load/simple_array.data‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,3 +1,5 @@
11
1/[1,2,3,4,5]/[32767,32768,32769]/[65534,65535,65536]/["a","b","c","d","e"]/["hello","world"]/["1991-01-01"]/["1991-01-01 00:00:00"]/[0.33,0.67]/[3.1415926,0.878787878]/[1,1.2,1.3]
22
2/[1,2,3,4,5]/[32767,32768,32769]/[65534,65535,65536]/["a","b","c","d","e"]/["hello","world"]/\N/\N/\N/\N/[1,\N,1.3]
3-
3/\N/\N/\N/\N/\N/\N/\N/\N/\N/\N
3+
3/\N/\N/\N/\N/\N/\N/\N/\N/\N/\N
4+
4/1,2,3,4,5/\N/\N/\N/\N/\N/\N/\N/\N/\N
5+
5/[1,2,3,4,5/\N/\N/\N/\N/\N/\N/\N/\N/\N

‎regression-test/suites/load_p0/broker_load/test_array_load.groovy‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -124,6 +124,7 @@ suite("test_array_load", "p0") {
124124
set 'where', where_expr
125125
set 'fuzzy_parse', fuzzy_flag
126126
set 'column_separator', column_sep
127+
set 'max_filter_ratio', '0.6'
127128
file file_name // import json file
128129
time 10000 // limit inflight 10s
129130

@@ -136,7 +137,7 @@ suite("test_array_load", "p0") {
136137
log.info("Stream load result: ${result}".toString())
137138
def json = parseJson(result)
138139
assertEquals("success", json.Status.toLowerCase())
139-
assertEquals(json.NumberTotalRows, json.NumberLoadedRows + json.NumberUnselectedRows)
140+
assertEquals(json.NumberTotalRows, json.NumberLoadedRows + json.NumberUnselectedRows + json.NumberFilteredRows)
140141
assertTrue(json.NumberLoadedRows > 0 && json.LoadBytes > 0)
141142
}
142143
}

0 commit comments

Comments
 (0)