1 | #include "duckdb/storage/write_ahead_log.hpp" |
---|---|
2 | |
3 | #include "duckdb/catalog/catalog_entry/scalar_macro_catalog_entry.hpp" |
4 | #include "duckdb/catalog/catalog_entry/schema_catalog_entry.hpp" |
5 | #include "duckdb/catalog/catalog_entry/table_catalog_entry.hpp" |
6 | #include "duckdb/catalog/catalog_entry/type_catalog_entry.hpp" |
7 | #include "duckdb/catalog/catalog_entry/view_catalog_entry.hpp" |
8 | #include "duckdb/main/database.hpp" |
9 | #include "duckdb/parser/parsed_data/alter_table_info.hpp" |
10 | #include <cstring> |
11 | |
12 | namespace duckdb { |
13 | |
14 | WriteAheadLog::WriteAheadLog(AttachedDatabase &database, const string &path) : skip_writing(false), database(database) { |
15 | wal_path = path; |
16 | writer = make_uniq<BufferedFileWriter>(args&: FileSystem::Get(db&: database), args: path.c_str(), |
17 | args: FileFlags::FILE_FLAGS_WRITE | FileFlags::FILE_FLAGS_FILE_CREATE | |
18 | FileFlags::FILE_FLAGS_APPEND); |
19 | } |
20 | |
21 | WriteAheadLog::~WriteAheadLog() { |
22 | } |
23 | |
24 | int64_t WriteAheadLog::GetWALSize() { |
25 | D_ASSERT(writer); |
26 | return writer->GetFileSize(); |
27 | } |
28 | |
29 | idx_t WriteAheadLog::GetTotalWritten() { |
30 | D_ASSERT(writer); |
31 | return writer->GetTotalWritten(); |
32 | } |
33 | |
34 | void WriteAheadLog::Truncate(int64_t size) { |
35 | writer->Truncate(size); |
36 | } |
37 | |
38 | void WriteAheadLog::Delete() { |
39 | if (!writer) { |
40 | return; |
41 | } |
42 | writer.reset(); |
43 | |
44 | auto &fs = FileSystem::Get(db&: database); |
45 | fs.RemoveFile(filename: wal_path); |
46 | } |
47 | |
48 | //===--------------------------------------------------------------------===// |
49 | // Write Entries |
50 | //===--------------------------------------------------------------------===// |
51 | void WriteAheadLog::WriteCheckpoint(block_id_t meta_block) { |
52 | writer->Write<WALType>(element: WALType::CHECKPOINT); |
53 | writer->Write<block_id_t>(element: meta_block); |
54 | } |
55 | |
56 | //===--------------------------------------------------------------------===// |
57 | // CREATE TABLE |
58 | //===--------------------------------------------------------------------===// |
59 | void WriteAheadLog::WriteCreateTable(const TableCatalogEntry &entry) { |
60 | if (skip_writing) { |
61 | return; |
62 | } |
63 | writer->Write<WALType>(element: WALType::CREATE_TABLE); |
64 | entry.Serialize(serializer&: *writer); |
65 | } |
66 | |
67 | //===--------------------------------------------------------------------===// |
68 | // DROP TABLE |
69 | //===--------------------------------------------------------------------===// |
70 | void WriteAheadLog::WriteDropTable(const TableCatalogEntry &entry) { |
71 | if (skip_writing) { |
72 | return; |
73 | } |
74 | writer->Write<WALType>(element: WALType::DROP_TABLE); |
75 | writer->WriteString(val: entry.schema.name); |
76 | writer->WriteString(val: entry.name); |
77 | } |
78 | |
79 | //===--------------------------------------------------------------------===// |
80 | // CREATE SCHEMA |
81 | //===--------------------------------------------------------------------===// |
82 | void WriteAheadLog::WriteCreateSchema(const SchemaCatalogEntry &entry) { |
83 | if (skip_writing) { |
84 | return; |
85 | } |
86 | writer->Write<WALType>(element: WALType::CREATE_SCHEMA); |
87 | writer->WriteString(val: entry.name); |
88 | } |
89 | |
90 | //===--------------------------------------------------------------------===// |
91 | // SEQUENCES |
92 | //===--------------------------------------------------------------------===// |
93 | void WriteAheadLog::WriteCreateSequence(const SequenceCatalogEntry &entry) { |
94 | if (skip_writing) { |
95 | return; |
96 | } |
97 | writer->Write<WALType>(element: WALType::CREATE_SEQUENCE); |
98 | entry.Serialize(serializer&: *writer); |
99 | } |
100 | |
101 | void WriteAheadLog::WriteDropSequence(const SequenceCatalogEntry &entry) { |
102 | if (skip_writing) { |
103 | return; |
104 | } |
105 | writer->Write<WALType>(element: WALType::DROP_SEQUENCE); |
106 | writer->WriteString(val: entry.schema.name); |
107 | writer->WriteString(val: entry.name); |
108 | } |
109 | |
110 | void WriteAheadLog::WriteSequenceValue(const SequenceCatalogEntry &entry, SequenceValue val) { |
111 | if (skip_writing) { |
112 | return; |
113 | } |
114 | writer->Write<WALType>(element: WALType::SEQUENCE_VALUE); |
115 | writer->WriteString(val: entry.schema.name); |
116 | writer->WriteString(val: entry.name); |
117 | writer->Write<uint64_t>(element: val.usage_count); |
118 | writer->Write<int64_t>(element: val.counter); |
119 | } |
120 | |
121 | //===--------------------------------------------------------------------===// |
122 | // MACROS |
123 | //===--------------------------------------------------------------------===// |
124 | void WriteAheadLog::WriteCreateMacro(const ScalarMacroCatalogEntry &entry) { |
125 | if (skip_writing) { |
126 | return; |
127 | } |
128 | writer->Write<WALType>(element: WALType::CREATE_MACRO); |
129 | entry.Serialize(serializer&: *writer); |
130 | } |
131 | |
132 | void WriteAheadLog::WriteDropMacro(const ScalarMacroCatalogEntry &entry) { |
133 | if (skip_writing) { |
134 | return; |
135 | } |
136 | writer->Write<WALType>(element: WALType::DROP_MACRO); |
137 | writer->WriteString(val: entry.schema.name); |
138 | writer->WriteString(val: entry.name); |
139 | } |
140 | |
141 | void WriteAheadLog::WriteCreateTableMacro(const TableMacroCatalogEntry &entry) { |
142 | if (skip_writing) { |
143 | return; |
144 | } |
145 | writer->Write<WALType>(element: WALType::CREATE_TABLE_MACRO); |
146 | entry.Serialize(serializer&: *writer); |
147 | } |
148 | |
149 | void WriteAheadLog::WriteDropTableMacro(const TableMacroCatalogEntry &entry) { |
150 | if (skip_writing) { |
151 | return; |
152 | } |
153 | writer->Write<WALType>(element: WALType::DROP_TABLE_MACRO); |
154 | writer->WriteString(val: entry.schema.name); |
155 | writer->WriteString(val: entry.name); |
156 | } |
157 | |
158 | //===--------------------------------------------------------------------===// |
159 | // Indexes |
160 | //===--------------------------------------------------------------------===// |
161 | void WriteAheadLog::WriteCreateIndex(const IndexCatalogEntry &entry) { |
162 | if (skip_writing) { |
163 | return; |
164 | } |
165 | writer->Write<WALType>(element: WALType::CREATE_INDEX); |
166 | entry.Serialize(serializer&: *writer); |
167 | } |
168 | |
169 | void WriteAheadLog::WriteDropIndex(const IndexCatalogEntry &entry) { |
170 | if (skip_writing) { |
171 | return; |
172 | } |
173 | writer->Write<WALType>(element: WALType::DROP_INDEX); |
174 | writer->WriteString(val: entry.schema.name); |
175 | writer->WriteString(val: entry.name); |
176 | } |
177 | |
178 | //===--------------------------------------------------------------------===// |
179 | // Custom Types |
180 | //===--------------------------------------------------------------------===// |
181 | void WriteAheadLog::WriteCreateType(const TypeCatalogEntry &entry) { |
182 | if (skip_writing) { |
183 | return; |
184 | } |
185 | writer->Write<WALType>(element: WALType::CREATE_TYPE); |
186 | entry.Serialize(serializer&: *writer); |
187 | } |
188 | |
189 | void WriteAheadLog::WriteDropType(const TypeCatalogEntry &entry) { |
190 | if (skip_writing) { |
191 | return; |
192 | } |
193 | writer->Write<WALType>(element: WALType::DROP_TYPE); |
194 | writer->WriteString(val: entry.schema.name); |
195 | writer->WriteString(val: entry.name); |
196 | } |
197 | |
198 | //===--------------------------------------------------------------------===// |
199 | // VIEWS |
200 | //===--------------------------------------------------------------------===// |
201 | void WriteAheadLog::WriteCreateView(const ViewCatalogEntry &entry) { |
202 | if (skip_writing) { |
203 | return; |
204 | } |
205 | writer->Write<WALType>(element: WALType::CREATE_VIEW); |
206 | entry.Serialize(serializer&: *writer); |
207 | } |
208 | |
209 | void WriteAheadLog::WriteDropView(const ViewCatalogEntry &entry) { |
210 | if (skip_writing) { |
211 | return; |
212 | } |
213 | writer->Write<WALType>(element: WALType::DROP_VIEW); |
214 | writer->WriteString(val: entry.schema.name); |
215 | writer->WriteString(val: entry.name); |
216 | } |
217 | |
218 | //===--------------------------------------------------------------------===// |
219 | // DROP SCHEMA |
220 | //===--------------------------------------------------------------------===// |
221 | void WriteAheadLog::WriteDropSchema(const SchemaCatalogEntry &entry) { |
222 | if (skip_writing) { |
223 | return; |
224 | } |
225 | writer->Write<WALType>(element: WALType::DROP_SCHEMA); |
226 | writer->WriteString(val: entry.name); |
227 | } |
228 | |
229 | //===--------------------------------------------------------------------===// |
230 | // DATA |
231 | //===--------------------------------------------------------------------===// |
232 | void WriteAheadLog::WriteSetTable(string &schema, string &table) { |
233 | if (skip_writing) { |
234 | return; |
235 | } |
236 | writer->Write<WALType>(element: WALType::USE_TABLE); |
237 | writer->WriteString(val: schema); |
238 | writer->WriteString(val: table); |
239 | } |
240 | |
241 | void WriteAheadLog::WriteInsert(DataChunk &chunk) { |
242 | if (skip_writing) { |
243 | return; |
244 | } |
245 | D_ASSERT(chunk.size() > 0); |
246 | chunk.Verify(); |
247 | |
248 | writer->Write<WALType>(element: WALType::INSERT_TUPLE); |
249 | chunk.Serialize(serializer&: *writer); |
250 | } |
251 | |
252 | void WriteAheadLog::WriteDelete(DataChunk &chunk) { |
253 | if (skip_writing) { |
254 | return; |
255 | } |
256 | D_ASSERT(chunk.size() > 0); |
257 | D_ASSERT(chunk.ColumnCount() == 1 && chunk.data[0].GetType() == LogicalType::ROW_TYPE); |
258 | chunk.Verify(); |
259 | |
260 | writer->Write<WALType>(element: WALType::DELETE_TUPLE); |
261 | chunk.Serialize(serializer&: *writer); |
262 | } |
263 | |
264 | void WriteAheadLog::WriteUpdate(DataChunk &chunk, const vector<column_t> &column_indexes) { |
265 | if (skip_writing) { |
266 | return; |
267 | } |
268 | D_ASSERT(chunk.size() > 0); |
269 | D_ASSERT(chunk.ColumnCount() == 2); |
270 | D_ASSERT(chunk.data[1].GetType().id() == LogicalType::ROW_TYPE); |
271 | chunk.Verify(); |
272 | |
273 | writer->Write<WALType>(element: WALType::UPDATE_TUPLE); |
274 | writer->Write<idx_t>(element: column_indexes.size()); |
275 | for (auto &col_idx : column_indexes) { |
276 | writer->Write<column_t>(element: col_idx); |
277 | } |
278 | chunk.Serialize(serializer&: *writer); |
279 | } |
280 | |
281 | //===--------------------------------------------------------------------===// |
282 | // Write ALTER Statement |
283 | //===--------------------------------------------------------------------===// |
284 | void WriteAheadLog::WriteAlter(data_ptr_t ptr, idx_t data_size) { |
285 | if (skip_writing) { |
286 | return; |
287 | } |
288 | writer->Write<WALType>(element: WALType::ALTER_INFO); |
289 | writer->WriteData(buffer: ptr, write_size: data_size); |
290 | } |
291 | |
292 | //===--------------------------------------------------------------------===// |
293 | // FLUSH |
294 | //===--------------------------------------------------------------------===// |
295 | void WriteAheadLog::Flush() { |
296 | if (skip_writing) { |
297 | return; |
298 | } |
299 | // write an empty entry |
300 | writer->Write<WALType>(element: WALType::WAL_FLUSH); |
301 | // flushes all changes made to the WAL to disk |
302 | writer->Sync(); |
303 | } |
304 | |
305 | } // namespace duckdb |
306 |