From f19dfca79393ff327648999e3b0c4960d50ef0c8 Mon Sep 17 00:00:00 2001 From: DmitriiAn Date: Mon, 28 Sep 2026 16:06:21 +0200 Subject: [PATCH 1/2] skip TIMELINE_HISTORY on the logical replication connection --- src/bin/pgcopydb/ld_stream.c | 10 -------- src/bin/pgcopydb/pgsql_timeline.c | 38 ------------------------------- 2 files changed, 48 deletions(-) diff --git a/src/bin/pgcopydb/ld_stream.c b/src/bin/pgcopydb/ld_stream.c index a12ce281e..1170102c3 100644 --- a/src/bin/pgcopydb/ld_stream.c +++ b/src/bin/pgcopydb/ld_stream.c @@ -2895,16 +2895,6 @@ stream_write_context(StreamSpecs *specs, LogicalStreamClient *stream) log_debug("Wrote tli %s timeline file \"%s\"", tli, specs->paths.tlifile); - /* read from the timeline history file and populate internal catalogs */ - if (stream->system.timeline > 1 && - !parse_timeline_history_file(stream->system.timelineHistoryFilename, - specs->sourceDB, - stream->system.timeline)) - { - /* errors have already been logged */ - return false; - } - return true; } diff --git a/src/bin/pgcopydb/pgsql_timeline.c b/src/bin/pgcopydb/pgsql_timeline.c index 8fbc0686f..223602a7e 100644 --- a/src/bin/pgcopydb/pgsql_timeline.c +++ b/src/bin/pgcopydb/pgsql_timeline.c @@ -84,44 +84,6 @@ pgsql_identify_system(PGSQL *pgsql, IdentifySystem *system, char *cdcPathDir) return false; } - /* while at it, we also run the TIMELINE_HISTORY command */ - if (system->timeline > 1) - { - TimelineHistoryResult hContext = { 0 }; - - char sql[BUFSIZE] = { 0 }; - sformat(sql, sizeof(sql), "TIMELINE_HISTORY %d", system->timeline); - - result = PQexec(connection, sql); - - if (!is_response_ok(result)) - { - log_error("Failed to request TIMELINE_HISTORY: %s", - PQerrorMessage(connection)); - PQclear(result); - clear_results(pgsql); - - PQfinish(connection); - - return false; - } - - (void) parseTimelineHistoryResult((void *) &hContext, result, cdcPathDir); - - PQclear(result); - clear_results(pgsql); - - if (!hContext.parsedOk) - { - log_error("Failed to get result from TIMELINE_HISTORY"); - PQfinish(connection); - return false; - } - - /* store the filename for the timeline history file */ - strlcpy(system->timelineHistoryFilename, hContext.filename, MAXPGPATH); - } - if (connIsOurs) { (void) pgsql_finish(pgsql); From 7ffaecc4f6ce1cc37171be224f0f28d24fa6cf7c Mon Sep 17 00:00:00 2001 From: Chris Munns Date: Mon, 28 Sep 2026 12:40:18 -0400 Subject: [PATCH 2/2] Remove the timeline history code and detect a timeline change Nothing read the timeline history. The rows filled IdentifySystem.currentTimeline, which no code reads, so the parser, the catalog table and the cdcPathDir plumbing that fed them are all removed. pgsql_identify_system loses its third parameter. In place of the collection, stream_write_context now compares the source timeline against the one recorded for this migration and stops when they differ. A promotion or a point-in-time restore forks the source history, and changes already applied to the target may belong to a branch the new source does not have. The check reads the timeline that IDENTIFY_SYSTEM already returns, so it needs no extra replication command and works on managed providers that restrict TIMELINE_HISTORY. The recorded timeline is left in place on failure so the operator can see both values. The cdc-low-level suite covers the new path. --- src/bin/pgcopydb/catalog.c | 173 +----------------------- src/bin/pgcopydb/catalog.h | 10 -- src/bin/pgcopydb/ld_stream.c | 58 ++++++-- src/bin/pgcopydb/pgsql.c | 2 +- src/bin/pgcopydb/pgsql.h | 17 +-- src/bin/pgcopydb/pgsql_timeline.c | 216 +----------------------------- src/bin/pgcopydb/pgsql_timeline.h | 6 +- tests/cdc-low-level/copydb.sh | 25 ++++ 8 files changed, 76 insertions(+), 431 deletions(-) diff --git a/src/bin/pgcopydb/catalog.c b/src/bin/pgcopydb/catalog.c index 02b025f4d..f0206018f 100644 --- a/src/bin/pgcopydb/catalog.c +++ b/src/bin/pgcopydb/catalog.c @@ -221,10 +221,7 @@ static char *sourceDBcreateDDLs[] = { "create table sentinel(" " id integer primary key check (id = 1), " " startpos pg_lsn, endpos pg_lsn, apply bool, " - " write_lsn pg_lsn, flush_lsn pg_lsn, replay_lsn pg_lsn)", - - "create table timeline_history(" - " tli integer primary key, startpos pg_lsn, endpos pg_lsn)" + " write_lsn pg_lsn, flush_lsn pg_lsn, replay_lsn pg_lsn)" }; @@ -517,8 +514,7 @@ static char *sourceDBdropDDLs[] = { "drop table if exists s_table_parts_done", "drop table if exists s_table_indexes_done", - "drop table if exists sentinel", - "drop table if exists timeline_history" + "drop table if exists sentinel" }; @@ -9601,171 +9597,6 @@ catalog_count_bytes_fetch(SQLiteQuery *query) } -/* - * catalog_add_timeline_history inserts a timeline history entry to our - * internal catalogs database. - */ -bool -catalog_add_timeline_history(DatabaseCatalog *catalog, TimelineHistoryEntry *entry) -{ - if (catalog == NULL) - { - log_error("BUG: catalog_add_timeline_history: catalog is NULL"); - return false; - } - - sqlite3 *db = catalog->db; - - if (db == NULL) - { - log_error("BUG: catalog_add_timeline_history: db is NULL"); - return false; - } - - char *sql = - "insert or replace into timeline_history(tli, startpos, endpos)" - "values($1, $2, $3)"; - - SQLiteQuery query = { 0 }; - - if (!catalog_sql_prepare(db, sql, &query)) - { - /* errors have already been logged */ - return false; - } - - char slsn[PG_LSN_MAXLENGTH] = { 0 }; - char elsn[PG_LSN_MAXLENGTH] = { 0 }; - - sformat(slsn, sizeof(slsn), "%X/%X", LSN_FORMAT_ARGS(entry->begin)); - sformat(elsn, sizeof(elsn), "%X/%X", LSN_FORMAT_ARGS(entry->end)); - - /* bind our parameters now */ - BindParam params[] = { - { BIND_PARAMETER_TYPE_INT, "tli", entry->tli, NULL }, - { BIND_PARAMETER_TYPE_TEXT, "startpos", 0, slsn }, - { BIND_PARAMETER_TYPE_TEXT, "endpos", 0, elsn } - }; - - int count = sizeof(params) / sizeof(params[0]); - - if (!catalog_sql_bind(&query, params, count)) - { - /* errors have already been logged */ - return false; - } - - /* now execute the query, which does not return any row */ - if (!catalog_sql_execute_once(&query)) - { - /* errors have already been logged */ - return false; - } - - return true; -} - - -/* - * catalog_lookup_timeline_history fetches the current TimelineHistoryEntry - * from our catalogs. - */ -bool -catalog_lookup_timeline_history(DatabaseCatalog *catalog, - int tli, - TimelineHistoryEntry *entry) -{ - sqlite3 *db = catalog->db; - - if (db == NULL) - { - log_error("BUG: catalog_lookup_timeline_history: db is NULL"); - return false; - } - - SQLiteQuery query = { - .context = entry, - .fetchFunction = &catalog_timeline_history_fetch - }; - - char *sql = - " select tli, startpos, endpos" - " from timeline_history" - " where tli = $1"; - - - if (!catalog_sql_prepare(db, sql, &query)) - { - /* errors have already been logged */ - return false; - } - - /* bind our parameters now */ - BindParam params[] = { - { BIND_PARAMETER_TYPE_INT, "tli", tli, NULL } - }; - - int count = sizeof(params) / sizeof(params[0]); - - if (!catalog_sql_bind(&query, params, count)) - { - /* errors have already been logged */ - return false; - } - - /* now execute the query, which return exactly one row */ - if (!catalog_sql_execute_once(&query)) - { - /* errors have already been logged */ - return false; - } - - return true; -} - - -/* - * catalog_timeline_history_fetch fetches a TimelineHistoryEntry from a query - * ppStmt result. - */ -bool -catalog_timeline_history_fetch(SQLiteQuery *query) -{ - TimelineHistoryEntry *entry = (TimelineHistoryEntry *) query->context; - - bzero(entry, sizeof(TimelineHistoryEntry)); - - /* tli */ - entry->tli = sqlite3_column_int(query->ppStmt, 0); - - /* begin LSN */ - if (sqlite3_column_type(query->ppStmt, 1) != SQLITE_NULL) - { - const char *startpos = (const char *) sqlite3_column_text(query->ppStmt, 1); - - if (!parseLSN(startpos, &entry->begin)) - { - log_error("Failed to parse LSN from \"%s\"", startpos); - return false; - } - } - - /* end LSN */ - if (sqlite3_column_type(query->ppStmt, 2) != SQLITE_NULL) - { - const char *endpos = (const char *) sqlite3_column_text(query->ppStmt, 2); - - if (!parseLSN(endpos, &entry->end)) - { - log_error("Failed to parse LSN from \"%s\"", endpos); - return false; - } - } - - return true; -} - - /* * catalog_execute executes sqlite query */ diff --git a/src/bin/pgcopydb/catalog.h b/src/bin/pgcopydb/catalog.h index af91f1c70..e3f73f1d2 100644 --- a/src/bin/pgcopydb/catalog.h +++ b/src/bin/pgcopydb/catalog.h @@ -748,16 +748,6 @@ bool catalog_count_bytes(DatabaseCatalog *catalog, CatalogBytesCounts *count); bool catalog_count_bytes_fetch(SQLiteQuery *query); -/* - * Logical decoding - */ -bool catalog_add_timeline_history(DatabaseCatalog *catalog, - TimelineHistoryEntry *entry); -bool catalog_lookup_timeline_history(DatabaseCatalog *catalog, - int tli, - TimelineHistoryEntry *entry); -bool catalog_timeline_history_fetch(SQLiteQuery *query); - /* * Internal tooling for catalogs management */ diff --git a/src/bin/pgcopydb/ld_stream.c b/src/bin/pgcopydb/ld_stream.c index 1170102c3..0c3f3b51e 100644 --- a/src/bin/pgcopydb/ld_stream.c +++ b/src/bin/pgcopydb/ld_stream.c @@ -467,7 +467,6 @@ startLogicalStreaming(StreamSpecs *specs) stream.closeFunction = &streamClose; stream.feedbackFunction = &streamFeedback; stream.keepaliveFunction = &streamKeepalive; - strlcpy(stream.cdcPathDir, specs->paths.dir, MAXPGPATH); /* * Read possibly already existing file to initialize the start LSN from a @@ -2855,10 +2854,46 @@ stream_fetch_current_lsn(uint64_t *lsn, } -/* - * stream_write_context writes the wal_segment_size and tli to files, as well as - * populate our internal catalogs with information in the timeline history file. - */ +static bool +stream_check_timeline(StreamSpecs *specs, uint32_t timeline) +{ + if (!file_exists(specs->paths.tlifile)) + { + return true; + } + + char *contents = NULL; + long size = 0L; + + if (!read_file(specs->paths.tlifile, &contents, &size)) + { + return false; + } + + uint32_t previous = 0; + + if (!stringToUInt(contents, &previous)) + { + log_error("Failed to parse timeline \"%s\" from \"%s\"", + contents, specs->paths.tlifile); + return false; + } + + if (previous == timeline) + { + return true; + } + + log_error("Source timeline changed from %u to %u", previous, timeline); + log_error("The source was promoted or restored to an earlier point in time"); + log_error("Changes already applied to the target may not exist on the " + "timeline the source follows now"); + log_error("Compare the source and the target, then start a new migration"); + + return false; +} + + bool stream_write_context(StreamSpecs *specs, LogicalStreamClient *stream) { @@ -2887,6 +2922,11 @@ stream_write_context(StreamSpecs *specs, LogicalStreamClient *stream) bytes = sformat(tli, sizeof(tli), "%d", system->timeline); + if (!stream_check_timeline(specs, system->timeline)) + { + return false; + } + if (!write_file(tli, bytes, specs->paths.tlifile)) { /* errors have already been logged */ @@ -3014,13 +3054,5 @@ stream_read_context(StreamSpecs *specs) return false; } - DatabaseCatalog *source = specs->sourceDB; - if (!catalog_lookup_timeline_history(source, system->timeline, - &system->currentTimeline)) - { - /* errors have already been logged */ - return false; - } - return true; } diff --git a/src/bin/pgcopydb/pgsql.c b/src/bin/pgcopydb/pgsql.c index 2dc183db0..b0a034fb2 100644 --- a/src/bin/pgcopydb/pgsql.c +++ b/src/bin/pgcopydb/pgsql.c @@ -4154,7 +4154,7 @@ pgsql_start_replication(LogicalStreamClient *client) } /* fetch the source timeline */ - if (!pgsql_identify_system(pgsql, &(client->system), client->cdcPathDir)) + if (!pgsql_identify_system(pgsql, &(client->system))) { /* errors have already been logged */ destroyPQExpBuffer(query); diff --git a/src/bin/pgcopydb/pgsql.h b/src/bin/pgcopydb/pgsql.h index e02b73d20..32bb542a8 100644 --- a/src/bin/pgcopydb/pgsql.h +++ b/src/bin/pgcopydb/pgsql.h @@ -379,11 +379,6 @@ bool pg_copy_large_object(PGSQL *src, #define PG_LSN_MAXLENGTH 18 /* - * TimelineHistoryEntry is taken from Postgres definitions and adapted to - * client-size code where we don't have all the necessary infrastruture. In - * particular we don't define a XLogRecPtr data type nor do we define a - * TimelineID data type. - * * Zero is used indicate an invalid pointer. Bootstrap skips the first possible * WAL segment, initializing the first WAL page at WAL segment size, so no XLOG * record can begin at zero. @@ -391,17 +386,10 @@ bool pg_copy_large_object(PGSQL *src, #define InvalidXLogRecPtr 0 #define XLogRecPtrIsInvalid(r) ((r) == InvalidXLogRecPtr) -typedef struct TimelineHistoryEntry -{ - uint32_t tli; - uint64_t begin; /* inclusive */ - uint64_t end; /* exclusive, InvalidXLogRecPtr means infinity */ -} TimelineHistoryEntry; - /* * The IdentifySystem contains information that is parsed from the - * IDENTIFY_SYSTEM replication command, and then the TIMELINE_HISTORY result. + * IDENTIFY_SYSTEM replication command. */ typedef struct IdentifySystem { @@ -409,8 +397,6 @@ typedef struct IdentifySystem uint32_t timeline; char xlogpos[PG_LSN_MAXLENGTH]; char dbname[NAMEDATALEN]; - TimelineHistoryEntry currentTimeline; - char timelineHistoryFilename[MAXPGPATH]; } IdentifySystem; @@ -470,7 +456,6 @@ typedef struct LogicalStreamClient { PGSQL pgsql; IdentifySystem system; - char cdcPathDir[MAXPGPATH]; char slotName[NAMEDATALEN]; diff --git a/src/bin/pgcopydb/pgsql_timeline.c b/src/bin/pgcopydb/pgsql_timeline.c index 223602a7e..fccaadf1b 100644 --- a/src/bin/pgcopydb/pgsql_timeline.c +++ b/src/bin/pgcopydb/pgsql_timeline.c @@ -20,18 +20,7 @@ typedef struct IdentifySystemResult } IdentifySystemResult; -typedef struct TimelineHistoryResult -{ - char sqlstate[6]; - bool parsedOk; - char filename[MAXPGPATH]; -} TimelineHistoryResult; - static void parseIdentifySystemResult(IdentifySystemResult *ctx, PGresult *result); -static void parseTimelineHistoryResult(TimelineHistoryResult *context, PGresult *result, - char *cdcPathDir); -static bool writeTimelineHistoryFile(char *filename, char *content); -static bool register_timeline_hook(void *ctx, char *line); /* @@ -40,7 +29,7 @@ static bool register_timeline_hook(void *ctx, char *line); * contain the 'replication=1' parameter. */ bool -pgsql_identify_system(PGSQL *pgsql, IdentifySystem *system, char *cdcPathDir) +pgsql_identify_system(PGSQL *pgsql, IdentifySystem *system) { bool connIsOurs = pgsql->connection == NULL; @@ -93,20 +82,6 @@ pgsql_identify_system(PGSQL *pgsql, IdentifySystem *system, char *cdcPathDir) } -/* - * writeTimelineHistoryFile writes the content of a timeline history file to disk. - * The filename is determined by the PostgreSQL TIMELINE_HISTORY command. - */ -static bool -writeTimelineHistoryFile(char *filename, char *content) -{ - log_debug("Writing timeline history file \"%s\"", filename); - - size_t size = strlen(content); - return write_file(content, size, filename); -} - - /* * parseIdentifySystemResult parses the result from a replication query * IDENTIFY_SYSTEM, and fills the given IdentifySystem structure. @@ -165,192 +140,3 @@ parseIdentifySystemResult(IdentifySystemResult *context, PGresult *result) context->parsedOk = true; } - - -/* - * parseTimelineHistoryResult parses the result of the TIMELINE_HISTORY - * replication command. The content is written to disk, and the filename - * is stored in the TimelineHistoryResult structure. - */ -static void -parseTimelineHistoryResult(TimelineHistoryResult *context, PGresult *result, - char *cdcPathDir) -{ - if (PQnfields(result) != 2) - { - log_error("Query returned %d columns, expected 2", PQnfields(result)); - context->parsedOk = false; - return; - } - - if (PQntuples(result) == 0) - { - log_sql("parseTimelineHistoryResult: query returned no rows"); - context->parsedOk = false; - return; - } - - if (PQntuples(result) != 1) - { - log_error("Query returned %d rows, expected 1", PQntuples(result)); - context->parsedOk = false; - return; - } - - /* filename (text) */ - char *value = PQgetvalue(result, 0, 0); - sformat(context->filename, sizeof(context->filename), "%s/%s", cdcPathDir, value); - - /* - * content (bytea) - * - * We do not want to store this value in memory. Instead we write it to disk - * as it is. - */ - value = PQgetvalue(result, 0, 1); - if (!writeTimelineHistoryFile(context->filename, value)) - { - log_error("Failed to write timeline history file \"%s\"", context->filename); - context->parsedOk = false; - return; - } - - context->parsedOk = true; -} - - -typedef struct TimeLineHistoryContext -{ - DatabaseCatalog *catalog; - uint32_t prevtli; - uint32_t prevend; -} TimelineHistoryContext; - - -/* - * parse_timeline_history_file is a wrapper for the iterator that prepares the - * context etc. - */ -bool -parse_timeline_history_file(char *filename, - DatabaseCatalog *catalog, - uint32_t currentTimeline) -{ - /* step 1: prepare the context */ - TimelineHistoryContext context = - { - .catalog = catalog, - .prevtli = 0, - .prevend = InvalidXLogRecPtr - }; - - /* step 2: iterate over the file */ - if (!file_iter_lines(filename, BUFSIZE, &context, register_timeline_hook)) - { - /* errors have already been logged */ - return false; - } - - /* step 3: add the current timeline to catalog */ - if (currentTimeline != context.prevtli + 1) - { - log_warn("parse_timeline_history_file: Expected timeline %d, got %d", - context.prevtli + 1, currentTimeline); - } - - TimelineHistoryEntry entry = { - .tli = currentTimeline, - .begin = context.prevend, - .end = InvalidXLogRecPtr - }; - - if (!catalog_add_timeline_history(catalog, &entry)) - { - log_error("Failed to add timeline history entry to catalog"); - return false; - } - - return true; -} - - -/* register_timeline_hook is the callback that is called for each line of a */ -/* timeline history file. */ -static bool -register_timeline_hook(void *ctx, char *line) -{ - TimelineHistoryContext *context = (TimelineHistoryContext *) ctx; - - char *ptr = line; - - /* skip leading whitespace */ - for (; *ptr; ptr++) - { - if (!isspace((unsigned char) *ptr)) - { - break; - } - } - - if (*ptr == '\0') - { - /* skip empty lines */ - return true; - } - - log_trace("register_timeline_hook: line is \"%s\"", line); - - char *tabptr = strchr(ptr, '\t'); - - if (tabptr == NULL) - { - log_error("Failed to parse history file line \"%s\"", line); - return false; - } - - *tabptr = '\0'; - - uint32_t current_tli = 0; - uint64_t current_lsn = InvalidXLogRecPtr; - - if (!stringToUInt(ptr, ¤t_tli)) - { - log_error("Failed to parse history timeline \"%s\"", line); - return false; - } - - char *lsn = tabptr + 1; - - for (char *lsnend = lsn; *lsnend; lsnend++) - { - if (!(isxdigit((unsigned char) *lsnend) || *lsnend == '/')) - { - *lsnend = '\0'; - break; - } - } - - if (!parseLSN(lsn, ¤t_lsn)) - { - log_error("Failed to parse history timeline %d LSN \"%s\"", - current_tli, lsn); - return false; - } - - TimelineHistoryEntry entry = { - .tli = current_tli, - .begin = context->prevend, - .end = current_lsn - }; - - if (!catalog_add_timeline_history(context->catalog, &entry)) - { - log_error("Failed to add timeline history entry to catalog"); - return false; - } - - context->prevtli = current_tli; - context->prevend = current_lsn; - - return true; -} diff --git a/src/bin/pgcopydb/pgsql_timeline.h b/src/bin/pgcopydb/pgsql_timeline.h index a3dabaee6..4c638df05 100644 --- a/src/bin/pgcopydb/pgsql_timeline.h +++ b/src/bin/pgcopydb/pgsql_timeline.h @@ -9,10 +9,6 @@ #include "schema.h" /* pgsql_timeline.c */ -bool pgsql_identify_system(PGSQL *pgsql, IdentifySystem *system, - char *cdcPathDir); -bool parse_timeline_history_file(char *filename, - DatabaseCatalog *catalog, - uint32_t currentTimeline); +bool pgsql_identify_system(PGSQL *pgsql, IdentifySystem *system); #endif /* PGSQL_TIMELINE_H */ diff --git a/tests/cdc-low-level/copydb.sh b/tests/cdc-low-level/copydb.sh index 83518f4cb..82d0b8fcb 100755 --- a/tests/cdc-low-level/copydb.sh +++ b/tests/cdc-low-level/copydb.sh @@ -161,5 +161,30 @@ test "${rows}" -eq 25000 # unpatched pgcopydb only syncs at COMMIT/EOF, which fails this assertion test "${syncs}" -ge 5 +TLIFILE=${SHAREDIR}/tli + +echo "recording timeline 2, as a source promoted before this run would" + +printf '2' > ${TLIFILE} + +psql -d ${PGCOPYDB_SOURCE_PGURI} -f /usr/src/pgcopydb/dml.sql + +lsn=`psql -At -d ${PGCOPYDB_SOURCE_PGURI} -c 'select pg_current_wal_lsn()'` + +set +e +pgcopydb stream receive --resume --endpos "${lsn}" > /tmp/timeline.log 2>&1 +rc=$? +set -e + +cat /tmp/timeline.log + +test "${rc}" -ne 0 + +grep -q "Source timeline changed from 2 to 1" /tmp/timeline.log + +test "$(cat ${TLIFILE})" = "2" + +rm -f ${TLIFILE} + # cleanup pgcopydb stream cleanup --verbose