Coverage Report

Created: 2026-07-12 18:43

next uncovered line (L), next uncovered region (R), next uncovered branch (B)
be/src/format/text/text_reader.cpp
Line
Count
Source
1
// Licensed to the Apache Software Foundation (ASF) under one
2
// or more contributor license agreements.  See the NOTICE file
3
// distributed with this work for additional information
4
// regarding copyright ownership.  The ASF licenses this file
5
// to you under the Apache License, Version 2.0 (the
6
// "License"); you may not use this file except in compliance
7
// with the License.  You may obtain a copy of the License at
8
//
9
//   http://www.apache.org/licenses/LICENSE-2.0
10
//
11
// Unless required by applicable law or agreed to in writing,
12
// software distributed under the License is distributed on an
13
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14
// KIND, either express or implied.  See the License for the
15
// specific language governing permissions and limitations
16
// under the License.
17
18
#include "format/text/text_reader.h"
19
20
#include <gen_cpp/PlanNodes_types.h>
21
#include <gen_cpp/Types_types.h>
22
#include <glog/logging.h>
23
24
#include <cstddef>
25
#include <utility>
26
#include <vector>
27
28
#include "common/compiler_util.h" // IWYU pragma: keep
29
#include "common/status.h"
30
#include "core/block/block.h"
31
#include "core/column/column_nullable.h"
32
#include "core/column/column_string.h"
33
#include "core/data_type_serde/data_type_string_serde.h"
34
#include "exec/scan/scanner.h"
35
#include "format/csv/csv_reader.h"
36
#include "format/file_reader/new_plain_text_line_reader.h"
37
#include "format/hive_text_util.h"
38
#include "format/line_reader.h"
39
#include "io/file_factory.h"
40
#include "io/fs/buffered_reader.h"
41
#include "io/fs/file_reader.h"
42
#include "io/fs/s3_file_reader.h"
43
#include "runtime/descriptors.h"
44
#include "runtime/runtime_state.h"
45
46
namespace doris {
47
48
39
void HiveTextFieldSplitter::do_split(const Slice& line, std::vector<Slice>* splitted_values) {
49
39
    if (_value_sep_len == 1) {
50
21
        _split_field_single_char(line, splitted_values);
51
21
    } else {
52
18
        _split_field_multi_char(line, splitted_values);
53
18
    }
54
39
}
55
56
void HiveTextFieldSplitter::_split_field_single_char(const Slice& line,
57
21
                                                     std::vector<Slice>* splitted_values) {
58
21
    const char* data = line.data;
59
21
    const size_t size = line.size;
60
21
    size_t value_start = 0;
61
1.01k
    for (size_t i = 0; i < size; ++i) {
62
990
        if (data[i] == _value_sep[0]) {
63
            // hive will escape the field separator in string
64
136
            if (_escape_char != 0 && is_hive_text_separator_escaped(data, i, _escape_char)) {
65
1
                continue;
66
1
            }
67
135
            process_value_func(data, value_start, i - value_start, _trimming_char, splitted_values);
68
135
            value_start = i + _value_sep_len;
69
135
        }
70
990
    }
71
21
    process_value_func(data, value_start, size - value_start, _trimming_char, splitted_values);
72
21
}
73
74
void HiveTextFieldSplitter::_split_field_multi_char(const Slice& line,
75
18
                                                    std::vector<Slice>* splitted_values) {
76
18
    const char* data = line.data;
77
18
    const size_t size = line.size;
78
18
    size_t start = 0;
79
80
18
    std::vector<int> next(_value_sep_len);
81
18
    next[0] = -1;
82
52
    for (int i = 1, j = -1; i < (int)_value_sep_len; i++) {
83
34
        while (j >= 0 && _value_sep[i] != _value_sep[j + 1]) {
84
0
            j = next[j];
85
0
        }
86
34
        if (_value_sep[i] == _value_sep[j + 1]) {
87
23
            j++;
88
23
        }
89
34
        next[i] = j;
90
34
    }
91
92
    // KMP search
93
205
    for (int i = 0, j = -1; i < (int)size; i++) {
94
207
        while (j >= 0 && data[i] != _value_sep[j + 1]) {
95
20
            j = next[j];
96
20
        }
97
187
        if (data[i] == _value_sep[j + 1]) {
98
92
            j++;
99
92
        }
100
187
        if (j == (int)_value_sep_len - 1) {
101
35
            size_t curpos = i - _value_sep_len + 1;
102
35
            if (_escape_char != 0 && is_hive_text_separator_escaped(data, curpos, _escape_char)) {
103
2
                j = next[j];
104
2
                continue;
105
2
            }
106
107
33
            if (curpos >= start) {
108
26
                process_value_func(data, start, curpos - start, _trimming_char, splitted_values);
109
26
                start = curpos + _value_sep_len;
110
26
            }
111
112
33
            j = next[j];
113
33
        }
114
187
    }
115
18
    process_value_func(data, start, size - start, _trimming_char, splitted_values);
116
18
}
117
118
TextReader::TextReader(RuntimeState* state, RuntimeProfile* profile, ScannerCounter* counter,
119
                       const TFileScanRangeParams& params, const TFileRangeDesc& range,
120
                       const std::vector<SlotDescriptor*>& file_slot_descs, size_t batch_size,
121
                       io::IOContext* io_ctx, std::shared_ptr<io::IOContext> io_ctx_holder)
122
7
        : CsvReader(state, profile, counter, params, range, file_slot_descs, batch_size, io_ctx,
123
7
                    std::move(io_ctx_holder)) {}
124
125
7
Status TextReader::_init_options() {
126
    // get column_separator and line_delimiter
127
7
    _value_separator = _params.file_attributes.text_params.column_separator;
128
7
    _value_separator_length = _value_separator.size();
129
7
    _line_delimiter = _params.file_attributes.text_params.line_delimiter;
130
7
    _line_delimiter_length = _line_delimiter.size();
131
132
7
    if (_params.file_attributes.text_params.__isset.escape) {
133
0
        _escape = _params.file_attributes.text_params.escape;
134
0
    }
135
7
    _options.escape_char = _escape;
136
137
7
    _options.collection_delim = _params.file_attributes.text_params.collection_delimiter[0];
138
7
    _options.map_key_delim = _params.file_attributes.text_params.mapkv_delimiter[0];
139
140
7
    _options.null_format = _params.file_attributes.text_params.null_format.data();
141
7
    _options.null_len = _params.file_attributes.text_params.null_format.length();
142
143
7
    return Status::OK();
144
7
}
145
146
0
Status TextReader::_deserialize_one_cell(DataTypeSerDeSPtr serde, IColumn& column, Slice& slice) {
147
0
    return serde->deserialize_one_cell_from_hive_text(column, slice, _options);
148
0
}
149
150
7
Status TextReader::_create_line_reader() {
151
7
    std::shared_ptr<TextLineReaderContextIf> text_line_reader_ctx;
152
153
7
    text_line_reader_ctx = std::make_shared<PlainTextLineReaderCtx>(_line_delimiter,
154
7
                                                                    _line_delimiter_length, false);
155
156
7
    _fields_splitter = std::make_unique<HiveTextFieldSplitter>(
157
7
            false, false, _value_separator, _value_separator_length, -1, _escape);
158
159
7
    _line_reader =
160
7
            NewPlainTextLineReader::create_unique(_profile, _file_reader, _decompressor.get(),
161
7
                                                  text_line_reader_ctx, _size, _start_offset);
162
163
7
    return Status::OK();
164
7
}
165
166
9
Status TextReader::_validate_line(const Slice& line, bool* success) {
167
    // text file do not need utf8 check
168
9
    *success = true;
169
9
    return Status::OK();
170
9
}
171
172
0
bool TextReader::_empty_line_as_record() const {
173
    // Hive TEXTFILE treats an empty physical line as a record. The splitter maps it
174
    // to one empty field and missing trailing fields are filled with null_format.
175
0
    return true;
176
0
}
177
178
126
Status TextReader::_deserialize_nullable_string(IColumn& column, Slice& slice) {
179
    // Hot path of hive text load, see CsvReader::_deserialize_nullable_string. The
180
    // column type was verified by the checked assert_cast in
181
    // _reserve_nullable_string_columns at the beginning of the batch.
182
126
    auto& null_column = assert_cast<ColumnNullable&, TypeCheckOnRelease::DISABLE>(column);
183
126
    auto& string_column = assert_cast<ColumnString&, TypeCheckOnRelease::DISABLE>(
184
126
            null_column.get_nested_column());
185
126
    if (slice.compare(Slice(_options.null_format, _options.null_len)) == 0) {
186
0
        string_column.insert_default();
187
0
        null_column.get_null_map_data().push_back(1);
188
0
        return Status::OK();
189
0
    }
190
    // Same as DataTypeStringSerDe::deserialize_one_cell_from_hive_text (which never
191
    // fails), written out here to skip the SerDe layer and its per-cell assert_cast.
192
126
    if (_options.escape_char != 0) {
193
0
        escape_string(slice.data, &slice.size, _options.escape_char);
194
0
    }
195
126
    string_column.insert_data(slice.data, slice.size);
196
126
    null_column.get_null_map_data().push_back(0);
197
126
    return Status::OK();
198
126
}
199
200
} // namespace doris