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 <vector> |
26 | | |
27 | | #include "common/compiler_util.h" // IWYU pragma: keep |
28 | | #include "common/status.h" |
29 | | #include "core/block/block.h" |
30 | | #include "exec/scan/scanner.h" |
31 | | #include "format/csv/csv_reader.h" |
32 | | #include "format/file_reader/new_plain_text_line_reader.h" |
33 | | #include "format/line_reader.h" |
34 | | #include "io/file_factory.h" |
35 | | #include "io/fs/buffered_reader.h" |
36 | | #include "io/fs/file_reader.h" |
37 | | #include "io/fs/s3_file_reader.h" |
38 | | #include "runtime/descriptors.h" |
39 | | #include "runtime/runtime_state.h" |
40 | | |
41 | | namespace doris { |
42 | | #include "common/compile_check_begin.h" |
43 | | |
44 | 24 | void HiveTextFieldSplitter::do_split(const Slice& line, std::vector<Slice>* splitted_values) { |
45 | 24 | if (_value_sep_len == 1) { |
46 | 7 | _split_field_single_char(line, splitted_values); |
47 | 17 | } else { |
48 | 17 | _split_field_multi_char(line, splitted_values); |
49 | 17 | } |
50 | 24 | } |
51 | | |
52 | | void HiveTextFieldSplitter::_split_field_single_char(const Slice& line, |
53 | 7 | std::vector<Slice>* splitted_values) { |
54 | 7 | const char* data = line.data; |
55 | 7 | const size_t size = line.size; |
56 | 7 | size_t value_start = 0; |
57 | 30 | for (size_t i = 0; i < size; ++i) { |
58 | 23 | if (data[i] == _value_sep[0]) { |
59 | | // hive will escape the field separator in string |
60 | 10 | if (_escape_char != 0 && i > 0 && data[i - 1] == _escape_char) { |
61 | 1 | continue; |
62 | 1 | } |
63 | 9 | process_value_func(data, value_start, i - value_start, _trimming_char, splitted_values); |
64 | 9 | value_start = i + _value_sep_len; |
65 | 9 | } |
66 | 23 | } |
67 | 7 | process_value_func(data, value_start, size - value_start, _trimming_char, splitted_values); |
68 | 7 | } |
69 | | |
70 | | void HiveTextFieldSplitter::_split_field_multi_char(const Slice& line, |
71 | 17 | std::vector<Slice>* splitted_values) { |
72 | 17 | const char* data = line.data; |
73 | 17 | const size_t size = line.size; |
74 | 17 | size_t start = 0; |
75 | | |
76 | 17 | std::vector<int> next(_value_sep_len); |
77 | 17 | next[0] = -1; |
78 | 50 | for (int i = 1, j = -1; i < (int)_value_sep_len; i++) { |
79 | 33 | while (j >= 0 && _value_sep[i] != _value_sep[j + 1]) { |
80 | 0 | j = next[j]; |
81 | 0 | } |
82 | 33 | if (_value_sep[i] == _value_sep[j + 1]) { |
83 | 22 | j++; |
84 | 22 | } |
85 | 33 | next[i] = j; |
86 | 33 | } |
87 | | |
88 | | // KMP search |
89 | 198 | for (int i = 0, j = -1; i < (int)size; i++) { |
90 | 200 | while (j >= 0 && data[i] != _value_sep[j + 1]) { |
91 | 19 | j = next[j]; |
92 | 19 | } |
93 | 181 | if (data[i] == _value_sep[j + 1]) { |
94 | 90 | j++; |
95 | 90 | } |
96 | 181 | if (j == (int)_value_sep_len - 1) { |
97 | 34 | size_t curpos = i - _value_sep_len + 1; |
98 | 34 | if (_escape_char != 0 && curpos > 0 && data[curpos - 1] == _escape_char) { |
99 | 2 | j = next[j]; |
100 | 2 | continue; |
101 | 2 | } |
102 | | |
103 | 32 | if (curpos >= start) { |
104 | 25 | process_value_func(data, start, curpos - start, _trimming_char, splitted_values); |
105 | 25 | start = curpos + _value_sep_len; |
106 | 25 | } |
107 | | |
108 | 32 | j = next[j]; |
109 | 32 | } |
110 | 181 | } |
111 | 17 | process_value_func(data, start, size - start, _trimming_char, splitted_values); |
112 | 17 | } |
113 | | |
114 | | TextReader::TextReader(RuntimeState* state, RuntimeProfile* profile, ScannerCounter* counter, |
115 | | const TFileScanRangeParams& params, const TFileRangeDesc& range, |
116 | | const std::vector<SlotDescriptor*>& file_slot_descs, io::IOContext* io_ctx) |
117 | 0 | : CsvReader(state, profile, counter, params, range, file_slot_descs, io_ctx) {} |
118 | | |
119 | 0 | Status TextReader::_init_options() { |
120 | | // get column_separator and line_delimiter |
121 | 0 | _value_separator = _params.file_attributes.text_params.column_separator; |
122 | 0 | _value_separator_length = _value_separator.size(); |
123 | 0 | _line_delimiter = _params.file_attributes.text_params.line_delimiter; |
124 | 0 | _line_delimiter_length = _line_delimiter.size(); |
125 | |
|
126 | 0 | if (_params.file_attributes.text_params.__isset.escape) { |
127 | 0 | _escape = _params.file_attributes.text_params.escape; |
128 | 0 | } |
129 | 0 | _options.escape_char = _escape; |
130 | |
|
131 | 0 | _options.collection_delim = _params.file_attributes.text_params.collection_delimiter[0]; |
132 | 0 | _options.map_key_delim = _params.file_attributes.text_params.mapkv_delimiter[0]; |
133 | |
|
134 | 0 | _options.null_format = _params.file_attributes.text_params.null_format.data(); |
135 | 0 | _options.null_len = _params.file_attributes.text_params.null_format.length(); |
136 | |
|
137 | 0 | return Status::OK(); |
138 | 0 | } |
139 | | |
140 | 0 | Status TextReader::_deserialize_one_cell(DataTypeSerDeSPtr serde, IColumn& column, Slice& slice) { |
141 | 0 | return serde->deserialize_one_cell_from_hive_text(column, slice, _options); |
142 | 0 | } |
143 | | |
144 | 0 | Status TextReader::_create_line_reader() { |
145 | 0 | std::shared_ptr<TextLineReaderContextIf> text_line_reader_ctx; |
146 | |
|
147 | 0 | text_line_reader_ctx = std::make_shared<PlainTextLineReaderCtx>(_line_delimiter, |
148 | 0 | _line_delimiter_length, false); |
149 | |
|
150 | 0 | _fields_splitter = std::make_unique<HiveTextFieldSplitter>( |
151 | 0 | false, false, _value_separator, _value_separator_length, -1, _escape); |
152 | |
|
153 | 0 | _line_reader = |
154 | 0 | NewPlainTextLineReader::create_unique(_profile, _file_reader, _decompressor.get(), |
155 | 0 | text_line_reader_ctx, _size, _start_offset); |
156 | |
|
157 | 0 | return Status::OK(); |
158 | 0 | } |
159 | | |
160 | 0 | Status TextReader::_validate_line(const Slice& line, bool* success) { |
161 | | // text file do not need utf8 check |
162 | 0 | *success = true; |
163 | 0 | return Status::OK(); |
164 | 0 | } |
165 | | |
166 | 0 | Status TextReader::_deserialize_nullable_string(IColumn& column, Slice& slice) { |
167 | 0 | auto& null_column = assert_cast<ColumnNullable&>(column); |
168 | 0 | if (slice.compare(Slice(_options.null_format, _options.null_len)) == 0) { |
169 | 0 | null_column.insert_data(nullptr, 0); |
170 | 0 | return Status::OK(); |
171 | 0 | } |
172 | 0 | static DataTypeStringSerDe stringSerDe(TYPE_STRING); |
173 | 0 | auto st = stringSerDe.deserialize_one_cell_from_hive_text(null_column.get_nested_column(), |
174 | 0 | slice, _options); |
175 | 0 | if (!st.ok()) { |
176 | | // fill null if fail |
177 | 0 | null_column.insert_data(nullptr, 0); // 0 is meaningless here |
178 | 0 | return Status::OK(); |
179 | 0 | } |
180 | | // fill not null if success |
181 | 0 | null_column.get_null_map_data().push_back(0); |
182 | 0 | return Status::OK(); |
183 | 0 | } |
184 | | |
185 | | #include "common/compile_check_end.h" |
186 | | } // namespace doris |