make sure the snapshot read TS greater than the TXN TS when create snapshot or clone.

pkg/frontend/clone.go

@@ -19,12 +19,15 @@ import (
 	"fmt"
 	"regexp"
 	"strings"
+	"time"
 
 	"github.com/matrixorigin/matrixone/pkg/common/moerr"
 	"github.com/matrixorigin/matrixone/pkg/defines"
 	plan2 "github.com/matrixorigin/matrixone/pkg/pb/plan"
+	"github.com/matrixorigin/matrixone/pkg/pb/timestamp"
 	"github.com/matrixorigin/matrixone/pkg/sql/parsers/tree"
 	"github.com/matrixorigin/matrixone/pkg/sql/plan"
+	"github.com/matrixorigin/matrixone/pkg/txn/client"
 )
 
 func getOpAndToAccountId(
@@ -182,6 +185,8 @@ func handleCloneDatabase(
 		sortedFkTbls  []string
 		fkTableMap    map[string]*tableInfo
 		snapCondition string
+
+		snapshotTS int64
 	)
 
 	bh = ses.GetBackgroundExec(reqCtx)
@@ -235,14 +240,36 @@ func handleCloneDatabase(
 		return err
 	}
 
+	// consider the following example:
+	// (within a session)
+	//   ...
+	// insert into t1 values (1) ---> commit ts (P2-L3)
+	// insert into t1 values (2) ---> commit ts (P2-L3)
+	// create table t2 clone t1 ---> the read snapshot ts is P2.
+	//
+	// limited by the format for the snapshot read TS, the logic TS is truncated,
+	// so in this example, the clone cannot read the newly inserted data.
+	//
+	// so we try to increase the txn physical ts here to make sure the snapshot TS
+	// the clone will get is greater than P2.
+	if snapshotTS, err = tryToIncreaseTxnPhysicalTS(
+		reqCtx, ses.proc.GetTxnOperator(),
+	); err != nil {
+		return err
+	}
+
 	ctx2 = defines.AttachAccountId(reqCtx, toAccountId)
 
 	cloneTable := func(dstDb, dstTbl, srcDb, srcTbl string) error {
 		sql := fmt.Sprintf(
-			"create table `%s`.`%s` clone `%s`.`%s`", dstDb, dstTbl, srcDb, srcTbl)
+			"create table `%s`.`%s` clone `%s`.`%s`",
+			dstDb, dstTbl, srcDb, srcTbl,
+		)
 
 		if snapCondition != "" {
 			sql = sql + " " + snapCondition
+		} else {
+			sql = sql + fmt.Sprintf(" {MO_TS = %d}", snapshotTS)
 		}
 
 		if err = bh.ExecRestore(ctx2, sql, opAccountId, toAccountId); err != nil {
@@ -318,3 +345,33 @@ func handleCloneDatabase(
 
 	return nil
 }
+
+func tryToIncreaseTxnPhysicalTS(
+	ctx context.Context, txnOp client.TxnOperator,
+) (updatedPhysical int64, err error) {
+
+	curTxnPhysicalTS := txnOp.SnapshotTS().PhysicalTime
+
+	if ctx.Value(defines.TenantIDKey{}) == nil {
+		return curTxnPhysicalTS, nil
+	}
+
+	// a slight increase added to the physical to make sure
+	// the updated ts is greater than the old txn timestamp (physical + logic)
+	curTxnPhysicalTS += int64(time.Microsecond)
+	if err = txnOp.UpdateSnapshot(ctx, timestamp.Timestamp{
+		PhysicalTime: curTxnPhysicalTS,
+	}); err != nil {
+		return
+	}
+
+	updatedPhysical = txnOp.SnapshotTS().PhysicalTime
+	if updatedPhysical <= curTxnPhysicalTS {
+		return 0, moerr.NewInternalErrorNoCtxf("try to update the snapshot ts failed in clone database")
+	}
+
+	// return a nanosecond precision
+	updatedPhysical -= int64(time.Nanosecond)
+
+	return updatedPhysical, nil
+}

pkg/frontend/snapshot.go

@@ -228,14 +228,25 @@ func doCreateSnapshot(ctx context.Context, ses *Session, stmt *tree.CreateSnapSh
 	}
 	snapshotId = newUUid.String()
 
+	var (
+		snapshotTS int64
+	)
+
+	// refer to the `handleCloneDatabase`
+	if snapshotTS, err = tryToIncreaseTxnPhysicalTS(
+		ctx, ses.proc.GetTxnOperator(),
+	); err != nil {
+		return err
+	}
+
 	// 3. get database name , table name  and objId according to the snapshot level
 	switch snapshotLevel {
 	case tree.SNAPSHOTLEVELCLUSTER:
 		sql, err = getSqlForCreateSnapshot(
 			ctx,
 			snapshotId,
 			snapshotName,
-			time.Now().UTC().UnixNano(),
+			snapshotTS,
 			snapshotLevel.String(),
 			"",
 			"",
@@ -296,7 +307,7 @@ func doCreateSnapshot(ctx context.Context, ses *Session, stmt *tree.CreateSnapSh
 			ctx,
 			snapshotId,
 			snapshotName,
-			time.Now().UTC().UnixNano(),
+			snapshotTS,
 			snapshotLevel.String(),
 			snapshotForAccount,
 			"",
@@ -350,7 +361,7 @@ func doCreateSnapshot(ctx context.Context, ses *Session, stmt *tree.CreateSnapSh
 			ctx,
 			snapshotId,
 			snapshotName,
-			time.Now().UTC().UnixNano(),
+			snapshotTS,
 			snapshotLevel.String(),
 			currentAccount,
 			databaseName,
@@ -414,7 +425,7 @@ func doCreateSnapshot(ctx context.Context, ses *Session, stmt *tree.CreateSnapSh
 			ctx,
 			snapshotId,
 			snapshotName,
-			time.Now().UTC().UnixNano(),
+			snapshotTS,
 			snapshotLevel.String(),
 			currentAccount,
 			databaseName,