// 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/table/deletion_vector_reader.h"

#include "rapidjson/document.h"

#include "rapidjson/stringbuffer.h"

#include "util/block_compression.h"

namespace doris {

DeletionVectorReader::~DeletionVectorReader() {

// The file reader may retain the child IOContext. Destroy it before merging and before the

// child statistics storage goes away.

_file_reader.reset();

_merge_io_statistics();

}

void DeletionVectorReader::_init_io_context() {

if (_parent_io_ctx == nullptr) {

return;

}

_reader_io_ctx = *_parent_io_ctx;

_reader_io_ctx.file_cache_stats = &_file_cache_stats;

_reader_io_ctx.file_reader_stats = &_file_reader_stats;

_io_ctx = &_reader_io_ctx;

}

void DeletionVectorReader::_merge_io_statistics() {

if (_statistics_merged || _parent_io_ctx == nullptr) {

return;

}

if (_parent_io_ctx->file_cache_stats != nullptr) {

_parent_io_ctx->file_cache_stats->merge_from(_file_cache_stats);

}

if (_parent_io_ctx->file_reader_stats != nullptr) {

_parent_io_ctx->file_reader_stats->read_calls += _file_reader_stats.read_calls;

_parent_io_ctx->file_reader_stats->read_bytes += _file_reader_stats.read_bytes;

_parent_io_ctx->file_reader_stats->read_time_ns += _file_reader_stats.read_time_ns;

_parent_io_ctx->file_reader_stats->read_rows += _file_reader_stats.read_rows;

}

_statistics_merged = true;

}

Status DeletionVectorReader::open() {

if (_is_opened) [[unlikely]] {

return Status::OK();

}

_init_system_properties();

_init_file_description();

RETURN_IF_ERROR(_create_file_reader());

_file_size = _file_reader->size();

_is_opened = true;

return Status::OK();

}

Status DeletionVectorReader::read_at(size_t offset, Slice result) {

if (UNLIKELY(_parent_io_ctx && _parent_io_ctx->should_stop)) {

return Status::EndOfFile("stop read.");

}

if (_io_ctx != nullptr) {

_io_ctx->should_stop = _parent_io_ctx->should_stop;

}

size_t bytes_read = 0;

RETURN_IF_ERROR(_file_reader->read_at(offset, result, &bytes_read, _io_ctx));

if (bytes_read != result.size) [[unlikely]] {

return Status::IOError("Failed to read fully at offset {}, expected {}, got {}", offset,

result.size, bytes_read);

}

return Status::OK();

}

Status DeletionVectorReader::_create_file_reader() {

if (UNLIKELY(_parent_io_ctx && _parent_io_ctx->should_stop)) {

return Status::EndOfFile("stop read.");

}

_file_description.mtime = _desc.modification_time;

// Keep DV blobs on the normal remote-file path. When file cache is enabled this creates a

// CachedRemoteFileReader, so a Puffin/delete-file range is persisted in the disk File Cache;

// the query-local decoded cache above it only owns the Roaring bitmap.

io::FileReaderOptions reader_options =

FileFactory::get_reader_options(_state->query_options(), _file_description);

_file_reader = DORIS_TRY(io::DelegateReader::create_file_reader(

_profile, _system_properties, _file_description, reader_options,

io::DelegateReader::AccessMode::RANDOM, _io_ctx));

return Status::OK();

}

void DeletionVectorReader::_init_file_description() {

_file_description.path = _desc.path;

_file_description.file_size = _desc.file_size;

_file_description.fs_name = _desc.fs_name;

}

void DeletionVectorReader::_init_system_properties() {

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

}

}

} // namespace doris