// 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 "cloud/cloud_delete_task.h"

#include <gen_cpp/AgentService_types.h>
#include <thrift/protocol/TDebugProtocol.h>

#include "cloud/cloud_meta_mgr.h"
#include "cloud/cloud_storage_engine.h"
#include "cloud/cloud_tablet.h"
#include "cloud/cloud_tablet_mgr.h"
#include "common/logging.h"
#include "storage/delete/delete_handler.h"

namespace doris {
using namespace ErrorCode;

Status CloudDeleteTask::execute(CloudStorageEngine& engine, const TPushReq& request) {
    VLOG_DEBUG << "begin to process delete data. request=" << ThriftDebugString(request);

    if (!request.__isset.transaction_id) {
        return Status::InvalidArgument("transaction_id is not set");
    }

    auto tablet = DORIS_TRY(engine.tablet_mgr().get_tablet(request.tablet_id));

    if (!request.__isset.schema_version) {
        return Status::InternalError("No valid schema version in request, tablet_id={}",
                                     tablet->tablet_id());
    }

    using namespace std::chrono;
    tablet->last_load_time_ms =
            duration_cast<milliseconds>(system_clock::now().time_since_epoch()).count();
    // check if version number exceed limit

    int32_t max_version_config = tablet->max_version_config();
    if (tablet->fetch_add_approximate_num_rowsets(0) > max_version_config) {
        LOG_WARNING("tablet exceeds max version num limit")
                .tag("limit", max_version_config)
                .tag("tablet_id", tablet->tablet_id());
        return Status::Error<TOO_MANY_VERSION>(
                "too many versions, versions={} tablet={}. Please reduce the frequency of loading "
                "data or adjust the max_tablet_version_num or time_series_max_tablet_version_num "
                "in be.conf to a larger value.",
                max_version_config, tablet->tablet_id());
    }

    // check delete condition if push for delete
    DeletePredicatePB del_pred;
    auto tablet_schema = std::make_shared<TabletSchema>();
    // FIXME(plat1ko): Rewrite columns updating logic
    tablet_schema->update_tablet_columns(*tablet->tablet_schema(), request.columns_desc);
    tablet_schema->update_indexes_from_thrift(request.index_list);

    tablet_schema->set_schema_version(request.schema_version);
    RETURN_IF_ERROR(DeleteHandler::generate_delete_predicate(*tablet_schema,
                                                             request.delete_conditions, &del_pred));

    PUniqueId load_id;
    load_id.set_hi(0);
    load_id.set_lo(0);
    RowsetWriterContext context;
    context.storage_resource = engine.get_storage_resource(request.storage_vault_id);
    if (!context.storage_resource) {
        return Status::InternalError("vault id not found, maybe not sync, vault id {}",
                                     request.storage_vault_id);
    }

    context.txn_id = request.transaction_id;
    context.load_id = load_id;
    context.rowset_state = PREPARED;
    context.segments_overlap = OVERLAP_UNKNOWN;
    context.tablet_schema = tablet_schema;
    // ATTN: `request.timeout` is always 0 in current version, so we MUST ensure that the retention time of the
    //  recycler is much longer than the duration of the push task
    context.txn_expiration = ::time(nullptr) + request.timeout;
    auto rowset_writer = DORIS_TRY(tablet->create_rowset_writer(context, false));

    RowsetSharedPtr rowset;
    RETURN_IF_ERROR(rowset_writer->build(rowset));
    rowset->rowset_meta()->set_delete_predicate(std::move(del_pred));

    auto st =
            engine.meta_mgr().prepare_rowset(*rowset_writer->rowset_meta(), "", tablet->table_id());
    if (!st.ok()) {
        LOG(WARNING) << "failed to prepare rowset, status=" << st.to_string();
        return st;
    }

    st = engine.meta_mgr().commit_rowset(*rowset->rowset_meta(), "", tablet->table_id());
    if (!st.ok()) {
        LOG(WARNING) << "failed to commit rowset, status=" << st.to_string();
        return st;
    }

    // Update tablet stats
    tablet->fetch_add_approximate_num_rowsets(1);
    tablet->fetch_add_approximate_cumu_num_rowsets(1);

    // TODO(liaoxin) delete operator don't send calculate delete bitmap task from fe,
    //  then we don't need to set_txn_related_info here.
    if (tablet->enable_unique_key_merge_on_write()) {
        DeleteBitmapPtr delete_bitmap = std::make_shared<DeleteBitmap>(tablet->tablet_id());
        RowsetIdUnorderedSet rowset_ids;
        engine.txn_delete_bitmap_cache().set_tablet_txn_info(
                request.transaction_id, tablet->tablet_id(), delete_bitmap, rowset_ids, rowset,
                request.timeout, nullptr);
    } else {
        if (config::enable_cloud_make_rs_visible_on_be) {
            engine.meta_mgr().cache_committed_rowset(rowset->rowset_meta(), context.txn_expiration);
        }
    }

    return st;
}

} // namespace doris