Received: from malur.postgresql.org ([217.196.149.56]) by arkaria.postgresql.org with esmtps (TLS1.3) tls TLS_ECDHE_RSA_WITH_AES_256_GCM_SHA384 (Exim 4.96) (envelope-from ) id 1vRDMf-009hgy-31 for pgsql-hackers@arkaria.postgresql.org; Thu, 04 Dec 2025 17:43:39 +0000 Received: from localhost ([127.0.0.1] helo=malur.postgresql.org) by malur.postgresql.org with esmtp (Exim 4.96) (envelope-from ) id 1vRDMd-003kPZ-2Y for pgsql-hackers@arkaria.postgresql.org; Thu, 04 Dec 2025 17:43:36 +0000 Received: from magus.postgresql.org ([2a02:c0:301:0:ffff::29]) by malur.postgresql.org with esmtps (TLS1.3) tls TLS_ECDHE_RSA_WITH_AES_256_GCM_SHA384 (Exim 4.96) (envelope-from ) id 1vRDMd-003kPP-09 for pgsql-hackers@lists.postgresql.org; Thu, 04 Dec 2025 17:43:35 +0000 Received: from mail-ej1-x62b.google.com ([2a00:1450:4864:20::62b]) by magus.postgresql.org with esmtps (TLS1.3) tls TLS_ECDHE_RSA_WITH_AES_128_GCM_SHA256 (Exim 4.96) (envelope-from ) id 1vRDMZ-0038js-2d for pgsql-hackers@lists.postgresql.org; Thu, 04 Dec 2025 17:43:35 +0000 Received: by mail-ej1-x62b.google.com with SMTP id a640c23a62f3a-b737502f77bso166256166b.2 for ; Thu, 04 Dec 2025 09:43:31 -0800 (PST) DKIM-Signature: v=1; a=rsa-sha256; c=relaxed/relaxed; d=cybertec.at; s=google; t=1764870209; x=1765475009; darn=lists.postgresql.org; h=message-id:date:mime-version:comments:references:in-reply-to :subject:cc:to:from:from:to:cc:subject:date:message-id:reply-to; bh=o/8KlvVKL0BvqcX+xbOynNRIy0tNHTcEyErzFb0WRlA=; b=joY7F0VuHY/6YKagGrvTsbXlhnRXRvN9PEfhSlLgBlCuBlUpe8CZc5hbb6Sj6xIO/1 56qmX82PgUtHLrisfAEH1QzVv5YVVHsfM8sMFO21lVuHHIveroJ5LswNZ+bu+J0LEMVI q3mr9mr/jtcyMqvPdzNc2Agj9gSEBYXVXD82t5BUo2d++sCrYN8IKLJ0vCDUJPwY4BSz XZl8ipP7fLeP2Eo3YHdU7Dx0KtgnzD87oi+kvdFAHZt+tSfB1Z1YhQ2d4r/+BvkAAVnt 7lJe/dnVV8kchFqjBf87JI4Mhmp5EFj59wq91WUSdRxi5wuCGMA1+mag+cG35eADijvM z0xw== X-Google-DKIM-Signature: v=1; a=rsa-sha256; c=relaxed/relaxed; d=1e100.net; s=20230601; t=1764870209; x=1765475009; h=message-id:date:mime-version:comments:references:in-reply-to :subject:cc:to:from:x-gm-gg:x-gm-message-state:from:to:cc:subject :date:message-id:reply-to; bh=o/8KlvVKL0BvqcX+xbOynNRIy0tNHTcEyErzFb0WRlA=; b=s8/0q5ev5kC0SLwtzzr48I6ffis8j6WKYqOfuvS3YrT7fp3aWF0+pBs5GiAFBaUGRb WOC/ClTcZMbZPIe1qBbsEeH/NWdezfdGY15wrgfHAuvBcMKN1WUCTz+Opq4WKYxMpREs IQLEE/qTdKyZsfOpj4OCnr7n6tYtbvnKRmcRcX4yAKMYjzcpshP+IH1GlXjugbX30ciQ Zetkr3UDIiTqhnq/j5tJz6adr/kMeyrlBiuZo+/k8YiRFK0ARzeYR63ByAoUxnnjwNdP 90RdzBYSG2CwVMxF+kLQDRWvQ5awm+Kl/y38CUNa0RKAMZRu8AGzGR71Bi+2vu8F/jpV ODSQ== X-Gm-Message-State: AOJu0Yw+vlRGOruNlqEfYAMXAEQSzDBr3+3Hj6M2CpRPVmSSd1TARI+g 9QplVAa6/+jpODT78NREY2S2yQDmVhesuVb2O1B07mJidsZFgOjHpz/tVeu/ZPvesdb6/AkDD/T ZGo8N X-Gm-Gg: ASbGncs6FTUKGkteg0xdn9Pz1qpd2TNjeZioKjrkO7fYFBDCmf2ifXerR7ViABXd5Wb +HIBNlrInJ9CA/DDHEFRlFXHBuX8pg4O5S2sq5gKaBmWwxfVXnDLpTlg0fJsd+lQxY9seW3s5ga S0JMK8VaNhnOFzXpDCTC1DJ64hm2E5v7kNWY+WKvmysmxpO2hfBUnpox2U6JBcNsa9aXUGwHOeC UGPdeZb6A0xdNz0zGezphP6xhHM5g40h15srWk8j3VHky+UG4DsGnYW5mM7VdQC5+vPAjl/2CcI 4qYyAF4r7flJ4ArUiCN48FGaVDSHi9ABIudg1Ob+032lQPJYpb7zl0JStj/O2KhS9KPKaCVKzi5 hkAElHWubUbJuK0l/CPst74LXM0obDsKyvy4kj+8U6OmKQQo40kh1B4m/dqK47VGf+fe6mUZnCx vF9XajMyPjWHHiHOl/aKPxLxtY X-Google-Smtp-Source: AGHT+IFvSh6LyWD3AOoYDgYWuJ6Oonu1JIh56cDnxR9ul5zQu/XpgO02HETDF8n2RlzdePkJo/bo/Q== X-Received: by 2002:a17:907:96a9:b0:b72:77c7:d8ad with SMTP id a640c23a62f3a-b79ec67dc11mr319523766b.35.1764870209023; Thu, 04 Dec 2025 09:43:29 -0800 (PST) Received: from localhost (109-81-168-246.rct.o2.cz. [109.81.168.246]) by smtp.gmail.com with ESMTPSA id a640c23a62f3a-b79fab22ae8sm103826066b.4.2025.12.04.09.43.27 (version=TLS1_3 cipher=TLS_AES_256_GCM_SHA384 bits=256/256); Thu, 04 Dec 2025 09:43:28 -0800 (PST) From: Antonin Houska To: Alvaro Herrera cc: Pg Hackers , Mihail Nikalayeu , Robert Treat Subject: Re: Adding REPACK [concurrently] In-reply-to: <202510301734.pj4uds3mqxx4@alvherre.pgsql> References: <202510301734.pj4uds3mqxx4@alvherre.pgsql> Comments: In-reply-to Alvaro Herrera message dated "Fri, 31 Oct 2025 00:17:05 +0100." X-Mailer: MH-E 8.6+git; nmh 1.8; GNU Emacs 28.3 MIME-Version: 1.0 Content-Type: multipart/mixed; boundary="=-=-=" Date: Thu, 04 Dec 2025 18:43:27 +0100 Message-ID: <116433.1764870207@localhost> List-Id: List-Help: List-Subscribe: List-Post: List-Owner: List-Archive: Archived-At: Precedence: bulk --=-=-= Content-Type: text/plain Alvaro Herrera wrote: > Here's a new installment of this series, v25, including the CONCURRENTLY > part, which required some conflict fixes on top of the much-changed > v24-0001 patch. v26 attached here. It's been rebased and reflects most of the feedback. A few incomplete items are marked as TBD here [1] and [2] is another thing that needs discussion. Besides that, I've done some refactoring in 0004: 1) move more code to setup_logical_decoding(), and 2) reduced the number of arguments of process_concurrent_changes() by using a new structure. Both these changes are a preparation for a background worker that will perform the logical decoding, but seem to be useful as such. (I have a PoC of the worker but will post it later, it doesn't seem to be the priority for now.) I've also removed support for decoding TRUNCATE because I realized that this command uses AccessExclusiveLock, so it cannot be executed on a table that REPACK (CONCURRENTLY) is just processing. Also I tried to fix TAB completion in psql. > I have not yet addressed Robert Treat's feedback from October 12th. These are still pending. [1] https://www.postgresql.org/message-id/23631.1764855372%40localhost [2] https://www.postgresql.org/message-id/CAJSLCQ2_jX8WmNOC4eu6hL5QyNHceOkgPbGhKHFw2X5onVEKDQ%40mail.gmail.com -- Antonin Houska Web: https://www.cybertec-postgresql.com --=-=-= Content-Type: text/x-diff; charset=utf-8 Content-Disposition: attachment; filename=v26-0001-Add-REPACK-command.patch Content-Transfer-Encoding: quoted-printable From 59f1075f4017398d83d399c336f6634700a02006 Mon Sep 17 00:00:00 2001 From: Antonin Houska Date: Thu, 4 Dec 2025 18:20:07 +0100 Subject: [PATCH 1/4] Add REPACK command MIME-Version: 1.0 Content-Type: text/plain; charset=3DUTF-8 Content-Transfer-Encoding: 8bit REPACK absorbs the functionality of VACUUM FULL and CLUSTER in a single command. Because this functionality is completely different from regular VACUUM, having it separate from VACUUM makes it easier for users to understand; as for CLUSTER, the term is heavily overloaded in the IT world and even in Postgres itself, so it's good that we can avoid it. This also adds pg_repackdb, a new utility that can invoke the new commands. This is heavily based on vacuumdb. Author: Antonin Houska Co-authored-by: =C3=81lvaro Herrera Reviewed-by: Mihail Nikalayeu Reviewed-by: Robert Treat Reviewed-by: Euler Taveira Reviewed-by: Matheus Alcantara Reviewed-by: Junwang Zhao Reviewed-by: jian he Discussion: https://postgr.es/m/82651.1720540558@antos Discussion: https://postgr.es/m/202507262156.sb455angijk6@alvherre.pgsql --- doc/src/sgml/monitoring.sgml | 223 +++++- doc/src/sgml/ref/allfiles.sgml | 2 + doc/src/sgml/ref/cluster.sgml | 97 +-- doc/src/sgml/ref/clusterdb.sgml | 5 + doc/src/sgml/ref/pg_repackdb.sgml | 488 +++++++++++++ doc/src/sgml/ref/repack.sgml | 328 +++++++++ doc/src/sgml/ref/vacuum.sgml | 33 +- doc/src/sgml/reference.sgml | 2 + src/backend/access/heap/heapam_handler.c | 32 +- src/backend/catalog/index.c | 2 +- src/backend/catalog/system_views.sql | 29 +- src/backend/commands/cluster.c | 851 +++++++++++++++-------- src/backend/commands/vacuum.c | 6 +- src/backend/parser/gram.y | 86 ++- src/backend/tcop/utility.c | 20 +- src/backend/utils/adt/pgstatfuncs.c | 2 + src/bin/psql/tab-complete.in.c | 42 +- src/bin/scripts/Makefile | 4 +- src/bin/scripts/meson.build | 2 + src/bin/scripts/pg_repackdb.c | 242 +++++++ src/bin/scripts/t/103_repackdb.pl | 47 ++ src/bin/scripts/vacuuming.c | 114 ++- src/bin/scripts/vacuuming.h | 3 + src/include/commands/cluster.h | 8 +- src/include/commands/progress.h | 50 +- src/include/nodes/parsenodes.h | 35 +- src/include/parser/kwlist.h | 1 + src/include/tcop/cmdtaglist.h | 1 + src/include/utils/backend_progress.h | 1 + src/test/regress/expected/cluster.out | 134 +++- src/test/regress/expected/rules.out | 72 +- src/test/regress/sql/cluster.sql | 70 +- src/tools/pgindent/typedefs.list | 2 + 33 files changed, 2490 insertions(+), 544 deletions(-) create mode 100644 doc/src/sgml/ref/pg_repackdb.sgml create mode 100644 doc/src/sgml/ref/repack.sgml create mode 100644 src/bin/scripts/pg_repackdb.c create mode 100644 src/bin/scripts/t/103_repackdb.pl diff --git a/doc/src/sgml/monitoring.sgml b/doc/src/sgml/monitoring.sgml index 039d73691be..b8da77b4d89 100644 --- a/doc/src/sgml/monitoring.sgml +++ b/doc/src/sgml/monitoring.sgml @@ -405,6 +405,14 @@ postgres 27093 0.0 0.0 30096 2752 ? Ss = 11:34 0:00 postgres: ser =20 + + pg_stat_progress_repackpg_stat_progress_repack + One row for each backend running + REPACK, showing current progress. See + . + + + pg_stat_progress_basebackuppg_stat_progress_basebackup One row for each WAL sender process streaming a base backup, @@ -5609,7 +5617,8 @@ FROM pg_stat_get_backend_idset() AS backendid; certain commands during command execution. Currently, the only commands which support progress reporting are ANALYZE, CLUSTER, - CREATE INDEX, VACUUM, + CREATE INDEX, REPACK, + VACUUM, COPY, and (i.e., replica= tion command that issues to take @@ -6068,6 +6077,218 @@ FROM pg_stat_get_backend_idset() AS backendid; =20 + + REPACK Progress Reporting + + + pg_stat_progress_repack + + + + Whenever REPACK is running, + the pg_stat_progress_repack view will contain a + row for each backend that is currently running the command. The tables + below describe the information that will be reported and provide + information about how to interpret it. + + + + <structname>pg_stat_progress_repack</structname> View + + + + + Column Type + + + Description + + + + + + + + pid integer + + + Process ID of backend. + + + + + + datid oid + + + OID of the database to which this backend is connected. + + + + + + datname name + + + Name of the database to which this backend is connected. + + + + + + relid oid + + + OID of the table being repacked. + + + + + + phase text + + + Current processing phase. See . + + + + + + repack_index_relid oid + + + If the table is being scanned using an index, this is the OID of the + index being used; otherwise, it is zero. + + + + + + heap_tuples_scanned bigint + + + Number of heap tuples scanned. + This counter only advances when the phase is + seq scanning heap, + index scanning heap + or writing new heap. + + + + + + heap_tuples_written bigint + + + Number of heap tuples written. + This counter only advances when the phase is + seq scanning heap, + index scanning heap + or writing new heap. + + + + + + heap_blks_total bigint + + + Total number of heap blocks in the table. This number is reported + as of the beginning of seq scanning heap. + + + + + + heap_blks_scanned bigint + + + Number of heap blocks scanned. This counter only advances when the + phase is seq scanning heap. + + + + + + index_rebuild_count bigint + + + Number of indexes rebuilt. This counter only advances when the pha= se + is rebuilding index. + + + + +
+ + + REPACK Phases + + + + + + Phase + Description + + + + + + initializing + + The command is preparing to begin scanning the heap. This phase is + expected to be very brief. + + + + seq scanning heap + + The command is currently scanning the table using a sequential scan. + + + + index scanning heap + + REPACK is currently scanning the table using an = index scan. + + + + sorting tuples + + REPACK is currently sorting tuples. + + + + writing new heap + + REPACK is currently writing the new heap. + + + + swapping relation files + + The command is currently swapping newly-built files into place. + + + + rebuilding index + + The command is currently rebuilding an index. + + + + performing final cleanup + + The command is performing final cleanup. When this phase is + completed, REPACK will end. + + + + +
+ + COPY Progress Reporting =20 diff --git a/doc/src/sgml/ref/allfiles.sgml b/doc/src/sgml/ref/allfiles.sgml index e167406c744..5df944d13ca 100644 --- a/doc/src/sgml/ref/allfiles.sgml +++ b/doc/src/sgml/ref/allfiles.sgml @@ -167,6 +167,7 @@ Complete list of usable sgml source files in this direc= tory. + @@ -213,6 +214,7 @@ Complete list of usable sgml source files in this direc= tory. + diff --git a/doc/src/sgml/ref/cluster.sgml b/doc/src/sgml/ref/cluster.sgml index 0b47460080b..2cda711bc9f 100644 --- a/doc/src/sgml/ref/cluster.sgml +++ b/doc/src/sgml/ref/cluster.sgml @@ -33,51 +33,13 @@ CLUSTER [ ( option [, ...] ) ] [ Description =20 - CLUSTER instructs PostgreSQL - to cluster the table specified - by table_name - based on the index specified by - index_name. The index mu= st - already have been defined on - table_name. + The CLUSTER command is equivalent to + with an USING INDEX + clause. See there for more details. =20 - - When a table is clustered, it is physically reordered - based on the index information. Clustering is a one-time operation: - when the table is subsequently updated, the changes are - not clustered. That is, no attempt is made to store new or - updated rows according to their index order. (If one wishes, one can - periodically recluster by issuing the command again. Also, setting - the table's fillfactor storage parameter to less than - 100% can aid in preserving cluster ordering during updates, since updat= ed - rows are kept on the same page if enough space is available there.) - - - - When a table is clustered, PostgreSQL - remembers which index it was clustered by. The form - CLUSTER table_name - reclusters the table using the same index as before. You can also - use the CLUSTER or SET WITHOUT CLUSTER - forms of ALTER TABLE to set the index to be used for - future cluster operations, or to clear any previous setting. - - - - CLUSTER without a - table_name reclusters al= l the - previously-clustered tables in the current database that the calling us= er - has privileges for. This form of CLUSTER cannot be - executed inside a transaction block. - + =20 - - When a table is being clustered, an ACCESS - EXCLUSIVE lock is acquired on it. This prevents any other - database operations (both reads and writes) from operating on the - table until the CLUSTER is finished. - =20 @@ -136,63 +98,12 @@ CLUSTER [ ( option [, ...] ) ] [ =20 - - In cases where you are accessing single rows randomly - within a table, the actual order of the data in the - table is unimportant. However, if you tend to access some - data more than others, and there is an index that groups - them together, you will benefit from using CLUSTER. - If you are requesting a range of indexed values from a table, or a - single indexed value that has multiple rows that match, - CLUSTER will help because once the index identifies= the - table page for the first row that matches, all other rows - that match are probably already on the same table page, - and so you save disk accesses and speed up the query. - - - - CLUSTER can re-sort the table using either an index= scan - on the specified index, or (if the index is a b-tree) a sequential - scan followed by sorting. It will attempt to choose the method that - will be faster, based on planner cost parameters and available statist= ical - information. - - While CLUSTER is running, the is temporarily changed to pg_ca= talog, pg_temp. =20 - - When an index scan is used, a temporary copy of the table is created t= hat - contains the table data in the index order. Temporary copies of each - index on the table are created as well. Therefore, you need free spac= e on - disk at least equal to the sum of the table size and the index sizes. - - - - When a sequential scan and sort is used, a temporary sort file is - also created, so that the peak temporary space requirement is as much - as double the table size, plus the index sizes. This method is often - faster than the index scan method, but if the disk space requirement is - intolerable, you can disable this choice by temporarily setting to off. - - - - It is advisable to set to - a reasonably large value (but not more than the amount of RAM you can - dedicate to the CLUSTER operation) before clusterin= g. - - - - Because the planner records statistics about the ordering of - tables, it is advisable to run = ANALYZE - on the newly clustered table. - Otherwise, the planner might make poor choices of query plans. - - Because CLUSTER remembers which indexes are cluster= ed, one can cluster the tables one wants clustered manually the first time, diff --git a/doc/src/sgml/ref/clusterdb.sgml b/doc/src/sgml/ref/clusterdb.s= gml index 0d2051bf6f1..546c1289c31 100644 --- a/doc/src/sgml/ref/clusterdb.sgml +++ b/doc/src/sgml/ref/clusterdb.sgml @@ -64,6 +64,11 @@ PostgreSQL documentation this utility and via other methods for accessing the server. =20 + + clusterdb has been superceded by + pg_repackdb. + + =20 =20 diff --git a/doc/src/sgml/ref/pg_repackdb.sgml b/doc/src/sgml/ref/pg_repack= db.sgml new file mode 100644 index 00000000000..b313b54ab63 --- /dev/null +++ b/doc/src/sgml/ref/pg_repackdb.sgml @@ -0,0 +1,488 @@ + + + + + pg_repackdb + + + + pg_repackdb + 1 + Application + + + + pg_repackdb + repack and analyze a PostgreSQL + database + + + + + pg_repackdb + connection-option + option + + + + + + + + table + ( column [,...] ) + + + + + + dbname + + + + + + + + pg_repackdb + connection-option + option + + + + + + + + schema + + + + + + dbname + + + + + + + + pg_repackdb + connection-option + option + + + + + + + + schema + + + + + + dbname + + + + + + + + + Description + + + pg_repackdb is a utility for repacking a + PostgreSQL database. + pg_repackdb will also generate internal + statistics used by the PostgreSQL query + optimizer. + + + + pg_repackdb is a wrapper around the SQL + command REPACK T= here + is no effective difference between repacking and analyzing databases via + this utility and via other methods for accessing the server. + + + + + Options + + + pg_repackdb accepts the following command-l= ine arguments: + + + + + + + Repack all databases. + + + + + + + + + + Specifies the name of the database to be repacked or analyzed, + when / is not used. If t= his + is not specified, the database name is read from the environment + variable PGDATABASE. If that is not set, the user = name + specified for the connection is used. + The dbname can be + a connection string. If= so, + connection string parameters will override any conflicting command + line options. + + + + + + + + + + Echo the commands that pg_repackdb + generates and sends to the server. + + + + + + + + + Pass the USING INDEX clause to REPACK<= /literal>, + and optionally the index name to specify. + + + + + + + + + + Execute the repack or analyze commands in parallel by running + njobs + commands simultaneously. This option may reduce the processing ti= me + but it also increases the load on the database server. + + + pg_repackdb will open + njobs connections t= o the + database, so make sure your + setting is high enough to accommodate all connections. + + + Note that using this mode might cause deadlock failures if certain + system catalogs are processed in parallel. + + + + + + + + + + Repack or analyze all tables in + schema only. Multi= ple + schemas can be repacked by writing multiple + switches. + + + + + + + + + + Do not repack or analyze any tables in + schema. Multiple s= chemas + can be excluded by writing multiple switches. + + + + + + + + + + Do not display progress messages. + + + + + + + + + + Repack or analyze table + only. Column names can be specified only in conjunction with + the option. Multiple tables can be + repacked by writing multiple + switches. + + + + If you specify columns, you probably have to escape the parenthes= es + from the shell. (See examples below.) + + + + + + + + + + + Print detailed information during processing. + + + + + + + + + + Print the pg_repackdb version and exit. + + + + + + + + + + Also calculate statistics for use by the optimizer. If a column n= ame + list is given, only compute statistics for those columns. + + + + + + + + + + Show help about pg_repackdb command line + arguments, and exit. + + + + + + + + + pg_repackdb also accepts + the following command-line arguments for connection parameters: + + + + + + + Specifies the host name of the machine on which the server + is running. If the value begins with a slash, it is used + as the directory for the Unix domain socket. + + + + + + + + + + Specifies the TCP port or local Unix domain socket file + extension on which the server + is listening for connections. + + + + + + + + + + User name to connect as. + + + + + + + + + + Never issue a password prompt. If the server requires + password authentication and a password is not available by + other means such as a .pgpass file, the + connection attempt will fail. This option can be useful in + batch jobs and scripts where no user is present to enter a + password. + + + + + + + + + + Force pg_repackdb to prompt for a + password before connecting to a database. + + + + This option is never essential, since + pg_repackdb will automatically prompt + for a password if the server demands password authentication. + However, pg_repackdb will waste a + connection attempt finding out that the server wants a password. + In some cases it is worth typing to avoid the = extra + connection attempt. + + + + + + + + + When the / is used, conne= ct + to this database to gather the list of databases to repack. + If not specified, the postgres database will be= used, + or if that does not exist, template1 will be us= ed. + This can be a connection + string. If so, connection string parameters will override = any + conflicting command line options. Also, connection string paramet= ers + other than the database name itself will be re-used when connecting + to other databases. + + + + + + + + + + Environment + + + + PGDATABASE + PGHOST + PGPORT + PGUSER + + + + Default connection parameters + + + + + + PG_COLOR + + + Specifies whether to use color in diagnostic messages. Possible valu= es + are always, auto and + never. + + + + + + + This utility, like most other PostgreSQL uti= lities, + also uses the environment variables supported by libpq + (see ). + + + + + + + Diagnostics + + + In case of difficulty, see + and for + discussions of potential problems and error messages. + The database server must be running at the + targeted host. Also, any default connection settings and environment + variables used by the libpq front-end + library will apply. + + + + + + Examples + + + To repack the database test: + +$ pg_repackdb test + + + + + To repack and analyze for the optimizer a database named + bigdb: + +$ pg_repackdb --analyze bigdb + + + + + To repack a single table + foo in a database named + xyzzy, and analyze a single column + bar of the table for the optimizer: + +$ pg_repackdb --analyze --verbose --table=3D'f= oo(bar)' xyzzy + + + + To repack all tables in the foo and bar schemas + in a database named xyzzy: + +$ pg_repackdb --schema=3D'foo' --schema=3D'bar= ' xyzzy + + + + + + + See Also + + + + + + + diff --git a/doc/src/sgml/ref/repack.sgml b/doc/src/sgml/ref/repack.sgml new file mode 100644 index 00000000000..61d5c2cdef1 --- /dev/null +++ b/doc/src/sgml/ref/repack.sgml @@ -0,0 +1,328 @@ + + + + + REPACK + + + + REPACK + 7 + SQL - Language Statements + + + + REPACK + rewrite a table to reclaim disk space + + + + +REPACK [ ( option [, ...] )= ] [ table_and_columns [ USI= NG INDEX [ index_name ] ] ] +REPACK [ ( option [, ...] )= ] USING INDEX + +where option can be= one of: + + VERBOSE [ boolean ] + ANALYZE [ boolean ] + +and table_and_columns is: + + table_name [ ( column_name [, ...] ) ] + + + + + Description + + + REPACK reclaims storage occupied by dead + tuples. Unlike VACUUM, it does so by rewriting the + entire contents of the table specified + by table_name into a new= disk + file with no extra space (except for the space guaranteed by + the fillfactor storage parameter), allowing unused s= pace + to be returned to the operating system. + + + + Without + a table_name, R= EPACK + processes every table and materialized view in the current database that + the current user has the MAINTAIN privilege on. This + form of REPACK cannot be executed inside a transacti= on + block. + + + + If a USING INDEX clause is specified, the rows are + physically reordered based on information from an index. Please see the + notes on clustering below. + + + + When a table is being repacked, an ACCESS EXCLUSIVE = lock + is acquired on it. This prevents any other database operations (both re= ads + and writes) from operating on the table until the REPACK + is finished. + + + + Notes on Clustering + + + If the USING INDEX clause is specified, the rows in + the table are physically reordered following an index: if an index name + is specified in the command, then that index is used; if no index name + is specified, then the index that has been configured as the index to + cluster on. If no index has been configured in this way, an error is + thrown. The index given in the USING INDEX clause + is configured as the index to cluster on, as well as an index given + to the CLUSTER command. An index can be set + manually using ALTER TABLE ... CLUSTER ON, and reset + with ALTER TABLE ... SET WITHOUT CLUSTER. + + + + If no table name is specified in REPACK USING INDEX, + all tables which have a clustering index defined and which the calling + user has privileges for are processed. + + + + Clustering is a one-time operation: when the table is + subsequently updated, the changes are not clustered. That is, no atte= mpt + is made to store new or updated rows according to their index order. = (If + one wishes, one can periodically recluster by issuing the command agai= n. + Also, setting the table's fillfactor storage parame= ter + to less than 100% can aid in preserving cluster ordering during update= s, + since updated rows are kept on the same page if enough space is availa= ble + there.) + + + + In cases where you are accessing single rows randomly within a table, = the + actual order of the data in the table is unimportant. However, if you = tend + to access some data more than others, and there is an index that groups + them together, you will benefit from using clustering. If + you are requesting a range of indexed values from a table, or a single + indexed value that has multiple rows that match, + REPACK will help because once the index identifies = the + table page for the first row that matches, all other rows that match a= re + probably already on the same table page, and so you save disk accesses= and + speed up the query. + + + + REPACK can re-sort the table using either an index = scan + on the specified index (if the index is a b-tree), or a sequential scan + followed by sorting. It will attempt to choose the method that will be + faster, based on planner cost parameters and available statistical + information. + + + + Because the planner records statistics about the ordering of tables, i= t is + advisable to + run ANALYZE on= the + newly repacked table. Otherwise, the planner might make poor choices = of + query plans. + + + + + Notes on Resources + + + When an index scan or a sequential scan without sort is used, a tempor= ary + copy of the table is created that contains the table data in the index + order. Temporary copies of each index on the table are created as wel= l. + Therefore, you need free space on disk at least equal to the sum of the + table size and the index sizes. + + + + When a sequential scan and sort is used, a temporary sort file is also + created, so that the peak temporary space requirement is as much as do= uble + the table size, plus the index sizes. This method is often faster than + the index scan method, but if the disk space requirement is intolerabl= e, + you can disable this choice by temporarily setting + to off. + + + + It is advisable to set to= a + reasonably large value (but not more than the amount of RAM you can + dedicate to the REPACK operation) before repacking. + + + + + + + Parameters + + + + table_name + + + The name (possibly schema-qualified) of a table. + + + + + + column_name + + + The name of a specific column to analyze. Defaults to all columns. + If a column list is specific, ANALYZE must also + be specified. + + + + + + index_name + + + The name of an index. + + + + + + VERBOSE + + + Prints a progress report as each table is repacked + at INFO level. + + + + + + ANALYZE + ANALYSE + + + Applies on the table after repacking= . This is + currently only supported when a single (non-partitioned) table is sp= ecified. + + + + + + boolean + + + Specifies whether the selected option should be turned on or off. + You can write TRUE, ON, or + 1 to enable the option, and FALSE, + OFF, or 0 to disable it. The + boolean value can also + be omitted, in which case TRUE is assumed. + + + + + + + + Notes + + + To repack a table, one must have the MAINTAIN privi= lege + on the table. + + + + While REPACK is running, the is temporarily changed to pg_ca= talog, + pg_temp. + + + + Each backend running REPACK will report its progress + in the pg_stat_progress_repack view. See + for details. + + + + Repacking a partitioned table repacks each of its partitions. If an in= dex + is specified, each partition is repacked using the partition of that + index. REPACK on a partitioned table cannot be exec= uted + inside a transaction block. + + + + + + Examples + + + Repack the table employees: + +REPACK employees; + + + + + Repack the table employees on the basis of its + index employees_ind (Since index is used here, this = is + effectively clustering): + +REPACK employees USING INDEX employees_ind; + + + + + Repack the table cases on physical ordering, + running an ANALYZE on the given columns once + repacking is done, showing informational messages: + +REPACK (ANALYZE, VERBOSE) cases (district, case_nr); + + + + + Repack all tables in the database on which you have + the MAINTAIN privilege: + +REPACK; + + + + + Repack all tables for which a clustering index has previously been + configured on which you have the MAINTAIN privilege, + showing informational messages: + +REPACK (VERBOSE) USING INDEX; + + + + + + + Compatibility + + + There is no REPACK statement in the SQL standard. + + + + + See Also + + + + + + + + diff --git a/doc/src/sgml/ref/vacuum.sgml b/doc/src/sgml/ref/vacuum.sgml index 6d0fdd43cfb..ac5d083d468 100644 --- a/doc/src/sgml/ref/vacuum.sgml +++ b/doc/src/sgml/ref/vacuum.sgml @@ -25,7 +25,6 @@ VACUUM [ ( option [, ...] ) ] [ where option can be= one of: =20 - FULL [ boolean ] FREEZE [ boolean ] VERBOSE [ boolean ] ANALYZE [ boolean ] @@ -39,6 +38,7 @@ VACUUM [ ( option [, ...] ) ] [ boolean ] ONLY_DATABASE_STATS [ boolean ] BUFFER_USAGE_LIMIT size + FULL [ boolean ] =20 and table_and_columns is: =20 @@ -95,20 +95,6 @@ VACUUM [ ( option [, ...] ) ] [ Parameters =20 - - FULL - - - Selects full vacuum, which can reclaim more - space, but takes much longer and exclusively locks the table. - This method also requires extra disk space, since it writes a - new copy of the table and doesn't release the old copy until - the operation is complete. Usually this should only be used when a - significant amount of space needs to be reclaimed from within the ta= ble. - - - - FREEZE @@ -362,6 +348,23 @@ VACUUM [ ( option [, ...] ) ] [ =20 + + FULL + + + This option, which is deprecated, makes VACUUM + behave like REPACK without a + USING INDEX clause. + This method of compacting the table takes much longer than + VACUUM and exclusively locks the table. + This method also requires extra disk space, since it writes a + new copy of the table and doesn't release the old copy until + the operation is complete. Usually this should only be used when a + significant amount of space needs to be reclaimed from within the ta= ble. + + + + boolean diff --git a/doc/src/sgml/reference.sgml b/doc/src/sgml/reference.sgml index 2cf02c37b17..5d9a8a25a02 100644 --- a/doc/src/sgml/reference.sgml +++ b/doc/src/sgml/reference.sgml @@ -195,6 +195,7 @@ &refreshMaterializedView; &reindex; &releaseSavepoint; + &repack; &reset; &revoke; &rollback; @@ -258,6 +259,7 @@ &pgIsready; &pgReceivewal; &pgRecvlogical; + &pgRepackdb; &pgRestore; &pgVerifyBackup; &psqlRef; diff --git a/src/backend/access/heap/heapam_handler.c b/src/backend/access/= heap/heapam_handler.c index bcbac844bb6..79f9de5d760 100644 --- a/src/backend/access/heap/heapam_handler.c +++ b/src/backend/access/heap/heapam_handler.c @@ -741,13 +741,13 @@ heapam_relation_copy_for_cluster(Relation OldHeap, Re= lation NewHeap, if (OldIndex !=3D NULL && !use_sort) { const int ci_index[] =3D { - PROGRESS_CLUSTER_PHASE, - PROGRESS_CLUSTER_INDEX_RELID + PROGRESS_REPACK_PHASE, + PROGRESS_REPACK_INDEX_RELID }; int64 ci_val[2]; =20 /* Set phase and OIDOldIndex to columns */ - ci_val[0] =3D PROGRESS_CLUSTER_PHASE_INDEX_SCAN_HEAP; + ci_val[0] =3D PROGRESS_REPACK_PHASE_INDEX_SCAN_HEAP; ci_val[1] =3D RelationGetRelid(OldIndex); pgstat_progress_update_multi_param(2, ci_index, ci_val); =20 @@ -759,15 +759,15 @@ heapam_relation_copy_for_cluster(Relation OldHeap, Re= lation NewHeap, else { /* In scan-and-sort mode and also VACUUM FULL, set phase */ - pgstat_progress_update_param(PROGRESS_CLUSTER_PHASE, - PROGRESS_CLUSTER_PHASE_SEQ_SCAN_HEAP); + pgstat_progress_update_param(PROGRESS_REPACK_PHASE, + PROGRESS_REPACK_PHASE_SEQ_SCAN_HEAP); =20 tableScan =3D table_beginscan(OldHeap, SnapshotAny, 0, (ScanKey) NULL); heapScan =3D (HeapScanDesc) tableScan; indexScan =3D NULL; =20 /* Set total heap blocks */ - pgstat_progress_update_param(PROGRESS_CLUSTER_TOTAL_HEAP_BLKS, + pgstat_progress_update_param(PROGRESS_REPACK_TOTAL_HEAP_BLKS, heapScan->rs_nblocks); } =20 @@ -809,7 +809,7 @@ heapam_relation_copy_for_cluster(Relation OldHeap, Rela= tion NewHeap, * is manually updated to the correct value when the table * scan finishes. */ - pgstat_progress_update_param(PROGRESS_CLUSTER_HEAP_BLKS_SCANNED, + pgstat_progress_update_param(PROGRESS_REPACK_HEAP_BLKS_SCANNED, heapScan->rs_nblocks); break; } @@ -825,7 +825,7 @@ heapam_relation_copy_for_cluster(Relation OldHeap, Rela= tion NewHeap, */ if (prev_cblock !=3D heapScan->rs_cblock) { - pgstat_progress_update_param(PROGRESS_CLUSTER_HEAP_BLKS_SCANNED, + pgstat_progress_update_param(PROGRESS_REPACK_HEAP_BLKS_SCANNED, (heapScan->rs_cblock + heapScan->rs_nblocks - heapScan->rs_startblock @@ -912,14 +912,14 @@ heapam_relation_copy_for_cluster(Relation OldHeap, Re= lation NewHeap, * In scan-and-sort mode, report increase in number of tuples * scanned */ - pgstat_progress_update_param(PROGRESS_CLUSTER_HEAP_TUPLES_SCANNED, + pgstat_progress_update_param(PROGRESS_REPACK_HEAP_TUPLES_SCANNED, *num_tuples); } else { const int ct_index[] =3D { - PROGRESS_CLUSTER_HEAP_TUPLES_SCANNED, - PROGRESS_CLUSTER_HEAP_TUPLES_WRITTEN + PROGRESS_REPACK_HEAP_TUPLES_SCANNED, + PROGRESS_REPACK_HEAP_TUPLES_WRITTEN }; int64 ct_val[2]; =20 @@ -952,14 +952,14 @@ heapam_relation_copy_for_cluster(Relation OldHeap, Re= lation NewHeap, double n_tuples =3D 0; =20 /* Report that we are now sorting tuples */ - pgstat_progress_update_param(PROGRESS_CLUSTER_PHASE, - PROGRESS_CLUSTER_PHASE_SORT_TUPLES); + pgstat_progress_update_param(PROGRESS_REPACK_PHASE, + PROGRESS_REPACK_PHASE_SORT_TUPLES); =20 tuplesort_performsort(tuplesort); =20 /* Report that we are now writing new heap */ - pgstat_progress_update_param(PROGRESS_CLUSTER_PHASE, - PROGRESS_CLUSTER_PHASE_WRITE_NEW_HEAP); + pgstat_progress_update_param(PROGRESS_REPACK_PHASE, + PROGRESS_REPACK_PHASE_WRITE_NEW_HEAP); =20 for (;;) { @@ -977,7 +977,7 @@ heapam_relation_copy_for_cluster(Relation OldHeap, Rela= tion NewHeap, values, isnull, rwstate); /* Report n_tuples */ - pgstat_progress_update_param(PROGRESS_CLUSTER_HEAP_TUPLES_WRITTEN, + pgstat_progress_update_param(PROGRESS_REPACK_HEAP_TUPLES_WRITTEN, n_tuples); } =20 diff --git a/src/backend/catalog/index.c b/src/backend/catalog/index.c index 5d9db167e59..08d4b8e44d7 100644 --- a/src/backend/catalog/index.c +++ b/src/backend/catalog/index.c @@ -4079,7 +4079,7 @@ reindex_relation(const ReindexStmt *stmt, Oid relid, = int flags, Assert(!ReindexIsProcessingIndex(indexOid)); =20 /* Set index rebuild count */ - pgstat_progress_update_param(PROGRESS_CLUSTER_INDEX_REBUILD_COUNT, + pgstat_progress_update_param(PROGRESS_REPACK_INDEX_REBUILD_COUNT, i); i++; } diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/sys= tem_views.sql index 086c4c8fb6f..024d219016d 100644 --- a/src/backend/catalog/system_views.sql +++ b/src/backend/catalog/system_views.sql @@ -1272,14 +1272,15 @@ CREATE VIEW pg_stat_progress_vacuum AS FROM pg_stat_get_progress_info('VACUUM') AS S LEFT JOIN pg_database D ON S.datid =3D D.oid; =20 -CREATE VIEW pg_stat_progress_cluster AS +CREATE VIEW pg_stat_progress_repack AS SELECT S.pid AS pid, S.datid AS datid, D.datname AS datname, S.relid AS relid, CASE S.param1 WHEN 1 THEN 'CLUSTER' - WHEN 2 THEN 'VACUUM FULL' + WHEN 2 THEN 'REPACK' + WHEN 3 THEN 'VACUUM FULL' END AS command, CASE S.param2 WHEN 0 THEN 'initializing' WHEN 1 THEN 'seq scanning heap' @@ -1290,15 +1291,35 @@ CREATE VIEW pg_stat_progress_cluster AS WHEN 6 THEN 'rebuilding index' WHEN 7 THEN 'performing final cleanup' END AS phase, - CAST(S.param3 AS oid) AS cluster_index_relid, + CAST(S.param3 AS oid) AS repack_index_relid, S.param4 AS heap_tuples_scanned, S.param5 AS heap_tuples_written, S.param6 AS heap_blks_total, S.param7 AS heap_blks_scanned, S.param8 AS index_rebuild_count - FROM pg_stat_get_progress_info('CLUSTER') AS S + FROM pg_stat_get_progress_info('REPACK') AS S LEFT JOIN pg_database D ON S.datid =3D D.oid; =20 +-- This view is as the one above, except for renaming a column and avoiding +-- 'REPACK' as a command name to report. +CREATE VIEW pg_stat_progress_cluster AS + SELECT + pid, + datid, + datname, + relid, + CASE WHEN command IN ('CLUSTER', 'VACUUM FULL') THEN command + WHEN repack_index_relid =3D 0 THEN 'VACUUM FULL' + ELSE 'CLUSTER' END AS command, + phase, + repack_index_relid AS cluster_index_relid, + heap_tuples_scanned, + heap_tuples_written, + heap_blks_total, + heap_blks_scanned, + index_rebuild_count + FROM pg_stat_progress_repack; + CREATE VIEW pg_stat_progress_create_index AS SELECT S.pid AS pid, S.datid AS datid, D.datname AS datname, diff --git a/src/backend/commands/cluster.c b/src/backend/commands/cluster.c index b55221d44cd..ba3c076ea7d 100644 --- a/src/backend/commands/cluster.c +++ b/src/backend/commands/cluster.c @@ -1,7 +1,8 @@ /*------------------------------------------------------------------------- * * cluster.c - * CLUSTER a table on an index. This is now also used for VACUUM FULL. + * CLUSTER a table on an index. This is now also used for VACUUM FULL a= nd + * REPACK. * * There is hardly anything left of Paul Brown's original implementation... * @@ -67,27 +68,36 @@ typedef struct Oid indexOid; } RelToCluster; =20 - -static void cluster_multiple_rels(List *rtcs, ClusterParams *params); -static void rebuild_relation(Relation OldHeap, Relation index, bool verbos= e); +static bool cluster_rel_recheck(RepackCommand cmd, Relation OldHeap, + Oid indexOid, Oid userid, int options); +static void rebuild_relation(RepackCommand cmd, + Relation OldHeap, Relation index, bool verbose); static void copy_table_data(Relation NewHeap, Relation OldHeap, Relation O= ldIndex, bool verbose, bool *pSwapToastByContent, TransactionId *pFreezeXid, MultiXactId *pCutoffMulti); -static List *get_tables_to_cluster(MemoryContext cluster_context); -static List *get_tables_to_cluster_partitioned(MemoryContext cluster_conte= xt, - Oid indexOid); -static bool cluster_is_permitted_for_relation(Oid relid, Oid userid); +static List *get_tables_to_repack(RepackCommand cmd, bool usingindex, + MemoryContext permcxt); +static List *get_tables_to_repack_partitioned(RepackCommand cmd, + Oid relid, bool rel_is_index, + MemoryContext permcxt); +static bool cluster_is_permitted_for_relation(RepackCommand cmd, + Oid relid, Oid userid); +static Relation process_single_relation(RepackStmt *stmt, + ClusterParams *params); +static Oid determine_clustered_index(Relation rel, bool usingindex, + const char *indexname); +static const char *RepackCommandAsString(RepackCommand cmd); =20 =20 -/*------------------------------------------------------------------------= --- - * This cluster code allows for clustering multiple tables at once. Because +/* + * The repack code allows for processing multiple tables at once. Because * of this, we cannot just run everything on a single transaction, or we * would be forced to acquire exclusive locks on all the tables being * clustered, simultaneously --- very likely leading to deadlock. * - * To solve this we follow a similar strategy to VACUUM code, - * clustering each relation in a separate transaction. For this to work, - * we need to: + * To solve this we follow a similar strategy to VACUUM code, processing e= ach + * relation in a separate transaction. For this to work, we need to: + * * - provide a separate memory context so that we can pass information in * a way that survives across transactions * - start a new transaction every time a new relation is clustered @@ -98,197 +108,165 @@ static bool cluster_is_permitted_for_relation(Oid re= lid, Oid userid); * * The single-relation case does not have any such overhead. * - * We also allow a relation to be specified without index. In that case, - * the indisclustered bit will be looked up, and an ERROR will be thrown - * if there is no index with the bit set. - *------------------------------------------------------------------------= --- + * We also allow a relation to be repacked following an index, but without + * naming a specific one. In that case, the indisclustered bit will be + * looked up, and an ERROR will be thrown if no so-marked index is found. */ void -cluster(ParseState *pstate, ClusterStmt *stmt, bool isTopLevel) +ExecRepack(ParseState *pstate, RepackStmt *stmt, bool isTopLevel) { - ListCell *lc; ClusterParams params =3D {0}; - bool verbose =3D false; Relation rel =3D NULL; - Oid indexOid =3D InvalidOid; - MemoryContext cluster_context; + MemoryContext repack_context; List *rtcs; =20 /* Parse option list */ - foreach(lc, stmt->params) + foreach_node(DefElem, opt, stmt->params) { - DefElem *opt =3D (DefElem *) lfirst(lc); - if (strcmp(opt->defname, "verbose") =3D=3D 0) - verbose =3D defGetBoolean(opt); + params.options |=3D defGetBoolean(opt) ? CLUOPT_VERBOSE : 0; + else if (strcmp(opt->defname, "analyze") =3D=3D 0 || + strcmp(opt->defname, "analyse") =3D=3D 0) + params.options |=3D defGetBoolean(opt) ? CLUOPT_ANALYZE : 0; else ereport(ERROR, - (errcode(ERRCODE_SYNTAX_ERROR), - errmsg("unrecognized CLUSTER option \"%s\"", - opt->defname), - parser_errposition(pstate, opt->location))); + errcode(ERRCODE_SYNTAX_ERROR), + errmsg("unrecognized %s option \"%s\"", + RepackCommandAsString(stmt->command), + opt->defname), + parser_errposition(pstate, opt->location)); } =20 - params.options =3D (verbose ? CLUOPT_VERBOSE : 0); - + /* + * If a single relation is specified, process it and we're done ... unless + * the relation is a partitioned table, in which case we fall through. + */ if (stmt->relation !=3D NULL) { - /* This is the single-relation case. */ - Oid tableOid; - - /* - * Find, lock, and check permissions on the table. We obtain - * AccessExclusiveLock right away to avoid lock-upgrade hazard in the - * single-transaction case. - */ - tableOid =3D RangeVarGetRelidExtended(stmt->relation, - AccessExclusiveLock, - 0, - RangeVarCallbackMaintainsTable, - NULL); - rel =3D table_open(tableOid, NoLock); - - /* - * Reject clustering a remote temp table ... their local buffer - * manager is not going to cope. - */ - if (RELATION_IS_OTHER_TEMP(rel)) - ereport(ERROR, - (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), - errmsg("cannot cluster temporary tables of other sessions"))); - - if (stmt->indexname =3D=3D NULL) - { - ListCell *index; - - /* We need to find the index that has indisclustered set. */ - foreach(index, RelationGetIndexList(rel)) - { - indexOid =3D lfirst_oid(index); - if (get_index_isclustered(indexOid)) - break; - indexOid =3D InvalidOid; - } - - if (!OidIsValid(indexOid)) - ereport(ERROR, - (errcode(ERRCODE_UNDEFINED_OBJECT), - errmsg("there is no previously clustered index for table \"%s\"", - stmt->relation->relname))); - } - else - { - /* - * The index is expected to be in the same namespace as the - * relation. - */ - indexOid =3D get_relname_relid(stmt->indexname, - rel->rd_rel->relnamespace); - if (!OidIsValid(indexOid)) - ereport(ERROR, - (errcode(ERRCODE_UNDEFINED_OBJECT), - errmsg("index \"%s\" for table \"%s\" does not exist", - stmt->indexname, stmt->relation->relname))); - } - - /* For non-partitioned tables, do what we came here to do. */ - if (rel->rd_rel->relkind !=3D RELKIND_PARTITIONED_TABLE) - { - cluster_rel(rel, indexOid, ¶ms); - /* cluster_rel closes the relation, but keeps lock */ - - return; - } + rel =3D process_single_relation(stmt, ¶ms); + if (rel =3D=3D NULL) + return; /* all done */ } =20 + /* + * Don't allow ANALYZE in the multiple-relation case for now. Maybe we + * can add support for this later. + */ + if (params.options & CLUOPT_ANALYZE) + ereport(ERROR, + errcode(ERRCODE_FEATURE_NOT_SUPPORTED), + errmsg("cannot %s multiple tables", "REPACK (ANALYZE)")); + /* * By here, we know we are in a multi-table situation. In order to avoid * holding locks for too long, we want to process each table in its own * transaction. This forces us to disallow running inside a user * transaction block. */ - PreventInTransactionBlock(isTopLevel, "CLUSTER"); + PreventInTransactionBlock(isTopLevel, RepackCommandAsString(stmt->command= )); =20 /* Also, we need a memory context to hold our list of relations */ - cluster_context =3D AllocSetContextCreate(PortalContext, - "Cluster", - ALLOCSET_DEFAULT_SIZES); + repack_context =3D AllocSetContextCreate(PortalContext, + "Repack", + ALLOCSET_DEFAULT_SIZES); + + params.options |=3D CLUOPT_RECHECK; =20 /* - * Either we're processing a partitioned table, or we were not given any - * table name at all. In either case, obtain a list of relations to - * process. - * - * In the former case, an index name must have been given, so we don't - * need to recheck its "indisclustered" bit, but we have to check that it - * is an index that we can cluster on. In the latter case, we set the - * option bit to have indisclustered verified. - * - * Rechecking the relation itself is necessary here in all cases. + * If we don't have a relation yet, determine a relation list. If we do, + * then it must be a partitioned table, and we want to process its + * partitions. */ - params.options |=3D CLUOPT_RECHECK; - if (rel !=3D NULL) + if (rel =3D=3D NULL) { - Assert(rel->rd_rel->relkind =3D=3D RELKIND_PARTITIONED_TABLE); - check_index_is_clusterable(rel, indexOid, AccessShareLock); - rtcs =3D get_tables_to_cluster_partitioned(cluster_context, indexOid); - - /* close relation, releasing lock on parent table */ - table_close(rel, AccessExclusiveLock); + Assert(stmt->indexname =3D=3D NULL); + rtcs =3D get_tables_to_repack(stmt->command, stmt->usingindex, + repack_context); } else { - rtcs =3D get_tables_to_cluster(cluster_context); - params.options |=3D CLUOPT_RECHECK_ISCLUSTERED; - } + Oid relid; + bool rel_is_index; =20 - /* Do the job. */ - cluster_multiple_rels(rtcs, ¶ms); + Assert(rel->rd_rel->relkind =3D=3D RELKIND_PARTITIONED_TABLE); =20 - /* Start a new transaction for the cleanup work. */ - StartTransactionCommand(); + /* + * If USING INDEX was specified, resolve the index name now and pass + * it down. + */ + if (stmt->usingindex) + { + /* + * If no index name was specified when repacking a partitioned + * table, punt for now. Maybe we can improve this later. + */ + if (!stmt->indexname) + ereport(ERROR, + errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), + errmsg("there is no previously clustered index for table \"%s\"", + RelationGetRelationName(rel))); + + relid =3D determine_clustered_index(rel, stmt->usingindex, + stmt->indexname); + if (!OidIsValid(relid)) + elog(ERROR, "unable to determine index to cluster on"); + /* XXX is this the right place for this check? */ + check_index_is_clusterable(rel, relid, AccessExclusiveLock); + rel_is_index =3D true; + } + else + { + relid =3D RelationGetRelid(rel); + rel_is_index =3D false; + } =20 - /* Clean up working storage */ - MemoryContextDelete(cluster_context); -} + rtcs =3D get_tables_to_repack_partitioned(stmt->command, + relid, rel_is_index, + repack_context); =20 -/* - * Given a list of relations to cluster, process each of them in a separate - * transaction. - * - * We expect to be in a transaction at start, but there isn't one when we - * return. - */ -static void -cluster_multiple_rels(List *rtcs, ClusterParams *params) -{ - ListCell *lc; + /* close parent relation, releasing lock on it */ + table_close(rel, AccessExclusiveLock); + rel =3D NULL; + } =20 /* Commit to get out of starting transaction */ PopActiveSnapshot(); CommitTransactionCommand(); =20 /* Cluster the tables, each in a separate transaction */ - foreach(lc, rtcs) + Assert(rel =3D=3D NULL); + foreach_ptr(RelToCluster, rtc, rtcs) { - RelToCluster *rtc =3D (RelToCluster *) lfirst(lc); - Relation rel; - /* Start a new transaction for each relation. */ StartTransactionCommand(); =20 + /* + * Open the target table, coping with the case where it has been + * dropped. + */ + rel =3D try_table_open(rtc->tableOid, AccessExclusiveLock); + if (rel =3D=3D NULL) + { + CommitTransactionCommand(); + continue; + } + /* functions in indexes may want a snapshot set */ PushActiveSnapshot(GetTransactionSnapshot()); =20 - rel =3D table_open(rtc->tableOid, AccessExclusiveLock); - /* Process this table */ - cluster_rel(rel, rtc->indexOid, params); + cluster_rel(stmt->command, rel, rtc->indexOid, ¶ms); /* cluster_rel closes the relation, but keeps lock */ =20 PopActiveSnapshot(); CommitTransactionCommand(); } + + /* Start a new transaction for the cleanup work. */ + StartTransactionCommand(); + + /* Clean up working storage */ + MemoryContextDelete(repack_context); } =20 /* @@ -304,11 +282,14 @@ cluster_multiple_rels(List *rtcs, ClusterParams *para= ms) * them incrementally while we load the table. * * If indexOid is InvalidOid, the table will be rewritten in physical order - * instead of index order. This is the new implementation of VACUUM FULL, - * and error messages should refer to the operation as VACUUM not CLUSTER. + * instead of index order. + * + * 'cmd' indicates which command is being executed, to be used for error + * messages. */ void -cluster_rel(Relation OldHeap, Oid indexOid, ClusterParams *params) +cluster_rel(RepackCommand cmd, Relation OldHeap, Oid indexOid, + ClusterParams *params) { Oid tableOid =3D RelationGetRelid(OldHeap); Oid save_userid; @@ -323,13 +304,8 @@ cluster_rel(Relation OldHeap, Oid indexOid, ClusterPar= ams *params) /* Check for user-requested abort. */ CHECK_FOR_INTERRUPTS(); =20 - pgstat_progress_start_command(PROGRESS_COMMAND_CLUSTER, tableOid); - if (OidIsValid(indexOid)) - pgstat_progress_update_param(PROGRESS_CLUSTER_COMMAND, - PROGRESS_CLUSTER_COMMAND_CLUSTER); - else - pgstat_progress_update_param(PROGRESS_CLUSTER_COMMAND, - PROGRESS_CLUSTER_COMMAND_VACUUM_FULL); + pgstat_progress_start_command(PROGRESS_COMMAND_REPACK, tableOid); + pgstat_progress_update_param(PROGRESS_REPACK_COMMAND, cmd); =20 /* * Switch to the table owner's userid, so that any index functions are run @@ -350,86 +326,38 @@ cluster_rel(Relation OldHeap, Oid indexOid, ClusterPa= rams *params) * *must* skip the one on indisclustered since it would reject an attempt * to cluster a not-previously-clustered index. */ - if (recheck) - { - /* Check that the user still has privileges for the relation */ - if (!cluster_is_permitted_for_relation(tableOid, save_userid)) - { - relation_close(OldHeap, AccessExclusiveLock); - goto out; - } - - /* - * Silently skip a temp table for a remote session. Only doing this - * check in the "recheck" case is appropriate (which currently means - * somebody is executing a database-wide CLUSTER or on a partitioned - * table), because there is another check in cluster() which will stop - * any attempt to cluster remote temp tables by name. There is - * another check in cluster_rel which is redundant, but we leave it - * for extra safety. - */ - if (RELATION_IS_OTHER_TEMP(OldHeap)) - { - relation_close(OldHeap, AccessExclusiveLock); - goto out; - } - - if (OidIsValid(indexOid)) - { - /* - * Check that the index still exists - */ - if (!SearchSysCacheExists1(RELOID, ObjectIdGetDatum(indexOid))) - { - relation_close(OldHeap, AccessExclusiveLock); - goto out; - } - - /* - * Check that the index is still the one with indisclustered set, - * if needed. - */ - if ((params->options & CLUOPT_RECHECK_ISCLUSTERED) !=3D 0 && - !get_index_isclustered(indexOid)) - { - relation_close(OldHeap, AccessExclusiveLock); - goto out; - } - } - } + if (recheck && + !cluster_rel_recheck(cmd, OldHeap, indexOid, save_userid, + params->options)) + goto out; =20 /* - * We allow VACUUM FULL, but not CLUSTER, on shared catalogs. CLUSTER - * would work in most respects, but the index would only get marked as - * indisclustered in the current database, leading to unexpected behavior - * if CLUSTER were later invoked in another database. + * We allow repacking shared catalogs only when not using an index. It + * would work to use an index in most respects, but the index would only + * get marked as indisclustered in the current database, leading to + * unexpected behavior if CLUSTER were later invoked in another database. */ if (OidIsValid(indexOid) && OldHeap->rd_rel->relisshared) ereport(ERROR, - (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), - errmsg("cannot cluster a shared catalog"))); + errcode(ERRCODE_FEATURE_NOT_SUPPORTED), + errmsg("cannot run %s on a shared catalog", + RepackCommandAsString(cmd))); =20 /* * Don't process temp tables of other backends ... their local buffer * manager is not going to cope. */ if (RELATION_IS_OTHER_TEMP(OldHeap)) - { - if (OidIsValid(indexOid)) - ereport(ERROR, - (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), - errmsg("cannot cluster temporary tables of other sessions"))); - else - ereport(ERROR, - (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), - errmsg("cannot vacuum temporary tables of other sessions"))); - } + ereport(ERROR, + errcode(ERRCODE_FEATURE_NOT_SUPPORTED), + errmsg("cannot run %s on temporary tables of other sessions", + RepackCommandAsString(cmd))); =20 /* * Also check for active uses of the relation in the current transaction, * including open scans and pending AFTER trigger events. */ - CheckTableNotInUse(OldHeap, OidIsValid(indexOid) ? "CLUSTER" : "VACUUM"); + CheckTableNotInUse(OldHeap, RepackCommandAsString(cmd)); =20 /* Check heap and index are valid to cluster on */ if (OidIsValid(indexOid)) @@ -442,6 +370,24 @@ cluster_rel(Relation OldHeap, Oid indexOid, ClusterPar= ams *params) else index =3D NULL; =20 + /* + * When allow_system_table_mods is turned off, we disallow repacking a + * catalog on a particular index unless that's already the clustered index + * for that catalog. + * + * XXX We don't check for this in CLUSTER, because it's historically been + * allowed. + */ + if (cmd !=3D REPACK_COMMAND_CLUSTER && + !allowSystemTableMods && OidIsValid(indexOid) && + IsCatalogRelation(OldHeap) && !index->rd_index->indisclustered) + ereport(ERROR, + errcode(ERRCODE_INSUFFICIENT_PRIVILEGE), + errmsg("permission denied: \"%s\" is a system catalog", + RelationGetRelationName(OldHeap)), + errdetail("System catalogs can only be clustered by the index they're = already clustered on, if any, unless \"%s\" is enabled.", + "allow_system_table_mods")); + /* * Quietly ignore the request if this is a materialized view which has not * been populated from its query. No harm is done because there is no data @@ -469,7 +415,7 @@ cluster_rel(Relation OldHeap, Oid indexOid, ClusterPara= ms *params) TransferPredicateLocksToHeapRelation(OldHeap); =20 /* rebuild_relation does all the dirty work */ - rebuild_relation(OldHeap, index, verbose); + rebuild_relation(cmd, OldHeap, index, verbose); /* rebuild_relation closes OldHeap, and index if valid */ =20 out: @@ -482,6 +428,63 @@ out: pgstat_progress_end_command(); } =20 +/* + * Check if the table (and its index) still meets the requirements of + * cluster_rel(). + */ +static bool +cluster_rel_recheck(RepackCommand cmd, Relation OldHeap, Oid indexOid, + Oid userid, int options) +{ + Oid tableOid =3D RelationGetRelid(OldHeap); + + /* Check that the user still has privileges for the relation */ + if (!cluster_is_permitted_for_relation(cmd, tableOid, userid)) + { + relation_close(OldHeap, AccessExclusiveLock); + return false; + } + + /* + * Silently skip a temp table for a remote session. Only doing this check + * in the "recheck" case is appropriate (which currently means somebody is + * executing a database-wide CLUSTER or on a partitioned table), because + * there is another check in cluster() which will stop any attempt to + * cluster remote temp tables by name. There is another check in + * cluster_rel which is redundant, but we leave it for extra safety. + */ + if (RELATION_IS_OTHER_TEMP(OldHeap)) + { + relation_close(OldHeap, AccessExclusiveLock); + return false; + } + + if (OidIsValid(indexOid)) + { + /* + * Check that the index still exists + */ + if (!SearchSysCacheExists1(RELOID, ObjectIdGetDatum(indexOid))) + { + relation_close(OldHeap, AccessExclusiveLock); + return false; + } + + /* + * Check that the index is still the one with indisclustered set, if + * needed. + */ + if ((options & CLUOPT_RECHECK_ISCLUSTERED) !=3D 0 && + !get_index_isclustered(indexOid)) + { + relation_close(OldHeap, AccessExclusiveLock); + return false; + } + } + + return true; +} + /* * Verify that the specified heap and index are valid to cluster on * @@ -626,7 +629,8 @@ mark_index_clustered(Relation rel, Oid indexOid, bool i= s_internal) * On exit, they are closed, but locks on them are not released. */ static void -rebuild_relation(Relation OldHeap, Relation index, bool verbose) +rebuild_relation(RepackCommand cmd, + Relation OldHeap, Relation index, bool verbose) { Oid tableOid =3D RelationGetRelid(OldHeap); Oid accessMethod =3D OldHeap->rd_rel->relam; @@ -642,8 +646,8 @@ rebuild_relation(Relation OldHeap, Relation index, bool= verbose) Assert(CheckRelationLockedByMe(OldHeap, AccessExclusiveLock, false) && (index =3D=3D NULL || CheckRelationLockedByMe(index, AccessExclusiveL= ock, false))); =20 - if (index) - /* Mark the correct index as clustered */ + /* for CLUSTER or REPACK USING INDEX, mark the index as the one to use */ + if (index !=3D NULL) mark_index_clustered(OldHeap, RelationGetRelid(index), true); =20 /* Remember info about rel before closing OldHeap */ @@ -958,20 +962,20 @@ copy_table_data(Relation NewHeap, Relation OldHeap, R= elation OldIndex, bool verb /* Log what we're doing */ if (OldIndex !=3D NULL && !use_sort) ereport(elevel, - (errmsg("clustering \"%s.%s\" using index scan on \"%s\"", - nspname, - RelationGetRelationName(OldHeap), - RelationGetRelationName(OldIndex)))); + errmsg("repacking \"%s.%s\" using index scan on \"%s\"", + nspname, + RelationGetRelationName(OldHeap), + RelationGetRelationName(OldIndex))); else if (use_sort) ereport(elevel, - (errmsg("clustering \"%s.%s\" using sequential scan and sort", - nspname, - RelationGetRelationName(OldHeap)))); + errmsg("repacking \"%s.%s\" using sequential scan and sort", + nspname, + RelationGetRelationName(OldHeap))); else ereport(elevel, - (errmsg("vacuuming \"%s.%s\"", - nspname, - RelationGetRelationName(OldHeap)))); + errmsg("repacking \"%s.%s\" in physical order", + nspname, + RelationGetRelationName(OldHeap))); =20 /* * Hand off the actual copying to AM specific function, the generic code @@ -1458,8 +1462,8 @@ finish_heap_swap(Oid OIDOldHeap, Oid OIDNewHeap, int i; =20 /* Report that we are now swapping relation files */ - pgstat_progress_update_param(PROGRESS_CLUSTER_PHASE, - PROGRESS_CLUSTER_PHASE_SWAP_REL_FILES); + pgstat_progress_update_param(PROGRESS_REPACK_PHASE, + PROGRESS_REPACK_PHASE_SWAP_REL_FILES); =20 /* Zero out possible results from swapped_relation_files */ memset(mapped_tables, 0, sizeof(mapped_tables)); @@ -1509,14 +1513,14 @@ finish_heap_swap(Oid OIDOldHeap, Oid OIDNewHeap, reindex_flags |=3D REINDEX_REL_FORCE_INDEXES_PERMANENT; =20 /* Report that we are now reindexing relations */ - pgstat_progress_update_param(PROGRESS_CLUSTER_PHASE, - PROGRESS_CLUSTER_PHASE_REBUILD_INDEX); + pgstat_progress_update_param(PROGRESS_REPACK_PHASE, + PROGRESS_REPACK_PHASE_REBUILD_INDEX); =20 reindex_relation(NULL, OIDOldHeap, reindex_flags, &reindex_params); =20 /* Report that we are now doing clean up */ - pgstat_progress_update_param(PROGRESS_CLUSTER_PHASE, - PROGRESS_CLUSTER_PHASE_FINAL_CLEANUP); + pgstat_progress_update_param(PROGRESS_REPACK_PHASE, + PROGRESS_REPACK_PHASE_FINAL_CLEANUP); =20 /* * If the relation being rebuilt is pg_class, swap_relation_files() @@ -1632,106 +1636,191 @@ finish_heap_swap(Oid OIDOldHeap, Oid OIDNewHeap, } } =20 - /* - * Get a list of tables that the current user has privileges on and - * have indisclustered set. Return the list in a List * of RelToCluster - * (stored in the specified memory context), each one giving the tableOid - * and the indexOid on which the table is already clustered. + * Determine which relations to process, when REPACK/CLUSTER is called + * without specifying a table name. The exact process depends on whether + * USING INDEX was given or not, and in any case we only return tables and + * materialized views that the current user has privileges to repack/clust= er. + * + * If USING INDEX was given, we scan pg_index to find those that have + * indisclustered set; if it was not given, scan pg_class and return all + * tables. + * + * Return it as a list of RelToCluster in the given memory context. */ static List * -get_tables_to_cluster(MemoryContext cluster_context) +get_tables_to_repack(RepackCommand cmd, bool usingindex, MemoryContext per= mcxt) { - Relation indRelation; + Relation catalog; TableScanDesc scan; - ScanKeyData entry; - HeapTuple indexTuple; - Form_pg_index index; - MemoryContext old_context; + HeapTuple tuple; List *rtcs =3D NIL; =20 - /* - * Get all indexes that have indisclustered set and that the current user - * has the appropriate privileges for. - */ - indRelation =3D table_open(IndexRelationId, AccessShareLock); - ScanKeyInit(&entry, - Anum_pg_index_indisclustered, - BTEqualStrategyNumber, F_BOOLEQ, - BoolGetDatum(true)); - scan =3D table_beginscan_catalog(indRelation, 1, &entry); - while ((indexTuple =3D heap_getnext(scan, ForwardScanDirection)) !=3D NUL= L) + if (usingindex) { - RelToCluster *rtc; + ScanKeyData entry; + + catalog =3D table_open(IndexRelationId, AccessShareLock); + ScanKeyInit(&entry, + Anum_pg_index_indisclustered, + BTEqualStrategyNumber, F_BOOLEQ, + BoolGetDatum(true)); + scan =3D table_beginscan_catalog(catalog, 1, &entry); + while ((tuple =3D heap_getnext(scan, ForwardScanDirection)) !=3D NULL) + { + RelToCluster *rtc; + Form_pg_index index; + MemoryContext oldcxt; =20 - index =3D (Form_pg_index) GETSTRUCT(indexTuple); + index =3D (Form_pg_index) GETSTRUCT(tuple); =20 - if (!cluster_is_permitted_for_relation(index->indrelid, GetUserId())) - continue; + /* + * Try to obtain a light lock on the index's table, to ensure it + * doesn't go away while we collect the list. If we cannot, just + * disregard it. + */ + if (!ConditionalLockRelationOid(index->indrelid, AccessShareLock)) + continue; =20 - /* Use a permanent memory context for the result list */ - old_context =3D MemoryContextSwitchTo(cluster_context); + /* Verify that the table still exists */ + if (!SearchSysCacheExists1(RELOID, ObjectIdGetDatum(index->indrelid))) + { + /* Release useless lock */ + UnlockRelationOid(index->indrelid, AccessShareLock); + continue; + } =20 - rtc =3D (RelToCluster *) palloc(sizeof(RelToCluster)); - rtc->tableOid =3D index->indrelid; - rtc->indexOid =3D index->indexrelid; - rtcs =3D lappend(rtcs, rtc); + if (!cluster_is_permitted_for_relation(cmd, index->indrelid, + GetUserId())) + continue; + + /* Use a permanent memory context for the result list */ + oldcxt =3D MemoryContextSwitchTo(permcxt); + rtc =3D palloc(sizeof(RelToCluster)); + rtc->tableOid =3D index->indrelid; + rtc->indexOid =3D index->indexrelid; + rtcs =3D lappend(rtcs, rtc); + MemoryContextSwitchTo(oldcxt); + } + } + else + { + catalog =3D table_open(RelationRelationId, AccessShareLock); + scan =3D table_beginscan_catalog(catalog, 0, NULL); + + while ((tuple =3D heap_getnext(scan, ForwardScanDirection)) !=3D NULL) + { + RelToCluster *rtc; + Form_pg_class class; + MemoryContext oldcxt; + + class =3D (Form_pg_class) GETSTRUCT(tuple); + + /* + * Try to obtain a light lock on the table, to ensure it doesn't + * go away while we collect the list. If we cannot, just + * disregard the table. + */ + if (!ConditionalLockRelationOid(class->oid, AccessShareLock)) + continue; =20 - MemoryContextSwitchTo(old_context); + /* Verify that the table still exists */ + if (!SearchSysCacheExists1(RELOID, ObjectIdGetDatum(class->oid))) + { + /* Release useless lock */ + UnlockRelationOid(class->oid, AccessShareLock); + continue; + } + + /* Can only process plain tables and matviews */ + if (class->relkind !=3D RELKIND_RELATION && + class->relkind !=3D RELKIND_MATVIEW) + continue; + + /* noisily skip rels which the user can't process */ + if (!cluster_is_permitted_for_relation(cmd, class->oid, + GetUserId())) + continue; + + /* Use a permanent memory context for the result list */ + oldcxt =3D MemoryContextSwitchTo(permcxt); + rtc =3D palloc(sizeof(RelToCluster)); + rtc->tableOid =3D class->oid; + rtc->indexOid =3D InvalidOid; + rtcs =3D lappend(rtcs, rtc); + MemoryContextSwitchTo(oldcxt); + } } - table_endscan(scan); =20 - relation_close(indRelation, AccessShareLock); + table_endscan(scan); + relation_close(catalog, AccessShareLock); =20 return rtcs; } =20 /* - * Given an index on a partitioned table, return a list of RelToCluster for + * Given a partitioned table or its index, return a list of RelToCluster f= or * all the children leaves tables/indexes. * * Like expand_vacuum_rel, but here caller must hold AccessExclusiveLock * on the table containing the index. + * + * 'rel_is_index' tells whether 'relid' is that of an index (true) or of t= he + * owning relation. */ static List * -get_tables_to_cluster_partitioned(MemoryContext cluster_context, Oid index= Oid) +get_tables_to_repack_partitioned(RepackCommand cmd, Oid relid, + bool rel_is_index, MemoryContext permcxt) { List *inhoids; - ListCell *lc; List *rtcs =3D NIL; - MemoryContext old_context; =20 - /* Do not lock the children until they're processed */ - inhoids =3D find_all_inheritors(indexOid, NoLock, NULL); - - foreach(lc, inhoids) + /* + * Do not lock the children until they're processed. Note that we do hold + * a lock on the parent partitioned table. + */ + inhoids =3D find_all_inheritors(relid, NoLock, NULL); + foreach_oid(child_oid, inhoids) { - Oid indexrelid =3D lfirst_oid(lc); - Oid relid =3D IndexGetRelation(indexrelid, false); + Oid table_oid, + index_oid; RelToCluster *rtc; + MemoryContext oldcxt; =20 - /* consider only leaf indexes */ - if (get_rel_relkind(indexrelid) !=3D RELKIND_INDEX) - continue; + if (rel_is_index) + { + /* consider only leaf indexes */ + if (get_rel_relkind(child_oid) !=3D RELKIND_INDEX) + continue; + + table_oid =3D IndexGetRelation(child_oid, false); + index_oid =3D child_oid; + } + else + { + /* consider only leaf relations */ + if (get_rel_relkind(child_oid) !=3D RELKIND_RELATION) + continue; + + table_oid =3D child_oid; + index_oid =3D InvalidOid; + } =20 /* * It's possible that the user does not have privileges to CLUSTER the - * leaf partition despite having such privileges on the partitioned - * table. We skip any partitions which the user is not permitted to - * CLUSTER. + * leaf partition despite having them on the partitioned table. Skip + * if so. */ - if (!cluster_is_permitted_for_relation(relid, GetUserId())) + if (!cluster_is_permitted_for_relation(cmd, table_oid, GetUserId())) continue; =20 /* Use a permanent memory context for the result list */ - old_context =3D MemoryContextSwitchTo(cluster_context); - - rtc =3D (RelToCluster *) palloc(sizeof(RelToCluster)); - rtc->tableOid =3D relid; - rtc->indexOid =3D indexrelid; + oldcxt =3D MemoryContextSwitchTo(permcxt); + rtc =3D palloc(sizeof(RelToCluster)); + rtc->tableOid =3D table_oid; + rtc->indexOid =3D index_oid; rtcs =3D lappend(rtcs, rtc); - - MemoryContextSwitchTo(old_context); + MemoryContextSwitchTo(oldcxt); } =20 return rtcs; @@ -1742,13 +1831,167 @@ get_tables_to_cluster_partitioned(MemoryContext cl= uster_context, Oid indexOid) * function emits a WARNING. */ static bool -cluster_is_permitted_for_relation(Oid relid, Oid userid) +cluster_is_permitted_for_relation(RepackCommand cmd, Oid relid, Oid userid) { + Assert(cmd =3D=3D REPACK_COMMAND_CLUSTER || cmd =3D=3D REPACK_COMMAND_REP= ACK); + if (pg_class_aclcheck(relid, userid, ACL_MAINTAIN) =3D=3D ACLCHECK_OK) return true; =20 ereport(WARNING, - (errmsg("permission denied to cluster \"%s\", skipping it", - get_rel_name(relid)))); + errmsg("permission denied to execute %s on \"%s\", skipping it", + RepackCommandAsString(cmd), + get_rel_name(relid))); + return false; } + + +/* + * Given a RepackStmt with an indicated relation name, resolve the relation + * name, obtain lock on it, then determine what to do based on the relation + * type: if it's table and not partitioned, repack it as indicated (using = an + * existing clustered index, or following the given one), and return NULL. + * + * On the other hand, if the table is partitioned, do nothing further and + * instead return the opened and locked relcache entry, so that caller can + * process the partitions using the multiple-table handling code. In this + * case, if an index name is given, it's up to the caller to resolve it. + */ +static Relation +process_single_relation(RepackStmt *stmt, ClusterParams *params) +{ + Relation rel; + Oid tableOid; + + Assert(stmt->relation !=3D NULL); + Assert(stmt->command =3D=3D REPACK_COMMAND_CLUSTER || + stmt->command =3D=3D REPACK_COMMAND_REPACK); + + /* + * Find, lock, and check permissions on the table. We obtain + * AccessExclusiveLock right away to avoid lock-upgrade hazard in the + * single-transaction case. + */ + tableOid =3D RangeVarGetRelidExtended(stmt->relation->relation, + AccessExclusiveLock, + 0, + RangeVarCallbackMaintainsTable, + NULL); + rel =3D table_open(tableOid, NoLock); + + /* + * Reject clustering a remote temp table ... their local buffer manager is + * not going to cope. + */ + if (RELATION_IS_OTHER_TEMP(rel)) + ereport(ERROR, + errcode(ERRCODE_FEATURE_NOT_SUPPORTED), + errmsg("cannot execute %s on temporary tables of other sessions", + RepackCommandAsString(stmt->command))); + + /* + * Make sure ANALYZE is specified if a column list is present. + */ + if ((params->options & CLUOPT_ANALYZE) =3D=3D 0 && stmt->relation->va_col= s !=3D NIL) + ereport(ERROR, + errcode(ERRCODE_FEATURE_NOT_SUPPORTED), + errmsg("ANALYZE option must be specified when a column list is provide= d")); + + /* + * For partitioned tables, let caller handle this. Otherwise, process it + * here and we're done. + */ + if (rel->rd_rel->relkind =3D=3D RELKIND_PARTITIONED_TABLE) + return rel; + else + { + Oid indexOid; + + indexOid =3D determine_clustered_index(rel, stmt->usingindex, + stmt->indexname); + if (OidIsValid(indexOid)) + check_index_is_clusterable(rel, indexOid, AccessExclusiveLock); + cluster_rel(stmt->command, rel, indexOid, params); + + /* Do an analyze, if requested */ + if (params->options & CLUOPT_ANALYZE) + { + VacuumParams vac_params =3D {0}; + + vac_params.options |=3D VACOPT_ANALYZE; + if (params->options & CLUOPT_VERBOSE) + vac_params.options |=3D VACOPT_VERBOSE; + analyze_rel(RelationGetRelid(rel), NULL, vac_params, + stmt->relation->va_cols, true, NULL); + } + + return NULL; + } +} + +/* + * Given a relation and the usingindex/indexname options in a + * REPACK USING INDEX or CLUSTER command, return the OID of the + * index to use for clustering the table. + * + * Caller must hold lock on the relation so that the set of indexes + * doesn't change, and must call check_index_is_clusterable. + */ +static Oid +determine_clustered_index(Relation rel, bool usingindex, const char *index= name) +{ + Oid indexOid; + + if (indexname =3D=3D NULL && usingindex) + { + /* + * If USING INDEX with no name is given, find a clustered index, or + * error out if none. + */ + indexOid =3D InvalidOid; + foreach_oid(idxoid, RelationGetIndexList(rel)) + { + if (get_index_isclustered(idxoid)) + { + indexOid =3D idxoid; + break; + } + } + + if (!OidIsValid(indexOid)) + ereport(ERROR, + errcode(ERRCODE_UNDEFINED_OBJECT), + errmsg("there is no previously clustered index for table \"%s\"", + RelationGetRelationName(rel))); + } + else if (indexname !=3D NULL) + { + /* An index was specified; obtain its OID. */ + indexOid =3D get_relname_relid(indexname, rel->rd_rel->relnamespace); + if (!OidIsValid(indexOid)) + ereport(ERROR, + errcode(ERRCODE_UNDEFINED_OBJECT), + errmsg("index \"%s\" for table \"%s\" does not exist", + indexname, RelationGetRelationName(rel))); + } + else + indexOid =3D InvalidOid; + + return indexOid; +} + +static const char * +RepackCommandAsString(RepackCommand cmd) +{ + switch (cmd) + { + case REPACK_COMMAND_REPACK: + return "REPACK"; + case REPACK_COMMAND_VACUUMFULL: + return "VACUUM"; + case REPACK_COMMAND_CLUSTER: + return "CLUSTER"; + } + return "???"; +} diff --git a/src/backend/commands/vacuum.c b/src/backend/commands/vacuum.c index e785dd55ce5..827e66724b5 100644 --- a/src/backend/commands/vacuum.c +++ b/src/backend/commands/vacuum.c @@ -349,7 +349,6 @@ ExecVacuum(ParseState *pstate, VacuumStmt *vacstmt, boo= l isTopLevel) } } =20 - /* * Sanity check DISABLE_PAGE_SKIPPING option. */ @@ -2287,8 +2286,9 @@ vacuum_rel(Oid relid, RangeVar *relation, VacuumParam= s params, if ((params.options & VACOPT_VERBOSE) !=3D 0) cluster_params.options |=3D CLUOPT_VERBOSE; =20 - /* VACUUM FULL is now a variant of CLUSTER; see cluster.c */ - cluster_rel(rel, InvalidOid, &cluster_params); + /* VACUUM FULL is a variant of REPACK; see cluster.c */ + cluster_rel(REPACK_COMMAND_VACUUMFULL, rel, InvalidOid, + &cluster_params); /* cluster_rel closes the relation, but keeps lock */ =20 rel =3D NULL; diff --git a/src/backend/parser/gram.y b/src/backend/parser/gram.y index c3a0a354a9c..c314c11e23d 100644 --- a/src/backend/parser/gram.y +++ b/src/backend/parser/gram.y @@ -286,7 +286,7 @@ static Node *makeRecursiveViewSelect(char *relname, Lis= t *aliases, Node *query); AlterCompositeTypeStmt AlterUserMappingStmt AlterRoleStmt AlterRoleSetStmt AlterPolicyStmt AlterStatsStmt AlterDefaultPrivilegesStmt DefACLAction - AnalyzeStmt CallStmt ClosePortalStmt ClusterStmt CommentStmt + AnalyzeStmt CallStmt ClosePortalStmt CommentStmt ConstraintsSetStmt CopyStmt CreateAsStmt CreateCastStmt CreateDomainStmt CreateExtensionStmt CreateGroupStmt CreateOpClassStmt CreateOpFamilyStmt AlterOpFamilyStmt CreatePLangStmt @@ -303,7 +303,7 @@ static Node *makeRecursiveViewSelect(char *relname, Lis= t *aliases, Node *query); GrantStmt GrantRoleStmt ImportForeignSchemaStmt IndexStmt InsertStmt ListenStmt LoadStmt LockStmt MergeStmt NotifyStmt ExplainableStmt Prepar= ableStmt CreateFunctionStmt AlterFunctionStmt ReindexStmt RemoveAggrStmt - RemoveFuncStmt RemoveOperStmt RenameStmt ReturnStmt RevokeStmt RevokeRol= eStmt + RemoveFuncStmt RemoveOperStmt RenameStmt RepackStmt ReturnStmt RevokeStm= t RevokeRoleStmt RuleActionStmt RuleActionStmtOrEmpty RuleStmt SecLabelStmt SelectStmt TransactionStmt TransactionStmtLegacy TruncateSt= mt UnlistenStmt UpdateStmt VacuumStmt @@ -322,7 +322,7 @@ static Node *makeRecursiveViewSelect(char *relname, Lis= t *aliases, Node *query); =20 %type opt_single_name %type opt_qualified_name -%type opt_concurrently +%type opt_concurrently opt_usingindex %type opt_drop_behavior %type opt_utility_option_list %type opt_wait_with_clause @@ -770,7 +770,7 @@ static Node *makeRecursiveViewSelect(char *relname, Lis= t *aliases, Node *query); QUOTE QUOTES =20 RANGE READ REAL REASSIGN RECURSIVE REF_P REFERENCES REFERENCING - REFRESH REINDEX RELATIVE_P RELEASE RENAME REPEATABLE REPLACE REPLICA + REFRESH REINDEX RELATIVE_P RELEASE RENAME REPACK REPEATABLE REPLACE REPLI= CA RESET RESPECT_P RESTART RESTRICT RETURN RETURNING RETURNS REVOKE RIGHT RO= LE ROLLBACK ROLLUP ROUTINE ROUTINES ROW ROWS RULE =20 @@ -1032,7 +1032,6 @@ stmt: | CallStmt | CheckPointStmt | ClosePortalStmt - | ClusterStmt | CommentStmt | ConstraintsSetStmt | CopyStmt @@ -1106,6 +1105,7 @@ stmt: | RemoveFuncStmt | RemoveOperStmt | RenameStmt + | RepackStmt | RevokeStmt | RevokeRoleStmt | RuleStmt @@ -1143,6 +1143,11 @@ opt_concurrently: | /*EMPTY*/ { $$ =3D false; } ; =20 +opt_usingindex: + USING INDEX { $$ =3D true; } + | /* EMPTY */ { $$ =3D false; } + ; + opt_drop_behavior: CASCADE { $$ =3D DROP_CASCADE; } | RESTRICT { $$ =3D DROP_RESTRICT; } @@ -11979,38 +11984,82 @@ CreateConversionStmt: /*************************************************************************= **** * * QUERY: + * REPACK [ (options) ] [ [ ] [ USING INDE= X ] ] + * + * obsolete variants: * CLUSTER (options) [ [ USING ] ] * CLUSTER [VERBOSE] [ [ USING ] ] * CLUSTER [VERBOSE] ON (for pre-8.3) * *************************************************************************= ****/ =20 -ClusterStmt: - CLUSTER '(' utility_option_list ')' qualified_name cluster_index_specif= ication +RepackStmt: + REPACK opt_utility_option_list vacuum_relation USING INDEX name { - ClusterStmt *n =3D makeNode(ClusterStmt); + RepackStmt *n =3D makeNode(RepackStmt); =20 - n->relation =3D $5; + n->command =3D REPACK_COMMAND_REPACK; + n->relation =3D (VacuumRelation *) $3; n->indexname =3D $6; + n->usingindex =3D true; + n->params =3D $2; + $$ =3D (Node *) n; + } + | REPACK opt_utility_option_list vacuum_relation opt_usingindex + { + RepackStmt *n =3D makeNode(RepackStmt); + + n->command =3D REPACK_COMMAND_REPACK; + n->relation =3D (VacuumRelation *) $3; + n->indexname =3D NULL; + n->usingindex =3D $4; + n->params =3D $2; + $$ =3D (Node *) n; + } + | REPACK opt_utility_option_list opt_usingindex + { + RepackStmt *n =3D makeNode(RepackStmt); + + n->command =3D REPACK_COMMAND_REPACK; + n->relation =3D NULL; + n->indexname =3D NULL; + n->usingindex =3D $3; + n->params =3D $2; + $$ =3D (Node *) n; + } + | CLUSTER '(' utility_option_list ')' qualified_name cluster_index_spec= ification + { + RepackStmt *n =3D makeNode(RepackStmt); + + n->command =3D REPACK_COMMAND_CLUSTER; + n->relation =3D makeNode(VacuumRelation); + n->relation->relation =3D $5; + n->indexname =3D $6; + n->usingindex =3D true; n->params =3D $3; $$ =3D (Node *) n; } | CLUSTER opt_utility_option_list { - ClusterStmt *n =3D makeNode(ClusterStmt); + RepackStmt *n =3D makeNode(RepackStmt); =20 + n->command =3D REPACK_COMMAND_CLUSTER; n->relation =3D NULL; n->indexname =3D NULL; + n->usingindex =3D true; n->params =3D $2; $$ =3D (Node *) n; } /* unparenthesized VERBOSE kept for pre-14 compatibility */ | CLUSTER opt_verbose qualified_name cluster_index_specification { - ClusterStmt *n =3D makeNode(ClusterStmt); + RepackStmt *n =3D makeNode(RepackStmt); =20 - n->relation =3D $3; + n->command =3D REPACK_COMMAND_CLUSTER; + n->relation =3D makeNode(VacuumRelation); + n->relation->relation =3D $3; n->indexname =3D $4; + n->usingindex =3D true; if ($2) n->params =3D list_make1(makeDefElem("verbose", NULL, @2)); $$ =3D (Node *) n; @@ -12018,20 +12067,25 @@ ClusterStmt: /* unparenthesized VERBOSE kept for pre-17 compatibility */ | CLUSTER VERBOSE { - ClusterStmt *n =3D makeNode(ClusterStmt); + RepackStmt *n =3D makeNode(RepackStmt); =20 + n->command =3D REPACK_COMMAND_CLUSTER; n->relation =3D NULL; n->indexname =3D NULL; + n->usingindex =3D true; n->params =3D list_make1(makeDefElem("verbose", NULL, @2)); $$ =3D (Node *) n; } /* kept for pre-8.3 compatibility */ | CLUSTER opt_verbose name ON qualified_name { - ClusterStmt *n =3D makeNode(ClusterStmt); + RepackStmt *n =3D makeNode(RepackStmt); =20 - n->relation =3D $5; + n->command =3D REPACK_COMMAND_CLUSTER; + n->relation =3D makeNode(VacuumRelation); + n->relation->relation =3D $5; n->indexname =3D $3; + n->usingindex =3D true; if ($2) n->params =3D list_make1(makeDefElem("verbose", NULL, @2)); $$ =3D (Node *) n; @@ -18069,6 +18123,7 @@ unreserved_keyword: | RELATIVE_P | RELEASE | RENAME + | REPACK | REPEATABLE | REPLACE | REPLICA @@ -18704,6 +18759,7 @@ bare_label_keyword: | RELATIVE_P | RELEASE | RENAME + | REPACK | REPEATABLE | REPLACE | REPLICA diff --git a/src/backend/tcop/utility.c b/src/backend/tcop/utility.c index d18a3a60a46..d01895d1cf1 100644 --- a/src/backend/tcop/utility.c +++ b/src/backend/tcop/utility.c @@ -279,9 +279,9 @@ ClassifyUtilityCommandAsReadOnly(Node *parsetree) return COMMAND_OK_IN_RECOVERY | COMMAND_OK_IN_READ_ONLY_TXN; } =20 - case T_ClusterStmt: case T_ReindexStmt: case T_VacuumStmt: + case T_RepackStmt: { /* * These commands write WAL, so they're not strictly @@ -856,14 +856,14 @@ standard_ProcessUtility(PlannedStmt *pstmt, ExecuteCallStmt(castNode(CallStmt, parsetree), params, isAtomicContext,= dest); break; =20 - case T_ClusterStmt: - cluster(pstate, (ClusterStmt *) parsetree, isTopLevel); - break; - case T_VacuumStmt: ExecVacuum(pstate, (VacuumStmt *) parsetree, isTopLevel); break; =20 + case T_RepackStmt: + ExecRepack(pstate, (RepackStmt *) parsetree, isTopLevel); + break; + case T_ExplainStmt: ExplainQuery(pstate, (ExplainStmt *) parsetree, params, dest); break; @@ -2864,10 +2864,6 @@ CreateCommandTag(Node *parsetree) tag =3D CMDTAG_CALL; break; =20 - case T_ClusterStmt: - tag =3D CMDTAG_CLUSTER; - break; - case T_VacuumStmt: if (((VacuumStmt *) parsetree)->is_vacuumcmd) tag =3D CMDTAG_VACUUM; @@ -2875,6 +2871,10 @@ CreateCommandTag(Node *parsetree) tag =3D CMDTAG_ANALYZE; break; =20 + case T_RepackStmt: + tag =3D CMDTAG_REPACK; + break; + case T_ExplainStmt: tag =3D CMDTAG_EXPLAIN; break; @@ -3516,7 +3516,7 @@ GetCommandLogLevel(Node *parsetree) lev =3D LOGSTMT_ALL; break; =20 - case T_ClusterStmt: + case T_RepackStmt: lev =3D LOGSTMT_DDL; break; =20 diff --git a/src/backend/utils/adt/pgstatfuncs.c b/src/backend/utils/adt/pg= statfuncs.c index 7e2ed69138a..1a82534e347 100644 --- a/src/backend/utils/adt/pgstatfuncs.c +++ b/src/backend/utils/adt/pgstatfuncs.c @@ -289,6 +289,8 @@ pg_stat_get_progress_info(PG_FUNCTION_ARGS) cmdtype =3D PROGRESS_COMMAND_ANALYZE; else if (pg_strcasecmp(cmd, "CLUSTER") =3D=3D 0) cmdtype =3D PROGRESS_COMMAND_CLUSTER; + else if (pg_strcasecmp(cmd, "REPACK") =3D=3D 0) + cmdtype =3D PROGRESS_COMMAND_REPACK; else if (pg_strcasecmp(cmd, "CREATE INDEX") =3D=3D 0) cmdtype =3D PROGRESS_COMMAND_CREATE_INDEX; else if (pg_strcasecmp(cmd, "BASEBACKUP") =3D=3D 0) diff --git a/src/bin/psql/tab-complete.in.c b/src/bin/psql/tab-complete.in.c index 20d7a65c614..626d9f1c98b 100644 --- a/src/bin/psql/tab-complete.in.c +++ b/src/bin/psql/tab-complete.in.c @@ -1267,7 +1267,7 @@ static const char *const sql_commands[] =3D { "DELETE FROM", "DISCARD", "DO", "DROP", "END", "EXECUTE", "EXPLAIN", "FETCH", "GRANT", "IMPORT FOREIGN SCHEMA", "INSERT INTO", "LISTEN", "LOAD= ", "LOCK", "MERGE INTO", "MOVE", "NOTIFY", "PREPARE", - "REASSIGN", "REFRESH MATERIALIZED VIEW", "REINDEX", "RELEASE", + "REASSIGN", "REFRESH MATERIALIZED VIEW", "REINDEX", "RELEASE", "REPACK", "RESET", "REVOKE", "ROLLBACK", "SAVEPOINT", "SECURITY LABEL", "SELECT", "SET", "SHOW", "START", "TABLE", "TRUNCATE", "UNLISTEN", "UPDATE", "VACUUM", "VALUES", @@ -5040,6 +5040,46 @@ match_previous_words(int pattern_id, COMPLETE_WITH_QUERY(Query_for_list_of_tablespaces); } =20 +/* REPACK */ + else if (Matches("REPACK")) + COMPLETE_WITH_SCHEMA_QUERY_PLUS(Query_for_list_of_clusterables, + "(", "USING INDEX"); + else if (Matches("REPACK", "(*)")) + COMPLETE_WITH_SCHEMA_QUERY_PLUS(Query_for_list_of_clusterables, + "USING INDEX"); + else if (Matches("REPACK", MatchAnyExcept("("))) + COMPLETE_WITH("USING INDEX"); + else if (Matches("REPACK", "(*)", MatchAnyExcept("("))) + COMPLETE_WITH("USING INDEX"); + else if (Matches("REPACK", MatchAny, "USING", "INDEX") || + Matches("REPACK", "(*)", MatchAny, "USING", "INDEX")) + { + set_completion_reference(prev3_wd); + COMPLETE_WITH_SCHEMA_QUERY(Query_for_index_of_table); + } + /* + * Complete ... [ (*) ] USING INDEX, with a list of indexes for + * . + */ + else if (TailMatches(MatchAny, "USING", "INDEX")) + { + set_completion_reference(prev3_wd); + COMPLETE_WITH_SCHEMA_QUERY(Query_for_index_of_table); + } + else if (HeadMatches("REPACK", "(*") && + !HeadMatches("REPACK", "(*)")) + { + /* + * This fires if we're in an unfinished parenthesized option list. + * get_previous_words treats a completed parenthesized option list as + * one word, so the above test is correct. + */ + if (ends_with(prev_wd, '(') || ends_with(prev_wd, ',')) + COMPLETE_WITH("ANALYZE", "VERBOSE"); + else if (TailMatches("ANALYZE", "VERBOSE")) + COMPLETE_WITH("ON", "OFF"); + } + /* SECURITY LABEL */ else if (Matches("SECURITY")) COMPLETE_WITH("LABEL"); diff --git a/src/bin/scripts/Makefile b/src/bin/scripts/Makefile index 019ca06455d..f0c1bd4175c 100644 --- a/src/bin/scripts/Makefile +++ b/src/bin/scripts/Makefile @@ -16,7 +16,7 @@ subdir =3D src/bin/scripts top_builddir =3D ../../.. include $(top_builddir)/src/Makefile.global =20 -PROGRAMS =3D createdb createuser dropdb dropuser clusterdb vacuumdb reinde= xdb pg_isready +PROGRAMS =3D createdb createuser dropdb dropuser clusterdb vacuumdb reinde= xdb pg_isready pg_repackdb =20 override CPPFLAGS :=3D -I$(libpq_srcdir) $(CPPFLAGS) LDFLAGS_INTERNAL +=3D -L$(top_builddir)/src/fe_utils -lpgfeutils $(libpq_p= gport) @@ -31,6 +31,7 @@ clusterdb: clusterdb.o common.o $(WIN32RES) | submake-lib= pq submake-libpgport su vacuumdb: vacuumdb.o vacuuming.o common.o $(WIN32RES) | submake-libpq subm= ake-libpgport submake-libpgfeutils reindexdb: reindexdb.o common.o $(WIN32RES) | submake-libpq submake-libpgp= ort submake-libpgfeutils pg_isready: pg_isready.o common.o $(WIN32RES) | submake-libpq submake-libp= gport submake-libpgfeutils +pg_repackdb: pg_repackdb.o vacuuming.o common.o $(WIN32RES) | submake-libp= q submake-libpgport submake-libpgfeutils =20 install: all installdirs $(INSTALL_PROGRAM) createdb$(X) '$(DESTDIR)$(bindir)'/createdb$(X) @@ -41,6 +42,7 @@ install: all installdirs $(INSTALL_PROGRAM) vacuumdb$(X) '$(DESTDIR)$(bindir)'/vacuumdb$(X) $(INSTALL_PROGRAM) reindexdb$(X) '$(DESTDIR)$(bindir)'/reindexdb$(X) $(INSTALL_PROGRAM) pg_isready$(X) '$(DESTDIR)$(bindir)'/pg_isready$(X) + $(INSTALL_PROGRAM) pg_repackdb$(X) '$(DESTDIR)$(bindir)'/pg_repackdb$(X) =20 installdirs: $(MKDIR_P) '$(DESTDIR)$(bindir)' diff --git a/src/bin/scripts/meson.build b/src/bin/scripts/meson.build index a4fed59d1c9..be573cae682 100644 --- a/src/bin/scripts/meson.build +++ b/src/bin/scripts/meson.build @@ -42,6 +42,7 @@ vacuuming_common =3D static_library('libvacuuming_common', =20 binaries =3D [ 'vacuumdb', + 'pg_repackdb', ] foreach binary : binaries binary_sources =3D files('@0@.c'.format(binary)) @@ -80,6 +81,7 @@ tests +=3D { 't/100_vacuumdb.pl', 't/101_vacuumdb_all.pl', 't/102_vacuumdb_stages.pl', + 't/103_repackdb.pl', 't/200_connstr.pl', ], }, diff --git a/src/bin/scripts/pg_repackdb.c b/src/bin/scripts/pg_repackdb.c new file mode 100644 index 00000000000..1edfa34ed0f --- /dev/null +++ b/src/bin/scripts/pg_repackdb.c @@ -0,0 +1,242 @@ +/*------------------------------------------------------------------------- + * + * pg_repackdb + * An utility to run REPACK + * + * Portions Copyright (c) 1996-2025, PostgreSQL Global Development Group + * Portions Copyright (c) 1994, Regents of the University of California + * + * FIXME: this is missing a way to specify the index to use to repack one + * table, or whether to pass a WITH INDEX clause when multiple tables are + * used. Something like --index[=3Dindexname]. Adding that bleeds into + * vacuuming.c as well. + * + * src/bin/scripts/pg_repackdb.c + * + *------------------------------------------------------------------------- + */ + +#include "postgres_fe.h" + +#include + +#include "common.h" +#include "common/logging.h" +#include "fe_utils/option_utils.h" +#include "vacuuming.h" + +static void help(const char *progname); +static void check_objfilter(bits32 objfilter); + +int +main(int argc, char *argv[]) +{ + static struct option long_options[] =3D { + {"host", required_argument, NULL, 'h'}, + {"port", required_argument, NULL, 'p'}, + {"username", required_argument, NULL, 'U'}, + {"no-password", no_argument, NULL, 'w'}, + {"password", no_argument, NULL, 'W'}, + {"echo", no_argument, NULL, 'e'}, + {"quiet", no_argument, NULL, 'q'}, + {"dbname", required_argument, NULL, 'd'}, + {"analyze", no_argument, NULL, 'z'}, + {"all", no_argument, NULL, 'a'}, + /* XXX this could be 'i', but optional_arg is messy */ + {"index", optional_argument, NULL, 1}, + {"table", required_argument, NULL, 't'}, + {"verbose", no_argument, NULL, 'v'}, + {"jobs", required_argument, NULL, 'j'}, + {"schema", required_argument, NULL, 'n'}, + {"exclude-schema", required_argument, NULL, 'N'}, + {"maintenance-db", required_argument, NULL, 2}, + {NULL, 0, NULL, 0} + }; + + const char *progname; + int optindex; + int c; + const char *dbname =3D NULL; + const char *maintenance_db =3D NULL; + ConnParams cparams; + bool echo =3D false; + bool quiet =3D false; + vacuumingOptions vacopts; + SimpleStringList objects =3D {NULL, NULL}; + int concurrentCons =3D 1; + int tbl_count =3D 0; + int ret; + + /* initialize options */ + memset(&vacopts, 0, sizeof(vacopts)); + vacopts.mode =3D MODE_REPACK; + + /* the same for connection parameters */ + memset(&cparams, 0, sizeof(cparams)); + cparams.prompt_password =3D TRI_DEFAULT; + + pg_logging_init(argv[0]); + progname =3D get_progname(argv[0]); + set_pglocale_pgservice(argv[0], PG_TEXTDOMAIN("pgscripts")); + + handle_help_version_opts(argc, argv, progname, help); + + while ((c =3D getopt_long(argc, argv, "ad:eh:j:n:N:p:qt:U:vwWz", + long_options, &optindex)) !=3D -1) + { + switch (c) + { + case 'a': + vacopts.objfilter |=3D OBJFILTER_ALL_DBS; + break; + case 'd': + vacopts.objfilter |=3D OBJFILTER_DATABASE; + dbname =3D pg_strdup(optarg); + break; + case 'e': + echo =3D true; + break; + case 'h': + cparams.pghost =3D pg_strdup(optarg); + break; + case 'j': + if (!option_parse_int(optarg, "-j/--jobs", 1, INT_MAX, + &concurrentCons)) + exit(1); + break; + case 'n': + vacopts.objfilter |=3D OBJFILTER_SCHEMA; + simple_string_list_append(&objects, optarg); + break; + case 'N': + vacopts.objfilter |=3D OBJFILTER_SCHEMA_EXCLUDE; + simple_string_list_append(&objects, optarg); + break; + case 'p': + cparams.pgport =3D pg_strdup(optarg); + break; + case 'q': + quiet =3D true; + break; + case 't': + vacopts.objfilter |=3D OBJFILTER_TABLE; + simple_string_list_append(&objects, optarg); + tbl_count++; + break; + case 'U': + cparams.pguser =3D pg_strdup(optarg); + break; + case 'v': + vacopts.verbose =3D true; + break; + case 'w': + cparams.prompt_password =3D TRI_NO; + break; + case 'W': + cparams.prompt_password =3D TRI_YES; + break; + case 'z': + vacopts.and_analyze =3D true; + break; + case 1: + vacopts.using_index =3D true; + if (optarg) + vacopts.indexname =3D pg_strdup(optarg); + else + vacopts.indexname =3D NULL; + break; + case 2: + maintenance_db =3D pg_strdup(optarg); + break; + default: + /* getopt_long already emitted a complaint */ + pg_log_error_hint("Try \"%s --help\" for more information.", progname); + exit(1); + } + } + + /* + * Non-option argument specifies database name as long as it wasn't + * already specified with -d / --dbname + */ + if (optind < argc && dbname =3D=3D NULL) + { + vacopts.objfilter |=3D OBJFILTER_DATABASE; + dbname =3D argv[optind]; + optind++; + } + + if (optind < argc) + { + pg_log_error("too many command-line arguments (first is \"%s\")", + argv[optind]); + pg_log_error_hint("Try \"%s --help\" for more information.", progname); + exit(1); + } + + /* + * Validate the combination of filters specified in the command-line + * options. + */ + check_objfilter(vacopts.objfilter); + + ret =3D vacuuming_main(&cparams, dbname, maintenance_db, &vacopts, + &objects, tbl_count, concurrentCons, + progname, echo, quiet); + exit(ret); +} + +/* + * Verify that the filters used at command line are compatible. + */ +void +check_objfilter(bits32 objfilter) +{ + if ((objfilter & OBJFILTER_ALL_DBS) && + (objfilter & OBJFILTER_DATABASE)) + pg_fatal("cannot repack all databases and a specific one at the same tim= e"); + + if ((objfilter & OBJFILTER_TABLE) && + (objfilter & OBJFILTER_SCHEMA)) + pg_fatal("cannot repack all tables in schema(s) and specific table(s) at= the same time"); + + if ((objfilter & OBJFILTER_TABLE) && + (objfilter & OBJFILTER_SCHEMA_EXCLUDE)) + pg_fatal("cannot repack specific table(s) and exclude schema(s) at the s= ame time"); + + if ((objfilter & OBJFILTER_SCHEMA) && + (objfilter & OBJFILTER_SCHEMA_EXCLUDE)) + pg_fatal("cannot repack all tables in schema(s) and exclude schema(s) at= the same time"); +} + +static void +help(const char *progname) +{ + printf(_("%s repacks a PostgreSQL database.\n\n"), progname); + printf(_("Usage:\n")); + printf(_(" %s [OPTION]... [DBNAME]\n"), progname); + printf(_("\nOptions:\n")); + printf(_(" -a, --all repack all databases\n")); + printf(_(" -d, --dbname=3DDBNAME database to repack\n")); + printf(_(" -e, --echo show the commands being sent = to the server\n")); + printf(_(" --index[=3DINDEX] repack following an index\n= ")); + printf(_(" -j, --jobs=3DNUM use this many concurrent co= nnections to repack\n")); + printf(_(" -n, --schema=3DSCHEMA repack tables in the specif= ied schema(s) only\n")); + printf(_(" -N, --exclude-schema=3DSCHEMA do not repack tables in the= specified schema(s)\n")); + printf(_(" -q, --quiet don't write any messages\n")); + printf(_(" -t, --table=3D'TABLE[(COLUMNS)]' repack specific table(s) on= ly\n")); + printf(_(" -v, --verbose write a lot of output\n")); + printf(_(" -V, --version output version information, t= hen exit\n")); + printf(_(" -z, --analyze update optimizer statistics\n= ")); + printf(_(" -?, --help show this help, then exit\n")= ); + printf(_("\nConnection options:\n")); + printf(_(" -h, --host=3DHOSTNAME database server host or socket di= rectory\n")); + printf(_(" -p, --port=3DPORT database server port\n")); + printf(_(" -U, --username=3DUSERNAME user name to connect as\n")); + printf(_(" -w, --no-password never prompt for password\n")); + printf(_(" -W, --password force password prompt\n")); + printf(_(" --maintenance-db=3DDBNAME alternate maintenance database\n"= )); + printf(_("\nRead the description of the SQL command REPACK for details.\n= ")); + printf(_("\nReport bugs to <%s>.\n"), PACKAGE_BUGREPORT); + printf(_("%s home page: <%s>\n"), PACKAGE_NAME, PACKAGE_URL); +} diff --git a/src/bin/scripts/t/103_repackdb.pl b/src/bin/scripts/t/103_repa= ckdb.pl new file mode 100644 index 00000000000..cadce9b837c --- /dev/null +++ b/src/bin/scripts/t/103_repackdb.pl @@ -0,0 +1,47 @@ +# Copyright (c) 2021-2025, PostgreSQL Global Development Group + +use strict; +use warnings FATAL =3D> 'all'; + +use PostgreSQL::Test::Cluster; +use PostgreSQL::Test::Utils; +use Test::More; + +program_help_ok('pg_repackdb'); +program_version_ok('pg_repackdb'); +program_options_handling_ok('pg_repackdb'); + +my $node =3D PostgreSQL::Test::Cluster->new('main'); +$node->init; +$node->start; + +$node->issues_sql_like( + [ 'pg_repackdb', 'postgres', '-t', 'pg_class'], + qr/statement: REPACK.*pg_class;/, + 'pg_repackdb processes a single table'); + +$node->safe_psql('postgres', 'CREATE USER testusr; + GRANT CREATE ON SCHEMA public TO testusr'); +$node->safe_psql('postgres', + 'CREATE TABLE cluster_1 (a int primary key); + ALTER TABLE cluster_1 CLUSTER ON cluster_1_pkey; + CREATE TABLE cluster_2 (a int unique); + ALTER TABLE cluster_2 CLUSTER ON cluster_2_a_key;', + extra_params =3D> ['-U' =3D> 'testusr']); + +$node->issues_sql_like( + [ 'pg_repackdb', 'postgres', '-U', 'testusr' ], + qr/statement: REPACK.*;/, + 'SQL REPACK run'); + +$node->issues_sql_like( + [ 'pg_repackdb', 'postgres', '--index'], + qr/statement: REPACK.*cluster_1 USING INDEX.*statement: REPACK.*cluster_2= USING INDEX/ms, + 'pg_repackdb --index chooses multiple tables'); + +$node->issues_sql_like( + [ 'pg_repackdb', 'postgres', '--analyze', '-t', 'cluster_1'], + qr/statement: REPACK \(ANALYZE\) public.cluster_1/, + 'pg_repackdb --analyze works'); + +done_testing(); diff --git a/src/bin/scripts/vacuuming.c b/src/bin/scripts/vacuuming.c index f836f21fb03..d8d77cabe43 100644 --- a/src/bin/scripts/vacuuming.c +++ b/src/bin/scripts/vacuuming.c @@ -1,6 +1,6 @@ /*------------------------------------------------------------------------- * vacuuming.c - * Helper routines for vacuumdb + * Helper routines for vacuumdb and pg_repackdb * * Portions Copyright (c) 1996-2025, PostgreSQL Global Development Group * Portions Copyright (c) 1994, Regents of the University of California @@ -43,8 +43,8 @@ static SimpleStringList *retrieve_objects(PGconn *conn, static void free_retrieved_objects(SimpleStringList *list); static void prepare_vacuum_command(PGconn *conn, PQExpBuffer sql, vacuumingOptions *vacopts, const char *table); -static void run_vacuum_command(PGconn *conn, const char *sql, bool echo, - const char *table); +static void run_vacuum_command(PGconn *conn, vacuumingOptions *vacopts, + const char *sql, bool echo, const char *table); =20 /* * Executes vacuum/analyze as indicated. Returns 0 if the plan is carried @@ -194,6 +194,14 @@ vacuum_one_database(ConnParams *cparams, =20 conn =3D connectDatabase(cparams, progname, echo, false, true); =20 + if (vacopts->mode =3D=3D MODE_REPACK && PQserverVersion(conn) < 190000) + { + /* XXX arguably, here we should use VACUUM FULL instead of failing */ + PQfinish(conn); + pg_fatal("cannot use the \"%s\" command on server versions older than Po= stgreSQL %s", + "REPACK", "19"); + } + if (vacopts->disable_page_skipping && PQserverVersion(conn) < 90600) { PQfinish(conn); @@ -286,9 +294,18 @@ vacuum_one_database(ConnParams *cparams, if (vacopts->mode =3D=3D MODE_ANALYZE_IN_STAGES) printf(_("%s: processing database \"%s\": %s\n"), progname, PQdb(conn), _(stage_messages[stage])); - else + else if (vacopts->mode =3D=3D MODE_ANALYZE) + printf(_("%s: analyzing database \"%s\"\n"), + progname, PQdb(conn)); + else if (vacopts->mode =3D=3D MODE_VACUUM) printf(_("%s: vacuuming database \"%s\"\n"), progname, PQdb(conn)); + else + { + Assert(vacopts->mode =3D=3D MODE_REPACK); + printf(_("%s: repacking database \"%s\"\n"), + progname, PQdb(conn)); + } fflush(stdout); } =20 @@ -383,7 +400,7 @@ vacuum_one_database(ConnParams *cparams, * through ParallelSlotsGetIdle. */ ParallelSlotSetHandler(free_slot, TableCommandResultHandler, NULL); - run_vacuum_command(free_slot->connection, sql.data, + run_vacuum_command(free_slot->connection, vacopts, sql.data, echo, tabname); =20 cell =3D cell->next; @@ -408,7 +425,7 @@ vacuum_one_database(ConnParams *cparams, } =20 ParallelSlotSetHandler(free_slot, TableCommandResultHandler, NULL); - run_vacuum_command(free_slot->connection, cmd, echo, NULL); + run_vacuum_command(free_slot->connection, vacopts, cmd, echo, NULL); =20 if (!ParallelSlotsWaitCompletion(sa)) ret =3D EXIT_FAILURE; /* error already reported by handler */ @@ -636,6 +653,35 @@ retrieve_objects(PGconn *conn, vacuumingOptions *vacop= ts, " AND listed_objects.object_oid IS NOT NULL\n"); } =20 + /* + * In REPACK mode, if the 'using_index' option was given but no index + * name, filter only tables that have an index with indisclustered set. + * (If an index name is given, we trust the user to pass a reasonable list + * of tables.) + * + * XXX it may be worth printing an error if an index name is given with no + * list of tables. + */ + if (vacopts->mode =3D=3D MODE_REPACK && + vacopts->using_index && !vacopts->indexname) + { + appendPQExpBufferStr(&catalog_query, + " AND EXISTS (SELECT 1 FROM pg_catalog.pg_index\n" + " WHERE indrelid =3D c.oid AND indisclustered)\n"); + } + + /* + * In REPACK mode, only consider the tables that the current user has + * MAINTAIN privileges on. XXX maybe we should do this in all cases, not + * just REPACK. The vacuumdb output is too noisy for no reason. + */ + if (vacopts->mode =3D=3D MODE_REPACK) + { + appendPQExpBufferStr(&catalog_query, + " AND pg_catalog.has_table_privilege(current_user, " + "c.oid, 'MAINTAIN')\n"); + } + /* * If no tables were listed, filter for the relevant relation types. If * tables were given via --table, don't bother filtering by relation type. @@ -874,8 +920,10 @@ prepare_vacuum_command(PGconn *conn, PQExpBuffer sql, if (vacopts->verbose) appendPQExpBufferStr(sql, " VERBOSE"); } + + appendPQExpBuffer(sql, " %s", table); } - else + else if (vacopts->mode =3D=3D MODE_VACUUM) { appendPQExpBufferStr(sql, "VACUUM"); =20 @@ -989,9 +1037,39 @@ prepare_vacuum_command(PGconn *conn, PQExpBuffer sql, if (vacopts->and_analyze) appendPQExpBufferStr(sql, " ANALYZE"); } + + appendPQExpBuffer(sql, " %s", table); } + else if (vacopts->mode =3D=3D MODE_REPACK) + { + appendPQExpBufferStr(sql, "REPACK"); =20 - appendPQExpBuffer(sql, " %s;", table); + if (vacopts->verbose) + { + appendPQExpBuffer(sql, "%sVERBOSE", sep); + sep =3D comma; + } + if (vacopts->and_analyze) + { + appendPQExpBuffer(sql, "%sANALYZE", sep); + sep =3D comma; + } + + if (sep !=3D paren) + appendPQExpBufferChar(sql, ')'); + + appendPQExpBuffer(sql, " %s", table); + + if (vacopts->using_index) + { + appendPQExpBuffer(sql, " USING INDEX"); + if (vacopts->indexname) + appendPQExpBuffer(sql, " %s", fmtIdEnc(vacopts->indexname, + PQclientEncoding(conn))); + } + } + + appendPQExpBufferChar(sql, ';'); } =20 /* @@ -1001,8 +1079,8 @@ prepare_vacuum_command(PGconn *conn, PQExpBuffer sql, * Any errors during command execution are reported to stderr. */ static void -run_vacuum_command(PGconn *conn, const char *sql, bool echo, - const char *table) +run_vacuum_command(PGconn *conn, vacuumingOptions *vacopts, + const char *sql, bool echo, const char *table) { bool status; =20 @@ -1015,13 +1093,21 @@ run_vacuum_command(PGconn *conn, const char *sql, b= ool echo, { if (table) { - pg_log_error("vacuuming of table \"%s\" in database \"%s\" failed: %s", - table, PQdb(conn), PQerrorMessage(conn)); + if (vacopts->mode =3D=3D MODE_VACUUM) + pg_log_error("vacuuming of table \"%s\" in database \"%s\" failed: %s", + table, PQdb(conn), PQerrorMessage(conn)); + else + pg_log_error("repacking of table \"%s\" in database \"%s\" failed: %s", + table, PQdb(conn), PQerrorMessage(conn)); } else { - pg_log_error("vacuuming of database \"%s\" failed: %s", - PQdb(conn), PQerrorMessage(conn)); + if (vacopts->mode =3D=3D MODE_VACUUM) + pg_log_error("vacuuming of database \"%s\" failed: %s", + PQdb(conn), PQerrorMessage(conn)); + else + pg_log_error("repacking of database \"%s\" failed: %s", + PQdb(conn), PQerrorMessage(conn)); } } } diff --git a/src/bin/scripts/vacuuming.h b/src/bin/scripts/vacuuming.h index 49f968b32e5..665dbaedfad 100644 --- a/src/bin/scripts/vacuuming.h +++ b/src/bin/scripts/vacuuming.h @@ -20,6 +20,7 @@ typedef enum { MODE_VACUUM, + MODE_REPACK, MODE_ANALYZE, MODE_ANALYZE_IN_STAGES } RunMode; @@ -37,6 +38,8 @@ typedef struct vacuumingOptions bool and_analyze; bool full; bool freeze; + bool using_index; + char *indexname; bool disable_page_skipping; bool skip_locked; int min_xid_age; diff --git a/src/include/commands/cluster.h b/src/include/commands/cluster.h index 60088a64cbb..652542e8e65 100644 --- a/src/include/commands/cluster.h +++ b/src/include/commands/cluster.h @@ -24,6 +24,7 @@ #define CLUOPT_RECHECK 0x02 /* recheck relation state */ #define CLUOPT_RECHECK_ISCLUSTERED 0x04 /* recheck relation state for * indisclustered */ +#define CLUOPT_ANALYZE 0x08 /* do an ANALYZE */ =20 /* options for CLUSTER */ typedef struct ClusterParams @@ -31,8 +32,11 @@ typedef struct ClusterParams bits32 options; /* bitmask of CLUOPT_* */ } ClusterParams; =20 -extern void cluster(ParseState *pstate, ClusterStmt *stmt, bool isTopLevel= ); -extern void cluster_rel(Relation OldHeap, Oid indexOid, ClusterParams *par= ams); + +extern void ExecRepack(ParseState *pstate, RepackStmt *stmt, bool isTopLev= el); + +extern void cluster_rel(RepackCommand command, Relation OldHeap, Oid index= Oid, + ClusterParams *params); extern void check_index_is_clusterable(Relation OldHeap, Oid indexOid, LOCKMODE lockmode); extern void mark_index_clustered(Relation rel, Oid indexOid, bool is_inter= nal); diff --git a/src/include/commands/progress.h b/src/include/commands/progres= s.h index 1cde4bd9bcf..ebf004b7aa5 100644 --- a/src/include/commands/progress.h +++ b/src/include/commands/progress.h @@ -56,28 +56,34 @@ #define PROGRESS_ANALYZE_PHASE_COMPUTE_EXT_STATS 4 #define PROGRESS_ANALYZE_PHASE_FINALIZE_ANALYZE 5 =20 -/* Progress parameters for cluster */ -#define PROGRESS_CLUSTER_COMMAND 0 -#define PROGRESS_CLUSTER_PHASE 1 -#define PROGRESS_CLUSTER_INDEX_RELID 2 -#define PROGRESS_CLUSTER_HEAP_TUPLES_SCANNED 3 -#define PROGRESS_CLUSTER_HEAP_TUPLES_WRITTEN 4 -#define PROGRESS_CLUSTER_TOTAL_HEAP_BLKS 5 -#define PROGRESS_CLUSTER_HEAP_BLKS_SCANNED 6 -#define PROGRESS_CLUSTER_INDEX_REBUILD_COUNT 7 - -/* Phases of cluster (as advertised via PROGRESS_CLUSTER_PHASE) */ -#define PROGRESS_CLUSTER_PHASE_SEQ_SCAN_HEAP 1 -#define PROGRESS_CLUSTER_PHASE_INDEX_SCAN_HEAP 2 -#define PROGRESS_CLUSTER_PHASE_SORT_TUPLES 3 -#define PROGRESS_CLUSTER_PHASE_WRITE_NEW_HEAP 4 -#define PROGRESS_CLUSTER_PHASE_SWAP_REL_FILES 5 -#define PROGRESS_CLUSTER_PHASE_REBUILD_INDEX 6 -#define PROGRESS_CLUSTER_PHASE_FINAL_CLEANUP 7 - -/* Commands of PROGRESS_CLUSTER */ -#define PROGRESS_CLUSTER_COMMAND_CLUSTER 1 -#define PROGRESS_CLUSTER_COMMAND_VACUUM_FULL 2 +/* + * Progress parameters for REPACK. + * + * Values for PROGRESS_REPACK_COMMAND are defined as in RepackCommand. + * + * Note: Since REPACK shares code with CLUSTER, these values are also + * used by CLUSTER. (CLUSTER is now deprecated, so it makes little sense to + * introduce a separate set of constants.) + */ +#define PROGRESS_REPACK_COMMAND 0 +#define PROGRESS_REPACK_PHASE 1 +#define PROGRESS_REPACK_INDEX_RELID 2 +#define PROGRESS_REPACK_HEAP_TUPLES_SCANNED 3 +#define PROGRESS_REPACK_HEAP_TUPLES_WRITTEN 4 +#define PROGRESS_REPACK_TOTAL_HEAP_BLKS 5 +#define PROGRESS_REPACK_HEAP_BLKS_SCANNED 6 +#define PROGRESS_REPACK_INDEX_REBUILD_COUNT 7 + +/* + * Phases of repack (as advertised via PROGRESS_REPACK_PHASE). + */ +#define PROGRESS_REPACK_PHASE_SEQ_SCAN_HEAP 1 +#define PROGRESS_REPACK_PHASE_INDEX_SCAN_HEAP 2 +#define PROGRESS_REPACK_PHASE_SORT_TUPLES 3 +#define PROGRESS_REPACK_PHASE_WRITE_NEW_HEAP 4 +#define PROGRESS_REPACK_PHASE_SWAP_REL_FILES 5 +#define PROGRESS_REPACK_PHASE_REBUILD_INDEX 6 +#define PROGRESS_REPACK_PHASE_FINAL_CLEANUP 7 =20 /* Progress parameters for CREATE INDEX */ /* 3, 4 and 5 reserved for "waitfor" metrics */ diff --git a/src/include/nodes/parsenodes.h b/src/include/nodes/parsenodes.h index d14294a4ece..94892042b8d 100644 --- a/src/include/nodes/parsenodes.h +++ b/src/include/nodes/parsenodes.h @@ -3951,18 +3951,6 @@ typedef struct AlterSystemStmt VariableSetStmt *setstmt; /* SET subcommand */ } AlterSystemStmt; =20 -/* ---------------------- - * Cluster Statement (support pbrown's cluster index implementation) - * ---------------------- - */ -typedef struct ClusterStmt -{ - NodeTag type; - RangeVar *relation; /* relation being indexed, or NULL if all */ - char *indexname; /* original index defined */ - List *params; /* list of DefElem nodes */ -} ClusterStmt; - /* ---------------------- * Vacuum and Analyze Statements * @@ -3975,7 +3963,7 @@ typedef struct VacuumStmt NodeTag type; List *options; /* list of DefElem nodes */ List *rels; /* list of VacuumRelation, or NIL for all */ - bool is_vacuumcmd; /* true for VACUUM, false for ANALYZE */ + bool is_vacuumcmd; /* true for VACUUM, false otherwise */ } VacuumStmt; =20 /* @@ -3993,6 +3981,27 @@ typedef struct VacuumRelation List *va_cols; /* list of column names, or NIL for all */ } VacuumRelation; =20 +/* ---------------------- + * Repack Statement + * ---------------------- + */ +typedef enum RepackCommand +{ + REPACK_COMMAND_CLUSTER =3D 1, + REPACK_COMMAND_REPACK, + REPACK_COMMAND_VACUUMFULL, +} RepackCommand; + +typedef struct RepackStmt +{ + NodeTag type; + RepackCommand command; /* type of command being run */ + VacuumRelation *relation; /* relation being repacked */ + char *indexname; /* order tuples by this index */ + bool usingindex; /* whether USING INDEX is specified */ + List *params; /* list of DefElem nodes */ +} RepackStmt; + /* ---------------------- * Explain Statement * diff --git a/src/include/parser/kwlist.h b/src/include/parser/kwlist.h index 5d4fe27ef96..f1a1d5e7a80 100644 --- a/src/include/parser/kwlist.h +++ b/src/include/parser/kwlist.h @@ -376,6 +376,7 @@ PG_KEYWORD("reindex", REINDEX, UNRESERVED_KEYWORD, BARE= _LABEL) PG_KEYWORD("relative", RELATIVE_P, UNRESERVED_KEYWORD, BARE_LABEL) PG_KEYWORD("release", RELEASE, UNRESERVED_KEYWORD, BARE_LABEL) PG_KEYWORD("rename", RENAME, UNRESERVED_KEYWORD, BARE_LABEL) +PG_KEYWORD("repack", REPACK, UNRESERVED_KEYWORD, BARE_LABEL) PG_KEYWORD("repeatable", REPEATABLE, UNRESERVED_KEYWORD, BARE_LABEL) PG_KEYWORD("replace", REPLACE, UNRESERVED_KEYWORD, BARE_LABEL) PG_KEYWORD("replica", REPLICA, UNRESERVED_KEYWORD, BARE_LABEL) diff --git a/src/include/tcop/cmdtaglist.h b/src/include/tcop/cmdtaglist.h index c4606d65043..66690f1134a 100644 --- a/src/include/tcop/cmdtaglist.h +++ b/src/include/tcop/cmdtaglist.h @@ -196,6 +196,7 @@ PG_CMDTAG(CMDTAG_REASSIGN_OWNED, "REASSIGN OWNED", fals= e, false, false) PG_CMDTAG(CMDTAG_REFRESH_MATERIALIZED_VIEW, "REFRESH MATERIALIZED VIEW", t= rue, false, false) PG_CMDTAG(CMDTAG_REINDEX, "REINDEX", true, false, false) PG_CMDTAG(CMDTAG_RELEASE, "RELEASE", false, false, false) +PG_CMDTAG(CMDTAG_REPACK, "REPACK", false, false, false) PG_CMDTAG(CMDTAG_RESET, "RESET", false, false, false) PG_CMDTAG(CMDTAG_REVOKE, "REVOKE", true, false, false) PG_CMDTAG(CMDTAG_REVOKE_ROLE, "REVOKE ROLE", false, false, false) diff --git a/src/include/utils/backend_progress.h b/src/include/utils/backe= nd_progress.h index dda813ab407..e69e366dcdc 100644 --- a/src/include/utils/backend_progress.h +++ b/src/include/utils/backend_progress.h @@ -28,6 +28,7 @@ typedef enum ProgressCommandType PROGRESS_COMMAND_CREATE_INDEX, PROGRESS_COMMAND_BASEBACKUP, PROGRESS_COMMAND_COPY, + PROGRESS_COMMAND_REPACK, } ProgressCommandType; =20 #define PGSTAT_NUM_PROGRESS_PARAM 20 diff --git a/src/test/regress/expected/cluster.out b/src/test/regress/expec= ted/cluster.out index 4d40a6809ab..277854418fa 100644 --- a/src/test/regress/expected/cluster.out +++ b/src/test/regress/expected/cluster.out @@ -495,6 +495,43 @@ ALTER TABLE clstrpart SET WITHOUT CLUSTER; ERROR: cannot mark index clustered in partitioned table ALTER TABLE clstrpart CLUSTER ON clstrpart_idx; ERROR: cannot mark index clustered in partitioned table +-- Check that REPACK sets new relfilenodes: it should process exactly the = same +-- tables as CLUSTER did. +DROP TABLE old_cluster_info; +DROP TABLE new_cluster_info; +CREATE TEMP TABLE old_cluster_info AS SELECT relname, level, relfilenode, = relkind FROM pg_partition_tree('clstrpart'::regclass) AS tree JOIN pg_class= c ON c.oid=3Dtree.relid ; +REPACK clstrpart USING INDEX clstrpart_idx; +CREATE TEMP TABLE new_cluster_info AS SELECT relname, level, relfilenode, = relkind FROM pg_partition_tree('clstrpart'::regclass) AS tree JOIN pg_class= c ON c.oid=3Dtree.relid ; +SELECT relname, old.level, old.relkind, old.relfilenode =3D new.relfilenod= e FROM old_cluster_info AS old JOIN new_cluster_info AS new USING (relname)= ORDER BY relname COLLATE "C"; + relname | level | relkind | ?column?=20 +-------------+-------+---------+---------- + clstrpart | 0 | p | t + clstrpart1 | 1 | p | t + clstrpart11 | 2 | r | f + clstrpart12 | 2 | p | t + clstrpart2 | 1 | r | f + clstrpart3 | 1 | p | t + clstrpart33 | 2 | r | f +(7 rows) + +-- And finally the same for REPACK w/o index. +DROP TABLE old_cluster_info; +DROP TABLE new_cluster_info; +CREATE TEMP TABLE old_cluster_info AS SELECT relname, level, relfilenode, = relkind FROM pg_partition_tree('clstrpart'::regclass) AS tree JOIN pg_class= c ON c.oid=3Dtree.relid ; +REPACK clstrpart; +CREATE TEMP TABLE new_cluster_info AS SELECT relname, level, relfilenode, = relkind FROM pg_partition_tree('clstrpart'::regclass) AS tree JOIN pg_class= c ON c.oid=3Dtree.relid ; +SELECT relname, old.level, old.relkind, old.relfilenode =3D new.relfilenod= e FROM old_cluster_info AS old JOIN new_cluster_info AS new USING (relname)= ORDER BY relname COLLATE "C"; + relname | level | relkind | ?column?=20 +-------------+-------+---------+---------- + clstrpart | 0 | p | t + clstrpart1 | 1 | p | t + clstrpart11 | 2 | r | f + clstrpart12 | 2 | p | t + clstrpart2 | 1 | r | f + clstrpart3 | 1 | p | t + clstrpart33 | 2 | r | f +(7 rows) + DROP TABLE clstrpart; -- Ownership of partitions is checked CREATE TABLE ptnowner(i int unique) PARTITION BY LIST (i); @@ -513,7 +550,7 @@ CREATE TEMP TABLE ptnowner_oldnodes AS JOIN pg_class AS c ON c.oid=3Dtree.relid; SET SESSION AUTHORIZATION regress_ptnowner; CLUSTER ptnowner USING ptnowner_i_idx; -WARNING: permission denied to cluster "ptnowner2", skipping it +WARNING: permission denied to execute CLUSTER on "ptnowner2", skipping it RESET SESSION AUTHORIZATION; SELECT a.relname, a.relfilenode=3Db.relfilenode FROM pg_class a JOIN ptnowner_oldnodes b USING (oid) ORDER BY a.relname COLLATE "C"; @@ -665,6 +702,101 @@ SELECT * FROM clstr_expression WHERE -a =3D -3 ORDER = BY -a, b; (4 rows) =20 COMMIT; +---------------------------------------------------------------------- +-- +-- REPACK +-- +---------------------------------------------------------------------- +-- REPACK handles individual tables identically to CLUSTER, but it's worth +-- checking if it handles table hierarchies identically as well. +REPACK clstr_tst USING INDEX clstr_tst_c; +-- Verify that inheritance link still works +INSERT INTO clstr_tst_inh VALUES (0, 100, 'in child table 2'); +SELECT a,b,c,substring(d for 30), length(d) from clstr_tst; + a | b | c | substring | length=20 +----+-----+------------------+--------------------------------+-------- + 10 | 14 | catorce | |=20=20=20= =20=20=20=20 + 18 | 5 | cinco | |=20=20=20= =20=20=20=20 + 9 | 4 | cuatro | |=20=20=20= =20=20=20=20 + 26 | 19 | diecinueve | |=20=20=20= =20=20=20=20 + 12 | 18 | dieciocho | |=20=20=20= =20=20=20=20 + 30 | 16 | dieciseis | |=20=20=20= =20=20=20=20 + 24 | 17 | diecisiete | |=20=20=20= =20=20=20=20 + 2 | 10 | diez | |=20=20=20= =20=20=20=20 + 23 | 12 | doce | |=20=20=20= =20=20=20=20 + 11 | 2 | dos | |=20=20=20= =20=20=20=20 + 25 | 9 | nueve | |=20=20=20= =20=20=20=20 + 31 | 8 | ocho | |=20=20=20= =20=20=20=20 + 1 | 11 | once | |=20=20=20= =20=20=20=20 + 28 | 15 | quince | |=20=20=20= =20=20=20=20 + 32 | 6 | seis | xyzzyxyzzyxyzzyxyzzyxyzzyxyzzy | 500000 + 29 | 7 | siete | |=20=20=20= =20=20=20=20 + 15 | 13 | trece | |=20=20=20= =20=20=20=20 + 22 | 30 | treinta | |=20=20=20= =20=20=20=20 + 17 | 32 | treinta y dos | |=20=20=20= =20=20=20=20 + 3 | 31 | treinta y uno | |=20=20=20= =20=20=20=20 + 5 | 3 | tres | |=20=20=20= =20=20=20=20 + 20 | 1 | uno | |=20=20=20= =20=20=20=20 + 6 | 20 | veinte | |=20=20=20= =20=20=20=20 + 14 | 25 | veinticinco | |=20=20=20= =20=20=20=20 + 21 | 24 | veinticuatro | |=20=20=20= =20=20=20=20 + 4 | 22 | veintidos | |=20=20=20= =20=20=20=20 + 19 | 29 | veintinueve | |=20=20=20= =20=20=20=20 + 16 | 28 | veintiocho | |=20=20=20= =20=20=20=20 + 27 | 26 | veintiseis | |=20=20=20= =20=20=20=20 + 13 | 27 | veintisiete | |=20=20=20= =20=20=20=20 + 7 | 23 | veintitres | |=20=20=20= =20=20=20=20 + 8 | 21 | veintiuno | |=20=20=20= =20=20=20=20 + 0 | 100 | in child table | |=20=20=20= =20=20=20=20 + 0 | 100 | in child table 2 | |=20=20=20= =20=20=20=20 +(34 rows) + +-- Verify that foreign key link still works +INSERT INTO clstr_tst (b, c) VALUES (1111, 'this should fail'); +ERROR: insert or update on table "clstr_tst" violates foreign key constra= int "clstr_tst_con" +DETAIL: Key (b)=3D(1111) is not present in table "clstr_tst_s". +SELECT conname FROM pg_constraint WHERE conrelid =3D 'clstr_tst'::regclass +ORDER BY 1; + conname=20=20=20=20=20=20=20=20 +---------------------- + clstr_tst_a_not_null + clstr_tst_con + clstr_tst_pkey +(3 rows) + +-- Verify partial analyze works +REPACK (ANALYZE) clstr_tst (a); +REPACK (ANALYZE) clstr_tst; +REPACK (VERBOSE) clstr_tst (a); +ERROR: ANALYZE option must be specified when a column list is provided +-- REPACK w/o argument performs no ordering, so we can only check which ta= bles +-- have the relfilenode changed. +RESET SESSION AUTHORIZATION; +CREATE TEMP TABLE relnodes_old AS +(SELECT relname, relfilenode +FROM pg_class +WHERE relname IN ('clstr_1', 'clstr_2', 'clstr_3')); +SET SESSION AUTHORIZATION regress_clstr_user; +SET client_min_messages =3D ERROR; -- order of "skipping" warnings may va= ry +REPACK; +RESET client_min_messages; +RESET SESSION AUTHORIZATION; +CREATE TEMP TABLE relnodes_new AS +(SELECT relname, relfilenode +FROM pg_class +WHERE relname IN ('clstr_1', 'clstr_2', 'clstr_3')); +-- Do the actual comparison. Unlike CLUSTER, clstr_3 should have been +-- processed because there is nothing like clustering index here. +SELECT o.relname FROM relnodes_old o +JOIN relnodes_new n ON o.relname =3D n.relname +WHERE o.relfilenode <> n.relfilenode +ORDER BY o.relname; + relname=20 +--------- + clstr_1 + clstr_3 +(2 rows) + -- clean up DROP TABLE clustertest; DROP TABLE clstr_1; diff --git a/src/test/regress/expected/rules.out b/src/test/regress/expecte= d/rules.out index 94e45dd4d57..1c957f12d27 100644 --- a/src/test/regress/expected/rules.out +++ b/src/test/regress/expected/rules.out @@ -1995,34 +1995,23 @@ pg_stat_progress_basebackup| SELECT pid, ELSE NULL::text END AS backup_type FROM pg_stat_get_progress_info('BASEBACKUP'::text) s(pid, datid, relid,= param1, param2, param3, param4, param5, param6, param7, param8, param9, pa= ram10, param11, param12, param13, param14, param15, param16, param17, param= 18, param19, param20); -pg_stat_progress_cluster| SELECT s.pid, - s.datid, - d.datname, - s.relid, - CASE s.param1 - WHEN 1 THEN 'CLUSTER'::text - WHEN 2 THEN 'VACUUM FULL'::text - ELSE NULL::text +pg_stat_progress_cluster| SELECT pid, + datid, + datname, + relid, + CASE + WHEN (command =3D ANY (ARRAY['CLUSTER'::text, 'VACUUM FULL'::t= ext])) THEN command + WHEN (repack_index_relid =3D (0)::oid) THEN 'VACUUM FULL'::text + ELSE 'CLUSTER'::text END AS command, - CASE s.param2 - WHEN 0 THEN 'initializing'::text - WHEN 1 THEN 'seq scanning heap'::text - WHEN 2 THEN 'index scanning heap'::text - WHEN 3 THEN 'sorting tuples'::text - WHEN 4 THEN 'writing new heap'::text - WHEN 5 THEN 'swapping relation files'::text - WHEN 6 THEN 'rebuilding index'::text - WHEN 7 THEN 'performing final cleanup'::text - ELSE NULL::text - END AS phase, - (s.param3)::oid AS cluster_index_relid, - s.param4 AS heap_tuples_scanned, - s.param5 AS heap_tuples_written, - s.param6 AS heap_blks_total, - s.param7 AS heap_blks_scanned, - s.param8 AS index_rebuild_count - FROM (pg_stat_get_progress_info('CLUSTER'::text) s(pid, datid, relid, p= aram1, param2, param3, param4, param5, param6, param7, param8, param9, para= m10, param11, param12, param13, param14, param15, param16, param17, param18= , param19, param20) - LEFT JOIN pg_database d ON ((s.datid =3D d.oid))); + phase, + repack_index_relid AS cluster_index_relid, + heap_tuples_scanned, + heap_tuples_written, + heap_blks_total, + heap_blks_scanned, + index_rebuild_count + FROM pg_stat_progress_repack; pg_stat_progress_copy| SELECT s.pid, s.datid, d.datname, @@ -2082,6 +2071,35 @@ pg_stat_progress_create_index| SELECT s.pid, s.param15 AS partitions_done FROM (pg_stat_get_progress_info('CREATE INDEX'::text) s(pid, datid, rel= id, param1, param2, param3, param4, param5, param6, param7, param8, param9,= param10, param11, param12, param13, param14, param15, param16, param17, pa= ram18, param19, param20) LEFT JOIN pg_database d ON ((s.datid =3D d.oid))); +pg_stat_progress_repack| SELECT s.pid, + s.datid, + d.datname, + s.relid, + CASE s.param1 + WHEN 1 THEN 'CLUSTER'::text + WHEN 2 THEN 'REPACK'::text + WHEN 3 THEN 'VACUUM FULL'::text + ELSE NULL::text + END AS command, + CASE s.param2 + WHEN 0 THEN 'initializing'::text + WHEN 1 THEN 'seq scanning heap'::text + WHEN 2 THEN 'index scanning heap'::text + WHEN 3 THEN 'sorting tuples'::text + WHEN 4 THEN 'writing new heap'::text + WHEN 5 THEN 'swapping relation files'::text + WHEN 6 THEN 'rebuilding index'::text + WHEN 7 THEN 'performing final cleanup'::text + ELSE NULL::text + END AS phase, + (s.param3)::oid AS repack_index_relid, + s.param4 AS heap_tuples_scanned, + s.param5 AS heap_tuples_written, + s.param6 AS heap_blks_total, + s.param7 AS heap_blks_scanned, + s.param8 AS index_rebuild_count + FROM (pg_stat_get_progress_info('REPACK'::text) s(pid, datid, relid, pa= ram1, param2, param3, param4, param5, param6, param7, param8, param9, param= 10, param11, param12, param13, param14, param15, param16, param17, param18,= param19, param20) + LEFT JOIN pg_database d ON ((s.datid =3D d.oid))); pg_stat_progress_vacuum| SELECT s.pid, s.datid, d.datname, diff --git a/src/test/regress/sql/cluster.sql b/src/test/regress/sql/cluste= r.sql index b7115f86104..c976823a3cb 100644 --- a/src/test/regress/sql/cluster.sql +++ b/src/test/regress/sql/cluster.sql @@ -76,7 +76,6 @@ INSERT INTO clstr_tst (b, c) VALUES (1111, 'this should f= ail'); SELECT conname FROM pg_constraint WHERE conrelid =3D 'clstr_tst'::regclass ORDER BY 1; =20 - SELECT relname, relkind, EXISTS(SELECT 1 FROM pg_class WHERE oid =3D c.reltoastrelid) AS hastoa= st FROM pg_class c WHERE relname LIKE 'clstr_tst%' ORDER BY relname; @@ -229,6 +228,24 @@ SELECT relname, old.level, old.relkind, old.relfilenod= e =3D new.relfilenode FROM o CLUSTER clstrpart; ALTER TABLE clstrpart SET WITHOUT CLUSTER; ALTER TABLE clstrpart CLUSTER ON clstrpart_idx; + +-- Check that REPACK sets new relfilenodes: it should process exactly the = same +-- tables as CLUSTER did. +DROP TABLE old_cluster_info; +DROP TABLE new_cluster_info; +CREATE TEMP TABLE old_cluster_info AS SELECT relname, level, relfilenode, = relkind FROM pg_partition_tree('clstrpart'::regclass) AS tree JOIN pg_class= c ON c.oid=3Dtree.relid ; +REPACK clstrpart USING INDEX clstrpart_idx; +CREATE TEMP TABLE new_cluster_info AS SELECT relname, level, relfilenode, = relkind FROM pg_partition_tree('clstrpart'::regclass) AS tree JOIN pg_class= c ON c.oid=3Dtree.relid ; +SELECT relname, old.level, old.relkind, old.relfilenode =3D new.relfilenod= e FROM old_cluster_info AS old JOIN new_cluster_info AS new USING (relname)= ORDER BY relname COLLATE "C"; + +-- And finally the same for REPACK w/o index. +DROP TABLE old_cluster_info; +DROP TABLE new_cluster_info; +CREATE TEMP TABLE old_cluster_info AS SELECT relname, level, relfilenode, = relkind FROM pg_partition_tree('clstrpart'::regclass) AS tree JOIN pg_class= c ON c.oid=3Dtree.relid ; +REPACK clstrpart; +CREATE TEMP TABLE new_cluster_info AS SELECT relname, level, relfilenode, = relkind FROM pg_partition_tree('clstrpart'::regclass) AS tree JOIN pg_class= c ON c.oid=3Dtree.relid ; +SELECT relname, old.level, old.relkind, old.relfilenode =3D new.relfilenod= e FROM old_cluster_info AS old JOIN new_cluster_info AS new USING (relname)= ORDER BY relname COLLATE "C"; + DROP TABLE clstrpart; =20 -- Ownership of partitions is checked @@ -313,6 +330,57 @@ EXPLAIN (COSTS OFF) SELECT * FROM clstr_expression WHE= RE -a =3D -3 ORDER BY -a, b; SELECT * FROM clstr_expression WHERE -a =3D -3 ORDER BY -a, b; COMMIT; =20 +---------------------------------------------------------------------- +-- +-- REPACK +-- +---------------------------------------------------------------------- + +-- REPACK handles individual tables identically to CLUSTER, but it's worth +-- checking if it handles table hierarchies identically as well. +REPACK clstr_tst USING INDEX clstr_tst_c; + +-- Verify that inheritance link still works +INSERT INTO clstr_tst_inh VALUES (0, 100, 'in child table 2'); +SELECT a,b,c,substring(d for 30), length(d) from clstr_tst; + +-- Verify that foreign key link still works +INSERT INTO clstr_tst (b, c) VALUES (1111, 'this should fail'); + +SELECT conname FROM pg_constraint WHERE conrelid =3D 'clstr_tst'::regclass +ORDER BY 1; + +-- Verify partial analyze works +REPACK (ANALYZE) clstr_tst (a); +REPACK (ANALYZE) clstr_tst; +REPACK (VERBOSE) clstr_tst (a); + +-- REPACK w/o argument performs no ordering, so we can only check which ta= bles +-- have the relfilenode changed. +RESET SESSION AUTHORIZATION; +CREATE TEMP TABLE relnodes_old AS +(SELECT relname, relfilenode +FROM pg_class +WHERE relname IN ('clstr_1', 'clstr_2', 'clstr_3')); + +SET SESSION AUTHORIZATION regress_clstr_user; +SET client_min_messages =3D ERROR; -- order of "skipping" warnings may va= ry +REPACK; +RESET client_min_messages; + +RESET SESSION AUTHORIZATION; +CREATE TEMP TABLE relnodes_new AS +(SELECT relname, relfilenode +FROM pg_class +WHERE relname IN ('clstr_1', 'clstr_2', 'clstr_3')); + +-- Do the actual comparison. Unlike CLUSTER, clstr_3 should have been +-- processed because there is nothing like clustering index here. +SELECT o.relname FROM relnodes_old o +JOIN relnodes_new n ON o.relname =3D n.relname +WHERE o.relfilenode <> n.relfilenode +ORDER BY o.relname; + -- clean up DROP TABLE clustertest; DROP TABLE clstr_1; diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs= .list index c1ad80a418d..4641da9b746 100644 --- a/src/tools/pgindent/typedefs.list +++ b/src/tools/pgindent/typedefs.list @@ -2555,6 +2555,8 @@ ReorderBufferTupleCidKey ReorderBufferUpdateProgressTxnCB ReorderTuple RepOriginId +RepackCommand +RepackStmt ReparameterizeForeignPathByChild_function ReplaceVarsFromTargetList_context ReplaceVarsNoMatchOption --=20 2.47.3 --=-=-= Content-Type: text/x-diff Content-Disposition: attachment; filename=v26-0002-Refactor-index_concurrently_create_copy-for-use-with.patch From 89b8c72639a7158726193dc799996ba0d9ddc74e Mon Sep 17 00:00:00 2001 From: Antonin Houska Date: Thu, 4 Dec 2025 18:20:07 +0100 Subject: [PATCH 2/4] Refactor index_concurrently_create_copy() for use with REPACK (CONCURRENTLY). This patch moves the code to index_create_copy() and adds a "concurrently" parameter so it can be used by REPACK (CONCURRENTLY). With the CONCURRENTLY option, REPACK cannot simply swap the heap file and rebuild its indexes. Instead, it needs to build a separate set of indexes (including system catalog entries) *before* the actual swap, to reduce the time AccessExclusiveLock needs to be held for. --- src/backend/catalog/index.c | 54 +++++++++++++++++++++++--------- src/backend/commands/indexcmds.c | 6 ++-- src/backend/nodes/makefuncs.c | 9 +++--- src/include/catalog/index.h | 3 ++ src/include/nodes/makefuncs.h | 4 ++- 5 files changed, 54 insertions(+), 22 deletions(-) diff --git a/src/backend/catalog/index.c b/src/backend/catalog/index.c index 08d4b8e44d7..cf2d0abf370 100644 --- a/src/backend/catalog/index.c +++ b/src/backend/catalog/index.c @@ -1290,15 +1290,32 @@ index_create(Relation heapRelation, /* * index_concurrently_create_copy * - * Create concurrently an index based on the definition of the one provided by - * caller. The index is inserted into catalogs and needs to be built later - * on. This is called during concurrent reindex processing. - * - * "tablespaceOid" is the tablespace to use for this index. + * Variant of index_create_copy(), called during concurrent reindex + * processing. */ Oid index_concurrently_create_copy(Relation heapRelation, Oid oldIndexId, Oid tablespaceOid, const char *newName) +{ + return index_create_copy(heapRelation, oldIndexId, tablespaceOid, newName, + true); +} + +/* + * index_create_copy + * + * Create an index based on the definition of the one provided by caller. The + * index is inserted into catalogs. If 'concurrently' is TRUE, it needs to be + * built later on, otherwise it's built immediately. + * + * "tablespaceOid" is the tablespace to use for this index. + * + * The actual implementation of index_concurrently_create_copy(), reusable for + * other purposes. + */ +Oid +index_create_copy(Relation heapRelation, Oid oldIndexId, Oid tablespaceOid, + const char *newName, bool concurrently) { Relation indexRelation; IndexInfo *oldInfo, @@ -1317,6 +1334,7 @@ index_concurrently_create_copy(Relation heapRelation, Oid oldIndexId, List *indexColNames = NIL; List *indexExprs = NIL; List *indexPreds = NIL; + int flags = 0; indexRelation = index_open(oldIndexId, RowExclusiveLock); @@ -1327,7 +1345,7 @@ index_concurrently_create_copy(Relation heapRelation, Oid oldIndexId, * Concurrent build of an index with exclusion constraints is not * supported. */ - if (oldInfo->ii_ExclusionOps != NULL) + if (oldInfo->ii_ExclusionOps != NULL && concurrently) ereport(ERROR, (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), errmsg("concurrent index creation for exclusion constraints is not supported"))); @@ -1383,9 +1401,7 @@ index_concurrently_create_copy(Relation heapRelation, Oid oldIndexId, } /* - * Build the index information for the new index. Note that rebuild of - * indexes with exclusion constraints is not supported, hence there is no - * need to fill all the ii_Exclusion* fields. + * Build the index information for the new index. */ newInfo = makeIndexInfo(oldInfo->ii_NumIndexAttrs, oldInfo->ii_NumIndexKeyAttrs, @@ -1394,10 +1410,13 @@ index_concurrently_create_copy(Relation heapRelation, Oid oldIndexId, indexPreds, oldInfo->ii_Unique, oldInfo->ii_NullsNotDistinct, - false, /* not ready for inserts */ - true, + !concurrently, /* isready */ + concurrently, /* concurrent */ indexRelation->rd_indam->amsummarizing, - oldInfo->ii_WithoutOverlaps); + oldInfo->ii_WithoutOverlaps, + oldInfo->ii_ExclusionOps, + oldInfo->ii_ExclusionProcs, + oldInfo->ii_ExclusionStrats); /* * Extract the list of column names and the column numbers for the new @@ -1435,6 +1454,9 @@ index_concurrently_create_copy(Relation heapRelation, Oid oldIndexId, stattargets[i].isnull = isnull; } + if (concurrently) + flags = INDEX_CREATE_SKIP_BUILD | INDEX_CREATE_CONCURRENT; + /* * Now create the new index. * @@ -1458,7 +1480,7 @@ index_concurrently_create_copy(Relation heapRelation, Oid oldIndexId, indcoloptions->values, stattargets, reloptionsDatum, - INDEX_CREATE_SKIP_BUILD | INDEX_CREATE_CONCURRENT, + flags, 0, true, /* allow table to be a system catalog? */ false, /* is_internal? */ @@ -2452,7 +2474,8 @@ BuildIndexInfo(Relation index) indexStruct->indisready, false, index->rd_indam->amsummarizing, - indexStruct->indisexclusion && indexStruct->indisunique); + indexStruct->indisexclusion && indexStruct->indisunique, + NULL, NULL, NULL); /* fill in attribute numbers */ for (i = 0; i < numAtts; i++) @@ -2512,7 +2535,8 @@ BuildDummyIndexInfo(Relation index) indexStruct->indisready, false, index->rd_indam->amsummarizing, - indexStruct->indisexclusion && indexStruct->indisunique); + indexStruct->indisexclusion && indexStruct->indisunique, + NULL, NULL, NULL); /* fill in attribute numbers */ for (i = 0; i < numAtts; i++) diff --git a/src/backend/commands/indexcmds.c b/src/backend/commands/indexcmds.c index a8033be4bff..9cc94884abc 100644 --- a/src/backend/commands/indexcmds.c +++ b/src/backend/commands/indexcmds.c @@ -242,7 +242,8 @@ CheckIndexCompatible(Oid oldId, */ indexInfo = makeIndexInfo(numberOfAttributes, numberOfAttributes, accessMethodId, NIL, NIL, false, false, - false, false, amsummarizing, isWithoutOverlaps); + false, false, amsummarizing, isWithoutOverlaps, + NULL, NULL, NULL); typeIds = palloc_array(Oid, numberOfAttributes); collationIds = palloc_array(Oid, numberOfAttributes); opclassIds = palloc_array(Oid, numberOfAttributes); @@ -927,7 +928,8 @@ DefineIndex(Oid tableId, !concurrent, concurrent, amissummarizing, - stmt->iswithoutoverlaps); + stmt->iswithoutoverlaps, + NULL, NULL, NULL); typeIds = palloc_array(Oid, numberOfAttributes); collationIds = palloc_array(Oid, numberOfAttributes); diff --git a/src/backend/nodes/makefuncs.c b/src/backend/nodes/makefuncs.c index e2d9e9be41a..c5d5a37f514 100644 --- a/src/backend/nodes/makefuncs.c +++ b/src/backend/nodes/makefuncs.c @@ -834,7 +834,8 @@ IndexInfo * makeIndexInfo(int numattrs, int numkeyattrs, Oid amoid, List *expressions, List *predicates, bool unique, bool nulls_not_distinct, bool isready, bool concurrent, bool summarizing, - bool withoutoverlaps) + bool withoutoverlaps, Oid *exclusion_ops, Oid *exclusion_procs, + uint16 *exclusion_strats) { IndexInfo *n = makeNode(IndexInfo); @@ -863,9 +864,9 @@ makeIndexInfo(int numattrs, int numkeyattrs, Oid amoid, List *expressions, n->ii_PredicateState = NULL; /* exclusion constraints */ - n->ii_ExclusionOps = NULL; - n->ii_ExclusionProcs = NULL; - n->ii_ExclusionStrats = NULL; + n->ii_ExclusionOps = exclusion_ops; + n->ii_ExclusionProcs = exclusion_procs; + n->ii_ExclusionStrats = exclusion_strats; /* speculative inserts */ n->ii_UniqueOps = NULL; diff --git a/src/include/catalog/index.h b/src/include/catalog/index.h index dda95e54903..4bf909078d8 100644 --- a/src/include/catalog/index.h +++ b/src/include/catalog/index.h @@ -99,6 +99,9 @@ extern Oid index_concurrently_create_copy(Relation heapRelation, Oid oldIndexId, Oid tablespaceOid, const char *newName); +extern Oid index_create_copy(Relation heapRelation, Oid oldIndexId, + Oid tablespaceOid, const char *newName, + bool concurrently); extern void index_concurrently_build(Oid heapRelationId, Oid indexRelationId); diff --git a/src/include/nodes/makefuncs.h b/src/include/nodes/makefuncs.h index 5473ce9a288..9ff7159ff0c 100644 --- a/src/include/nodes/makefuncs.h +++ b/src/include/nodes/makefuncs.h @@ -99,7 +99,9 @@ extern IndexInfo *makeIndexInfo(int numattrs, int numkeyattrs, Oid amoid, List *expressions, List *predicates, bool unique, bool nulls_not_distinct, bool isready, bool concurrent, - bool summarizing, bool withoutoverlaps); + bool summarizing, bool withoutoverlaps, + Oid *exclusion_ops, Oid *exclusion_procs, + uint16 *exclusion_strats); extern Node *makeStringConst(char *str, int location); extern DefElem *makeDefElem(char *name, Node *arg, int location); -- 2.47.3 --=-=-= Content-Type: text/x-diff Content-Disposition: attachment; filename=v26-0003-Move-conversion-of-a-historic-to-MVCC-snapshot-to-a-.patch From 79b67f90e1be61945797252ce8c5ffc481ab67f9 Mon Sep 17 00:00:00 2001 From: Antonin Houska Date: Thu, 4 Dec 2025 18:20:07 +0100 Subject: [PATCH 3/4] Move conversion of a "historic" to MVCC snapshot to a separate function. The conversion is now handled by SnapBuildMVCCFromHistoric(). REPACK CONCURRENTLY will also need it. --- src/backend/replication/logical/snapbuild.c | 51 +++++++++++++++++---- src/backend/utils/time/snapmgr.c | 3 +- src/include/replication/snapbuild.h | 1 + src/include/utils/snapmgr.h | 1 + 4 files changed, 45 insertions(+), 11 deletions(-) diff --git a/src/backend/replication/logical/snapbuild.c b/src/backend/replication/logical/snapbuild.c index 6e18baa33cb..dcf32101c4c 100644 --- a/src/backend/replication/logical/snapbuild.c +++ b/src/backend/replication/logical/snapbuild.c @@ -440,10 +440,7 @@ Snapshot SnapBuildInitialSnapshot(SnapBuild *builder) { Snapshot snap; - TransactionId xid; TransactionId safeXid; - TransactionId *newxip; - int newxcnt = 0; Assert(XactIsoLevel == XACT_REPEATABLE_READ); Assert(builder->building_full_snapshot); @@ -485,6 +482,31 @@ SnapBuildInitialSnapshot(SnapBuild *builder) MyProc->xmin = snap->xmin; + /* Convert the historic snapshot to MVCC snapshot. */ + return SnapBuildMVCCFromHistoric(snap, true); +} + +/* + * Turn a historic MVCC snapshot into an ordinary MVCC snapshot. + * + * Unlike a regular (non-historic) MVCC snapshot, the xip array of this + * snapshot contains not only running main transactions, but also their + * subtransactions. This difference does has no impact on XidInMVCCSnapshot(). + * + * Pass true for 'in_place' if you don't care about modifying the source + * snapshot. If you need a new instance, and one that was allocated as a + * single chunk of memory, pass false. + */ +Snapshot +SnapBuildMVCCFromHistoric(Snapshot snapshot, bool in_place) +{ + TransactionId xid; + TransactionId *oldxip = snapshot->xip; + uint32 oldxcnt = snapshot->xcnt; + TransactionId *newxip; + int newxcnt = 0; + Snapshot result; + /* allocate in transaction context */ newxip = (TransactionId *) palloc(sizeof(TransactionId) * GetMaxSnapshotXidCount()); @@ -495,7 +517,7 @@ SnapBuildInitialSnapshot(SnapBuild *builder) * classical snapshot by marking all non-committed transactions as * in-progress. This can be expensive. */ - for (xid = snap->xmin; NormalTransactionIdPrecedes(xid, snap->xmax);) + for (xid = snapshot->xmin; NormalTransactionIdPrecedes(xid, snapshot->xmax);) { void *test; @@ -503,7 +525,7 @@ SnapBuildInitialSnapshot(SnapBuild *builder) * Check whether transaction committed using the decoding snapshot * meaning of ->xip. */ - test = bsearch(&xid, snap->xip, snap->xcnt, + test = bsearch(&xid, snapshot->xip, snapshot->xcnt, sizeof(TransactionId), xidComparator); if (test == NULL) @@ -520,11 +542,22 @@ SnapBuildInitialSnapshot(SnapBuild *builder) } /* adjust remaining snapshot fields as needed */ - snap->snapshot_type = SNAPSHOT_MVCC; - snap->xcnt = newxcnt; - snap->xip = newxip; + snapshot->xcnt = newxcnt; + snapshot->xip = newxip; + + if (in_place) + result = snapshot; + else + { + result = CopySnapshot(snapshot); + + /* Restore the original values so the source is intact. */ + snapshot->xip = oldxip; + snapshot->xcnt = oldxcnt; + } + result->snapshot_type = SNAPSHOT_MVCC; - return snap; + return result; } /* diff --git a/src/backend/utils/time/snapmgr.c b/src/backend/utils/time/snapmgr.c index 24f73a49d27..886060305f5 100644 --- a/src/backend/utils/time/snapmgr.c +++ b/src/backend/utils/time/snapmgr.c @@ -213,7 +213,6 @@ typedef struct ExportedSnapshot static List *exportedSnapshots = NIL; /* Prototypes for local functions */ -static Snapshot CopySnapshot(Snapshot snapshot); static void UnregisterSnapshotNoOwner(Snapshot snapshot); static void FreeSnapshot(Snapshot snapshot); static void SnapshotResetXmin(void); @@ -604,7 +603,7 @@ SetTransactionSnapshot(Snapshot sourcesnap, VirtualTransactionId *sourcevxid, * The copy is palloc'd in TopTransactionContext and has initial refcounts set * to 0. The returned snapshot has the copied flag set. */ -static Snapshot +Snapshot CopySnapshot(Snapshot snapshot) { Snapshot newsnap; diff --git a/src/include/replication/snapbuild.h b/src/include/replication/snapbuild.h index 44031dcf6e3..6d4d2d1814c 100644 --- a/src/include/replication/snapbuild.h +++ b/src/include/replication/snapbuild.h @@ -73,6 +73,7 @@ extern void FreeSnapshotBuilder(SnapBuild *builder); extern void SnapBuildSnapDecRefcount(Snapshot snap); extern Snapshot SnapBuildInitialSnapshot(SnapBuild *builder); +extern Snapshot SnapBuildMVCCFromHistoric(Snapshot snapshot, bool in_place); extern const char *SnapBuildExportSnapshot(SnapBuild *builder); extern void SnapBuildClearExportedSnapshot(void); extern void SnapBuildResetExportedSnapshotState(void); diff --git a/src/include/utils/snapmgr.h b/src/include/utils/snapmgr.h index 604c1f90216..f65f83c85cd 100644 --- a/src/include/utils/snapmgr.h +++ b/src/include/utils/snapmgr.h @@ -63,6 +63,7 @@ extern Snapshot GetTransactionSnapshot(void); extern Snapshot GetLatestSnapshot(void); extern void SnapshotSetCommandId(CommandId curcid); +extern Snapshot CopySnapshot(Snapshot snapshot); extern Snapshot GetCatalogSnapshot(Oid relid); extern Snapshot GetNonHistoricCatalogSnapshot(Oid relid); extern void InvalidateCatalogSnapshot(void); -- 2.47.3 --=-=-= Content-Type: text/plain Content-Disposition: attachment; filename=v26-0004-Add-CONCURRENTLY-option-to-REPACK-command.patch From 1a181bb4b334554dafd1d8fd3ba5974effffed78 Mon Sep 17 00:00:00 2001 From: Antonin Houska Date: Thu, 4 Dec 2025 18:20:07 +0100 Subject: [PATCH 4/4] Add CONCURRENTLY option to REPACK command. The REPACK command copies the relation data into a new file, creates new indexes and eventually swaps the files. To make sure that the old file does not change during the copying, the relation is locked in an exclusive mode, which prevents applications from both reading and writing. (To keep the data consistent, we'd only need to prevent the applications from writing, but even reading needs to be blocked before we can swap the files - otherwise some applications could continue using the old file. Since we should not request a stronger lock without releasing the weaker one first, we acquire the exclusive lock in the beginning and keep it till the end of the processing.) This patch introduces an alternative workflow, which only requires the exclusive lock when the relation (and index) files are being swapped. (Supposedly, the swapping should be pretty fast.) On the other hand, when we copy the data to the new file, we allow applications to read from the relation and even to write to it. First, we scan the relation using a "historic snapshot", and insert all the tuples satisfying this snapshot into the new file. Second, logical decoding is used to capture the data changes done by applications during the copying (i.e. changes that do not satisfy the historic snapshot mentioned above), and those are applied to the new file before we acquire the exclusive lock that we need to swap the files. (Of course, more data changes can take place while we are waiting for the lock - these will be applied to the new file after we have acquired the lock, before we swap the files.) Since the logical decoding system, during its startup, waits until all the transactions which already have XID assigned have finished, there is a risk of deadlock if a transaction that already changed anything in the database tries to acquire a conflicting lock on the table REPACK CONCURRENTLY is working on. As an example, consider transaction running CREATE INDEX command on the table that is being REPACKed CONCURRENTLY. On the other hand, DML commands (INSERT, UPDATE, DELETE) are not a problem as their lock does not conflict with REPACK CONCURRENTLY. The current approach is that we accept the risk. If we tried to avoid it, it'd be necessary to unlock the table before the logical decoding is setup and lock it again afterwards. Such temporary unlocking would imply re-checking if the table still meets all the requirements for REPACK CONCURRENTLY. Like the existing implementation of REPACK, the variant with the CONCURRENTLY option also requires an extra space for the new relation and index files (which coexist with the old files for some time). In addition, the CONCURRENTLY option might introduce a lag in releasing WAL segments for archiving / recycling. This is due to the decoding of the data changes done by applications concurrently. When copying the table contents into the new file, we check the lag periodically. If it exceeds the size of a WAL segment, we decode all the available WAL before resuming the copying. (Of course, the changes are not applied until the whole table contents is copied.) A background worker might be a better approach for the decoding - let's consider implementing it in the future. The WAL records produced by running DML commands on the new relation do not contain enough information to be processed by the logical decoding system. All we need from the new relation is the file (relfilenode), while the actual relation is eventually dropped. Thus there is no point in replaying the DMLs anywhere. --- doc/src/sgml/monitoring.sgml | 37 +- doc/src/sgml/mvcc.sgml | 12 +- doc/src/sgml/ref/repack.sgml | 129 +- src/Makefile | 1 + src/backend/access/heap/heapam.c | 34 +- src/backend/access/heap/heapam_handler.c | 227 ++- src/backend/access/heap/rewriteheap.c | 6 +- src/backend/access/transam/xact.c | 11 +- src/backend/catalog/system_views.sql | 19 +- src/backend/commands/cluster.c | 1659 ++++++++++++++++- src/backend/commands/matview.c | 2 +- src/backend/commands/tablecmds.c | 1 + src/backend/commands/vacuum.c | 12 +- src/backend/meson.build | 1 + src/backend/replication/logical/decode.c | 93 + src/backend/replication/logical/snapbuild.c | 23 +- .../replication/pgoutput_repack/Makefile | 32 + .../replication/pgoutput_repack/meson.build | 18 + .../pgoutput_repack/pgoutput_repack.c | 248 +++ .../storage/lmgr/generate-lwlocknames.pl | 2 +- src/backend/utils/cache/relcache.c | 1 + src/backend/utils/time/snapmgr.c | 3 +- src/bin/psql/tab-complete.in.c | 4 +- src/include/access/heapam.h | 9 +- src/include/access/heapam_xlog.h | 2 + src/include/access/tableam.h | 10 + src/include/commands/cluster.h | 90 +- src/include/commands/progress.h | 17 +- src/include/replication/snapbuild.h | 1 + src/include/storage/lockdefs.h | 4 +- src/include/utils/snapmgr.h | 2 + src/test/modules/injection_points/Makefile | 2 + .../injection_points/expected/repack.out | 113 ++ .../modules/injection_points/logical.conf | 1 + src/test/modules/injection_points/meson.build | 3 + .../injection_points/specs/repack.spec | 143 ++ src/test/regress/expected/rules.out | 19 +- src/tools/pgindent/typedefs.list | 5 + 38 files changed, 2761 insertions(+), 235 deletions(-) create mode 100644 src/backend/replication/pgoutput_repack/Makefile create mode 100644 src/backend/replication/pgoutput_repack/meson.build create mode 100644 src/backend/replication/pgoutput_repack/pgoutput_repack.c create mode 100644 src/test/modules/injection_points/expected/repack.out create mode 100644 src/test/modules/injection_points/logical.conf create mode 100644 src/test/modules/injection_points/specs/repack.spec diff --git a/doc/src/sgml/monitoring.sgml b/doc/src/sgml/monitoring.sgml index b8da77b4d89..7d9e60c1b20 100644 --- a/doc/src/sgml/monitoring.sgml +++ b/doc/src/sgml/monitoring.sgml @@ -6177,14 +6177,35 @@ FROM pg_stat_get_backend_idset() AS backendid; - heap_tuples_written bigint + heap_tuples_inserted bigint - Number of heap tuples written. + Number of heap tuples inserted. This counter only advances when the phase is seq scanning heap, - index scanning heap - or writing new heap. + index scanning heap, + writing new heap + or catch-up. + + + + + + heap_tuples_updated bigint + + + Number of heap tuples updated. + This counter only advances when the phase is catch-up. + + + + + + heap_tuples_deleted bigint + + + Number of heap tuples deleted. + This counter only advances when the phase is catch-up. @@ -6265,6 +6286,14 @@ FROM pg_stat_get_backend_idset() AS backendid; REPACK is currently writing the new heap. + + catch-up + + REPACK CONCURRENTLY is currently processing the DML + commands that other transactions executed during any of the preceding + phase. + + swapping relation files diff --git a/doc/src/sgml/mvcc.sgml b/doc/src/sgml/mvcc.sgml index 049ee75a4ba..0f5c34af542 100644 --- a/doc/src/sgml/mvcc.sgml +++ b/doc/src/sgml/mvcc.sgml @@ -1833,15 +1833,17 @@ SELECT pg_advisory_lock(q.id) FROM Caveats - Some DDL commands, currently only TRUNCATE and the - table-rewriting forms of ALTER TABLE, are not + Some commands, currently only TRUNCATE, the + table-rewriting forms of ALTER + TABLE and REPACK with + the CONCURRENTLY option, are not MVCC-safe. This means that after the truncation or rewrite commits, the table will appear empty to concurrent transactions, if they are using a - snapshot taken before the DDL command committed. This will only be an + snapshot taken before the command committed. This will only be an issue for a transaction that did not access the table in question - before the DDL command started — any transaction that has done so + before the command started — any transaction that has done so would hold at least an ACCESS SHARE table lock, - which would block the DDL command until that transaction completes. + which would block the truncating or rewriting command until that transaction completes. So these commands will not cause any apparent inconsistency in the table contents for successive queries on the target table, but they could cause visible inconsistency between the contents of the target diff --git a/doc/src/sgml/ref/repack.sgml b/doc/src/sgml/ref/repack.sgml index 61d5c2cdef1..2765ce4100e 100644 --- a/doc/src/sgml/ref/repack.sgml +++ b/doc/src/sgml/ref/repack.sgml @@ -28,6 +28,7 @@ REPACK [ ( option [, ...] ) ] USING VERBOSE [ boolean ] ANALYZE [ boolean ] + CONCURRENTLY [ boolean ] and table_and_columns is: @@ -54,7 +55,8 @@ REPACK [ ( option [, ...] ) ] USING processes every table and materialized view in the current database that the current user has the MAINTAIN privilege on. This form of REPACK cannot be executed inside a transaction - block. + block. Also, this form is not allowed if + the CONCURRENTLY option is used. @@ -67,7 +69,8 @@ REPACK [ ( option [, ...] ) ] USING When a table is being repacked, an ACCESS EXCLUSIVE lock is acquired on it. This prevents any other database operations (both reads and writes) from operating on the table until the REPACK - is finished. + is finished. If you want to keep the table accessible during the repacking, + consider using the CONCURRENTLY option. @@ -195,6 +198,128 @@ REPACK [ ( option [, ...] ) ] USING + + CONCURRENTLY + + + Allow other transactions to use the table while it is being repacked. + + + + Internally, REPACK copies the contents of the table + (ignoring dead tuples) into a new file, sorted by the specified index, + and also creates a new file for each index. Then it swaps the old and + new files for the table and all the indexes, and deletes the old + files. The ACCESS EXCLUSIVE lock is needed to make + sure that the old files do not change during the processing because the + changes would get lost due to the swap. + + + + With the CONCURRENTLY option, the ACCESS + EXCLUSIVE lock is only acquired to swap the table and index + files. The data changes that took place during the creation of the new + table and index files are captured using logical decoding + () and applied before + the ACCESS EXCLUSIVE lock is requested. Thus the lock + is typically held only for the time needed to swap the files, which + should be pretty short. However, the time might still be noticeable if + too many data changes have been done to the table while + REPACK was waiting for the lock: those changes must + be processed just before the files are swapped, while the + ACCESS EXCLUSIVE lock is being held. + + + + Note that REPACK with the + the CONCURRENTLY option does not try to order the + rows inserted into the table after the repacking started. Also + note REPACK might fail to complete due to DDL + commands executed on the table by other transactions during the + repacking. + + + + + In addition to the temporary space requirements explained in + , + the CONCURRENTLY option can add to the usage of + temporary space a bit more. The reason is that other transactions can + perform DML operations which cannot be applied to the new file until + REPACK has copied all the tuples from the old + file. Thus the tuples inserted into the old file during the copying are + also stored separately in a temporary file, so they can eventually be + applied to the new file. + + + + Furthermore, the data changes performed during the copying are + extracted from write-ahead log (WAL), and + this extraction (decoding) only takes place when certain amount of WAL + has been written. Therefore, WAL removal can be delayed by this + threshold. Currently the threshold is equal to the value of + the wal_segment_size + configuration parameter. + + + + + The CONCURRENTLY option cannot be used in the + following cases: + + + + + The table is UNLOGGED. + + + + + + The table is partitioned. + + + + + + The table is a system catalog or a TOAST table. + + + + + + REPACK is executed inside a transaction block. + + + + + + The wal_level + configuration parameter is less than logical. + + + + + + The max_replication_slots + configuration parameter does not allow for creation of an additional + replication slot. + + + + + + + + REPACK with the CONCURRENTLY + option is not MVCC-safe, see for + details. + + + + + + VERBOSE diff --git a/src/Makefile b/src/Makefile index 2f31a2f20a7..b18c9a14ffa 100644 --- a/src/Makefile +++ b/src/Makefile @@ -23,6 +23,7 @@ SUBDIRS = \ interfaces \ backend/replication/libpqwalreceiver \ backend/replication/pgoutput \ + backend/replication/pgoutput_repack \ fe_utils \ bin \ pl \ diff --git a/src/backend/access/heap/heapam.c b/src/backend/access/heap/heapam.c index 4d382a04338..f019f93bca6 100644 --- a/src/backend/access/heap/heapam.c +++ b/src/backend/access/heap/heapam.c @@ -60,7 +60,8 @@ static HeapTuple heap_prepare_insert(Relation relation, HeapTuple tup, static XLogRecPtr log_heap_update(Relation reln, Buffer oldbuf, Buffer newbuf, HeapTuple oldtup, HeapTuple newtup, HeapTuple old_key_tuple, - bool all_visible_cleared, bool new_all_visible_cleared); + bool all_visible_cleared, bool new_all_visible_cleared, + bool wal_logical); #ifdef USE_ASSERT_CHECKING static void check_lock_if_inplace_updateable_rel(Relation relation, const ItemPointerData *otid, @@ -2803,7 +2804,7 @@ xmax_infomask_changed(uint16 new_infomask, uint16 old_infomask) TM_Result heap_delete(Relation relation, const ItemPointerData *tid, CommandId cid, Snapshot crosscheck, bool wait, - TM_FailureData *tmfd, bool changingPart) + TM_FailureData *tmfd, bool changingPart, bool wal_logical) { TM_Result result; TransactionId xid = GetCurrentTransactionId(); @@ -3050,7 +3051,8 @@ l1: * Compute replica identity tuple before entering the critical section so * we don't PANIC upon a memory allocation failure. */ - old_key_tuple = ExtractReplicaIdentity(relation, &tp, true, &old_key_copied); + old_key_tuple = wal_logical ? + ExtractReplicaIdentity(relation, &tp, true, &old_key_copied) : NULL; /* * If this is the first possibly-multixact-able operation in the current @@ -3140,6 +3142,15 @@ l1: xlrec.flags |= XLH_DELETE_CONTAINS_OLD_KEY; } + /* + * Unlike UPDATE, DELETE is decoded even if there is no old key, so it + * does not help to clear both XLH_DELETE_CONTAINS_OLD_TUPLE and + * XLH_DELETE_CONTAINS_OLD_KEY. Thus we need an extra flag. TODO + * Consider not decoding tuples w/o the old tuple/key instead. + */ + if (!wal_logical) + xlrec.flags |= XLH_DELETE_NO_LOGICAL; + XLogBeginInsert(); XLogRegisterData(&xlrec, SizeOfHeapDelete); @@ -3232,7 +3243,8 @@ simple_heap_delete(Relation relation, const ItemPointerData *tid) result = heap_delete(relation, tid, GetCurrentCommandId(true), InvalidSnapshot, true /* wait for commit */ , - &tmfd, false /* changingPart */ ); + &tmfd, false, /* changingPart */ + true /* wal_logical */ ); switch (result) { case TM_SelfModified: @@ -3273,7 +3285,7 @@ TM_Result heap_update(Relation relation, const ItemPointerData *otid, HeapTuple newtup, CommandId cid, Snapshot crosscheck, bool wait, TM_FailureData *tmfd, LockTupleMode *lockmode, - TU_UpdateIndexes *update_indexes) + TU_UpdateIndexes *update_indexes, bool wal_logical) { TM_Result result; TransactionId xid = GetCurrentTransactionId(); @@ -4166,7 +4178,8 @@ l2: newbuf, &oldtup, heaptup, old_key_tuple, all_visible_cleared, - all_visible_cleared_new); + all_visible_cleared_new, + wal_logical); if (newbuf != buffer) { PageSetLSN(BufferGetPage(newbuf), recptr); @@ -4524,7 +4537,8 @@ simple_heap_update(Relation relation, const ItemPointerData *otid, HeapTuple tup result = heap_update(relation, otid, tup, GetCurrentCommandId(true), InvalidSnapshot, true /* wait for commit */ , - &tmfd, &lockmode, update_indexes); + &tmfd, &lockmode, update_indexes, + true /* wal_logical */ ); switch (result) { case TM_SelfModified: @@ -8864,7 +8878,8 @@ static XLogRecPtr log_heap_update(Relation reln, Buffer oldbuf, Buffer newbuf, HeapTuple oldtup, HeapTuple newtup, HeapTuple old_key_tuple, - bool all_visible_cleared, bool new_all_visible_cleared) + bool all_visible_cleared, bool new_all_visible_cleared, + bool wal_logical) { xl_heap_update xlrec; xl_heap_header xlhdr; @@ -8875,7 +8890,8 @@ log_heap_update(Relation reln, Buffer oldbuf, suffixlen = 0; XLogRecPtr recptr; Page page = BufferGetPage(newbuf); - bool need_tuple_data = RelationIsLogicallyLogged(reln); + bool need_tuple_data = RelationIsLogicallyLogged(reln) && + wal_logical; bool init; int bufflags; diff --git a/src/backend/access/heap/heapam_handler.c b/src/backend/access/heap/heapam_handler.c index 79f9de5d760..01be29eb405 100644 --- a/src/backend/access/heap/heapam_handler.c +++ b/src/backend/access/heap/heapam_handler.c @@ -33,6 +33,7 @@ #include "catalog/index.h" #include "catalog/storage.h" #include "catalog/storage_xlog.h" +#include "commands/cluster.h" #include "commands/progress.h" #include "executor/executor.h" #include "miscadmin.h" @@ -309,7 +310,8 @@ heapam_tuple_delete(Relation relation, ItemPointer tid, CommandId cid, * the storage itself is cleaning the dead tuples by itself, it is the * time to call the index tuple deletion also. */ - return heap_delete(relation, tid, cid, crosscheck, wait, tmfd, changingPart); + return heap_delete(relation, tid, cid, crosscheck, wait, tmfd, changingPart, + true); } @@ -328,7 +330,7 @@ heapam_tuple_update(Relation relation, ItemPointer otid, TupleTableSlot *slot, tuple->t_tableOid = slot->tts_tableOid; result = heap_update(relation, otid, tuple, cid, crosscheck, wait, - tmfd, lockmode, update_indexes); + tmfd, lockmode, update_indexes, true); ItemPointerCopy(&tuple->t_self, &slot->tts_tid); /* @@ -685,13 +687,15 @@ static void heapam_relation_copy_for_cluster(Relation OldHeap, Relation NewHeap, Relation OldIndex, bool use_sort, TransactionId OldestXmin, + Snapshot snapshot, + LogicalDecodingContext *decoding_ctx, TransactionId *xid_cutoff, MultiXactId *multi_cutoff, double *num_tuples, double *tups_vacuumed, double *tups_recently_dead) { - RewriteState rwstate; + RewriteState rwstate = NULL; IndexScanDesc indexScan; TableScanDesc tableScan; HeapScanDesc heapScan; @@ -705,6 +709,8 @@ heapam_relation_copy_for_cluster(Relation OldHeap, Relation NewHeap, bool *isnull; BufferHeapTupleTableSlot *hslot; BlockNumber prev_cblock = InvalidBlockNumber; + bool concurrent = snapshot != NULL; + XLogRecPtr end_of_wal_prev = GetFlushRecPtr(NULL); /* Remember if it's a system catalog */ is_system_catalog = IsSystemRelation(OldHeap); @@ -720,9 +726,12 @@ heapam_relation_copy_for_cluster(Relation OldHeap, Relation NewHeap, values = (Datum *) palloc(natts * sizeof(Datum)); isnull = (bool *) palloc(natts * sizeof(bool)); - /* Initialize the rewrite operation */ - rwstate = begin_heap_rewrite(OldHeap, NewHeap, OldestXmin, *xid_cutoff, - *multi_cutoff); + /* + * Initialize the rewrite operation. + */ + if (!concurrent) + rwstate = begin_heap_rewrite(OldHeap, NewHeap, OldestXmin, + *xid_cutoff, *multi_cutoff); /* Set up sorting if wanted */ @@ -737,6 +746,9 @@ heapam_relation_copy_for_cluster(Relation OldHeap, Relation NewHeap, * Prepare to scan the OldHeap. To ensure we see recently-dead tuples * that still need to be copied, we scan with SnapshotAny and use * HeapTupleSatisfiesVacuum for the visibility test. + * + * In the CONCURRENTLY case, we do regular MVCC visibility tests, using + * the snapshot passed by the caller. */ if (OldIndex != NULL && !use_sort) { @@ -753,7 +765,9 @@ heapam_relation_copy_for_cluster(Relation OldHeap, Relation NewHeap, tableScan = NULL; heapScan = NULL; - indexScan = index_beginscan(OldHeap, OldIndex, SnapshotAny, NULL, 0, 0); + indexScan = index_beginscan(OldHeap, OldIndex, + snapshot ? snapshot : SnapshotAny, + NULL, 0, 0); index_rescan(indexScan, NULL, 0, NULL, 0); } else @@ -762,7 +776,9 @@ heapam_relation_copy_for_cluster(Relation OldHeap, Relation NewHeap, pgstat_progress_update_param(PROGRESS_REPACK_PHASE, PROGRESS_REPACK_PHASE_SEQ_SCAN_HEAP); - tableScan = table_beginscan(OldHeap, SnapshotAny, 0, (ScanKey) NULL); + tableScan = table_beginscan(OldHeap, + snapshot ? snapshot : SnapshotAny, + 0, (ScanKey) NULL); heapScan = (HeapScanDesc) tableScan; indexScan = NULL; @@ -837,70 +853,77 @@ heapam_relation_copy_for_cluster(Relation OldHeap, Relation NewHeap, tuple = ExecFetchSlotHeapTuple(slot, false, NULL); buf = hslot->buffer; - LockBuffer(buf, BUFFER_LOCK_SHARE); - - switch (HeapTupleSatisfiesVacuum(tuple, OldestXmin, buf)) + /* + * Regarding CONCURRENTLY, see the comments on MVCC snapshot above. + */ + if (!concurrent) { - case HEAPTUPLE_DEAD: - /* Definitely dead */ - isdead = true; - break; - case HEAPTUPLE_RECENTLY_DEAD: - *tups_recently_dead += 1; - /* fall through */ - case HEAPTUPLE_LIVE: - /* Live or recently dead, must copy it */ - isdead = false; - break; - case HEAPTUPLE_INSERT_IN_PROGRESS: + LockBuffer(buf, BUFFER_LOCK_SHARE); - /* - * Since we hold exclusive lock on the relation, normally the - * only way to see this is if it was inserted earlier in our - * own transaction. However, it can happen in system - * catalogs, since we tend to release write lock before commit - * there. Give a warning if neither case applies; but in any - * case we had better copy it. - */ - if (!is_system_catalog && - !TransactionIdIsCurrentTransactionId(HeapTupleHeaderGetXmin(tuple->t_data))) - elog(WARNING, "concurrent insert in progress within table \"%s\"", - RelationGetRelationName(OldHeap)); - /* treat as live */ - isdead = false; - break; - case HEAPTUPLE_DELETE_IN_PROGRESS: + switch (HeapTupleSatisfiesVacuum(tuple, OldestXmin, buf)) + { + case HEAPTUPLE_DEAD: + /* Definitely dead */ + isdead = true; + break; + case HEAPTUPLE_RECENTLY_DEAD: + *tups_recently_dead += 1; + /* fall through */ + case HEAPTUPLE_LIVE: + /* Live or recently dead, must copy it */ + isdead = false; + break; + case HEAPTUPLE_INSERT_IN_PROGRESS: - /* - * Similar situation to INSERT_IN_PROGRESS case. - */ - if (!is_system_catalog && - !TransactionIdIsCurrentTransactionId(HeapTupleHeaderGetUpdateXid(tuple->t_data))) - elog(WARNING, "concurrent delete in progress within table \"%s\"", - RelationGetRelationName(OldHeap)); - /* treat as recently dead */ - *tups_recently_dead += 1; - isdead = false; - break; - default: - elog(ERROR, "unexpected HeapTupleSatisfiesVacuum result"); - isdead = false; /* keep compiler quiet */ - break; - } + /* + * As long as we hold exclusive lock on the relation, + * normally the only way to see this is if it was inserted + * earlier in our own transaction. However, it can happen + * in system catalogs, since we tend to release write lock + * before commit there. Give a warning if neither case + * applies; but in any case we had better copy it. + */ + if (!is_system_catalog && + !TransactionIdIsCurrentTransactionId(HeapTupleHeaderGetXmin(tuple->t_data))) + elog(WARNING, "concurrent insert in progress within table \"%s\"", + RelationGetRelationName(OldHeap)); + /* treat as live */ + isdead = false; + break; + case HEAPTUPLE_DELETE_IN_PROGRESS: + + /* + * Similar situation to INSERT_IN_PROGRESS case. + */ + if (!is_system_catalog && + !TransactionIdIsCurrentTransactionId(HeapTupleHeaderGetUpdateXid(tuple->t_data))) + elog(WARNING, "concurrent delete in progress within table \"%s\"", + RelationGetRelationName(OldHeap)); + /* treat as recently dead */ + *tups_recently_dead += 1; + isdead = false; + break; + default: + elog(ERROR, "unexpected HeapTupleSatisfiesVacuum result"); + isdead = false; /* keep compiler quiet */ + break; + } - LockBuffer(buf, BUFFER_LOCK_UNLOCK); + LockBuffer(buf, BUFFER_LOCK_UNLOCK); - if (isdead) - { - *tups_vacuumed += 1; - /* heap rewrite module still needs to see it... */ - if (rewrite_heap_dead_tuple(rwstate, tuple)) + if (isdead) { - /* A previous recently-dead tuple is now known dead */ *tups_vacuumed += 1; - *tups_recently_dead -= 1; + /* heap rewrite module still needs to see it... */ + if (rewrite_heap_dead_tuple(rwstate, tuple)) + { + /* A previous recently-dead tuple is now known dead */ + *tups_vacuumed += 1; + *tups_recently_dead -= 1; + } + + continue; } - continue; } *num_tuples += 1; @@ -919,7 +942,7 @@ heapam_relation_copy_for_cluster(Relation OldHeap, Relation NewHeap, { const int ct_index[] = { PROGRESS_REPACK_HEAP_TUPLES_SCANNED, - PROGRESS_REPACK_HEAP_TUPLES_WRITTEN + PROGRESS_REPACK_HEAP_TUPLES_INSERTED }; int64 ct_val[2]; @@ -934,6 +957,31 @@ heapam_relation_copy_for_cluster(Relation OldHeap, Relation NewHeap, ct_val[1] = *num_tuples; pgstat_progress_update_multi_param(2, ct_index, ct_val); } + + /* + * Process the WAL produced by the load, as well as by other + * transactions, so that the replication slot can advance and WAL does + * not pile up. Use wal_segment_size as a threshold so that we do not + * introduce the decoding overhead too often. + * + * Of course, we must not apply the changes until the initial load has + * completed. + * + * Note that our insertions into the new table should not be decoded + * as we (intentionally) do not write the logical decoding specific + * information to WAL. + */ + if (concurrent) + { + XLogRecPtr end_of_wal; + + end_of_wal = GetFlushRecPtr(NULL); + if ((end_of_wal - end_of_wal_prev) > wal_segment_size) + { + repack_decode_concurrent_changes(decoding_ctx, end_of_wal); + end_of_wal_prev = end_of_wal; + } + } } if (indexScan != NULL) @@ -977,15 +1025,32 @@ heapam_relation_copy_for_cluster(Relation OldHeap, Relation NewHeap, values, isnull, rwstate); /* Report n_tuples */ - pgstat_progress_update_param(PROGRESS_REPACK_HEAP_TUPLES_WRITTEN, + pgstat_progress_update_param(PROGRESS_REPACK_HEAP_TUPLES_INSERTED, n_tuples); + + /* + * Try to keep the amount of not-yet-decoded WAL small, like + * above. + */ + if (concurrent) + { + XLogRecPtr end_of_wal; + + end_of_wal = GetFlushRecPtr(NULL); + if ((end_of_wal - end_of_wal_prev) > wal_segment_size) + { + repack_decode_concurrent_changes(decoding_ctx, end_of_wal); + end_of_wal_prev = end_of_wal; + } + } } tuplesort_end(tuplesort); } /* Write out any remaining tuples, and fsync if needed */ - end_heap_rewrite(rwstate); + if (rwstate) + end_heap_rewrite(rwstate); /* Clean up */ pfree(values); @@ -2376,6 +2441,10 @@ heapam_scan_sample_next_tuple(TableScanDesc scan, SampleScanState *scanstate, * SET WITHOUT OIDS. * * So, we must reconstruct the tuple from component Datums. + * + * If rwstate=NULL, use simple_heap_insert() instead of rewriting - in that + * case we still need to deform/form the tuple. TODO Shouldn't we rename the + * function, as might not do any rewrite? */ static void reform_and_rewrite_tuple(HeapTuple tuple, @@ -2398,8 +2467,28 @@ reform_and_rewrite_tuple(HeapTuple tuple, copiedTuple = heap_form_tuple(newTupDesc, values, isnull); - /* The heap rewrite module does the rest */ - rewrite_heap_tuple(rwstate, tuple, copiedTuple); + if (rwstate) + /* The heap rewrite module does the rest */ + rewrite_heap_tuple(rwstate, tuple, copiedTuple); + else + { + /* + * Insert tuple when processing REPACK CONCURRENTLY. + * + * rewriteheap.c is not used in the CONCURRENTLY case because it'd be + * difficult to do the same in the catch-up phase (as the logical + * decoding does not provide us with sufficient visibility + * information). Thus we must use heap_insert() both during the + * catch-up and here. + * + * The following is like simple_heap_insert() except that we pass the + * flag to skip logical decoding: as soon as REPACK CONCURRENTLY swaps + * the relation files, it drops this relation, so no logical + * replication subscription should need the data. + */ + heap_insert(NewHeap, copiedTuple, GetCurrentCommandId(true), + HEAP_INSERT_NO_LOGICAL, NULL); + } heap_freetuple(copiedTuple); } diff --git a/src/backend/access/heap/rewriteheap.c b/src/backend/access/heap/rewriteheap.c index 66ab48f0fe0..ee83a0fc91d 100644 --- a/src/backend/access/heap/rewriteheap.c +++ b/src/backend/access/heap/rewriteheap.c @@ -617,9 +617,9 @@ raw_heap_insert(RewriteState state, HeapTuple tup) int options = HEAP_INSERT_SKIP_FSM; /* - * While rewriting the heap for VACUUM FULL / CLUSTER, make sure data - * for the TOAST table are not logically decoded. The main heap is - * WAL-logged as XLOG FPI records, which are not logically decoded. + * While rewriting the heap for REPACK, make sure data for the TOAST + * table are not logically decoded. The main heap is WAL-logged as + * XLOG FPI records, which are not logically decoded. */ options |= HEAP_INSERT_NO_LOGICAL; diff --git a/src/backend/access/transam/xact.c b/src/backend/access/transam/xact.c index 092e197eba3..edf467a9125 100644 --- a/src/backend/access/transam/xact.c +++ b/src/backend/access/transam/xact.c @@ -216,6 +216,7 @@ typedef struct TransactionStateData bool parallelChildXact; /* is any parent transaction parallel? */ bool chain; /* start a new block after this one */ bool topXidLogged; /* for a subxact: is top-level XID logged? */ + bool internal; /* for a subxact: launched internally? */ struct TransactionStateData *parent; /* back link to parent */ } TransactionStateData; @@ -4741,6 +4742,7 @@ BeginInternalSubTransaction(const char *name) /* Normal subtransaction start */ PushTransaction(); s = CurrentTransactionState; /* changed by push */ + s->internal = true; /* * Savepoint names, like the TransactionState block itself, live @@ -5257,7 +5259,13 @@ AbortSubTransaction(void) LWLockReleaseAll(); pgstat_report_wait_end(); - pgstat_progress_end_command(); + + /* + * Internal subtransacion might be used by an user command, in which case + * the command outlives the subtransaction. + */ + if (!s->internal) + pgstat_progress_end_command(); pgaio_error_cleanup(); @@ -5474,6 +5482,7 @@ PushTransaction(void) s->parallelModeLevel = 0; s->parallelChildXact = (p->parallelModeLevel != 0 || p->parallelChildXact); s->topXidLogged = false; + s->internal = false; CurrentTransactionState = s; diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/system_views.sql index 024d219016d..941cd46ef67 100644 --- a/src/backend/catalog/system_views.sql +++ b/src/backend/catalog/system_views.sql @@ -1287,16 +1287,19 @@ CREATE VIEW pg_stat_progress_repack AS WHEN 2 THEN 'index scanning heap' WHEN 3 THEN 'sorting tuples' WHEN 4 THEN 'writing new heap' - WHEN 5 THEN 'swapping relation files' - WHEN 6 THEN 'rebuilding index' - WHEN 7 THEN 'performing final cleanup' + WHEN 5 THEN 'catch-up' + WHEN 6 THEN 'swapping relation files' + WHEN 7 THEN 'rebuilding index' + WHEN 8 THEN 'performing final cleanup' END AS phase, CAST(S.param3 AS oid) AS repack_index_relid, S.param4 AS heap_tuples_scanned, - S.param5 AS heap_tuples_written, - S.param6 AS heap_blks_total, - S.param7 AS heap_blks_scanned, - S.param8 AS index_rebuild_count + S.param5 AS heap_tuples_inserted, + S.param6 AS heap_tuples_updated, + S.param7 AS heap_tuples_deleted, + S.param8 AS heap_blks_total, + S.param9 AS heap_blks_scanned, + S.param10 AS index_rebuild_count FROM pg_stat_get_progress_info('REPACK') AS S LEFT JOIN pg_database D ON S.datid = D.oid; @@ -1314,7 +1317,7 @@ CREATE VIEW pg_stat_progress_cluster AS phase, repack_index_relid AS cluster_index_relid, heap_tuples_scanned, - heap_tuples_written, + heap_tuples_inserted + heap_tuples_updated AS heap_tuples_written, heap_blks_total, heap_blks_scanned, index_rebuild_count diff --git a/src/backend/commands/cluster.c b/src/backend/commands/cluster.c index ba3c076ea7d..260d7cc07e0 100644 --- a/src/backend/commands/cluster.c +++ b/src/backend/commands/cluster.c @@ -26,6 +26,10 @@ #include "access/toast_internals.h" #include "access/transam.h" #include "access/xact.h" +#include "access/xlog.h" +#include "access/xlog_internal.h" +#include "access/xloginsert.h" +#include "access/xlogutils.h" #include "catalog/catalog.h" #include "catalog/dependency.h" #include "catalog/heap.h" @@ -33,6 +37,7 @@ #include "catalog/namespace.h" #include "catalog/objectaccess.h" #include "catalog/pg_am.h" +#include "catalog/pg_control.h" #include "catalog/pg_inherits.h" #include "catalog/toasting.h" #include "commands/cluster.h" @@ -40,15 +45,21 @@ #include "commands/progress.h" #include "commands/tablecmds.h" #include "commands/vacuum.h" +#include "executor/executor.h" #include "miscadmin.h" #include "optimizer/optimizer.h" #include "pgstat.h" +#include "replication/decode.h" +#include "replication/logical.h" +#include "replication/snapbuild.h" #include "storage/bufmgr.h" +#include "storage/ipc.h" #include "storage/lmgr.h" #include "storage/predicate.h" #include "utils/acl.h" #include "utils/fmgroids.h" #include "utils/guc.h" +#include "utils/injection_point.h" #include "utils/inval.h" #include "utils/lsyscache.h" #include "utils/memutils.h" @@ -68,13 +79,68 @@ typedef struct Oid indexOid; } RelToCluster; +/* + * The following definitions are used for concurrent processing. + */ + +/* + * The locators are used to avoid logical decoding of data that we do not need + * for our table. + */ +RelFileLocator repacked_rel_locator = {.relNumber = InvalidOid}; +RelFileLocator repacked_rel_toast_locator = {.relNumber = InvalidOid}; + +/* + * Everything we need to call ExecInsertIndexTuples(). + */ +typedef struct IndexInsertState +{ + ResultRelInfo *rri; + EState *estate; +} IndexInsertState; + +/* The WAL segment being decoded. */ +static XLogSegNo repack_current_segment = 0; + +/* + * Information needed to apply concurrent data changes. + */ +typedef struct ChangeDest +{ + /* The relation the changes are applied to. */ + Relation rel; + + /* + * If valid, set rel->rd_toastoid to this for the time the changes are + * being applied. + */ + Oid toastrelid; + + /* + * The following is needed to find the existing tuple if the change is + * UPDATE or DELETE. 'ident_key' should have all the fields except for + * 'sk_argument' initialized. + */ + Relation ident_index; + ScanKey ident_key; + int ident_key_nentries; + + /* Needed to update indexes of rel_dst. */ + IndexInsertState *iistate; +} ChangeDest; + static bool cluster_rel_recheck(RepackCommand cmd, Relation OldHeap, - Oid indexOid, Oid userid, int options); -static void rebuild_relation(RepackCommand cmd, - Relation OldHeap, Relation index, bool verbose); + Oid indexOid, Oid userid, LOCKMODE lmode, + int options); +static void check_repack_concurrently_requirements(Relation rel); +static void rebuild_relation(RepackCommand cmd, Relation OldHeap, Relation index, + bool verbose, bool concurrently); static void copy_table_data(Relation NewHeap, Relation OldHeap, Relation OldIndex, - bool verbose, bool *pSwapToastByContent, - TransactionId *pFreezeXid, MultiXactId *pCutoffMulti); + Snapshot snapshot, LogicalDecodingContext *decoding_ctx, + bool verbose, + bool *pSwapToastByContent, + TransactionId *pFreezeXid, + MultiXactId *pCutoffMulti); static List *get_tables_to_repack(RepackCommand cmd, bool usingindex, MemoryContext permcxt); static List *get_tables_to_repack_partitioned(RepackCommand cmd, @@ -82,13 +148,54 @@ static List *get_tables_to_repack_partitioned(RepackCommand cmd, MemoryContext permcxt); static bool cluster_is_permitted_for_relation(RepackCommand cmd, Oid relid, Oid userid); + +static void begin_concurrent_repack(Relation rel); +static void end_concurrent_repack(void); +static LogicalDecodingContext *setup_logical_decoding(Oid relid); +static HeapTuple get_changed_tuple(char *change); +static void apply_concurrent_changes(RepackDecodingState *dstate, + ChangeDest *dest); +static void apply_concurrent_insert(Relation rel, ConcurrentChange *change, + HeapTuple tup, IndexInsertState *iistate, + TupleTableSlot *index_slot); +static void apply_concurrent_update(Relation rel, HeapTuple tup, + HeapTuple tup_target, + ConcurrentChange *change, + IndexInsertState *iistate, + TupleTableSlot *index_slot); +static void apply_concurrent_delete(Relation rel, HeapTuple tup_target, + ConcurrentChange *change); +static HeapTuple find_target_tuple(Relation rel, ChangeDest *dest, + HeapTuple tup_key, + TupleTableSlot *ident_slot); +static void process_concurrent_changes(LogicalDecodingContext *decoding_ctx, + XLogRecPtr end_of_wal, + ChangeDest *dest); +static IndexInsertState *get_index_insert_state(Relation relation, + Oid ident_index_id, + Relation *ident_index_p); +static ScanKey build_identity_key(Oid ident_idx_oid, Relation rel_src, + int *nentries); +static void free_index_insert_state(IndexInsertState *iistate); +static void cleanup_logical_decoding(LogicalDecodingContext *ctx); +static void rebuild_relation_finish_concurrent(Relation NewHeap, Relation OldHeap, + Relation cl_index, + LogicalDecodingContext *decoding_ctx, + bool swap_toast_by_content, + TransactionId frozenXid, + MultiXactId cutoffMulti); +static List *build_new_indexes(Relation NewHeap, Relation OldHeap, List *OldIndexes); static Relation process_single_relation(RepackStmt *stmt, + LOCKMODE lockmode, + bool isTopLevel, ClusterParams *params); static Oid determine_clustered_index(Relation rel, bool usingindex, const char *indexname); static const char *RepackCommandAsString(RepackCommand cmd); +#define REPL_PLUGIN_NAME "pgoutput_repack" + /* * The repack code allows for processing multiple tables at once. Because * of this, we cannot just run everything on a single transaction, or we @@ -118,6 +225,7 @@ ExecRepack(ParseState *pstate, RepackStmt *stmt, bool isTopLevel) ClusterParams params = {0}; Relation rel = NULL; MemoryContext repack_context; + LOCKMODE lockmode; List *rtcs; /* Parse option list */ @@ -128,6 +236,16 @@ ExecRepack(ParseState *pstate, RepackStmt *stmt, bool isTopLevel) else if (strcmp(opt->defname, "analyze") == 0 || strcmp(opt->defname, "analyse") == 0) params.options |= defGetBoolean(opt) ? CLUOPT_ANALYZE : 0; + else if (strcmp(opt->defname, "concurrently") == 0 && + defGetBoolean(opt)) + { + if (stmt->command != REPACK_COMMAND_REPACK) + ereport(ERROR, + errcode(ERRCODE_FEATURE_NOT_SUPPORTED), + errmsg("CONCURRENTLY option not supported for %s", + RepackCommandAsString(stmt->command))); + params.options |= CLUOPT_CONCURRENT; + } else ereport(ERROR, errcode(ERRCODE_SYNTAX_ERROR), @@ -137,13 +255,25 @@ ExecRepack(ParseState *pstate, RepackStmt *stmt, bool isTopLevel) parser_errposition(pstate, opt->location)); } + /* + * Determine the lock mode expected by cluster_rel(). + * + * In the exclusive case, we obtain AccessExclusiveLock right away to + * avoid lock-upgrade hazard in the single-transaction case. In the + * CONCURRENTLY case, the AccessExclusiveLock will only be used at the end + * of processing, supposedly for very short time. Until then, we'll have + * to unlock the relation temporarily, so there's no lock-upgrade hazard. + */ + lockmode = (params.options & CLUOPT_CONCURRENT) == 0 ? + AccessExclusiveLock : ShareUpdateExclusiveLock; + /* * If a single relation is specified, process it and we're done ... unless * the relation is a partitioned table, in which case we fall through. */ if (stmt->relation != NULL) { - rel = process_single_relation(stmt, ¶ms); + rel = process_single_relation(stmt, lockmode, isTopLevel, ¶ms); if (rel == NULL) return; /* all done */ } @@ -158,10 +288,29 @@ ExecRepack(ParseState *pstate, RepackStmt *stmt, bool isTopLevel) errmsg("cannot %s multiple tables", "REPACK (ANALYZE)")); /* - * By here, we know we are in a multi-table situation. In order to avoid - * holding locks for too long, we want to process each table in its own - * transaction. This forces us to disallow running inside a user - * transaction block. + * By here, we know we are in a multi-table situation. + * + * Concurrent processing is currently considered rather special (e.g. in + * terms of resources consumed) so it is not performed in bulk. + */ + if (params.options & CLUOPT_CONCURRENT) + { + if (rel != NULL) + { + Assert(rel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE); + ereport(ERROR, + errmsg("REPACK CONCURRENTLY not supported for partitioned tables"), + errhint("Consider running the command for individual partitions.")); + } + else + ereport(ERROR, + errmsg("REPACK CONCURRENTLY requires explicit table name")); + } + + /* + * In order to avoid holding locks for too long, we want to process each + * table in its own transaction. This forces us to disallow running + * inside a user transaction block. */ PreventInTransactionBlock(isTopLevel, RepackCommandAsString(stmt->command)); @@ -244,7 +393,7 @@ ExecRepack(ParseState *pstate, RepackStmt *stmt, bool isTopLevel) * Open the target table, coping with the case where it has been * dropped. */ - rel = try_table_open(rtc->tableOid, AccessExclusiveLock); + rel = try_table_open(rtc->tableOid, lockmode); if (rel == NULL) { CommitTransactionCommand(); @@ -255,7 +404,7 @@ ExecRepack(ParseState *pstate, RepackStmt *stmt, bool isTopLevel) PushActiveSnapshot(GetTransactionSnapshot()); /* Process this table */ - cluster_rel(stmt->command, rel, rtc->indexOid, ¶ms); + cluster_rel(stmt->command, rel, rtc->indexOid, ¶ms, isTopLevel); /* cluster_rel closes the relation, but keeps lock */ PopActiveSnapshot(); @@ -284,22 +433,53 @@ ExecRepack(ParseState *pstate, RepackStmt *stmt, bool isTopLevel) * If indexOid is InvalidOid, the table will be rewritten in physical order * instead of index order. * + * Note that, in the concurrent case, the function releases the lock at some + * point, in order to get AccessExclusiveLock for the final steps (i.e. to + * swap the relation files). To make things simpler, the caller should expect + * OldHeap to be closed on return, regardless CLUOPT_CONCURRENT. (The + * AccessExclusiveLock is kept till the end of the transaction.) + * * 'cmd' indicates which command is being executed, to be used for error * messages. */ void cluster_rel(RepackCommand cmd, Relation OldHeap, Oid indexOid, - ClusterParams *params) + ClusterParams *params, bool isTopLevel) { Oid tableOid = RelationGetRelid(OldHeap); + Relation index; + LOCKMODE lmode; Oid save_userid; int save_sec_context; int save_nestlevel; bool verbose = ((params->options & CLUOPT_VERBOSE) != 0); bool recheck = ((params->options & CLUOPT_RECHECK) != 0); - Relation index; + bool concurrent = ((params->options & CLUOPT_CONCURRENT) != 0); + + /* + * The lock mode is AccessExclusiveLock for normal processing and + * ShareUpdateExclusiveLock for concurrent processing (so that SELECT, + * INSERT, UPDATE and DELETE commands work, but cluster_rel() cannot be + * called concurrently for the same relation). + */ + lmode = !concurrent ? AccessExclusiveLock : ShareUpdateExclusiveLock; + + /* There are specific requirements on concurrent processing. */ + if (concurrent) + { + /* + * Make sure we have no XID assigned, otherwise call of + * setup_logical_decoding() can cause a deadlock. + * + * The existence of transaction block actually does not imply that XID + * was already assigned, but it very likely is. We might want to check + * the result of GetCurrentTransactionIdIfAny() instead, but that + * would be less clear from user's perspective. + */ + PreventInTransactionBlock(isTopLevel, "REPACK (CONCURRENTLY)"); - Assert(CheckRelationLockedByMe(OldHeap, AccessExclusiveLock, false)); + check_repack_concurrently_requirements(OldHeap); + } /* Check for user-requested abort. */ CHECK_FOR_INTERRUPTS(); @@ -325,10 +505,13 @@ cluster_rel(RepackCommand cmd, Relation OldHeap, Oid indexOid, * If this is a single-transaction CLUSTER, we can skip these tests. We * *must* skip the one on indisclustered since it would reject an attempt * to cluster a not-previously-clustered index. + * + * XXX move [some of] these comments to where the RECHECK flag is + * determined? */ if (recheck && !cluster_rel_recheck(cmd, OldHeap, indexOid, save_userid, - params->options)) + lmode, params->options)) goto out; /* @@ -343,6 +526,12 @@ cluster_rel(RepackCommand cmd, Relation OldHeap, Oid indexOid, errmsg("cannot run %s on a shared catalog", RepackCommandAsString(cmd))); + /* + * The CONCURRENTLY case should have been rejected earlier because it does + * not support system catalogs. + */ + Assert(!(OldHeap->rd_rel->relisshared && concurrent)); + /* * Don't process temp tables of other backends ... their local buffer * manager is not going to cope. @@ -363,7 +552,7 @@ cluster_rel(RepackCommand cmd, Relation OldHeap, Oid indexOid, if (OidIsValid(indexOid)) { /* verify the index is good and lock it */ - check_index_is_clusterable(OldHeap, indexOid, AccessExclusiveLock); + check_index_is_clusterable(OldHeap, indexOid, lmode); /* also open it */ index = index_open(indexOid, NoLock); } @@ -398,7 +587,9 @@ cluster_rel(RepackCommand cmd, Relation OldHeap, Oid indexOid, if (OldHeap->rd_rel->relkind == RELKIND_MATVIEW && !RelationIsPopulated(OldHeap)) { - relation_close(OldHeap, AccessExclusiveLock); + if (index) + index_close(index, lmode); + relation_close(OldHeap, lmode); goto out; } @@ -411,11 +602,35 @@ cluster_rel(RepackCommand cmd, Relation OldHeap, Oid indexOid, * invalid, because we move tuples around. Promote them to relation * locks. Predicate locks on indexes will be promoted when they are * reindexed. + * + * During concurrent processing, the heap as well as its indexes stay in + * operation, so we postpone this step until they are locked using + * AccessExclusiveLock near the end of the processing. */ - TransferPredicateLocksToHeapRelation(OldHeap); + if (!concurrent) + TransferPredicateLocksToHeapRelation(OldHeap); /* rebuild_relation does all the dirty work */ - rebuild_relation(cmd, OldHeap, index, verbose); + PG_TRY(); + { + /* + * For concurrent processing, make sure that our logical decoding + * ignores data changes of other tables than the one we are + * processing. + */ + if (concurrent) + begin_concurrent_repack(OldHeap); + + rebuild_relation(cmd, OldHeap, index, /* save_userid, */ + verbose, concurrent); + } + PG_FINALLY(); + { + if (concurrent) + end_concurrent_repack(); + } + PG_END_TRY(); + /* rebuild_relation closes OldHeap, and index if valid */ out: @@ -434,14 +649,14 @@ out: */ static bool cluster_rel_recheck(RepackCommand cmd, Relation OldHeap, Oid indexOid, - Oid userid, int options) + Oid userid, LOCKMODE lmode, int options) { Oid tableOid = RelationGetRelid(OldHeap); /* Check that the user still has privileges for the relation */ if (!cluster_is_permitted_for_relation(cmd, tableOid, userid)) { - relation_close(OldHeap, AccessExclusiveLock); + relation_close(OldHeap, lmode); return false; } @@ -455,7 +670,7 @@ cluster_rel_recheck(RepackCommand cmd, Relation OldHeap, Oid indexOid, */ if (RELATION_IS_OTHER_TEMP(OldHeap)) { - relation_close(OldHeap, AccessExclusiveLock); + relation_close(OldHeap, lmode); return false; } @@ -466,7 +681,7 @@ cluster_rel_recheck(RepackCommand cmd, Relation OldHeap, Oid indexOid, */ if (!SearchSysCacheExists1(RELOID, ObjectIdGetDatum(indexOid))) { - relation_close(OldHeap, AccessExclusiveLock); + relation_close(OldHeap, lmode); return false; } @@ -477,7 +692,7 @@ cluster_rel_recheck(RepackCommand cmd, Relation OldHeap, Oid indexOid, if ((options & CLUOPT_RECHECK_ISCLUSTERED) != 0 && !get_index_isclustered(indexOid)) { - relation_close(OldHeap, AccessExclusiveLock); + relation_close(OldHeap, lmode); return false; } } @@ -489,7 +704,7 @@ cluster_rel_recheck(RepackCommand cmd, Relation OldHeap, Oid indexOid, * Verify that the specified heap and index are valid to cluster on * * Side effect: obtains lock on the index. The caller may - * in some cases already have AccessExclusiveLock on the table, but + * in some cases already have a lock of the same strength on the table, but * not in all cases so we can't rely on the table-level lock for * protection here. */ @@ -618,19 +833,88 @@ mark_index_clustered(Relation rel, Oid indexOid, bool is_internal) table_close(pg_index, RowExclusiveLock); } +/* + * Check if the CONCURRENTLY option is legal for the relation. + */ +static void +check_repack_concurrently_requirements(Relation rel) +{ + char relpersistence, + replident; + Oid ident_idx; + + /* Data changes in system relations are not logically decoded. */ + if (IsCatalogRelation(rel)) + ereport(ERROR, + (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), + errmsg("cannot repack relation \"%s\"", + RelationGetRelationName(rel)), + errhint("REPACK CONCURRENTLY is not supported for catalog relations."))); + + /* + * reorderbuffer.c does not seem to handle processing of TOAST relation + * alone. + */ + if (IsToastRelation(rel)) + ereport(ERROR, + (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), + errmsg("cannot repack relation \"%s\"", + RelationGetRelationName(rel)), + errhint("REPACK CONCURRENTLY is not supported for TOAST relations, unless the main relation is repacked too."))); + + relpersistence = rel->rd_rel->relpersistence; + if (relpersistence != RELPERSISTENCE_PERMANENT) + ereport(ERROR, + (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), + errmsg("cannot repack relation \"%s\"", + RelationGetRelationName(rel)), + errhint("REPACK CONCURRENTLY is only allowed for permanent relations."))); + + /* With NOTHING, WAL does not contain the old tuple. */ + replident = rel->rd_rel->relreplident; + if (replident == REPLICA_IDENTITY_NOTHING) + ereport(ERROR, + (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), + errmsg("cannot repack relation \"%s\"", + RelationGetRelationName(rel)), + errhint("Relation \"%s\" has insufficient replication identity.", + RelationGetRelationName(rel)))); + + /* + * Identity index is not set if the replica identity is FULL, but PK might + * exist in such a case. + */ + ident_idx = RelationGetReplicaIndex(rel); + if (!OidIsValid(ident_idx) && OidIsValid(rel->rd_pkindex)) + ident_idx = rel->rd_pkindex; + if (!OidIsValid(ident_idx)) + ereport(ERROR, + (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), + errmsg("cannot process relation \"%s\"", + RelationGetRelationName(rel)), + (errhint("Relation \"%s\" has no identity index.", + RelationGetRelationName(rel))))); +} + + /* * rebuild_relation: rebuild an existing relation in index or physical order * - * OldHeap: table to rebuild. + * OldHeap: table to rebuild. See cluster_rel() for comments on the required + * lock strength. + * * index: index to cluster by, or NULL to rewrite in physical order. * - * On entry, heap and index (if one is given) must be open, and - * AccessExclusiveLock held on them. - * On exit, they are closed, but locks on them are not released. + * On entry, heap and index (if one is given) must be open, and the + * appropriate lock held on them -- AccessExclusiveLock for exclusive + * processing and ShareUpdateExclusiveLock for concurrent processing. + * + * On exit, they are closed, but still locked with AccessExclusiveLock. + * (The function handles the lock upgrade if 'concurrent' is true.) */ static void -rebuild_relation(RepackCommand cmd, - Relation OldHeap, Relation index, bool verbose) +rebuild_relation(RepackCommand cmd, Relation OldHeap, Relation index, + bool verbose, bool concurrent) { Oid tableOid = RelationGetRelid(OldHeap); Oid accessMethod = OldHeap->rd_rel->relam; @@ -638,13 +922,38 @@ rebuild_relation(RepackCommand cmd, Oid OIDNewHeap; Relation NewHeap; char relpersistence; - bool is_system_catalog; bool swap_toast_by_content; TransactionId frozenXid; MultiXactId cutoffMulti; + LogicalDecodingContext *decoding_ctx = NULL; + Snapshot snapshot = NULL; +#if USE_ASSERT_CHECKING + LOCKMODE lmode; + + lmode = concurrent ? ShareUpdateExclusiveLock : AccessExclusiveLock; + + Assert(CheckRelationLockedByMe(OldHeap, lmode, false)); + Assert(index == NULL || CheckRelationLockedByMe(index, lmode, false)); +#endif - Assert(CheckRelationLockedByMe(OldHeap, AccessExclusiveLock, false) && - (index == NULL || CheckRelationLockedByMe(index, AccessExclusiveLock, false))); + if (concurrent) + { + /* + * Prepare to capture the concurrent data changes. + * + * Note that this call waits for all transactions with XID already + * assigned to finish. If some of those transactions is waiting for a + * lock conflicting with ShareUpdateExclusiveLock on our table (e.g. + * it runs CREATE INDEX), we can end up in a deadlock. Not sure this + * risk is worth unlocking/locking the table (and its clustering + * index) and checking again if its still eligible for REPACK + * CONCURRENTLY. + */ + decoding_ctx = setup_logical_decoding(tableOid); + + snapshot = SnapBuildInitialSnapshotForRepack(decoding_ctx->snapshot_builder); + PushActiveSnapshot(snapshot); + } /* for CLUSTER or REPACK USING INDEX, mark the index as the one to use */ if (index != NULL) @@ -652,7 +961,6 @@ rebuild_relation(RepackCommand cmd, /* Remember info about rel before closing OldHeap */ relpersistence = OldHeap->rd_rel->relpersistence; - is_system_catalog = IsSystemRelation(OldHeap); /* * Create the transient table that will receive the re-ordered data. @@ -668,30 +976,65 @@ rebuild_relation(RepackCommand cmd, NewHeap = table_open(OIDNewHeap, NoLock); /* Copy the heap data into the new table in the desired order */ - copy_table_data(NewHeap, OldHeap, index, verbose, + copy_table_data(NewHeap, OldHeap, index, snapshot, decoding_ctx, verbose, &swap_toast_by_content, &frozenXid, &cutoffMulti); + /* The historic snapshot won't be needed anymore. */ + if (snapshot) + PopActiveSnapshot(); - /* Close relcache entries, but keep lock until transaction commit */ - table_close(OldHeap, NoLock); - if (index) - index_close(index, NoLock); + if (concurrent) + { + /* + * Push a snapshot that we will use to find old versions of rows when + * processing concurrent UPDATE and DELETE commands. (That snapshot + * should also be used by index expressions.) + */ + PushActiveSnapshot(GetTransactionSnapshot()); - /* - * Close the new relation so it can be dropped as soon as the storage is - * swapped. The relation is not visible to others, so no need to unlock it - * explicitly. - */ - table_close(NewHeap, NoLock); + /* + * Make sure we can find the tuples just inserted when applying DML + * commands on top of those. + */ + CommandCounterIncrement(); + UpdateActiveSnapshotCommandId(); - /* - * Swap the physical files of the target and transient tables, then - * rebuild the target's indexes and throw away the transient table. - */ - finish_heap_swap(tableOid, OIDNewHeap, is_system_catalog, - swap_toast_by_content, false, true, - frozenXid, cutoffMulti, - relpersistence); + rebuild_relation_finish_concurrent(NewHeap, OldHeap, index, + decoding_ctx, swap_toast_by_content, + frozenXid, cutoffMulti); + PopActiveSnapshot(); + + pgstat_progress_update_param(PROGRESS_REPACK_PHASE, + PROGRESS_REPACK_PHASE_FINAL_CLEANUP); + + /* Done with decoding. */ + cleanup_logical_decoding(decoding_ctx); + } + else + { + bool is_system_catalog = IsSystemRelation(OldHeap); + + /* Close relcache entries, but keep lock until transaction commit */ + table_close(OldHeap, NoLock); + if (index) + index_close(index, NoLock); + + /* + * Close the new relation so it can be dropped as soon as the storage + * is swapped. The relation is not visible to others, so no need to + * unlock it explicitly. + */ + table_close(NewHeap, NoLock); + + /* + * Swap the physical files of the target and transient tables, then + * rebuild the target's indexes and throw away the transient table. + */ + finish_heap_swap(tableOid, OIDNewHeap, is_system_catalog, + swap_toast_by_content, false, true, true, + frozenXid, cutoffMulti, + relpersistence); + } } @@ -826,15 +1169,19 @@ make_new_heap(Oid OIDOldHeap, Oid NewTableSpace, Oid NewAccessMethod, /* * Do the physical copying of table data. * + * 'snapshot' and 'decoding_ctx': see table_relation_copy_for_cluster(). Pass + * iff concurrent processing is required. + * * There are three output parameters: * *pSwapToastByContent is set true if toast tables must be swapped by content. * *pFreezeXid receives the TransactionId used as freeze cutoff point. * *pCutoffMulti receives the MultiXactId used as a cutoff point. */ static void -copy_table_data(Relation NewHeap, Relation OldHeap, Relation OldIndex, bool verbose, - bool *pSwapToastByContent, TransactionId *pFreezeXid, - MultiXactId *pCutoffMulti) +copy_table_data(Relation NewHeap, Relation OldHeap, Relation OldIndex, + Snapshot snapshot, LogicalDecodingContext *decoding_ctx, + bool verbose, bool *pSwapToastByContent, + TransactionId *pFreezeXid, MultiXactId *pCutoffMulti) { Relation relRelation; HeapTuple reltup; @@ -852,6 +1199,8 @@ copy_table_data(Relation NewHeap, Relation OldHeap, Relation OldIndex, bool verb PGRUsage ru0; char *nspname; + bool concurrent = snapshot != NULL; + pg_rusage_init(&ru0); /* Store a copy of the namespace name for logging purposes */ @@ -954,8 +1303,48 @@ copy_table_data(Relation NewHeap, Relation OldHeap, Relation OldIndex, bool verb * provided, else plain seqscan. */ if (OldIndex != NULL && OldIndex->rd_rel->relam == BTREE_AM_OID) + { + ResourceOwner oldowner = NULL; + ResourceOwner resowner = NULL; + + /* + * In the CONCURRENT case, use a dedicated resource owner so we don't + * leave any additional locks behind us that we cannot release easily. + */ + if (concurrent) + { + Assert(CheckRelationLockedByMe(OldHeap, ShareUpdateExclusiveLock, + false)); + Assert(CheckRelationLockedByMe(OldIndex, ShareUpdateExclusiveLock, + false)); + + resowner = ResourceOwnerCreate(CurrentResourceOwner, + "plan_cluster_use_sort"); + oldowner = CurrentResourceOwner; + CurrentResourceOwner = resowner; + } + use_sort = plan_cluster_use_sort(RelationGetRelid(OldHeap), RelationGetRelid(OldIndex)); + + if (concurrent) + { + CurrentResourceOwner = oldowner; + + /* + * We are primarily concerned about locks, but if the planner + * happened to allocate any other resources, we should release + * them too because we're going to delete the whole resowner. + */ + ResourceOwnerRelease(resowner, RESOURCE_RELEASE_BEFORE_LOCKS, + false, false); + ResourceOwnerRelease(resowner, RESOURCE_RELEASE_LOCKS, + false, false); + ResourceOwnerRelease(resowner, RESOURCE_RELEASE_AFTER_LOCKS, + false, false); + ResourceOwnerDelete(resowner); + } + } else use_sort = false; @@ -984,7 +1373,9 @@ copy_table_data(Relation NewHeap, Relation OldHeap, Relation OldIndex, bool verb * values (e.g. because the AM doesn't use freezing). */ table_relation_copy_for_cluster(OldHeap, NewHeap, OldIndex, use_sort, - cutoffs.OldestXmin, &cutoffs.FreezeLimit, + cutoffs.OldestXmin, snapshot, + decoding_ctx, + &cutoffs.FreezeLimit, &cutoffs.MultiXactCutoff, &num_tuples, &tups_vacuumed, &tups_recently_dead); @@ -993,7 +1384,11 @@ copy_table_data(Relation NewHeap, Relation OldHeap, Relation OldIndex, bool verb *pFreezeXid = cutoffs.FreezeLimit; *pCutoffMulti = cutoffs.MultiXactCutoff; - /* Reset rd_toastoid just to be tidy --- it shouldn't be looked at again */ + /* + * Reset rd_toastoid just to be tidy --- it shouldn't be looked at again. + * In the CONCURRENTLY case, we need to set it again before applying the + * concurrent changes. + */ NewHeap->rd_toastoid = InvalidOid; num_pages = RelationGetNumberOfBlocks(NewHeap); @@ -1451,14 +1846,13 @@ finish_heap_swap(Oid OIDOldHeap, Oid OIDNewHeap, bool swap_toast_by_content, bool check_constraints, bool is_internal, + bool reindex, TransactionId frozenXid, MultiXactId cutoffMulti, char newrelpersistence) { ObjectAddress object; Oid mapped_tables[4]; - int reindex_flags; - ReindexParams reindex_params = {0}; int i; /* Report that we are now swapping relation files */ @@ -1484,39 +1878,47 @@ finish_heap_swap(Oid OIDOldHeap, Oid OIDNewHeap, if (is_system_catalog) CacheInvalidateCatalog(OIDOldHeap); - /* - * Rebuild each index on the relation (but not the toast table, which is - * all-new at this point). It is important to do this before the DROP - * step because if we are processing a system catalog that will be used - * during DROP, we want to have its indexes available. There is no - * advantage to the other order anyway because this is all transactional, - * so no chance to reclaim disk space before commit. We do not need a - * final CommandCounterIncrement() because reindex_relation does it. - * - * Note: because index_build is called via reindex_relation, it will never - * set indcheckxmin true for the indexes. This is OK even though in some - * sense we are building new indexes rather than rebuilding existing ones, - * because the new heap won't contain any HOT chains at all, let alone - * broken ones, so it can't be necessary to set indcheckxmin. - */ - reindex_flags = REINDEX_REL_SUPPRESS_INDEX_USE; - if (check_constraints) - reindex_flags |= REINDEX_REL_CHECK_CONSTRAINTS; + if (reindex) + { + int reindex_flags; + ReindexParams reindex_params = {0}; - /* - * Ensure that the indexes have the same persistence as the parent - * relation. - */ - if (newrelpersistence == RELPERSISTENCE_UNLOGGED) - reindex_flags |= REINDEX_REL_FORCE_INDEXES_UNLOGGED; - else if (newrelpersistence == RELPERSISTENCE_PERMANENT) - reindex_flags |= REINDEX_REL_FORCE_INDEXES_PERMANENT; + /* + * Rebuild each index on the relation (but not the toast table, which + * is all-new at this point). It is important to do this before the + * DROP step because if we are processing a system catalog that will + * be used during DROP, we want to have its indexes available. There + * is no advantage to the other order anyway because this is all + * transactional, so no chance to reclaim disk space before commit. We + * do not need a final CommandCounterIncrement() because + * reindex_relation does it. + * + * Note: because index_build is called via reindex_relation, it will + * never set indcheckxmin true for the indexes. This is OK even + * though in some sense we are building new indexes rather than + * rebuilding existing ones, because the new heap won't contain any + * HOT chains at all, let alone broken ones, so it can't be necessary + * to set indcheckxmin. + */ + reindex_flags = REINDEX_REL_SUPPRESS_INDEX_USE; + if (check_constraints) + reindex_flags |= REINDEX_REL_CHECK_CONSTRAINTS; - /* Report that we are now reindexing relations */ - pgstat_progress_update_param(PROGRESS_REPACK_PHASE, - PROGRESS_REPACK_PHASE_REBUILD_INDEX); + /* + * Ensure that the indexes have the same persistence as the parent + * relation. + */ + if (newrelpersistence == RELPERSISTENCE_UNLOGGED) + reindex_flags |= REINDEX_REL_FORCE_INDEXES_UNLOGGED; + else if (newrelpersistence == RELPERSISTENCE_PERMANENT) + reindex_flags |= REINDEX_REL_FORCE_INDEXES_PERMANENT; - reindex_relation(NULL, OIDOldHeap, reindex_flags, &reindex_params); + /* Report that we are now reindexing relations */ + pgstat_progress_update_param(PROGRESS_REPACK_PHASE, + PROGRESS_REPACK_PHASE_REBUILD_INDEX); + + reindex_relation(NULL, OIDOldHeap, reindex_flags, &reindex_params); + } /* Report that we are now doing clean up */ pgstat_progress_update_param(PROGRESS_REPACK_PHASE, @@ -1859,7 +2261,8 @@ cluster_is_permitted_for_relation(RepackCommand cmd, Oid relid, Oid userid) * case, if an index name is given, it's up to the caller to resolve it. */ static Relation -process_single_relation(RepackStmt *stmt, ClusterParams *params) +process_single_relation(RepackStmt *stmt, LOCKMODE lockmode, bool isTopLevel, + ClusterParams *params) { Relation rel; Oid tableOid; @@ -1868,13 +2271,9 @@ process_single_relation(RepackStmt *stmt, ClusterParams *params) Assert(stmt->command == REPACK_COMMAND_CLUSTER || stmt->command == REPACK_COMMAND_REPACK); - /* - * Find, lock, and check permissions on the table. We obtain - * AccessExclusiveLock right away to avoid lock-upgrade hazard in the - * single-transaction case. - */ + /* Find, lock, and check permissions on the table. */ tableOid = RangeVarGetRelidExtended(stmt->relation->relation, - AccessExclusiveLock, + lockmode, 0, RangeVarCallbackMaintainsTable, NULL); @@ -1906,13 +2305,14 @@ process_single_relation(RepackStmt *stmt, ClusterParams *params) return rel; else { - Oid indexOid; + Oid indexOid = InvalidOid; indexOid = determine_clustered_index(rel, stmt->usingindex, stmt->indexname); if (OidIsValid(indexOid)) - check_index_is_clusterable(rel, indexOid, AccessExclusiveLock); - cluster_rel(stmt->command, rel, indexOid, params); + check_index_is_clusterable(rel, indexOid, lockmode); + + cluster_rel(stmt->command, rel, indexOid, params, isTopLevel); /* Do an analyze, if requested */ if (params->options & CLUOPT_ANALYZE) @@ -1995,3 +2395,1058 @@ RepackCommandAsString(RepackCommand cmd) } return "???"; } + + +/* + * Call this function before REPACK CONCURRENTLY starts to setup logical + * decoding. It makes sure that other users of the table put enough + * information into WAL. + * + * The point is that at various places we expect that the table we're + * processing is treated like a system catalog. For example, we need to be + * able to scan it using a "historic snapshot" anytime during the processing + * (as opposed to scanning only at the start point of the decoding, as logical + * replication does during initial table synchronization), in order to apply + * concurrent UPDATE / DELETE commands. + * + * Note that TOAST table needs no attention here as it's not scanned using + * historic snapshot. + */ +static void +begin_concurrent_repack(Relation rel) +{ + Oid toastrelid; + + /* + * Avoid logical decoding of other relations by this backend. The lock we + * have guarantees that the actual locator cannot be changed concurrently: + * TRUNCATE needs AccessExclusiveLock. + */ + Assert(CheckRelationLockedByMe(rel, ShareUpdateExclusiveLock, false)); + repacked_rel_locator = rel->rd_locator; + toastrelid = rel->rd_rel->reltoastrelid; + if (OidIsValid(toastrelid)) + { + Relation toastrel; + + /* Avoid logical decoding of other TOAST relations. */ + toastrel = table_open(toastrelid, AccessShareLock); + repacked_rel_toast_locator = toastrel->rd_locator; + table_close(toastrel, AccessShareLock); + } +} + +/* + * Call this when done with REPACK CONCURRENTLY. + */ +static void +end_concurrent_repack(void) +{ + /* + * Restore normal function of (future) logical decoding for this backend. + */ + repacked_rel_locator.relNumber = InvalidOid; + repacked_rel_toast_locator.relNumber = InvalidOid; +} + +/* + * This function is much like pg_create_logical_replication_slot() except that + * the new slot is neither released (if anyone else could read changes from + * our slot, we could miss changes other backends do while we copy the + * existing data into temporary table), nor persisted (it's easier to handle + * crash by restarting all the work from scratch). + */ +static LogicalDecodingContext * +setup_logical_decoding(Oid relid) +{ + Relation rel; + TupleDesc tupdesc; + LogicalDecodingContext *ctx; + RepackDecodingState *dstate = palloc0_object(RepackDecodingState); + + /* + * REPACK CONCURRENTLY is not allowed in a transaction block, so this + * should never fire. + */ + Assert(!TransactionIdIsValid(GetTopTransactionIdIfAny())); + + /* + * A single backend should not execute multiple REPACK commands at a time, + * so use PID to make the slot unique. + */ + snprintf(NameStr(dstate->slotname), NAMEDATALEN, "repack_%d", MyProcPid); + + /* + * Check if we can use logical decoding. + */ + CheckSlotPermissions(); + CheckLogicalDecodingRequirements(); + + /* RS_TEMPORARY so that the slot gets cleaned up on ERROR. */ + ReplicationSlotCreate(NameStr(dstate->slotname), true, RS_TEMPORARY, + false, false, false); + + /* + * Neither prepare_write nor do_write callback nor update_progress is + * useful for us. + */ + ctx = CreateInitDecodingContext(REPL_PLUGIN_NAME, + NIL, + true, + InvalidXLogRecPtr, + XL_ROUTINE(.page_read = read_local_xlog_page, + .segment_open = wal_segment_open, + .segment_close = wal_segment_close), + NULL, NULL, NULL); + + /* + * We don't have control on setting fast_forward, so at least check it. + */ + Assert(!ctx->fast_forward); + + DecodingContextFindStartpoint(ctx); + + /* Some WAL records should have been read. */ + Assert(ctx->reader->EndRecPtr != InvalidXLogRecPtr); + + XLByteToSeg(ctx->reader->EndRecPtr, repack_current_segment, + wal_segment_size); + + /* + * Setup structures to store decoded changes. + */ + dstate->relid = relid; + dstate->tstore = tuplestore_begin_heap(false, false, + maintenance_work_mem); + + /* Caller should already have the table locked. */ + rel = table_open(relid, NoLock); + tupdesc = CreateTupleDescCopy(RelationGetDescr(rel)); + dstate->tupdesc = tupdesc; + table_close(rel, NoLock); + + /* Initialize the descriptor to store the changes ... */ + dstate->tupdesc_change = CreateTemplateTupleDesc(1); + + TupleDescInitEntry(dstate->tupdesc_change, 1, NULL, BYTEAOID, -1, 0); + /* ... as well as the corresponding slot. */ + dstate->tsslot = MakeSingleTupleTableSlot(dstate->tupdesc_change, + &TTSOpsMinimalTuple); + + dstate->resowner = ResourceOwnerCreate(CurrentResourceOwner, + "logical decoding"); + + ctx->output_writer_private = dstate; + return ctx; +} + +/* + * Retrieve tuple from ConcurrentChange structure. + * + * The input data starts with the structure but it might not be appropriately + * aligned. + */ +static HeapTuple +get_changed_tuple(char *change) +{ + HeapTupleData tup_data; + HeapTuple result; + char *src; + + /* + * Ensure alignment before accessing the fields. (This is why we can't use + * heap_copytuple() instead of this function.) + */ + src = change + offsetof(ConcurrentChange, tup_data); + memcpy(&tup_data, src, sizeof(HeapTupleData)); + + result = (HeapTuple) palloc(HEAPTUPLESIZE + tup_data.t_len); + memcpy(result, &tup_data, sizeof(HeapTupleData)); + result->t_data = (HeapTupleHeader) ((char *) result + HEAPTUPLESIZE); + src = change + SizeOfConcurrentChange; + memcpy(result->t_data, src, result->t_len); + + return result; +} + +/* + * Decode logical changes from the WAL sequence up to end_of_wal. + */ +void +repack_decode_concurrent_changes(LogicalDecodingContext *ctx, + XLogRecPtr end_of_wal) +{ + RepackDecodingState *dstate; + ResourceOwner resowner_old; + + /* + * Invalidate the "present" cache before moving to "(recent) history". + */ + InvalidateSystemCaches(); + + dstate = (RepackDecodingState *) ctx->output_writer_private; + resowner_old = CurrentResourceOwner; + CurrentResourceOwner = dstate->resowner; + + PG_TRY(); + { + while (ctx->reader->EndRecPtr < end_of_wal) + { + XLogRecord *record; + XLogSegNo segno_new; + char *errm = NULL; + XLogRecPtr end_lsn; + + record = XLogReadRecord(ctx->reader, &errm); + if (errm) + elog(ERROR, "%s", errm); + + if (record != NULL) + LogicalDecodingProcessRecord(ctx, ctx->reader); + + /* + * If WAL segment boundary has been crossed, inform the decoding + * system that the catalog_xmin can advance. (We can confirm more + * often, but a filling a single WAL segment should not take much + * time.) + */ + end_lsn = ctx->reader->EndRecPtr; + XLByteToSeg(end_lsn, segno_new, wal_segment_size); + if (segno_new != repack_current_segment) + { + LogicalConfirmReceivedLocation(end_lsn); + elog(DEBUG1, "REPACK: confirmed receive location %X/%X", + (uint32) (end_lsn >> 32), (uint32) end_lsn); + repack_current_segment = segno_new; + } + + CHECK_FOR_INTERRUPTS(); + } + InvalidateSystemCaches(); + CurrentResourceOwner = resowner_old; + } + PG_CATCH(); + { + /* clear all timetravel entries */ + InvalidateSystemCaches(); + CurrentResourceOwner = resowner_old; + PG_RE_THROW(); + } + PG_END_TRY(); +} + +/* + * Apply changes that happened during the initial load. + * + * Scan key is passed by caller, so it does not have to be constructed + * multiple times. Key entries have all fields initialized, except for + * sk_argument. + */ +static void +apply_concurrent_changes(RepackDecodingState *dstate, ChangeDest *dest) +{ + Relation rel = dest->rel; + TupleTableSlot *index_slot, + *ident_slot; + HeapTuple tup_old = NULL; + + if (dstate->nchanges == 0) + return; + + /* TupleTableSlot is needed to pass the tuple to ExecInsertIndexTuples(). */ + index_slot = MakeSingleTupleTableSlot(dstate->tupdesc, &TTSOpsHeapTuple); + + /* A slot to fetch tuples from identity index. */ + ident_slot = table_slot_create(rel, NULL); + + while (tuplestore_gettupleslot(dstate->tstore, true, false, + dstate->tsslot)) + { + bool shouldFree; + HeapTuple tup_change, + tup, + tup_exist; + char *change_raw, + *src; + ConcurrentChange change; + bool isnull[1]; + Datum values[1]; + + CHECK_FOR_INTERRUPTS(); + + /* Get the change from the single-column tuple. */ + tup_change = ExecFetchSlotHeapTuple(dstate->tsslot, false, &shouldFree); + heap_deform_tuple(tup_change, dstate->tupdesc_change, values, isnull); + Assert(!isnull[0]); + + /* Make sure we access aligned data. */ + change_raw = (char *) DatumGetByteaP(values[0]); + src = (char *) VARDATA(change_raw); + memcpy(&change, src, SizeOfConcurrentChange); + + /* + * Extract the tuple from the change. The tuple is copied here because + * it might be assigned to 'tup_old', in which case it needs to + * survive into the next iteration. + */ + tup = get_changed_tuple(src); + + if (change.kind == CHANGE_UPDATE_OLD) + { + Assert(tup_old == NULL); + tup_old = tup; + } + else if (change.kind == CHANGE_INSERT) + { + Assert(tup_old == NULL); + + apply_concurrent_insert(rel, &change, tup, dest->iistate, + index_slot); + + pfree(tup); + } + else if (change.kind == CHANGE_UPDATE_NEW || + change.kind == CHANGE_DELETE) + { + HeapTuple tup_key; + + if (change.kind == CHANGE_UPDATE_NEW) + { + tup_key = tup_old != NULL ? tup_old : tup; + } + else + { + Assert(tup_old == NULL); + tup_key = tup; + } + + /* + * Find the tuple to be updated or deleted. + */ + tup_exist = find_target_tuple(rel, dest, tup_key, ident_slot); + if (tup_exist == NULL) + elog(ERROR, "Failed to find target tuple"); + + if (change.kind == CHANGE_UPDATE_NEW) + apply_concurrent_update(rel, tup, tup_exist, &change, + dest->iistate, index_slot); + else + apply_concurrent_delete(rel, tup_exist, &change); + + if (tup_old != NULL) + { + pfree(tup_old); + tup_old = NULL; + } + + pfree(tup); + } + else + elog(ERROR, "Unrecognized kind of change: %d", change.kind); + + /* + * If a change was applied now, increment CID for next writes and + * update the snapshot so it sees the changes we've applied so far. + */ + if (change.kind != CHANGE_UPDATE_OLD) + { + CommandCounterIncrement(); + UpdateActiveSnapshotCommandId(); + } + + /* TTSOpsMinimalTuple has .get_heap_tuple==NULL. */ + Assert(shouldFree); + pfree(tup_change); + } + + tuplestore_clear(dstate->tstore); + dstate->nchanges = 0; + + /* Cleanup. */ + ExecDropSingleTupleTableSlot(index_slot); + ExecDropSingleTupleTableSlot(ident_slot); +} + +static void +apply_concurrent_insert(Relation rel, ConcurrentChange *change, HeapTuple tup, + IndexInsertState *iistate, TupleTableSlot *index_slot) +{ + List *recheck; + + + /* + * Like simple_heap_insert(), but make sure that the INSERT is not + * logically decoded - see reform_and_rewrite_tuple() for more + * information. + */ + heap_insert(rel, tup, GetCurrentCommandId(true), HEAP_INSERT_NO_LOGICAL, + NULL); + + /* + * Update indexes. + * + * In case functions in the index need the active snapshot and caller + * hasn't set one. + */ + ExecStoreHeapTuple(tup, index_slot, false); + recheck = ExecInsertIndexTuples(iistate->rri, + index_slot, + iistate->estate, + false, /* update */ + false, /* noDupErr */ + NULL, /* specConflict */ + NIL, /* arbiterIndexes */ + false /* onlySummarizing */ + ); + + /* + * If recheck is required, it must have been preformed on the source + * relation by now. (All the logical changes we process here are already + * committed.) + */ + list_free(recheck); + + pgstat_progress_incr_param(PROGRESS_REPACK_HEAP_TUPLES_INSERTED, 1); +} + +static void +apply_concurrent_update(Relation rel, HeapTuple tup, HeapTuple tup_target, + ConcurrentChange *change, IndexInsertState *iistate, + TupleTableSlot *index_slot) +{ + LockTupleMode lockmode; + TM_FailureData tmfd; + TU_UpdateIndexes update_indexes; + TM_Result res; + List *recheck; + + /* + * Write the new tuple into the new heap. ('tup' gets the TID assigned + * here.) + * + * Do it like in simple_heap_update(), except for 'wal_logical' (and + * except for 'wait'). + */ + res = heap_update(rel, &tup_target->t_self, tup, + GetCurrentCommandId(true), + InvalidSnapshot, + false, /* no wait - only we are doing changes */ + &tmfd, &lockmode, &update_indexes, + false /* wal_logical */ ); + if (res != TM_Ok) + ereport(ERROR, (errmsg("failed to apply concurrent UPDATE"))); + + ExecStoreHeapTuple(tup, index_slot, false); + + if (update_indexes != TU_None) + { + recheck = ExecInsertIndexTuples(iistate->rri, + index_slot, + iistate->estate, + true, /* update */ + false, /* noDupErr */ + NULL, /* specConflict */ + NIL, /* arbiterIndexes */ + /* onlySummarizing */ + update_indexes == TU_Summarizing); + list_free(recheck); + } + + pgstat_progress_incr_param(PROGRESS_REPACK_HEAP_TUPLES_UPDATED, 1); +} + +static void +apply_concurrent_delete(Relation rel, HeapTuple tup_target, + ConcurrentChange *change) +{ + TM_Result res; + TM_FailureData tmfd; + + /* + * Delete tuple from the new heap. + * + * Do it like in simple_heap_delete(), except for 'wal_logical' (and + * except for 'wait'). + */ + res = heap_delete(rel, &tup_target->t_self, GetCurrentCommandId(true), + InvalidSnapshot, false, + &tmfd, + false, /* no wait - only we are doing changes */ + false /* wal_logical */ ); + + if (res != TM_Ok) + ereport(ERROR, (errmsg("failed to apply concurrent DELETE"))); + + pgstat_progress_incr_param(PROGRESS_REPACK_HEAP_TUPLES_DELETED, 1); +} + +/* + * Find the tuple to be updated or deleted. + * + * 'tup_key' is a tuple containing the key values for the scan. + * + * On exit,'*scan_p' contains the scan descriptor used. The caller must close + * it when he no longer needs the tuple returned. + */ +static HeapTuple +find_target_tuple(Relation rel, ChangeDest *dest, HeapTuple tup_key, + TupleTableSlot *ident_slot) +{ + Relation ident_index = dest->ident_index; + IndexScanDesc scan; + Form_pg_index ident_form; + int2vector *ident_indkey; + HeapTuple result = NULL; + + /* XXX no instrumentation for now */ + scan = index_beginscan(rel, ident_index, GetActiveSnapshot(), + NULL, dest->ident_key_nentries, 0); + index_rescan(scan, dest->ident_key, dest->ident_key_nentries, NULL, 0); + + /* Info needed to retrieve key values from heap tuple. */ + ident_form = ident_index->rd_index; + ident_indkey = &ident_form->indkey; + + /* Use the incoming tuple to finalize the scan key. */ + for (int i = 0; i < scan->numberOfKeys; i++) + { + ScanKey entry; + bool isnull; + int16 attno_heap; + + entry = &scan->keyData[i]; + attno_heap = ident_indkey->values[i]; + entry->sk_argument = heap_getattr(tup_key, + attno_heap, + rel->rd_att, + &isnull); + Assert(!isnull); + } + if (index_getnext_slot(scan, ForwardScanDirection, ident_slot)) + { + bool shouldFree; + + result = ExecFetchSlotHeapTuple(ident_slot, false, &shouldFree); + /* TTSOpsBufferHeapTuple has .get_heap_tuple != NULL. */ + Assert(!shouldFree); + } + index_endscan(scan); + + return result; +} + +/* + * Decode and apply concurrent changes. + */ +static void +process_concurrent_changes(LogicalDecodingContext *decoding_ctx, + XLogRecPtr end_of_wal, ChangeDest *dest) +{ + RepackDecodingState *dstate; + + pgstat_progress_update_param(PROGRESS_REPACK_PHASE, + PROGRESS_REPACK_PHASE_CATCH_UP); + + dstate = (RepackDecodingState *) decoding_ctx->output_writer_private; + + repack_decode_concurrent_changes(decoding_ctx, end_of_wal); + + if (dstate->nchanges == 0) + return; + + PG_TRY(); + { + /* + * Make sure that TOAST values can eventually be accessed via the old + * relation - see comment in copy_table_data(). + */ + if (OidIsValid(dest->toastrelid)) + dest->rel->rd_toastoid = dest->toastrelid; + + apply_concurrent_changes(dstate, dest); + } + + /* + * TODO Consider if the setting needs to be reverted: rel_dst will + * eventually be dropped, w/o other transactions being able to access it. + */ + PG_FINALLY(); + { + if (OidIsValid(dest->toastrelid)) + dest->rel->rd_toastoid = InvalidOid; + } + PG_END_TRY(); +} + +/* + * Initialize IndexInsertState for index specified by ident_index_id. + * + * While doing that, also return the identity index in *ident_index_p. + */ +static IndexInsertState * +get_index_insert_state(Relation relation, Oid ident_index_id, + Relation *ident_index_p) +{ + EState *estate; + int i; + IndexInsertState *result; + Relation ident_index = NULL; + + result = (IndexInsertState *) palloc0(sizeof(IndexInsertState)); + estate = CreateExecutorState(); + + result->rri = (ResultRelInfo *) palloc(sizeof(ResultRelInfo)); + InitResultRelInfo(result->rri, relation, 0, 0, 0); + ExecOpenIndices(result->rri, false); + + /* + * Find the relcache entry of the identity index so that we spend no extra + * effort to open / close it. + */ + for (i = 0; i < result->rri->ri_NumIndices; i++) + { + Relation ind_rel; + + ind_rel = result->rri->ri_IndexRelationDescs[i]; + if (ind_rel->rd_id == ident_index_id) + ident_index = ind_rel; + } + if (ident_index == NULL) + elog(ERROR, "Failed to open identity index"); + + /* Only initialize fields needed by ExecInsertIndexTuples(). */ + result->estate = estate; + + *ident_index_p = ident_index; + return result; +} + +/* + * Build scan key to process logical changes. + */ +static ScanKey +build_identity_key(Oid ident_idx_oid, Relation rel_src, int *nentries) +{ + Relation ident_idx_rel; + Form_pg_index ident_idx; + int n, + i; + ScanKey result; + + Assert(OidIsValid(ident_idx_oid)); + ident_idx_rel = index_open(ident_idx_oid, AccessShareLock); + ident_idx = ident_idx_rel->rd_index; + n = ident_idx->indnatts; + result = (ScanKey) palloc(sizeof(ScanKeyData) * n); + for (i = 0; i < n; i++) + { + ScanKey entry; + int16 relattno; + Form_pg_attribute att; + Oid opfamily, + opcintype, + opno, + opcode; + + entry = &result[i]; + relattno = ident_idx->indkey.values[i]; + if (relattno >= 1) + { + TupleDesc desc; + + desc = rel_src->rd_att; + att = TupleDescAttr(desc, relattno - 1); + } + else + elog(ERROR, "Unexpected attribute number %d in index", relattno); + + opfamily = ident_idx_rel->rd_opfamily[i]; + opcintype = ident_idx_rel->rd_opcintype[i]; + opno = get_opfamily_member(opfamily, opcintype, opcintype, + BTEqualStrategyNumber); + + if (!OidIsValid(opno)) + elog(ERROR, "Failed to find = operator for type %u", opcintype); + + opcode = get_opcode(opno); + if (!OidIsValid(opcode)) + elog(ERROR, "Failed to find = operator for operator %u", opno); + + /* Initialize everything but argument. */ + ScanKeyInit(entry, + i + 1, + BTEqualStrategyNumber, opcode, + (Datum) NULL); + entry->sk_collation = att->attcollation; + } + index_close(ident_idx_rel, AccessShareLock); + + *nentries = n; + return result; +} + +static void +free_index_insert_state(IndexInsertState *iistate) +{ + ExecCloseIndices(iistate->rri); + FreeExecutorState(iistate->estate); + pfree(iistate->rri); + pfree(iistate); +} + +static void +cleanup_logical_decoding(LogicalDecodingContext *ctx) +{ + RepackDecodingState *dstate; + + dstate = (RepackDecodingState *) ctx->output_writer_private; + + ExecDropSingleTupleTableSlot(dstate->tsslot); + FreeTupleDesc(dstate->tupdesc_change); + FreeTupleDesc(dstate->tupdesc); + tuplestore_end(dstate->tstore); + + FreeDecodingContext(ctx); + + ReplicationSlotRelease(); + ReplicationSlotDrop(NameStr(dstate->slotname), false); + pfree(dstate); +} + +/* + * The final steps of rebuild_relation() for concurrent processing. + * + * On entry, NewHeap is locked in AccessExclusiveLock mode. OldHeap and its + * clustering index (if one is passed) are still locked in a mode that allows + * concurrent data changes. On exit, both tables and their indexes are closed, + * but locked in AccessExclusiveLock mode. + */ +static void +rebuild_relation_finish_concurrent(Relation NewHeap, Relation OldHeap, + Relation cl_index, + LogicalDecodingContext *decoding_ctx, + bool swap_toast_by_content, + TransactionId frozenXid, + MultiXactId cutoffMulti) +{ + LOCKMODE lockmode_old PG_USED_FOR_ASSERTS_ONLY; + List *ind_oids_new; + Oid old_table_oid = RelationGetRelid(OldHeap); + Oid new_table_oid = RelationGetRelid(NewHeap); + List *ind_oids_old = RelationGetIndexList(OldHeap); + ListCell *lc, + *lc2; + char relpersistence; + bool is_system_catalog; + Oid ident_idx_old, + ident_idx_new; + XLogRecPtr wal_insert_ptr, + end_of_wal; + char dummy_rec_data = '\0'; + Relation *ind_refs, + *ind_refs_p; + int nind; + ChangeDest chgdst; + + /* Like in cluster_rel(). */ + lockmode_old = ShareUpdateExclusiveLock; + Assert(CheckRelationLockedByMe(OldHeap, lockmode_old, false)); + Assert(cl_index == NULL || + CheckRelationLockedByMe(cl_index, lockmode_old, false)); + /* This is expected from the caller. */ + Assert(CheckRelationLockedByMe(NewHeap, AccessExclusiveLock, false)); + + ident_idx_old = RelationGetReplicaIndex(OldHeap); + + /* + * Unlike the exclusive case, we build new indexes for the new relation + * rather than swapping the storage and reindexing the old relation. The + * point is that the index build can take some time, so we do it before we + * get AccessExclusiveLock on the old heap and therefore we cannot swap + * the heap storage yet. + * + * index_create() will lock the new indexes using AccessExclusiveLock - no + * need to change that. + * + * We assume that ShareUpdateExclusiveLock on the table prevents anyone + * from dropping the existing indexes or adding new ones, so the lists of + * old and new indexes should match at the swap time. On the other hand we + * do not block ALTER INDEX commands that do not require table lock (e.g. + * ALTER INDEX ... SET ...). + * + * XXX Should we check a the end of our work if another transaction + * executed such a command and issue a NOTICE that we might have discarded + * its effects? (For example, someone changes storage parameter after we + * have created the new index, the new value of that parameter is lost.) + * Alternatively, we can lock all the indexes now in a mode that blocks + * all the ALTER INDEX commands (ShareUpdateExclusiveLock ?), and keep + * them locked till the end of the transactions. That might increase the + * risk of deadlock during the lock upgrade below, however SELECT / DML + * queries should not be involved in such a deadlock. + */ + ind_oids_new = build_new_indexes(NewHeap, OldHeap, ind_oids_old); + + /* + * Processing shouldn't start w/o valid identity index. + */ + Assert(OidIsValid(ident_idx_old)); + + /* Find "identity index" on the new relation. */ + ident_idx_new = InvalidOid; + forboth(lc, ind_oids_old, lc2, ind_oids_new) + { + Oid ind_old = lfirst_oid(lc); + Oid ind_new = lfirst_oid(lc2); + + if (ident_idx_old == ind_old) + { + ident_idx_new = ind_new; + break; + } + } + if (!OidIsValid(ident_idx_new)) + + /* + * Should not happen, given our lock on the old relation. + */ + ereport(ERROR, + (errmsg("Identity index missing on the new relation"))); + + /* Gather information to apply concurrent changes. */ + chgdst.rel = NewHeap; + chgdst.toastrelid = swap_toast_by_content ? + OldHeap->rd_rel->reltoastrelid : InvalidOid; + chgdst.iistate = get_index_insert_state(NewHeap, ident_idx_new, + &chgdst.ident_index); + chgdst.ident_key = build_identity_key(ident_idx_new, OldHeap, + &chgdst.ident_key_nentries); + + /* + * During testing, wait for another backend to perform concurrent data + * changes which we will process below. + */ + INJECTION_POINT("repack-concurrently-before-lock", NULL); + + /* + * Flush all WAL records inserted so far (possibly except for the last + * incomplete page, see GetInsertRecPtr), to minimize the amount of data + * we need to flush while holding exclusive lock on the source table. + */ + wal_insert_ptr = GetInsertRecPtr(); + XLogFlush(wal_insert_ptr); + end_of_wal = GetFlushRecPtr(NULL); + + /* + * Apply concurrent changes first time, to minimize the time we need to + * hold AccessExclusiveLock. (Quite some amount of WAL could have been + * written during the data copying and index creation.) + */ + process_concurrent_changes(decoding_ctx, end_of_wal, &chgdst); + + /* + * Acquire AccessExclusiveLock on the table, its TOAST relation (if there + * is one), all its indexes, so that we can swap the files. + * + * Before that, unlock the index temporarily to avoid deadlock in case + * another transaction is trying to lock it while holding the lock on the + * table. + */ + if (cl_index) + { + index_close(cl_index, ShareUpdateExclusiveLock); + cl_index = NULL; + } + /* For the same reason, unlock TOAST relation. */ + if (OldHeap->rd_rel->reltoastrelid) + LockRelationOid(OldHeap->rd_rel->reltoastrelid, AccessExclusiveLock); + /* Finally lock the table */ + LockRelationOid(old_table_oid, AccessExclusiveLock); + + /* + * Lock all indexes now, not only the clustering one: all indexes need to + * have their files swapped. While doing that, store their relation + * references in an array, to handle predicate locks below. + */ + ind_refs_p = ind_refs = palloc_array(Relation, list_length(ind_oids_old)); + nind = 0; + foreach(lc, ind_oids_old) + { + Oid ind_oid; + Relation index; + + ind_oid = lfirst_oid(lc); + index = index_open(ind_oid, AccessExclusiveLock); + + /* + * TODO 1) Do we need to check if ALTER INDEX was executed since the + * new index was created in build_new_indexes()? 2) Specifically for + * the clustering index, should check_index_is_clusterable() be called + * here? (Not sure about the latter: ShareUpdateExclusiveLock on the + * table probably blocks all commands that affect the result of + * check_index_is_clusterable().) + */ + *ind_refs_p = index; + ind_refs_p++; + nind++; + } + + /* + * In addition, lock the OldHeap's TOAST relation exclusively - again, the + * lock is needed to swap the files. + */ + if (OidIsValid(OldHeap->rd_rel->reltoastrelid)) + LockRelationOid(OldHeap->rd_rel->reltoastrelid, AccessExclusiveLock); + + /* + * Tuples and pages of the old heap will be gone, but the heap will stay. + */ + TransferPredicateLocksToHeapRelation(OldHeap); + /* The same for indexes. */ + for (int i = 0; i < nind; i++) + { + Relation index = ind_refs[i]; + + TransferPredicateLocksToHeapRelation(index); + + /* + * References to indexes on the old relation are not needed anymore, + * however locks stay till the end of the transaction. + */ + index_close(index, NoLock); + } + pfree(ind_refs); + + /* + * Flush anything we see in WAL, to make sure that all changes committed + * while we were waiting for the exclusive lock are available for + * decoding. This should not be necessary if all backends had + * synchronous_commit set, but we can't rely on this setting. + * + * Unfortunately, GetInsertRecPtr() may lag behind the actual insert + * position, and GetLastImportantRecPtr() points at the start of the last + * record rather than at the end. Thus the simplest way to determine the + * insert position is to insert a dummy record and use its LSN. + * + * XXX Consider using GetLastImportantRecPtr() and adding the size of the + * last record (plus the total size of all the page headers the record + * spans)? + */ + XLogBeginInsert(); + XLogRegisterData(&dummy_rec_data, 1); + wal_insert_ptr = XLogInsert(RM_XLOG_ID, XLOG_NOOP); + XLogFlush(wal_insert_ptr); + end_of_wal = GetFlushRecPtr(NULL); + + /* Apply the concurrent changes again. */ + process_concurrent_changes(decoding_ctx, end_of_wal, &chgdst); + + /* Remember info about rel before closing OldHeap */ + relpersistence = OldHeap->rd_rel->relpersistence; + is_system_catalog = IsSystemRelation(OldHeap); + + pgstat_progress_update_param(PROGRESS_REPACK_PHASE, + PROGRESS_REPACK_PHASE_SWAP_REL_FILES); + + /* + * Even ShareUpdateExclusiveLock should have prevented others from + * creating / dropping indexes (even using the CONCURRENTLY option), so we + * do not need to check whether the lists match. + */ + forboth(lc, ind_oids_old, lc2, ind_oids_new) + { + Oid ind_old = lfirst_oid(lc); + Oid ind_new = lfirst_oid(lc2); + Oid mapped_tables[4]; + + /* Zero out possible results from swapped_relation_files */ + memset(mapped_tables, 0, sizeof(mapped_tables)); + + swap_relation_files(ind_old, ind_new, + (old_table_oid == RelationRelationId), + swap_toast_by_content, + true, + InvalidTransactionId, + InvalidMultiXactId, + mapped_tables); + +#ifdef USE_ASSERT_CHECKING + + /* + * Concurrent processing is not supported for system relations, so + * there should be no mapped tables. + */ + for (int i = 0; i < 4; i++) + Assert(mapped_tables[i] == 0); +#endif + } + + /* The new indexes must be visible for deletion. */ + CommandCounterIncrement(); + + /* Close the old heap but keep lock until transaction commit. */ + table_close(OldHeap, NoLock); + /* Close the new heap. (We didn't have to open its indexes). */ + table_close(NewHeap, NoLock); + + /* Cleanup what we don't need anymore. (And close the identity index.) */ + pfree(chgdst.ident_key); + free_index_insert_state(chgdst.iistate); + + /* + * Swap the relations and their TOAST relations and TOAST indexes. This + * also drops the new relation and its indexes. + * + * (System catalogs are currently not supported.) + */ + Assert(!is_system_catalog); + finish_heap_swap(old_table_oid, new_table_oid, + is_system_catalog, + swap_toast_by_content, + false, true, false, + frozenXid, cutoffMulti, + relpersistence); +} + +/* + * Build indexes on NewHeap according to those on OldHeap. + * + * OldIndexes is the list of index OIDs on OldHeap. + * + * A list of OIDs of the corresponding indexes created on NewHeap is + * returned. The order of items does match, so we can use these arrays to swap + * index storage. + */ +static List * +build_new_indexes(Relation NewHeap, Relation OldHeap, List *OldIndexes) +{ + ListCell *lc; + List *result = NIL; + + pgstat_progress_update_param(PROGRESS_REPACK_PHASE, + PROGRESS_REPACK_PHASE_REBUILD_INDEX); + + foreach(lc, OldIndexes) + { + Oid ind_oid, + ind_oid_new; + char *newName; + Relation ind; + + ind_oid = lfirst_oid(lc); + ind = index_open(ind_oid, AccessShareLock); + + newName = ChooseRelationName(get_rel_name(ind_oid), + NULL, + "repacknew", + get_rel_namespace(ind->rd_index->indrelid), + false); + ind_oid_new = index_create_copy(NewHeap, ind_oid, + ind->rd_rel->reltablespace, newName, + false); + result = lappend_oid(result, ind_oid_new); + + index_close(ind, AccessShareLock); + } + + return result; +} diff --git a/src/backend/commands/matview.c b/src/backend/commands/matview.c index ef7c0d624f1..d12e2d0f2e0 100644 --- a/src/backend/commands/matview.c +++ b/src/backend/commands/matview.c @@ -892,7 +892,7 @@ refresh_by_match_merge(Oid matviewOid, Oid tempOid, Oid relowner, static void refresh_by_heap_swap(Oid matviewOid, Oid OIDNewHeap, char relpersistence) { - finish_heap_swap(matviewOid, OIDNewHeap, false, false, true, true, + finish_heap_swap(matviewOid, OIDNewHeap, false, false, true, true, true, RecentXmin, ReadNextMultiXactId(), relpersistence); } diff --git a/src/backend/commands/tablecmds.c b/src/backend/commands/tablecmds.c index 07e5b95782e..1bce85e4232 100644 --- a/src/backend/commands/tablecmds.c +++ b/src/backend/commands/tablecmds.c @@ -5992,6 +5992,7 @@ ATRewriteTables(AlterTableStmt *parsetree, List **wqueue, LOCKMODE lockmode, finish_heap_swap(tab->relid, OIDNewHeap, false, false, true, !OidIsValid(tab->newTableSpace), + true, RecentXmin, ReadNextMultiXactId(), persistence); diff --git a/src/backend/commands/vacuum.c b/src/backend/commands/vacuum.c index 827e66724b5..03478ecd5cc 100644 --- a/src/backend/commands/vacuum.c +++ b/src/backend/commands/vacuum.c @@ -126,7 +126,7 @@ static void vac_truncate_clog(TransactionId frozenXID, TransactionId lastSaneFrozenXid, MultiXactId lastSaneMinMulti); static bool vacuum_rel(Oid relid, RangeVar *relation, VacuumParams params, - BufferAccessStrategy bstrategy); + BufferAccessStrategy bstrategy, bool isTopLevel); static double compute_parallel_delay(void); static VacOptValue get_vacoptval_from_boolean(DefElem *def); static bool vac_tid_reaped(ItemPointer itemptr, void *state); @@ -627,7 +627,8 @@ vacuum(List *relations, const VacuumParams params, BufferAccessStrategy bstrateg if (params.options & VACOPT_VACUUM) { - if (!vacuum_rel(vrel->oid, vrel->relation, params, bstrategy)) + if (!vacuum_rel(vrel->oid, vrel->relation, params, bstrategy, + isTopLevel)) continue; } @@ -1997,7 +1998,7 @@ vac_truncate_clog(TransactionId frozenXID, */ static bool vacuum_rel(Oid relid, RangeVar *relation, VacuumParams params, - BufferAccessStrategy bstrategy) + BufferAccessStrategy bstrategy, bool isTopLevel) { LOCKMODE lmode; Relation rel; @@ -2288,7 +2289,7 @@ vacuum_rel(Oid relid, RangeVar *relation, VacuumParams params, /* VACUUM FULL is a variant of REPACK; see cluster.c */ cluster_rel(REPACK_COMMAND_VACUUMFULL, rel, InvalidOid, - &cluster_params); + &cluster_params, isTopLevel); /* cluster_rel closes the relation, but keeps lock */ rel = NULL; @@ -2331,7 +2332,8 @@ vacuum_rel(Oid relid, RangeVar *relation, VacuumParams params, toast_vacuum_params.options |= VACOPT_PROCESS_MAIN; toast_vacuum_params.toast_parent = relid; - vacuum_rel(toast_relid, NULL, toast_vacuum_params, bstrategy); + vacuum_rel(toast_relid, NULL, toast_vacuum_params, bstrategy, + isTopLevel); } /* diff --git a/src/backend/meson.build b/src/backend/meson.build index b831a541652..5c148131217 100644 --- a/src/backend/meson.build +++ b/src/backend/meson.build @@ -194,5 +194,6 @@ pg_test_mod_args = pg_mod_args + { subdir('jit/llvm') subdir('replication/libpqwalreceiver') subdir('replication/pgoutput') +subdir('replication/pgoutput_repack') subdir('snowball') subdir('utils/mb/conversion_procs') diff --git a/src/backend/replication/logical/decode.c b/src/backend/replication/logical/decode.c index cc03f0706e9..73fc4d30c67 100644 --- a/src/backend/replication/logical/decode.c +++ b/src/backend/replication/logical/decode.c @@ -33,6 +33,7 @@ #include "access/xlogreader.h" #include "access/xlogrecord.h" #include "catalog/pg_control.h" +#include "commands/cluster.h" #include "replication/decode.h" #include "replication/logical.h" #include "replication/message.h" @@ -472,6 +473,88 @@ heap_decode(LogicalDecodingContext *ctx, XLogRecordBuffer *buf) TransactionId xid = XLogRecGetXid(buf->record); SnapBuild *builder = ctx->snapshot_builder; + /* + * If the change is not intended for logical decoding, do not even + * establish transaction for it - REPACK CONCURRENTLY is the typical use + * case. + * + * First, check if REPACK CONCURRENTLY is being performed by this backend. + * If so, only decode data changes of the table that it is processing, and + * the changes of its TOAST relation. + * + * (TOAST locator should not be set unless the main is.) + */ + Assert(!OidIsValid(repacked_rel_toast_locator.relNumber) || + OidIsValid(repacked_rel_locator.relNumber)); + + if (OidIsValid(repacked_rel_locator.relNumber)) + { + XLogReaderState *r = buf->record; + RelFileLocator locator; + + /* Not all records contain the block. */ + if (XLogRecGetBlockTagExtended(r, 0, &locator, NULL, NULL, NULL) && + !RelFileLocatorEquals(locator, repacked_rel_locator) && + (!OidIsValid(repacked_rel_toast_locator.relNumber) || + !RelFileLocatorEquals(locator, repacked_rel_toast_locator))) + return; + } + + /* + * Second, skip records which do not contain sufficient information for + * the decoding. + * + * The problem we solve here is that REPACK CONCURRENTLY generates WAL + * when doing changes in the new table. Those changes should not be useful + * for any other user (such as logical replication subscription) because + * the new table will eventually be dropped (after REPACK CONCURRENTLY has + * assigned its file to the "old table"). + */ + switch (info) + { + case XLOG_HEAP_INSERT: + { + xl_heap_insert *rec; + + rec = (xl_heap_insert *) XLogRecGetData(buf->record); + + /* + * This does happen when 1) raw_heap_insert marks the TOAST + * record as HEAP_INSERT_NO_LOGICAL, 2) REPACK CONCURRENTLY + * replays inserts performed by other backends. + */ + if ((rec->flags & XLH_INSERT_CONTAINS_NEW_TUPLE) == 0) + return; + + break; + } + + case XLOG_HEAP_HOT_UPDATE: + case XLOG_HEAP_UPDATE: + { + xl_heap_update *rec; + + rec = (xl_heap_update *) XLogRecGetData(buf->record); + if ((rec->flags & + (XLH_UPDATE_CONTAINS_NEW_TUPLE | + XLH_UPDATE_CONTAINS_OLD_TUPLE | + XLH_UPDATE_CONTAINS_OLD_KEY)) == 0) + return; + + break; + } + + case XLOG_HEAP_DELETE: + { + xl_heap_delete *rec; + + rec = (xl_heap_delete *) XLogRecGetData(buf->record); + if (rec->flags & XLH_DELETE_NO_LOGICAL) + return; + break; + } + } + ReorderBufferProcessXid(ctx->reorder, xid, buf->origptr); /* @@ -512,6 +595,16 @@ heap_decode(LogicalDecodingContext *ctx, XLogRecordBuffer *buf) break; case XLOG_HEAP_TRUNCATE: + /* Is REPACK (CONCURRENTLY) being run by this backend? */ + if (OidIsValid(repacked_rel_locator.relNumber)) + /* + * TRUNCATE changes rd_locator of the relation, so it'd break + * REPACK (CONCURRENTLY). In fact it should not happen because + * TRUNCATE needs AccessExclusiveLock on the table. Should we + * only use Assert() here? + */ + ereport(ERROR, + (errmsg("TRUNCATE encountered while doing REPACK (CONCURRENTLY)"))); if (SnapBuildProcessChange(builder, xid, buf->origptr) && !ctx->fast_forward) DecodeTruncate(ctx, buf); diff --git a/src/backend/replication/logical/snapbuild.c b/src/backend/replication/logical/snapbuild.c index dcf32101c4c..63efb577975 100644 --- a/src/backend/replication/logical/snapbuild.c +++ b/src/backend/replication/logical/snapbuild.c @@ -486,12 +486,33 @@ SnapBuildInitialSnapshot(SnapBuild *builder) return SnapBuildMVCCFromHistoric(snap, true); } +/* + * Build an MVCC snapshot for the initial data load performed by REPACK + * CONCURRENTLY command. + * + * The snapshot will only be used to scan one particular relation, which is + * treated like a catalog (therefore ->building_full_snapshot is not + * important), and the caller should already have a replication slot setup (so + * we do not set MyProc->xmin). XXX Do we yet need to add some restrictions? + */ +Snapshot +SnapBuildInitialSnapshotForRepack(SnapBuild *builder) +{ + Snapshot snap; + + Assert(builder->state == SNAPBUILD_CONSISTENT); + Assert(builder->building_full_snapshot); + + snap = SnapBuildBuildSnapshot(builder); + return SnapBuildMVCCFromHistoric(snap, false); +} + /* * Turn a historic MVCC snapshot into an ordinary MVCC snapshot. * * Unlike a regular (non-historic) MVCC snapshot, the xip array of this * snapshot contains not only running main transactions, but also their - * subtransactions. This difference does has no impact on XidInMVCCSnapshot(). + * subtransactions. This difference has no impact on XidInMVCCSnapshot(). * * Pass true for 'in_place' if you don't care about modifying the source * snapshot. If you need a new instance, and one that was allocated as a diff --git a/src/backend/replication/pgoutput_repack/Makefile b/src/backend/replication/pgoutput_repack/Makefile new file mode 100644 index 00000000000..4efeb713b70 --- /dev/null +++ b/src/backend/replication/pgoutput_repack/Makefile @@ -0,0 +1,32 @@ +#------------------------------------------------------------------------- +# +# Makefile-- +# Makefile for src/backend/replication/pgoutput_repack +# +# IDENTIFICATION +# src/backend/replication/pgoutput_repack +# +#------------------------------------------------------------------------- + +subdir = src/backend/replication/pgoutput_repack +top_builddir = ../../../.. +include $(top_builddir)/src/Makefile.global + +OBJS = \ + $(WIN32RES) \ + pgoutput_repack.o +PGFILEDESC = "pgoutput_repack - logical replication output plugin for REPACK command" +NAME = pgoutput_repack + +all: all-shared-lib + +include $(top_srcdir)/src/Makefile.shlib + +install: all installdirs install-lib + +installdirs: installdirs-lib + +uninstall: uninstall-lib + +clean distclean: clean-lib + rm -f $(OBJS) diff --git a/src/backend/replication/pgoutput_repack/meson.build b/src/backend/replication/pgoutput_repack/meson.build new file mode 100644 index 00000000000..133e865a4a0 --- /dev/null +++ b/src/backend/replication/pgoutput_repack/meson.build @@ -0,0 +1,18 @@ +# Copyright (c) 2022-2024, PostgreSQL Global Development Group + +pgoutput_repack_sources = files( + 'pgoutput_repack.c', +) + +if host_system == 'windows' + pgoutput_repack_sources += rc_lib_gen.process(win32ver_rc, extra_args: [ + '--NAME', 'pgoutput_repack', + '--FILEDESC', 'pgoutput_repack - logical replication output plugin for REPACK command',]) +endif + +pgoutput_repack = shared_module('pgoutput_repack', + pgoutput_repack_sources, + kwargs: pg_mod_args, +) + +backend_targets += pgoutput_repack diff --git a/src/backend/replication/pgoutput_repack/pgoutput_repack.c b/src/backend/replication/pgoutput_repack/pgoutput_repack.c new file mode 100644 index 00000000000..178e47bdce4 --- /dev/null +++ b/src/backend/replication/pgoutput_repack/pgoutput_repack.c @@ -0,0 +1,248 @@ +/*------------------------------------------------------------------------- + * + * pgoutput_repack.c + * Logical Replication output plugin for REPACK command + * + * Copyright (c) 2012-2024, PostgreSQL Global Development Group + * + * IDENTIFICATION + * src/backend/replication/pgoutput_repack/pgoutput_repack.c + * + *------------------------------------------------------------------------- + */ +#include "postgres.h" + +#include "access/heaptoast.h" +#include "commands/cluster.h" +#include "replication/snapbuild.h" + +PG_MODULE_MAGIC; + +static void plugin_startup(LogicalDecodingContext *ctx, + OutputPluginOptions *opt, bool is_init); +static void plugin_shutdown(LogicalDecodingContext *ctx); +static void plugin_begin_txn(LogicalDecodingContext *ctx, + ReorderBufferTXN *txn); +static void plugin_commit_txn(LogicalDecodingContext *ctx, + ReorderBufferTXN *txn, XLogRecPtr commit_lsn); +static void plugin_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, + Relation rel, ReorderBufferChange *change); +static void store_change(LogicalDecodingContext *ctx, + ConcurrentChangeKind kind, HeapTuple tuple); + +void +_PG_output_plugin_init(OutputPluginCallbacks *cb) +{ + AssertVariableIsOfType(&_PG_output_plugin_init, LogicalOutputPluginInit); + + cb->startup_cb = plugin_startup; + cb->begin_cb = plugin_begin_txn; + cb->change_cb = plugin_change; + cb->commit_cb = plugin_commit_txn; + cb->shutdown_cb = plugin_shutdown; +} + + +/* initialize this plugin */ +static void +plugin_startup(LogicalDecodingContext *ctx, OutputPluginOptions *opt, + bool is_init) +{ + ctx->output_plugin_private = NULL; + + /* Probably unnecessary, as we don't use the SQL interface ... */ + opt->output_type = OUTPUT_PLUGIN_BINARY_OUTPUT; + + if (ctx->output_plugin_options != NIL) + { + ereport(ERROR, + (errcode(ERRCODE_INVALID_PARAMETER_VALUE), + errmsg("This plugin does not expect any options"))); + } +} + +static void +plugin_shutdown(LogicalDecodingContext *ctx) +{ +} + +/* + * As we don't release the slot during processing of particular table, there's + * no room for SQL interface, even for debugging purposes. Therefore we need + * neither OutputPluginPrepareWrite() nor OutputPluginWrite() in the plugin + * callbacks. (Although we might want to write custom callbacks, this API + * seems to be unnecessarily generic for our purposes.) + */ + +/* BEGIN callback */ +static void +plugin_begin_txn(LogicalDecodingContext *ctx, ReorderBufferTXN *txn) +{ +} + +/* COMMIT callback */ +static void +plugin_commit_txn(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, + XLogRecPtr commit_lsn) +{ +} + +/* + * Callback for individual changed tuples + */ +static void +plugin_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, + Relation relation, ReorderBufferChange *change) +{ + RepackDecodingState *dstate; + + dstate = (RepackDecodingState *) ctx->output_writer_private; + + /* Only interested in one particular relation. */ + if (relation->rd_id != dstate->relid) + return; + + /* Decode entry depending on its type */ + switch (change->action) + { + case REORDER_BUFFER_CHANGE_INSERT: + { + HeapTuple newtuple; + + newtuple = change->data.tp.newtuple != NULL ? + change->data.tp.newtuple : NULL; + + /* + * Identity checks in the main function should have made this + * impossible. + */ + if (newtuple == NULL) + elog(ERROR, "Incomplete insert info."); + + store_change(ctx, CHANGE_INSERT, newtuple); + } + break; + case REORDER_BUFFER_CHANGE_UPDATE: + { + HeapTuple oldtuple, + newtuple; + + oldtuple = change->data.tp.oldtuple != NULL ? + change->data.tp.oldtuple : NULL; + newtuple = change->data.tp.newtuple != NULL ? + change->data.tp.newtuple : NULL; + + if (newtuple == NULL) + elog(ERROR, "Incomplete update info."); + + if (oldtuple != NULL) + store_change(ctx, CHANGE_UPDATE_OLD, oldtuple); + + store_change(ctx, CHANGE_UPDATE_NEW, newtuple); + } + break; + case REORDER_BUFFER_CHANGE_DELETE: + { + HeapTuple oldtuple; + + oldtuple = change->data.tp.oldtuple ? + change->data.tp.oldtuple : NULL; + + if (oldtuple == NULL) + elog(ERROR, "Incomplete delete info."); + + store_change(ctx, CHANGE_DELETE, oldtuple); + } + break; + default: + /* Should not come here */ + Assert(false); + break; + } +} + +/* Store concurrent data change. */ +static void +store_change(LogicalDecodingContext *ctx, ConcurrentChangeKind kind, + HeapTuple tuple) +{ + RepackDecodingState *dstate; + char *change_raw; + ConcurrentChange change; + bool flattened = false; + Size size; + Datum values[1]; + bool isnull[1]; + char *dst, + *dst_start; + + dstate = (RepackDecodingState *) ctx->output_writer_private; + + size = MAXALIGN(VARHDRSZ) + SizeOfConcurrentChange; + + if (tuple) + { + /* + * ReorderBufferCommit() stores the TOAST chunks in its private memory + * context and frees them after having called apply_change(). + * Therefore we need flat copy (including TOAST) that we eventually + * copy into the memory context which is available to + * decode_concurrent_changes(). + */ + if (HeapTupleHasExternal(tuple)) + { + /* + * toast_flatten_tuple_to_datum() might be more convenient but we + * don't want the decompression it does. + */ + tuple = toast_flatten_tuple(tuple, dstate->tupdesc); + flattened = true; + } + + size += tuple->t_len; + } + + /* XXX Isn't there any function / macro to do this? */ + if (size >= 0x3FFFFFFF) + elog(ERROR, "Change is too big."); + + /* Construct the change. */ + change_raw = (char *) palloc0(size); + SET_VARSIZE(change_raw, size); + + /* + * Since the varlena alignment might not be sufficient for the structure, + * set the fields in a local instance and remember where it should + * eventually be copied. + */ + change.kind = kind; + dst_start = (char *) VARDATA(change_raw); + + /* + * Copy the tuple. + * + * CAUTION: change->tup_data.t_data must be fixed on retrieval! + */ + memcpy(&change.tup_data, tuple, sizeof(HeapTupleData)); + dst = dst_start + SizeOfConcurrentChange; + memcpy(dst, tuple->t_data, tuple->t_len); + + /* The data has been copied. */ + if (flattened) + pfree(tuple); + + /* Copy the structure so it can be stored. */ + memcpy(dst_start, &change, SizeOfConcurrentChange); + + /* Store as tuple of 1 bytea column. */ + values[0] = PointerGetDatum(change_raw); + isnull[0] = false; + tuplestore_putvalues(dstate->tstore, dstate->tupdesc_change, + values, isnull); + + /* Accounting. */ + dstate->nchanges++; + + /* Cleanup. */ + pfree(change_raw); +} diff --git a/src/backend/storage/lmgr/generate-lwlocknames.pl b/src/backend/storage/lmgr/generate-lwlocknames.pl index cd3e43c448a..519f3953638 100644 --- a/src/backend/storage/lmgr/generate-lwlocknames.pl +++ b/src/backend/storage/lmgr/generate-lwlocknames.pl @@ -162,7 +162,7 @@ while (<$lwlocklist>) die "$wait_event_lwlocks[$lwlock_count] defined in wait_event_names.txt but " - . " missing from lwlocklist.h" + . "missing from lwlocklist.h" if $lwlock_count < scalar @wait_event_lwlocks; die diff --git a/src/backend/utils/cache/relcache.c b/src/backend/utils/cache/relcache.c index 915d0bc9084..d0f01d85bd3 100644 --- a/src/backend/utils/cache/relcache.c +++ b/src/backend/utils/cache/relcache.c @@ -64,6 +64,7 @@ #include "catalog/pg_type.h" #include "catalog/schemapg.h" #include "catalog/storage.h" +#include "commands/cluster.h" #include "commands/policy.h" #include "commands/publicationcmds.h" #include "commands/trigger.h" diff --git a/src/backend/utils/time/snapmgr.c b/src/backend/utils/time/snapmgr.c index 886060305f5..fbb3d66bbd9 100644 --- a/src/backend/utils/time/snapmgr.c +++ b/src/backend/utils/time/snapmgr.c @@ -214,7 +214,6 @@ static List *exportedSnapshots = NIL; /* Prototypes for local functions */ static void UnregisterSnapshotNoOwner(Snapshot snapshot); -static void FreeSnapshot(Snapshot snapshot); static void SnapshotResetXmin(void); /* ResourceOwner callbacks to track snapshot references */ @@ -659,7 +658,7 @@ CopySnapshot(Snapshot snapshot) * FreeSnapshot * Free the memory associated with a snapshot. */ -static void +void FreeSnapshot(Snapshot snapshot) { Assert(snapshot->regd_count == 0); diff --git a/src/bin/psql/tab-complete.in.c b/src/bin/psql/tab-complete.in.c index 626d9f1c98b..0fcf343d3af 100644 --- a/src/bin/psql/tab-complete.in.c +++ b/src/bin/psql/tab-complete.in.c @@ -5075,8 +5075,8 @@ match_previous_words(int pattern_id, * one word, so the above test is correct. */ if (ends_with(prev_wd, '(') || ends_with(prev_wd, ',')) - COMPLETE_WITH("ANALYZE", "VERBOSE"); - else if (TailMatches("ANALYZE", "VERBOSE")) + COMPLETE_WITH("ANALYZE", "CONCURRENTLY", "VERBOSE"); + else if (TailMatches("ANALYZE", "CONCURRENTLY", "VERBOSE")) COMPLETE_WITH("ON", "OFF"); } diff --git a/src/include/access/heapam.h b/src/include/access/heapam.h index 632c4332a8c..b0d6af0474c 100644 --- a/src/include/access/heapam.h +++ b/src/include/access/heapam.h @@ -361,14 +361,15 @@ extern void heap_multi_insert(Relation relation, TupleTableSlot **slots, BulkInsertState bistate); extern TM_Result heap_delete(Relation relation, const ItemPointerData *tid, CommandId cid, Snapshot crosscheck, bool wait, - TM_FailureData *tmfd, bool changingPart); + TM_FailureData *tmfd, bool changingPart, + bool wal_logical); extern void heap_finish_speculative(Relation relation, const ItemPointerData *tid); extern void heap_abort_speculative(Relation relation, const ItemPointerData *tid); extern TM_Result heap_update(Relation relation, const ItemPointerData *otid, HeapTuple newtup, CommandId cid, Snapshot crosscheck, bool wait, TM_FailureData *tmfd, LockTupleMode *lockmode, - TU_UpdateIndexes *update_indexes); + TU_UpdateIndexes *update_indexes, bool wal_logical); extern TM_Result heap_lock_tuple(Relation relation, HeapTuple tuple, CommandId cid, LockTupleMode mode, LockWaitPolicy wait_policy, bool follow_updates, @@ -445,6 +446,10 @@ extern HTSV_Result HeapTupleSatisfiesVacuumHorizon(HeapTuple htup, Buffer buffer TransactionId *dead_after); extern void HeapTupleSetHintBits(HeapTupleHeader tuple, Buffer buffer, uint16 infomask, TransactionId xid); +extern bool HeapTupleMVCCInserted(HeapTuple htup, Snapshot snapshot, + Buffer buffer); +extern bool HeapTupleMVCCNotDeleted(HeapTuple htup, Snapshot snapshot, + Buffer buffer); extern bool HeapTupleHeaderIsOnlyLocked(HeapTupleHeader tuple); extern bool HeapTupleIsSurelyDead(HeapTuple htup, GlobalVisState *vistest); diff --git a/src/include/access/heapam_xlog.h b/src/include/access/heapam_xlog.h index 16c2b2e3c9c..2cc49fd48de 100644 --- a/src/include/access/heapam_xlog.h +++ b/src/include/access/heapam_xlog.h @@ -104,6 +104,8 @@ #define XLH_DELETE_CONTAINS_OLD_KEY (1<<2) #define XLH_DELETE_IS_SUPER (1<<3) #define XLH_DELETE_IS_PARTITION_MOVE (1<<4) +/* See heap_delete() */ +#define XLH_DELETE_NO_LOGICAL (1<<5) /* convenience macro for checking whether any form of old tuple was logged */ #define XLH_DELETE_CONTAINS_OLD \ diff --git a/src/include/access/tableam.h b/src/include/access/tableam.h index 2fa790b6bf5..d8f76d325f9 100644 --- a/src/include/access/tableam.h +++ b/src/include/access/tableam.h @@ -22,6 +22,7 @@ #include "access/xact.h" #include "commands/vacuum.h" #include "executor/tuptable.h" +#include "replication/logical.h" #include "storage/read_stream.h" #include "utils/rel.h" #include "utils/snapshot.h" @@ -629,6 +630,8 @@ typedef struct TableAmRoutine Relation OldIndex, bool use_sort, TransactionId OldestXmin, + Snapshot snapshot, + LogicalDecodingContext *decoding_ctx, TransactionId *xid_cutoff, MultiXactId *multi_cutoff, double *num_tuples, @@ -1646,6 +1649,10 @@ table_relation_copy_data(Relation rel, const RelFileLocator *newrlocator) * not needed for the relation's AM * - *xid_cutoff - ditto * - *multi_cutoff - ditto + * - snapshot - if != NULL, ignore data changes done by transactions that this + * (MVCC) snapshot considers still in-progress or in the future. + * - decoding_ctx - logical decoding context, to capture concurrent data + * changes. * * Output parameters: * - *xid_cutoff - rel's new relfrozenxid value, may be invalid @@ -1658,6 +1665,8 @@ table_relation_copy_for_cluster(Relation OldTable, Relation NewTable, Relation OldIndex, bool use_sort, TransactionId OldestXmin, + Snapshot snapshot, + LogicalDecodingContext *decoding_ctx, TransactionId *xid_cutoff, MultiXactId *multi_cutoff, double *num_tuples, @@ -1666,6 +1675,7 @@ table_relation_copy_for_cluster(Relation OldTable, Relation NewTable, { OldTable->rd_tableam->relation_copy_for_cluster(OldTable, NewTable, OldIndex, use_sort, OldestXmin, + snapshot, decoding_ctx, xid_cutoff, multi_cutoff, num_tuples, tups_vacuumed, tups_recently_dead); diff --git a/src/include/commands/cluster.h b/src/include/commands/cluster.h index 652542e8e65..b43a1740053 100644 --- a/src/include/commands/cluster.h +++ b/src/include/commands/cluster.h @@ -13,10 +13,15 @@ #ifndef CLUSTER_H #define CLUSTER_H +#include "nodes/execnodes.h" #include "nodes/parsenodes.h" #include "parser/parse_node.h" +#include "replication/logical.h" #include "storage/lock.h" +#include "storage/relfilelocator.h" #include "utils/relcache.h" +#include "utils/resowner.h" +#include "utils/tuplestore.h" /* flag bits for ClusterParams->options */ @@ -25,6 +30,8 @@ #define CLUOPT_RECHECK_ISCLUSTERED 0x04 /* recheck relation state for * indisclustered */ #define CLUOPT_ANALYZE 0x08 /* do an ANALYZE */ +#define CLUOPT_CONCURRENT 0x10 /* allow concurrent data changes */ + /* options for CLUSTER */ typedef struct ClusterParams @@ -33,14 +40,94 @@ typedef struct ClusterParams } ClusterParams; +/* + * The following definitions are used by REPACK CONCURRENTLY. + */ + +extern RelFileLocator repacked_rel_locator; +extern RelFileLocator repacked_rel_toast_locator; + +typedef enum +{ + CHANGE_INSERT, + CHANGE_UPDATE_OLD, + CHANGE_UPDATE_NEW, + CHANGE_DELETE +} ConcurrentChangeKind; + +typedef struct ConcurrentChange +{ + /* See the enum above. */ + ConcurrentChangeKind kind; + + /* + * The actual tuple. + * + * The tuple data follows the ConcurrentChange structure. Before use make + * sure the tuple is correctly aligned (ConcurrentChange can be stored as + * bytea) and that tuple->t_data is fixed. + */ + HeapTupleData tup_data; +} ConcurrentChange; + +#define SizeOfConcurrentChange (offsetof(ConcurrentChange, tup_data) + \ + sizeof(HeapTupleData)) + +/* + * Logical decoding state. + * + * Here we store the data changes that we decode from WAL while the table + * contents is being copied to a new storage. Also the necessary metadata + * needed to apply these changes to the table is stored here. + */ +typedef struct RepackDecodingState +{ + /* The relation whose changes we're decoding. */ + Oid relid; + + /* Replication slot name. */ + NameData slotname; + + /* + * Decoded changes are stored here. Although we try to avoid excessive + * batches, it can happen that the changes need to be stored to disk. The + * tuplestore does this transparently. + */ + Tuplestorestate *tstore; + + /* The current number of changes in tstore. */ + double nchanges; + + /* + * Descriptor to store the ConcurrentChange structure serialized (bytea). + * We can't store the tuple directly because tuplestore only supports + * minimum tuple and we may need to transfer OID system column from the + * output plugin. Also we need to transfer the change kind, so it's better + * to put everything in the structure than to use 2 tuplestores "in + * parallel". + */ + TupleDesc tupdesc_change; + + /* Tuple descriptor needed to update indexes. */ + TupleDesc tupdesc; + + /* Slot to retrieve data from tstore. */ + TupleTableSlot *tsslot; + + ResourceOwner resowner; +} RepackDecodingState; + extern void ExecRepack(ParseState *pstate, RepackStmt *stmt, bool isTopLevel); extern void cluster_rel(RepackCommand command, Relation OldHeap, Oid indexOid, - ClusterParams *params); + ClusterParams *params, bool isTopLevel); extern void check_index_is_clusterable(Relation OldHeap, Oid indexOid, LOCKMODE lockmode); extern void mark_index_clustered(Relation rel, Oid indexOid, bool is_internal); +extern void repack_decode_concurrent_changes(LogicalDecodingContext *ctx, + XLogRecPtr end_of_wal); + extern Oid make_new_heap(Oid OIDOldHeap, Oid NewTableSpace, Oid NewAccessMethod, char relpersistence, LOCKMODE lockmode); extern void finish_heap_swap(Oid OIDOldHeap, Oid OIDNewHeap, @@ -48,6 +135,7 @@ extern void finish_heap_swap(Oid OIDOldHeap, Oid OIDNewHeap, bool swap_toast_by_content, bool check_constraints, bool is_internal, + bool reindex, TransactionId frozenXid, MultiXactId cutoffMulti, char newrelpersistence); diff --git a/src/include/commands/progress.h b/src/include/commands/progress.h index ebf004b7aa5..5024fea5e2e 100644 --- a/src/include/commands/progress.h +++ b/src/include/commands/progress.h @@ -69,10 +69,12 @@ #define PROGRESS_REPACK_PHASE 1 #define PROGRESS_REPACK_INDEX_RELID 2 #define PROGRESS_REPACK_HEAP_TUPLES_SCANNED 3 -#define PROGRESS_REPACK_HEAP_TUPLES_WRITTEN 4 -#define PROGRESS_REPACK_TOTAL_HEAP_BLKS 5 -#define PROGRESS_REPACK_HEAP_BLKS_SCANNED 6 -#define PROGRESS_REPACK_INDEX_REBUILD_COUNT 7 +#define PROGRESS_REPACK_HEAP_TUPLES_INSERTED 4 +#define PROGRESS_REPACK_HEAP_TUPLES_UPDATED 5 +#define PROGRESS_REPACK_HEAP_TUPLES_DELETED 6 +#define PROGRESS_REPACK_TOTAL_HEAP_BLKS 7 +#define PROGRESS_REPACK_HEAP_BLKS_SCANNED 8 +#define PROGRESS_REPACK_INDEX_REBUILD_COUNT 9 /* * Phases of repack (as advertised via PROGRESS_REPACK_PHASE). @@ -81,9 +83,10 @@ #define PROGRESS_REPACK_PHASE_INDEX_SCAN_HEAP 2 #define PROGRESS_REPACK_PHASE_SORT_TUPLES 3 #define PROGRESS_REPACK_PHASE_WRITE_NEW_HEAP 4 -#define PROGRESS_REPACK_PHASE_SWAP_REL_FILES 5 -#define PROGRESS_REPACK_PHASE_REBUILD_INDEX 6 -#define PROGRESS_REPACK_PHASE_FINAL_CLEANUP 7 +#define PROGRESS_REPACK_PHASE_CATCH_UP 5 +#define PROGRESS_REPACK_PHASE_SWAP_REL_FILES 6 +#define PROGRESS_REPACK_PHASE_REBUILD_INDEX 7 +#define PROGRESS_REPACK_PHASE_FINAL_CLEANUP 8 /* Progress parameters for CREATE INDEX */ /* 3, 4 and 5 reserved for "waitfor" metrics */ diff --git a/src/include/replication/snapbuild.h b/src/include/replication/snapbuild.h index 6d4d2d1814c..802fc4b0823 100644 --- a/src/include/replication/snapbuild.h +++ b/src/include/replication/snapbuild.h @@ -73,6 +73,7 @@ extern void FreeSnapshotBuilder(SnapBuild *builder); extern void SnapBuildSnapDecRefcount(Snapshot snap); extern Snapshot SnapBuildInitialSnapshot(SnapBuild *builder); +extern Snapshot SnapBuildInitialSnapshotForRepack(SnapBuild *builder); extern Snapshot SnapBuildMVCCFromHistoric(Snapshot snapshot, bool in_place); extern const char *SnapBuildExportSnapshot(SnapBuild *builder); extern void SnapBuildClearExportedSnapshot(void); diff --git a/src/include/storage/lockdefs.h b/src/include/storage/lockdefs.h index 7f3ba0352f6..2739327b0da 100644 --- a/src/include/storage/lockdefs.h +++ b/src/include/storage/lockdefs.h @@ -36,8 +36,8 @@ typedef int LOCKMODE; #define AccessShareLock 1 /* SELECT */ #define RowShareLock 2 /* SELECT FOR UPDATE/FOR SHARE */ #define RowExclusiveLock 3 /* INSERT, UPDATE, DELETE */ -#define ShareUpdateExclusiveLock 4 /* VACUUM (non-FULL), ANALYZE, CREATE - * INDEX CONCURRENTLY */ +#define ShareUpdateExclusiveLock 4 /* VACUUM (non-exclusive), ANALYZE, CREATE + * INDEX CONCURRENTLY, REPACK CONCURRENTLY */ #define ShareLock 5 /* CREATE INDEX (WITHOUT CONCURRENTLY) */ #define ShareRowExclusiveLock 6 /* like EXCLUSIVE MODE, but allows ROW * SHARE */ diff --git a/src/include/utils/snapmgr.h b/src/include/utils/snapmgr.h index f65f83c85cd..1f821fd2ccd 100644 --- a/src/include/utils/snapmgr.h +++ b/src/include/utils/snapmgr.h @@ -64,6 +64,8 @@ extern Snapshot GetLatestSnapshot(void); extern void SnapshotSetCommandId(CommandId curcid); extern Snapshot CopySnapshot(Snapshot snapshot); +extern void FreeSnapshot(Snapshot snapshot); + extern Snapshot GetCatalogSnapshot(Oid relid); extern Snapshot GetNonHistoricCatalogSnapshot(Oid relid); extern void InvalidateCatalogSnapshot(void); diff --git a/src/test/modules/injection_points/Makefile b/src/test/modules/injection_points/Makefile index a618e6a9899..e477a1ba5ff 100644 --- a/src/test/modules/injection_points/Makefile +++ b/src/test/modules/injection_points/Makefile @@ -16,12 +16,14 @@ REGRESS_OPTS = --dlpath=$(top_builddir)/src/test/regress ISOLATION = basic \ inplace \ + repack \ syscache-update-pruned \ index-concurrently-upsert \ index-concurrently-upsert-predicate \ reindex-concurrently-upsert \ reindex-concurrently-upsert-on-constraint \ reindex-concurrently-upsert-partitioned +ISOLATION_OPTS = --temp-config $(top_srcdir)/src/test/modules/injection_points/logical.conf TAP_TESTS = 1 diff --git a/src/test/modules/injection_points/expected/repack.out b/src/test/modules/injection_points/expected/repack.out new file mode 100644 index 00000000000..b575e9052ee --- /dev/null +++ b/src/test/modules/injection_points/expected/repack.out @@ -0,0 +1,113 @@ +Parsed test spec with 2 sessions + +starting permutation: wait_before_lock change_existing change_new change_subxact1 change_subxact2 check2 wakeup_before_lock check1 +injection_points_attach +----------------------- + +(1 row) + +step wait_before_lock: + REPACK (CONCURRENTLY) repack_test USING INDEX repack_test_pkey; + +step change_existing: + UPDATE repack_test SET i=10 where i=1; + UPDATE repack_test SET j=20 where i=2; + UPDATE repack_test SET i=30 where i=3; + UPDATE repack_test SET i=40 where i=30; + DELETE FROM repack_test WHERE i=4; + +step change_new: + INSERT INTO repack_test(i, j) VALUES (5, 5), (6, 6), (7, 7), (8, 8); + UPDATE repack_test SET i=50 where i=5; + UPDATE repack_test SET j=60 where i=6; + DELETE FROM repack_test WHERE i=7; + +step change_subxact1: + BEGIN; + INSERT INTO repack_test(i, j) VALUES (100, 100); + SAVEPOINT s1; + UPDATE repack_test SET i=101 where i=100; + SAVEPOINT s2; + UPDATE repack_test SET i=102 where i=101; + COMMIT; + +step change_subxact2: + BEGIN; + SAVEPOINT s1; + INSERT INTO repack_test(i, j) VALUES (110, 110); + ROLLBACK TO SAVEPOINT s1; + INSERT INTO repack_test(i, j) VALUES (110, 111); + COMMIT; + +step check2: + INSERT INTO relfilenodes(node) + SELECT relfilenode FROM pg_class WHERE relname='repack_test'; + + SELECT i, j FROM repack_test ORDER BY i, j; + + INSERT INTO data_s2(i, j) + SELECT i, j FROM repack_test; + + i| j +---+--- + 2| 20 + 6| 60 + 8| 8 + 10| 1 + 40| 3 + 50| 5 +102|100 +110|111 +(8 rows) + +step wakeup_before_lock: + SELECT injection_points_wakeup('repack-concurrently-before-lock'); + +injection_points_wakeup +----------------------- + +(1 row) + +step wait_before_lock: <... completed> +step check1: + INSERT INTO relfilenodes(node) + SELECT relfilenode FROM pg_class WHERE relname='repack_test'; + + SELECT count(DISTINCT node) FROM relfilenodes; + + SELECT i, j FROM repack_test ORDER BY i, j; + + INSERT INTO data_s1(i, j) + SELECT i, j FROM repack_test; + + SELECT count(*) + FROM data_s1 d1 FULL JOIN data_s2 d2 USING (i, j) + WHERE d1.i ISNULL OR d2.i ISNULL; + +count +----- + 2 +(1 row) + + i| j +---+--- + 2| 20 + 6| 60 + 8| 8 + 10| 1 + 40| 3 + 50| 5 +102|100 +110|111 +(8 rows) + +count +----- + 0 +(1 row) + +injection_points_detach +----------------------- + +(1 row) + diff --git a/src/test/modules/injection_points/logical.conf b/src/test/modules/injection_points/logical.conf new file mode 100644 index 00000000000..c8f264bc6cb --- /dev/null +++ b/src/test/modules/injection_points/logical.conf @@ -0,0 +1 @@ +wal_level = logical \ No newline at end of file diff --git a/src/test/modules/injection_points/meson.build b/src/test/modules/injection_points/meson.build index 1a2af8a26c4..f8957d82afa 100644 --- a/src/test/modules/injection_points/meson.build +++ b/src/test/modules/injection_points/meson.build @@ -47,6 +47,7 @@ tests += { 'specs': [ 'basic', 'inplace', + 'repack', 'syscache-update-pruned', 'index-concurrently-upsert', 'index-concurrently-upsert-predicate', @@ -57,6 +58,8 @@ tests += { 'runningcheck': false, # see syscache-update-pruned # Some tests wait for all snapshots, so avoid parallel execution 'runningcheck-parallel': false, + # 'repack' requires wal_level = 'logical'. + 'regress_args': ['--temp-config', files('logical.conf')], }, 'tap': { 'env': { diff --git a/src/test/modules/injection_points/specs/repack.spec b/src/test/modules/injection_points/specs/repack.spec new file mode 100644 index 00000000000..75850334986 --- /dev/null +++ b/src/test/modules/injection_points/specs/repack.spec @@ -0,0 +1,143 @@ +# Prefix the system columns with underscore as they are not allowed as column +# names. +setup +{ + CREATE EXTENSION injection_points; + + CREATE TABLE repack_test(i int PRIMARY KEY, j int); + INSERT INTO repack_test(i, j) VALUES (1, 1), (2, 2), (3, 3), (4, 4); + + CREATE TABLE relfilenodes(node oid); + + CREATE TABLE data_s1(i int, j int); + CREATE TABLE data_s2(i int, j int); +} + +teardown +{ + DROP TABLE repack_test; + DROP EXTENSION injection_points; + + DROP TABLE relfilenodes; + DROP TABLE data_s1; + DROP TABLE data_s2; +} + +session s1 +setup +{ + SELECT injection_points_set_local(); + SELECT injection_points_attach('repack-concurrently-before-lock', 'wait'); +} +# Perform the initial load and wait for s2 to do some data changes. +step wait_before_lock +{ + REPACK (CONCURRENTLY) repack_test USING INDEX repack_test_pkey; +} +# Check the table from the perspective of s1. +# +# Besides the contents, we also check that relfilenode has changed. + +# Have each session write the contents into a table and use FULL JOIN to check +# if the outputs are identical. +step check1 +{ + INSERT INTO relfilenodes(node) + SELECT relfilenode FROM pg_class WHERE relname='repack_test'; + + SELECT count(DISTINCT node) FROM relfilenodes; + + SELECT i, j FROM repack_test ORDER BY i, j; + + INSERT INTO data_s1(i, j) + SELECT i, j FROM repack_test; + + SELECT count(*) + FROM data_s1 d1 FULL JOIN data_s2 d2 USING (i, j) + WHERE d1.i ISNULL OR d2.i ISNULL; +} +teardown +{ + SELECT injection_points_detach('repack-concurrently-before-lock'); +} + +session s2 +# Change the existing data. UPDATE changes both key and non-key columns. Also +# update one row twice to test whether tuple version generated by this session +# can be found. +step change_existing +{ + UPDATE repack_test SET i=10 where i=1; + UPDATE repack_test SET j=20 where i=2; + UPDATE repack_test SET i=30 where i=3; + UPDATE repack_test SET i=40 where i=30; + DELETE FROM repack_test WHERE i=4; +} +# Insert new rows and UPDATE / DELETE some of them. Again, update both key and +# non-key column. +step change_new +{ + INSERT INTO repack_test(i, j) VALUES (5, 5), (6, 6), (7, 7), (8, 8); + UPDATE repack_test SET i=50 where i=5; + UPDATE repack_test SET j=60 where i=6; + DELETE FROM repack_test WHERE i=7; +} + +# When applying concurrent data changes, we should see the effects of an +# in-progress subtransaction. +# +# XXX Not sure this test is useful now - it was designed for the patch that +# preserves tuple visibility and which therefore modifies +# TransactionIdIsCurrentTransactionId(). +step change_subxact1 +{ + BEGIN; + INSERT INTO repack_test(i, j) VALUES (100, 100); + SAVEPOINT s1; + UPDATE repack_test SET i=101 where i=100; + SAVEPOINT s2; + UPDATE repack_test SET i=102 where i=101; + COMMIT; +} + +# When applying concurrent data changes, we should not see the effects of a +# rolled back subtransaction. +# +# XXX Is this test useful? See above. +step change_subxact2 +{ + BEGIN; + SAVEPOINT s1; + INSERT INTO repack_test(i, j) VALUES (110, 110); + ROLLBACK TO SAVEPOINT s1; + INSERT INTO repack_test(i, j) VALUES (110, 111); + COMMIT; +} + +# Check the table from the perspective of s2. +step check2 +{ + INSERT INTO relfilenodes(node) + SELECT relfilenode FROM pg_class WHERE relname='repack_test'; + + SELECT i, j FROM repack_test ORDER BY i, j; + + INSERT INTO data_s2(i, j) + SELECT i, j FROM repack_test; +} +step wakeup_before_lock +{ + SELECT injection_points_wakeup('repack-concurrently-before-lock'); +} + +# Test if data changes introduced while one session is performing REPACK +# CONCURRENTLY find their way into the table. +permutation + wait_before_lock + change_existing + change_new + change_subxact1 + change_subxact2 + check2 + wakeup_before_lock + check1 diff --git a/src/test/regress/expected/rules.out b/src/test/regress/expected/rules.out index 1c957f12d27..e23751afeaf 100644 --- a/src/test/regress/expected/rules.out +++ b/src/test/regress/expected/rules.out @@ -2007,7 +2007,7 @@ pg_stat_progress_cluster| SELECT pid, phase, repack_index_relid AS cluster_index_relid, heap_tuples_scanned, - heap_tuples_written, + (heap_tuples_inserted + heap_tuples_updated) AS heap_tuples_written, heap_blks_total, heap_blks_scanned, index_rebuild_count @@ -2087,17 +2087,20 @@ pg_stat_progress_repack| SELECT s.pid, WHEN 2 THEN 'index scanning heap'::text WHEN 3 THEN 'sorting tuples'::text WHEN 4 THEN 'writing new heap'::text - WHEN 5 THEN 'swapping relation files'::text - WHEN 6 THEN 'rebuilding index'::text - WHEN 7 THEN 'performing final cleanup'::text + WHEN 5 THEN 'catch-up'::text + WHEN 6 THEN 'swapping relation files'::text + WHEN 7 THEN 'rebuilding index'::text + WHEN 8 THEN 'performing final cleanup'::text ELSE NULL::text END AS phase, (s.param3)::oid AS repack_index_relid, s.param4 AS heap_tuples_scanned, - s.param5 AS heap_tuples_written, - s.param6 AS heap_blks_total, - s.param7 AS heap_blks_scanned, - s.param8 AS index_rebuild_count + s.param5 AS heap_tuples_inserted, + s.param6 AS heap_tuples_updated, + s.param7 AS heap_tuples_deleted, + s.param8 AS heap_blks_total, + s.param9 AS heap_blks_scanned, + s.param10 AS index_rebuild_count FROM (pg_stat_get_progress_info('REPACK'::text) s(pid, datid, relid, param1, param2, param3, param4, param5, param6, param7, param8, param9, param10, param11, param12, param13, param14, param15, param16, param17, param18, param19, param20) LEFT JOIN pg_database d ON ((s.datid = d.oid))); pg_stat_progress_vacuum| SELECT s.pid, diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list index 4641da9b746..5e546160a2f 100644 --- a/src/tools/pgindent/typedefs.list +++ b/src/tools/pgindent/typedefs.list @@ -411,6 +411,7 @@ CatCacheHeader CatalogId CatalogIdMapEntry CatalogIndexState +ChangeDest ChangeVarNodes_callback ChangeVarNodes_context CheckPoint @@ -487,6 +488,8 @@ CompressFileHandle CompressionLocation CompressorState ComputeXidHorizonsResult +ConcurrentChange +ConcurrentChangeKind ConditionVariable ConditionVariableMinimallyPadded ConditionalStack @@ -1264,6 +1267,7 @@ IndexElem IndexFetchHeapData IndexFetchTableData IndexInfo +IndexInsertState IndexList IndexOnlyScan IndexOnlyScanState @@ -2556,6 +2560,7 @@ ReorderBufferUpdateProgressTxnCB ReorderTuple RepOriginId RepackCommand +RepackDecodingState RepackStmt ReparameterizeForeignPathByChild_function ReplaceVarsFromTargetList_context -- 2.47.3 --=-=-=--