be/src/format/json/new_json_reader.h
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 | | #pragma once |
19 | | |
20 | | #include <rapidjson/allocators.h> |
21 | | #include <rapidjson/document.h> |
22 | | #include <rapidjson/encodings.h> |
23 | | #include <rapidjson/rapidjson.h> |
24 | | #include <simdjson/common_defs.h> |
25 | | #include <simdjson/simdjson.h> // IWYU pragma: keep |
26 | | |
27 | | #include <memory> |
28 | | #include <string> |
29 | | #include <string_view> |
30 | | #include <unordered_map> |
31 | | #include <unordered_set> |
32 | | #include <vector> |
33 | | |
34 | | #include "common/status.h" |
35 | | #include "core/custom_allocator.h" |
36 | | #include "core/string_ref.h" |
37 | | #include "core/types.h" |
38 | | #include "exprs/json_functions.h" |
39 | | #include "format/line_reader.h" |
40 | | #include "format/table/table_format_reader.h" |
41 | | #include "io/file_factory.h" |
42 | | #include "io/fs/file_reader_writer_fwd.h" |
43 | | #include "runtime/runtime_profile.h" |
44 | | #include "util/decompressor.h" |
45 | | |
46 | | namespace simdjson::fallback::ondemand { |
47 | | class object; |
48 | | } // namespace simdjson::fallback::ondemand |
49 | | |
50 | | namespace doris { |
51 | | class SlotDescriptor; |
52 | | class RuntimeState; |
53 | | class TFileRangeDesc; |
54 | | class TFileScanRangeParams; |
55 | | |
56 | | namespace io { |
57 | | class FileSystem; |
58 | | struct IOContext; |
59 | | } // namespace io |
60 | | |
61 | | struct ScannerCounter; |
62 | | class Block; |
63 | | class IColumn; |
64 | | |
65 | | /// JSON-specific initialization context. |
66 | | /// Extends ReaderInitContext with default value context (unique to JSON reader). |
67 | | struct JsonInitContext final : public ReaderInitContext { |
68 | | const std::unordered_map<std::string, VExprContextSPtr>* col_default_value_ctx = nullptr; |
69 | | bool is_load = false; |
70 | | }; |
71 | | |
72 | | class NewJsonReader : public TableFormatReader { |
73 | | ENABLE_FACTORY_CREATOR(NewJsonReader); |
74 | | |
75 | | public: |
76 | | NewJsonReader(RuntimeState* state, RuntimeProfile* profile, ScannerCounter* counter, |
77 | | const TFileScanRangeParams& params, const TFileRangeDesc& range, |
78 | | const std::vector<SlotDescriptor*>& file_slot_descs, bool* scanner_eof, |
79 | | size_t batch_size, io::IOContext* io_ctx, |
80 | | std::shared_ptr<io::IOContext> io_ctx_holder = nullptr); |
81 | | |
82 | | NewJsonReader(RuntimeProfile* profile, const TFileScanRangeParams& params, |
83 | | const TFileRangeDesc& range, const std::vector<SlotDescriptor*>& file_slot_descs, |
84 | | size_t batch_size, io::IOContext* io_ctx, |
85 | | std::shared_ptr<io::IOContext> io_ctx_holder = nullptr); |
86 | 2 | ~NewJsonReader() override = default; |
87 | | |
88 | | Status init_reader( |
89 | | const std::unordered_map<std::string, VExprContextSPtr>& col_default_value_ctx, |
90 | | bool is_load); |
91 | | |
92 | | Status _do_get_next_block(Block* block, size_t* read_rows, bool* eof) override; |
93 | | Status _get_columns_impl(std::unordered_map<std::string, DataTypePtr>* name_to_type) override; |
94 | | |
95 | | // Row-based readers control throughput via row count, not byte budget. |
96 | | // The FileScanner's AdaptiveBlockSizePredictor converts the byte budget |
97 | | // into a predicted row count and calls set_batch_size() with it. |
98 | | void set_batch_size(size_t batch_size) override; |
99 | 6 | size_t get_batch_size() const override { return _batch_size; } |
100 | | |
101 | | Status init_schema_reader() override; |
102 | | Status get_parsed_schema(std::vector<std::string>* col_names, |
103 | | std::vector<DataTypePtr>* col_types) override; |
104 | | |
105 | | protected: |
106 | | // ---- Unified init_reader(ReaderInitContext*) overrides ---- |
107 | | Status _open_file_reader(ReaderInitContext* ctx) override; |
108 | | Status _do_init_reader(ReaderInitContext* ctx) override; |
109 | | |
110 | | void _collect_profile_before_close() override; |
111 | | |
112 | | private: |
113 | | Status _get_range_params(); |
114 | | void _init_system_properties(); |
115 | | void _init_file_description(); |
116 | | Status _open_file_reader(bool need_schema); |
117 | | Status _open_line_reader(); |
118 | | Status _parse_jsonpath_and_json_root(); |
119 | | |
120 | | Status _read_json_column(RuntimeState* state, Block& block, |
121 | | const std::vector<SlotDescriptor*>& slot_descs, bool* is_empty_row, |
122 | | bool* eof); |
123 | | |
124 | | Status _read_one_message(DorisUniqueBufferPtr<uint8_t>* file_buf, size_t* read_size); |
125 | | |
126 | | // StreamLoadPipe::read_one_message only reads a portion of the data when stream loading with a chunked transfer HTTP request. |
127 | | // Need to read all the data before performing JSON parsing. |
128 | | Status _read_one_message_from_pipe(DorisUniqueBufferPtr<uint8_t>* file_buf, size_t* read_size); |
129 | | |
130 | | // simdjson, replace none simdjson function if it is ready |
131 | | Status _simdjson_init_reader(); |
132 | | Status _simdjson_parse_json(size_t* size, bool* is_empty_row, bool* eof, |
133 | | simdjson::error_code* error); |
134 | | Status _get_json_value(size_t* size, bool* eof, simdjson::error_code* error, |
135 | | bool* is_empty_row); |
136 | | Status _judge_empty_row(size_t size, bool eof, bool* is_empty_row); |
137 | | |
138 | | Status _handle_simdjson_error(simdjson::simdjson_error& error, Block& block, size_t num_rows, |
139 | | bool* eof); |
140 | | |
141 | | Status _simdjson_handle_simple_json(RuntimeState* state, Block& block, |
142 | | const std::vector<SlotDescriptor*>& slot_descs, |
143 | | bool* is_empty_row, bool* eof); |
144 | | |
145 | | Status _simdjson_handle_simple_json_write_columns( |
146 | | Block& block, const std::vector<SlotDescriptor*>& slot_descs, bool* is_empty_row, |
147 | | bool* eof); |
148 | | |
149 | | Status _simdjson_handle_flat_array_complex_json(RuntimeState* state, Block& block, |
150 | | const std::vector<SlotDescriptor*>& slot_descs, |
151 | | bool* is_empty_row, bool* eof); |
152 | | |
153 | | Status _simdjson_handle_flat_array_complex_json_write_columns( |
154 | | Block& block, const std::vector<SlotDescriptor*>& slot_descs, bool* is_empty_row, |
155 | | bool* eof); |
156 | | |
157 | | Status _simdjson_handle_nested_complex_json(RuntimeState* state, Block& block, |
158 | | const std::vector<SlotDescriptor*>& slot_descs, |
159 | | bool* is_empty_row, bool* eof); |
160 | | |
161 | | Status _simdjson_set_column_value(simdjson::ondemand::object* value, Block& block, |
162 | | const std::vector<SlotDescriptor*>& slot_descs, bool* valid); |
163 | | |
164 | | template <bool use_string_cache> |
165 | | Status _simdjson_write_data_to_column(simdjson::ondemand::value& value, |
166 | | const DataTypePtr& type_desc, IColumn* column_ptr, |
167 | | const std::string& column_name, DataTypeSerDeSPtr serde, |
168 | | bool* valid); |
169 | | |
170 | | Status _simdjson_write_columns_by_jsonpath(simdjson::ondemand::object* value, |
171 | | const std::vector<SlotDescriptor*>& slot_descs, |
172 | | Block& block, bool* valid); |
173 | | Status _append_error_msg(simdjson::ondemand::object* obj, std::string error_msg, |
174 | | std::string col_name, bool* valid); |
175 | | |
176 | | size_t _column_index(const StringRef& name, size_t key_index); |
177 | | |
178 | | Status (NewJsonReader::*_vhandle_json_callback)(RuntimeState* state, Block& block, |
179 | | const std::vector<SlotDescriptor*>& slot_descs, |
180 | | bool* is_empty_row, bool* eof); |
181 | | Status _get_column_default_value( |
182 | | const std::vector<SlotDescriptor*>& slot_descs, |
183 | | const std::unordered_map<std::string, VExprContextSPtr>& col_default_value_ctx); |
184 | | |
185 | | Status _fill_missing_column(SlotDescriptor* slot_desc, DataTypeSerDeSPtr serde, |
186 | | IColumn* column_ptr, bool* valid); |
187 | | |
188 | | // fe will add skip_bitmap_col to _file_slot_descs iff the target olap table has skip_bitmap_col |
189 | | // and the current load is a flexible partial update |
190 | | // flexible partial update can not be used when user specify jsonpaths, so we just fill the skip bitmap |
191 | | // in `_simdjson_handle_simple_json` and `_vhandle_simple_json` (which will be used when jsonpaths is not specified) |
192 | 0 | bool _should_process_skip_bitmap_col() const { return skip_bitmap_col_idx != -1; } |
193 | | void _append_empty_skip_bitmap_value(Block& block, size_t cur_row_count); |
194 | | void _set_skip_bitmap_mark(SlotDescriptor* slot_desc, IColumn* column_ptr, Block& block, |
195 | | size_t cur_row_count, bool* valid); |
196 | | RuntimeState* _state = nullptr; |
197 | | RuntimeProfile* _profile = nullptr; |
198 | | ScannerCounter* _counter = nullptr; |
199 | | const TFileScanRangeParams& _params; |
200 | | const TFileRangeDesc& _range; |
201 | | io::FileSystemProperties _system_properties; |
202 | | io::FileDescription _file_description; |
203 | | const std::vector<SlotDescriptor*>& _file_slot_descs; |
204 | | |
205 | | io::FileReaderSPtr _file_reader; |
206 | | std::unique_ptr<LineReader> _line_reader; |
207 | | bool _reader_eof; |
208 | | std::unique_ptr<Decompressor> _decompressor; |
209 | | TFileCompressType::type _file_compress_type; |
210 | | |
211 | | // When we fetch range doesn't start from 0 will always skip the first line |
212 | | bool _skip_first_line; |
213 | | |
214 | | std::string _line_delimiter; |
215 | | size_t _line_delimiter_length; |
216 | | |
217 | | uint32_t _next_row; |
218 | | size_t _total_rows; |
219 | | |
220 | | std::string _jsonpaths; |
221 | | std::string _json_root; |
222 | | bool _read_json_by_line; |
223 | | bool _strip_outer_array; |
224 | | bool _num_as_string; |
225 | | bool _fuzzy_parse; |
226 | | |
227 | | std::vector<std::vector<JsonPath>> _parsed_jsonpaths; |
228 | | std::vector<JsonPath> _parsed_json_root; |
229 | | bool _parsed_from_json_root = false; // to avoid parsing json root multiple times |
230 | | |
231 | | char _value_buffer[4 * 1024 * 1024]; // 4MB |
232 | | char _parse_buffer[512 * 1024]; // 512KB |
233 | | |
234 | | using Document = rapidjson::GenericDocument<rapidjson::UTF8<>, rapidjson::MemoryPoolAllocator<>, |
235 | | rapidjson::MemoryPoolAllocator<>>; |
236 | | rapidjson::MemoryPoolAllocator<> _value_allocator; |
237 | | rapidjson::MemoryPoolAllocator<> _parse_allocator; |
238 | | Document _origin_json_doc; // origin json document object from parsed json string |
239 | | rapidjson::Value* _json_doc; // _json_doc equals _final_json_doc iff not set `json_root` |
240 | | std::unordered_map<std::string, int> _name_map; |
241 | | |
242 | | bool* _scanner_eof = nullptr; |
243 | | |
244 | | size_t _current_offset; |
245 | | |
246 | | io::IOContext* _io_ctx = nullptr; |
247 | | std::shared_ptr<io::IOContext> _io_ctx_holder; |
248 | | |
249 | | RuntimeProfile::Counter* _read_timer = nullptr; |
250 | | |
251 | | // ======SIMD JSON====== |
252 | | // name mapping |
253 | | /// Hash table match `field name -> position in the block`. NOTE You can use perfect hash map. |
254 | | using NameMap = phmap::flat_hash_map<StringRef, size_t, StringRefHash>; |
255 | | NameMap _slot_desc_index; |
256 | | /// Cached search results for previous row (keyed as index in JSON object) - used as a hint. |
257 | | std::vector<NameMap::iterator> _prev_positions; |
258 | | /// Set of columns which already met in row. Exception is thrown if there are more than one column with the same name. |
259 | | std::vector<UInt8> _seen_columns; |
260 | | // simdjson |
261 | | DorisUniqueBufferPtr<uint8_t> _json_str_ptr; |
262 | | const uint8_t* _json_str = nullptr; |
263 | | static constexpr size_t _init_buffer_size = 1024 * 1024 * 8; |
264 | | size_t _padded_size = _init_buffer_size + simdjson::SIMDJSON_PADDING; |
265 | | size_t _original_doc_size = 0; |
266 | | std::string _simdjson_ondemand_padding_buffer; |
267 | | std::string _simdjson_ondemand_unscape_padding_buffer; |
268 | | // char _simdjson_ondemand_padding_buffer[_padded_size]; |
269 | | simdjson::ondemand::document _original_json_doc; |
270 | | simdjson::ondemand::value _json_value; |
271 | | // for strip outer array |
272 | | // array_iter pointed to _array |
273 | | simdjson::ondemand::array_iterator _array_iter; |
274 | | simdjson::ondemand::array _array; |
275 | | std::unique_ptr<simdjson::ondemand::parser> _ondemand_json_parser; |
276 | | // column to default value string map |
277 | | std::unordered_map<std::string, std::string> _col_default_value_map; |
278 | | |
279 | | // From document of simdjson: |
280 | | // ``` |
281 | | // Important: a value should be consumed once. Calling get_string() twice on the same value is an error. |
282 | | // ``` |
283 | | // We should cache the string_views to avoid multiple get_string() calls. |
284 | | struct StringViewHash { |
285 | 0 | size_t operator()(const std::string_view& str) const { |
286 | 0 | return std::hash<int64_t>()(reinterpret_cast<int64_t>(str.data())); |
287 | 0 | } |
288 | | }; |
289 | | struct StringViewEqual { |
290 | 0 | bool operator()(const std::string_view& lhs, const std::string_view& rhs) const { |
291 | 0 | return lhs.data() == rhs.data() && lhs.size() == rhs.size(); |
292 | 0 | } |
293 | | }; |
294 | | std::unordered_map<std::string_view, std::string_view, StringViewHash, StringViewEqual> |
295 | | _cached_string_values; |
296 | | |
297 | | int32_t skip_bitmap_col_idx {-1}; |
298 | | |
299 | | //Used to indicate whether it is a stream load. When loading, only data will be inserted into columnString. |
300 | | //If an illegal value is encountered during the load process, `_append_error_msg` should be called |
301 | | //instead of directly returning `Status::DataQualityError` |
302 | | bool _is_load = true; |
303 | | |
304 | | // In hive : create table xxx ROW FORMAT SERDE 'org.apache.hive.hcatalog.data.JsonSerDe'; |
305 | | // Hive will not allow you to create columns with the same name but different case, including field names inside |
306 | | // structs, and will automatically convert uppercase names in create sql to lowercase.However, when Hive loads data |
307 | | // to table, the column names in the data may be uppercase,and there may be multiple columns with |
308 | | // the same name but different capitalization.We refer to the behavior of hive, convert all column names |
309 | | // in the data to lowercase,and use the last one as the insertion value |
310 | | bool _is_hive_table = false; |
311 | | |
312 | | // hive : org.openx.data.jsonserde.JsonSerDe, `ignore.malformed.json` prop. |
313 | | // If the variable is true, `null` will be inserted for llegal json format instead of return error. |
314 | | bool _openx_json_ignore_malformed = false; |
315 | | |
316 | | DataTypeSerDeSPtrs _serdes; |
317 | | DataTypeSerDe::FormatOptions _serde_options; |
318 | | // Adaptive batch size set by FileScanner. |
319 | | size_t _batch_size; |
320 | | }; |
321 | | |
322 | | } // namespace doris |