Coverage Report

Created: 2026-07-27 16:12

next uncovered line (L), next uncovered region (R), next uncovered branch (B)
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
40
    size_t operator()(std::string_view value) const {
48
40
        return std::hash<std::string_view> {}(value);
49
40
    }
50
};
51
52
struct CaseInsensitiveStringHash {
53
    using is_transparent = void;
54
4
    size_t operator()(std::string_view value) const {
55
4
        size_t hash = 1469598103934665603ULL;
56
12
        for (const unsigned char c : value) {
57
12
            const unsigned char folded = c >= 'A' && c <= 'Z' ? c + ('a' - 'A') : c;
58
12
            hash = (hash ^ folded) * 1099511628211ULL;
59
12
        }
60
4
        return hash;
61
4
    }
62
};
63
64
struct CaseInsensitiveStringEqual {
65
    using is_transparent = void;
66
2
    bool operator()(std::string_view lhs, std::string_view rhs) const {
67
2
        if (lhs.size() != rhs.size()) return false;
68
8
        for (size_t i = 0; i < lhs.size(); ++i) {
69
6
            const unsigned char left =
70
6
                    lhs[i] >= 'A' && lhs[i] <= 'Z' ? lhs[i] + ('a' - 'A') : lhs[i];
71
6
            const unsigned char right =
72
6
                    rhs[i] >= 'A' && rhs[i] <= 'Z' ? rhs[i] + ('a' - 'A') : rhs[i];
73
6
            if (left != right) return false;
74
6
        }
75
2
        return true;
76
2
    }
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
12
    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