to 3.0: clone with snapshot should match the snapshot level and owner (#22429)

A clone with a snapshot should match the snapshot's level and owner.

Approved by: @heni02, @XuPeng-SH, @aunjgr, @daviszhen

Author: gouhongshen

pkg/frontend/authenticate.go

@@ -1297,10 +1297,7 @@ const (
 
 	checkDatabaseWithOwnerFormat = `select dat_id, owner from mo_catalog.mo_database where datname = "%s" and account_id = %d;`
 
-	checkDatabaseTableFormat = `select t.rel_id from mo_catalog.mo_database d, mo_catalog.mo_tables t
-										where d.dat_id = t.reldatabase_id
-											and d.datname = "%s"
-											and t.relname = "%s";`
+	checkDatabaseTableFormat = `select rel_id from mo_catalog.mo_tables where relname = "%s" and reldatabase = "%s" and account_id = %d;`
 
 	//TODO:fix privilege_level string and obj_type string
 	//For object_type : table, privilege_level : *.*
@@ -1941,12 +1938,30 @@ func getSqlForCheckDatabaseWithOwner(ctx context.Context, dbName string, account
 	return fmt.Sprintf(checkDatabaseWithOwnerFormat, dbName, accountId), nil
 }
 
-func getSqlForCheckDatabaseTable(ctx context.Context, dbName, tableName string) (string, error) {
+func getSqlForCheckDatabaseTable(
+	ctx context.Context,
+	dbName string,
+	tableName string,
+) (string, error) {
+
 	err := inputNameIsInvalid(ctx, dbName, tableName)
 	if err != nil {
 		return "", err
 	}
-	return fmt.Sprintf(checkDatabaseTableFormat, dbName, tableName), nil
+
+	var (
+		account uint32
+	)
+
+	if v := ctx.Value(defines.TenantIDKey{}); v != nil {
+		account = v.(uint32)
+	} else {
+		return "", moerr.NewInternalErrorNoCtx("no account id found in the ctx")
+	}
+
+	// we need the account id here to filter out the same dbName and tableName that exist in the
+	// different accounts.
+	return fmt.Sprintf(checkDatabaseTableFormat, tableName, dbName, account), nil
 }
 
 func getSqlForDeleteRole(roleId int64) []string {
@@ -6714,6 +6729,8 @@ func authenticateUserCanExecuteStatementWithObjectTypeAccountAndDatabase(ctx con
 			}
 			tbName := string(st.Names[0].ObjectName)
 			return checkRoleWhetherTableOwner(ctx, ses, dbName, tbName, ok)
+		case *tree.CloneTable, *tree.CloneDatabase:
+			return true, stats, nil
 		}
 	}
 	return ok, stats, nil

pkg/frontend/authenticate_test.go

@@ -5134,6 +5134,8 @@ func Test_doGrantPrivilege(t *testing.T) {
 			},
 		}
 
+		ctx := context.WithValue(context.TODO(), defines.TenantIDKey{}, uint32(sysAccountID))
+
 		for _, stmt := range stmts {
 			priv := determinePrivilegeSetOfStatement(stmt)
 			ses := newSes(priv, ctrl)
@@ -5165,14 +5167,14 @@ func Test_doGrantPrivilege(t *testing.T) {
 				bh.sql2result[sql] = mrs
 			} else if stmt.Level.Level == tree.PRIVILEGE_LEVEL_TYPE_TABLE ||
 				stmt.Level.Level == tree.PRIVILEGE_LEVEL_TYPE_DATABASE_TABLE {
-				sql, _ := getSqlForCheckDatabaseTable(context.TODO(), dbName, tableName)
+				sql, _ := getSqlForCheckDatabaseTable(ctx, dbName, tableName)
 				mrs := newMrsForCheckDatabaseTable([][]interface{}{
 					{0},
 				})
 				bh.sql2result[sql] = mrs
 			}
 
-			_, objId, err := checkPrivilegeObjectTypeAndPrivilegeLevel(context.TODO(), ses, bh, stmt.ObjType, *stmt.Level)
+			_, objId, err := checkPrivilegeObjectTypeAndPrivilegeLevel(ctx, ses, bh, stmt.ObjType, *stmt.Level)
 			convey.So(err, convey.ShouldBeNil)
 
 			for _, p := range stmt.Privileges {
@@ -5436,6 +5438,8 @@ func Test_doRevokePrivilege(t *testing.T) {
 			},
 		}
 
+		ctx := context.WithValue(context.TODO(), defines.TenantIDKey{}, uint32(sysAccountID))
+
 		for _, stmt := range stmts {
 			priv := determinePrivilegeSetOfStatement(stmt)
 			ses := newSes(priv, ctrl)
@@ -5467,14 +5471,14 @@ func Test_doRevokePrivilege(t *testing.T) {
 				bh.sql2result[sql] = mrs
 			} else if stmt.Level.Level == tree.PRIVILEGE_LEVEL_TYPE_TABLE ||
 				stmt.Level.Level == tree.PRIVILEGE_LEVEL_TYPE_DATABASE_TABLE {
-				sql, _ := getSqlForCheckDatabaseTable(context.TODO(), dbName, tableName)
+				sql, _ := getSqlForCheckDatabaseTable(ctx, dbName, tableName)
 				mrs := newMrsForCheckDatabaseTable([][]interface{}{
 					{0},
 				})
 				bh.sql2result[sql] = mrs
 			}
 
-			_, objId, err := checkPrivilegeObjectTypeAndPrivilegeLevel(context.TODO(), ses, bh, stmt.ObjType, *stmt.Level)
+			_, objId, err := checkPrivilegeObjectTypeAndPrivilegeLevel(ctx, ses, bh, stmt.ObjType, *stmt.Level)
 			convey.So(err, convey.ShouldBeNil)
 
 			for _, p := range stmt.Privileges {

pkg/frontend/snapshot.go

@@ -1627,6 +1627,11 @@ func doResolveSnapshotWithSnapshotName(ctx context.Context, ses FeSession, snaps
 			TenantName: record.accountName,
 			TenantID:   accountId,
 		},
+		ExtraInfo: &pbplan.SnapshotExtraInfo{
+			Level: record.level,
+			ObjId: record.objId,
+			Name:  record.snapshotName,
+		},
 	}, nil
 }