[#23867] YSQL: Support postgres timestamp in YSQL clone AS OF syntax

Summary:
This diff adds support for PG's timestamptz type as the clone time parameter when cloning a
database as of a point in time:
```
CREATE DATABASE db2 TEMPLATE db1 AS OF '2024-09-10 21:52:46.782360';
```
The Unix microseconds timestamp format is still supported.
Jira: DB-12773

Test Plan: `./yb_build.sh release --cxx-test integration-tests_minicluster-snapshot-test --gtest_filter *CloneYsqlSyntax/*`

Reviewers: mhaddad, kfranz

Reviewed By: kfranz

Subscribers: jason, yql, ybase

Differential Revision: https://phorge.dev.yugabyte.com/D42966

Author: SrivastavaAnubhav

docs/content/preview/manage/backup-restore/instant-db-cloning.md

@@ -71,10 +71,12 @@ CREATE DATABASE clone_db TEMPLATE original_db;
 
 In this example, `clone_db` is created as a clone of `original_db`, and contains the latest schema and data of `original_db` as of current time.
 
-To create a clone of the original database at a specific point in time (within the history retention period specified when creating the snapshot schedule), you can specify the [Unix timestamp](https://www.unixtimestamp.com/) in microseconds using the `AS OF` option as follows:
+To create a clone of the original database at a specific point in time (within the history retention period specified when creating the snapshot schedule), you can specify a timestamp using the `AS OF` option. The timestamp may be either a [Unix timestamp](https://www.unixtimestamp.com/) in microseconds (as below), or a [PostgreSQL TIMESTAMP](https://www.postgresql.org/docs/current/datatype-datetime.html) in single quotes.
 
 ```sql
 CREATE DATABASE clone_db TEMPLATE original_db AS OF 1723146703674480;
+# Alternatively:
+CREATE DATABASE clone_db TEMPLATE original_db AS OF '2024-08-08 19:51:43.674480';
 ```
 
 ### Clone a YCQL keyspace

src/postgres/src/backend/commands/dbcommands.c

@@ -982,7 +982,33 @@ createdb(ParseState *pstate, const CreatedbStmt *stmt)
 	else
 		dbcolocated = YBColocateDatabaseByDefault();
 	if (dclonetime && dclonetime->arg)
-		dbclonetime = defGetInt64(dclonetime);
+	{
+		/*
+		 * This is a Float and not an Integer because Integer is too small to
+		 * contain a Unix epoch timestamp in microseconds (which is the format we
+		 * require).
+		 */
+		if (IsA(dclonetime->arg, Float))
+			dbclonetime = defGetInt64(dclonetime);
+		else if (IsA(dclonetime->arg, String))
+		{
+			const char *clone_time_str = defGetString(dclonetime);
+			TimestampTz clone_time = DirectFunctionCall3(timestamptz_in,
+					CStringGetDatum(clone_time_str),
+					ObjectIdGetDatum(InvalidOid),
+					Int32GetDatum(-1));
+			dbclonetime =
+					yb_timestamptz_to_micros_time_t(DatumGetTimestampTz(clone_time));
+		}
+		else
+		{
+			ereport(ERROR,
+					(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+					 errmsg("invalid clone time type: %s (must be a Unix microseconds "
+									"timestamp or a timestamptz-formatted string).",
+							nodeToString(dclonetime->arg))));
+		}
+	}
 	if (dcollversion)
 		dbcollversion = defGetString(dcollversion);
 

src/postgres/src/backend/utils/adt/timestamp.c

@@ -5940,3 +5940,17 @@ generate_series_timestamptz(PG_FUNCTION_ARGS)
 		SRF_RETURN_DONE(funcctx);
 	}
 }
+
+/*
+ * Same as timestamptz_to_time_t, but with microsecond precision.
+ */
+pg_time_t
+yb_timestamptz_to_micros_time_t(TimestampTz t)
+{
+	pg_time_t	result;
+
+	result = (pg_time_t) (t +
+						  ((POSTGRES_EPOCH_JDATE - UNIX_EPOCH_JDATE) * USECS_PER_DAY));
+
+	return result;
+}

src/postgres/src/include/utils/timestamp.h

@@ -114,4 +114,6 @@ extern int	date2isoyearday(int year, int mon, int mday);
 
 extern bool TimestampTimestampTzRequiresRewrite(void);
 
+extern pg_time_t yb_timestamptz_to_micros_time_t(TimestampTz t);
+
 #endif							/* TIMESTAMP_H */

src/yb/integration-tests/minicluster-snapshot-test.cc

@@ -718,42 +718,56 @@ TEST_P(PgCloneTestWithColocatedDBParam, YB_DISABLE_TEST_IN_SANITIZERS(CloneYsqlS
   const std::vector<std::tuple<int32_t, int32_t>> kRows = {{1, 10}, {2, 20}};
   ASSERT_OK(source_conn_->ExecuteFormat(
       "INSERT INTO t1 VALUES ($0, $1)", std::get<0>(kRows[0]), std::get<1>(kRows[0])));
-
   // Write a second row after recording the hybrid time.
-  auto ht = ASSERT_RESULT(GetCurrentTime()).ToInt64();
+  auto timestamp = ASSERT_RESULT(GetCurrentTime());
   ASSERT_OK(source_conn_->ExecuteFormat(
       "INSERT INTO t1 VALUES ($0, $1)", std::get<0>(kRows[1]), std::get<1>(kRows[1])));
 
-  // Perform the first clone operation to ht.
-  ASSERT_OK(source_conn_->ExecuteFormat(
-      "CREATE DATABASE $0 TEMPLATE $1 AS OF $2", kTargetNamespaceName1, kSourceNamespaceName, ht));
-
-  // Perform the second clone operation to clone the source DB using the current timestamp (AS OF is
-  // not specified)
+  // Clone to ht and verify clone only has the first row.
   ASSERT_OK(source_conn_->ExecuteFormat(
-      "CREATE DATABASE $0 TEMPLATE $1", kTargetNamespaceName2, kSourceNamespaceName));
-
-  // Verify source rows are unchanged.
-  auto rows = ASSERT_RESULT((source_conn_->FetchRows<int32_t, int32_t>("SELECT * FROM t1")));
-  ASSERT_VECTORS_EQ(rows, kRows);
-
-  // Verify first clone only has the first row.
+      "CREATE DATABASE $0 TEMPLATE $1 AS OF $2", kTargetNamespaceName1, kSourceNamespaceName,
+      timestamp.ToInt64()));
   // Use a scope here and below so we can drop the cloned databases after.
   {
     auto target_conn1 = ASSERT_RESULT(ConnectToDB(kTargetNamespaceName1));
     auto row = ASSERT_RESULT((target_conn1.FetchRow<int32_t, int32_t>("SELECT * FROM t1")));
     ASSERT_EQ(row, kRows[0]);
   }
+  ASSERT_OK(source_conn_->ExecuteFormat("DROP DATABASE $0", kTargetNamespaceName1));
 
-  // Verify second clone has both rows.
+  // Clone using the current timestamp (AS OF is not specified) and check that it has both rows.
+  ASSERT_OK(source_conn_->ExecuteFormat(
+      "CREATE DATABASE $0 TEMPLATE $1", kTargetNamespaceName2, kSourceNamespaceName));
   {
     auto target_conn2 = ASSERT_RESULT(ConnectToDB(kTargetNamespaceName2));
-    rows = ASSERT_RESULT((target_conn2.FetchRows<int32_t, int32_t>("SELECT * FROM t1")));
+    auto rows = ASSERT_RESULT((target_conn2.FetchRows<int32_t, int32_t>("SELECT * FROM t1")));
     ASSERT_VECTORS_EQ(rows, kRows);
   }
-
-  ASSERT_OK(source_conn_->ExecuteFormat("DROP DATABASE $0", kTargetNamespaceName1));
   ASSERT_OK(source_conn_->ExecuteFormat("DROP DATABASE $0", kTargetNamespaceName2));
+
+  // Clone using a timestamp in PG's timestamptz format. It should only have the first row.
+  const auto kTargetNamespaceName3 = "testdb_clone3";
+  ASSERT_OK(source_conn_->ExecuteFormat(
+      "CREATE DATABASE $0 TEMPLATE $1 AS OF '$2'", kTargetNamespaceName3, kSourceNamespaceName,
+      timestamp.ToHumanReadableTime()));
+  {
+    auto target_conn3 = ASSERT_RESULT(ConnectToDB(kTargetNamespaceName3));
+    auto row = ASSERT_RESULT((target_conn3.FetchRow<int32_t, int32_t>("SELECT * FROM t1")));
+    ASSERT_EQ(row, kRows[0]);
+  }
+  ASSERT_OK(source_conn_->ExecuteFormat("DROP DATABASE $0", kTargetNamespaceName3));
+
+  // Cloning using a timestamp in Unix seconds should not work.
+  const auto kTargetNamespaceName4 = "testdb_clone4";
+  auto status = source_conn_->ExecuteFormat(
+      "CREATE DATABASE $0 TEMPLATE $1 AS OF $2", kTargetNamespaceName4, kSourceNamespaceName,
+      timestamp.ToInt64() / 1000000);
+  ASSERT_NOK(status);
+  ASSERT_STR_CONTAINS(status.message().ToBuffer(), "invalid clone time type");
+
+  // Verify source rows are unchanged.
+  auto rows = ASSERT_RESULT((source_conn_->FetchRows<int32_t, int32_t>("SELECT * FROM t1")));
+  ASSERT_VECTORS_EQ(rows, kRows);
 }
 
 class TsDataSizeMetricsTest : public PgCloneTest {