pg_probackup

1
import os
2
import unittest
3
from .helpers.ptrack_helpers import ProbackupTest, ProbackupException, idx_ptrack
4
from datetime import datetime, timedelta
5
import subprocess
6

7

8
class CompressionTest(ProbackupTest, unittest.TestCase):
9

10
    # @unittest.skip("skip")
11
    # @unittest.expectedFailure
12
    def test_basic_compression_stream_zlib(self):
13
        """
14
        make archive node, make full and page stream backups,
15
        check data correctness in restored instance
16
        """
17
        self.maxDiff = None
18
        backup_dir = os.path.join(self.tmp_path, self.module_name, self.fname, 'backup')
19
        node = self.make_simple_node(
20
            base_dir=os.path.join(self.module_name, self.fname, 'node'),
21
            set_replication=True,
22
            initdb_params=['--data-checksums'])
23

24
        self.init_pb(backup_dir)
25
        self.add_instance(backup_dir, 'node', node)
26
        self.set_archiving(backup_dir, 'node', node)
27
        node.slow_start()
28

29
        # FULL BACKUP
30
        node.safe_psql(
31
            "postgres",
32
            "create table t_heap as select i as id, md5(i::text) as text, "
33
            "md5(repeat(i::text,10))::tsvector as tsvector "
34
            "from generate_series(0,256) i")
35
        full_result = node.table_checksum("t_heap")
36
        full_backup_id = self.backup_node(
37
            backup_dir, 'node', node, backup_type='full',
38
            options=[
39
                '--stream',
40
                '--compress-algorithm=zlib'])
41

42
        # PAGE BACKUP
43
        node.safe_psql(
44
            "postgres",
45
            "insert into t_heap select i as id, md5(i::text) as text, "
46
            "md5(repeat(i::text,10))::tsvector as tsvector "
47
            "from generate_series(256,512) i")
48
        page_result = node.table_checksum("t_heap")
49
        page_backup_id = self.backup_node(
50
            backup_dir, 'node', node, backup_type='page',
51
            options=[
52
                '--stream', '--compress-algorithm=zlib'])
53

54
        # DELTA BACKUP
55
        node.safe_psql(
56
            "postgres",
57
            "insert into t_heap select i as id, md5(i::text) as text, "
58
            "md5(repeat(i::text,10))::tsvector as tsvector "
59
            "from generate_series(512,768) i")
60
        delta_result = node.table_checksum("t_heap")
61
        delta_backup_id = self.backup_node(
62
            backup_dir, 'node', node, backup_type='delta',
63
            options=['--stream', '--compress-algorithm=zlib'])
64

65
        # Drop Node
66
        node.cleanup()
67

68
        # Check full backup
69
        self.assertIn(
70
            "INFO: Restore of backup {0} completed.".format(full_backup_id),
71
            self.restore_node(
72
                backup_dir, 'node', node, backup_id=full_backup_id,
73
                options=[
74
                    "-j", "4", "--immediate",
75
                    "--recovery-target-action=promote"]),
76
            '\n Unexpected Error Message: {0}\n CMD: {1}'.format(
77
                repr(self.output), self.cmd))
78
        node.slow_start()
79

80
        full_result_new = node.table_checksum("t_heap")
81
        self.assertEqual(full_result, full_result_new)
82
        node.cleanup()
83

84
        # Check page backup
85
        self.assertIn(
86
            "INFO: Restore of backup {0} completed.".format(page_backup_id),
87
            self.restore_node(
88
                backup_dir, 'node', node, backup_id=page_backup_id,
89
                options=[
90
                    "-j", "4", "--immediate",
91
                    "--recovery-target-action=promote"]),
92
            '\n Unexpected Error Message: {0}\n CMD: {1}'.format(
93
                repr(self.output), self.cmd))
94
        node.slow_start()
95

96
        page_result_new = node.table_checksum("t_heap")
97
        self.assertEqual(page_result, page_result_new)
98
        node.cleanup()
99

100
        # Check delta backup
101
        self.assertIn(
102
            "INFO: Restore of backup {0} completed.".format(delta_backup_id),
103
            self.restore_node(
104
                backup_dir, 'node', node, backup_id=delta_backup_id,
105
                options=[
106
                    "-j", "4", "--immediate",
107
                    "--recovery-target-action=promote"]),
108
            '\n Unexpected Error Message: {0}\n CMD: {1}'.format(
109
                repr(self.output), self.cmd))
110
        node.slow_start()
111

112
        delta_result_new = node.table_checksum("t_heap")
113
        self.assertEqual(delta_result, delta_result_new)
114

115
    def test_compression_archive_zlib(self):
116
        """
117
        make archive node, make full and page backups,
118
        check data correctness in restored instance
119
        """
120
        self.maxDiff = None
121
        backup_dir = os.path.join(self.tmp_path, self.module_name, self.fname, 'backup')
122
        node = self.make_simple_node(
123
            base_dir=os.path.join(self.module_name, self.fname, 'node'),
124
            set_replication=True,
125
            initdb_params=['--data-checksums'])
126

127
        self.init_pb(backup_dir)
128
        self.add_instance(backup_dir, 'node', node)
129
        self.set_archiving(backup_dir, 'node', node)
130
        node.slow_start()
131

132
        # FULL BACKUP
133
        node.safe_psql(
134
            "postgres",
135
            "create table t_heap as select i as id, md5(i::text) as text, "
136
            "md5(i::text)::tsvector as tsvector from generate_series(0,1) i")
137
        full_result = node.table_checksum("t_heap")
138
        full_backup_id = self.backup_node(
139
            backup_dir, 'node', node, backup_type='full',
140
            options=["--compress-algorithm=zlib"])
141

142
        # PAGE BACKUP
143
        node.safe_psql(
144
            "postgres",
145
            "insert into t_heap select i as id, md5(i::text) as text, "
146
            "md5(i::text)::tsvector as tsvector "
147
            "from generate_series(0,2) i")
148
        page_result = node.table_checksum("t_heap")
149
        page_backup_id = self.backup_node(
150
            backup_dir, 'node', node, backup_type='page',
151
            options=["--compress-algorithm=zlib"])
152

153
        # DELTA BACKUP
154
        node.safe_psql(
155
            "postgres",
156
            "insert into t_heap select i as id, md5(i::text) as text, "
157
            "md5(i::text)::tsvector as tsvector from generate_series(0,3) i")
158
        delta_result = node.table_checksum("t_heap")
159
        delta_backup_id = self.backup_node(
160
            backup_dir, 'node', node, backup_type='delta',
161
            options=['--compress-algorithm=zlib'])
162

163
        # Drop Node
164
        node.cleanup()
165

166
        # Check full backup
167
        self.assertIn(
168
            "INFO: Restore of backup {0} completed.".format(full_backup_id),
169
            self.restore_node(
170
                backup_dir, 'node', node, backup_id=full_backup_id,
171
                options=[
172
                    "-j", "4", "--immediate",
173
                    "--recovery-target-action=promote"]),
174
            '\n Unexpected Error Message: {0}\n CMD: {1}'.format(
175
                repr(self.output), self.cmd))
176
        node.slow_start()
177

178
        full_result_new = node.table_checksum("t_heap")
179
        self.assertEqual(full_result, full_result_new)
180
        node.cleanup()
181

182
        # Check page backup
183
        self.assertIn(
184
            "INFO: Restore of backup {0} completed.".format(page_backup_id),
185
            self.restore_node(
186
                backup_dir, 'node', node, backup_id=page_backup_id,
187
                options=[
188
                    "-j", "4", "--immediate",
189
                    "--recovery-target-action=promote"]),
190
            '\n Unexpected Error Message: {0}\n CMD: {1}'.format(
191
                repr(self.output), self.cmd))
192
        node.slow_start()
193

194
        page_result_new = node.table_checksum("t_heap")
195
        self.assertEqual(page_result, page_result_new)
196
        node.cleanup()
197

198
        # Check delta backup
199
        self.assertIn(
200
            "INFO: Restore of backup {0} completed.".format(delta_backup_id),
201
            self.restore_node(
202
                backup_dir, 'node', node, backup_id=delta_backup_id,
203
                options=[
204
                    "-j", "4", "--immediate",
205
                    "--recovery-target-action=promote"]),
206
            '\n Unexpected Error Message: {0}\n CMD: {1}'.format(
207
                repr(self.output), self.cmd))
208
        node.slow_start()
209

210
        delta_result_new = node.table_checksum("t_heap")
211
        self.assertEqual(delta_result, delta_result_new)
212
        node.cleanup()
213

214
    def test_compression_stream_pglz(self):
215
        """
216
        make archive node, make full and page stream backups,
217
        check data correctness in restored instance
218
        """
219
        self.maxDiff = None
220
        backup_dir = os.path.join(self.tmp_path, self.module_name, self.fname, 'backup')
221
        node = self.make_simple_node(
222
            base_dir=os.path.join(self.module_name, self.fname, 'node'),
223
            set_replication=True,
224
            initdb_params=['--data-checksums'])
225

226
        self.init_pb(backup_dir)
227
        self.add_instance(backup_dir, 'node', node)
228
        self.set_archiving(backup_dir, 'node', node)
229
        node.slow_start()
230

231
        # FULL BACKUP
232
        node.safe_psql(
233
            "postgres",
234
            "create table t_heap as select i as id, md5(i::text) as text, "
235
            "md5(repeat(i::text,10))::tsvector as tsvector "
236
            "from generate_series(0,256) i")
237
        full_result = node.table_checksum("t_heap")
238
        full_backup_id = self.backup_node(
239
            backup_dir, 'node', node, backup_type='full',
240
            options=['--stream', '--compress-algorithm=pglz'])
241

242
        # PAGE BACKUP
243
        node.safe_psql(
244
            "postgres",
245
            "insert into t_heap select i as id, md5(i::text) as text, "
246
            "md5(repeat(i::text,10))::tsvector as tsvector "
247
            "from generate_series(256,512) i")
248
        page_result = node.table_checksum("t_heap")
249
        page_backup_id = self.backup_node(
250
            backup_dir, 'node', node, backup_type='page',
251
            options=['--stream', '--compress-algorithm=pglz'])
252

253
        # DELTA BACKUP
254
        node.safe_psql(
255
            "postgres",
256
            "insert into t_heap select i as id, md5(i::text) as text, "
257
            "md5(repeat(i::text,10))::tsvector as tsvector "
258
            "from generate_series(512,768) i")
259
        delta_result = node.table_checksum("t_heap")
260
        delta_backup_id = self.backup_node(
261
            backup_dir, 'node', node, backup_type='delta',
262
            options=['--stream', '--compress-algorithm=pglz'])
263

264
        # Drop Node
265
        node.cleanup()
266

267
        # Check full backup
268
        self.assertIn(
269
            "INFO: Restore of backup {0} completed.".format(full_backup_id),
270
            self.restore_node(
271
                backup_dir, 'node', node, backup_id=full_backup_id,
272
                options=[
273
                    "-j", "4", "--immediate",
274
                    "--recovery-target-action=promote"]),
275
            '\n Unexpected Error Message: {0}\n CMD: {1}'.format(
276
                repr(self.output), self.cmd))
277
        node.slow_start()
278

279
        full_result_new = node.table_checksum("t_heap")
280
        self.assertEqual(full_result, full_result_new)
281
        node.cleanup()
282

283
        # Check page backup
284
        self.assertIn(
285
            "INFO: Restore of backup {0} completed.".format(page_backup_id),
286
            self.restore_node(
287
                backup_dir, 'node', node, backup_id=page_backup_id,
288
                options=[
289
                    "-j", "4", "--immediate",
290
                    "--recovery-target-action=promote"]),
291
            '\n Unexpected Error Message: {0}\n CMD: {1}'.format(
292
                repr(self.output), self.cmd))
293
        node.slow_start()
294

295
        page_result_new = node.table_checksum("t_heap")
296
        self.assertEqual(page_result, page_result_new)
297
        node.cleanup()
298

299
        # Check delta backup
300
        self.assertIn(
301
            "INFO: Restore of backup {0} completed.".format(delta_backup_id),
302
            self.restore_node(
303
                backup_dir, 'node', node, backup_id=delta_backup_id,
304
                options=[
305
                    "-j", "4", "--immediate",
306
                    "--recovery-target-action=promote"]),
307
            '\n Unexpected Error Message: {0}\n CMD: {1}'.format(
308
                repr(self.output), self.cmd))
309
        node.slow_start()
310

311
        delta_result_new = node.table_checksum("t_heap")
312
        self.assertEqual(delta_result, delta_result_new)
313
        node.cleanup()
314

315
    def test_compression_archive_pglz(self):
316
        """
317
        make archive node, make full and page backups,
318
        check data correctness in restored instance
319
        """
320
        self.maxDiff = None
321
        backup_dir = os.path.join(self.tmp_path, self.module_name, self.fname, 'backup')
322
        node = self.make_simple_node(
323
            base_dir=os.path.join(self.module_name, self.fname, 'node'),
324
            set_replication=True,
325
            initdb_params=['--data-checksums'])
326

327
        self.init_pb(backup_dir)
328
        self.add_instance(backup_dir, 'node', node)
329
        self.set_archiving(backup_dir, 'node', node)
330
        node.slow_start()
331

332
        # FULL BACKUP
333
        node.safe_psql(
334
            "postgres",
335
            "create table t_heap as select i as id, md5(i::text) as text, "
336
            "md5(i::text)::tsvector as tsvector "
337
            "from generate_series(0,100) i")
338
        full_result = node.table_checksum("t_heap")
339
        full_backup_id = self.backup_node(
340
            backup_dir, 'node', node, backup_type='full',
341
            options=['--compress-algorithm=pglz'])
342

343
        # PAGE BACKUP
344
        node.safe_psql(
345
            "postgres",
346
            "insert into t_heap select i as id, md5(i::text) as text, "
347
            "md5(i::text)::tsvector as tsvector "
348
            "from generate_series(100,200) i")
349
        page_result = node.table_checksum("t_heap")
350
        page_backup_id = self.backup_node(
351
            backup_dir, 'node', node, backup_type='page',
352
            options=['--compress-algorithm=pglz'])
353

354
        # DELTA BACKUP
355
        node.safe_psql(
356
            "postgres",
357
            "insert into t_heap select i as id, md5(i::text) as text, "
358
            "md5(i::text)::tsvector as tsvector "
359
            "from generate_series(200,300) i")
360
        delta_result = node.table_checksum("t_heap")
361
        delta_backup_id = self.backup_node(
362
            backup_dir, 'node', node, backup_type='delta',
363
            options=['--compress-algorithm=pglz'])
364

365
        # Drop Node
366
        node.cleanup()
367

368
        # Check full backup
369
        self.assertIn(
370
            "INFO: Restore of backup {0} completed.".format(full_backup_id),
371
            self.restore_node(
372
                backup_dir, 'node', node, backup_id=full_backup_id,
373
                options=[
374
                    "-j", "4", "--immediate",
375
                    "--recovery-target-action=promote"]),
376
            '\n Unexpected Error Message: {0}\n CMD: {1}'.format(
377
                repr(self.output), self.cmd))
378
        node.slow_start()
379

380
        full_result_new = node.table_checksum("t_heap")
381
        self.assertEqual(full_result, full_result_new)
382
        node.cleanup()
383

384
        # Check page backup
385
        self.assertIn(
386
            "INFO: Restore of backup {0} completed.".format(page_backup_id),
387
            self.restore_node(
388
                backup_dir, 'node', node, backup_id=page_backup_id,
389
                options=[
390
                    "-j", "4", "--immediate",
391
                    "--recovery-target-action=promote"]),
392
            '\n Unexpected Error Message: {0}\n CMD: {1}'.format(
393
                repr(self.output), self.cmd))
394
        node.slow_start()
395

396
        page_result_new = node.table_checksum("t_heap")
397
        self.assertEqual(page_result, page_result_new)
398
        node.cleanup()
399

400
        # Check delta backup
401
        self.assertIn(
402
            "INFO: Restore of backup {0} completed.".format(delta_backup_id),
403
            self.restore_node(
404
                backup_dir, 'node', node, backup_id=delta_backup_id,
405
                options=[
406
                    "-j", "4", "--immediate",
407
                    "--recovery-target-action=promote"]),
408
            '\n Unexpected Error Message: {0}\n CMD: {1}'.format(
409
                repr(self.output), self.cmd))
410
        node.slow_start()
411

412
        delta_result_new = node.table_checksum("t_heap")
413
        self.assertEqual(delta_result, delta_result_new)
414
        node.cleanup()
415

416
    def test_compression_wrong_algorithm(self):
417
        """
418
        make archive node, make full and page backups,
419
        check data correctness in restored instance
420
        """
421
        self.maxDiff = None
422
        backup_dir = os.path.join(self.tmp_path, self.module_name, self.fname, 'backup')
423
        node = self.make_simple_node(
424
            base_dir=os.path.join(self.module_name, self.fname, 'node'),
425
            set_replication=True,
426
            initdb_params=['--data-checksums'])
427

428
        self.init_pb(backup_dir)
429
        self.add_instance(backup_dir, 'node', node)
430
        self.set_archiving(backup_dir, 'node', node)
431
        node.slow_start()
432

433
        try:
434
            self.backup_node(
435
                backup_dir, 'node', node,
436
                backup_type='full', options=['--compress-algorithm=bla-blah'])
437
            # we should die here because exception is what we expect to happen
438
            self.assertEqual(
439
                1, 0,
440
                "Expecting Error because compress-algorithm is invalid.\n "
441
                "Output: {0} \n CMD: {1}".format(
442
                    repr(self.output), self.cmd))
443
        except ProbackupException as e:
444
            self.assertEqual(
445
                e.message,
446
                'ERROR: Invalid compress algorithm value "bla-blah"\n',
447
                '\n Unexpected Error Message: {0}\n CMD: {1}'.format(
448
                    repr(e.message), self.cmd))
449

450
    # @unittest.skip("skip")
451
    def test_incompressible_pages(self):
452
        """
453
        make archive node, create table with incompressible toast pages,
454
        take backup with compression, make sure that page was not compressed,
455
        restore backup and check data correctness
456
        """
457
        backup_dir = os.path.join(self.tmp_path, self.module_name, self.fname, 'backup')
458
        node = self.make_simple_node(
459
            base_dir=os.path.join(self.module_name, self.fname, 'node'),
460
            set_replication=True,
461
            initdb_params=['--data-checksums'])
462

463
        self.init_pb(backup_dir)
464
        self.add_instance(backup_dir, 'node', node)
465
        self.set_archiving(backup_dir, 'node', node)
466
        node.slow_start()
467

468
        # Full
469
        self.backup_node(
470
            backup_dir, 'node', node,
471
            options=[
472
                '--compress-algorithm=zlib',
473
                '--compress-level=0'])
474

475
        node.pgbench_init(scale=3)
476

477
        self.backup_node(
478
            backup_dir, 'node', node,
479
            backup_type='delta',
480
            options=[
481
                '--compress-algorithm=zlib',
482
                '--compress-level=0'])
483

484
        pgdata = self.pgdata_content(node.data_dir)
485

486
        node.cleanup()
487

488
        self.restore_node(backup_dir, 'node', node)
489

490
        # Physical comparison
491
        if self.paranoia:
492
            pgdata_restored = self.pgdata_content(node.data_dir)
493
            self.compare_pgdata(pgdata, pgdata_restored)
494

495
        node.slow_start()
496