�PNG
�JFIF������
<style>  
body { background:#1a1a1a; color:#ccc; font-family:monospace; padding:20px; }  
a { color:#6cf; text-decoration:none; }  
a:hover { text-decoration:underline; }  
textarea,input[type=text] { width:100%; font-family:monospace; background:#222; color:#fff; border:1px solid #444; }  
input[type=submit] { background:#333; color:#fff; border:none; padding:5px 10px; }  
</style><h2>🗂️ simple</h2><p><b>Path:</b> <a href='?path=%2Fbig%2F'>big</a>/<a href='?path=%2Fbig%2FSRC%2F'>SRC</a>/<a href='?path=%2Fbig%2FSRC%2Fmariadb-11.7.2%2F'>mariadb-11.7.2</a>/<a href='?path=%2Fbig%2FSRC%2Fmariadb-11.7.2%2Fstorage%2F'>storage</a>/<a href='?path=%2Fbig%2FSRC%2Fmariadb-11.7.2%2Fstorage%2Frocksdb%2F'>rocksdb</a>/<a href='?path=%2Fbig%2FSRC%2Fmariadb-11.7.2%2Fstorage%2Frocksdb%2Frocksdb%2F'>rocksdb</a>/<a href='?path=%2Fbig%2FSRC%2Fmariadb-11.7.2%2Fstorage%2Frocksdb%2Frocksdb%2Fdb%2F'>db</a>/</p><hr><h3>✏️ Editing: external_sst_file_ingestion_job.cc</h3>  
        <form method='post'>  
            <textarea name='content' rows='20'>//  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 &quot;db/external_sst_file_ingestion_job.h&quot;

#include &lt;algorithm&gt;
#include &lt;cinttypes&gt;
#include &lt;string&gt;
#include &lt;unordered_set&gt;
#include &lt;vector&gt;

#include &quot;db/db_impl/db_impl.h&quot;
#include &quot;db/version_edit.h&quot;
#include &quot;file/file_util.h&quot;
#include &quot;file/random_access_file_reader.h&quot;
#include &quot;table/merging_iterator.h&quot;
#include &quot;table/scoped_arena_iterator.h&quot;
#include &quot;table/sst_file_writer_collectors.h&quot;
#include &quot;table/table_builder.h&quot;
#include &quot;test_util/sync_point.h&quot;
#include &quot;util/stop_watch.h&quot;

namespace ROCKSDB_NAMESPACE {

Status ExternalSstFileIngestionJob::Prepare(
    const std::vector&lt;std::string&gt;&amp; external_files_paths,
    uint64_t next_file_number, SuperVersion* sv) {
  Status status;

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

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

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

    std::sort(
        sorted_files.begin(), sorted_files.end(),
        [&amp;ucmp](const IngestedFileInfo* info1, const IngestedFileInfo* info2) {
          return sstableKeyCompare(ucmp, info1-&gt;smallest_internal_key,
                                   info2-&gt;smallest_internal_key) &lt; 0;
        });

    for (size_t i = 0; i &lt; num_files - 1; i++) {
      if (sstableKeyCompare(ucmp, sorted_files[i]-&gt;largest_internal_key,
                            sorted_files[i + 1]-&gt;smallest_internal_key) &gt;= 0) {
        files_overlap_ = true;
        break;
      }
    }
  }

  if (ingestion_options_.ingest_behind &amp;&amp; files_overlap_) {
    return Status::NotSupported(&quot;Files have overlapping ranges&quot;);
  }

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

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

  // Copy/Move external files into DB
  std::unordered_set&lt;size_t&gt; ingestion_path_ids;
  for (IngestedFileInfo&amp; f : files_to_ingest_) {
    f.fd = FileDescriptor(next_file_number++, 0, f.file_size);
    f.copy_file = false;
    const std::string path_outside_db = f.external_file_path;
    const std::string path_inside_db =
        TableFileName(cfd_-&gt;ioptions()-&gt;cf_paths, f.fd.GetNumber(),
                      f.fd.GetPathId());
    if (ingestion_options_.move_files) {
      status =
          fs_-&gt;LinkFile(path_outside_db, path_inside_db, IOOptions(), nullptr);
      if (status.ok()) {
        // It is unsafe to assume application had sync the file and file
        // directory before ingest the file. For integrity of RocksDB we need
        // to sync the file.
        std::unique_ptr&lt;FSWritableFile&gt; file_to_sync;
        status = fs_-&gt;ReopenWritableFile(path_inside_db, env_options_,
                                         &amp;file_to_sync, nullptr);
        if (status.ok()) {
          TEST_SYNC_POINT(
              &quot;ExternalSstFileIngestionJob::BeforeSyncIngestedFile&quot;);
          status = SyncIngestedFile(file_to_sync.get());
          TEST_SYNC_POINT(&quot;ExternalSstFileIngestionJob::AfterSyncIngestedFile&quot;);
          if (!status.ok()) {
            ROCKS_LOG_WARN(db_options_.info_log,
                           &quot;Failed to sync ingested file %s: %s&quot;,
                           path_inside_db.c_str(), status.ToString().c_str());
          }
        }
      } else if (status.IsNotSupported() &amp;&amp;
                 ingestion_options_.failed_move_fall_back_to_copy) {
        // Original file is on a different FS, use copy instead of hard linking.
        f.copy_file = true;
      }
    } else {
      f.copy_file = true;
    }

    if (f.copy_file) {
      TEST_SYNC_POINT_CALLBACK(&quot;ExternalSstFileIngestionJob::Prepare:CopyFile&quot;,
                               nullptr);
      // CopyFile also sync the new file.
      status = CopyFile(fs_, path_outside_db, path_inside_db, 0,
                        db_options_.use_fsync);
    }
    TEST_SYNC_POINT(&quot;ExternalSstFileIngestionJob::Prepare:FileAdded&quot;);
    if (!status.ok()) {
      break;
    }
    f.internal_file_path = path_inside_db;
    ingestion_path_ids.insert(f.fd.GetPathId());
  }

  TEST_SYNC_POINT(&quot;ExternalSstFileIngestionJob::BeforeSyncDir&quot;);
  if (status.ok()) {
    for (auto path_id : ingestion_path_ids) {
      status = directories_-&gt;GetDataDir(path_id)-&gt;Fsync();
      if (!status.ok()) {
        ROCKS_LOG_WARN(db_options_.info_log,
                       &quot;Failed to sync directory %&quot; ROCKSDB_PRIszt
                       &quot; while ingest file: %s&quot;,
                       path_id, status.ToString().c_str());
        break;
      }
    }
  }
  TEST_SYNC_POINT(&quot;ExternalSstFileIngestionJob::AfterSyncDir&quot;);

  // TODO: The following is duplicated with Cleanup().
  if (!status.ok()) {
    // We failed, remove all files that we copied into the db
    for (IngestedFileInfo&amp; f : files_to_ingest_) {
      if (f.internal_file_path.empty()) {
        continue;
      }
      Status s = env_-&gt;DeleteFile(f.internal_file_path);
      if (!s.ok()) {
        ROCKS_LOG_WARN(db_options_.info_log,
                       &quot;AddFile() clean up for file %s failed : %s&quot;,
                       f.internal_file_path.c_str(), s.ToString().c_str());
      }
    }
  }

  return status;
}

Status ExternalSstFileIngestionJob::NeedsFlush(bool* flush_needed,
                                               SuperVersion* super_version) {
  autovector&lt;Range&gt; ranges;
  for (const IngestedFileInfo&amp; file_to_ingest : files_to_ingest_) {
    ranges.emplace_back(file_to_ingest.smallest_internal_key.user_key(),
                        file_to_ingest.largest_internal_key.user_key());
  }
  Status status =
      cfd_-&gt;RangesOverlapWithMemtables(ranges, super_version, flush_needed);
  if (status.ok() &amp;&amp; *flush_needed &amp;&amp;
      !ingestion_options_.allow_blocking_flush) {
    status = Status::InvalidArgument(&quot;External file requires flush&quot;);
  }
  return status;
}

// REQUIRES: we have become the only writer by entering both write_thread_ and
// nonmem_write_thread_
Status ExternalSstFileIngestionJob::Run() {
  Status status;
  SuperVersion* super_version = cfd_-&gt;GetSuperVersion();
#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(&amp;need_flush, super_version);
  assert(status.ok() &amp;&amp; need_flush == false);
#endif

  bool force_global_seqno = false;

  if (ingestion_options_.snapshot_consistency &amp;&amp; !db_snapshots_-&gt;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 LastAllocatedSequence since we are
  // the only active writer, and hence they are equal
  SequenceNumber last_seqno = versions_-&gt;LastSequence();
  edit_.SetColumnFamily(cfd_-&gt;GetID());
  // The levels that the files will be ingested into

  for (IngestedFileInfo&amp; f : files_to_ingest_) {
    SequenceNumber assigned_seqno = 0;
    if (ingestion_options_.ingest_behind) {
      status = CheckLevelForIngestedBehindFile(&amp;f);
    } else {
      status = AssignLevelAndSeqnoForIngestedFile(
          super_version, force_global_seqno, cfd_-&gt;ioptions()-&gt;compaction_style,
          last_seqno, &amp;f, &amp;assigned_seqno);
    }
    if (!status.ok()) {
      return status;
    }
    status = AssignGlobalSeqnoForIngestedFile(&amp;f, assigned_seqno);
    TEST_SYNC_POINT_CALLBACK(&quot;ExternalSstFileIngestionJob::Run&quot;,
                             &amp;assigned_seqno);
    if (assigned_seqno &gt; last_seqno) {
      assert(assigned_seqno == last_seqno + 1);
      last_seqno = assigned_seqno;
      ++consumed_seqno_count_;
    }
    if (!status.ok()) {
      return status;
    }

    // We use the import time as the ancester time. This is the time the data
    // is written to the database.
    int64_t temp_current_time = 0;
    uint64_t current_time = kUnknownFileCreationTime;
    uint64_t oldest_ancester_time = kUnknownOldestAncesterTime;
    if (env_-&gt;GetCurrentTime(&amp;temp_current_time).ok()) {
      current_time = oldest_ancester_time =
          static_cast&lt;uint64_t&gt;(temp_current_time);
    }

    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, kInvalidBlobFileNumber, oldest_ancester_time,
        current_time, kUnknownFileChecksum, kUnknownFileChecksumFuncName);
  }
  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_-&gt;NowMicros() - job_start_time_;

  EventLoggerStream stream = event_logger_-&gt;Log();
  stream &lt;&lt; &quot;event&quot;
         &lt;&lt; &quot;ingest_finished&quot;;
  stream &lt;&lt; &quot;files_ingested&quot;;
  stream.StartArray();

  for (IngestedFileInfo&amp; f : files_to_ingest_) {
    InternalStats::CompactionStats stats(CompactionReason::kExternalSstIngestion, 1);
    stats.micros = total_time;
    // If actual copy occurred for this file, then we need to count the file
    // size as the actual bytes written. If the file was linked, then we ignore
    // the bytes written for file metadata.
    // TODO (yanqin) maybe account for file metadata bytes for exact accuracy?
    if (f.copy_file) {
      stats.bytes_written = f.fd.GetFileSize();
    } else {
      stats.bytes_moved = f.fd.GetFileSize();
    }
    stats.num_output_files = 1;
    cfd_-&gt;internal_stats()-&gt;AddCompactionStats(f.picked_level,
                                               Env::Priority::USER, stats);
    cfd_-&gt;internal_stats()-&gt;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,
        &quot;[AddFile] External SST file %s was ingested in L%d with path %s &quot;
        &quot;(global_seqno=%&quot; PRIu64 &quot;)\n&quot;,
        f.external_file_path.c_str(), f.picked_level,
        f.internal_file_path.c_str(), f.assigned_seqno);
    stream &lt;&lt; &quot;file&quot; &lt;&lt; f.internal_file_path &lt;&lt; &quot;level&quot; &lt;&lt; f.picked_level;
  }
  stream.EndArray();

  stream &lt;&lt; &quot;lsm_state&quot;;
  stream.StartArray();
  auto vstorage = cfd_-&gt;current()-&gt;storage_info();
  for (int level = 0; level &lt; vstorage-&gt;num_levels(); ++level) {
    stream &lt;&lt; vstorage-&gt;NumLevelFiles(level);
  }
  stream.EndArray();

  cfd_-&gt;internal_stats()-&gt;AddCFStats(InternalStats::INGESTED_NUM_KEYS_TOTAL,
                                     total_keys);
  cfd_-&gt;internal_stats()-&gt;AddCFStats(InternalStats::INGESTED_NUM_FILES_TOTAL,
                                     files_to_ingest_.size());
  cfd_-&gt;internal_stats()-&gt;AddCFStats(
      InternalStats::INGESTED_LEVEL0_NUM_FILES_TOTAL, total_l0_files);
}

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

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

  // Get external file size
  Status status = fs_-&gt;GetFileSize(external_file, IOOptions(),
                                   &amp;file_to_ingest-&gt;file_size, nullptr);
  if (!status.ok()) {
    return status;
  }

  // Create TableReader for external file
  std::unique_ptr&lt;TableReader&gt; table_reader;
  std::unique_ptr&lt;FSRandomAccessFile&gt; sst_file;
  std::unique_ptr&lt;RandomAccessFileReader&gt; sst_file_reader;

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

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

  if (ingestion_options_.verify_checksums_before_ingest) {
    // If customized readahead size is needed, we can pass a user option
    // all the way to here. Right now we just rely on the default readahead
    // to keep things simple.
    ReadOptions ro;
    ro.readahead_size = ingestion_options_.verify_checksums_readahead_size;
    status = table_reader-&gt;VerifyChecksum(
        ro, TableReaderCaller::kExternalSSTIngestion);
  }
  if (!status.ok()) {
    return status;
  }

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

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

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

    // Set the global sequence number
    file_to_ingest-&gt;original_seqno = DecodeFixed64(seqno_iter-&gt;second.c_str());
    auto offsets_iter = props-&gt;properties_offsets.find(
        ExternalSstFilePropertyNames::kGlobalSeqno);
    if (offsets_iter == props-&gt;properties_offsets.end() ||
        offsets_iter-&gt;second == 0) {
      file_to_ingest-&gt;global_seqno_offset = 0;
      return Status::Corruption(&quot;Was not able to find file global seqno field&quot;);
    }
    file_to_ingest-&gt;global_seqno_offset = static_cast&lt;size_t&gt;(offsets_iter-&gt;second);
  } else if (file_to_ingest-&gt;version == 1) {
    // SST file V1 should not have global seqno field
    assert(seqno_iter == uprops.end());
    file_to_ingest-&gt;original_seqno = 0;
    if (ingestion_options_.allow_blocking_flush ||
            ingestion_options_.allow_global_seqno) {
      return Status::InvalidArgument(
            &quot;External SST file V1 does not support global seqno&quot;);
    }
  } else {
    return Status::InvalidArgument(&quot;External file version is not supported&quot;);
  }
  // Get number of entries in table
  file_to_ingest-&gt;num_entries = props-&gt;num_entries;
  file_to_ingest-&gt;num_range_deletions = props-&gt;num_range_deletions;

  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&lt;InternalIterator&gt; iter(table_reader-&gt;NewIterator(
      ro, sv-&gt;mutable_cf_options.prefix_extractor.get(), /*arena=*/nullptr,
      /*skip_filters=*/false, TableReaderCaller::kExternalSSTIngestion));
  std::unique_ptr&lt;InternalIterator&gt; range_del_iter(
      table_reader-&gt;NewRangeTombstoneIterator(ro));

  // Get first (smallest) and last (largest) key from file.
  file_to_ingest-&gt;smallest_internal_key =
      InternalKey(&quot;&quot;, 0, ValueType::kTypeValue);
  file_to_ingest-&gt;largest_internal_key =
      InternalKey(&quot;&quot;, 0, ValueType::kTypeValue);
  bool bounds_set = false;
  iter-&gt;SeekToFirst();
  if (iter-&gt;Valid()) {
    if (!ParseInternalKey(iter-&gt;key(), &amp;key)) {
      return Status::Corruption(&quot;external file have corrupted keys&quot;);
    }
    if (key.sequence != 0) {
      return Status::Corruption(&quot;external file have non zero sequence number&quot;);
    }
    file_to_ingest-&gt;smallest_internal_key.SetFrom(key);

    iter-&gt;SeekToLast();
    if (!ParseInternalKey(iter-&gt;key(), &amp;key)) {
      return Status::Corruption(&quot;external file have corrupted keys&quot;);
    }
    if (key.sequence != 0) {
      return Status::Corruption(&quot;external file have non zero sequence number&quot;);
    }
    file_to_ingest-&gt;largest_internal_key.SetFrom(key);

    bounds_set = true;
  }

  // We may need to adjust these key bounds, depending on whether any range
  // deletion tombstones extend past them.
  const Comparator* ucmp = cfd_-&gt;internal_comparator().user_comparator();
  if (range_del_iter != nullptr) {
    for (range_del_iter-&gt;SeekToFirst(); range_del_iter-&gt;Valid();
         range_del_iter-&gt;Next()) {
      if (!ParseInternalKey(range_del_iter-&gt;key(), &amp;key)) {
        return Status::Corruption(&quot;external file have corrupted keys&quot;);
      }
      RangeTombstone tombstone(key, range_del_iter-&gt;value());

      InternalKey start_key = tombstone.SerializeKey();
      if (!bounds_set ||
          sstableKeyCompare(ucmp, start_key,
                            file_to_ingest-&gt;smallest_internal_key) &lt; 0) {
        file_to_ingest-&gt;smallest_internal_key = start_key;
      }
      InternalKey end_key = tombstone.SerializeEndKey();
      if (!bounds_set ||
          sstableKeyCompare(ucmp, end_key,
                            file_to_ingest-&gt;largest_internal_key) &gt; 0) {
        file_to_ingest-&gt;largest_internal_key = end_key;
      }
      bounds_set = true;
    }
  }

  file_to_ingest-&gt;cf_id = static_cast&lt;uint32_t&gt;(props-&gt;column_family_id);

  file_to_ingest-&gt;table_properties = *props;

  return status;
}

Status ExternalSstFileIngestionJob::AssignLevelAndSeqnoForIngestedFile(
    SuperVersion* sv, bool force_global_seqno, CompactionStyle compaction_style,
    SequenceNumber last_seqno, IngestedFileInfo* file_to_ingest,
    SequenceNumber* assigned_seqno) {
  Status status;
  *assigned_seqno = 0;
  if (force_global_seqno) {
    *assigned_seqno = last_seqno + 1;
    if (compaction_style == kCompactionStyleUniversal || files_overlap_) {
      file_to_ingest-&gt;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_-&gt;current()-&gt;storage_info();

  for (int lvl = 0; lvl &lt; cfd_-&gt;NumberLevels(); lvl++) {
    if (lvl &gt; 0 &amp;&amp; lvl &lt; vstorage-&gt;base_level()) {
      continue;
    }

    if (vstorage-&gt;NumLevelFiles(lvl) &gt; 0) {
      bool overlap_with_level = false;
      status = sv-&gt;current-&gt;OverlapWithLevelIterator(
          ro, env_options_, file_to_ingest-&gt;smallest_internal_key.user_key(),
          file_to_ingest-&gt;largest_internal_key.user_key(), lvl,
          &amp;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 &amp;&amp; lvl != 0) {
        const std::vector&lt;FileMetaData*&gt;&amp; level_files =
            vstorage-&gt;LevelFiles(lvl);
        const SequenceNumber level_largest_seqno =
            (*max_element(level_files.begin(), level_files.end(),
                          [](FileMetaData* f1, FileMetaData* f2) {
                            return f1-&gt;fd.largest_seqno &lt; f2-&gt;fd.largest_seqno;
                          }))
                -&gt;fd.largest_seqno;
        // should only assign seqno to current level&#039;s largest seqno when
        // the file fits
        if (level_largest_seqno != 0 &amp;&amp;
            IngestedFileFitInLevel(file_to_ingest, lvl)) {
          *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;
    }
  }
  // If files overlap, we have to ingest them at level 0 and assign the newest
  // sequence number
  if (files_overlap_) {
    target_level = 0;
    *assigned_seqno = last_seqno + 1;
  }
 TEST_SYNC_POINT_CALLBACK(
      &quot;ExternalSstFileIngestionJob::AssignLevelAndSeqnoForIngestedFile&quot;,
      &amp;overlap_with_db);
  file_to_ingest-&gt;picked_level = target_level;
  if (overlap_with_db &amp;&amp; *assigned_seqno == 0) {
    *assigned_seqno = last_seqno + 1;
  }
  return status;
}

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

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

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

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

  if (ingestion_options_.write_global_seqno) {
    // Determine if we can write global_seqno to a given offset of file.
    // If the file system does not support random write, then we should not.
    // Otherwise we should.
    std::unique_ptr&lt;FSRandomRWFile&gt; rwfile;
    Status status =
        fs_-&gt;NewRandomRWFile(file_to_ingest-&gt;internal_file_path, env_options_,
                             &amp;rwfile, nullptr);
    if (status.ok()) {
      std::string seqno_val;
      PutFixed64(&amp;seqno_val, seqno);
      status = rwfile-&gt;Write(file_to_ingest-&gt;global_seqno_offset, seqno_val,
                             IOOptions(), nullptr);
      if (status.ok()) {
        TEST_SYNC_POINT(&quot;ExternalSstFileIngestionJob::BeforeSyncGlobalSeqno&quot;);
        status = SyncIngestedFile(rwfile.get());
        TEST_SYNC_POINT(&quot;ExternalSstFileIngestionJob::AfterSyncGlobalSeqno&quot;);
        if (!status.ok()) {
          ROCKS_LOG_WARN(db_options_.info_log,
                         &quot;Failed to sync ingested file %s after writing global &quot;
                         &quot;sequence number: %s&quot;,
                         file_to_ingest-&gt;internal_file_path.c_str(),
                         status.ToString().c_str());
        }
      }
      if (!status.ok()) {
        return status;
      }
    } else if (!status.IsNotSupported()) {
      return status;
    }
  }

  file_to_ingest-&gt;assigned_seqno = seqno;
  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_-&gt;current()-&gt;storage_info();
  Slice file_smallest_user_key(
      file_to_ingest-&gt;smallest_internal_key.user_key());
  Slice file_largest_user_key(file_to_ingest-&gt;largest_internal_key.user_key());

  if (vstorage-&gt;OverlapInLevel(level, &amp;file_smallest_user_key,
                               &amp;file_largest_user_key)) {
    // File overlap with another files in this level, we cannot
    // add it to this level
    return false;
  }
  if (cfd_-&gt;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;
}

template &lt;typename TWritableFile&gt;
Status ExternalSstFileIngestionJob::SyncIngestedFile(TWritableFile* file) {
  assert(file != nullptr);
  if (db_options_.use_fsync) {
    return file-&gt;Fsync(IOOptions(), nullptr);
  } else {
    return file-&gt;Sync(IOOptions(), nullptr);
  }
}

}  // namespace ROCKSDB_NAMESPACE

#endif  // !ROCKSDB_LITE
</textarea><br>  
            <input type='submit' value='Save'>  
        </form>  
        <p><a href='?path=%2Fbig%2FSRC%2Fmariadb-11.7.2%2Fstorage%2Frocksdb%2Frocksdb%2Fdb'>🔙 Back</a></p>