be/src/format_v2/json/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 <simdjson/simdjson.h> // IWYU pragma: keep |
21 | | |
22 | | #include <memory> |
23 | | #include <optional> |
24 | | #include <string> |
25 | | #include <string_view> |
26 | | #include <unordered_map> |
27 | | #include <vector> |
28 | | |
29 | | #include "core/custom_allocator.h" |
30 | | #include "core/data_type_serde/data_type_serde.h" |
31 | | #include "exprs/json_functions.h" |
32 | | #include "format_v2/file_reader.h" |
33 | | #include "gen_cpp/PlanNodes_types.h" |
34 | | #include "runtime/runtime_profile.h" |
35 | | |
36 | | namespace doris { |
37 | | class Decompressor; |
38 | | class LineReader; |
39 | | class SlotDescriptor; |
40 | | class IColumn; |
41 | | } // namespace doris |
42 | | |
43 | | namespace doris::format::json { |
44 | | |
45 | | struct TransparentStringHash { |
46 | | using is_transparent = void; |
47 | 80 | size_t operator()(std::string_view value) const { |
48 | 80 | return std::hash<std::string_view> {}(value); |
49 | 80 | } |
50 | | }; |
51 | | |
52 | | struct CaseInsensitiveStringHash { |
53 | | using is_transparent = void; |
54 | 8 | size_t operator()(std::string_view value) const { |
55 | 8 | size_t hash = 1469598103934665603ULL; |
56 | 24 | for (const unsigned char c : value) { |
57 | 24 | const unsigned char folded = c >= 'A' && c <= 'Z' ? c + ('a' - 'A') : c; |
58 | 24 | hash = (hash ^ folded) * 1099511628211ULL; |
59 | 24 | } |
60 | 8 | return hash; |
61 | 8 | } |
62 | | }; |
63 | | |
64 | | struct CaseInsensitiveStringEqual { |
65 | | using is_transparent = void; |
66 | 4 | bool operator()(std::string_view lhs, std::string_view rhs) const { |
67 | 4 | if (lhs.size() != rhs.size()) return false; |
68 | 16 | for (size_t i = 0; i < lhs.size(); ++i) { |
69 | 12 | const unsigned char left = |
70 | 12 | lhs[i] >= 'A' && lhs[i] <= 'Z' ? lhs[i] + ('a' - 'A') : lhs[i]; |
71 | 12 | const unsigned char right = |
72 | 12 | rhs[i] >= 'A' && rhs[i] <= 'Z' ? rhs[i] + ('a' - 'A') : rhs[i]; |
73 | 12 | if (left != right) return false; |
74 | 12 | } |
75 | 4 | return true; |
76 | 4 | } |
77 | | }; |
78 | | |
79 | | // FileScannerV2 JSON reader. |
80 | | // |
81 | | // JSON files do not carry an embedded physical schema. The v2 table layer still needs a |
82 | | // file-local schema and FileScanRequest contract, so this reader exposes FE-provided file slots as |
83 | | // v2 file-local columns and performs JSON parsing/materialization directly in the v2 path. |
84 | | class JsonReader final : public FileReader { |
85 | | public: |
86 | | // `file_slot_descs` is the FE-planned file schema. JSON has no physical schema, so the reader |
87 | | // exposes these slots as synthetic file-local columns and materializes only the columns |
88 | | // requested by FileScanRequest. |
89 | | JsonReader(std::shared_ptr<io::FileSystemProperties>& system_properties, |
90 | | std::unique_ptr<io::FileDescription>& file_description, |
91 | | std::shared_ptr<io::IOContext> io_ctx, RuntimeProfile* profile, |
92 | | const TFileScanRangeParams* scan_params, const TFileRangeDesc& range, |
93 | | const std::vector<SlotDescriptor*>& file_slot_descs, |
94 | | TFileCompressType::type range_compress_type = TFileCompressType::UNKNOWN, |
95 | | std::optional<TUniqueId> stream_load_id = std::nullopt); |
96 | | ~JsonReader() override; |
97 | | |
98 | | // Initializes scan attributes and builds the synthetic schema from FE slots. |
99 | | Status init(RuntimeState* state) override; |
100 | | Status get_schema(std::vector<ColumnDefinition>* file_schema) const override; |
101 | | std::unique_ptr<TableColumnMapper> create_column_mapper( |
102 | | TableColumnMapperOptions options) const override; |
103 | | // Opens the underlying file or stream and binds requested local column ids to output block |
104 | | // positions. After this call, `get_block` can be called until it returns eof. |
105 | | Status open(std::shared_ptr<FileScanRequest> request) override; |
106 | | // Appends rows into `file_block` according to the FileScanRequest order. The block must already |
107 | | // contain columns matching the requested positions. |
108 | | Status get_block(Block* file_block, size_t* rows, bool* eof) override; |
109 | | Status close() override; |
110 | | |
111 | | #ifdef BE_TEST |
112 | 24 | size_t TEST_document_buffer_size() const { return _document_buffer.size(); } |
113 | | #endif |
114 | | |
115 | | private: |
116 | | void _init_profile() override; |
117 | | // A requested column keeps both identities: |
118 | | // - `source_index`: index in FE file slots, used for jsonpaths and SerDe lookup. |
119 | | // - `block_position`: index in the caller's output block, used for materialization. |
120 | | struct RequestedColumn { |
121 | | LocalColumnId file_column_id = LocalColumnId::invalid(); |
122 | | LocalIndex block_position; |
123 | | size_t source_index = 0; |
124 | | SlotDescriptor* slot_desc = nullptr; |
125 | | DataTypeSerDeSPtr serde; |
126 | | }; |
127 | | |
128 | | Status _build_requested_columns(const FileScanRequest& request, |
129 | | std::vector<RequestedColumn>* columns) const; |
130 | | // Reconciles TableReader's split/range descriptor with FileReader's concrete file description. |
131 | | TFileRangeDesc _json_range() const; |
132 | | Status _open_file_reader(); |
133 | | Status _create_decompressor(); |
134 | | Status _create_line_reader(); |
135 | | Status _parse_jsonpath_and_json_root(); |
136 | | // Reads one logical JSON document: one line for JSON Lines, or the whole range/pipe payload for |
137 | | // single-document mode. |
138 | | Status _read_one_document(size_t* size, bool* eof); |
139 | | Status _read_one_document_from_pipe(size_t* read_size); |
140 | | // Moves the logical document into a simdjson-padded buffer and creates an ondemand document. |
141 | | Status _parse_next_json(size_t* size, bool* eof); |
142 | | // Applies json_root and validates the object/array shape required by strip_outer_array. |
143 | | Status _extract_json_value(size_t size, bool* eof, bool* is_empty_row); |
144 | | Status _append_rows_from_current_value(Block* block, bool* is_empty_row, bool* eof); |
145 | | Status _append_simple_json_rows(Block* block, bool* is_empty_row, bool* eof); |
146 | | Status _append_flat_array_jsonpath_rows(Block* block, bool* is_empty_row, bool* eof); |
147 | | Status _append_nested_jsonpath_row(Block* block, bool* is_empty_row, bool* eof); |
148 | | Status _set_column_values_from_object(simdjson::ondemand::object* object_value, Block* block, |
149 | | bool* valid); |
150 | | Status _write_columns_by_jsonpath(simdjson::ondemand::object* object_value, Block* block, |
151 | | bool* valid); |
152 | | template <bool use_string_cache> |
153 | | Status _write_data_to_column(simdjson::ondemand::value& value, const DataTypePtr& type_desc, |
154 | | IColumn* column_ptr, const std::string& column_name, |
155 | | const DataTypeSerDeSPtr& serde, bool* valid); |
156 | | Status _fill_missing_column(const RequestedColumn& column, IColumn* column_ptr, bool* valid); |
157 | | Status _append_null_for_malformed_json(Block* block); |
158 | | Status _handle_json_error(const Status& status, Block* block, size_t original_rows, |
159 | | bool* is_empty_row); |
160 | | Status _apply_filters(Block* file_block, size_t* rows); |
161 | | void _truncate_block_to_rows(Block* block, size_t num_rows); |
162 | | void _pop_back_last_inserted_value(Block* block, size_t column_index); |
163 | | size_t _column_index(std::string_view key, size_t key_index); |
164 | | bool _is_root_path_for_column(const RequestedColumn& column) const; |
165 | | |
166 | | const TFileScanRangeParams* _scan_params = nullptr; |
167 | | TFileRangeDesc _range; |
168 | | TFileRangeDesc _reader_range; |
169 | | std::vector<SlotDescriptor*> _source_file_slot_descs; |
170 | | DataTypeSerDeSPtrs _source_serdes; |
171 | | std::vector<ColumnDefinition> _file_schema; |
172 | | RuntimeState* _runtime_state = nullptr; |
173 | | TFileCompressType::type _range_compress_type = TFileCompressType::UNKNOWN; |
174 | | std::optional<TUniqueId> _stream_load_id; |
175 | | std::vector<RequestedColumn> _requested_columns; |
176 | | std::unordered_map<std::string, size_t, TransparentStringHash, std::equal_to<>> |
177 | | _slot_name_to_index; |
178 | | std::unordered_map<std::string, size_t, CaseInsensitiveStringHash, CaseInsensitiveStringEqual> |
179 | | _hive_slot_name_to_index; |
180 | | std::vector<size_t> _previous_positions; |
181 | | |
182 | | RuntimeProfile::Counter* _total_time = nullptr; |
183 | | RuntimeProfile::Counter* _open_time = nullptr; |
184 | | RuntimeProfile::Counter* _read_document_time = nullptr; |
185 | | RuntimeProfile::Counter* _parse_time = nullptr; |
186 | | RuntimeProfile::Counter* _materialize_time = nullptr; |
187 | | RuntimeProfile::Counter* _filter_time = nullptr; |
188 | | |
189 | | io::FileReaderSPtr _physical_file_reader; |
190 | | std::unique_ptr<Decompressor> _decompressor; |
191 | | std::unique_ptr<LineReader> _line_reader; |
192 | | int64_t _current_offset = 0; |
193 | | bool _reader_eof = false; |
194 | | bool _skip_first_line = false; |
195 | | bool _single_document_read = false; |
196 | | |
197 | | std::string _line_delimiter; |
198 | | size_t _line_delimiter_length = 0; |
199 | | std::string _jsonpaths; |
200 | | std::string _json_root; |
201 | | bool _read_json_by_line = false; |
202 | | bool _strip_outer_array = false; |
203 | | bool _num_as_string = false; |
204 | | bool _fuzzy_parse = false; |
205 | | bool _is_hive_table = false; |
206 | | bool _openx_json_ignore_malformed = false; |
207 | | TFileCompressType::type _file_compress_type = TFileCompressType::UNKNOWN; |
208 | | |
209 | | std::vector<std::vector<JsonPath>> _parsed_jsonpaths; |
210 | | std::vector<JsonPath> _parsed_json_root; |
211 | | bool _parsed_from_json_root = false; |
212 | | DataTypeSerDe::FormatOptions _serde_options; |
213 | | |
214 | | // simdjson ondemand values point into `_padding_buffer`, so the buffer must outlive all values |
215 | | // created from the current document. |
216 | | std::unique_ptr<simdjson::ondemand::parser> _json_parser; |
217 | | simdjson::ondemand::document _original_json_doc; |
218 | | simdjson::ondemand::value _json_value; |
219 | | simdjson::ondemand::array _array; |
220 | | simdjson::ondemand::array_iterator _array_iter; |
221 | | std::string _document_buffer; |
222 | | std::string_view _document_view; |
223 | | std::string _padding_buffer; |
224 | | size_t _original_doc_size = 0; |
225 | | size_t _padded_size = 1024 * 1024 * 8 + simdjson::SIMDJSON_PADDING; |
226 | | std::unordered_map<std::string_view, std::string_view> _cached_string_values; |
227 | | }; |
228 | | |
229 | | } // namespace doris::format::json |