fix(restore): Set kv version to restoreTs for rolled up keys and sche…

…ma keys (#7930)

The kv version should be set to restore timestamp for rolled-up keys 
and schema keys as well.

worker/restore_map.go

@@ -371,7 +371,9 @@ func (m *mapper) processReqCh(ctx context.Context) error {
 					return err
 				}
 				for _, kv := range kvs {
-					if err := toBuffer(kv, kv.Version); err != nil {
+					version := kv.Version
+					kv.Version = m.restoreTs
+					if err := toBuffer(kv, version); err != nil {
 						return err
 					}
 				}
@@ -448,8 +450,10 @@ func (m *mapper) processReqCh(ctx context.Context) error {
 			kv.StreamId = 0
 			// Schema and type keys are not stored in an intermediate format so their
 			// value can be written as is.
+			version := kv.Version
+			kv.Version = m.restoreTs
 			kv.Key = restoreKey
-			if err := toBuffer(kv, kv.Version); err != nil {
+			if err := toBuffer(kv, version); err != nil {
 				return err
 			}