be/src/format/json/new_json_reader.cpp

Source
// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements.  See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership.  The ASF licenses this file
// to you under the Apache License, Version 2.0 (the
// "License"); you may not use this file except in compliance
// with the License.  You may obtain a copy of the License at
//
//   http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing,
// software distributed under the License is distributed on an
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
// KIND, either express or implied.  See the License for the
// specific language governing permissions and limitations
// under the License.

#include "format/json/new_json_reader.h"

#include <fmt/format.h>
#include <gen_cpp/Metrics_types.h>
#include <gen_cpp/PlanNodes_types.h>
#include <gen_cpp/Types_types.h>
#include <glog/logging.h>
#include <rapidjson/error/en.h>
#include <rapidjson/reader.h>
#include <rapidjson/stringbuffer.h>
#include <rapidjson/writer.h>
#include <simdjson/simdjson.h> // IWYU pragma: keep

#include <algorithm>
#include <cinttypes>
#include <cstdio>
#include <cstring>
#include <map>
#include <memory>
#include <string_view>
#include <utility>

#include "common/compiler_util.h" // IWYU pragma: keep
#include "common/config.h"
#include "common/exception.h"
#include "common/status.h"
#include "core/assert_cast.h"
#include "core/block/column_with_type_and_name.h"
#include "core/column/column.h"
#include "core/column/column_array.h"
#include "core/column/column_map.h"
#include "core/column/column_nullable.h"
#include "core/column/column_string.h"
#include "core/column/column_struct.h"
#include "core/column/column_variant.h"
#include "core/custom_allocator.h"
#include "core/data_type/data_type_array.h"
#include "core/data_type/data_type_factory.hpp"
#include "core/data_type/data_type_map.h"
#include "core/data_type/data_type_number.h" // IWYU pragma: keep
#include "core/data_type/data_type_struct.h"
#include "core/data_type/define_primitive_type.h"
#include "exec/common/variant_util.h"
#include "exec/scan/scanner.h"
#include "exprs/json_functions.h"
#include "format/file_reader/new_plain_text_line_reader.h"
#include "io/file_factory.h"
#include "io/fs/buffered_reader.h"
#include "io/fs/file_reader.h"
#include "io/fs/stream_load_pipe.h"
#include "io/fs/tracing_file_reader.h"
#include "runtime/descriptors.h"
#include "runtime/runtime_state.h"
#include "util/slice.h"

namespace doris::io {
struct IOContext;
enum class FileCachePolicy : uint8_t;
} // namespace doris::io

namespace doris {
using namespace ErrorCode;

NewJsonReader::NewJsonReader(RuntimeState* state, RuntimeProfile* profile, ScannerCounter* counter,
                             const TFileScanRangeParams& params, const TFileRangeDesc& range,
                             const std::vector<SlotDescriptor*>& file_slot_descs, bool* scanner_eof,
                             size_t batch_size, io::IOContext* io_ctx,
                             std::shared_ptr<io::IOContext> io_ctx_holder)
        : _vhandle_json_callback(nullptr),
          _state(state),
          _profile(profile),
          _counter(counter),
          _params(params),
          _range(range),
          _file_slot_descs(file_slot_descs),
          _file_reader(nullptr),
          _line_reader(nullptr),
          _reader_eof(false),
          _decompressor(nullptr),
          _skip_first_line(false),
          _next_row(0),
          _total_rows(0),
          _value_allocator(_value_buffer, sizeof(_value_buffer)),
          _parse_allocator(_parse_buffer, sizeof(_parse_buffer)),
          _origin_json_doc(&_value_allocator, sizeof(_parse_buffer), &_parse_allocator),
          _scanner_eof(scanner_eof),
          _current_offset(0),
          _io_ctx(io_ctx),
          _io_ctx_holder(std::move(io_ctx_holder)),
          _batch_size(std::max(batch_size, 1UL)) {
    if (_io_ctx == nullptr && _io_ctx_holder) {
        _io_ctx = _io_ctx_holder.get();
    }
    _read_timer = ADD_TIMER(_profile, "ReadTime");
    if (_range.__isset.compress_type) {
        // for compatibility
        _file_compress_type = _range.compress_type;
    } else {
        _file_compress_type = _params.compress_type;
    }
    _init_system_properties();
    _init_file_description();
}

NewJsonReader::NewJsonReader(RuntimeProfile* profile, const TFileScanRangeParams& params,
                             const TFileRangeDesc& range,
                             const std::vector<SlotDescriptor*>& file_slot_descs, size_t batch_size,
                             io::IOContext* io_ctx, std::shared_ptr<io::IOContext> io_ctx_holder)
        : _vhandle_json_callback(nullptr),
          _state(nullptr),
          _profile(profile),
          _params(params),
          _range(range),
          _file_slot_descs(file_slot_descs),
          _line_reader(nullptr),
          _reader_eof(false),
          _decompressor(nullptr),
          _skip_first_line(false),
          _next_row(0),
          _total_rows(0),
          _value_allocator(_value_buffer, sizeof(_value_buffer)),
          _parse_allocator(_parse_buffer, sizeof(_parse_buffer)),
          _origin_json_doc(&_value_allocator, sizeof(_parse_buffer), &_parse_allocator),
          _io_ctx(io_ctx),
          _io_ctx_holder(std::move(io_ctx_holder)),
          _batch_size(std::max(batch_size, 1UL)) {
    if (_io_ctx == nullptr && _io_ctx_holder) {
        _io_ctx = _io_ctx_holder.get();
    }
    if (_range.__isset.compress_type) {
        // for compatibility
        _file_compress_type = _range.compress_type;
    } else {
        _file_compress_type = _params.compress_type;
    }
    _init_system_properties();
    _init_file_description();
}

void NewJsonReader::_init_system_properties() {
    if (_range.__isset.file_type) {
        // for compatibility
        _system_properties.system_type = _range.file_type;
    } else {
        _system_properties.system_type = _params.file_type;
    }
    _system_properties.properties = _params.properties;
    _system_properties.hdfs_params = _params.hdfs_params;
    if (_params.__isset.broker_addresses) {
        _system_properties.broker_addresses.assign(_params.broker_addresses.begin(),
                                                   _params.broker_addresses.end());
    }
}

void NewJsonReader::_init_file_description() {
    _file_description.path = _range.path;
    _file_description.file_size = _range.__isset.file_size ? _range.file_size : -1;

    if (_range.__isset.fs_name) {
        _file_description.fs_name = _range.fs_name;
    }
    if (_range.__isset.file_cache_admission) {
        _file_description.file_cache_admission = _range.file_cache_admission;
    }
}

Status NewJsonReader::init_reader(
        const std::unordered_map<std::string, VExprContextSPtr>& col_default_value_ctx,
        bool is_load) {
    _is_load = is_load;

    // generate _col_default_value_map
    RETURN_IF_ERROR(_get_column_default_value(_file_slot_descs, col_default_value_ctx));

    //use serde insert data to column.
    for (auto* slot_desc : _file_slot_descs) {
        _serdes.emplace_back(slot_desc->get_data_type_ptr()->get_serde());
    }

    // create decompressor.
    // _decompressor may be nullptr if this is not a compressed file
    RETURN_IF_ERROR(Decompressor::create_decompressor(_file_compress_type, &_decompressor));

    RETURN_IF_ERROR(_simdjson_init_reader());
    return Status::OK();
}

// ---- Unified init_reader(ReaderInitContext*) overrides ----

Status NewJsonReader::_open_file_reader(ReaderInitContext* /*ctx*/) {
    RETURN_IF_ERROR(_get_range_params());
    RETURN_IF_ERROR(_open_file_reader(false));
    return Status::OK();
}

Status NewJsonReader::_do_init_reader(ReaderInitContext* base_ctx) {
    auto* ctx = checked_context_cast<JsonInitContext>(base_ctx);
    _is_load = ctx->is_load;

    RETURN_IF_ERROR(_get_column_default_value(_file_slot_descs, *ctx->col_default_value_ctx));
    for (auto* slot_desc : _file_slot_descs) {
        _serdes.emplace_back(slot_desc->get_data_type_ptr()->get_serde());
    }

    // Create decompressor (needed by line reader below)
    RETURN_IF_ERROR(Decompressor::create_decompressor(_file_compress_type, &_decompressor));

    if (LIKELY(_read_json_by_line)) {
        RETURN_IF_ERROR(_open_line_reader());
    }
    RETURN_IF_ERROR(_parse_jsonpath_and_json_root());

    if (_parsed_jsonpaths.empty()) {
        _vhandle_json_callback = &NewJsonReader::_simdjson_handle_simple_json;
    } else {
        if (_strip_outer_array) {
            _vhandle_json_callback = &NewJsonReader::_simdjson_handle_flat_array_complex_json;
        } else {
            _vhandle_json_callback = &NewJsonReader::_simdjson_handle_nested_complex_json;
        }
    }
    _ondemand_json_parser = std::make_unique<simdjson::ondemand::parser>();
    for (int i = 0; i < _file_slot_descs.size(); ++i) {
        _slot_desc_index[StringRef {_file_slot_descs[i]->col_name()}] = i;
        if (_file_slot_descs[i]->is_skip_bitmap_col()) {
            skip_bitmap_col_idx = i;
        }
    }
    _simdjson_ondemand_padding_buffer.resize(_padded_size);
    _simdjson_ondemand_unscape_padding_buffer.resize(_padded_size);
    return Status::OK();
}

void NewJsonReader::set_batch_size(size_t batch_size) {
    // 0 means "not set" / "use default" for the row-based readers; we must
    // never let _batch_size be 0 because _do_get_next_block uses it as the
    // upper bound of a `while (block->rows() < batch_size)` loop and a 0
    // would make the reader return without setting eof, causing the scanner
    // to spin on empty blocks.
    _batch_size = std::max(batch_size, 1UL);
}

Status NewJsonReader::_do_get_next_block(Block* block, size_t* read_rows, bool* eof) {
    if (_reader_eof) {
        *eof = true;
        return Status::OK();
    }

    const auto batch_size = _batch_size;
    const auto max_block_bytes = _state->preferred_block_size_bytes();

    while (block->rows() < batch_size && !_reader_eof && (block->bytes() < max_block_bytes)) {
        if (UNLIKELY(_read_json_by_line && _skip_first_line)) {
            size_t size = 0;
            const uint8_t* line_ptr = nullptr;
            RETURN_IF_ERROR(_line_reader->read_line(&line_ptr, &size, &_reader_eof, _io_ctx));
            _skip_first_line = false;
            continue;
        }

        bool is_empty_row = false;

        RETURN_IF_ERROR(
                _read_json_column(_state, *block, _file_slot_descs, &is_empty_row, &_reader_eof));
        if (is_empty_row) {
            // Read empty row, just continue
            continue;
        }
        ++(*read_rows);
    }

    return Status::OK();
}

Status NewJsonReader::_get_columns_impl(
        std::unordered_map<std::string, DataTypePtr>* name_to_type) {
    for (const auto& slot : _file_slot_descs) {
        name_to_type->emplace(slot->col_name(), slot->type());
    }
    return Status::OK();
}

// init decompressor, file reader and line reader for parsing schema
Status NewJsonReader::init_schema_reader() {
    RETURN_IF_ERROR(_get_range_params());
    // create decompressor.
    // _decompressor may be nullptr if this is not a compressed file
    RETURN_IF_ERROR(Decompressor::create_decompressor(_file_compress_type, &_decompressor));
    RETURN_IF_ERROR(_open_file_reader(true));
    if (_read_json_by_line) {
        RETURN_IF_ERROR(_open_line_reader());
    }
    // generate _parsed_jsonpaths and _parsed_json_root
    RETURN_IF_ERROR(_parse_jsonpath_and_json_root());
    return Status::OK();
}

Status NewJsonReader::get_parsed_schema(std::vector<std::string>* col_names,
                                        std::vector<DataTypePtr>* col_types) {
    bool eof = false;
    const uint8_t* json_str = nullptr;
    DorisUniqueBufferPtr<uint8_t> json_str_ptr;
    size_t size = 0;
    if (_line_reader != nullptr) {
        RETURN_IF_ERROR(_line_reader->read_line(&json_str, &size, &eof, _io_ctx));
    } else {
        size_t read_size = 0;
        RETURN_IF_ERROR(_read_one_message(&json_str_ptr, &read_size));
        json_str = json_str_ptr.get();
        size = read_size;
        if (read_size == 0) {
            eof = true;
        }
    }

    if (size == 0 || eof) {
        return Status::EndOfFile("Empty file.");
    }

    // clear memory here.
    _value_allocator.Clear();
    _parse_allocator.Clear();
    bool has_parse_error = false;

    // parse jsondata to JsonDoc
    // As the issue: https://github.com/Tencent/rapidjson/issues/1458
    // Now, rapidjson only support uint64_t, So lagreint load cause bug. We use kParseNumbersAsStringsFlag.
    if (_num_as_string) {
        has_parse_error =
                _origin_json_doc.Parse<rapidjson::kParseNumbersAsStringsFlag>((char*)json_str, size)
                        .HasParseError();
    } else {
        has_parse_error = _origin_json_doc.Parse((char*)json_str, size).HasParseError();
    }

    if (has_parse_error) {
        return Status::DataQualityError(
                "Parse json data for JsonDoc failed. code: {}, error info: {}",
                _origin_json_doc.GetParseError(),
                rapidjson::GetParseError_En(_origin_json_doc.GetParseError()));
    }

    // set json root
    if (!_parsed_json_root.empty()) {
        _json_doc = JsonFunctions::get_json_object_from_parsed_json(
                _parsed_json_root, &_origin_json_doc, _origin_json_doc.GetAllocator());
        if (_json_doc == nullptr) {
            return Status::DataQualityError("JSON Root not found.");
        }
    } else {
        _json_doc = &_origin_json_doc;
    }

    if (_json_doc->IsArray() && !_strip_outer_array) {
        return Status::DataQualityError(
                "JSON data is array-object, `strip_outer_array` must be TRUE.");
    }
    if (!_json_doc->IsArray() && _strip_outer_array) {
        return Status::DataQualityError(
                "JSON data is not an array-object, `strip_outer_array` must be FALSE.");
    }

    rapidjson::Value* objectValue = nullptr;
    if (_json_doc->IsArray()) {
        if (_json_doc->Size() == 0) {
            // may be passing an empty json, such as "[]"
            return Status::InternalError<false>("Empty first json line");
        }
        objectValue = &(*_json_doc)[0];
    } else {
        objectValue = _json_doc;
    }

    if (!objectValue->IsObject()) {
        return Status::DataQualityError("JSON data is not an object. but: {}",
                                        objectValue->GetType());
    }

    // use jsonpaths to col_names
    if (!_parsed_jsonpaths.empty()) {
        for (auto& _parsed_jsonpath : _parsed_jsonpaths) {
            size_t len = _parsed_jsonpath.size();
            if (len == 0) {
                return Status::InvalidArgument("It's invalid jsonpaths.");
            }
            std::string key = _parsed_jsonpath[len - 1].key;
            col_names->emplace_back(key);
            col_types->emplace_back(
                    DataTypeFactory::instance().create_data_type(PrimitiveType::TYPE_STRING, true));
        }
        return Status::OK();
    }

    for (int i = 0; i < objectValue->MemberCount(); ++i) {
        auto it = objectValue->MemberBegin() + i;
        col_names->emplace_back(it->name.GetString());
        col_types->emplace_back(make_nullable(std::make_shared<DataTypeString>()));
    }
    return Status::OK();
}

Status NewJsonReader::_get_range_params() {
    if (!_params.__isset.file_attributes) {
        return Status::InternalError<false>("BE cat get file_attributes");
    }

    // get line_delimiter
    if (_params.file_attributes.__isset.text_params &&
        _params.file_attributes.text_params.__isset.line_delimiter) {
        _line_delimiter = _params.file_attributes.text_params.line_delimiter;
        _line_delimiter_length = _line_delimiter.size();
    }

    if (_params.file_attributes.__isset.jsonpaths) {
        _jsonpaths = _params.file_attributes.jsonpaths;
    }
    if (_params.file_attributes.__isset.json_root) {
        _json_root = _params.file_attributes.json_root;
    }
    if (_params.file_attributes.__isset.read_json_by_line) {
        _read_json_by_line = _params.file_attributes.read_json_by_line;
    }
    if (_params.file_attributes.__isset.strip_outer_array) {
        _strip_outer_array = _params.file_attributes.strip_outer_array;
    }
    if (_params.file_attributes.__isset.num_as_string) {
        _num_as_string = _params.file_attributes.num_as_string;
    }
    if (_params.file_attributes.__isset.fuzzy_parse) {
        _fuzzy_parse = _params.file_attributes.fuzzy_parse;
    }
    if (_range.table_format_params.table_format_type == "hive") {
        _is_hive_table = true;
    }
    if (_params.file_attributes.__isset.openx_json_ignore_malformed) {
        _openx_json_ignore_malformed = _params.file_attributes.openx_json_ignore_malformed;
    }
    return Status::OK();
}

static Status ignore_malformed_json_append_null(Block& block) {
    for (auto& column : block.get_columns()) {
        if (!column->is_nullable()) [[unlikely]] {
            return Status::DataQualityError("malformed json, but the column `{}` is not nullable.",
                                            column->get_name());
        }
        static_cast<ColumnNullable*>(column->assume_mutable().get())->insert_default();
    }
    return Status::OK();
}

Status NewJsonReader::_open_file_reader(bool need_schema) {
    int64_t start_offset = _range.start_offset;
    if (start_offset != 0) {
        start_offset -= 1;
    }

    _current_offset = start_offset;

    if (_params.file_type == TFileType::FILE_STREAM) {
        // Due to http_stream needs to pre read a portion of the data to parse column information, so it is set to true here
        RETURN_IF_ERROR(FileFactory::create_pipe_reader(_range.load_id, &_file_reader, _state,
                                                        need_schema));
    } else {
        _file_description.mtime = _range.__isset.modification_time ? _range.modification_time : 0;
        io::FileReaderOptions reader_options =
                FileFactory::get_reader_options(_state, _file_description);
        io::FileReaderSPtr file_reader;
        if (_io_ctx_holder) {
            file_reader = DORIS_TRY(io::DelegateReader::create_file_reader(
                    _profile, _system_properties, _file_description, reader_options,
                    io::DelegateReader::AccessMode::SEQUENTIAL,
                    std::static_pointer_cast<const io::IOContext>(_io_ctx_holder),
                    io::PrefetchRange(_range.start_offset, _range.size)));
        } else {
            file_reader = DORIS_TRY(io::DelegateReader::create_file_reader(
                    _profile, _system_properties, _file_description, reader_options,
                    io::DelegateReader::AccessMode::SEQUENTIAL, _io_ctx,
                    io::PrefetchRange(_range.start_offset, _range.size)));
        }
        _file_reader = _io_ctx ? std::make_shared<io::TracingFileReader>(std::move(file_reader),
                                                                         _io_ctx->file_reader_stats)
                               : file_reader;
    }
    return Status::OK();
}

Status NewJsonReader::_open_line_reader() {
    int64_t size = _range.size;
    if (_range.start_offset != 0) {
        // When we fetch range doesn't start from 0, size will += 1.
        size += 1;
        _skip_first_line = true;
    } else {
        _skip_first_line = false;
    }
    _line_reader = NewPlainTextLineReader::create_unique(
            _profile, _file_reader, _decompressor.get(),
            std::make_shared<PlainTextLineReaderCtx>(_line_delimiter, _line_delimiter_length,
                                                     false),
            size, _current_offset);
    return Status::OK();
}

Status NewJsonReader::_parse_jsonpath_and_json_root() {
    // parse jsonpaths
    if (!_jsonpaths.empty()) {
        rapidjson::Document jsonpaths_doc;
        if (!jsonpaths_doc.Parse(_jsonpaths.c_str(), _jsonpaths.length()).HasParseError()) {
            if (!jsonpaths_doc.IsArray()) {
                return Status::InvalidJsonPath("Invalid json path: {}", _jsonpaths);
            }
            for (int i = 0; i < jsonpaths_doc.Size(); i++) {
                const rapidjson::Value& path = jsonpaths_doc[i];
                if (!path.IsString()) {
                    return Status::InvalidJsonPath("Invalid json path: {}", _jsonpaths);
                }
                std::string json_path = path.GetString();
                // $ -> $. in json_path
                if (UNLIKELY(json_path.size() == 1 && json_path[0] == '$')) {
                    json_path.insert(1, ".");
                }
                std::vector<JsonPath> parsed_paths;
                JsonFunctions::parse_json_paths(json_path, &parsed_paths);
                _parsed_jsonpaths.push_back(std::move(parsed_paths));
            }

        } else {
            return Status::InvalidJsonPath("Invalid json path: {}", _jsonpaths);
        }
    }

    // parse jsonroot
    if (!_json_root.empty()) {
        std::string json_root = _json_root;
        //  $ -> $. in json_root
        if (json_root.size() == 1 && json_root[0] == '$') {
            json_root.insert(1, ".");
        }
        JsonFunctions::parse_json_paths(json_root, &_parsed_json_root);
    }
    return Status::OK();
}

Status NewJsonReader::_read_json_column(RuntimeState* state, Block& block,
                                        const std::vector<SlotDescriptor*>& slot_descs,
                                        bool* is_empty_row, bool* eof) {
    return (this->*_vhandle_json_callback)(state, block, slot_descs, is_empty_row, eof);
}

Status NewJsonReader::_read_one_message(DorisUniqueBufferPtr<uint8_t>* file_buf,
                                        size_t* read_size) {
    switch (_params.file_type) {
    case TFileType::FILE_LOCAL:
        [[fallthrough]];
    case TFileType::FILE_HDFS:
    case TFileType::FILE_HTTP:
        [[fallthrough]];
    case TFileType::FILE_S3: {
        size_t file_size = _file_reader->size();
        *file_buf = make_unique_buffer<uint8_t>(file_size);
        Slice result(file_buf->get(), file_size);
        RETURN_IF_ERROR(_file_reader->read_at(_current_offset, result, read_size, _io_ctx));
        _current_offset += *read_size;
        break;
    }
    case TFileType::FILE_STREAM: {
        RETURN_IF_ERROR(_read_one_message_from_pipe(file_buf, read_size));
        break;
    }
    default: {
        return Status::NotSupported<false>("no supported file reader type: {}", _params.file_type);
    }
    }
    return Status::OK();
}

Status NewJsonReader::_read_one_message_from_pipe(DorisUniqueBufferPtr<uint8_t>* file_buf,
                                                  size_t* read_size) {
    auto* stream_load_pipe = dynamic_cast<io::StreamLoadPipe*>(_file_reader.get());

    // first read: read from the pipe once.
    RETURN_IF_ERROR(stream_load_pipe->read_one_message(file_buf, read_size));

    // When the file is not chunked, the entire file has already been read.
    if (!stream_load_pipe->is_chunked_transfer()) {
        return Status::OK();
    }

    std::vector<uint8_t> buf;
    uint64_t cur_size = 0;

    // second read: continuously read data from the pipe until all data is read.
    DorisUniqueBufferPtr<uint8_t> read_buf;
    size_t read_buf_size = 0;
    while (true) {
        RETURN_IF_ERROR(stream_load_pipe->read_one_message(&read_buf, &read_buf_size));
        if (read_buf_size == 0) {
            break;
        } else {
            buf.insert(buf.end(), read_buf.get(), read_buf.get() + read_buf_size);
            cur_size += read_buf_size;
            read_buf_size = 0;
            read_buf.reset();
        }
    }

    // No data is available during the second read.
    if (cur_size == 0) {
        return Status::OK();
    }

    DorisUniqueBufferPtr<uint8_t> total_buf = make_unique_buffer<uint8_t>(cur_size + *read_size);

    // copy the data during the first read
    memcpy(total_buf.get(), file_buf->get(), *read_size);

    // copy the data during the second read
    memcpy(total_buf.get() + *read_size, buf.data(), cur_size);
    *file_buf = std::move(total_buf);
    *read_size += cur_size;
    return Status::OK();
}

// ---------SIMDJSON----------
// simdjson, replace none simdjson function if it is ready
Status NewJsonReader::_simdjson_init_reader() {
    RETURN_IF_ERROR(_get_range_params());

    RETURN_IF_ERROR(_open_file_reader(false));
    if (LIKELY(_read_json_by_line)) {
        RETURN_IF_ERROR(_open_line_reader());
    }

    // generate _parsed_jsonpaths and _parsed_json_root
    RETURN_IF_ERROR(_parse_jsonpath_and_json_root());

    //improve performance
    if (_parsed_jsonpaths.empty()) { // input is a simple json-string
        _vhandle_json_callback = &NewJsonReader::_simdjson_handle_simple_json;
    } else { // input is a complex json-string and a json-path
        if (_strip_outer_array) {
            _vhandle_json_callback = &NewJsonReader::_simdjson_handle_flat_array_complex_json;
        } else {
            _vhandle_json_callback = &NewJsonReader::_simdjson_handle_nested_complex_json;
        }
    }
    _ondemand_json_parser = std::make_unique<simdjson::ondemand::parser>();
    for (int i = 0; i < _file_slot_descs.size(); ++i) {
        _slot_desc_index[StringRef {_file_slot_descs[i]->col_name()}] = i;
        if (_file_slot_descs[i]->is_skip_bitmap_col()) {
            skip_bitmap_col_idx = i;
        }
    }
    _simdjson_ondemand_padding_buffer.resize(_padded_size);
    _simdjson_ondemand_unscape_padding_buffer.resize(_padded_size);
    return Status::OK();
}

Status NewJsonReader::_handle_simdjson_error(simdjson::simdjson_error& error, Block& block,
                                             size_t num_rows, bool* eof) {
    fmt::memory_buffer error_msg;
    fmt::format_to(error_msg, "Parse json data failed. code: {}, error info: {}", error.error(),
                   error.what());
    _counter->num_rows_filtered++;
    // Before continuing to process other rows, we need to first clean the fail parsed row.
    for (int i = 0; i < block.columns(); ++i) {
        auto column = block.get_by_position(i).column->assume_mutable();
        if (column->size() > num_rows) {
            column->pop_back(column->size() - num_rows);
        }
    }

    RETURN_IF_ERROR(_state->append_error_msg_to_file(
            [&]() -> std::string {
                return std::string(_simdjson_ondemand_padding_buffer.data(), _original_doc_size);
            },
            [&]() -> std::string { return fmt::to_string(error_msg); }));
    return Status::OK();
}

Status NewJsonReader::_simdjson_handle_simple_json(RuntimeState* /*state*/, Block& block,
                                                   const std::vector<SlotDescriptor*>& slot_descs,
                                                   bool* is_empty_row, bool* eof) {
    // simple json
    size_t size = 0;
    simdjson::error_code error;
    size_t num_rows = block.rows();
    try {
        // step1: get and parse buf to get json doc
        RETURN_IF_ERROR(_simdjson_parse_json(&size, is_empty_row, eof, &error));
        if (size == 0 || *eof) {
            *is_empty_row = true;
            return Status::OK();
        }

        // step2: get json value by json doc
        Status st = _get_json_value(&size, eof, &error, is_empty_row);
        if (st.is<DATA_QUALITY_ERROR>()) {
            if (_is_load) {
                return Status::OK();
            } else if (_openx_json_ignore_malformed) {
                RETURN_IF_ERROR(ignore_malformed_json_append_null(block));
                return Status::OK();
            }
        }

        RETURN_IF_ERROR(st);
        if (*is_empty_row || *eof) {
            return Status::OK();
        }

        // step 3: write columns by json value
        RETURN_IF_ERROR(
                _simdjson_handle_simple_json_write_columns(block, slot_descs, is_empty_row, eof));
    } catch (simdjson::simdjson_error& e) {
        RETURN_IF_ERROR(_handle_simdjson_error(e, block, num_rows, eof));
        if (*_scanner_eof) {
            // When _scanner_eof is true and valid is false, it means that we have encountered
            // unqualified data and decided to stop the scan.
            *is_empty_row = true;
            return Status::OK();
        }
    }

    return Status::OK();
}

Status NewJsonReader::_simdjson_handle_simple_json_write_columns(
        Block& block, const std::vector<SlotDescriptor*>& slot_descs, bool* is_empty_row,
        bool* eof) {
    simdjson::ondemand::object objectValue;
    size_t num_rows = block.rows();
    bool valid = false;
    try {
        if (_json_value.type() == simdjson::ondemand::json_type::array) {
            _array = _json_value.get_array();
            if (_array.count_elements() == 0) {
                // may be passing an empty json, such as "[]"
                RETURN_IF_ERROR(_append_error_msg(nullptr, "Empty json line", "", nullptr));
                if (*_scanner_eof) {
                    *is_empty_row = true;
                    return Status::OK();
                }
                return Status::OK();
            }

            _array_iter = _array.begin();
            while (true) {
                objectValue = *_array_iter;
                RETURN_IF_ERROR(
                        _simdjson_set_column_value(&objectValue, block, slot_descs, &valid));
                if (!valid) {
                    if (*_scanner_eof) {
                        // When _scanner_eof is true and valid is false, it means that we have encountered
                        // unqualified data and decided to stop the scan.
                        *is_empty_row = true;
                        return Status::OK();
                    }
                }
                ++_array_iter;
                if (_array_iter == _array.end()) {
                    // Hint to read next json doc
                    break;
                }
            }
        } else {
            objectValue = _json_value;
            RETURN_IF_ERROR(_simdjson_set_column_value(&objectValue, block, slot_descs, &valid));
            if (!valid) {
                if (*_scanner_eof) {
                    *is_empty_row = true;
                    return Status::OK();
                }
            }
            *is_empty_row = false;
        }
    } catch (simdjson::simdjson_error& e) {
        RETURN_IF_ERROR(_handle_simdjson_error(e, block, num_rows, eof));
        if (!valid) {
            if (*_scanner_eof) {
                *is_empty_row = true;
                return Status::OK();
            }
        }
    }
    return Status::OK();
}

Status NewJsonReader::_simdjson_handle_flat_array_complex_json(
        RuntimeState* /*state*/, Block& block, const std::vector<SlotDescriptor*>& slot_descs,
        bool* is_empty_row, bool* eof) {
    // array complex json
    size_t size = 0;
    simdjson::error_code error;
    size_t num_rows = block.rows();
    try {
        // step1: get and parse buf to get json doc
        RETURN_IF_ERROR(_simdjson_parse_json(&size, is_empty_row, eof, &error));
        if (size == 0 || *eof) {
            *is_empty_row = true;
            return Status::OK();
        }

        // step2: get json value by json doc
        Status st = _get_json_value(&size, eof, &error, is_empty_row);
        if (st.is<DATA_QUALITY_ERROR>()) {
            return Status::OK();
        }
        RETURN_IF_ERROR(st);
        if (*is_empty_row) {
            return Status::OK();
        }

        // step 3: write columns by json value
        RETURN_IF_ERROR(_simdjson_handle_flat_array_complex_json_write_columns(block, slot_descs,
                                                                               is_empty_row, eof));
    } catch (simdjson::simdjson_error& e) {
        RETURN_IF_ERROR(_handle_simdjson_error(e, block, num_rows, eof));
        if (*_scanner_eof) {
            // When _scanner_eof is true and valid is false, it means that we have encountered
            // unqualified data and decided to stop the scan.
            *is_empty_row = true;
            return Status::OK();
        }
    }

    return Status::OK();
}

Status NewJsonReader::_simdjson_handle_flat_array_complex_json_write_columns(
        Block& block, const std::vector<SlotDescriptor*>& slot_descs, bool* is_empty_row,
        bool* eof) {
// Advance one row in array list, if it is the endpoint, stop advance and break the loop
#define ADVANCE_ROW()                  \
    ++_array_iter;                     \
    if (_array_iter == _array.end()) { \
        break;                         \
    }

    simdjson::ondemand::object cur;
    size_t num_rows = block.rows();
    try {
        bool valid = true;
        _array = _json_value.get_array();
        _array_iter = _array.begin();

        while (true) {
            cur = (*_array_iter).get_object();
            // extract root
            if (!_parsed_from_json_root && !_parsed_json_root.empty()) {
                simdjson::ondemand::value val;
                Status st = JsonFunctions::extract_from_object(cur, _parsed_json_root, &val);
                if (UNLIKELY(!st.ok())) {
                    if (st.is<NOT_FOUND>()) {
                        RETURN_IF_ERROR(_append_error_msg(nullptr, st.to_string(), "", nullptr));
                        ADVANCE_ROW();
                        continue;
                    }
                    return st;
                }
                if (val.type() != simdjson::ondemand::json_type::object) {
                    RETURN_IF_ERROR(_append_error_msg(nullptr, "Not object item", "", nullptr));
                    ADVANCE_ROW();
                    continue;
                }
                cur = val.get_object();
            }
            RETURN_IF_ERROR(_simdjson_write_columns_by_jsonpath(&cur, slot_descs, block, &valid));
            ADVANCE_ROW();
            if (!valid) {
                continue; // process next line
            }
            *is_empty_row = false;
        }
    } catch (simdjson::simdjson_error& e) {
        RETURN_IF_ERROR(_handle_simdjson_error(e, block, num_rows, eof));
        if (*_scanner_eof) {
            // When _scanner_eof is true and valid is false, it means that we have encountered
            // unqualified data and decided to stop the scan.
            *is_empty_row = true;
            return Status::OK();
        }
    }

    return Status::OK();
}

Status NewJsonReader::_simdjson_handle_nested_complex_json(
        RuntimeState* /*state*/, Block& block, const std::vector<SlotDescriptor*>& slot_descs,
        bool* is_empty_row, bool* eof) {
    // nested complex json
    while (true) {
        size_t num_rows = block.rows();
        simdjson::ondemand::object cur;
        size_t size = 0;
        simdjson::error_code error;
        try {
            RETURN_IF_ERROR(_simdjson_parse_json(&size, is_empty_row, eof, &error));
            if (size == 0 || *eof) {
                *is_empty_row = true;
                return Status::OK();
            }
            Status st = _get_json_value(&size, eof, &error, is_empty_row);
            if (st.is<DATA_QUALITY_ERROR>()) {
                continue; // continue to read next
            }
            RETURN_IF_ERROR(st);
            if (*is_empty_row) {
                return Status::OK();
            }
            *is_empty_row = false;
            bool valid = true;
            if (_json_value.type() != simdjson::ondemand::json_type::object) {
                RETURN_IF_ERROR(_append_error_msg(nullptr, "Not object item", "", nullptr));
                continue;
            }
            cur = _json_value.get_object();
            st = _simdjson_write_columns_by_jsonpath(&cur, slot_descs, block, &valid);
            if (!st.ok()) {
                RETURN_IF_ERROR(_append_error_msg(nullptr, st.to_string(), "", nullptr));
                // Before continuing to process other rows, we need to first clean the fail parsed row.
                for (int i = 0; i < block.columns(); ++i) {
                    auto column = block.get_by_position(i).column->assume_mutable();
                    if (column->size() > num_rows) {
                        column->pop_back(column->size() - num_rows);
                    }
                }
                continue;
            }
            if (!valid) {
                // there is only one line in this case, so if it return false, just set is_empty_row true
                // so that the caller will continue reading next line.
                *is_empty_row = true;
            }
            break; // read a valid row
        } catch (simdjson::simdjson_error& e) {
            RETURN_IF_ERROR(_handle_simdjson_error(e, block, num_rows, eof));
            if (*_scanner_eof) {
                // When _scanner_eof is true and valid is false, it means that we have encountered
                // unqualified data and decided to stop the scan.
                *is_empty_row = true;
                return Status::OK();
            }
            continue;
        }
    }
    return Status::OK();
}

size_t NewJsonReader::_column_index(const StringRef& name, size_t key_index) {
    /// Optimization by caching the order of fields (which is almost always the same)
    /// and a quick check to match the next expected field, instead of searching the hash table.
    if (_prev_positions.size() > key_index && name == _prev_positions[key_index]->first) {
        return _prev_positions[key_index]->second;
    }
    auto it = _slot_desc_index.find(name);
    if (it != _slot_desc_index.end()) {
        if (key_index < _prev_positions.size()) {
            _prev_positions[key_index] = it;
        }
        return it->second;
    }
    return size_t(-1);
}

Status NewJsonReader::_simdjson_set_column_value(simdjson::ondemand::object* value, Block& block,
                                                 const std::vector<SlotDescriptor*>& slot_descs,
                                                 bool* valid) {
    // set
    _seen_columns.assign(block.columns(), false);
    size_t cur_row_count = block.rows();
    auto pop_current_row = [&]() {
        for (size_t index = 0; index < block.columns(); ++index) {
            auto column = block.get_by_position(index).column->assume_mutable();
            if (column->size() > cur_row_count) {
                DCHECK(column->size() == cur_row_count + 1);
                column->pop_back(column->size() - cur_row_count);
                DCHECK(column->size() == cur_row_count);
            }
        }
    };
    bool has_valid_value = false;
    // iterate through object, simdjson::ondemond will parsing on the fly
    size_t key_index = 0;

    for (auto field : *value) {
        std::string_view key = field.unescaped_key();
        StringRef name_ref(key.data(), key.size());
        std::string key_string;
        if (_is_hive_table) {
            key_string = name_ref.to_string();
            std::transform(key_string.begin(), key_string.end(), key_string.begin(), ::tolower);
            name_ref = StringRef(key_string);
        }
        const size_t column_index = _column_index(name_ref, key_index++);
        if (UNLIKELY(ssize_t(column_index) < 0)) {
            // This key is not exist in slot desc, just ignore
            continue;
        }
        if (skip_bitmap_col_idx >= 0 && std::cmp_equal(column_index, skip_bitmap_col_idx)) {
            continue;
        }
        if (_seen_columns[column_index]) {
            if (_is_hive_table) {
                //Since value can only be traversed once,
                // we can only insert the original value first, then delete it, and then reinsert the new value
                block.get_by_position(column_index).column->assume_mutable()->pop_back(1);
            } else {
                continue;
            }
        }
        simdjson::ondemand::value val = field.value();
        auto* column_ptr = block.get_by_position(column_index).column->assume_mutable().get();
        RETURN_IF_ERROR(_simdjson_write_data_to_column<false>(
                val, slot_descs[column_index]->type(), column_ptr,
                slot_descs[column_index]->col_name(), _serdes[column_index], valid,
                _is_flexible_variant_column(*slot_descs[column_index])));
        if (!(*valid)) {
            pop_current_row();
            return Status::OK();
        }
        _seen_columns[column_index] = true;
        has_valid_value = true;
    }

    if (!has_valid_value && _is_load) {
        std::string col_names;
        for (auto* slot_desc : slot_descs) {
            col_names.append(slot_desc->col_name() + ", ");
        }
        RETURN_IF_ERROR(_append_error_msg(value,
                                          "There is no column matching jsonpaths in the json file, "
                                          "columns:[{}], please check columns "
                                          "and jsonpaths:" +
                                                  _jsonpaths,
                                          col_names, valid));
        return Status::OK();
    }

    if (_should_process_skip_bitmap_col()) {
        _append_empty_skip_bitmap_value(block, cur_row_count);
    }

    // fill missing slot
    int nullcount = 0;
    for (size_t i = 0; i < slot_descs.size(); ++i) {
        if (_seen_columns[i]) {
            continue;
        }
        if (skip_bitmap_col_idx >= 0 && std::cmp_equal(i, skip_bitmap_col_idx)) {
            continue;
        }

        auto* slot_desc = slot_descs[i];
        auto* column_ptr = block.get_by_position(i).column->assume_mutable().get();

        // Quick path to insert default value, instead of using default values in the value map.
        if (!_should_process_skip_bitmap_col() &&
            (_col_default_value_map.empty() ||
             _col_default_value_map.find(slot_desc->col_name()) == _col_default_value_map.end())) {
            column_ptr->insert_default();
            continue;
        }
        if (column_ptr->size() < cur_row_count + 1) {
            DCHECK(column_ptr->size() == cur_row_count);
            if (_should_process_skip_bitmap_col()) {
                // not found, skip this column in flexible partial update
                if (slot_desc->is_key() && !slot_desc->is_auto_increment()) {
                    RETURN_IF_ERROR(
                            _append_error_msg(value,
                                              "The key columns can not be ommited in flexible "
                                              "partial update, missing key column: {}",
                                              slot_desc->col_name(), valid));
                    pop_current_row();
                    return Status::OK();
                }
                _set_skip_bitmap_mark(slot_desc, column_ptr, block, cur_row_count, valid);
                column_ptr->insert_default();
            } else {
                RETURN_IF_ERROR(_fill_missing_column(slot_desc, _serdes[i], column_ptr, valid));
                if (!(*valid)) {
                    return Status::OK();
                }
            }
            ++nullcount;
        }
        DCHECK(column_ptr->size() == cur_row_count + 1);
    }

    // There is at least one valid value here
    DCHECK(nullcount < block.columns());
    *valid = true;
    return Status::OK();
}

template <bool use_string_cache>
Status NewJsonReader::_simdjson_write_data_to_column(simdjson::ondemand::value& value,
                                                     const DataTypePtr& type_desc,
                                                     IColumn* column_ptr,
                                                     const std::string& column_name,
                                                     DataTypeSerDeSPtr serde, bool* valid,
                                                     bool is_flexible_variant_column) {
    ColumnNullable* nullable_column = nullptr;
    IColumn* data_column_ptr = column_ptr;
    DataTypeSerDeSPtr data_serde = serde;

    auto primitive_type = remove_nullable(type_desc)->get_primitive_type();
    const IColumn* nested_column_ptr =
            column_ptr->is_nullable()
                    ? assert_cast<const ColumnNullable&>(*column_ptr).get_nested_column_ptr().get()
                    : column_ptr;
    const bool is_flexible_variant_patch_column =
            _should_process_skip_bitmap_col() &&
            (primitive_type == TYPE_VARIANT ||
             check_and_get_column<ColumnVariant>(nested_column_ptr) != nullptr ||
             is_flexible_variant_column);
    if (is_flexible_variant_patch_column && value.type() != simdjson::ondemand::json_type::object) {
        if (_is_load) {
            RETURN_IF_ERROR(_append_error_msg(
                    nullptr,
                    "VARIANT flexible partial update only supports JSON object patch values", "",
                    valid));
            return Status::OK();
        }
        return Status::NotSupported(
                "VARIANT flexible partial update only supports JSON object patch values");
    }

    if (column_ptr->is_nullable()) {
        nullable_column = reinterpret_cast<ColumnNullable*>(column_ptr);

        data_column_ptr = nullable_column->get_nested_column().get_ptr().get();
        data_serde = serde->get_nested_serdes()[0];

        // kNullType will put 1 into the Null map, so there is no need to push 0 for kNullType.
        if (value.type() == simdjson::ondemand::json_type::null) {
            nullable_column->insert_default();
            *valid = true;
            return Status::OK();
        }
    } else if (value.type() == simdjson::ondemand::json_type::null) [[unlikely]] {
        if (_is_load) {
            RETURN_IF_ERROR(_append_error_msg(
                    nullptr, "Json value is null, but the column `{}` is not nullable.",
                    column_name, valid));
            return Status::OK();
        } else {
            return Status::DataQualityError(
                    "Json value is null, but the column `{}` is not nullable.", column_name);
        }
    }

    if (_is_load || !is_complex_type(primitive_type)) {
        if (is_flexible_variant_patch_column && primitive_type == TYPE_VARIANT) {
            ParseConfig parse_config;
            parse_config.check_duplicate_json_path =
                    config::variant_enable_duplicate_json_path_check;
            parse_config.reject_json_null_value = true;
            parse_config.record_empty_object_path = true;
            std::string_view json_str = simdjson::to_json_string(value);
            StringRef json_ref {json_str.data(), json_str.size()};
            try {
                variant_util::parse_json_to_variant(*data_column_ptr, json_ref, nullptr,
                                                    parse_config);
            } catch (const Exception& e) {
                return e.to_status();
            }
            if (nullable_column) {
                nullable_column->get_null_map_data().push_back(0);
            }
            *valid = true;
            return Status::OK();
        }
        if (value.type() == simdjson::ondemand::json_type::string) {
            std::string_view value_string;
            if constexpr (use_string_cache) {
                const auto cache_key = value.raw_json().value();
                if (_cached_string_values.contains(cache_key)) {
                    value_string = _cached_string_values[cache_key];
                } else {
                    value_string = value.get_string();
                    _cached_string_values.emplace(cache_key, value_string);
                }
            } else {
                DCHECK(_cached_string_values.empty());
                value_string = value.get_string();
            }

            Slice slice {value_string.data(), value_string.size()};
            RETURN_IF_ERROR(data_serde->deserialize_one_cell_from_json(*data_column_ptr, slice,
                                                                       _serde_options));

        } else if (value.type() == simdjson::ondemand::json_type::boolean) {
            const char* str_value = nullptr;
            // insert "1"/"0" , not "true"/"false".
            if (value.get_bool()) {
                str_value = (char*)"1";
            } else {
                str_value = (char*)"0";
            }
            Slice slice {str_value, 1};
            RETURN_IF_ERROR(data_serde->deserialize_one_cell_from_json(*data_column_ptr, slice,
                                                                       _serde_options));
        } else {
            // Maybe we can `switch (value->GetType()) case: kNumberType`.
            // Note that `if (value->IsInt())`, but column is FloatColumn.
            std::string_view json_str = simdjson::to_json_string(value);
            Slice slice {json_str.data(), json_str.size()};
            RETURN_IF_ERROR(data_serde->deserialize_one_cell_from_json(*data_column_ptr, slice,
                                                                       _serde_options));
        }
    } else if (primitive_type == TYPE_STRUCT) {
        if (value.type() != simdjson::ondemand::json_type::object) [[unlikely]] {
            return Status::DataQualityError(
                    "Json value isn't object, but the column `{}` is struct.", column_name);
        }

        const auto* type_struct =
                assert_cast<const DataTypeStruct*>(remove_nullable(type_desc).get());
        auto sub_col_size = type_struct->get_elements().size();
        simdjson::ondemand::object struct_value = value.get_object();
        auto sub_serdes = data_serde->get_nested_serdes();
        auto* struct_column_ptr = assert_cast<ColumnStruct*>(data_column_ptr);

        std::map<std::string, size_t> sub_col_name_to_idx;
        for (size_t sub_col_idx = 0; sub_col_idx < sub_col_size; sub_col_idx++) {
            sub_col_name_to_idx.emplace(type_struct->get_element_name(sub_col_idx), sub_col_idx);
        }
        std::vector<bool> has_value(sub_col_size, false);
        for (simdjson::ondemand::field sub : struct_value) {
            std::string_view sub_key_view = sub.unescaped_key();
            std::string sub_key(sub_key_view.data(), sub_key_view.length());
            std::transform(sub_key.begin(), sub_key.end(), sub_key.begin(), ::tolower);

            if (sub_col_name_to_idx.find(sub_key) == sub_col_name_to_idx.end()) [[unlikely]] {
                continue;
            }
            size_t sub_column_idx = sub_col_name_to_idx[sub_key];
            auto sub_column_ptr = struct_column_ptr->get_column(sub_column_idx).get_ptr();

            if (has_value[sub_column_idx]) [[unlikely]] {
                // Since struct_value can only be traversed once, we can only insert
                // the original value first, then delete it, and then reinsert the new value.
                sub_column_ptr->pop_back(1);
            }
            has_value[sub_column_idx] = true;

            const auto& sub_col_type = type_struct->get_element(sub_column_idx);
            RETURN_IF_ERROR(_simdjson_write_data_to_column<use_string_cache>(
                    sub.value(), sub_col_type, sub_column_ptr.get(), column_name + "." + sub_key,
                    sub_serdes[sub_column_idx], valid));
        }

        //fill missing subcolumn
        for (size_t sub_col_idx = 0; sub_col_idx < sub_col_size; sub_col_idx++) {
            if (has_value[sub_col_idx]) {
                continue;
            }

            auto sub_column_ptr = struct_column_ptr->get_column(sub_col_idx).get_ptr();
            if (sub_column_ptr->is_nullable()) {
                sub_column_ptr->insert_default();
                continue;
            } else [[unlikely]] {
                return Status::DataQualityError(
                        "Json file structColumn miss field {} and this column isn't nullable.",
                        column_name + "." + type_struct->get_element_name(sub_col_idx));
            }
        }
    } else if (primitive_type == TYPE_MAP) {
        if (value.type() != simdjson::ondemand::json_type::object) [[unlikely]] {
            return Status::DataQualityError("Json value isn't object, but the column `{}` is map.",
                                            column_name);
        }
        simdjson::ondemand::object object_value = value.get_object();

        auto sub_serdes = data_serde->get_nested_serdes();
        auto* map_column_ptr = assert_cast<ColumnMap*>(data_column_ptr);

        size_t field_count = 0;
        for (simdjson::ondemand::field member_value : object_value) {
            auto f = [](std::string_view key_view, const DataTypePtr& type_desc,
                        IColumn* column_ptr, DataTypeSerDeSPtr serde,
                        DataTypeSerDe::FormatOptions serde_options, bool* valid) {
                auto* data_column_ptr = column_ptr;
                auto data_serde = serde;
                if (column_ptr->is_nullable()) {
                    auto* nullable_column = static_cast<ColumnNullable*>(column_ptr);

                    nullable_column->get_null_map_data().push_back(0);
                    data_column_ptr = nullable_column->get_nested_column().get_ptr().get();
                    data_serde = serde->get_nested_serdes()[0];
                }
                Slice slice(key_view.data(), key_view.length());

                RETURN_IF_ERROR(data_serde->deserialize_one_cell_from_json(*data_column_ptr, slice,
                                                                           serde_options));
                return Status::OK();
            };

            RETURN_IF_ERROR(f(member_value.unescaped_key(),
                              assert_cast<const DataTypeMap*>(remove_nullable(type_desc).get())
                                      ->get_key_type(),
                              map_column_ptr->get_keys_ptr()->assume_mutable()->get_ptr().get(),
                              sub_serdes[0], _serde_options, valid));

            simdjson::ondemand::value field_value = member_value.value();
            RETURN_IF_ERROR(_simdjson_write_data_to_column<use_string_cache>(
                    field_value,
                    assert_cast<const DataTypeMap*>(remove_nullable(type_desc).get())
                            ->get_value_type(),
                    map_column_ptr->get_values_ptr()->assume_mutable()->get_ptr().get(),
                    column_name + ".value", sub_serdes[1], valid));
            field_count++;
        }

        auto& offsets = map_column_ptr->get_offsets();
        offsets.emplace_back(offsets.back() + field_count);

    } else if (primitive_type == TYPE_ARRAY) {
        if (value.type() != simdjson::ondemand::json_type::array) [[unlikely]] {
            return Status::DataQualityError("Json value isn't array, but the column `{}` is array.",
                                            column_name);
        }

        simdjson::ondemand::array array_value = value.get_array();

        auto sub_serdes = data_serde->get_nested_serdes();
        auto* array_column_ptr = assert_cast<ColumnArray*>(data_column_ptr);

        int field_count = 0;
        for (simdjson::ondemand::value sub_value : array_value) {
            RETURN_IF_ERROR(_simdjson_write_data_to_column<use_string_cache>(
                    sub_value,
                    assert_cast<const DataTypeArray*>(remove_nullable(type_desc).get())
                            ->get_nested_type(),
                    array_column_ptr->get_data().get_ptr().get(), column_name + ".element",
                    sub_serdes[0], valid));
            field_count++;
        }
        auto& offsets = array_column_ptr->get_offsets();
        offsets.emplace_back(offsets.back() + field_count);

    } else {
        return Status::InternalError("Not support load to complex column.");
    }
    //We need to finally set the nullmap of column_nullable to keep the size consistent with data_column
    if (nullable_column && value.type() != simdjson::ondemand::json_type::null) {
        nullable_column->get_null_map_data().push_back(0);
    }
    *valid = true;
    return Status::OK();
}

Status NewJsonReader::_append_error_msg(simdjson::ondemand::object* obj, std::string error_msg,
                                        std::string col_name, bool* valid) {
    std::string err_msg;
    if (!col_name.empty()) {
        fmt::memory_buffer error_buf;
        fmt::format_to(error_buf, error_msg, col_name, _jsonpaths);
        err_msg = fmt::to_string(error_buf);
    } else {
        err_msg = error_msg;
    }

    _counter->num_rows_filtered++;
    if (valid != nullptr) {
        // current row is invalid
        *valid = false;
    }

    RETURN_IF_ERROR(_state->append_error_msg_to_file(
            [&]() -> std::string {
                if (!obj) {
                    return "";
                }
                std::string_view str_view;
                (void)!obj->raw_json().get(str_view);
                return std::string(str_view.data(), str_view.size());
            },
            [&]() -> std::string { return err_msg; }));
    return Status::OK();
}

Status NewJsonReader::_simdjson_parse_json(size_t* size, bool* is_empty_row, bool* eof,
                                           simdjson::error_code* error) {
    SCOPED_TIMER(_read_timer);
    // step1: read buf from pipe.
    if (_line_reader != nullptr) {
        RETURN_IF_ERROR(_line_reader->read_line(&_json_str, size, eof, _io_ctx));
    } else {
        size_t length = 0;
        RETURN_IF_ERROR(_read_one_message(&_json_str_ptr, &length));
        _json_str = _json_str_ptr.get();
        *size = length;
        if (length == 0) {
            *eof = true;
        }
    }
    if (*eof) {
        return Status::OK();
    }

    // step2: init json parser iterate.
    if (*size + simdjson::SIMDJSON_PADDING > _padded_size) {
        // For efficiency reasons, simdjson requires a string with a few bytes (simdjson::SIMDJSON_PADDING) at the end.
        // Hence, a re-allocation is needed if the space is not enough.
        _simdjson_ondemand_padding_buffer.resize(*size + simdjson::SIMDJSON_PADDING);
        _simdjson_ondemand_unscape_padding_buffer.resize(*size + simdjson::SIMDJSON_PADDING);
        _padded_size = *size + simdjson::SIMDJSON_PADDING;
    }
    // trim BOM since simdjson does not handle UTF-8 Unicode (with BOM)
    if (*size >= 3 && static_cast<char>(_json_str[0]) == '\xEF' &&
        static_cast<char>(_json_str[1]) == '\xBB' && static_cast<char>(_json_str[2]) == '\xBF') {
        // skip the first three BOM bytes
        _json_str += 3;
        *size -= 3;
    }
    memcpy(&_simdjson_ondemand_padding_buffer.front(), _json_str, *size);
    _original_doc_size = *size;
    *error = _ondemand_json_parser
                     ->iterate(std::string_view(_simdjson_ondemand_padding_buffer.data(), *size),
                               _padded_size)
                     .get(_original_json_doc);
    return Status::OK();
}

Status NewJsonReader::_judge_empty_row(size_t size, bool eof, bool* is_empty_row) {
    if (size == 0 || eof) {
        *is_empty_row = true;
        return Status::OK();
    }

    if (!_parsed_jsonpaths.empty() && _strip_outer_array) {
        _total_rows = _json_value.count_elements().value();
        _next_row = 0;

        if (_total_rows == 0) {
            // meet an empty json array.
            *is_empty_row = true;
        }
    }
    return Status::OK();
}

Status NewJsonReader::_get_json_value(size_t* size, bool* eof, simdjson::error_code* error,
                                      bool* is_empty_row) {
    SCOPED_TIMER(_read_timer);
    auto return_quality_error = [&](fmt::memory_buffer& error_msg,
                                    const std::string& doc_info) -> Status {
        _counter->num_rows_filtered++;
        RETURN_IF_ERROR(_state->append_error_msg_to_file(
                [&]() -> std::string { return doc_info; },
                [&]() -> std::string { return fmt::to_string(error_msg); }));
        if (*_scanner_eof) {
            // Case A: if _scanner_eof is set to true in "append_error_msg_to_file", which means
            // we meet enough invalid rows and the scanner should be stopped.
            // So we set eof to true and return OK, the caller will stop the process as we meet the end of file.
            *eof = true;
            return Status::OK();
        }
        return Status::DataQualityError(fmt::to_string(error_msg));
    };
    if (*error != simdjson::error_code::SUCCESS) {
        fmt::memory_buffer error_msg;
        fmt::format_to(error_msg, "Parse json data for JsonDoc failed. code: {}, error info: {}",
                       *error, simdjson::error_message(*error));
        return return_quality_error(error_msg, std::string((char*)_json_str, *size));
    }
    auto type_res = _original_json_doc.type();
    if (type_res.error() != simdjson::error_code::SUCCESS) {
        fmt::memory_buffer error_msg;
        fmt::format_to(error_msg, "Parse json data for JsonDoc failed. code: {}, error info: {}",
                       type_res.error(), simdjson::error_message(type_res.error()));
        return return_quality_error(error_msg, std::string((char*)_json_str, *size));
    }
    simdjson::ondemand::json_type type = type_res.value();
    if (type != simdjson::ondemand::json_type::object &&
        type != simdjson::ondemand::json_type::array) {
        fmt::memory_buffer error_msg;
        fmt::format_to(error_msg, "Not an json object or json array");
        return return_quality_error(error_msg, std::string((char*)_json_str, *size));
    }
    if (!_parsed_json_root.empty() && type == simdjson::ondemand::json_type::object) {
        try {
            // set json root
            // if it is an array at top level, then we should iterate the entire array in
            // ::_simdjson_handle_flat_array_complex_json
            simdjson::ondemand::object object = _original_json_doc;
            Status st = JsonFunctions::extract_from_object(object, _parsed_json_root, &_json_value);
            if (!st.ok()) {
                fmt::memory_buffer error_msg;
                fmt::format_to(error_msg, "{}", st.to_string());
                return return_quality_error(error_msg, std::string((char*)_json_str, *size));
            }
            _parsed_from_json_root = true;
        } catch (simdjson::simdjson_error& e) {
            fmt::memory_buffer error_msg;
            fmt::format_to(error_msg, "Encounter error while extract_from_object, error: {}",
                           e.what());
            return return_quality_error(error_msg, std::string((char*)_json_str, *size));
        }
    } else {
        _json_value = _original_json_doc;
    }

    if (_json_value.type() == simdjson::ondemand::json_type::array && !_strip_outer_array) {
        fmt::memory_buffer error_msg;
        fmt::format_to(error_msg, "{}",
                       "JSON data is array-object, `strip_outer_array` must be TRUE.");
        return return_quality_error(error_msg, std::string((char*)_json_str, *size));
    }

    if (_json_value.type() != simdjson::ondemand::json_type::array && _strip_outer_array) {
        fmt::memory_buffer error_msg;
        fmt::format_to(error_msg, "{}",
                       "JSON data is not an array-object, `strip_outer_array` must be FALSE.");
        return return_quality_error(error_msg, std::string((char*)_json_str, *size));
    }
    RETURN_IF_ERROR(_judge_empty_row(*size, *eof, is_empty_row));
    return Status::OK();
}

Status NewJsonReader::_simdjson_write_columns_by_jsonpath(
        simdjson::ondemand::object* value, const std::vector<SlotDescriptor*>& slot_descs,
        Block& block, bool* valid) {
    // write by jsonpath
    bool has_valid_value = false;

    Defer clear_defer([this]() { _cached_string_values.clear(); });

    for (size_t i = 0; i < slot_descs.size(); i++) {
        auto* slot_desc = slot_descs[i];
        auto* column_ptr = block.get_by_position(i).column->assume_mutable().get();
        simdjson::ondemand::value json_value;
        Status st;
        if (i < _parsed_jsonpaths.size()) {
            st = JsonFunctions::extract_from_object(*value, _parsed_jsonpaths[i], &json_value);
            if (!st.ok() && !st.is<NOT_FOUND>()) {
                return st;
            }
        }
        if (i < _parsed_jsonpaths.size() && JsonFunctions::is_root_path(_parsed_jsonpaths[i])) {
            // Indicate that the jsonpath is "$" or "$.", read the full root json object, insert the original doc directly
            ColumnNullable* nullable_column = nullptr;
            IColumn* target_column_ptr = nullptr;
            if (slot_desc->is_nullable()) {
                nullable_column = assert_cast<ColumnNullable*>(column_ptr);
                target_column_ptr = &nullable_column->get_nested_column();
                nullable_column->get_null_map_data().push_back(0);
            }
            auto* column_string = assert_cast<ColumnString*>(target_column_ptr);
            column_string->insert_data(_simdjson_ondemand_padding_buffer.data(),
                                       _original_doc_size);
            has_valid_value = true;
        } else if (i >= _parsed_jsonpaths.size() || st.is<NOT_FOUND>()) {
            // not match in jsondata, filling with default value
            RETURN_IF_ERROR(_fill_missing_column(slot_desc, _serdes[i], column_ptr, valid));
            if (!(*valid)) {
                return Status::OK();
            }
        } else {
            RETURN_IF_ERROR(_simdjson_write_data_to_column<true>(json_value, slot_desc->type(),
                                                                 column_ptr, slot_desc->col_name(),
                                                                 _serdes[i], valid));
            if (!(*valid)) {
                return Status::OK();
            }
            has_valid_value = true;
        }
    }
    if (!has_valid_value) {
        // there is no valid value in json line but has filled with default value before
        // so remove this line in block
        std::string col_names;
        for (int i = 0; i < block.columns(); ++i) {
            auto column = block.get_by_position(i).column->assume_mutable();
            column->pop_back(1);
        }
        for (auto* slot_desc : slot_descs) {
            col_names.append(slot_desc->col_name() + ", ");
        }
        RETURN_IF_ERROR(_append_error_msg(value,
                                          "There is no column matching jsonpaths in the json file, "
                                          "columns:[{}], please check columns "
                                          "and jsonpaths:" +
                                                  _jsonpaths,
                                          col_names, valid));
        return Status::OK();
    }
    *valid = true;
    return Status::OK();
}

Status NewJsonReader::_get_column_default_value(
        const std::vector<SlotDescriptor*>& slot_descs,
        const std::unordered_map<std::string, VExprContextSPtr>& col_default_value_ctx) {
    for (auto* slot_desc : slot_descs) {
        auto it = col_default_value_ctx.find(slot_desc->col_name());
        if (it != col_default_value_ctx.end() && it->second != nullptr) {
            const auto& ctx = it->second;
            // NULL_LITERAL means no valid value of current column
            if (ctx->root()->node_type() == TExprNodeType::type::NULL_LITERAL) {
                continue;
            }
            ColumnWithTypeAndName result;
            RETURN_IF_ERROR(ctx->execute_const_expr(result));
            DCHECK(result.column->size() == 1);
            _col_default_value_map.emplace(slot_desc->col_name(),
                                           result.column->get_data_at(0).to_string());
        }
    }
    return Status::OK();
}

Status NewJsonReader::_fill_missing_column(SlotDescriptor* slot_desc, DataTypeSerDeSPtr serde,
                                           IColumn* column_ptr, bool* valid) {
    auto col_value = _col_default_value_map.find(slot_desc->col_name());
    if (col_value == _col_default_value_map.end()) {
        if (slot_desc->is_nullable()) {
            auto* nullable_column = static_cast<ColumnNullable*>(column_ptr);
            nullable_column->insert_default();
        } else {
            if (_is_load) {
                RETURN_IF_ERROR(_append_error_msg(
                        nullptr, "The column `{}` is not nullable, but it's not found in jsondata.",
                        slot_desc->col_name(), valid));
            } else {
                return Status::DataQualityError(
                        "The column `{}` is not nullable, but it's not found in jsondata.",
                        slot_desc->col_name());
            }
        }
    } else {
        const std::string& v_str = col_value->second;
        Slice column_default_value {v_str};
        RETURN_IF_ERROR(serde->deserialize_one_cell_from_json(*column_ptr, column_default_value,
                                                              _serde_options));
    }
    *valid = true;
    return Status::OK();
}

bool NewJsonReader::_is_flexible_variant_column(const SlotDescriptor& slot_desc) const {
    if (!_should_process_skip_bitmap_col()) {
        return false;
    }
    if (remove_nullable(slot_desc.type())->get_primitive_type() == TYPE_VARIANT) {
        return true;
    }
    DORIS_CHECK(_state != nullptr);
    DORIS_CHECK(_params.__isset.dest_tuple_id);
    const auto* dest_tuple_desc = _state->desc_tbl().get_tuple_descriptor(_params.dest_tuple_id);
    DORIS_CHECK(dest_tuple_desc != nullptr);
    for (const auto* dest_slot_desc : dest_tuple_desc->slots()) {
        if (dest_slot_desc->col_name() == slot_desc.col_name()) {
            return remove_nullable(dest_slot_desc->type())->get_primitive_type() == TYPE_VARIANT;
        }
    }
    return false;
}

void NewJsonReader::_append_empty_skip_bitmap_value(Block& block, size_t cur_row_count) const {
    auto* skip_bitmap_nullable_col_ptr = assert_cast<ColumnNullable*>(
            block.get_by_position(skip_bitmap_col_idx).column->assume_mutable().get());
    auto* skip_bitmap_col_ptr =
            assert_cast<ColumnBitmap*>(skip_bitmap_nullable_col_ptr->get_nested_column_ptr().get());
    DCHECK(skip_bitmap_nullable_col_ptr->size() == cur_row_count);
    // should append an empty bitmap for every row wheather this line misses columns
    skip_bitmap_nullable_col_ptr->get_null_map_data().push_back(0);
    skip_bitmap_col_ptr->insert_default();
    DCHECK(skip_bitmap_col_ptr->size() == cur_row_count + 1);
}

void NewJsonReader::_set_skip_bitmap_mark(SlotDescriptor* slot_desc, IColumn* column_ptr,
                                          Block& block, size_t cur_row_count, bool* valid) const {
    // we record the missing column's column unique id in skip bitmap
    // to indicate which columns need to do the alignment process
    auto* skip_bitmap_nullable_col_ptr = assert_cast<ColumnNullable*>(
            block.get_by_position(skip_bitmap_col_idx).column->assume_mutable().get());
    auto* skip_bitmap_col_ptr =
            assert_cast<ColumnBitmap*>(skip_bitmap_nullable_col_ptr->get_nested_column_ptr().get());
    DCHECK(skip_bitmap_col_ptr->size() == cur_row_count + 1);
    auto& skip_bitmap = skip_bitmap_col_ptr->get_data().back();
    skip_bitmap.add(slot_desc->col_unique_id());
}

void NewJsonReader::_collect_profile_before_close() {
    if (_line_reader != nullptr) {
        _line_reader->collect_profile_before_close();
    }
    if (_file_reader != nullptr) {
        _file_reader->collect_profile_before_close();
    }
}

} // namespace doris

Coverage Report

Created: 2026-05-17 04:12