/*------------------------------------------------------------------------- * * pg_amcheck.c * Detects corruption within database relations. * * Copyright (c) 2017-2025, PostgreSQL Global Development Group * * IDENTIFICATION * src/bin/pg_amcheck/pg_amcheck.c * *------------------------------------------------------------------------- */ #include "postgres_fe.h" #include #include #include "catalog/pg_am_d.h" #include "catalog/pg_class_d.h" #include "catalog/pg_namespace_d.h" #include "common/logging.h" #include "common/username.h" #include "fe_utils/cancel.h" #include "fe_utils/option_utils.h" #include "fe_utils/parallel_slot.h" #include "fe_utils/query_utils.h" #include "fe_utils/simple_list.h" #include "fe_utils/string_utils.h" #include "getopt_long.h" #include "pgtime.h" #include "storage/block.h" typedef struct PatternInfo { const char *pattern; /* Unaltered pattern from the command line */ char *db_regex; /* Database regexp parsed from pattern, or * NULL */ char *nsp_regex; /* Schema regexp parsed from pattern, or NULL */ char *rel_regex; /* Relation regexp parsed from pattern, or * NULL */ bool heap_only; /* true if rel_regex should only match heap * tables */ bool btree_only; /* true if rel_regex should only match btree * indexes */ bool matched; /* true if the pattern matched in any database */ } PatternInfo; typedef struct PatternInfoArray { PatternInfo *data; size_t len; } PatternInfoArray; /* pg_amcheck command line options controlled by user flags */ typedef struct AmcheckOptions { bool dbpattern; bool alldb; bool echo; bool verbose; bool strict_names; bool show_progress; int jobs; /* * Whether to install missing extensions, and optionally the name of the * schema in which to install the extension's objects. */ bool install_missing; char *install_schema; /* Objects to check or not to check, as lists of PatternInfo structs. */ PatternInfoArray include; PatternInfoArray exclude; /* * As an optimization, if any pattern in the exclude list applies to heap * tables, or similarly if any such pattern applies to btree indexes, or * to schemas, then these will be true, otherwise false. These should * always agree with what you'd conclude by grep'ing through the exclude * list. */ bool excludetbl; bool excludeidx; bool excludensp; /* * If any inclusion pattern exists, then we should only be checking * matching relations rather than all relations, so this is true iff * include is empty. */ bool allrel; /* heap table checking options */ bool no_toast_expansion; bool reconcile_toast; bool on_error_stop; int64 startblock; int64 endblock; const char *skip; /* btree index checking options */ bool parent_check; bool rootdescend; bool heapallindexed; bool checkunique; /* heap and btree hybrid option */ bool no_btree_expansion; } AmcheckOptions; static AmcheckOptions opts = { .dbpattern = false, .alldb = false, .echo = false, .verbose = false, .strict_names = true, .show_progress = false, .jobs = 1, .install_missing = false, .install_schema = "pg_catalog", .include = {NULL, 0}, .exclude = {NULL, 0}, .excludetbl = false, .excludeidx = false, .excludensp = false, .allrel = true, .no_toast_expansion = false, .reconcile_toast = true, .on_error_stop = false, .startblock = -1, .endblock = -1, .skip = "none", .parent_check = false, .rootdescend = false, .heapallindexed = false, .checkunique = false, .no_btree_expansion = false }; static const char *progname = NULL; /* Whether all relations have so far passed their corruption checks */ static bool all_checks_pass = true; /* Time last progress report was displayed */ static pg_time_t last_progress_report = 0; static bool progress_since_last_stderr = false; typedef struct DatabaseInfo { char *datname; char *amcheck_schema; /* escaped, quoted literal */ bool is_checkunique; } DatabaseInfo; typedef struct RelationInfo { const DatabaseInfo *datinfo; /* shared by other relinfos */ Oid reloid; bool is_heap; /* true if heap, false if btree */ char *nspname; char *relname; int relpages; int blocks_to_check; char *sql; /* set during query run, pg_free'd after */ } RelationInfo; /* * Query for determining if contrib's amcheck is installed. If so, selects the * namespace name where amcheck's functions can be found. */ static const char *const amcheck_sql = "SELECT n.nspname, x.extversion FROM pg_catalog.pg_extension x" "\nJOIN pg_catalog.pg_namespace n ON x.extnamespace = n.oid" "\nWHERE x.extname = 'amcheck'"; static void prepare_heap_command(PQExpBuffer sql, RelationInfo *rel, PGconn *conn); static void prepare_btree_command(PQExpBuffer sql, RelationInfo *rel, PGconn *conn); static void run_command(ParallelSlot *slot, const char *sql); static bool verify_heap_slot_handler(PGresult *res, PGconn *conn, void *context); static bool verify_btree_slot_handler(PGresult *res, PGconn *conn, void *context); static void help(const char *progname); static void progress_report(uint64 relations_total, uint64 relations_checked, uint64 relpages_total, uint64 relpages_checked, const char *datname, bool force, bool finished); static void append_database_pattern(PatternInfoArray *pia, const char *pattern, int encoding); static void append_schema_pattern(PatternInfoArray *pia, const char *pattern, int encoding); static void append_relation_pattern(PatternInfoArray *pia, const char *pattern, int encoding); static void append_heap_pattern(PatternInfoArray *pia, const char *pattern, int encoding); static void append_btree_pattern(PatternInfoArray *pia, const char *pattern, int encoding); static void compile_database_list(PGconn *conn, SimplePtrList *databases, const char *initial_dbname); static void compile_relation_list_one_db(PGconn *conn, SimplePtrList *relations, const DatabaseInfo *dat, uint64 *pagecount); #define log_no_match(...) do { \ if (opts.strict_names) \ pg_log_error(__VA_ARGS__); \ else \ pg_log_warning(__VA_ARGS__); \ } while(0) #define FREE_AND_SET_NULL(x) do { \ pg_free(x); \ (x) = NULL; \ } while (0) int main(int argc, char *argv[]) { PGconn *conn = NULL; SimplePtrListCell *cell; SimplePtrList databases = {NULL, NULL}; SimplePtrList relations = {NULL, NULL}; bool failed = false; const char *latest_datname; int parallel_workers; ParallelSlotArray *sa; PQExpBufferData sql; uint64 reltotal = 0; uint64 pageschecked = 0; uint64 pagestotal = 0; uint64 relprogress = 0; int pattern_id; static struct option long_options[] = { /* Connection options */ {"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'}, {"maintenance-db", required_argument, NULL, 1}, /* check options */ {"all", no_argument, NULL, 'a'}, {"database", required_argument, NULL, 'd'}, {"exclude-database", required_argument, NULL, 'D'}, {"echo", no_argument, NULL, 'e'}, {"index", required_argument, NULL, 'i'}, {"exclude-index", required_argument, NULL, 'I'}, {"jobs", required_argument, NULL, 'j'}, {"progress", no_argument, NULL, 'P'}, {"relation", required_argument, NULL, 'r'}, {"exclude-relation", required_argument, NULL, 'R'}, {"schema", required_argument, NULL, 's'}, {"exclude-schema", required_argument, NULL, 'S'}, {"table", required_argument, NULL, 't'}, {"exclude-table", required_argument, NULL, 'T'}, {"verbose", no_argument, NULL, 'v'}, {"no-dependent-indexes", no_argument, NULL, 2}, {"no-dependent-toast", no_argument, NULL, 3}, {"exclude-toast-pointers", no_argument, NULL, 4}, {"on-error-stop", no_argument, NULL, 5}, {"skip", required_argument, NULL, 6}, {"startblock", required_argument, NULL, 7}, {"endblock", required_argument, NULL, 8}, {"rootdescend", no_argument, NULL, 9}, {"no-strict-names", no_argument, NULL, 10}, {"heapallindexed", no_argument, NULL, 11}, {"parent-check", no_argument, NULL, 12}, {"install-missing", optional_argument, NULL, 13}, {"checkunique", no_argument, NULL, 14}, {NULL, 0, NULL, 0} }; int optindex; int c; const char *db = NULL; const char *maintenance_db = NULL; const char *host = NULL; const char *port = NULL; const char *username = NULL; enum trivalue prompt_password = TRI_DEFAULT; int encoding = pg_get_encoding_from_locale(NULL, false); ConnParams cparams; pg_logging_init(argv[0]); progname = get_progname(argv[0]); set_pglocale_pgservice(argv[0], PG_TEXTDOMAIN("pg_amcheck")); handle_help_version_opts(argc, argv, progname, help); /* process command-line options */ while ((c = getopt_long(argc, argv, "ad:D:eh:Hi:I:j:p:Pr:R:s:S:t:T:U:vwW", long_options, &optindex)) != -1) { char *endptr; unsigned long optval; switch (c) { case 'a': opts.alldb = true; break; case 'd': opts.dbpattern = true; append_database_pattern(&opts.include, optarg, encoding); break; case 'D': opts.dbpattern = true; append_database_pattern(&opts.exclude, optarg, encoding); break; case 'e': opts.echo = true; break; case 'h': host = pg_strdup(optarg); break; case 'i': opts.allrel = false; append_btree_pattern(&opts.include, optarg, encoding); break; case 'I': opts.excludeidx = true; append_btree_pattern(&opts.exclude, optarg, encoding); break; case 'j': if (!option_parse_int(optarg, "-j/--jobs", 1, INT_MAX, &opts.jobs)) exit(1); break; case 'p': port = pg_strdup(optarg); break; case 'P': opts.show_progress = true; break; case 'r': opts.allrel = false; append_relation_pattern(&opts.include, optarg, encoding); break; case 'R': opts.excludeidx = true; opts.excludetbl = true; append_relation_pattern(&opts.exclude, optarg, encoding); break; case 's': opts.allrel = false; append_schema_pattern(&opts.include, optarg, encoding); break; case 'S': opts.excludensp = true; append_schema_pattern(&opts.exclude, optarg, encoding); break; case 't': opts.allrel = false; append_heap_pattern(&opts.include, optarg, encoding); break; case 'T': opts.excludetbl = true; append_heap_pattern(&opts.exclude, optarg, encoding); break; case 'U': username = pg_strdup(optarg); break; case 'v': opts.verbose = true; pg_logging_increase_verbosity(); break; case 'w': prompt_password = TRI_NO; break; case 'W': prompt_password = TRI_YES; break; case 1: maintenance_db = pg_strdup(optarg); break; case 2: opts.no_btree_expansion = true; break; case 3: opts.no_toast_expansion = true; break; case 4: opts.reconcile_toast = false; break; case 5: opts.on_error_stop = true; break; case 6: if (pg_strcasecmp(optarg, "all-visible") == 0) opts.skip = "all-visible"; else if (pg_strcasecmp(optarg, "all-frozen") == 0) opts.skip = "all-frozen"; else if (pg_strcasecmp(optarg, "none") == 0) opts.skip = "none"; else pg_fatal("invalid argument for option %s", "--skip"); break; case 7: errno = 0; optval = strtoul(optarg, &endptr, 10); if (endptr == optarg || *endptr != '\0' || errno != 0) pg_fatal("invalid start block"); if (optval > MaxBlockNumber) pg_fatal("start block out of bounds"); opts.startblock = optval; break; case 8: errno = 0; optval = strtoul(optarg, &endptr, 10); if (endptr == optarg || *endptr != '\0' || errno != 0) pg_fatal("invalid end block"); if (optval > MaxBlockNumber) pg_fatal("end block out of bounds"); opts.endblock = optval; break; case 9: opts.rootdescend = true; opts.parent_check = true; break; case 10: opts.strict_names = false; break; case 11: opts.heapallindexed = true; break; case 12: opts.parent_check = true; break; case 13: opts.install_missing = true; if (optarg) opts.install_schema = pg_strdup(optarg); break; case 14: opts.checkunique = true; break; default: /* getopt_long already emitted a complaint */ pg_log_error_hint("Try \"%s --help\" for more information.", progname); exit(1); } } if (opts.endblock >= 0 && opts.endblock < opts.startblock) pg_fatal("end block precedes start block"); /* * A single non-option arguments specifies a database name or connection * string. */ if (optind < argc) { db = 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); } /* fill cparams except for dbname, which is set below */ cparams.pghost = host; cparams.pgport = port; cparams.pguser = username; cparams.prompt_password = prompt_password; cparams.dbname = NULL; cparams.override_dbname = NULL; setup_cancel_handler(NULL); /* choose the database for our initial connection */ if (opts.alldb) { if (db != NULL) pg_fatal("cannot specify a database name with --all"); cparams.dbname = maintenance_db; } else if (db != NULL) { if (opts.dbpattern) pg_fatal("cannot specify both a database name and database patterns"); cparams.dbname = db; } if (opts.alldb || opts.dbpattern) { conn = connectMaintenanceDatabase(&cparams, progname, opts.echo); compile_database_list(conn, &databases, NULL); } else { if (cparams.dbname == NULL) { if (getenv("PGDATABASE")) cparams.dbname = getenv("PGDATABASE"); else if (getenv("PGUSER")) cparams.dbname = getenv("PGUSER"); else cparams.dbname = get_user_name_or_exit(progname); } conn = connectDatabase(&cparams, progname, opts.echo, false, true); compile_database_list(conn, &databases, PQdb(conn)); } if (databases.head == NULL) { if (conn != NULL) disconnectDatabase(conn); pg_log_warning("no databases to check"); exit(0); } /* * Compile a list of all relations spanning all databases to be checked. */ for (cell = databases.head; cell; cell = cell->next) { PGresult *result; int ntups; const char *amcheck_schema = NULL; DatabaseInfo *dat = (DatabaseInfo *) cell->ptr; cparams.override_dbname = dat->datname; if (conn == NULL || strcmp(PQdb(conn), dat->datname) != 0) { if (conn != NULL) disconnectDatabase(conn); conn = connectDatabase(&cparams, progname, opts.echo, false, true); } /* * Optionally install amcheck if not already installed in this * database. */ if (opts.install_missing) { char *schema; char *install_sql; /* * Must re-escape the schema name for each database, as the * escaping rules may change. */ schema = PQescapeIdentifier(conn, opts.install_schema, strlen(opts.install_schema)); install_sql = psprintf("CREATE EXTENSION IF NOT EXISTS amcheck WITH SCHEMA %s", schema); executeCommand(conn, install_sql, opts.echo); pfree(install_sql); PQfreemem(schema); } /* * Verify that amcheck is installed for this next database. User * error could result in a database not having amcheck that should * have it, but we also could be iterating over multiple databases * where not all of them have amcheck installed (for example, * 'template1'). */ result = executeQuery(conn, amcheck_sql, opts.echo); if (PQresultStatus(result) != PGRES_TUPLES_OK) { /* Querying the catalog failed. */ pg_log_error("database \"%s\": %s", PQdb(conn), PQerrorMessage(conn)); pg_log_error_detail("Query was: %s", amcheck_sql); PQclear(result); disconnectDatabase(conn); exit(1); } ntups = PQntuples(result); if (ntups == 0) { /* Querying the catalog succeeded, but amcheck is missing. */ pg_log_warning("skipping database \"%s\": amcheck is not installed", PQdb(conn)); PQclear(result); disconnectDatabase(conn); conn = NULL; continue; } amcheck_schema = PQgetvalue(result, 0, 0); if (opts.verbose) pg_log_info("in database \"%s\": using amcheck version \"%s\" in schema \"%s\"", PQdb(conn), PQgetvalue(result, 0, 1), amcheck_schema); dat->amcheck_schema = PQescapeIdentifier(conn, amcheck_schema, strlen(amcheck_schema)); /* * Check the version of amcheck extension. Skip requested unique * constraint check with warning if it is not yet supported by * amcheck. */ if (opts.checkunique == true) { /* * Now amcheck has only major and minor versions in the string but * we also support revision just in case. Now it is expected to be * zero. */ int vmaj = 0, vmin = 0, vrev = 0; const char *amcheck_version = PQgetvalue(result, 0, 1); sscanf(amcheck_version, "%d.%d.%d", &vmaj, &vmin, &vrev); /* * checkunique option is supported in amcheck since version 1.4 */ if ((vmaj == 1 && vmin < 4) || vmaj == 0) { pg_log_warning("option %s is not supported by amcheck version %s", "--checkunique", amcheck_version); dat->is_checkunique = false; } else dat->is_checkunique = true; } PQclear(result); compile_relation_list_one_db(conn, &relations, dat, &pagestotal); } /* * Check that all inclusion patterns matched at least one schema or * relation that we can check. */ for (pattern_id = 0; pattern_id < opts.include.len; pattern_id++) { PatternInfo *pat = &opts.include.data[pattern_id]; if (!pat->matched && (pat->nsp_regex != NULL || pat->rel_regex != NULL)) { failed = opts.strict_names; if (pat->heap_only) log_no_match("no heap tables to check matching \"%s\"", pat->pattern); else if (pat->btree_only) log_no_match("no btree indexes to check matching \"%s\"", pat->pattern); else if (pat->rel_regex == NULL) log_no_match("no relations to check in schemas matching \"%s\"", pat->pattern); else log_no_match("no relations to check matching \"%s\"", pat->pattern); } } if (failed) { if (conn != NULL) disconnectDatabase(conn); exit(1); } /* * Set parallel_workers to the lesser of opts.jobs and the number of * relations. */ parallel_workers = 0; for (cell = relations.head; cell; cell = cell->next) { reltotal++; if (parallel_workers < opts.jobs) parallel_workers++; } if (reltotal == 0) { if (conn != NULL) disconnectDatabase(conn); pg_fatal("no relations to check"); } progress_report(reltotal, relprogress, pagestotal, pageschecked, NULL, true, false); /* * Main event loop. * * We use server-side parallelism to check up to parallel_workers * relations in parallel. The list of relations was computed in database * order, which minimizes the number of connects and disconnects as we * process the list. */ latest_datname = NULL; sa = ParallelSlotsSetup(parallel_workers, &cparams, progname, opts.echo, NULL); if (conn != NULL) { ParallelSlotsAdoptConn(sa, conn); conn = NULL; } initPQExpBuffer(&sql); for (relprogress = 0, cell = relations.head; cell; cell = cell->next) { ParallelSlot *free_slot; RelationInfo *rel; rel = (RelationInfo *) cell->ptr; if (CancelRequested) { failed = true; break; } /* * The list of relations is in database sorted order. If this next * relation is in a different database than the last one seen, we are * about to start checking this database. Note that other slots may * still be working on relations from prior databases. */ latest_datname = rel->datinfo->datname; progress_report(reltotal, relprogress, pagestotal, pageschecked, latest_datname, false, false); relprogress++; pageschecked += rel->blocks_to_check; /* * Get a parallel slot for the next amcheck command, blocking if * necessary until one is available, or until a previously issued slot * command fails, indicating that we should abort checking the * remaining objects. */ free_slot = ParallelSlotsGetIdle(sa, rel->datinfo->datname); if (!free_slot) { /* * Something failed. We don't need to know what it was, because * the handler should already have emitted the necessary error * messages. */ failed = true; break; } if (opts.verbose) PQsetErrorVerbosity(free_slot->connection, PQERRORS_VERBOSE); /* * Execute the appropriate amcheck command for this relation using our * slot's database connection. We do not wait for the command to * complete, nor do we perform any error checking, as that is done by * the parallel slots and our handler callback functions. */ if (rel->is_heap) { if (opts.verbose) { if (opts.show_progress && progress_since_last_stderr) fprintf(stderr, "\n"); pg_log_info("checking heap table \"%s.%s.%s\"", rel->datinfo->datname, rel->nspname, rel->relname); progress_since_last_stderr = false; } prepare_heap_command(&sql, rel, free_slot->connection); rel->sql = pstrdup(sql.data); /* pg_free'd after command */ ParallelSlotSetHandler(free_slot, verify_heap_slot_handler, rel); run_command(free_slot, rel->sql); } else { if (opts.verbose) { if (opts.show_progress && progress_since_last_stderr) fprintf(stderr, "\n"); pg_log_info("checking btree index \"%s.%s.%s\"", rel->datinfo->datname, rel->nspname, rel->relname); progress_since_last_stderr = false; } prepare_btree_command(&sql, rel, free_slot->connection); rel->sql = pstrdup(sql.data); /* pg_free'd after command */ ParallelSlotSetHandler(free_slot, verify_btree_slot_handler, rel); run_command(free_slot, rel->sql); } } termPQExpBuffer(&sql); if (!failed) { /* * Wait for all slots to complete, or for one to indicate that an * error occurred. Like above, we rely on the handler emitting the * necessary error messages. */ if (sa && !ParallelSlotsWaitCompletion(sa)) failed = true; progress_report(reltotal, relprogress, pagestotal, pageschecked, NULL, true, true); } if (sa) { ParallelSlotsTerminate(sa); FREE_AND_SET_NULL(sa); } if (failed) exit(1); if (!all_checks_pass) exit(2); } /* * prepare_heap_command * * Creates a SQL command for running amcheck checking on the given heap * relation. The command is phrased as a SQL query, with column order and * names matching the expectations of verify_heap_slot_handler, which will * receive and handle each row returned from the verify_heapam() function. * * The constructed SQL command will silently skip temporary tables, as checking * them would needlessly draw errors from the underlying amcheck function. * * sql: buffer into which the heap table checking command will be written * rel: relation information for the heap table to be checked * conn: the connection to be used, for string escaping purposes */ static void prepare_heap_command(PQExpBuffer sql, RelationInfo *rel, PGconn *conn) { resetPQExpBuffer(sql); appendPQExpBuffer(sql, "SELECT v.blkno, v.offnum, v.attnum, v.msg " "FROM pg_catalog.pg_class c, %s.verify_heapam(" "\nrelation := c.oid, on_error_stop := %s, check_toast := %s, skip := '%s'", rel->datinfo->amcheck_schema, opts.on_error_stop ? "true" : "false", opts.reconcile_toast ? "true" : "false", opts.skip); if (opts.startblock >= 0) appendPQExpBuffer(sql, ", startblock := " INT64_FORMAT, opts.startblock); if (opts.endblock >= 0) appendPQExpBuffer(sql, ", endblock := " INT64_FORMAT, opts.endblock); appendPQExpBuffer(sql, "\n) v WHERE c.oid = %u " "AND c.relpersistence != " CppAsString2(RELPERSISTENCE_TEMP), rel->reloid); } /* * prepare_btree_command * * Creates a SQL command for running amcheck checking on the given btree index * relation. The command does not select any columns, as btree checking * functions do not return any, but rather return corruption information by * raising errors, which verify_btree_slot_handler expects. * * The constructed SQL command will silently skip temporary indexes, and * indexes being reindexed concurrently, as checking them would needlessly draw * errors from the underlying amcheck functions. * * sql: buffer into which the heap table checking command will be written * rel: relation information for the index to be checked * conn: the connection to be used, for string escaping purposes */ static void prepare_btree_command(PQExpBuffer sql, RelationInfo *rel, PGconn *conn) { resetPQExpBuffer(sql); if (opts.parent_check) appendPQExpBuffer(sql, "SELECT %s.bt_index_parent_check(" "index := c.oid, heapallindexed := %s, rootdescend := %s " "%s)" "\nFROM pg_catalog.pg_class c, pg_catalog.pg_index i " "WHERE c.oid = %u " "AND c.oid = i.indexrelid " "AND c.relpersistence != " CppAsString2(RELPERSISTENCE_TEMP) " " "AND i.indisready AND i.indisvalid AND i.indislive", rel->datinfo->amcheck_schema, (opts.heapallindexed ? "true" : "false"), (opts.rootdescend ? "true" : "false"), (rel->datinfo->is_checkunique ? ", checkunique := true" : ""), rel->reloid); else appendPQExpBuffer(sql, "SELECT %s.bt_index_check(" "index := c.oid, heapallindexed := %s " "%s)" "\nFROM pg_catalog.pg_class c, pg_catalog.pg_index i " "WHERE c.oid = %u " "AND c.oid = i.indexrelid " "AND c.relpersistence != " CppAsString2(RELPERSISTENCE_TEMP) " " "AND i.indisready AND i.indisvalid AND i.indislive", rel->datinfo->amcheck_schema, (opts.heapallindexed ? "true" : "false"), (rel->datinfo->is_checkunique ? ", checkunique := true" : ""), rel->reloid); } /* * run_command * * Sends a command to the server without waiting for the command to complete. * Logs an error if the command cannot be sent, but otherwise any errors are * expected to be handled by a ParallelSlotHandler. * * If reconnecting to the database is necessary, the cparams argument may be * modified. * * slot: slot with connection to the server we should use for the command * sql: query to send */ static void run_command(ParallelSlot *slot, const char *sql) { if (opts.echo) printf("%s\n", sql); if (PQsendQuery(slot->connection, sql) == 0) { pg_log_error("error sending command to database \"%s\": %s", PQdb(slot->connection), PQerrorMessage(slot->connection)); pg_log_error_detail("Command was: %s", sql); exit(1); } } /* * should_processing_continue * * Checks a query result returned from a query (presumably issued on a slot's * connection) to determine if parallel slots should continue issuing further * commands. * * Note: Heap relation corruption is reported by verify_heapam() via the result * set, rather than an ERROR, but running verify_heapam() on a corrupted heap * table may still result in an error being returned from the server due to * missing relation files, bad checksums, etc. The btree corruption checking * functions always use errors to communicate corruption messages. We can't * just abort processing because we got a mere ERROR. * * res: result from an executed sql query */ static bool should_processing_continue(PGresult *res) { const char *severity; switch (PQresultStatus(res)) { /* These are expected and ok */ case PGRES_COMMAND_OK: case PGRES_TUPLES_OK: case PGRES_NONFATAL_ERROR: break; /* This is expected but requires closer scrutiny */ case PGRES_FATAL_ERROR: severity = PQresultErrorField(res, PG_DIAG_SEVERITY_NONLOCALIZED); if (severity == NULL) return false; /* libpq failure, probably lost connection */ if (strcmp(severity, "FATAL") == 0) return false; if (strcmp(severity, "PANIC") == 0) return false; break; /* These are unexpected */ case PGRES_BAD_RESPONSE: case PGRES_EMPTY_QUERY: case PGRES_COPY_OUT: case PGRES_COPY_IN: case PGRES_COPY_BOTH: case PGRES_SINGLE_TUPLE: case PGRES_PIPELINE_SYNC: case PGRES_PIPELINE_ABORTED: case PGRES_TUPLES_CHUNK: return false; } return true; } /* * Returns a copy of the argument string with all lines indented four spaces. * * The caller should pg_free the result when finished with it. */ static char * indent_lines(const char *str) { PQExpBufferData buf; const char *c; char *result; initPQExpBuffer(&buf); appendPQExpBufferStr(&buf, " "); for (c = str; *c; c++) { appendPQExpBufferChar(&buf, *c); if (c[0] == '\n' && c[1] != '\0') appendPQExpBufferStr(&buf, " "); } result = pstrdup(buf.data); termPQExpBuffer(&buf); return result; } /* * verify_heap_slot_handler * * ParallelSlotHandler that receives results from a heap table checking command * created by prepare_heap_command and outputs the results for the user. * * res: result from an executed sql query * conn: connection on which the sql query was executed * context: the sql query being handled, as a cstring */ static bool verify_heap_slot_handler(PGresult *res, PGconn *conn, void *context) { RelationInfo *rel = (RelationInfo *) context; if (PQresultStatus(res) == PGRES_TUPLES_OK) { int i; int ntups = PQntuples(res); if (ntups > 0) all_checks_pass = false; for (i = 0; i < ntups; i++) { const char *msg; /* The message string should never be null, but check */ if (PQgetisnull(res, i, 3)) msg = "NO MESSAGE"; else msg = PQgetvalue(res, i, 3); if (!PQgetisnull(res, i, 2)) printf(_("heap table \"%s.%s.%s\", block %s, offset %s, attribute %s:\n"), rel->datinfo->datname, rel->nspname, rel->relname, PQgetvalue(res, i, 0), /* blkno */ PQgetvalue(res, i, 1), /* offnum */ PQgetvalue(res, i, 2)); /* attnum */ else if (!PQgetisnull(res, i, 1)) printf(_("heap table \"%s.%s.%s\", block %s, offset %s:\n"), rel->datinfo->datname, rel->nspname, rel->relname, PQgetvalue(res, i, 0), /* blkno */ PQgetvalue(res, i, 1)); /* offnum */ else if (!PQgetisnull(res, i, 0)) printf(_("heap table \"%s.%s.%s\", block %s:\n"), rel->datinfo->datname, rel->nspname, rel->relname, PQgetvalue(res, i, 0)); /* blkno */ else printf(_("heap table \"%s.%s.%s\":\n"), rel->datinfo->datname, rel->nspname, rel->relname); printf(" %s\n", msg); } } else if (PQresultStatus(res) != PGRES_TUPLES_OK) { char *msg = indent_lines(PQerrorMessage(conn)); all_checks_pass = false; printf(_("heap table \"%s.%s.%s\":\n"), rel->datinfo->datname, rel->nspname, rel->relname); printf("%s", msg); if (opts.verbose) printf(_("query was: %s\n"), rel->sql); FREE_AND_SET_NULL(msg); } FREE_AND_SET_NULL(rel->sql); FREE_AND_SET_NULL(rel->nspname); FREE_AND_SET_NULL(rel->relname); return should_processing_continue(res); } /* * verify_btree_slot_handler * * ParallelSlotHandler that receives results from a btree checking command * created by prepare_btree_command and outputs them for the user. The results * from the btree checking command is assumed to be empty, but when the results * are an error code, the useful information about the corruption is expected * in the connection's error message. * * res: result from an executed sql query * conn: connection on which the sql query was executed * context: unused */ static bool verify_btree_slot_handler(PGresult *res, PGconn *conn, void *context) { RelationInfo *rel = (RelationInfo *) context; if (PQresultStatus(res) == PGRES_TUPLES_OK) { int ntups = PQntuples(res); if (ntups > 1) { /* * We expect the btree checking functions to return one void row * each, or zero rows if the check was skipped due to the object * being in the wrong state to be checked, so we should output * some sort of warning if we get anything more, not because it * indicates corruption, but because it suggests a mismatch * between amcheck and pg_amcheck versions. * * In conjunction with --progress, anything written to stderr at * this time would present strangely to the user without an extra * newline, so we print one. If we were multithreaded, we'd have * to avoid splitting this across multiple calls, but we're in an * event loop, so it doesn't matter. */ if (opts.show_progress && progress_since_last_stderr) fprintf(stderr, "\n"); pg_log_warning("btree index \"%s.%s.%s\": btree checking function returned unexpected number of rows: %d", rel->datinfo->datname, rel->nspname, rel->relname, ntups); if (opts.verbose) pg_log_warning_detail("Query was: %s", rel->sql); pg_log_warning_hint("Are %s's and amcheck's versions compatible?", progname); progress_since_last_stderr = false; } } else { char *msg = indent_lines(PQerrorMessage(conn)); all_checks_pass = false; printf(_("btree index \"%s.%s.%s\":\n"), rel->datinfo->datname, rel->nspname, rel->relname); printf("%s", msg); if (opts.verbose) printf(_("query was: %s\n"), rel->sql); FREE_AND_SET_NULL(msg); } FREE_AND_SET_NULL(rel->sql); FREE_AND_SET_NULL(rel->nspname); FREE_AND_SET_NULL(rel->relname); return should_processing_continue(res); } /* * help * * Prints help page for the program * * progname: the name of the executed program, such as "pg_amcheck" */ static void help(const char *progname) { printf(_("%s checks objects in a PostgreSQL database for corruption.\n\n"), progname); printf(_("Usage:\n")); printf(_(" %s [OPTION]... [DBNAME]\n"), progname); printf(_("\nTarget options:\n")); printf(_(" -a, --all check all databases\n")); printf(_(" -d, --database=PATTERN check matching database(s)\n")); printf(_(" -D, --exclude-database=PATTERN do NOT check matching database(s)\n")); printf(_(" -i, --index=PATTERN check matching index(es)\n")); printf(_(" -I, --exclude-index=PATTERN do NOT check matching index(es)\n")); printf(_(" -r, --relation=PATTERN check matching relation(s)\n")); printf(_(" -R, --exclude-relation=PATTERN do NOT check matching relation(s)\n")); printf(_(" -s, --schema=PATTERN check matching schema(s)\n")); printf(_(" -S, --exclude-schema=PATTERN do NOT check matching schema(s)\n")); printf(_(" -t, --table=PATTERN check matching table(s)\n")); printf(_(" -T, --exclude-table=PATTERN do NOT check matching table(s)\n")); printf(_(" --no-dependent-indexes do NOT expand list of relations to include indexes\n")); printf(_(" --no-dependent-toast do NOT expand list of relations to include TOAST tables\n")); printf(_(" --no-strict-names do NOT require patterns to match objects\n")); printf(_("\nTable checking options:\n")); printf(_(" --exclude-toast-pointers do NOT follow relation TOAST pointers\n")); printf(_(" --on-error-stop stop checking at end of first corrupt page\n")); printf(_(" --skip=OPTION do NOT check \"all-frozen\" or \"all-visible\" blocks\n")); printf(_(" --startblock=BLOCK begin checking table(s) at the given block number\n")); printf(_(" --endblock=BLOCK check table(s) only up to the given block number\n")); printf(_("\nB-tree index checking options:\n")); printf(_(" --checkunique check unique constraint if index is unique\n")); printf(_(" --heapallindexed check that all heap tuples are found within indexes\n")); printf(_(" --parent-check check index parent/child relationships\n")); printf(_(" --rootdescend search from root page to refind tuples\n")); printf(_("\nConnection options:\n")); printf(_(" -h, --host=HOSTNAME database server host or socket directory\n")); printf(_(" -p, --port=PORT database server port\n")); printf(_(" -U, --username=USERNAME 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=DBNAME alternate maintenance database\n")); printf(_("\nOther options:\n")); printf(_(" -e, --echo show the commands being sent to the server\n")); printf(_(" -j, --jobs=NUM use this many concurrent connections to the server\n")); printf(_(" -P, --progress show progress information\n")); printf(_(" -v, --verbose write a lot of output\n")); printf(_(" -V, --version output version information, then exit\n")); printf(_(" --install-missing install missing extensions\n")); printf(_(" -?, --help show this help, then exit\n")); printf(_("\nReport bugs to <%s>.\n"), PACKAGE_BUGREPORT); printf(_("%s home page: <%s>\n"), PACKAGE_NAME, PACKAGE_URL); } /* * Print a progress report based on the global variables. * * Progress report is written at maximum once per second, unless the force * parameter is set to true. * * If finished is set to true, this is the last progress report. The cursor * is moved to the next line. */ static void progress_report(uint64 relations_total, uint64 relations_checked, uint64 relpages_total, uint64 relpages_checked, const char *datname, bool force, bool finished) { int percent_rel = 0; int percent_pages = 0; char checked_rel[32]; char total_rel[32]; char checked_pages[32]; char total_pages[32]; pg_time_t now; if (!opts.show_progress) return; now = time(NULL); if (now == last_progress_report && !force && !finished) return; /* Max once per second */ last_progress_report = now; if (relations_total) percent_rel = (int) (relations_checked * 100 / relations_total); if (relpages_total) percent_pages = (int) (relpages_checked * 100 / relpages_total); snprintf(checked_rel, sizeof(checked_rel), UINT64_FORMAT, relations_checked); snprintf(total_rel, sizeof(total_rel), UINT64_FORMAT, relations_total); snprintf(checked_pages, sizeof(checked_pages), UINT64_FORMAT, relpages_checked); snprintf(total_pages, sizeof(total_pages), UINT64_FORMAT, relpages_total); #define VERBOSE_DATNAME_LENGTH 35 if (opts.verbose) { if (!datname) /* * No datname given, so clear the status line (used for first and * last call) */ fprintf(stderr, _("%*s/%s relations (%d%%), %*s/%s pages (%d%%) %*s"), (int) strlen(total_rel), checked_rel, total_rel, percent_rel, (int) strlen(total_pages), checked_pages, total_pages, percent_pages, VERBOSE_DATNAME_LENGTH + 2, ""); else { bool truncate = (strlen(datname) > VERBOSE_DATNAME_LENGTH); fprintf(stderr, _("%*s/%s relations (%d%%), %*s/%s pages (%d%%) (%s%-*.*s)"), (int) strlen(total_rel), checked_rel, total_rel, percent_rel, (int) strlen(total_pages), checked_pages, total_pages, percent_pages, /* Prefix with "..." if we do leading truncation */ truncate ? "..." : "", truncate ? VERBOSE_DATNAME_LENGTH - 3 : VERBOSE_DATNAME_LENGTH, truncate ? VERBOSE_DATNAME_LENGTH - 3 : VERBOSE_DATNAME_LENGTH, /* Truncate datname at beginning if it's too long */ truncate ? datname + strlen(datname) - VERBOSE_DATNAME_LENGTH + 3 : datname); } } else fprintf(stderr, _("%*s/%s relations (%d%%), %*s/%s pages (%d%%)"), (int) strlen(total_rel), checked_rel, total_rel, percent_rel, (int) strlen(total_pages), checked_pages, total_pages, percent_pages); /* * Stay on the same line if reporting to a terminal and we're not done * yet. */ if (!finished && isatty(fileno(stderr))) { fputc('\r', stderr); progress_since_last_stderr = true; } else fputc('\n', stderr); } /* * Extend the pattern info array to hold one additional initialized pattern * info entry. * * Returns a pointer to the new entry. */ static PatternInfo * extend_pattern_info_array(PatternInfoArray *pia) { PatternInfo *result; pia->len++; pia->data = (PatternInfo *) pg_realloc(pia->data, pia->len * sizeof(PatternInfo)); result = &pia->data[pia->len - 1]; memset(result, 0, sizeof(*result)); return result; } /* * append_database_pattern * * Adds the given pattern interpreted as a database name pattern. * * pia: the pattern info array to be appended * pattern: the database name pattern * encoding: client encoding for parsing the pattern */ static void append_database_pattern(PatternInfoArray *pia, const char *pattern, int encoding) { PQExpBufferData buf; int dotcnt; PatternInfo *info = extend_pattern_info_array(pia); initPQExpBuffer(&buf); patternToSQLRegex(encoding, NULL, NULL, &buf, pattern, false, false, &dotcnt); if (dotcnt > 0) { pg_log_error("improper qualified name (too many dotted names): %s", pattern); exit(2); } info->pattern = pattern; info->db_regex = pstrdup(buf.data); termPQExpBuffer(&buf); } /* * append_schema_pattern * * Adds the given pattern interpreted as a schema name pattern. * * pia: the pattern info array to be appended * pattern: the schema name pattern * encoding: client encoding for parsing the pattern */ static void append_schema_pattern(PatternInfoArray *pia, const char *pattern, int encoding) { PQExpBufferData dbbuf; PQExpBufferData nspbuf; int dotcnt; PatternInfo *info = extend_pattern_info_array(pia); initPQExpBuffer(&dbbuf); initPQExpBuffer(&nspbuf); patternToSQLRegex(encoding, NULL, &dbbuf, &nspbuf, pattern, false, false, &dotcnt); if (dotcnt > 1) { pg_log_error("improper qualified name (too many dotted names): %s", pattern); exit(2); } info->pattern = pattern; if (dbbuf.data[0]) { opts.dbpattern = true; info->db_regex = pstrdup(dbbuf.data); } if (nspbuf.data[0]) info->nsp_regex = pstrdup(nspbuf.data); termPQExpBuffer(&dbbuf); termPQExpBuffer(&nspbuf); } /* * append_relation_pattern_helper * * Adds to a list the given pattern interpreted as a relation pattern. * * pia: the pattern info array to be appended * pattern: the relation name pattern * encoding: client encoding for parsing the pattern * heap_only: whether the pattern should only be matched against heap tables * btree_only: whether the pattern should only be matched against btree indexes */ static void append_relation_pattern_helper(PatternInfoArray *pia, const char *pattern, int encoding, bool heap_only, bool btree_only) { PQExpBufferData dbbuf; PQExpBufferData nspbuf; PQExpBufferData relbuf; int dotcnt; PatternInfo *info = extend_pattern_info_array(pia); initPQExpBuffer(&dbbuf); initPQExpBuffer(&nspbuf); initPQExpBuffer(&relbuf); patternToSQLRegex(encoding, &dbbuf, &nspbuf, &relbuf, pattern, false, false, &dotcnt); if (dotcnt > 2) { pg_log_error("improper relation name (too many dotted names): %s", pattern); exit(2); } info->pattern = pattern; if (dbbuf.data[0]) { opts.dbpattern = true; info->db_regex = pstrdup(dbbuf.data); } if (nspbuf.data[0]) info->nsp_regex = pstrdup(nspbuf.data); if (relbuf.data[0]) info->rel_regex = pstrdup(relbuf.data); termPQExpBuffer(&dbbuf); termPQExpBuffer(&nspbuf); termPQExpBuffer(&relbuf); info->heap_only = heap_only; info->btree_only = btree_only; } /* * append_relation_pattern * * Adds the given pattern interpreted as a relation pattern, to be matched * against both heap tables and btree indexes. * * pia: the pattern info array to be appended * pattern: the relation name pattern * encoding: client encoding for parsing the pattern */ static void append_relation_pattern(PatternInfoArray *pia, const char *pattern, int encoding) { append_relation_pattern_helper(pia, pattern, encoding, false, false); } /* * append_heap_pattern * * Adds the given pattern interpreted as a relation pattern, to be matched only * against heap tables. * * pia: the pattern info array to be appended * pattern: the relation name pattern * encoding: client encoding for parsing the pattern */ static void append_heap_pattern(PatternInfoArray *pia, const char *pattern, int encoding) { append_relation_pattern_helper(pia, pattern, encoding, true, false); } /* * append_btree_pattern * * Adds the given pattern interpreted as a relation pattern, to be matched only * against btree indexes. * * pia: the pattern info array to be appended * pattern: the relation name pattern * encoding: client encoding for parsing the pattern */ static void append_btree_pattern(PatternInfoArray *pia, const char *pattern, int encoding) { append_relation_pattern_helper(pia, pattern, encoding, false, true); } /* * append_db_pattern_cte * * Appends to the buffer the body of a Common Table Expression (CTE) containing * the database portions filtered from the list of patterns expressed as two * columns: * * pattern_id: the index of this pattern in pia->data[] * rgx: the database regular expression parsed from the pattern * * Patterns without a database portion are skipped. Patterns with more than * just a database portion are optionally skipped, depending on argument * 'inclusive'. * * buf: the buffer to be appended * pia: the array of patterns to be inserted into the CTE * conn: the database connection * inclusive: whether to include patterns with schema and/or relation parts * * Returns whether any database patterns were appended. */ static bool append_db_pattern_cte(PQExpBuffer buf, const PatternInfoArray *pia, PGconn *conn, bool inclusive) { int pattern_id; const char *comma; bool have_values; comma = ""; have_values = false; for (pattern_id = 0; pattern_id < pia->len; pattern_id++) { PatternInfo *info = &pia->data[pattern_id]; if (info->db_regex != NULL && (inclusive || (info->nsp_regex == NULL && info->rel_regex == NULL))) { if (!have_values) appendPQExpBufferStr(buf, "\nVALUES"); have_values = true; appendPQExpBuffer(buf, "%s\n(%d, ", comma, pattern_id); appendStringLiteralConn(buf, info->db_regex, conn); appendPQExpBufferChar(buf, ')'); comma = ","; } } if (!have_values) appendPQExpBufferStr(buf, "\nSELECT NULL, NULL, NULL WHERE false"); return have_values; } /* * compile_database_list * * If any database patterns exist, or if --all was given, compiles a distinct * list of databases to check using a SQL query based on the patterns plus the * literal initial database name, if given. If no database patterns exist and * --all was not given, the query is not necessary, and only the initial * database name (if any) is added to the list. * * conn: connection to the initial database * databases: the list onto which databases should be appended * initial_dbname: an optional extra database name to include in the list */ static void compile_database_list(PGconn *conn, SimplePtrList *databases, const char *initial_dbname) { PGresult *res; PQExpBufferData sql; int ntups; int i; bool fatal; if (initial_dbname) { DatabaseInfo *dat = (DatabaseInfo *) pg_malloc0(sizeof(DatabaseInfo)); /* This database is included. Add to list */ if (opts.verbose) pg_log_info("including database \"%s\"", initial_dbname); dat->datname = pstrdup(initial_dbname); simple_ptr_list_append(databases, dat); } initPQExpBuffer(&sql); /* Append the include patterns CTE. */ appendPQExpBufferStr(&sql, "WITH include_raw (pattern_id, rgx) AS ("); if (!append_db_pattern_cte(&sql, &opts.include, conn, true) && !opts.alldb) { /* * None of the inclusion patterns (if any) contain database portions, * so there is no need to query the database to resolve database * patterns. * * Since we're also not operating under --all, we don't need to query * the exhaustive list of connectable databases, either. */ termPQExpBuffer(&sql); return; } /* Append the exclude patterns CTE. */ appendPQExpBufferStr(&sql, "),\nexclude_raw (pattern_id, rgx) AS ("); append_db_pattern_cte(&sql, &opts.exclude, conn, false); appendPQExpBufferStr(&sql, "),"); /* * Append the database CTE, which includes whether each database is * connectable and also joins against exclude_raw to determine whether * each database is excluded. */ appendPQExpBufferStr(&sql, "\ndatabase (datname) AS (" "\nSELECT d.datname " "FROM pg_catalog.pg_database d " "LEFT OUTER JOIN exclude_raw e " "ON d.datname ~ e.rgx " "\nWHERE d.datallowconn AND datconnlimit != -2 " "AND e.pattern_id IS NULL" ")," /* * Append the include_pat CTE, which joins the include_raw CTE against the * databases CTE to determine if all the inclusion patterns had matches, * and whether each matched pattern had the misfortune of only matching * excluded or unconnectable databases. */ "\ninclude_pat (pattern_id, checkable) AS (" "\nSELECT i.pattern_id, " "COUNT(*) FILTER (" "WHERE d IS NOT NULL" ") AS checkable" "\nFROM include_raw i " "LEFT OUTER JOIN database d " "ON d.datname ~ i.rgx" "\nGROUP BY i.pattern_id" ")," /* * Append the filtered_databases CTE, which selects from the database CTE * optionally joined against the include_raw CTE to only select databases * that match an inclusion pattern. This appears to duplicate what the * include_pat CTE already did above, but here we want only databases, and * there we wanted patterns. */ "\nfiltered_databases (datname) AS (" "\nSELECT DISTINCT d.datname " "FROM database d"); if (!opts.alldb) appendPQExpBufferStr(&sql, " INNER JOIN include_raw i " "ON d.datname ~ i.rgx"); appendPQExpBufferStr(&sql, ")" /* * Select the checkable databases and the unmatched inclusion patterns. */ "\nSELECT pattern_id, datname FROM (" "\nSELECT pattern_id, NULL::TEXT AS datname " "FROM include_pat " "WHERE checkable = 0 " "UNION ALL" "\nSELECT NULL, datname " "FROM filtered_databases" ") AS combined_records" "\nORDER BY pattern_id NULLS LAST, datname"); res = executeQuery(conn, sql.data, opts.echo); if (PQresultStatus(res) != PGRES_TUPLES_OK) { pg_log_error("query failed: %s", PQerrorMessage(conn)); pg_log_error_detail("Query was: %s", sql.data); disconnectDatabase(conn); exit(1); } termPQExpBuffer(&sql); ntups = PQntuples(res); for (fatal = false, i = 0; i < ntups; i++) { int pattern_id = -1; const char *datname = NULL; if (!PQgetisnull(res, i, 0)) pattern_id = atoi(PQgetvalue(res, i, 0)); if (!PQgetisnull(res, i, 1)) datname = PQgetvalue(res, i, 1); if (pattern_id >= 0) { /* * Current record pertains to an inclusion pattern that matched no * checkable databases. */ fatal = opts.strict_names; if (pattern_id >= opts.include.len) pg_fatal("internal error: received unexpected database pattern_id %d", pattern_id); log_no_match("no connectable databases to check matching \"%s\"", opts.include.data[pattern_id].pattern); } else { DatabaseInfo *dat; /* Current record pertains to a database */ Assert(datname != NULL); /* Avoid entering a duplicate entry matching the initial_dbname */ if (initial_dbname != NULL && strcmp(initial_dbname, datname) == 0) continue; /* This database is included. Add to list */ if (opts.verbose) pg_log_info("including database \"%s\"", datname); dat = (DatabaseInfo *) pg_malloc0(sizeof(DatabaseInfo)); dat->datname = pstrdup(datname); simple_ptr_list_append(databases, dat); } } PQclear(res); if (fatal) { if (conn != NULL) disconnectDatabase(conn); exit(1); } } /* * append_rel_pattern_raw_cte * * Appends to the buffer the body of a Common Table Expression (CTE) containing * the given patterns as six columns: * * pattern_id: the index of this pattern in pia->data[] * db_regex: the database regexp parsed from the pattern, or NULL if the * pattern had no database part * nsp_regex: the namespace regexp parsed from the pattern, or NULL if the * pattern had no namespace part * rel_regex: the relname regexp parsed from the pattern, or NULL if the * pattern had no relname part * heap_only: true if the pattern applies only to heap tables (not indexes) * btree_only: true if the pattern applies only to btree indexes (not tables) * * buf: the buffer to be appended * patterns: the array of patterns to be inserted into the CTE * conn: the database connection */ static void append_rel_pattern_raw_cte(PQExpBuffer buf, const PatternInfoArray *pia, PGconn *conn) { int pattern_id; const char *comma; bool have_values; comma = ""; have_values = false; for (pattern_id = 0; pattern_id < pia->len; pattern_id++) { PatternInfo *info = &pia->data[pattern_id]; if (!have_values) appendPQExpBufferStr(buf, "\nVALUES"); have_values = true; appendPQExpBuffer(buf, "%s\n(%d::INTEGER, ", comma, pattern_id); if (info->db_regex == NULL) appendPQExpBufferStr(buf, "NULL"); else appendStringLiteralConn(buf, info->db_regex, conn); appendPQExpBufferStr(buf, "::TEXT, "); if (info->nsp_regex == NULL) appendPQExpBufferStr(buf, "NULL"); else appendStringLiteralConn(buf, info->nsp_regex, conn); appendPQExpBufferStr(buf, "::TEXT, "); if (info->rel_regex == NULL) appendPQExpBufferStr(buf, "NULL"); else appendStringLiteralConn(buf, info->rel_regex, conn); if (info->heap_only) appendPQExpBufferStr(buf, "::TEXT, true::BOOLEAN"); else appendPQExpBufferStr(buf, "::TEXT, false::BOOLEAN"); if (info->btree_only) appendPQExpBufferStr(buf, ", true::BOOLEAN"); else appendPQExpBufferStr(buf, ", false::BOOLEAN"); appendPQExpBufferChar(buf, ')'); comma = ","; } if (!have_values) appendPQExpBufferStr(buf, "\nSELECT NULL::INTEGER, NULL::TEXT, NULL::TEXT, " "NULL::TEXT, NULL::BOOLEAN, NULL::BOOLEAN " "WHERE false"); } /* * append_rel_pattern_filtered_cte * * Appends to the buffer a Common Table Expression (CTE) which selects * all patterns from the named raw CTE, filtered by database. All patterns * which have no database portion or whose database portion matches our * connection's database name are selected, with other patterns excluded. * * The basic idea here is that if we're connected to database "foo" and we have * patterns "foo.bar.baz", "alpha.beta" and "one.two.three", we only want to * use the first two while processing relations in this database, as the third * one is not relevant. * * buf: the buffer to be appended * raw: the name of the CTE to select from * filtered: the name of the CTE to create * conn: the database connection */ static void append_rel_pattern_filtered_cte(PQExpBuffer buf, const char *raw, const char *filtered, PGconn *conn) { appendPQExpBuffer(buf, "\n%s (pattern_id, nsp_regex, rel_regex, heap_only, btree_only) AS (" "\nSELECT pattern_id, nsp_regex, rel_regex, heap_only, btree_only " "FROM %s r" "\nWHERE (r.db_regex IS NULL " "OR ", filtered, raw); appendStringLiteralConn(buf, PQdb(conn), conn); appendPQExpBufferStr(buf, " ~ r.db_regex)"); appendPQExpBufferStr(buf, " AND (r.nsp_regex IS NOT NULL" " OR r.rel_regex IS NOT NULL)" "),"); } /* * compile_relation_list_one_db * * Compiles a list of relations to check within the currently connected * database based on the user supplied options, sorted by descending size, * and appends them to the given list of relations. * * The cells of the constructed list contain all information about the relation * necessary to connect to the database and check the object, including which * database to connect to, where contrib/amcheck is installed, and the Oid and * type of object (heap table vs. btree index). Rather than duplicating the * database details per relation, the relation structs use references to the * same database object, provided by the caller. * * conn: connection to this next database, which should be the same as in 'dat' * relations: list onto which the relations information should be appended * dat: the database info struct for use by each relation * pagecount: gets incremented by the number of blocks to check in all * relations added */ static void compile_relation_list_one_db(PGconn *conn, SimplePtrList *relations, const DatabaseInfo *dat, uint64 *pagecount) { PGresult *res; PQExpBufferData sql; int ntups; int i; initPQExpBuffer(&sql); appendPQExpBufferStr(&sql, "WITH"); /* Append CTEs for the relation inclusion patterns, if any */ if (!opts.allrel) { appendPQExpBufferStr(&sql, " include_raw (pattern_id, db_regex, nsp_regex, rel_regex, heap_only, btree_only) AS ("); append_rel_pattern_raw_cte(&sql, &opts.include, conn); appendPQExpBufferStr(&sql, "\n),"); append_rel_pattern_filtered_cte(&sql, "include_raw", "include_pat", conn); } /* Append CTEs for the relation exclusion patterns, if any */ if (opts.excludetbl || opts.excludeidx || opts.excludensp) { appendPQExpBufferStr(&sql, " exclude_raw (pattern_id, db_regex, nsp_regex, rel_regex, heap_only, btree_only) AS ("); append_rel_pattern_raw_cte(&sql, &opts.exclude, conn); appendPQExpBufferStr(&sql, "\n),"); append_rel_pattern_filtered_cte(&sql, "exclude_raw", "exclude_pat", conn); } /* Append the relation CTE. */ appendPQExpBufferStr(&sql, " relation (pattern_id, oid, nspname, relname, reltoastrelid, relpages, is_heap, is_btree) AS (" "\nSELECT DISTINCT ON (c.oid"); if (!opts.allrel) appendPQExpBufferStr(&sql, ", ip.pattern_id) ip.pattern_id,"); else appendPQExpBufferStr(&sql, ") NULL::INTEGER AS pattern_id,"); appendPQExpBuffer(&sql, "\nc.oid, n.nspname, c.relname, c.reltoastrelid, c.relpages, " "c.relam = %u AS is_heap, " "c.relam = %u AS is_btree" "\nFROM pg_catalog.pg_class c " "INNER JOIN pg_catalog.pg_namespace n " "ON c.relnamespace = n.oid", HEAP_TABLE_AM_OID, BTREE_AM_OID); if (!opts.allrel) appendPQExpBuffer(&sql, "\nINNER JOIN include_pat ip" "\nON (n.nspname ~ ip.nsp_regex OR ip.nsp_regex IS NULL)" "\nAND (c.relname ~ ip.rel_regex OR ip.rel_regex IS NULL)" "\nAND (c.relam = %u OR NOT ip.heap_only)" "\nAND (c.relam = %u OR NOT ip.btree_only)", HEAP_TABLE_AM_OID, BTREE_AM_OID); if (opts.excludetbl || opts.excludeidx || opts.excludensp) appendPQExpBuffer(&sql, "\nLEFT OUTER JOIN exclude_pat ep" "\nON (n.nspname ~ ep.nsp_regex OR ep.nsp_regex IS NULL)" "\nAND (c.relname ~ ep.rel_regex OR ep.rel_regex IS NULL)" "\nAND (c.relam = %u OR NOT ep.heap_only OR ep.rel_regex IS NULL)" "\nAND (c.relam = %u OR NOT ep.btree_only OR ep.rel_regex IS NULL)", HEAP_TABLE_AM_OID, BTREE_AM_OID); /* * Exclude temporary tables and indexes, which must necessarily belong to * other sessions. (We don't create any ourselves.) We must ultimately * exclude indexes marked invalid or not ready, but we delay that decision * until firing off the amcheck command, as the state of an index may * change by then. */ appendPQExpBufferStr(&sql, "\nWHERE c.relpersistence != " CppAsString2(RELPERSISTENCE_TEMP)); if (opts.excludetbl || opts.excludeidx || opts.excludensp) appendPQExpBufferStr(&sql, "\nAND ep.pattern_id IS NULL"); /* * We need to be careful not to break the --no-dependent-toast and * --no-dependent-indexes options. By default, the btree indexes, toast * tables, and toast table btree indexes associated with primary heap * tables are included, using their own CTEs below. We implement the * --exclude-* options by not creating those CTEs, but that's no use if * we've already selected the toast and indexes here. On the other hand, * we want inclusion patterns that match indexes or toast tables to be * honored. So, if inclusion patterns were given, we want to select all * tables, toast tables, or indexes that match the patterns. But if no * inclusion patterns were given, and we're simply matching all relations, * then we only want to match the primary tables here. */ if (opts.allrel) appendPQExpBuffer(&sql, " AND c.relam = %u " "AND c.relkind IN (" CppAsString2(RELKIND_RELATION) ", " CppAsString2(RELKIND_SEQUENCE) ", " CppAsString2(RELKIND_MATVIEW) ", " CppAsString2(RELKIND_TOASTVALUE) ") " "AND c.relnamespace != %u", HEAP_TABLE_AM_OID, PG_TOAST_NAMESPACE); else appendPQExpBuffer(&sql, " AND c.relam IN (%u, %u)" "AND c.relkind IN (" CppAsString2(RELKIND_RELATION) ", " CppAsString2(RELKIND_SEQUENCE) ", " CppAsString2(RELKIND_MATVIEW) ", " CppAsString2(RELKIND_TOASTVALUE) ", " CppAsString2(RELKIND_INDEX) ") " "AND ((c.relam = %u AND c.relkind IN (" CppAsString2(RELKIND_RELATION) ", " CppAsString2(RELKIND_SEQUENCE) ", " CppAsString2(RELKIND_MATVIEW) ", " CppAsString2(RELKIND_TOASTVALUE) ")) OR " "(c.relam = %u AND c.relkind = " CppAsString2(RELKIND_INDEX) "))", HEAP_TABLE_AM_OID, BTREE_AM_OID, HEAP_TABLE_AM_OID, BTREE_AM_OID); appendPQExpBufferStr(&sql, "\nORDER BY c.oid)"); if (!opts.no_toast_expansion) { /* * Include a CTE for toast tables associated with primary heap tables * selected above, filtering by exclusion patterns (if any) that match * toast table names. */ appendPQExpBufferStr(&sql, ", toast (oid, nspname, relname, relpages) AS (" "\nSELECT t.oid, 'pg_toast', t.relname, t.relpages" "\nFROM pg_catalog.pg_class t " "INNER JOIN relation r " "ON r.reltoastrelid = t.oid"); if (opts.excludetbl || opts.excludensp) appendPQExpBufferStr(&sql, "\nLEFT OUTER JOIN exclude_pat ep" "\nON ('pg_toast' ~ ep.nsp_regex OR ep.nsp_regex IS NULL)" "\nAND (t.relname ~ ep.rel_regex OR ep.rel_regex IS NULL)" "\nAND ep.heap_only" "\nWHERE ep.pattern_id IS NULL" "\nAND t.relpersistence != " CppAsString2(RELPERSISTENCE_TEMP)); appendPQExpBufferStr(&sql, "\n)"); } if (!opts.no_btree_expansion) { /* * Include a CTE for btree indexes associated with primary heap tables * selected above, filtering by exclusion patterns (if any) that match * btree index names. */ appendPQExpBufferStr(&sql, ", index (oid, nspname, relname, relpages) AS (" "\nSELECT c.oid, r.nspname, c.relname, c.relpages " "FROM relation r" "\nINNER JOIN pg_catalog.pg_index i " "ON r.oid = i.indrelid " "INNER JOIN pg_catalog.pg_class c " "ON i.indexrelid = c.oid " "AND c.relpersistence != " CppAsString2(RELPERSISTENCE_TEMP)); if (opts.excludeidx || opts.excludensp) appendPQExpBufferStr(&sql, "\nINNER JOIN pg_catalog.pg_namespace n " "ON c.relnamespace = n.oid" "\nLEFT OUTER JOIN exclude_pat ep " "ON (n.nspname ~ ep.nsp_regex OR ep.nsp_regex IS NULL) " "AND (c.relname ~ ep.rel_regex OR ep.rel_regex IS NULL) " "AND ep.btree_only" "\nWHERE ep.pattern_id IS NULL"); else appendPQExpBufferStr(&sql, "\nWHERE true"); appendPQExpBuffer(&sql, " AND c.relam = %u " "AND c.relkind = " CppAsString2(RELKIND_INDEX), BTREE_AM_OID); if (opts.no_toast_expansion) appendPQExpBuffer(&sql, " AND c.relnamespace != %u", PG_TOAST_NAMESPACE); appendPQExpBufferStr(&sql, "\n)"); } if (!opts.no_toast_expansion && !opts.no_btree_expansion) { /* * Include a CTE for btree indexes associated with toast tables of * primary heap tables selected above, filtering by exclusion patterns * (if any) that match the toast index names. */ appendPQExpBufferStr(&sql, ", toast_index (oid, nspname, relname, relpages) AS (" "\nSELECT c.oid, 'pg_toast', c.relname, c.relpages " "FROM toast t " "INNER JOIN pg_catalog.pg_index i " "ON t.oid = i.indrelid" "\nINNER JOIN pg_catalog.pg_class c " "ON i.indexrelid = c.oid " "AND c.relpersistence != " CppAsString2(RELPERSISTENCE_TEMP)); if (opts.excludeidx) appendPQExpBufferStr(&sql, "\nLEFT OUTER JOIN exclude_pat ep " "ON ('pg_toast' ~ ep.nsp_regex OR ep.nsp_regex IS NULL) " "AND (c.relname ~ ep.rel_regex OR ep.rel_regex IS NULL) " "AND ep.btree_only " "WHERE ep.pattern_id IS NULL"); else appendPQExpBufferStr(&sql, "\nWHERE true"); appendPQExpBuffer(&sql, " AND c.relam = %u" " AND c.relkind = " CppAsString2(RELKIND_INDEX) ")", BTREE_AM_OID); } /* * Roll-up distinct rows from CTEs. * * Relations that match more than one pattern may occur more than once in * the list, and indexes and toast for primary relations may also have * matched in their own right, so we rely on UNION to deduplicate the * list. */ appendPQExpBufferStr(&sql, "\nSELECT pattern_id, is_heap, is_btree, oid, nspname, relname, relpages " "FROM ("); appendPQExpBufferStr(&sql, /* Inclusion patterns that failed to match */ "\nSELECT pattern_id, is_heap, is_btree, " "NULL::OID AS oid, " "NULL::TEXT AS nspname, " "NULL::TEXT AS relname, " "NULL::INTEGER AS relpages" "\nFROM relation " "WHERE pattern_id IS NOT NULL " "UNION" /* Primary relations */ "\nSELECT NULL::INTEGER AS pattern_id, " "is_heap, is_btree, oid, nspname, relname, relpages " "FROM relation"); if (!opts.no_toast_expansion) appendPQExpBufferStr(&sql, " UNION" /* Toast tables for primary relations */ "\nSELECT NULL::INTEGER AS pattern_id, TRUE AS is_heap, " "FALSE AS is_btree, oid, nspname, relname, relpages " "FROM toast"); if (!opts.no_btree_expansion) appendPQExpBufferStr(&sql, " UNION" /* Indexes for primary relations */ "\nSELECT NULL::INTEGER AS pattern_id, FALSE AS is_heap, " "TRUE AS is_btree, oid, nspname, relname, relpages " "FROM index"); if (!opts.no_toast_expansion && !opts.no_btree_expansion) appendPQExpBufferStr(&sql, " UNION" /* Indexes for toast relations */ "\nSELECT NULL::INTEGER AS pattern_id, FALSE AS is_heap, " "TRUE AS is_btree, oid, nspname, relname, relpages " "FROM toast_index"); appendPQExpBufferStr(&sql, "\n) AS combined_records " "ORDER BY relpages DESC NULLS FIRST, oid"); res = executeQuery(conn, sql.data, opts.echo); if (PQresultStatus(res) != PGRES_TUPLES_OK) { pg_log_error("query failed: %s", PQerrorMessage(conn)); pg_log_error_detail("Query was: %s", sql.data); disconnectDatabase(conn); exit(1); } termPQExpBuffer(&sql); ntups = PQntuples(res); for (i = 0; i < ntups; i++) { int pattern_id = -1; bool is_heap = false; bool is_btree PG_USED_FOR_ASSERTS_ONLY = false; Oid oid = InvalidOid; const char *nspname = NULL; const char *relname = NULL; int relpages = 0; if (!PQgetisnull(res, i, 0)) pattern_id = atoi(PQgetvalue(res, i, 0)); if (!PQgetisnull(res, i, 1)) is_heap = (PQgetvalue(res, i, 1)[0] == 't'); if (!PQgetisnull(res, i, 2)) is_btree = (PQgetvalue(res, i, 2)[0] == 't'); if (!PQgetisnull(res, i, 3)) oid = atooid(PQgetvalue(res, i, 3)); if (!PQgetisnull(res, i, 4)) nspname = PQgetvalue(res, i, 4); if (!PQgetisnull(res, i, 5)) relname = PQgetvalue(res, i, 5); if (!PQgetisnull(res, i, 6)) relpages = atoi(PQgetvalue(res, i, 6)); if (pattern_id >= 0) { /* * Current record pertains to an inclusion pattern. Record that * it matched. */ if (pattern_id >= opts.include.len) pg_fatal("internal error: received unexpected relation pattern_id %d", pattern_id); opts.include.data[pattern_id].matched = true; } else { /* Current record pertains to a relation */ RelationInfo *rel = (RelationInfo *) pg_malloc0(sizeof(RelationInfo)); Assert(OidIsValid(oid)); Assert((is_heap && !is_btree) || (is_btree && !is_heap)); rel->datinfo = dat; rel->reloid = oid; rel->is_heap = is_heap; rel->nspname = pstrdup(nspname); rel->relname = pstrdup(relname); rel->relpages = relpages; rel->blocks_to_check = relpages; if (is_heap && (opts.startblock >= 0 || opts.endblock >= 0)) { /* * We apply --startblock and --endblock to heap tables, but * not btree indexes, and for progress purposes we need to * track how many blocks we expect to check. */ if (opts.endblock >= 0 && rel->blocks_to_check > opts.endblock) rel->blocks_to_check = opts.endblock + 1; if (opts.startblock >= 0) { if (rel->blocks_to_check > opts.startblock) rel->blocks_to_check -= opts.startblock; else rel->blocks_to_check = 0; } } *pagecount += rel->blocks_to_check; simple_ptr_list_append(relations, rel); } } PQclear(res); }