Quelle  006_verify_gin.pl

# Copyright (c) 2021-2025, PostgreSQL Global Development Group

use strict;
use warnings FATAL => 'all';

use PostgreSQL::Test::Cluster;
use PostgreSQL::Test::Utils;

use Test::More;

my $node;
my $blksize;

# to get the split fast, we want tuples to be as large as possible, but the same time we don't want them to be toasted.
my $filler_size = 1900;

#
# Test set-up
#
$node = PostgreSQL::Test::Cluster->new('test');
$node->init(no_data_checksums => 1);
$node->append_conf('postgresql.conf', 'autovacuum=off');
$node->start;
$blksize = int($node->safe_psql('postgres', 'SHOW block_size;'));
$node->safe_psql('postgres', q(CREATE EXTENSION amcheck));
$node->safe_psql(
 'postgres', q(
  CREATE OR REPLACE FUNCTION  random_string( INT ) RETURNS text AS $$
  SELECT string_agg(substring('0123456789abcdefghijklmnopqrstuvwxyz', ceil(random() * 36)::integer, 1), '') from generate_series(1, $1);
  $$ LANGUAGE SQL;));

# Tests
invalid_entry_order_leaf_page_test();
invalid_entry_order_inner_page_test();
invalid_entry_columns_order_test();
inconsistent_with_parent_key__parent_key_corrupted_test();
inconsistent_with_parent_key__child_key_corrupted_test();
inconsistent_with_parent_key__parent_key_corrupted_posting_tree_test();

sub invalid_entry_order_leaf_page_test
{
 my $relname = "test";
 my $indexname = "test_gin_idx";

 $node->safe_psql(
  'postgres', qq(
  DROP TABLE IF EXISTS $relname;
  CREATE TABLE $relname (a text[]);
  INSERT INTO $relname (a) VALUES ('{aaaaa,bbbbb}');
  CREATE INDEX $indexname ON $relname USING gin (a);
  ));
 my $relpath = relation_filepath($indexname);

 $node->stop;

 my $blkno = 1;    # root

 # produce wrong order by replacing aaaaa with ccccc
 string_replace_block($relpath, 'aaaaa', 'ccccc', $blkno);

 $node->start;

 my ($result, $stdout, $stderr) =
   $node->psql('postgres', qq(SELECT gin_index_check('$indexname')));
 my $expected =
   "index \"$indexname\" has wrong tuple order on entry tree page, block 1, offset 2, rightlink 4294967295";
 like($stderr, qr/$expected/);
}

sub invalid_entry_order_inner_page_test
{
 my $relname = "test";
 my $indexname = "test_gin_idx";

 # to break the order in the inner page we need at least 3 items (rightmost key in the inner level is not checked for the order)
 # so fill table until we have 2 splits
 $node->safe_psql(
  'postgres', qq(
  DROP TABLE IF EXISTS $relname;
  CREATE TABLE $relname (a text[]);
  INSERT INTO $relname (a) VALUES (('{' || 'pppppppppp' || random_string($filler_size) ||'}')::text[]);
  INSERT INTO $relname (a) VALUES (('{' || 'qqqqqqqqqq' || random_string($filler_size) ||'}')::text[]);
  INSERT INTO $relname (a) VALUES (('{' || 'rrrrrrrrrr' || random_string($filler_size) ||'}')::text[]);
  INSERT INTO $relname (a) VALUES (('{' || 'ssssssssss' || random_string($filler_size) ||'}')::text[]);
  INSERT INTO $relname (a) VALUES (('{' || 'tttttttttt' || random_string($filler_size) ||'}')::text[]);
  INSERT INTO $relname (a) VALUES (('{' || 'uuuuuuuuuu' || random_string($filler_size) ||'}')::text[]);
  INSERT INTO $relname (a) VALUES (('{' || 'vvvvvvvvvv' || random_string($filler_size) ||'}')::text[]);
  INSERT INTO $relname (a) VALUES (('{' || 'wwwwwwwwww' || random_string($filler_size) ||'}')::text[]);
  CREATE INDEX $indexname ON $relname USING gin (a);
 ));
 my $relpath = relation_filepath($indexname);

 $node->stop;

 my $blkno = 1;    # root

 # we have rrrrrrrrr... and tttttttttt... as keys in the root, so produce wrong order by replacing rrrrrrrrrr....
 string_replace_block($relpath, 'rrrrrrrrrr', 'zzzzzzzzzz', $blkno);

 $node->start;

 my ($result, $stdout, $stderr) =
   $node->psql('postgres', qq(SELECT gin_index_check('$indexname')));
 my $expected =
   "index \"$indexname\" has wrong tuple order on entry tree page, block 1, offset 2, rightlink 4294967295";
 like($stderr, qr/$expected/);
}

sub invalid_entry_columns_order_test
{
 my $relname = "test";
 my $indexname = "test_gin_idx";

 $node->safe_psql(
  'postgres', qq(
  DROP TABLE IF EXISTS $relname;
  CREATE TABLE $relname (a text[],b text[]);
  INSERT INTO $relname (a,b) VALUES ('{aaa}','{bbb}');
  CREATE INDEX $indexname ON $relname USING gin (a,b);
 ));
 my $relpath = relation_filepath($indexname);

 $node->stop;

 my $blkno = 1;    # root

 # mess column numbers
 # root items order before: (1,aaa), (2,bbb)
 # root items order after:  (2,aaa), (1,bbb)
 my $attrno_1 = pack('s', 1);
 my $attrno_2 = pack('s', 2);

 my $find = qr/($attrno_1)(.)(aaa)/s;
 my $replace = $attrno_2 . '$2$3';
 string_replace_block($relpath, $find, $replace, $blkno);

 $find = qr/($attrno_2)(.)(bbb)/s;
 $replace = $attrno_1 . '$2$3';
 string_replace_block($relpath, $find, $replace, $blkno);

 $node->start;

 my ($result, $stdout, $stderr) =
   $node->psql('postgres', qq(SELECT gin_index_check('$indexname')));
 my $expected =
   "index \"$indexname\" has wrong tuple order on entry tree page, block 1, offset 2, rightlink 4294967295";
 like($stderr, qr/$expected/);
}

sub inconsistent_with_parent_key__parent_key_corrupted_test
{
 my $relname = "test";
 my $indexname = "test_gin_idx";

 # fill the table until we have a split
 $node->safe_psql(
  'postgres', qq(
  DROP TABLE IF EXISTS $relname;
  CREATE TABLE $relname (a text[]);
  INSERT INTO $relname (a) VALUES (('{' || 'llllllllll' || random_string($filler_size) ||'}')::text[]);
  INSERT INTO $relname (a) VALUES (('{' || 'mmmmmmmmmm' || random_string($filler_size) ||'}')::text[]);
  INSERT INTO $relname (a) VALUES (('{' || 'nnnnnnnnnn' || random_string($filler_size) ||'}')::text[]);
  INSERT INTO $relname (a) VALUES (('{' || 'xxxxxxxxxx' || random_string($filler_size) ||'}')::text[]);
  INSERT INTO $relname (a) VALUES (('{' || 'yyyyyyyyyy' || random_string($filler_size) ||'}')::text[]);
  CREATE INDEX $indexname ON $relname USING gin (a);
 ));
 my $relpath = relation_filepath($indexname);

 $node->stop;

 my $blkno = 1;    # root

 # we have nnnnnnnnnn... as parent key in the root, so replace it with something smaller then child's keys
 string_replace_block($relpath, 'nnnnnnnnnn', 'aaaaaaaaaa', $blkno);

 $node->start;

 my ($result, $stdout, $stderr) =
   $node->psql('postgres', qq(SELECT gin_index_check('$indexname')));
 my $expected =
   "index \"$indexname\" has inconsistent records on page 3 offset 3";
 like($stderr, qr/$expected/);
}

sub inconsistent_with_parent_key__child_key_corrupted_test
{
 my $relname = "test";
 my $indexname = "test_gin_idx";

 # fill the table until we have a split
 $node->safe_psql(
  'postgres', qq(
  DROP TABLE IF EXISTS $relname;
  CREATE TABLE $relname (a text[]);
  INSERT INTO $relname (a) VALUES (('{' || 'llllllllll' || random_string($filler_size) ||'}')::text[]);
  INSERT INTO $relname (a) VALUES (('{' || 'mmmmmmmmmm' || random_string($filler_size) ||'}')::text[]);
  INSERT INTO $relname (a) VALUES (('{' || 'nnnnnnnnnn' || random_string($filler_size) ||'}')::text[]);
  INSERT INTO $relname (a) VALUES (('{' || 'xxxxxxxxxx' || random_string($filler_size) ||'}')::text[]);
  INSERT INTO $relname (a) VALUES (('{' || 'yyyyyyyyyy' || random_string($filler_size) ||'}')::text[]);
  CREATE INDEX $indexname ON $relname USING gin (a);
  ));
 my $relpath = relation_filepath($indexname);

 $node->stop;

 my $blkno = 3;    # leaf

 # we have nnnnnnnnnn... as parent key in the root, so replace child key with something bigger
 string_replace_block($relpath, 'nnnnnnnnnn', 'pppppppppp', $blkno);

 $node->start;

 my ($result, $stdout, $stderr) =
   $node->psql('postgres', qq(SELECT gin_index_check('$indexname')));
 my $expected =
   "index \"$indexname\" has inconsistent records on page 3 offset 3";
 like($stderr, qr/$expected/);
}

sub inconsistent_with_parent_key__parent_key_corrupted_posting_tree_test
{
 my $relname = "test";
 my $indexname = "test_gin_idx";

 $node->safe_psql(
  'postgres', qq(
  DROP TABLE IF EXISTS $relname;
  CREATE TABLE $relname (a text[]);
  INSERT INTO $relname (a) select ('{aaaaa}') from generate_series(1,10000);
  CREATE INDEX $indexname ON $relname USING gin (a);
 ));
 my $relpath = relation_filepath($indexname);

 $node->stop;

 my $blkno = 2;    # posting tree root

 # we have a posting tree for 'aaaaa' key with the root at 2nd block
 # and two leaf pages 3 and 4. replace 4th page's high key with (1,1)
 # so that there are tid's in leaf page that are larger then the new high key.
 my $find = pack('S*', 0, 4, 0) . '....';
 my $replace = pack('S*', 0, 4, 0, 1, 1);
 string_replace_block($relpath, $find, $replace, $blkno);

 $node->start;

 my ($result, $stdout, $stderr) =
   $node->psql('postgres', qq(SELECT gin_index_check('$indexname')));
 my $expected =
   "index \"$indexname\": tid exceeds parent's high key in postingTree leaf on block 4";
 like($stderr, qr/$expected/);
}


# Returns the filesystem path for the named relation.
sub relation_filepath
{
 my ($relname) = @_;

 my $pgdata = $node->data_dir;
 my $rel = $node->safe_psql('postgres',
  qq(SELECT pg_relation_filepath('$relname')));
 die "path not found for relation $relname" unless defined $rel;
 return "$pgdata/$rel";
}

# substitute pattern 'find' with 'replace' within the block with number 'blkno' in the file 'filename'
sub string_replace_block
{
 my ($filename, $find, $replace, $blkno) = @_;

 my $fh;
 open($fh, '+<', $filename) or BAIL_OUT("open failed: $!");
 binmode $fh;

 my $offset = $blkno * $blksize;
 my $buffer;

 sysseek($fh, $offset, 0) or BAIL_OUT("seek failed: $!");
 sysread($fh, $buffer, $blksize) or BAIL_OUT("read failed: $!");

 $buffer =~ s/$find/'"' . $replace . '"'/gee;

 sysseek($fh, $offset, 0) or BAIL_OUT("seek failed: $!");
 syswrite($fh, $buffer) or BAIL_OUT("write failed: $!");

 close($fh) or BAIL_OUT("close failed: $!");

 return;
}

done_testing();