//  Copyright (c) 2011-present, Facebook, Inc.  All rights reserved.
//  This source code is licensed under both the GPLv2 (found in the
//  COPYING file in the root directory) and Apache 2.0 License
//  (found in the LICENSE.Apache file in the root directory).

#ifndef ROCKSDB_LITE

#include "db/external_sst_file_ingestion_job.h"

#ifndef __STDC_FORMAT_MACROS
#define __STDC_FORMAT_MACROS
#endif

#include <inttypes.h>
#include <algorithm>
#include <string>
#include <vector>

#include "db/version_edit.h"
#include "table/merging_iterator.h"
#include "table/scoped_arena_iterator.h"
#include "table/sst_file_writer_collectors.h"
#include "table/table_builder.h"
#include "util/file_reader_writer.h"
#include "util/file_util.h"
#include "util/stop_watch.h"
#include "util/sync_point.h"

namespace rocksdb {

Status ExternalSstFileIngestionJob::Prepare(
    const std::vector<std::string>& external_files_paths) {
  Status status;

  // Read the information of files we are ingesting
  for (const std::string& file_path : external_files_paths) {
    IngestedFileInfo file_to_ingest;
    status = GetIngestedFileInfo(file_path, &file_to_ingest);
    if (!status.ok()) {
      return status;
    }
    files_to_ingest_.push_back(file_to_ingest);
  }

  for (const IngestedFileInfo& f : files_to_ingest_) {
    if (f.cf_id !=
            TablePropertiesCollectorFactory::Context::kUnknownColumnFamily &&
        f.cf_id != cfd_->GetID()) {
      return Status::InvalidArgument(
          "External file column family id dont match");
    }
  }

  const Comparator* ucmp = cfd_->internal_comparator().user_comparator();
  auto num_files = files_to_ingest_.size();
  if (num_files == 0) {
    return Status::InvalidArgument("The list of files is empty");
  } else if (num_files > 1) {
    // Verify that passed files dont have overlapping ranges
    autovector<const IngestedFileInfo*> sorted_files;
    for (size_t i = 0; i < num_files; i++) {
      sorted_files.push_back(&files_to_ingest_[i]);
    }

    std::sort(
        sorted_files.begin(), sorted_files.end(),
        [&ucmp](const IngestedFileInfo* info1, const IngestedFileInfo* info2) {
          return ucmp->Compare(info1->smallest_user_key,
                               info2->smallest_user_key) < 0;
        });

    for (size_t i = 0; i < num_files - 1; i++) {
      if (ucmp->Compare(sorted_files[i]->largest_user_key,
                        sorted_files[i + 1]->smallest_user_key) >= 0) {
        return Status::NotSupported("Files have overlapping ranges");
      }
    }
  }

  for (IngestedFileInfo& f : files_to_ingest_) {
    if (f.num_entries == 0) {
      return Status::InvalidArgument("File contain no entries");
    }

    if (!f.smallest_internal_key().Valid() ||
        !f.largest_internal_key().Valid()) {
      return Status::Corruption("Generated table have corrupted keys");
    }
  }

  // Copy/Move external files into DB
  for (IngestedFileInfo& f : files_to_ingest_) {
    f.fd = FileDescriptor(versions_->NewFileNumber(), 0, f.file_size);

    const std::string path_outside_db = f.external_file_path;
    const std::string path_inside_db =
        TableFileName(db_options_.db_paths, f.fd.GetNumber(), f.fd.GetPathId());

    if (ingestion_options_.move_files) {
      status = env_->LinkFile(path_outside_db, path_inside_db);
      if (status.IsNotSupported()) {
        // Original file is on a different FS, use copy instead of hard linking
        status = CopyFile(env_, path_outside_db, path_inside_db, 0,
                          db_options_.use_fsync);
      }
    } else {
      status = CopyFile(env_, path_outside_db, path_inside_db, 0,
                        db_options_.use_fsync);
    }
    TEST_SYNC_POINT("DBImpl::AddFile:FileCopied");
    if (!status.ok()) {
      break;
    }
    f.internal_file_path = path_inside_db;
  }

  if (!status.ok()) {
    // We failed, remove all files that we copied into the db
    for (IngestedFileInfo& f : files_to_ingest_) {
      if (f.internal_file_path == "") {
        break;
      }
      Status s = env_->DeleteFile(f.internal_file_path);
      if (!s.ok()) {
        ROCKS_LOG_WARN(db_options_.info_log,
                       "AddFile() clean up for file %s failed : %s",
                       f.internal_file_path.c_str(), s.ToString().c_str());
      }
    }
  }

  return status;
}

Status ExternalSstFileIngestionJob::NeedsFlush(bool* flush_needed) {
  SuperVersion* super_version = cfd_->GetSuperVersion();
  Status status =
      IngestedFilesOverlapWithMemtables(super_version, flush_needed);

  if (status.ok() && *flush_needed &&
      !ingestion_options_.allow_blocking_flush) {
    status = Status::InvalidArgument("External file requires flush");
  }
  return status;
}

// REQUIRES: we have become the only writer by entering both write_thread_ and
// nonmem_write_thread_
Status ExternalSstFileIngestionJob::Run() {
  Status status;
#ifndef NDEBUG
  // We should never run the job with a memtable that is overlapping
  // with the files we are ingesting
  bool need_flush = false;
  status = NeedsFlush(&need_flush);
  assert(status.ok() && need_flush == false);
#endif

  bool consumed_seqno = false;
  bool force_global_seqno = false;

  if (ingestion_options_.snapshot_consistency && !db_snapshots_->empty()) {
    // We need to assign a global sequence number to all the files even
    // if the dont overlap with any ranges since we have snapshots
    force_global_seqno = true;
  }
  // It is safe to use this instead of LastToBeWrittenSequence since we are
  // the only active writer, and hence they are equal
  const SequenceNumber last_seqno = versions_->LastSequence();
  SuperVersion* super_version = cfd_->GetSuperVersion();
  edit_.SetColumnFamily(cfd_->GetID());
  // The levels that the files will be ingested into

  for (IngestedFileInfo& f : files_to_ingest_) {
    SequenceNumber assigned_seqno = 0;
    if (ingestion_options_.ingest_behind) {
      status = CheckLevelForIngestedBehindFile(&f);
    } else {
      status = AssignLevelAndSeqnoForIngestedFile(
         super_version, force_global_seqno, cfd_->ioptions()->compaction_style,
         &f, &assigned_seqno);
    }
    if (!status.ok()) {
      return status;
    }
    status = AssignGlobalSeqnoForIngestedFile(&f, assigned_seqno);
    TEST_SYNC_POINT_CALLBACK("ExternalSstFileIngestionJob::Run",
                             &assigned_seqno);
    if (assigned_seqno == last_seqno + 1) {
      consumed_seqno = true;
    }
    if (!status.ok()) {
      return status;
    }
    edit_.AddFile(f.picked_level, f.fd.GetNumber(), f.fd.GetPathId(),
                  f.fd.GetFileSize(), f.smallest_internal_key(),
                  f.largest_internal_key(), f.assigned_seqno, f.assigned_seqno,
                  false);
  }

  if (consumed_seqno) {
    versions_->SetLastToBeWrittenSequence(last_seqno + 1);
    versions_->SetLastSequence(last_seqno + 1);
  }

  return status;
}

void ExternalSstFileIngestionJob::UpdateStats() {
  // Update internal stats for new ingested files
  uint64_t total_keys = 0;
  uint64_t total_l0_files = 0;
  uint64_t total_time = env_->NowMicros() - job_start_time_;
  for (IngestedFileInfo& f : files_to_ingest_) {
    InternalStats::CompactionStats stats(1);
    stats.micros = total_time;
    stats.bytes_written = f.fd.GetFileSize();
    stats.num_output_files = 1;
    cfd_->internal_stats()->AddCompactionStats(f.picked_level, stats);
    cfd_->internal_stats()->AddCFStats(InternalStats::BYTES_INGESTED_ADD_FILE,
                                       f.fd.GetFileSize());
    total_keys += f.num_entries;
    if (f.picked_level == 0) {
      total_l0_files += 1;
    }
    ROCKS_LOG_INFO(
        db_options_.info_log,
        "[AddFile] External SST file %s was ingested in L%d with path %s "
        "(global_seqno=%" PRIu64 ")\n",
        f.external_file_path.c_str(), f.picked_level,
        f.internal_file_path.c_str(), f.assigned_seqno);
  }
  cfd_->internal_stats()->AddCFStats(InternalStats::INGESTED_NUM_KEYS_TOTAL,
                                     total_keys);
  cfd_->internal_stats()->AddCFStats(InternalStats::INGESTED_NUM_FILES_TOTAL,
                                     files_to_ingest_.size());
  cfd_->internal_stats()->AddCFStats(
      InternalStats::INGESTED_LEVEL0_NUM_FILES_TOTAL, total_l0_files);
}

void ExternalSstFileIngestionJob::Cleanup(const Status& status) {
  if (!status.ok()) {
    // We failed to add the files to the database
    // remove all the files we copied
    for (IngestedFileInfo& f : files_to_ingest_) {
      Status s = env_->DeleteFile(f.internal_file_path);
      if (!s.ok()) {
        ROCKS_LOG_WARN(db_options_.info_log,
                       "AddFile() clean up for file %s failed : %s",
                       f.internal_file_path.c_str(), s.ToString().c_str());
      }
    }
  } else if (status.ok() && ingestion_options_.move_files) {
    // The files were moved and added successfully, remove original file links
    for (IngestedFileInfo& f : files_to_ingest_) {
      Status s = env_->DeleteFile(f.external_file_path);
      if (!s.ok()) {
        ROCKS_LOG_WARN(
            db_options_.info_log,
            "%s was added to DB successfully but failed to remove original "
            "file link : %s",
            f.external_file_path.c_str(), s.ToString().c_str());
      }
    }
  }
}

Status ExternalSstFileIngestionJob::GetIngestedFileInfo(
    const std::string& external_file, IngestedFileInfo* file_to_ingest) {
  file_to_ingest->external_file_path = external_file;

  // Get external file size
  Status status = env_->GetFileSize(external_file, &file_to_ingest->file_size);
  if (!status.ok()) {
    return status;
  }

  // Create TableReader for external file
  std::unique_ptr<TableReader> table_reader;
  std::unique_ptr<RandomAccessFile> sst_file;
  std::unique_ptr<RandomAccessFileReader> sst_file_reader;

  status = env_->NewRandomAccessFile(external_file, &sst_file, env_options_);
  if (!status.ok()) {
    return status;
  }
  sst_file_reader.reset(new RandomAccessFileReader(std::move(sst_file),
                                                   external_file));

  status = cfd_->ioptions()->table_factory->NewTableReader(
      TableReaderOptions(*cfd_->ioptions(), env_options_,
                         cfd_->internal_comparator()),
      std::move(sst_file_reader), file_to_ingest->file_size, &table_reader);
  if (!status.ok()) {
    return status;
  }

  // Get the external file properties
  auto props = table_reader->GetTableProperties();
  const auto& uprops = props->user_collected_properties;

  // Get table version
  auto version_iter = uprops.find(ExternalSstFilePropertyNames::kVersion);
  if (version_iter == uprops.end()) {
    return Status::Corruption("External file version not found");
  }
  file_to_ingest->version = DecodeFixed32(version_iter->second.c_str());

  auto seqno_iter = uprops.find(ExternalSstFilePropertyNames::kGlobalSeqno);
  if (file_to_ingest->version == 2) {
    // version 2 imply that we have global sequence number
    if (seqno_iter == uprops.end()) {
      return Status::Corruption(
          "External file global sequence number not found");
    }

    // Set the global sequence number
    file_to_ingest->original_seqno = DecodeFixed64(seqno_iter->second.c_str());
    file_to_ingest->global_seqno_offset = props->properties_offsets.at(
        ExternalSstFilePropertyNames::kGlobalSeqno);

    if (file_to_ingest->global_seqno_offset == 0) {
      return Status::Corruption("Was not able to find file global seqno field");
    }
  } else if (file_to_ingest->version == 1) {
    // SST file V1 should not have global seqno field
    assert(seqno_iter == uprops.end());
    file_to_ingest->original_seqno = 0;
    if (ingestion_options_.allow_blocking_flush ||
            ingestion_options_.allow_global_seqno) {
      return Status::InvalidArgument(
            "External SST file V1 does not support global seqno");
    }
  } else {
    return Status::InvalidArgument("External file version is not supported");
  }
  // Get number of entries in table
  file_to_ingest->num_entries = props->num_entries;

  ParsedInternalKey key;
  ReadOptions ro;
  // During reading the external file we can cache blocks that we read into
  // the block cache, if we later change the global seqno of this file, we will
  // have block in cache that will include keys with wrong seqno.
  // We need to disable fill_cache so that we read from the file without
  // updating the block cache.
  ro.fill_cache = false;
  std::unique_ptr<InternalIterator> iter(table_reader->NewIterator(ro));

  // Get first (smallest) key from file
  iter->SeekToFirst();
  if (!ParseInternalKey(iter->key(), &key)) {
    return Status::Corruption("external file have corrupted keys");
  }
  if (key.sequence != 0) {
    return Status::Corruption("external file have non zero sequence number");
  }
  file_to_ingest->smallest_user_key = key.user_key.ToString();

  // Get last (largest) key from file
  iter->SeekToLast();
  if (!ParseInternalKey(iter->key(), &key)) {
    return Status::Corruption("external file have corrupted keys");
  }
  if (key.sequence != 0) {
    return Status::Corruption("external file have non zero sequence number");
  }
  file_to_ingest->largest_user_key = key.user_key.ToString();

  file_to_ingest->cf_id = static_cast<uint32_t>(props->column_family_id);

  file_to_ingest->table_properties = *props;

  return status;
}

Status ExternalSstFileIngestionJob::IngestedFilesOverlapWithMemtables(
    SuperVersion* sv, bool* overlap) {
  // Create an InternalIterator over all memtables
  Arena arena;
  ReadOptions ro;
  ro.total_order_seek = true;
  MergeIteratorBuilder merge_iter_builder(&cfd_->internal_comparator(), &arena);
  merge_iter_builder.AddIterator(sv->mem->NewIterator(ro, &arena));
  sv->imm->AddIterators(ro, &merge_iter_builder);
  ScopedArenaIterator memtable_iter(merge_iter_builder.Finish());

  std::vector<InternalIterator*> memtable_range_del_iters;
  auto* active_range_del_iter = sv->mem->NewRangeTombstoneIterator(ro);
  if (active_range_del_iter != nullptr) {
    memtable_range_del_iters.push_back(active_range_del_iter);
  }
  sv->imm->AddRangeTombstoneIterators(ro, &memtable_range_del_iters);
  std::unique_ptr<InternalIterator> memtable_range_del_iter(NewMergingIterator(
      &cfd_->internal_comparator(),
      memtable_range_del_iters.empty() ? nullptr : &memtable_range_del_iters[0],
      static_cast<int>(memtable_range_del_iters.size())));

  Status status;
  *overlap = false;
  for (IngestedFileInfo& f : files_to_ingest_) {
    status =
        IngestedFileOverlapWithIteratorRange(&f, memtable_iter.get(), overlap);
    if (!status.ok() || *overlap == true) {
      break;
    }
    status = IngestedFileOverlapWithRangeDeletions(
        &f, memtable_range_del_iter.get(), overlap);
    if (!status.ok() || *overlap == true) {
      break;
    }
  }

  return status;
}

Status ExternalSstFileIngestionJob::AssignLevelAndSeqnoForIngestedFile(
    SuperVersion* sv, bool force_global_seqno, CompactionStyle compaction_style,
    IngestedFileInfo* file_to_ingest, SequenceNumber* assigned_seqno) {
  Status status;
  *assigned_seqno = 0;
  const SequenceNumber last_seqno = versions_->LastSequence();
  if (force_global_seqno) {
    *assigned_seqno = last_seqno + 1;
    if (compaction_style == kCompactionStyleUniversal) {
      file_to_ingest->picked_level = 0;
      return status;
    }
  }

  bool overlap_with_db = false;
  Arena arena;
  ReadOptions ro;
  ro.total_order_seek = true;
  int target_level = 0;
  auto* vstorage = cfd_->current()->storage_info();

  for (int lvl = 0; lvl < cfd_->NumberLevels(); lvl++) {
    if (lvl > 0 && lvl < vstorage->base_level()) {
      continue;
    }

    if (vstorage->NumLevelFiles(lvl) > 0) {
      bool overlap_with_level = false;
      status = IngestedFileOverlapWithLevel(sv, file_to_ingest, lvl,
        &overlap_with_level);
      if (!status.ok()) {
        return status;
      }
      if (overlap_with_level) {
        // We must use L0 or any level higher than `lvl` to be able to overwrite
        // the keys that we overlap with in this level, We also need to assign
        // this file a seqno to overwrite the existing keys in level `lvl`
        overlap_with_db = true;
        break;
      }

      if (compaction_style == kCompactionStyleUniversal && lvl != 0) {
        const std::vector<FileMetaData*>& level_files =
            vstorage->LevelFiles(lvl);
        const SequenceNumber level_largest_seqno =
            (*max_element(level_files.begin(), level_files.end(),
                          [](FileMetaData* f1, FileMetaData* f2) {
                            return f1->largest_seqno < f2->largest_seqno;
                          }))
                ->largest_seqno;
        if (level_largest_seqno != 0) {
          *assigned_seqno = level_largest_seqno;
        } else {
          continue;
        }
      }
    } else if (compaction_style == kCompactionStyleUniversal) {
      continue;
    }

    // We dont overlap with any keys in this level, but we still need to check
    // if our file can fit in it
    if (IngestedFileFitInLevel(file_to_ingest, lvl)) {
      target_level = lvl;
    }
  }
 TEST_SYNC_POINT_CALLBACK(
      "ExternalSstFileIngestionJob::AssignLevelAndSeqnoForIngestedFile",
      &overlap_with_db);
  file_to_ingest->picked_level = target_level;
  if (overlap_with_db && *assigned_seqno == 0) {
    *assigned_seqno = last_seqno + 1;
  }
  return status;
}

Status ExternalSstFileIngestionJob::CheckLevelForIngestedBehindFile(
    IngestedFileInfo* file_to_ingest) {
  auto* vstorage = cfd_->current()->storage_info();
  // first check if new files fit in the bottommost level
  int bottom_lvl = cfd_->NumberLevels() - 1;
  if(!IngestedFileFitInLevel(file_to_ingest, bottom_lvl)) {
    return Status::InvalidArgument(
      "Can't ingest_behind file as it doesn't fit "
      "at the bottommost level!");
  }

  // second check if despite allow_ingest_behind=true we still have 0 seqnums
  // at some upper level
  for (int lvl = 0; lvl < cfd_->NumberLevels() - 1; lvl++) {
    for (auto file : vstorage->LevelFiles(lvl)) {
      if (file->smallest_seqno == 0) {
        return Status::InvalidArgument(
          "Can't ingest_behind file as despite allow_ingest_behind=true "
          "there are files with 0 seqno in database at upper levels!");
      }
    }
  }

  file_to_ingest->picked_level = bottom_lvl;
  return Status::OK();
}

Status ExternalSstFileIngestionJob::AssignGlobalSeqnoForIngestedFile(
    IngestedFileInfo* file_to_ingest, SequenceNumber seqno) {
  if (file_to_ingest->original_seqno == seqno) {
    // This file already have the correct global seqno
    return Status::OK();
  } else if (!ingestion_options_.allow_global_seqno) {
    return Status::InvalidArgument("Global seqno is required, but disabled");
  } else if (file_to_ingest->global_seqno_offset == 0) {
    return Status::InvalidArgument(
        "Trying to set global seqno for a file that dont have a global seqno "
        "field");
  }

  std::unique_ptr<RandomRWFile> rwfile;
  Status status = env_->NewRandomRWFile(file_to_ingest->internal_file_path,
                                        &rwfile, env_options_);
  if (!status.ok()) {
    return status;
  }

  // Write the new seqno in the global sequence number field in the file
  std::string seqno_val;
  PutFixed64(&seqno_val, seqno);
  status = rwfile->Write(file_to_ingest->global_seqno_offset, seqno_val);
  if (status.ok()) {
    file_to_ingest->assigned_seqno = seqno;
  }
  return status;
}

Status ExternalSstFileIngestionJob::IngestedFileOverlapWithIteratorRange(
    const IngestedFileInfo* file_to_ingest, InternalIterator* iter,
    bool* overlap) {
  auto* vstorage = cfd_->current()->storage_info();
  auto* ucmp = vstorage->InternalComparator()->user_comparator();
  InternalKey range_start(file_to_ingest->smallest_user_key, kMaxSequenceNumber,
                          kValueTypeForSeek);
  iter->Seek(range_start.Encode());
  if (!iter->status().ok()) {
    return iter->status();
  }

  *overlap = false;
  if (iter->Valid()) {
    ParsedInternalKey seek_result;
    if (!ParseInternalKey(iter->key(), &seek_result)) {
      return Status::Corruption("DB have corrupted keys");
    }

    if (ucmp->Compare(seek_result.user_key, file_to_ingest->largest_user_key) <=
        0) {
      *overlap = true;
    }
  }

  return iter->status();
}

Status ExternalSstFileIngestionJob::IngestedFileOverlapWithRangeDeletions(
    const IngestedFileInfo* file_to_ingest, InternalIterator* range_del_iter,
    bool* overlap) {
  auto* vstorage = cfd_->current()->storage_info();
  auto* ucmp = vstorage->InternalComparator()->user_comparator();

  *overlap = false;
  if (range_del_iter != nullptr) {
    for (range_del_iter->SeekToFirst(); range_del_iter->Valid();
         range_del_iter->Next()) {
      ParsedInternalKey parsed_key;
      if (!ParseInternalKey(range_del_iter->key(), &parsed_key)) {
        return Status::Corruption("corrupted range deletion key: " +
                                  range_del_iter->key().ToString());
      }
      RangeTombstone range_del(parsed_key, range_del_iter->value());
      if (ucmp->Compare(range_del.start_key_,
                        file_to_ingest->largest_user_key) <= 0 &&
          ucmp->Compare(file_to_ingest->smallest_user_key,
                        range_del.end_key_) <= 0) {
        *overlap = true;
        break;
      }
    }
  }
  return Status::OK();
}

bool ExternalSstFileIngestionJob::IngestedFileFitInLevel(
    const IngestedFileInfo* file_to_ingest, int level) {
  if (level == 0) {
    // Files can always fit in L0
    return true;
  }

  auto* vstorage = cfd_->current()->storage_info();
  Slice file_smallest_user_key(file_to_ingest->smallest_user_key);
  Slice file_largest_user_key(file_to_ingest->largest_user_key);

  if (vstorage->OverlapInLevel(level, &file_smallest_user_key,
                               &file_largest_user_key)) {
    // File overlap with another files in this level, we cannot
    // add it to this level
    return false;
  }
  if (cfd_->RangeOverlapWithCompaction(file_smallest_user_key,
                                       file_largest_user_key, level)) {
    // File overlap with a running compaction output that will be stored
    // in this level, we cannot add this file to this level
    return false;
  }

  // File did not overlap with level files, our compaction output
  return true;
}

Status ExternalSstFileIngestionJob::IngestedFileOverlapWithLevel(
    SuperVersion* sv, IngestedFileInfo* file_to_ingest, int lvl,
    bool* overlap_with_level) {
  Arena arena;
  ReadOptions ro;
  ro.total_order_seek = true;
  MergeIteratorBuilder merge_iter_builder(&cfd_->internal_comparator(),
                                          &arena);
  sv->current->AddIteratorsForLevel(ro, env_options_, &merge_iter_builder, lvl,
                                    nullptr /* range_del_agg */);
  ScopedArenaIterator level_iter(merge_iter_builder.Finish());

  std::vector<InternalIterator*> level_range_del_iters;
  sv->current->AddRangeDelIteratorsForLevel(ro, env_options_, lvl,
                                            &level_range_del_iters);
  std::unique_ptr<InternalIterator> level_range_del_iter(NewMergingIterator(
      &cfd_->internal_comparator(),
      level_range_del_iters.empty() ? nullptr : &level_range_del_iters[0],
      static_cast<int>(level_range_del_iters.size())));

  Status status = IngestedFileOverlapWithIteratorRange(
      file_to_ingest, level_iter.get(), overlap_with_level);
  if (status.ok() && *overlap_with_level == false) {
    status = IngestedFileOverlapWithRangeDeletions(
        file_to_ingest, level_range_del_iter.get(), overlap_with_level);
  }
  return status;
}

}  // namespace rocksdb

#endif  // !ROCKSDB_LITE