1 //  Copyright (c) 2011-present, Facebook, Inc.  All rights reserved.
2 //  This source code is licensed under both the GPLv2 (found in the
3 //  COPYING file in the root directory) and Apache 2.0 License
4 //  (found in the LICENSE.Apache file in the root directory).
5 
6 #pragma once
7 #include <string>
8 #include <unordered_set>
9 #include <vector>
10 
11 #include "db/column_family.h"
12 #include "db/dbformat.h"
13 #include "db/internal_stats.h"
14 #include "db/snapshot_impl.h"
15 #include "logging/event_logger.h"
16 #include "options/db_options.h"
17 #include "rocksdb/db.h"
18 #include "rocksdb/env.h"
19 #include "rocksdb/sst_file_writer.h"
20 #include "util/autovector.h"
21 
22 namespace ROCKSDB_NAMESPACE {
23 
24 class Directories;
25 
26 struct IngestedFileInfo {
27   // External file path
28   std::string external_file_path;
29   // Smallest internal key in external file
30   InternalKey smallest_internal_key;
31   // Largest internal key in external file
32   InternalKey largest_internal_key;
33   // Sequence number for keys in external file
34   SequenceNumber original_seqno;
35   // Offset of the global sequence number field in the file, will
36   // be zero if version is 1 (global seqno is not supported)
37   size_t global_seqno_offset;
38   // External file size
39   uint64_t file_size;
40   // total number of keys in external file
41   uint64_t num_entries;
42   // total number of range deletions in external file
43   uint64_t num_range_deletions;
44   // Id of column family this file shoule be ingested into
45   uint32_t cf_id;
46   // TableProperties read from external file
47   TableProperties table_properties;
48   // Version of external file
49   int version;
50 
51   // FileDescriptor for the file inside the DB
52   FileDescriptor fd;
53   // file path that we picked for file inside the DB
54   std::string internal_file_path;
55   // Global sequence number that we picked for the file inside the DB
56   SequenceNumber assigned_seqno = 0;
57   // Level inside the DB we picked for the external file.
58   int picked_level = 0;
59   // Whether to copy or link the external sst file. copy_file will be set to
60   // false if ingestion_options.move_files is true and underlying FS
61   // supports link operation. Need to provide a default value to make the
62   // undefined-behavior sanity check of llvm happy. Since
63   // ingestion_options.move_files is false by default, thus copy_file is true
64   // by default.
65   bool copy_file = true;
66 };
67 
68 class ExternalSstFileIngestionJob {
69  public:
ExternalSstFileIngestionJob(Env * env,VersionSet * versions,ColumnFamilyData * cfd,const ImmutableDBOptions & db_options,const EnvOptions & env_options,SnapshotList * db_snapshots,const IngestExternalFileOptions & ingestion_options,Directories * directories,EventLogger * event_logger)70   ExternalSstFileIngestionJob(
71       Env* env, VersionSet* versions, ColumnFamilyData* cfd,
72       const ImmutableDBOptions& db_options, const EnvOptions& env_options,
73       SnapshotList* db_snapshots,
74       const IngestExternalFileOptions& ingestion_options,
75       Directories* directories, EventLogger* event_logger)
76       : env_(env),
77         fs_(db_options.fs.get()),
78         versions_(versions),
79         cfd_(cfd),
80         db_options_(db_options),
81         env_options_(env_options),
82         db_snapshots_(db_snapshots),
83         ingestion_options_(ingestion_options),
84         directories_(directories),
85         event_logger_(event_logger),
86         job_start_time_(env_->NowMicros()),
87         consumed_seqno_count_(0) {
88     assert(directories != nullptr);
89   }
90 
91   // Prepare the job by copying external files into the DB.
92   Status Prepare(const std::vector<std::string>& external_files_paths,
93                  uint64_t next_file_number, SuperVersion* sv);
94 
95   // Check if we need to flush the memtable before running the ingestion job
96   // This will be true if the files we are ingesting are overlapping with any
97   // key range in the memtable.
98   //
99   // @param super_version A referenced SuperVersion that will be held for the
100   //    duration of this function.
101   //
102   // Thread-safe
103   Status NeedsFlush(bool* flush_needed, SuperVersion* super_version);
104 
105   // Will execute the ingestion job and prepare edit() to be applied.
106   // REQUIRES: Mutex held
107   Status Run();
108 
109   // Update column family stats.
110   // REQUIRES: Mutex held
111   void UpdateStats();
112 
113   // Cleanup after successful/failed job
114   void Cleanup(const Status& status);
115 
edit()116   VersionEdit* edit() { return &edit_; }
117 
files_to_ingest()118   const autovector<IngestedFileInfo>& files_to_ingest() const {
119     return files_to_ingest_;
120   }
121 
122   // How many sequence numbers did we consume as part of the ingest job?
ConsumedSequenceNumbersCount()123   int ConsumedSequenceNumbersCount() const { return consumed_seqno_count_; }
124 
125  private:
126   // Open the external file and populate `file_to_ingest` with all the
127   // external information we need to ingest this file.
128   Status GetIngestedFileInfo(const std::string& external_file,
129                              IngestedFileInfo* file_to_ingest,
130                              SuperVersion* sv);
131 
132   // Assign `file_to_ingest` the appropriate sequence number and  the lowest
133   // possible level that it can be ingested to according to compaction_style.
134   // REQUIRES: Mutex held
135   Status AssignLevelAndSeqnoForIngestedFile(SuperVersion* sv,
136                                             bool force_global_seqno,
137                                             CompactionStyle compaction_style,
138                                             SequenceNumber last_seqno,
139                                             IngestedFileInfo* file_to_ingest,
140                                             SequenceNumber* assigned_seqno);
141 
142   // File that we want to ingest behind always goes to the lowest level;
143   // we just check that it fits in the level, that DB allows ingest_behind,
144   // and that we don't have 0 seqnums at the upper levels.
145   // REQUIRES: Mutex held
146   Status CheckLevelForIngestedBehindFile(IngestedFileInfo* file_to_ingest);
147 
148   // Set the file global sequence number to `seqno`
149   Status AssignGlobalSeqnoForIngestedFile(IngestedFileInfo* file_to_ingest,
150                                           SequenceNumber seqno);
151 
152   // Check if `file_to_ingest` can fit in level `level`
153   // REQUIRES: Mutex held
154   bool IngestedFileFitInLevel(const IngestedFileInfo* file_to_ingest,
155                               int level);
156 
157   // Helper method to sync given file.
158   template <typename TWritableFile>
159   Status SyncIngestedFile(TWritableFile* file);
160 
161   Env* env_;
162   FileSystem* fs_;
163   VersionSet* versions_;
164   ColumnFamilyData* cfd_;
165   const ImmutableDBOptions& db_options_;
166   const EnvOptions& env_options_;
167   SnapshotList* db_snapshots_;
168   autovector<IngestedFileInfo> files_to_ingest_;
169   const IngestExternalFileOptions& ingestion_options_;
170   Directories* directories_;
171   EventLogger* event_logger_;
172   VersionEdit edit_;
173   uint64_t job_start_time_;
174   int consumed_seqno_count_;
175   // Set in ExternalSstFileIngestionJob::Prepare(), if true all files are
176   // ingested in L0
177   bool files_overlap_{false};
178 };
179 
180 }  // namespace ROCKSDB_NAMESPACE
181