// 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_v2/json/json_reader.h"

#include <rapidjson/document.h>

#include <algorithm>

#include <cstring>

#include <limits>

#include <map>

#include <string_view>

#include <utility>

#include "common/cast_set.h"

#include "core/assert_cast.h"

#include "core/block/block.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/data_type/data_type_array.h"

#include "core/data_type/data_type_map.h"

#include "core/data_type/data_type_nullable.h"

#include "core/data_type/data_type_string.h"

#include "core/data_type/data_type_struct.h"

#include "format/file_reader/new_plain_text_line_reader.h"

#include "format_v2/column_mapper.h"

#include "format_v2/materialized_reader_util.h"

#include "io/file_factory.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/file_scan_profile.h"

#include "runtime/runtime_state.h"

#include "util/decompressor.h"

#include "util/slice.h"

namespace doris::format::json {

namespace {

DataTypePtr json_file_type_from_slot_type(const DataTypePtr& type) {

    if (type == nullptr) {

        return nullptr;

    }

    // Text-like file readers expose CHAR/VARCHAR as STRING and let the table column mapper cast to

    // the destination slot type. JSON follows the same file-schema convention so that v2 mapping

    // behaves consistently across text formats.

    const bool is_nullable = type->is_nullable();

    const auto nested_type = remove_nullable(type);

    DataTypePtr file_type;

    switch (nested_type->get_primitive_type()) {

    case TYPE_CHAR:

    case TYPE_VARCHAR:

        file_type = std::make_shared<DataTypeString>();

        break;

    case TYPE_ARRAY: {

        const auto* array_type = assert_cast<const DataTypeArray*>(nested_type.get());

        file_type = std::make_shared<DataTypeArray>(

                json_file_type_from_slot_type(array_type->get_nested_type()));

        break;

    }

    case TYPE_MAP: {

        const auto* map_type = assert_cast<const DataTypeMap*>(nested_type.get());

        file_type = std::make_shared<DataTypeMap>(

                json_file_type_from_slot_type(map_type->get_key_type()),

                json_file_type_from_slot_type(map_type->get_value_type()));

        break;

    }

    case TYPE_STRUCT: {

        const auto* struct_type = assert_cast<const DataTypeStruct*>(nested_type.get());

        DataTypes file_children;

        file_children.reserve(struct_type->get_elements().size());

        for (const auto& child_type : struct_type->get_elements()) {

            file_children.push_back(json_file_type_from_slot_type(child_type));

        }

        file_type =

                std::make_shared<DataTypeStruct>(file_children, struct_type->get_element_names());

        break;

    }

    default:

        file_type = nested_type;

        break;

    }

    return is_nullable ? make_nullable(file_type) : file_type;

}

ColumnDefinition synthetic_file_child(const std::string& name, DataTypePtr type, int32_t local_id);

std::vector<ColumnDefinition> synthesize_file_children_from_type(const DataTypePtr& type) {

    std::vector<ColumnDefinition> children;

    if (type == nullptr) {

        return children;

    }

    const auto nested_type = remove_nullable(type);

    switch (nested_type->get_primitive_type()) {

    case TYPE_ARRAY: {

        const auto* array_type = assert_cast<const DataTypeArray*>(nested_type.get());

        children.push_back(synthetic_file_child("element", array_type->get_nested_type(), 0));

        break;

    }

    case TYPE_MAP: {

        const auto* map_type = assert_cast<const DataTypeMap*>(nested_type.get());

        children.push_back(synthetic_file_child("key", map_type->get_key_type(), 0));

        children.push_back(synthetic_file_child("value", map_type->get_value_type(), 1));

        break;

    }

    case TYPE_STRUCT: {

        const auto* struct_type = assert_cast<const DataTypeStruct*>(nested_type.get());

        children.reserve(struct_type->get_elements().size());

        for (size_t idx = 0; idx < struct_type->get_elements().size(); ++idx) {

            children.push_back(synthetic_file_child(struct_type->get_element_name(idx),

                                                    struct_type->get_element(idx),

                                                    cast_set<int32_t>(idx)));

        }

        break;

    }

    default:

        break;

    }

    return children;

}

ColumnDefinition synthetic_file_child(const std::string& name, DataTypePtr type, int32_t local_id) {

    ColumnDefinition child;

    child.identifier = Field::create_field<TYPE_STRING>(name);

    child.local_id = local_id;

    child.name = name;

    child.type = std::move(type);

    child.children = synthesize_file_children_from_type(child.type);

    return child;

}

std::string lower_key(std::string_view key) {

    std::string lowered(key.data(), key.size());

    std::transform(lowered.begin(), lowered.end(), lowered.begin(), ::tolower);

    return lowered;

}

} // namespace

JsonReader::JsonReader(std::shared_ptr<io::FileSystemProperties>& system_properties,

                       std::unique_ptr<io::FileDescription>& file_description,

                       std::shared_ptr<io::IOContext> io_ctx, RuntimeProfile* profile,

                       const TFileScanRangeParams* scan_params, const TFileRangeDesc& range,

                       const std::vector<SlotDescriptor*>& file_slot_descs,

                       TFileCompressType::type range_compress_type,

                       std::optional<TUniqueId> stream_load_id)

        : FileReader(system_properties, file_description, std::move(io_ctx), profile),

          _scan_params(scan_params),

          _range(range),

          _source_file_slot_descs(file_slot_descs),

          _range_compress_type(range_compress_type),

          _stream_load_id(std::move(stream_load_id)) {}

JsonReader::~JsonReader() {

    static_cast<void>(close());

}

void JsonReader::_init_profile() {

    if (_profile == nullptr) {

        return;

    }

    file_scan_profile::ensure_hierarchy(_profile);

    static const char* json_profile = "JsonReader";

    _total_time =

            ADD_CHILD_TIMER_WITH_LEVEL(_profile, json_profile, file_scan_profile::FILE_READER, 1);

    _open_time = ADD_CHILD_TIMER_WITH_LEVEL(_profile, "JsonOpenTime", json_profile, 1);

    _read_document_time =

            ADD_CHILD_TIMER_WITH_LEVEL(_profile, "JsonReadDocumentTime", json_profile, 1);

    _parse_time = ADD_CHILD_TIMER_WITH_LEVEL(_profile, "JsonParseTime", json_profile, 1);

    _materialize_time =

            ADD_CHILD_TIMER_WITH_LEVEL(_profile, "JsonMaterializeTime", json_profile, 1);

    _filter_time = ADD_CHILD_TIMER_WITH_LEVEL(_profile, "JsonFilterTime", json_profile, 1);

}

Status JsonReader::init(RuntimeState* state) {

    _init_profile();

    SCOPED_TIMER(_total_time);

    _runtime_state = state;

    if (_scan_params == nullptr) {

        return Status::InvalidArgument("JSON v2 reader requires scan params");

    }

    if (_file_description == nullptr) {

        return Status::InvalidArgument("JSON v2 reader requires file description");

    }

    if (_runtime_state == nullptr) {

        return Status::InvalidArgument("JSON v2 reader requires runtime state");

    }

    if (!_scan_params->__isset.file_attributes) {

        return Status::InvalidArgument("JSON v2 reader requires file attributes");

    }

    const auto& attributes = _scan_params->file_attributes;

    if (attributes.__isset.text_params && attributes.text_params.__isset.line_delimiter) {

        _line_delimiter = attributes.text_params.line_delimiter;

    } else {

        _line_delimiter = "\n";

    }

    _line_delimiter_length = _line_delimiter.size();

    _jsonpaths = attributes.__isset.jsonpaths ? attributes.jsonpaths : "";

    _json_root = attributes.__isset.json_root ? attributes.json_root : "";

    _read_json_by_line = attributes.__isset.read_json_by_line && attributes.read_json_by_line;

    _strip_outer_array = attributes.__isset.strip_outer_array && attributes.strip_outer_array;

    _num_as_string = attributes.__isset.num_as_string && attributes.num_as_string;

    _fuzzy_parse = attributes.__isset.fuzzy_parse && attributes.fuzzy_parse;

    _openx_json_ignore_malformed = attributes.__isset.openx_json_ignore_malformed &&

                                   attributes.openx_json_ignore_malformed;

    _is_hive_table = _range.table_format_params.table_format_type == "hive";

    _file_compress_type = _range_compress_type != TFileCompressType::UNKNOWN

                                  ? _range_compress_type

                                  : _scan_params->compress_type;

    _source_serdes = create_data_type_serdes(_source_file_slot_descs);

    _file_schema.clear();

    _file_schema.reserve(_source_file_slot_descs.size());

    // JSON has no physical footer schema. The FE file slots are therefore the authoritative schema

    // for both field names and source local ids.

    for (size_t idx = 0; idx < _source_file_slot_descs.size(); ++idx) {

        const auto* slot = _source_file_slot_descs[idx];

        DORIS_CHECK(slot != nullptr);

        ColumnDefinition field;

        field.identifier = Field::create_field<TYPE_STRING>(slot->col_name());

        field.local_id = cast_set<int32_t>(idx);

        field.name = slot->col_name();

        field.type = json_file_type_from_slot_type(slot->get_data_type_ptr());

        field.children = synthesize_file_children_from_type(field.type);

        _file_schema.push_back(std::move(field));

    }

    _eof = false;

    return Status::OK();

}

Status JsonReader::get_schema(std::vector<ColumnDefinition>* file_schema) const {

    SCOPED_TIMER(_total_time);

    if (file_schema == nullptr) {

        return Status::InvalidArgument("JSON v2 file_schema is null");

    }

    *file_schema = _file_schema;

    return Status::OK();

}

std::unique_ptr<TableColumnMapper> JsonReader::create_column_mapper(

        TableColumnMapperOptions options) const {

    return std::make_unique<MaterializedColumnMapper>(std::move(options));

}

Status JsonReader::open(std::shared_ptr<FileScanRequest> request) {

    SCOPED_TIMER(_total_time);

    SCOPED_TIMER(_open_time);

    RETURN_IF_ERROR(FileReader::open(std::move(request)));

    DORIS_CHECK(_request != nullptr);

    RETURN_IF_ERROR(_build_requested_columns(*_request, &_requested_columns));

    _slot_name_to_index.clear();

    _hive_slot_name_to_index.clear();

    _slot_name_to_index.reserve(_requested_columns.size());

    _hive_slot_name_to_index.reserve(_requested_columns.size());

    for (size_t idx = 0; idx < _requested_columns.size(); ++idx) {

        auto name = _requested_columns[idx].slot_desc->col_name();

        if (_is_hive_table) {

            _hive_slot_name_to_index.emplace(std::move(name), idx);

        } else {

            _slot_name_to_index.emplace(std::move(name), idx);

        }

    }

    _previous_positions.clear();

    _reader_range = _json_range();

    RETURN_IF_ERROR(_open_file_reader());

    RETURN_IF_ERROR(_create_decompressor());

    if (_read_json_by_line) {

        RETURN_IF_ERROR(_create_line_reader());

    }

    RETURN_IF_ERROR(_parse_jsonpath_and_json_root());

    _json_parser = std::make_unique<simdjson::ondemand::parser>();

    _padding_buffer.resize(_padded_size);

    _reader_eof = false;

    _single_document_read = false;

    _eof = false;

    return Status::OK();

}

Status JsonReader::get_block(Block* file_block, size_t* rows, bool* eof) {

    SCOPED_TIMER(_total_time);

    DORIS_CHECK(file_block != nullptr);

    DORIS_CHECK(rows != nullptr);

    DORIS_CHECK(eof != nullptr);

    if (_json_parser == nullptr || _physical_file_reader == nullptr) {

        return Status::InternalError("JSON v2 reader is not open");

    }

    const auto batch_size = _runtime_state->batch_size();

    const auto max_block_bytes = _runtime_state->preferred_block_size_bytes();

    *rows = 0;

    *eof = false;

    while (file_block->rows() < batch_size && !_reader_eof &&

           file_block->bytes() < max_block_bytes) {

        if (_read_json_by_line && _skip_first_line) {

            size_t skipped_size = 0;

            const uint8_t* skipped_line = nullptr;

            RETURN_IF_ERROR(_line_reader->read_line(&skipped_line, &skipped_size, &_reader_eof,

                                                    _io_ctx.get()));

            _skip_first_line = false;

            continue;

        }

        const size_t original_rows = file_block->rows();

        size_t size = 0;

        bool is_empty_row = false;

        Status st = Status::OK();

        try {

            {

                SCOPED_TIMER(_parse_time);

                st = _parse_next_json(&size, &_reader_eof);

            }

            if (st.ok() && !_reader_eof) {

                if (size == 0) {

                    is_empty_row = true;

                } else {

                    st = _extract_json_value(size, &_reader_eof, &is_empty_row);

                }

            }

            if (st.ok() && !_reader_eof && !is_empty_row) {

                SCOPED_TIMER(_materialize_time);

                st = _append_rows_from_current_value(file_block, &is_empty_row, &_reader_eof);

            }

        } catch (simdjson::simdjson_error& e) {

            st = Status::DataQualityError("Parse json data failed. code: {}, error info: {}",

                                          e.error(), e.what());

        }

        if (!st.ok()) {

            RETURN_IF_ERROR(_handle_json_error(st, file_block, original_rows, &is_empty_row));

        }

        // An ignored or empty JSON object can produce no row. Avoid spinning forever on a document

        // that was consumed but produced no materialized value.

        if (!is_empty_row && file_block->rows() == original_rows) {

            break;

        }

    }

    *rows = file_block->rows();

    _record_scan_rows(cast_set<int64_t>(*rows));

    {

        SCOPED_TIMER(_filter_time);

        RETURN_IF_ERROR(_apply_filters(file_block, rows));

    }

    *eof = _reader_eof && *rows == 0;

    _eof = *eof;

    return Status::OK();

}

Status JsonReader::close() {

    SCOPED_TIMER(_total_time);

    if (_line_reader != nullptr) {

        _line_reader->close();

        _line_reader.reset();

    }

    _json_parser.reset();

    _decompressor.reset();

    _physical_file_reader.reset();

    _tracing_file_reader.reset();

    _file_reader.reset();

    _requested_columns.clear();

    _slot_name_to_index.clear();

    _previous_positions.clear();

    _cached_string_values.clear();

    return Status::OK();

}

Status JsonReader::_build_requested_columns(const FileScanRequest& request,

                                            std::vector<RequestedColumn>* columns) const {

    DORIS_CHECK(columns != nullptr);

    columns->clear();

    // FileScanRequest stores a map from file-local id to output block position. Materialization is

    // position-driven, so normalize it into a dense vector ordered by block position while keeping

    // the original source index for jsonpaths.

    std::vector<RequestedColumn> by_position(request.local_positions.size());

    for (const auto& [file_column_id, block_position] : request.local_positions) {

        if (file_column_id.value() < 0 ||

            static_cast<size_t>(file_column_id.value()) >= _source_file_slot_descs.size()) {

            return Status::InvalidArgument("JSON v2 request references unknown local column id {}",

                                           file_column_id.value());

        }

        if (block_position.value() >= by_position.size()) {

            return Status::InvalidArgument("JSON v2 request has invalid block position {}",

                                           block_position.value());

        }

        const auto source_index = cast_set<size_t>(file_column_id.value());

        RequestedColumn requested_column;

        requested_column.file_column_id = file_column_id;

        requested_column.block_position = block_position;

        requested_column.source_index = source_index;

        requested_column.slot_desc = _source_file_slot_descs[source_index];

        requested_column.serde = _source_serdes[source_index];

        by_position[block_position.value()] = std::move(requested_column);

    }

    for (size_t pos = 0; pos < by_position.size(); ++pos) {

        if (!by_position[pos].file_column_id.is_valid()) {

            return Status::InvalidArgument("JSON v2 request misses block position {}", pos);

        }

    }

    *columns = std::move(by_position);

    return Status::OK();

}

TFileRangeDesc JsonReader::_json_range() const {

    auto range = _range;

    range.__set_path(_file_description->path);

    range.__set_start_offset(_file_description->range_start_offset);

    range.__set_size(_file_description->range_size);

    if (_file_description->file_size >= 0) {

        range.__set_file_size(_file_description->file_size);

    }

    if (!_file_description->fs_name.empty()) {

        range.__set_fs_name(_file_description->fs_name);

    }

    range.__set_file_cache_admission(_file_description->file_cache_admission);

    if (_range_compress_type != TFileCompressType::UNKNOWN) {

        range.__set_compress_type(_range_compress_type);

    }

    if (_stream_load_id.has_value()) {

        range.__set_load_id(*_stream_load_id);

    }

    return range;

}

Status JsonReader::_open_file_reader() {

    _current_offset = _reader_range.start_offset;

    if (_current_offset != 0) {

        --_current_offset;

    }

    if (_scan_params->file_type == TFileType::FILE_STREAM) {

        if (!_stream_load_id.has_value()) {

            return Status::InvalidArgument("JSON v2 stream reader requires load id");

        }

        RETURN_IF_ERROR(FileFactory::create_pipe_reader(*_stream_load_id, &_physical_file_reader,

                                                        _runtime_state, /*need_schema=*/false));

    } else {

        _file_description->mtime =

                _reader_range.__isset.modification_time ? _reader_range.modification_time : 0;

        auto reader_options = FileFactory::get_reader_options(_runtime_state->query_options(),

                                                              *_file_description);

        auto file_reader = DORIS_TRY(FileFactory::create_file_reader(

                *_system_properties, *_file_description, reader_options, _profile));

        _physical_file_reader =

                _io_ctx && _io_ctx->file_reader_stats

                        ? std::make_shared<io::TracingFileReader>(std::move(file_reader),

                                                                  _io_ctx->file_reader_stats)

                        : file_reader;

    }

    _file_reader = _physical_file_reader;

    _tracing_file_reader = _physical_file_reader;

    return Status::OK();

}

Status JsonReader::_create_decompressor() {

    return Decompressor::create_decompressor(_file_compress_type, &_decompressor);

}

Status JsonReader::_create_line_reader() {

    int64_t size = _reader_range.size;

    if (_reader_range.start_offset != 0) {

        // Start one byte earlier and discard the first partial line, matching split semantics used

        // by text readers.

        ++size;

        _skip_first_line = true;

    } else {

        _skip_first_line = false;

    }

    _line_reader = NewPlainTextLineReader::create_unique(

            _profile, _physical_file_reader, _decompressor.get(),

            std::make_shared<PlainTextLineReaderCtx>(_line_delimiter, _line_delimiter_length,

                                                     false),

            size, _current_offset);

    return Status::OK();

}

Status JsonReader::_parse_jsonpath_and_json_root() {

    _parsed_jsonpaths.clear();

    _parsed_json_root.clear();

    if (!_jsonpaths.empty()) {

        rapidjson::Document jsonpaths_doc;

        if (jsonpaths_doc.Parse(_jsonpaths.c_str(), _jsonpaths.length()).HasParseError() ||

            !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();

            if (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));

        }

    }

    if (!_json_root.empty()) {

        std::string json_root = _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 JsonReader::_read_one_document(size_t* size, bool* eof) {

    SCOPED_TIMER(_read_document_time);

    DORIS_CHECK(size != nullptr);

    DORIS_CHECK(eof != nullptr);

    *size = 0;

    *eof = false;

    if (_line_reader != nullptr) {

        const uint8_t* line = nullptr;

        RETURN_IF_ERROR(_line_reader->read_line(&line, size, eof, _io_ctx.get()));

        if (*eof) {

            return Status::OK();

        }

        // The line reader owns this span until the next read, and parsing copies it immediately

        // into the padded buffer. Borrowing it avoids a redundant line-sized string copy.

        _document_view = std::string_view(reinterpret_cast<const char*>(line), *size);

        return Status::OK();

    }

    // Non-line mode treats the split as one JSON document. This supports a single object or an

    // array with strip_outer_array=true.

    if (_single_document_read) {

        *eof = true;

        return Status::OK();

    }

    _single_document_read = true;

    if (_scan_params->file_type == TFileType::FILE_STREAM) {

        return _read_one_document_from_pipe(size);

    }

    auto read_size = _reader_range.size;

    if (read_size <= 0 && _reader_range.__isset.file_size) {

        read_size = _reader_range.file_size - _current_offset;

    }

    if (read_size <= 0) {

        *eof = true;

        return Status::OK();

    }

    _document_buffer.resize(cast_set<size_t>(read_size));

    Slice result(_document_buffer.data(), _document_buffer.size());

    RETURN_IF_ERROR(_physical_file_reader->read_at(_current_offset, result, size, _io_ctx.get()));

    _document_buffer.resize(*size);

    _document_view = _document_buffer;

    if (*size == 0) {

        *eof = true;

    }

    return Status::OK();

}

Status JsonReader::_read_one_document_from_pipe(size_t* read_size) {

    auto* stream_load_pipe = dynamic_cast<io::StreamLoadPipe*>(_physical_file_reader.get());

    if (stream_load_pipe == nullptr) {

        return Status::InternalError("JSON v2 stream reader requires StreamLoadPipe");

    }

    DorisUniqueBufferPtr<uint8_t> file_buf;

    RETURN_IF_ERROR(stream_load_pipe->read_one_message(&file_buf, read_size));

    _document_buffer.assign(reinterpret_cast<const char*>(file_buf.get()), *read_size);

    _document_view = _document_buffer;

    if (!stream_load_pipe->is_chunked_transfer()) {

        return Status::OK();

    }

    while (true) {

        DorisUniqueBufferPtr<uint8_t> next_buf;

        size_t next_size = 0;

        RETURN_IF_ERROR(stream_load_pipe->read_one_message(&next_buf, &next_size));

        if (next_size == 0) {

            break;

        }

        _document_buffer.append(reinterpret_cast<const char*>(next_buf.get()), next_size);

        *read_size += next_size;

    }

    _document_view = _document_buffer;

    return Status::OK();

}

Status JsonReader::_parse_next_json(size_t* size, bool* eof) {

    RETURN_IF_ERROR(_read_one_document(size, eof));

    if (*eof || *size == 0) {

        return Status::OK();

    }

    if (*size >= 3 && static_cast<unsigned char>(_document_view[0]) == 0xEF &&

        static_cast<unsigned char>(_document_view[1]) == 0xBB &&

        static_cast<unsigned char>(_document_view[2]) == 0xBF) {

        _document_view.remove_prefix(3);

        *size -= 3;

    }

    if (*size + simdjson::SIMDJSON_PADDING > _padded_size) {

        _padded_size = *size + simdjson::SIMDJSON_PADDING;

        _padding_buffer.resize(_padded_size);

    }

    // Ondemand values reference the input buffer. Keep the padded bytes in a member buffer until the

    // current document is fully materialized.

    std::memcpy(_padding_buffer.data(), _document_view.data(), *size);

    _original_doc_size = *size;

    const auto error =

            _json_parser->iterate(std::string_view(_padding_buffer.data(), *size), _padded_size)

                    .get(_original_json_doc);

    if (error != simdjson::error_code::SUCCESS) {

        return Status::DataQualityError(

                "Parse json data for JsonDoc failed. code: {}, error info: {}", error,

                simdjson::error_message(error));

    }

    return Status::OK();

}

Status JsonReader::_extract_json_value(size_t size, bool* eof, bool* is_empty_row) {

    DORIS_CHECK(eof != nullptr);

    DORIS_CHECK(is_empty_row != nullptr);

    *is_empty_row = false;

    if (size == 0 || *eof) {

        *is_empty_row = true;

        return Status::OK();

    }

    auto type_res = _original_json_doc.type();

    if (type_res.error() != simdjson::error_code::SUCCESS) {

        return Status::DataQualityError(

                "Parse json data for JsonDoc failed. code: {}, error info: {}", type_res.error(),

                simdjson::error_message(type_res.error()));

    }

    const auto type = type_res.value();

    if (type != simdjson::ondemand::json_type::object &&

        type != simdjson::ondemand::json_type::array) {

        return Status::DataQualityError("Not an json object or json array");

    }

    _parsed_from_json_root = false;

    if (!_parsed_json_root.empty() && type == simdjson::ondemand::json_type::object) {

        // In object mode json_root can be applied once here. In outer-array mode each array element

        // needs its own root extraction, which is handled while iterating the array.

        simdjson::ondemand::object object = _original_json_doc;

        Status st = JsonFunctions::extract_from_object(object, _parsed_json_root, &_json_value);

        if (!st.ok()) {

            return Status::DataQualityError("{}", st.to_string());

        }

        _parsed_from_json_root = true;

    } else {

        _json_value = _original_json_doc;

    }

    const auto value_type = _json_value.type().value();

    if (value_type == simdjson::ondemand::json_type::array && !_strip_outer_array) {

        return Status::DataQualityError(

                "JSON data is array-object, `strip_outer_array` must be TRUE.");

    }

    if (value_type != simdjson::ondemand::json_type::array && _strip_outer_array) {

        return Status::DataQualityError(

                "JSON data is not an array-object, `strip_outer_array` must be FALSE.");

    }

    if (!_parsed_jsonpaths.empty() && _strip_outer_array &&

        _json_value.count_elements().value() == 0) {

        *is_empty_row = true;

    }

    return Status::OK();

}

Status JsonReader::_append_rows_from_current_value(Block* block, bool* is_empty_row, bool* eof) {

    if (_parsed_jsonpaths.empty()) {

        return _append_simple_json_rows(block, is_empty_row, eof);

    }

    if (_strip_outer_array) {

        return _append_flat_array_jsonpath_rows(block, is_empty_row, eof);

    }

    return _append_nested_jsonpath_row(block, is_empty_row, eof);

}

Status JsonReader::_append_simple_json_rows(Block* block, bool* is_empty_row, bool* eof) {

    DORIS_CHECK(block != nullptr);

    DORIS_CHECK(is_empty_row != nullptr);

    DORIS_CHECK(eof != nullptr);

    bool valid = false;

    if (_json_value.type().value() == simdjson::ondemand::json_type::array) {

        _array = _json_value.get_array();

        if (_array.count_elements() == 0) {

            *is_empty_row = true;

            return Status::OK();

        }

        _array_iter = _array.begin();

        while (_array_iter != _array.end()) {

            simdjson::ondemand::object object_value = (*_array_iter).get_object();

            RETURN_IF_ERROR(_set_column_values_from_object(&object_value, block, &valid));

            ++_array_iter;

            if (!valid) {

                *is_empty_row = true;

                return Status::OK();

            }

        }

    } else {

        simdjson::ondemand::object object_value = _json_value.get_object();

        RETURN_IF_ERROR(_set_column_values_from_object(&object_value, block, &valid));

        if (!valid) {

            *is_empty_row = true;

            return Status::OK();

        }

    }

    *is_empty_row = false;

    return Status::OK();

}

Status JsonReader::_append_flat_array_jsonpath_rows(Block* block, bool* is_empty_row, bool* eof) {

    DORIS_CHECK(block != nullptr);

    DORIS_CHECK(is_empty_row != nullptr);

    DORIS_CHECK(eof != nullptr);

    const size_t original_rows = block->rows();

    bool valid = true;

    _array = _json_value.get_array();

    _array_iter = _array.begin();

    while (_array_iter != _array.end()) {

        simdjson::ondemand::object object_value = (*_array_iter).get_object();

        if (!_parsed_from_json_root && !_parsed_json_root.empty()) {

            // For strip_outer_array, json_root is evaluated against each element. Elements without

            // the requested root do not produce rows, matching the load reader behavior.

            simdjson::ondemand::value rooted_value;

            Status st = JsonFunctions::extract_from_object(object_value, _parsed_json_root,

                                                           &rooted_value);

            if (!st.ok()) {

                if (st.is<ErrorCode::NOT_FOUND>()) {

                    ++_array_iter;

                    continue;

                }

                return st;

            }

            if (rooted_value.type().value() != simdjson::ondemand::json_type::object) {

                ++_array_iter;

                continue;

            }

            object_value = rooted_value.get_object();

        }

        RETURN_IF_ERROR(_write_columns_by_jsonpath(&object_value, block, &valid));

        ++_array_iter;

    }

    *is_empty_row = block->rows() == original_rows;

    return Status::OK();

}

Status JsonReader::_append_nested_jsonpath_row(Block* block, bool* is_empty_row, bool* eof) {

    DORIS_CHECK(block != nullptr);

    DORIS_CHECK(is_empty_row != nullptr);

    DORIS_CHECK(eof != nullptr);

    if (_json_value.type().value() != simdjson::ondemand::json_type::object) {

        return Status::DataQualityError("Not object item");

    }

    bool valid = true;

    simdjson::ondemand::object object_value = _json_value.get_object();

    RETURN_IF_ERROR(_write_columns_by_jsonpath(&object_value, block, &valid));

    *is_empty_row = !valid;

    return Status::OK();

}

Status JsonReader::_set_column_values_from_object(simdjson::ondemand::object* object_value,

                                                  Block* block, bool* valid) {

    DORIS_CHECK(object_value != nullptr);

    DORIS_CHECK(block != nullptr);

    DORIS_CHECK(valid != nullptr);

    std::vector<bool> seen_columns(block->columns(), false);

    const size_t cur_row_count = block->rows();

    bool has_valid_value = false;

    size_t key_index = 0;

    for (auto field : *object_value) {

        std::string_view key = field.unescaped_key().value();

        const size_t column_index = _column_index(key, key_index++);

        if (column_index == static_cast<size_t>(-1)) {

            continue;

        }

        if (seen_columns[column_index]) {

            if (_is_hive_table) {

                // Hive JSON keeps the last duplicate key ignoring case. The earlier value has

                // already been appended, so remove it before writing the replacement.

                _pop_back_last_inserted_value(block, column_index);

            } else {

                continue;

            }

        }

        simdjson::ondemand::value value = field.value().value();

        const auto& requested = _requested_columns[column_index];

        auto* column_ptr = block->get_by_position(column_index).column->assert_mutable().get();

        RETURN_IF_ERROR(_write_data_to_column<false>(

                value, requested.slot_desc->get_data_type_ptr(), column_ptr,

                requested.slot_desc->col_name(), requested.serde, valid));

        if (!*valid) {

            return Status::OK();

        }

        seen_columns[column_index] = true;

        has_valid_value = true;

    }

    for (size_t i = 0; i < _requested_columns.size(); ++i) {

        if (seen_columns[i]) {

            continue;

        }

        auto* column_ptr = block->get_by_position(i).column->assert_mutable().get();

        RETURN_IF_ERROR(_fill_missing_column(_requested_columns[i], column_ptr, valid));

        if (!*valid) {

            _truncate_block_to_rows(block, cur_row_count);

            return Status::OK();

        }

    }

    *valid = true;

    if (!has_valid_value) {

        return Status::OK();

    }

    return Status::OK();

}

Status JsonReader::_write_columns_by_jsonpath(simdjson::ondemand::object* object_value,

                                              Block* block, bool* valid) {

    DORIS_CHECK(object_value != nullptr);

    DORIS_CHECK(block != nullptr);

    DORIS_CHECK(valid != nullptr);

    bool has_valid_value = false;

    const size_t cur_row_count = block->rows();

    _cached_string_values.clear();

    for (size_t i = 0; i < _requested_columns.size(); ++i) {

        const auto& requested = _requested_columns[i];

        auto* column_ptr = block->get_by_position(i).column->assert_mutable().get();

        simdjson::ondemand::value json_value;

        Status st = Status::OK();

        if (requested.source_index < _parsed_jsonpaths.size()) {

            st = JsonFunctions::extract_from_object(

                    *object_value, _parsed_jsonpaths[requested.source_index], &json_value);

            if (!st.ok() && !st.is<ErrorCode::NOT_FOUND>()) {

                return st;

            }

        }

        if (_is_root_path_for_column(requested)) {

            // A root jsonpath means "materialize the whole current JSON document" instead of a

            // field under it. Use the original bytes so callers receive the same document text.

            if (is_column_nullable(*column_ptr)) {

                auto* nullable_column = assert_cast<ColumnNullable*>(column_ptr);

                nullable_column->get_null_map_data().push_back(0);

                auto* column_string =

                        assert_cast<ColumnString*>(nullable_column->get_nested_column_ptr().get());

                column_string->insert_data(_padding_buffer.data(), _original_doc_size);

            } else {

                auto* column_string = assert_cast<ColumnString*>(column_ptr);

                column_string->insert_data(_padding_buffer.data(), _original_doc_size);

            }

            has_valid_value = true;

        } else if (requested.source_index >= _parsed_jsonpaths.size() ||

                   st.is<ErrorCode::NOT_FOUND>()) {

            RETURN_IF_ERROR(_fill_missing_column(requested, column_ptr, valid));

            if (!*valid) {

                _truncate_block_to_rows(block, cur_row_count);

                return Status::OK();

            }

        } else {

            RETURN_IF_ERROR(_write_data_to_column<true>(

                    json_value, requested.slot_desc->get_data_type_ptr(), column_ptr,

                    requested.slot_desc->col_name(), requested.serde, valid));

            if (!*valid) {

                _truncate_block_to_rows(block, cur_row_count);

                return Status::OK();

            }

            has_valid_value = true;

        }

    }

    if (!has_valid_value) {

        // jsonpaths can legally match nothing. Roll the row back so an all-missing path set does

        // not create a synthetic row of nulls.

        _truncate_block_to_rows(block, cur_row_count);

        *valid = false;

        return Status::OK();

    }

    *valid = true;

    return Status::OK();

}

template <bool use_string_cache>

Status JsonReader::_write_data_to_column(simdjson::ondemand::value& value,

                                         const DataTypePtr& type_desc, IColumn* column_ptr,

                                         const std::string& column_name,

                                         const DataTypeSerDeSPtr& serde, bool* valid) {

    ColumnNullable* nullable_column = nullptr;

    IColumn* data_column_ptr = column_ptr;

    DataTypeSerDeSPtr data_serde = serde;

    const auto value_type = value.type().value();

    if (is_column_nullable(*column_ptr)) {

        nullable_column = assert_cast<ColumnNullable*>(column_ptr);

        data_column_ptr = nullable_column->get_nested_column().get_ptr().get();

        if (type_desc->is_nullable()) {

            data_serde = serde->get_nested_serdes()[0];

        }

        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) {

        return Status::DataQualityError("Json value is null, but the column `{}` is not nullable.",

                                        column_name);

    }

    const auto primitive_type = type_desc->get_primitive_type();

    if (!is_complex_type(primitive_type)) {

        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 {

                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 = value.get_bool() ? "1" : "0";

            Slice slice {str_value, 1};

            RETURN_IF_ERROR(data_serde->deserialize_one_cell_from_json(*data_column_ptr, slice,

                                                                       _serde_options));

        } else {

            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) {

            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* struct_column_ptr = assert_cast<ColumnStruct*>(data_column_ptr);

        const auto sub_serdes = data_serde->get_nested_serdes();

        std::map<std::string, size_t> sub_col_name_to_idx;

        for (size_t sub_col_idx = 0; sub_col_idx < type_struct->get_elements().size();

             ++sub_col_idx) {

            sub_col_name_to_idx.emplace(lower_key(type_struct->get_element_name(sub_col_idx)),

                                        sub_col_idx);

        }

        std::vector<bool> has_value(type_struct->get_elements().size(), false);

        simdjson::ondemand::object struct_value = value.get_object();

        for (auto sub : struct_value) {

            const auto sub_key = lower_key(sub.unescaped_key().value());