Store timelines in internal catalogs

src/bin/pgcopydb/catalog.c

@@ -183,7 +183,10 @@ 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)"
+	" 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)"
 };
 
 
@@ -434,7 +437,8 @@ 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 sentinel",
+	"drop table if exists timeline_history"
 };
 
 
@@ -7700,6 +7704,169 @@ catalog_count_summary_done_fetch(SQLiteQuery *query)
 }
 
 
+/*
+ * catalog_add_timeline_history inserts a timeline history entry to our
+ * internal catalogs database.
+ */
+bool
+catalog_add_timeline_history(void *ctx, TimelineHistoryEntry *entry)
+{
+	TimelineHistoryContext *context = (TimelineHistoryContext *) ctx;
+	DatabaseCatalog *catalog = context->source;
+	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 fetches the current TimelineHistoryEntry
+ * from our catalogs.
+ */
+bool
+catalog_lookup_timeline(DatabaseCatalog *catalog,
+						int tli,
+						TimelineHistoryEntry *entry)
+{
+	sqlite3 *db = catalog->db;
+
+	if (db == NULL)
+	{
+		log_error("BUG: catalog_lookup_timeline: 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;
+}
+
+
+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

@@ -622,6 +622,16 @@ bool catalog_count_summary_done(DatabaseCatalog *catalog,
 bool catalog_count_summary_done_fetch(SQLiteQuery *query);
 
 
+/*
+ * Logical decoding
+ */
+bool catalog_add_timeline_history(void *context,
+								  TimelineHistoryEntry *entry);
+bool catalog_lookup_timeline(DatabaseCatalog *catalog,
+							 int tli,
+							 TimelineHistoryEntry *entry);
+bool catalog_timeline_history_fetch(SQLiteQuery *query);
+
 /*
  * Internal tooling for catalogs management
  */

src/bin/pgcopydb/follow.c

@@ -419,12 +419,11 @@ follow_prepare_mode_switch(StreamSpecs *streamSpecs,
 
 	if (streamSpecs->system.timeline == 0)
 	{
-		if (!stream_read_context(&(streamSpecs->paths),
-								 &(streamSpecs->system),
-								 &(streamSpecs->WalSegSz)))
+		if (!stream_read_context(streamSpecs))
 		{
 			log_error("Failed to read the streaming context information "
-					  "from the source database, see above for details");
+					  "from the source database and internal catalogs, "
+					  "see above for details");
 			return false;
 		}
 	}

src/bin/pgcopydb/ld_apply.c

@@ -208,12 +208,11 @@ stream_apply_setup(StreamSpecs *specs, StreamApplyContext *context)
 
 	if (specs->system.timeline == 0)
 	{
-		if (!stream_read_context(&(specs->paths),
-								 &(specs->system),
-								 &(specs->WalSegSz)))
+		if (!stream_read_context(specs))
 		{
 			log_error("Failed to read the streaming context information "
-					  "from the source database, see above for details");
+					  "from the source database and internal catalogs, "
+					  "see above for details");
 			return false;
 		}
 	}

src/bin/pgcopydb/ld_stream.c

@@ -490,7 +490,10 @@ startLogicalStreaming(StreamSpecs *specs)
 				  OutputPluginToString(specs->slot.plugin),
 				  specs->pluginOptions.count);
 
-		if (!pgsql_start_replication(&stream))
+		/* prepare the context for timeline history */
+		TimelineHistoryContext timelineContext = { .source = specs->sourceDB };
+
+		if (!pgsql_start_replication(&stream, &timelineContext))
 		{
 			/* errors have already been logged */
 			return false;
@@ -2613,8 +2616,7 @@ stream_fetch_current_lsn(uint64_t *lsn,
 
 
 /*
- * stream_write_context writes the wal_segment_size and timeline history to
- * files.
+ * stream_write_context writes the wal_segment_size and tli to files.
  */
 bool
 stream_write_context(StreamSpecs *specs, LogicalStreamClient *stream)
@@ -2652,16 +2654,6 @@ stream_write_context(StreamSpecs *specs, LogicalStreamClient *stream)
 
 	log_debug("Wrote tli %s timeline file \"%s\"", tli, specs->paths.tlifile);
 
-	if (!write_file(system->timelines.content,
-					strlen(system->timelines.content),
-					specs->paths.tlihistfile))
-	{
-		/* errors have already been logged */
-		return false;
-	}
-
-	log_debug("Wrote timeline history file \"%s\"", specs->paths.tlihistfile);
-
 	return true;
 }
 
@@ -2677,7 +2669,6 @@ stream_cleanup_context(StreamSpecs *specs)
 
 	success = success && unlink_file(specs->paths.walsegsizefile);
 	success = success && unlink_file(specs->paths.tlifile);
-	success = success && unlink_file(specs->paths.tlihistfile);
 
 	/* reset the timeline, so that we always read from the disk */
 	specs->system.timeline = 0;
@@ -2691,14 +2682,14 @@ stream_cleanup_context(StreamSpecs *specs)
  * wal_segment_size and timeline history.
  */
 bool
-stream_read_context(CDCPaths *paths,
-					IdentifySystem *system,
-					uint32_t *WalSegSz)
+stream_read_context(StreamSpecs *specs)
 {
+	CDCPaths *paths = &(specs->paths);
+	IdentifySystem *system = &(specs->system);
+	uint32_t *WalSegSz = &(specs->WalSegSz);
+
 	char *wal_segment_size = NULL;
 	char *tli = NULL;
-	char *history = NULL;
-
 	long size = 0L;
 
 	/*
@@ -2727,8 +2718,7 @@ stream_read_context(CDCPaths *paths,
 		}
 
 		if (file_exists(paths->walsegsizefile) &&
-			file_exists(paths->tlifile) &&
-			file_exists(paths->tlihistfile))
+			file_exists(paths->tlifile))
 		{
 			/*
 			 * File did exist, but might have been deleted now (race condition
@@ -2742,9 +2732,6 @@ stream_read_context(CDCPaths *paths,
 			success = success &&
 					  read_file(paths->tlifile, &tli, &size);
 
-			success = success &&
-					  read_file(paths->tlihistfile, &history, &size);
-
 			if (success)
 			{
 				/* success: break out of the retry loop */
@@ -2765,8 +2752,7 @@ stream_read_context(CDCPaths *paths,
 
 	/* did retry policy expire before the files are created? */
 	if (!(file_exists(paths->walsegsizefile) &&
-		  file_exists(paths->tlifile) &&
-		  file_exists(paths->tlihistfile)))
+		  file_exists(paths->tlifile)))
 	{
 		log_error("Failed to read stream context file: retry policy expired");
 		return false;
@@ -2787,12 +2773,12 @@ stream_read_context(CDCPaths *paths,
 		return false;
 	}
 
-	if (!parseTimeLineHistory(paths->tlihistfile, history, system))
+	DatabaseCatalog *source = specs->sourceDB;
+	if (!catalog_lookup_timeline(source, system->timeline, &system->currentTimeline))
 	{
 		/* errors have already been logged */
 		return false;
 	}
 
-
 	return true;
 }

src/bin/pgcopydb/ld_stream.h

@@ -561,9 +561,7 @@ bool stream_fetch_current_lsn(uint64_t *lsn,
 
 bool stream_write_context(StreamSpecs *specs, LogicalStreamClient *stream);
 bool stream_cleanup_context(StreamSpecs *specs);
-bool stream_read_context(CDCPaths *paths,
-						 IdentifySystem *system,
-						 uint32_t *WalSegSz);
+bool stream_read_context(StreamSpecs *specs);
 
 StreamAction StreamActionFromChar(char action);
 char * StreamActionToString(StreamAction action);