skip TIMELINE_HISTORY on the logical replication connection

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
  */

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
  */

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,24 +2922,19 @@ stream_write_context(StreamSpecs *specs, LogicalStreamClient *stream)
 
 	bytes = sformat(tli, sizeof(tli), "%d", system->timeline);
 
-	if (!write_file(tli, bytes, specs->paths.tlifile))
+	if (!stream_check_timeline(specs, system->timeline))
 	{
-		/* errors have already been logged */
 		return false;
 	}
 
-	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))
+	if (!write_file(tli, bytes, specs->paths.tlifile))
 	{
 		/* errors have already been logged */
 		return false;
 	}
 
+	log_debug("Wrote tli %s timeline file \"%s\"", tli, specs->paths.tlifile);
+
 	return true;
 }
 
@@ -3024,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;
 }

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);

src/bin/pgcopydb/pgsql.h

@@ -379,38 +379,24 @@ 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.
  */
 #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
 {
 	uint64_t identifier;
 	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];