compare: implement --snapshot option for source database

docs/ref/pgcopydb_compare.rst

@@ -45,6 +45,7 @@ of tables, indexes, constraints and sequences there.
      --source         Postgres URI to the source database
      --target         Postgres URI to the target database
      --dir            Work directory to use
+     --snapshot       Use snapshot obtained with pg_export_snapshot on the source database
 
 
 .. _pgcopydb_compare_data:
@@ -92,13 +93,14 @@ Running such a query on a large table can take a lot of time.
      --target         Postgres URI to the target database
      --dir            Work directory to use
      --json           Format the output using JSON
+     --snapshot       Use snapshot obtained with pg_export_snapshot on the source database
 
 
 Options
 -------
 
-The following options are available to ``pgcopydb create`` and ``pgcopydb
-drop`` subcommands:
+The following options are available to ``pgcopydb compare schema`` and ``pgcopydb
+compare data`` subcommands:
 
 --source
 
@@ -126,6 +128,11 @@ drop`` subcommands:
   The output of the command is formatted in JSON, when supported. Ignored
   otherwise.
 
+--snapshot
+
+  Instead of comparing the live data pgcopydb will use an already exported
+  snapshot on the source database for comparison.
+
 --verbose
 
   Increase current verbosity. The default level of verbosity is INFO. In

src/bin/pgcopydb/cli_compare.c

@@ -33,7 +33,8 @@ static CommandLine compare_schema_command =
 		" --source ... ",
 		"  --source         Postgres URI to the source database\n"
 		"  --target         Postgres URI to the target database\n"
-		"  --dir            Work directory to use\n",
+		"  --dir            Work directory to use\n"
+		"  --snapshot       Use snapshot obtained with pg_export_snapshot on the source database\n",
 		cli_compare_getopts,
 		cli_compare_schema);
 
@@ -45,7 +46,8 @@ static CommandLine compare_data_command =
 		"  --source         Postgres URI to the source database\n"
 		"  --target         Postgres URI to the target database\n"
 		"  --dir            Work directory to use\n"
-		"  --json           Format the output using JSON\n",
+		"  --json           Format the output using JSON\n"
+		"  --snapshot       Use snapshot obtained with pg_export_snapshot on the source database\n",
 		cli_compare_getopts,
 		cli_compare_data);
 
@@ -77,6 +79,7 @@ cli_compare_getopts(int argc, char **argv)
 		{ "jobs", required_argument, NULL, 'j' },
 		{ "table-jobs", required_argument, NULL, 'j' },
 		{ "json", no_argument, NULL, 'J' },
+		{ "snapshot", required_argument, NULL, 'N' },
 		{ "version", no_argument, NULL, 'V' },
 		{ "verbose", no_argument, NULL, 'v' },
 		{ "notice", no_argument, NULL, 'v' },
@@ -199,6 +202,13 @@ cli_compare_getopts(int argc, char **argv)
 				break;
 			}
 
+			case 'N':
+			{
+				strlcpy(options.snapshot, optarg, sizeof(options.snapshot));
+				log_trace("--snapshot %s", options.snapshot);
+				break;
+			}
+
 			case 'd':
 			{
 				verboseCount = 3;
@@ -279,9 +289,10 @@ cli_compare_schema(int argc, char **argv)
 	bool service = true;
 	char *serviceName = "snapshot";
 
-	/* pretend that --resume --not-consistent have been used */
+	/* pretend that --resume has been used */
 	compareOptions.resume = true;
-	compareOptions.notConsistent = true;
+	/* pretend that --not-consistent has been used (unless a snapshot is used) */
+	compareOptions.notConsistent = IS_EMPTY_STRING_BUFFER(compareOptions.snapshot);
 
 	if (!copydb_init_workdir(&copySpecs,
 							 dir,
@@ -301,11 +312,23 @@ cli_compare_schema(int argc, char **argv)
 		exit(EXIT_CODE_INTERNAL_ERROR);
 	}
 
+	if (!IS_EMPTY_STRING_BUFFER(compareOptions.snapshot) && !copydb_set_snapshot(&copySpecs))
+	{
+		/* errors have already been logged */
+		exit(EXIT_CODE_INTERNAL_ERROR);
+	}
+
 	if (!compare_schemas(&copySpecs))
 	{
 		log_fatal("Comparing the schemas failed, see above for details");
 		exit(EXIT_CODE_INTERNAL_ERROR);
 	}
+
+	if (!IS_EMPTY_STRING_BUFFER(compareOptions.snapshot))
+	{
+		/* close our snapshot: commit transaction and finish connection */
+		(void) copydb_close_snapshot(&copySpecs);
+	}
 }
 
 
@@ -328,9 +351,10 @@ cli_compare_data(int argc, char **argv)
 	bool service = true;
 	char *serviceName = "snapshot";
 
-	/* pretend that --resume --not-consistent have been used */
+	/* pretend that --resume has been used */
 	compareOptions.resume = true;
-	compareOptions.notConsistent = true;
+	/* pretend that --not-consistent has been used (unless a snapshot is used) */
+	compareOptions.notConsistent = IS_EMPTY_STRING_BUFFER(compareOptions.snapshot);
 
 	if (!copydb_init_workdir(&copySpecs,
 							 dir,
@@ -350,6 +374,12 @@ cli_compare_data(int argc, char **argv)
 		exit(EXIT_CODE_INTERNAL_ERROR);
 	}
 
+	if (!IS_EMPTY_STRING_BUFFER(compareOptions.snapshot) && !copydb_set_snapshot(&copySpecs))
+	{
+		/* errors have already been logged */
+		exit(EXIT_CODE_INTERNAL_ERROR);
+	}
+
 	if (!compare_data(&copySpecs))
 	{
 		log_fatal("Failed to compute checksums, see above for details");
@@ -358,6 +388,12 @@ cli_compare_data(int argc, char **argv)
 
 	SourceTableArray *tableArray = &(copySpecs.catalog.sourceTableArray);
 
+	if (!IS_EMPTY_STRING_BUFFER(compareOptions.snapshot))
+	{
+		/* close our snapshot: commit transaction and finish connection */
+		(void) copydb_close_snapshot(&copySpecs);
+	}
+
 	if (outputJSON)
 	{
 		JSON_Value *js = json_value_init_array();

src/bin/pgcopydb/compare.c

@@ -410,6 +410,27 @@ compare_table(CopyDataSpec *copySpecs, SourceTable *source)
 		return false;
 	}
 
+	if (!IS_EMPTY_STRING_BUFFER(copySpecs->sourceSnapshot.snapshot))
+	{
+		log_debug("compare_table: %s using snapshot: %s", source->relname, copySpecs->sourceSnapshot.snapshot);
+
+		if (!pgsql_set_transaction(&src, ISOLATION_REPEATABLE_READ, false, true))
+		{
+			/* errors have already been logged */
+			(void) pgsql_finish(&src);
+			return false;
+		}
+
+		if (!pgsql_set_snapshot(&src, copySpecs->sourceSnapshot.snapshot))
+		{
+			/* errors have already been logged */
+			(void) pgsql_finish(&src);
+			return false;
+		}
+
+		copySpecs->sourceSnapshot.state = SNAPSHOT_STATE_SET;
+	}
+
 	if (!pgsql_init(&dst, dsn->target_pguri, PGSQL_CONN_TARGET))
 	{
 		/* errors have already been logged */
@@ -801,6 +822,10 @@ compare_fetch_schemas(CopyDataSpec *copySpecs,
 	*sourceSpecs = *copySpecs;
 	*targetSpecs = *copySpecs;
 
+	/* unset sourceSnapshot snapshot for target */
+	targetSpecs->consistent = false;
+	(void) memset(&(targetSpecs->sourceSnapshot.snapshot), 0, sizeof(targetSpecs->sourceSnapshot.snapshot));
+
 	ConnStrings *sourceConnStrings = &(sourceSpecs->connStrings);
 
 	sourceConnStrings->target_pguri = NULL;