Update code to support namespace mapping when perform the in-place restore with block data mover (#10461)

* Update code to support namespace mapping when perform the in-place restore with block data mover

Update code to support namespace mapping when perform the in-p
lace restore with block data mover

Signed-off-by: Wenkai Yin(尹文开) <yinw@vmware.com>

changelogs/unreleased/10461-ywk253100

@@ -0,0 +1 @@
+Update code to support namespace mapping when perform the in-p lace restore with block data mover

config/crd/v1/bases/velero.io_restores.yaml

@@ -68,6 +68,11 @@ spec:
                   BackupName is the unique name of the Velero backup to restore
                   from.
                 type: string
+              csiSnapshotTimeout:
+                description: |-
+                  CSISnapshotTimeout specifies the time used to wait for CSI VolumeSnapshot ready to use during creation, before returning error as timeout.
+                  The default value is 30 minute.
+                type: string
               excludedNamespaces:
                 description: |-
                   ExcludedNamespaces contains a list of namespaces that are not

config/crd/v2alpha1/bases/velero.io_datadownloads.yaml

@@ -99,6 +99,10 @@ spec:
                   used to do the incremental restore.
                 nullable: true
                 properties:
+                  cleanUp:
+                    description: CleanUp indicates request to clean up the volume
+                      snapshot after the backup/restore is completed.
+                    type: boolean
                   driver:
                     description: Driver is the driver used by the VolumeSnapshotContent
                     type: string

config/crd/v2alpha1/bases/velero.io_datauploads.yaml

@@ -95,6 +95,10 @@ spec:
                   of the CSI snapshot.
                 nullable: true
                 properties:
+                  cleanUp:
+                    description: CleanUp indicates request to clean up the volume
+                      snapshot after the backup/restore is completed.
+                    type: boolean
                   driver:
                     description: Driver is the driver used by the VolumeSnapshotContent
                     type: string

pkg/apis/velero/v1/restore_types.go

@@ -120,6 +120,11 @@ type RestoreSpec struct {
 	// +nullable
 	ExistingVolumeDataPolicy VolumeDataPolicyType `json:"existingVolumeDataPolicy,omitempty"`
 
+	// CSISnapshotTimeout specifies the time used to wait for CSI VolumeSnapshot ready to use during creation, before returning error as timeout.
+	// The default value is 30 minute.
+	// +optional
+	CSISnapshotTimeout metav1.Duration `json:"csiSnapshotTimeout,omitempty"`
+
 	// ItemOperationTimeout specifies the time used to wait for RestoreItemAction operations
 	// The default value is 4 hour.
 	// +optional

pkg/apis/velero/v1/zz_generated.deepcopy.go

@@ -1426,6 +1426,7 @@ func (in *RestoreSpec) DeepCopyInto(out *RestoreSpec) {
 		**out = **in
 	}
 	in.Hooks.DeepCopyInto(&out.Hooks)
+	out.CSISnapshotTimeout = in.CSISnapshotTimeout
 	out.ItemOperationTimeout = in.ItemOperationTimeout
 	if in.ResourceModifier != nil {
 		in, out := &in.ResourceModifier, &out.ResourceModifier

pkg/apis/velero/v2alpha1/data_upload_types.go

@@ -97,6 +97,10 @@ type CSISnapshotSpec struct {
 	// Driver is the driver used by the VolumeSnapshotContent
 	// +optional
 	Driver string `json:"driver,omitempty"`
+
+	// CleanUp indicates request to clean up the volume snapshot after the backup/restore is completed.
+	// +optional
+	CleanUp bool `json:"cleanUp,omitempty"`
 }
 
 // DataUploadPhase represents the lifecycle phase of a DataUpload.

pkg/builder/restore_builder.go

@@ -179,6 +179,12 @@ func (b *RestoreBuilder) ItemOperationTimeout(timeout time.Duration) *RestoreBui
 	return b
 }
 
+// CSISnapshotTimeout sets the Restore's CSISnapshotTimeout
+func (b *RestoreBuilder) CSISnapshotTimeout(timeout time.Duration) *RestoreBuilder {
+	b.object.Spec.CSISnapshotTimeout.Duration = timeout
+	return b
+}
+
 // ResourcePoliciesConfigmap sets the Restore's resource policies configmap.
 func (b *RestoreBuilder) ResourcePoliciesConfigmap(name string) *RestoreBuilder {
 	b.object.Spec.ResourcePolicy = &corev1api.TypedLocalObjectReference{

pkg/builder/restore_builder_test.go

@@ -18,6 +18,7 @@ package builder
 
 import (
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/assert"
 )
@@ -34,3 +35,11 @@ func TestRestoreBuilder_ResourcePoliciesConfigmap(t *testing.T) {
 	assert.Equal(t, "my-policy-cm", restore.Spec.ResourcePolicy.Name)
 	assert.Equal(t, (*string)(nil), restore.Spec.ResourcePolicy.APIGroup)
 }
+
+func TestRestoreBuilder_CSISnapshotTimeout(t *testing.T) {
+	restore := ForRestore("velero", "my-restore").
+		CSISnapshotTimeout(25 * time.Minute).
+		Result()
+
+	assert.Equal(t, 25*time.Minute, restore.Spec.CSISnapshotTimeout.Duration)
+}

pkg/cmd/server/config/config.go

@@ -43,8 +43,9 @@ const (
 	// the default TTL for a backup
 	defaultBackupTTL = 30 * 24 * time.Hour
 
-	defaultCSISnapshotTimeout   = 10 * time.Minute
-	defaultItemOperationTimeout = 4 * time.Hour
+	defaultBackupCSISnapshotTimeout  = 10 * time.Minute
+	defaultRestoreCSISnapshotTimeout = 30 * time.Minute
+	defaultItemOperationTimeout      = 4 * time.Hour
 
 	resourceTimeout = defaultResourceTerminatingTimeout
 
@@ -159,7 +160,8 @@ type Config struct {
 	DefaultBackupTTL                    time.Duration
 	DefaultVGSLabelKey                  string
 	StoreValidationFrequency            time.Duration
-	DefaultCSISnapshotTimeout           time.Duration
+	DefaultBackupCSISnapshotTimeout     time.Duration
+	DefaultRestoreCSISnapshotTimeout    time.Duration
 	DefaultItemOperationTimeout         time.Duration
 	ResourceTimeout                     time.Duration
 	RestoreResourcePriorities           types.Priorities
@@ -193,35 +195,36 @@ type Config struct {
 
 func GetDefaultConfig() *Config {
 	config := &Config{
-		PluginDir:                      "/plugins",
-		MetricsAddress:                 defaultMetricsAddress,
-		DefaultBackupLocation:          "default",
-		DefaultVolumeSnapshotLocations: flag.NewMap().WithKeyValueDelimiter(':'),
-		BackupSyncPeriod:               defaultBackupSyncPeriod,
-		DefaultBackupTTL:               defaultBackupTTL,
-		DefaultVGSLabelKey:             velerov1api.DefaultVGSLabelKey,
-		DefaultCSISnapshotTimeout:      defaultCSISnapshotTimeout,
-		DefaultItemOperationTimeout:    defaultItemOperationTimeout,
-		ResourceTimeout:                resourceTimeout,
-		StoreValidationFrequency:       defaultStoreValidationFrequency,
-		PodVolumeOperationTimeout:      defaultPodVolumeOperationTimeout,
-		RestoreResourcePriorities:      defaultRestorePriorities,
-		ClientQPS:                      defaultClientQPS,
-		ClientBurst:                    defaultClientBurst,
-		ClientPageSize:                 defaultClientPageSize,
-		ProfilerAddress:                defaultProfilerAddress,
-		ResourceTerminatingTimeout:     defaultResourceTerminatingTimeout,
-		LogLevel:                       logging.LogLevelFlag(logrus.InfoLevel),
-		LogFormat:                      logging.NewFormatFlag(),
-		DefaultVolumesToFsBackup:       podvolumeconfigs.DefaultVolumesToFsBackup,
-		UploaderType:                   uploader.KopiaType,
-		MaxConcurrentK8SConnections:    defaultMaxConcurrentK8SConnections,
-		DefaultSnapshotMoveData:        false,
-		DisableInformerCache:           defaultDisableInformerCache,
-		ScheduleSkipImmediately:        false,
-		CredentialsDirectory:           credentials.DefaultStoreDirectory(),
-		ItemBlockWorkerCount:           DefaultItemBlockWorkerCount,
-		ConcurrentBackups:              DefaultConcurrentBackups,
+		PluginDir:                        "/plugins",
+		MetricsAddress:                   defaultMetricsAddress,
+		DefaultBackupLocation:            "default",
+		DefaultVolumeSnapshotLocations:   flag.NewMap().WithKeyValueDelimiter(':'),
+		BackupSyncPeriod:                 defaultBackupSyncPeriod,
+		DefaultBackupTTL:                 defaultBackupTTL,
+		DefaultVGSLabelKey:               velerov1api.DefaultVGSLabelKey,
+		DefaultBackupCSISnapshotTimeout:  defaultBackupCSISnapshotTimeout,
+		DefaultRestoreCSISnapshotTimeout: defaultRestoreCSISnapshotTimeout,
+		DefaultItemOperationTimeout:      defaultItemOperationTimeout,
+		ResourceTimeout:                  resourceTimeout,
+		StoreValidationFrequency:         defaultStoreValidationFrequency,
+		PodVolumeOperationTimeout:        defaultPodVolumeOperationTimeout,
+		RestoreResourcePriorities:        defaultRestorePriorities,
+		ClientQPS:                        defaultClientQPS,
+		ClientBurst:                      defaultClientBurst,
+		ClientPageSize:                   defaultClientPageSize,
+		ProfilerAddress:                  defaultProfilerAddress,
+		ResourceTerminatingTimeout:       defaultResourceTerminatingTimeout,
+		LogLevel:                         logging.LogLevelFlag(logrus.InfoLevel),
+		LogFormat:                        logging.NewFormatFlag(),
+		DefaultVolumesToFsBackup:         podvolumeconfigs.DefaultVolumesToFsBackup,
+		UploaderType:                     uploader.KopiaType,
+		MaxConcurrentK8SConnections:      defaultMaxConcurrentK8SConnections,
+		DefaultSnapshotMoveData:          false,
+		DisableInformerCache:             defaultDisableInformerCache,
+		ScheduleSkipImmediately:          false,
+		CredentialsDirectory:             credentials.DefaultStoreDirectory(),
+		ItemBlockWorkerCount:             DefaultItemBlockWorkerCount,
+		ConcurrentBackups:                DefaultConcurrentBackups,
 	}
 
 	return config

pkg/cmd/server/config/config_test.go

@@ -2,6 +2,7 @@ package config
 
 import (
 	"testing"
+	"time"
 
 	"github.com/spf13/pflag"
 	"github.com/stretchr/testify/assert"
@@ -11,6 +12,8 @@ import (
 func TestGetDefaultConfig(t *testing.T) {
 	config := GetDefaultConfig()
 	assert.Equal(t, 1, config.ItemBlockWorkerCount)
+	assert.Equal(t, 10*time.Minute, config.DefaultBackupCSISnapshotTimeout)
+	assert.Equal(t, 30*time.Minute, config.DefaultRestoreCSISnapshotTimeout)
 }
 
 func TestBindFlags(t *testing.T) {

pkg/cmd/server/server.go

@@ -668,7 +668,7 @@ func (s *server) runControllers(defaultVolumeSnapshotLocations map[string]string
 			s.config.DefaultVolumesToFsBackup,
 			s.config.DefaultBackupTTL,
 			s.config.DefaultVGSLabelKey,
-			s.config.DefaultCSISnapshotTimeout,
+			s.config.DefaultBackupCSISnapshotTimeout,
 			s.config.ResourceTimeout,
 			s.config.DefaultItemOperationTimeout,
 			defaultVolumeSnapshotLocations,
@@ -878,6 +878,7 @@ func (s *server) runControllers(defaultVolumeSnapshotLocations map[string]string
 			backupStoreGetter,
 			s.metrics,
 			s.config.LogFormat.Parse(),
+			s.config.DefaultRestoreCSISnapshotTimeout,
 			s.config.DefaultItemOperationTimeout,
 			s.config.DisableInformerCache,
 			s.crClient,

pkg/controller/restore_controller.go

@@ -107,6 +107,7 @@ type restoreReconciler struct {
 	metrics                     *metrics.ServerMetrics
 	logFormat                   logging.Format
 	clock                       clock.WithTickerAndDelayedExecution
+	defaultCSISnapshotTimeout   time.Duration
 	defaultItemOperationTimeout time.Duration
 	disableInformerCache        bool
 
@@ -133,6 +134,7 @@ func NewRestoreReconciler(
 	backupStoreGetter persistence.ObjectBackupStoreGetter,
 	metrics *metrics.ServerMetrics,
 	logFormat logging.Format,
+	defaultCSISnapshotTimeout time.Duration,
 	defaultItemOperationTimeout time.Duration,
 	disableInformerCache bool,
 	globalCrClient client.Client,
@@ -149,6 +151,7 @@ func NewRestoreReconciler(
 		metrics:                     metrics,
 		logFormat:                   logFormat,
 		clock:                       &clock.RealClock{},
+		defaultCSISnapshotTimeout:   defaultCSISnapshotTimeout,
 		defaultItemOperationTimeout: defaultItemOperationTimeout,
 		disableInformerCache:        disableInformerCache,
 
@@ -250,6 +253,10 @@ func (r *restoreReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ct
 		restore.Status.StartTimestamp = &metav1.Time{Time: r.clock.Now()}
 		restore.Status.Phase = api.RestorePhaseInProgress
 	}
+	if restore.Spec.CSISnapshotTimeout.Duration == 0 {
+		// set default CSI snapshot timeout
+		restore.Spec.CSISnapshotTimeout.Duration = r.defaultCSISnapshotTimeout
+	}
 	if restore.Spec.ItemOperationTimeout.Duration == 0 {
 		// set default item operation timeout
 		restore.Spec.ItemOperationTimeout.Duration = r.defaultItemOperationTimeout

pkg/controller/restore_controller_test.go

@@ -112,6 +112,7 @@ func TestFetchBackupInfo(t *testing.T) {
 				NewFakeSingleObjectBackupStoreGetter(backupStore),
 				metrics.NewServerMetrics(),
 				formatFlag,
+				30*time.Minute,
 				60*time.Minute,
 				false,
 				fakeGlobalClient,
@@ -194,6 +195,7 @@ func TestProcessQueueItemSkips(t *testing.T) {
 				nil, // backupStoreGetter
 				metrics.NewServerMetrics(),
 				formatFlag,
+				30*time.Minute,
 				60*time.Minute,
 				false,
 				fakeGlobalClient,
@@ -211,6 +213,96 @@ func TestProcessQueueItemSkips(t *testing.T) {
 	}
 }
 
+func TestRestoreReconcile_CSISnapshotTimeoutDefaulting(t *testing.T) {
+	formatFlag := logging.FormatText
+	defaultCSITimeout := 45 * time.Minute
+
+	tests := []struct {
+		name               string
+		initialCSITimeout  time.Duration
+		expectedCSITimeout time.Duration
+	}{
+		{
+			name:               "CSISnapshotTimeout is 0, should default",
+			initialCSITimeout:  0,
+			expectedCSITimeout: defaultCSITimeout,
+		},
+		{
+			name:               "CSISnapshotTimeout is set, should be preserved",
+			initialCSITimeout:  15 * time.Minute,
+			expectedCSITimeout: 15 * time.Minute,
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			fakeClient := velerotest.NewFakeControllerRuntimeClient(t)
+			fakeGlobalClient := velerotest.NewFakeControllerRuntimeClient(t)
+			restorer := &fakeRestorer{kbClient: fakeClient}
+			backupStore := &persistencemocks.BackupStore{}
+			pluginManager := &pluginmocks.Manager{}
+
+			restore := builder.ForRestore("velero", "restore-1").
+				Phase(velerov1api.RestorePhaseNew).
+				Backup("backup-1").
+				CSISnapshotTimeout(tc.initialCSITimeout).
+				ItemOperationTimeout(60 * time.Minute).
+				Result()
+
+			require.NoError(t, fakeClient.Create(t.Context(), restore))
+
+			r := NewRestoreReconciler(
+				t.Context(),
+				velerov1api.DefaultNamespace,
+				restorer,
+				fakeClient,
+				velerotest.NewLogger(),
+				logrus.InfoLevel,
+				func(logrus.FieldLogger) clientmgmt.Manager { return pluginManager },
+				NewFakeSingleObjectBackupStoreGetter(backupStore),
+				metrics.NewServerMetrics(),
+				formatFlag,
+				defaultCSITimeout,
+				60*time.Minute,
+				false,
+				fakeGlobalClient,
+				10*time.Minute,
+				"",
+			)
+
+			location := builder.ForBackupStorageLocation("velero", "default").Provider("myCloud").Bucket("bucket").Phase(velerov1api.BackupStorageLocationPhaseAvailable).Result()
+			require.NoError(t, fakeClient.Create(t.Context(), location))
+
+			backup := defaultBackup().ObjectMeta(builder.WithName("backup-1")).StorageLocation("default").Phase(velerov1api.BackupPhaseCompleted).Result()
+			require.NoError(t, fakeClient.Create(t.Context(), backup))
+
+			backupStore.On("GetBackupContents", "backup-1").Return(io.NopCloser(bytes.NewReader([]byte("hello world"))), nil)
+			backupStore.On("GetCSIVolumeSnapshots", "backup-1").Return([]*snapshotv1api.VolumeSnapshot{}, nil)
+			backupStore.On("GetBackupVolumeInfos", "backup-1").Return([]*volume.BackupVolumeInfo{}, nil)
+			backupStore.On("GetBackupVolumeSnapshots", "backup-1").Return([]*volume.Snapshot{}, nil)
+			backupStore.On("PutRestoreLog", "backup-1", "restore-1", mock.Anything).Return(nil)
+			backupStore.On("PutRestoreResults", "backup-1", "restore-1", mock.Anything).Return(nil)
+			backupStore.On("PutRestoredResourceList", "restore-1", mock.Anything).Return(nil)
+			backupStore.On("PutRestoreItemOperations", mock.Anything, mock.Anything).Return(nil)
+			backupStore.On("PutRestoreVolumeInfo", "restore-1", mock.Anything).Return(nil)
+
+			pluginManager.On("GetRestoreItemActionsV2").Return(nil, nil)
+			pluginManager.On("CleanupClients").Return()
+
+			restorer.On("RestoreWithResolvers", mock.Anything, mock.Anything, mock.Anything, mock.Anything,
+				mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(results.Result{}, results.Result{})
+
+			_, err := r.Reconcile(t.Context(), ctrl.Request{NamespacedName: types.NamespacedName{
+				Namespace: "velero",
+				Name:      "restore-1",
+			}})
+			require.NoError(t, err)
+
+			assert.Equal(t, tc.expectedCSITimeout, restorer.calledWithArg.Spec.CSISnapshotTimeout.Duration)
+		})
+	}
+}
+
 func TestRestoreReconcile(t *testing.T) {
 	defaultStorageLocation := builder.ForBackupStorageLocation("velero", "default").Provider("myCloud").Bucket("bucket").Phase(velerov1api.BackupStorageLocationPhaseAvailable).Result()
 
@@ -610,6 +702,7 @@ func TestRestoreReconcile(t *testing.T) {
 				NewFakeSingleObjectBackupStoreGetter(backupStore),
 				metrics.NewServerMetrics(),
 				formatFlag,
+				30*time.Minute,
 				60*time.Minute,
 				false,
 				fakeGlobalClient,
@@ -799,6 +892,7 @@ func TestValidateAndCompleteWhenScheduleNameSpecified(t *testing.T) {
 		NewFakeSingleObjectBackupStoreGetter(backupStore),
 		metrics.NewServerMetrics(),
 		formatFlag,
+		30*time.Minute,
 		60*time.Minute,
 		false,
 		fakeGlobalClient,
@@ -896,6 +990,7 @@ func TestValidateAndCompleteWithResourcePolicySpecified(t *testing.T) {
 		NewFakeSingleObjectBackupStoreGetter(backupStore),
 		metrics.NewServerMetrics(),
 		formatFlag,
+		30*time.Minute,
 		60*time.Minute,
 		false,
 		fakeGlobalClient,
@@ -1026,6 +1121,7 @@ func TestValidateAndCompleteWithResourceModifierSpecified(t *testing.T) {
 		NewFakeSingleObjectBackupStoreGetter(backupStore),
 		metrics.NewServerMetrics(),
 		formatFlag,
+		30*time.Minute,
 		60*time.Minute,
 		false,
 		fakeGlobalClient,
@@ -1174,6 +1270,7 @@ func TestValidateAndCompleteWithDefaultResourceModifier(t *testing.T) {
 			NewFakeSingleObjectBackupStoreGetter(backupStore),
 			metrics.NewServerMetrics(),
 			formatFlag,
+			30*time.Minute,
 			60*time.Minute,
 			false,
 			fakeGlobalClient,
@@ -1417,7 +1514,7 @@ func TestMostRecentCompletedBackup(t *testing.T) {
 }
 
 func NewRestore(ns, name, backup, includeNS, includeResource string, phase velerov1api.RestorePhase) *builder.RestoreBuilder {
-	restore := builder.ForRestore(ns, name).Phase(phase).Backup(backup).ItemOperationTimeout(60 * time.Minute)
+	restore := builder.ForRestore(ns, name).Phase(phase).Backup(backup).ItemOperationTimeout(60 * time.Minute).CSISnapshotTimeout(30 * time.Minute)
 
 	if includeNS != "" {
 		restore = restore.IncludedNamespaces(includeNS)

pkg/exposer/generic_restore.go

@@ -512,7 +512,7 @@ func (e *genericRestoreExposer) CleanUp(ctx context.Context, ownerObject corev1a
 	kube.DeleteConfigMapsWithLabel(ctx, e.kubeClient.CoreV1(), ownerObject.Namespace,
 		BackupPVCSecretLabel, string(ownerObject.UID), e.log)
 
-	if param.Snapshot != nil {
+	if param.Snapshot != nil && param.Snapshot.CleanUp {
 		kube.EnsureDeleteVolumeSnapshotIfAny(ctx, e.ctrlClient, param.Snapshot.VolumeSnapshotNamespace,
 			param.Snapshot.VolumeSnapshot, 0, e.log)
 	}

pkg/exposer/generic_restore_test.go

@@ -32,6 +32,7 @@ import (
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/client-go/kubernetes/fake"
 	clientTesting "k8s.io/client-go/testing"
+	crclient "sigs.k8s.io/controller-runtime/pkg/client"
 
 	velerov1 "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
 	velerov2alpha1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v2alpha1"
@@ -2249,3 +2250,195 @@ func TestCreateRestorePod(t *testing.T) {
 		})
 	}
 }
+
+func TestGenericRestoreCleanUp(t *testing.T) {
+	ownerObject := corev1api.ObjectReference{
+		Kind:       "Restore",
+		Namespace:  "velero",
+		Name:       "restore-item",
+		UID:        "owner-uid",
+		APIVersion: "velero.io/v1",
+	}
+
+	tests := []struct {
+		name                 string
+		param                *GenericRestoreCleanUpParam
+		ctrlClientObjects    []crclient.Object
+		expectSnapshotExists bool
+	}{
+		{
+			name: "param has nil snapshot: pod, pvcs, pvs, secrets, cms cleaned up",
+			param: &GenericRestoreCleanUpParam{
+				Snapshot: nil,
+			},
+		},
+		{
+			name: "param snapshot with CleanUp false: snapshot is not deleted",
+			param: &GenericRestoreCleanUpParam{
+				Snapshot: &velerov2alpha1api.CSISnapshotSpec{
+					VolumeSnapshot:          "test-vs",
+					VolumeSnapshotNamespace: "velero",
+					CleanUp:                 false,
+				},
+			},
+			ctrlClientObjects: []crclient.Object{
+				&snapshotv1api.VolumeSnapshot{
+					ObjectMeta: metav1.ObjectMeta{
+						Name:      "test-vs",
+						Namespace: "velero",
+					},
+				},
+			},
+			expectSnapshotExists: true,
+		},
+		{
+			name: "param snapshot with CleanUp true: snapshot is deleted",
+			param: &GenericRestoreCleanUpParam{
+				Snapshot: &velerov2alpha1api.CSISnapshotSpec{
+					VolumeSnapshot:          "test-vs",
+					VolumeSnapshotNamespace: "velero",
+					CleanUp:                 true,
+				},
+			},
+			ctrlClientObjects: []crclient.Object{
+				&snapshotv1api.VolumeSnapshot{
+					ObjectMeta: metav1.ObjectMeta{
+						Name:      "test-vs",
+						Namespace: "velero",
+					},
+				},
+			},
+			expectSnapshotExists: false,
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			restorePod := &corev1api.Pod{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "restore-item",
+					Namespace: "velero",
+				},
+			}
+			restorePVC := &corev1api.PersistentVolumeClaim{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "restore-item",
+					Namespace: "velero",
+				},
+				Spec: corev1api.PersistentVolumeClaimSpec{
+					VolumeName: "pv-restore",
+				},
+			}
+			restorePV := &corev1api.PersistentVolume{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: "pv-restore",
+				},
+			}
+			cachePVC := &corev1api.PersistentVolumeClaim{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "restore-item-cache",
+					Namespace: "velero",
+				},
+				Spec: corev1api.PersistentVolumeClaimSpec{
+					VolumeName: "pv-cache",
+				},
+			}
+			cachePV := &corev1api.PersistentVolume{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: "pv-cache",
+				},
+			}
+			secret := &corev1api.Secret{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "owned-secret",
+					Namespace: "velero",
+					Labels:    map[string]string{BackupPVCSecretLabel: string(ownerObject.UID)},
+				},
+			}
+			cm := &corev1api.ConfigMap{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "owned-cm",
+					Namespace: "velero",
+					Labels:    map[string]string{BackupPVCSecretLabel: string(ownerObject.UID)},
+				},
+			}
+			unrelatedSecret := &corev1api.Secret{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "other-secret",
+					Namespace: "velero",
+					Labels:    map[string]string{BackupPVCSecretLabel: "other-uid"},
+				},
+			}
+			unrelatedCM := &corev1api.ConfigMap{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "other-cm",
+					Namespace: "velero",
+					Labels:    map[string]string{BackupPVCSecretLabel: "other-uid"},
+				},
+			}
+
+			fakeKubeClient := fake.NewSimpleClientset(
+				restorePod, restorePVC, restorePV, cachePVC, cachePV,
+				secret, cm, unrelatedSecret, unrelatedCM,
+			)
+
+			runtimeObjs := make([]runtime.Object, len(tc.ctrlClientObjects))
+			for i, obj := range tc.ctrlClientObjects {
+				runtimeObjs[i] = obj
+			}
+			fakeCtrlClient := velerotest.NewFakeControllerRuntimeClient(t, runtimeObjs...)
+
+			e := &genericRestoreExposer{
+				kubeClient: fakeKubeClient,
+				ctrlClient: fakeCtrlClient,
+				log:        velerotest.NewLogger(),
+			}
+
+			e.CleanUp(t.Context(), ownerObject, tc.param)
+
+			// Verify restore pod is deleted
+			_, err := fakeKubeClient.CoreV1().Pods("velero").Get(t.Context(), "restore-item", metav1.GetOptions{})
+			require.True(t, apierrors.IsNotFound(err), "restore pod should be deleted")
+
+			// Verify restore PVC is deleted and PV reclaim policy is set to Delete
+			_, err = fakeKubeClient.CoreV1().PersistentVolumeClaims("velero").Get(t.Context(), "restore-item", metav1.GetOptions{})
+			require.True(t, apierrors.IsNotFound(err), "restore PVC should be deleted")
+			retrievedPV, err := fakeKubeClient.CoreV1().PersistentVolumes().Get(t.Context(), "pv-restore", metav1.GetOptions{})
+			require.NoError(t, err)
+			assert.Equal(t, corev1api.PersistentVolumeReclaimDelete, retrievedPV.Spec.PersistentVolumeReclaimPolicy)
+
+			// Verify cache PVC is deleted and cache PV reclaim policy is set to Delete
+			_, err = fakeKubeClient.CoreV1().PersistentVolumeClaims("velero").Get(t.Context(), "restore-item-cache", metav1.GetOptions{})
+			require.True(t, apierrors.IsNotFound(err), "cache PVC should be deleted")
+			retrievedCachePV, err := fakeKubeClient.CoreV1().PersistentVolumes().Get(t.Context(), "pv-cache", metav1.GetOptions{})
+			require.NoError(t, err)
+			assert.Equal(t, corev1api.PersistentVolumeReclaimDelete, retrievedCachePV.Spec.PersistentVolumeReclaimPolicy)
+
+			// Verify owned secrets and configmaps are deleted
+			_, err = fakeKubeClient.CoreV1().Secrets("velero").Get(t.Context(), "owned-secret", metav1.GetOptions{})
+			require.True(t, apierrors.IsNotFound(err), "owned secret should be deleted")
+			_, err = fakeKubeClient.CoreV1().ConfigMaps("velero").Get(t.Context(), "owned-cm", metav1.GetOptions{})
+			require.True(t, apierrors.IsNotFound(err), "owned configmap should be deleted")
+
+			// Verify unrelated secrets and configmaps are preserved
+			_, err = fakeKubeClient.CoreV1().Secrets("velero").Get(t.Context(), "other-secret", metav1.GetOptions{})
+			require.NoError(t, err, "unrelated secret should not be deleted")
+			_, err = fakeKubeClient.CoreV1().ConfigMaps("velero").Get(t.Context(), "other-cm", metav1.GetOptions{})
+			require.NoError(t, err, "unrelated configmap should not be deleted")
+
+			// Verify VolumeSnapshot state if applicable
+			if tc.param.Snapshot != nil {
+				vs := &snapshotv1api.VolumeSnapshot{}
+				err = fakeCtrlClient.Get(t.Context(), crclient.ObjectKey{
+					Namespace: tc.param.Snapshot.VolumeSnapshotNamespace,
+					Name:      tc.param.Snapshot.VolumeSnapshot,
+				}, vs)
+				if tc.expectSnapshotExists {
+					require.NoError(t, err, "VolumeSnapshot should still exist")
+				} else {
+					require.True(t, apierrors.IsNotFound(err), "VolumeSnapshot should be deleted")
+				}
+			}
+		})
+	}
+}

pkg/restore/actions/csi/pvc_action.go

@@ -232,7 +232,18 @@ func (p *pvcRestoreItemAction) executeWithDataMove(logger *logrus.Entry, input *
 		return nil, errors.Wrapf(err, "failed to get DataUploadResult for restore: %s", input.Restore.Name)
 	}
 
+	// If cross-namespace restore is configured, change the namespace
+	// for PVC object to be restored
+	newNamespace, namespaceMapped := input.Restore.Spec.NamespaceMapping[pvc.GetNamespace()]
+	// make sure the namespace mapping is not the same as the original namespace
+	namespaceMapped = namespaceMapped && newNamespace != pvc.Namespace
+	if !namespaceMapped {
+		// Use original namespace
+		newNamespace = pvc.Namespace
+	}
+
 	var volumeSnapshot *snapshotv1api.VolumeSnapshot
+	cleanUpVolumeSnapshot := false
 	restoreType := input.Restore.Spec.ExistingVolumeDataPolicy
 	if pvcExists {
 		// Pre-flight checks must pass before any side effect on the existing PVC/PV.
@@ -248,16 +259,34 @@ func (p *pvcRestoreItemAction) executeWithDataMove(logger *logrus.Entry, input *
 
 		// take a CSI snapshot of the existing PVC as the baseline of CBT
 		if input.Restore.IsVolumeDataInplaceIncrementalRestore() && datamover.IsVeleroBlockDataMover(dataUploadResult.DataMover) {
-			logger.Info("ExistingVolumeDataPolicy is in-place incremental restore and data mover is velero-block. Taking a CSI snapshot of the existing PVC as the baseline of CBT...")
-			volumeSnapshot, err = p.createVolumeSnapshot(ctx, logger, input.Restore, *existingPVC, dataUploadResult.SnapshotClass, backup.Spec.CSISnapshotTimeout.Duration)
-			if err != nil {
-				logger.Warnf("Fail to create VolumeSnapshot for existing PVC %s/%s: %s, incremental restore will be suppressed", existingPVC.Namespace, existingPVC.Name, err.Error())
+			// take a CSI snapshot of the existing PVC as the baseline of CBT
+			if !namespaceMapped {
+				logger.Info("requesting an in-place incremental restore with block data mover, taking a CSI snapshot of the existing PVC as the baseline of CBT...")
+				volumeSnapshot, err = p.createVolumeSnapshot(ctx, logger, input.Restore, *existingPVC, dataUploadResult.SnapshotClass, input.Restore.Spec.CSISnapshotTimeout.Duration)
+				if err != nil {
+					logger.Warnf("fail to create VolumeSnapshot for existing PVC %s/%s: %s, fallback to in-place full restore", existingPVC.Namespace, existingPVC.Name, err.Error())
+					restoreType = velerov1api.VolumeDataPolicyTypeFull
+				} else {
+					cleanUpVolumeSnapshot = true
+					defer func() {
+						if err != nil {
+							csi.CleanupVolumeSnapshot(ctx, volumeSnapshot, p.crClient, p.log)
+						}
+					}()
+				}
 			} else {
-				defer func() {
-					if err != nil {
-						csi.CleanupVolumeSnapshot(ctx, volumeSnapshot, p.crClient, logger)
-					}
-				}()
+				var ok bool
+				volumeSnapshot, ok, err = p.isCreatedFromSnapshot(ctx, existingPVC)
+				if err != nil {
+					return nil, errors.WithStack(err)
+				}
+				if !ok {
+					return nil, fmt.Errorf("the existing PVC %s/%s should be created from a VolumeSnapshot when triggering the in-place incremental restore with block data mover and namespace-mapping set, fail the restore", existingPVC.Namespace, existingPVC.Name)
+				}
+				logger.Infof("existing PVC %s/%s is created from VolumeSnapshot %s/%s", existingPVC.Namespace, existingPVC.Name, volumeSnapshot.Namespace, volumeSnapshot.Name)
+				if !kube.IsPVCBound(existingPVC) {
+					return nil, fmt.Errorf("the existing PVC %s/%s should be bound before triggering the in-place incremental restore with block data mover and namespace-mapping set, fail the restore", existingPVC.Namespace, existingPVC.Name)
+				}
 			}
 		}
 
@@ -272,18 +301,10 @@ func (p *pvcRestoreItemAction) executeWithDataMove(logger *logrus.Entry, input *
 		string(velerov1api.AsyncOperationIDPrefixDataDownload) +
 			string(input.Restore.UID) + "." + string(pvcFromBackup.UID))
 
-	// If cross-namespace restore is configured, change the namespace
-	// for PVC object to be restored
-	newNamespace, ok := input.Restore.Spec.NamespaceMapping[pvc.GetNamespace()]
-	if !ok {
-		// Use original namespace
-		newNamespace = pvc.Namespace
-	}
-
 	var dataDownload *velerov2alpha1.DataDownload
 	dataDownload, err = restoreFromDataUploadResult(
 		ctx, dataUploadResult, input.Restore, backup, pvc, existingPV, newNamespace,
-		operationID, string(restoreType), volumeSnapshot, p.crClient)
+		operationID, string(restoreType), volumeSnapshot, cleanUpVolumeSnapshot, p.crClient)
 	if err != nil {
 		logger.Errorf("Failed to restore from DataUploadResult: %s", err.Error())
 		return nil, errors.WithStack(err)
@@ -512,6 +533,7 @@ func newDataDownload(
 	pv *corev1api.PersistentVolume,
 	newNamespace, operationID, restoreType string,
 	volumeSnapshot *snapshotv1api.VolumeSnapshot,
+	cleanUpVolumeSnapshot bool,
 ) *velerov2alpha1.DataDownload {
 	pvName := ""
 	if pv != nil {
@@ -561,6 +583,7 @@ func newDataDownload(
 		dataDownload.Spec.CSISnapshot = &velerov2alpha1.CSISnapshotSpec{
 			VolumeSnapshot:          volumeSnapshot.Name,
 			VolumeSnapshotNamespace: volumeSnapshot.Namespace,
+			CleanUp:                 cleanUpVolumeSnapshot,
 		}
 	}
 	if restore.Spec.UploaderConfig != nil {
@@ -578,6 +601,7 @@ func restoreFromDataUploadResult(
 	pv *corev1api.PersistentVolume,
 	newNamespace, operationID, restoreType string,
 	volumeSnapshot *snapshotv1api.VolumeSnapshot,
+	cleanUpVolumeSnapshot bool,
 	crClient crclient.Client,
 ) (*velerov2alpha1.DataDownload, error) {
 	pvc.Spec.VolumeName = ""
@@ -601,6 +625,7 @@ func restoreFromDataUploadResult(
 		operationID,
 		restoreType,
 		volumeSnapshot,
+		cleanUpVolumeSnapshot,
 	)
 	err := crClient.Create(ctx, dataDownload)
 	if err != nil {
@@ -679,7 +704,7 @@ func (p *pvcRestoreItemAction) deleteExistingPVC(ctx context.Context, logger *lo
 	return pv, nil
 }
 
-func (p *pvcRestoreItemAction) createVolumeSnapshot(ctx context.Context, logger *logrus.Entry, restore *velerov1api.Restore, pvc corev1api.PersistentVolumeClaim, vsClass string, operationTimeout time.Duration) (vs *snapshotv1api.VolumeSnapshot, err error) {
+func (p *pvcRestoreItemAction) createVolumeSnapshot(ctx context.Context, logger *logrus.Entry, restore *velerov1api.Restore, pvc corev1api.PersistentVolumeClaim, vsClass string, timeout time.Duration) (vs *snapshotv1api.VolumeSnapshot, err error) {
 	logger.Infof("creating VolumeSnapshot for PVC %s/%s with VolumeSnapshotClass %s", pvc.Namespace, pvc.Name, vsClass)
 
 	labels := map[string]string{
@@ -711,19 +736,19 @@ func (p *pvcRestoreItemAction) createVolumeSnapshot(ctx context.Context, logger
 	vsName := vs.Name
 	vsNamespace := vs.Namespace
 
-	_, err = csi.WaitUntilVSCHandleIsReady(vs, p.crClient, logger, operationTimeout)
+	_, err = csi.WaitUntilVSCHandleIsReady(vs, p.crClient, logger, timeout)
 	if err != nil {
 		csi.CleanupVolumeSnapshot(ctx, vs, p.crClient, logger)
 		return nil, errors.Wrapf(err, "failed to wait for VolumeSnapshotContent of VolumeSnapshot %s/%s to be ready within timeout %v",
-			vsNamespace, vsName, operationTimeout)
+			vsNamespace, vsName, timeout)
 	}
 
 	var updatedVS *snapshotv1api.VolumeSnapshot
-	updatedVS, err = csi.WaitVolumeSnapshotReady(ctx, p.csiSnapshotClient, vs.Name, vs.Namespace, operationTimeout, logger)
+	updatedVS, err = csi.WaitVolumeSnapshotReady(ctx, p.csiSnapshotClient, vs.Name, vs.Namespace, timeout, logger)
 	if err != nil {
 		csi.CleanupVolumeSnapshot(ctx, vs, p.crClient, logger)
 		return nil, errors.Wrapf(err, "failed to wait for VolumeSnapshot %s/%s to become Ready within timeout %v",
-			vsNamespace, vsName, operationTimeout)
+			vsNamespace, vsName, timeout)
 	}
 	vs = updatedVS
 
@@ -739,6 +764,39 @@ func sourceSizeFromCarrier(pvc *corev1api.PersistentVolumeClaim) int64 {
 	return size
 }
 
+func (p *pvcRestoreItemAction) isCreatedFromSnapshot(
+	ctx context.Context,
+	pvc *corev1api.PersistentVolumeClaim,
+) (*snapshotv1api.VolumeSnapshot, bool, error) {
+	var vsName string
+	var vsNamespace string
+
+	if pvc.Spec.DataSource != nil && pvc.Spec.DataSource.Kind == "VolumeSnapshot" &&
+		pvc.Spec.DataSource.APIGroup != nil && *pvc.Spec.DataSource.APIGroup == snapshotv1api.SchemeGroupVersion.Group {
+		vsName = pvc.Spec.DataSource.Name
+		vsNamespace = pvc.Namespace
+	} else if pvc.Spec.DataSourceRef != nil && pvc.Spec.DataSourceRef.Kind == "VolumeSnapshot" &&
+		pvc.Spec.DataSourceRef.APIGroup != nil && *pvc.Spec.DataSourceRef.APIGroup == snapshotv1api.SchemeGroupVersion.Group {
+		vsName = pvc.Spec.DataSourceRef.Name
+		if pvc.Spec.DataSourceRef.Namespace != nil {
+			vsNamespace = *pvc.Spec.DataSourceRef.Namespace
+		} else {
+			vsNamespace = pvc.Namespace
+		}
+	}
+
+	if vsName == "" {
+		return nil, false, nil
+	}
+
+	vs := new(snapshotv1api.VolumeSnapshot)
+	if err := p.crClient.Get(ctx, crclient.ObjectKey{Namespace: vsNamespace, Name: vsName}, vs); err != nil {
+		return nil, false, errors.Wrapf(err, "fail to get VolumeSnapshot %s/%s", vsNamespace, vsName)
+	}
+
+	return vs, true, nil
+}
+
 func NewPvcRestoreItemAction(f client.Factory) plugincommon.HandlerInitializer {
 	return func(logger logrus.FieldLogger) (any, error) {
 		crClient, err := f.KubebuilderClient()

pkg/restore/actions/csi/pvc_action_test.go

@@ -24,6 +24,7 @@ import (
 	"github.com/google/go-cmp/cmp"
 	"github.com/google/go-cmp/cmp/cmpopts"
 	snapshotv1api "github.com/kubernetes-csi/external-snapshotter/client/v8/apis/volumesnapshot/v1"
+	snapshotFake "github.com/kubernetes-csi/external-snapshotter/client/v8/clientset/versioned/fake"
 	"github.com/sirupsen/logrus"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -37,6 +38,7 @@ import (
 	"k8s.io/apimachinery/pkg/util/validation"
 	"k8s.io/client-go/kubernetes/fake"
 	"k8s.io/client-go/rest"
+	"k8s.io/utils/ptr"
 	crclient "sigs.k8s.io/controller-runtime/pkg/client"
 
 	"github.com/vmware-tanzu/velero/pkg/apis/velero/shared"
@@ -383,6 +385,8 @@ func TestExecute(t *testing.T) {
 		expectedPVC          *corev1api.PersistentVolumeClaim
 		preCreatePVC         bool
 		kubeClientObj        []runtime.Object
+		crObjects            []runtime.Object
+		snapshotClientObj    []runtime.Object
 	}{
 		{
 			name:        "Don't restore PV",
@@ -528,11 +532,94 @@ func TestExecute(t *testing.T) {
 					ObjectMeta(builder.WithOwnerReference([]metav1.OwnerReference{{APIVersion: velerov1api.SchemeGroupVersion.String(), Kind: "Restore", Name: "testRestore", UID: "uid", Controller: boolptr.True()}}),
 						builder.WithLabelsMap(map[string]string{velerov1api.AsyncOperationIDLabel: "dd-uid.", velerov1api.RestoreNameLabel: "testRestore", velerov1api.RestoreUIDLabel: "uid"}),
 						builder.WithGenerateName("testRestore-")).Result()
-				d.Spec.RestoreType = "incremental"
+				d.Spec.RestoreType = "full"
 				d.Spec.DataMover = "velero-block"
 				return d
 			}(),
 		},
+		{
+			name:             "PVC exists and in-place incremental restore set with namespace mapping, existing PVC created from VolumeSnapshot",
+			backup:           builder.ForBackup("velero", "testBackup").SnapshotMoveData(true).Result(),
+			restore:          builder.ForRestore("velero", "testRestore").Backup("testBackup").NamespaceMappings("velero", "restore").ExistingVolumeDataPolicy(string(velerov1api.VolumeDataPolicyTypeIncremental)).ItemOperationTimeout(time.Minute * 10).ObjectMeta(builder.WithUID("uid")).Result(),
+			pvc:              builder.ForPersistentVolumeClaim("velero", "testPVC").ObjectMeta(builder.WithAnnotations(velerov1api.VolumeSnapshotLabel, "vsName", velerov1api.VolumeSnapshotRestoreSize, "10Gi", velerov1api.DataUploadNameAnnotation, "velero/")).Result(),
+			pv:               builder.ForPersistentVolume("testPV").ReclaimPolicy(corev1api.PersistentVolumeReclaimRetain).Result(),
+			dataUploadResult: builder.ForConfigMap("velero", "testCM").Data("uid", "{\"DataMover\":\"velero-block\", \"SnapshotClass\":\"test-snapclass\"}").ObjectMeta(builder.WithLabels(velerov1api.RestoreUIDLabel, "uid", velerov1api.PVCNamespaceNameLabel, "velero.testPVC", velerov1api.ResourceUsageLabel, label.GetValidName(string(velerov1api.VeleroResourceUsageDataUploadResult)))).Result(),
+			kubeClientObj: []runtime.Object{
+				builder.ForPersistentVolumeClaim("restore", "testPVC").VolumeName("testPV").Phase(corev1api.ClaimBound).ObjectMeta(builder.WithAnnotations(velerov1api.VolumeSnapshotLabel, "vsName", velerov1api.VolumeSnapshotRestoreSize, "10Gi", velerov1api.DataUploadNameAnnotation, "velero/")).DataSource(&corev1api.TypedLocalObjectReference{APIGroup: ptr.To(snapshotv1api.SchemeGroupVersion.Group), Kind: "VolumeSnapshot", Name: "source-snap"}).Result(),
+				builder.ForPersistentVolume("testPV").ReclaimPolicy(corev1api.PersistentVolumeReclaimRetain).Result(),
+			},
+			crObjects: []runtime.Object{
+				builder.ForPersistentVolumeClaim("restore", "testPVC").VolumeName("testPV").Phase(corev1api.ClaimBound).ObjectMeta(builder.WithAnnotations(velerov1api.VolumeSnapshotLabel, "vsName", velerov1api.VolumeSnapshotRestoreSize, "10Gi", velerov1api.DataUploadNameAnnotation, "velero/")).DataSource(&corev1api.TypedLocalObjectReference{APIGroup: ptr.To(snapshotv1api.SchemeGroupVersion.Group), Kind: "VolumeSnapshot", Name: "source-snap"}).Result(),
+				builder.ForVolumeSnapshot("restore", "source-snap").Status().BoundVolumeSnapshotContentName("source-vsc").ReadyToUse(true).Result(),
+				builder.ForVolumeSnapshotContent("source-vsc").Status(&snapshotv1api.VolumeSnapshotContentStatus{SnapshotHandle: ptr.To("handle-1")}).Result(),
+			},
+			snapshotClientObj: []runtime.Object{
+				builder.ForVolumeSnapshot("restore", "source-snap").Status().ReadyToUse(true).Result(),
+			},
+			expectedDataDownload: func() *velerov2alpha1.DataDownload {
+				d := builder.ForDataDownload("velero", "name").TargetVolume(velerov2alpha1.TargetVolumeSpec{PVC: "testPVC", Namespace: "restore", PV: "testPV"}).
+					ObjectMeta(builder.WithOwnerReference([]metav1.OwnerReference{{APIVersion: velerov1api.SchemeGroupVersion.String(), Kind: "Restore", Name: "testRestore", UID: "uid", Controller: boolptr.True()}}),
+						builder.WithLabelsMap(map[string]string{velerov1api.AsyncOperationIDLabel: "dd-uid.", velerov1api.RestoreNameLabel: "testRestore", velerov1api.RestoreUIDLabel: "uid"}),
+						builder.WithGenerateName("testRestore-")).Result()
+				d.Spec.RestoreType = "incremental"
+				d.Spec.DataMover = "velero-block"
+				d.Spec.CSISnapshot = &velerov2alpha1.CSISnapshotSpec{
+					VolumeSnapshot:          "source-snap",
+					VolumeSnapshotNamespace: "restore",
+					CleanUp:                 false,
+				}
+				return d
+			}(),
+		},
+		{
+			name:             "PVC exists and in-place incremental restore set with namespace mapping, existing PVC not created from VolumeSnapshot",
+			backup:           builder.ForBackup("velero", "testBackup").SnapshotMoveData(true).Result(),
+			restore:          builder.ForRestore("velero", "testRestore").Backup("testBackup").NamespaceMappings("velero", "restore").ExistingVolumeDataPolicy(string(velerov1api.VolumeDataPolicyTypeIncremental)).ItemOperationTimeout(time.Minute * 10).ObjectMeta(builder.WithUID("uid")).Result(),
+			pvc:              builder.ForPersistentVolumeClaim("velero", "testPVC").ObjectMeta(builder.WithAnnotations(velerov1api.VolumeSnapshotLabel, "vsName", velerov1api.VolumeSnapshotRestoreSize, "10Gi", velerov1api.DataUploadNameAnnotation, "velero/")).Result(),
+			pv:               builder.ForPersistentVolume("testPV").ReclaimPolicy(corev1api.PersistentVolumeReclaimRetain).Result(),
+			dataUploadResult: builder.ForConfigMap("velero", "testCM").Data("uid", "{\"DataMover\":\"velero-block\", \"SnapshotClass\":\"test-snapclass\"}").ObjectMeta(builder.WithLabels(velerov1api.RestoreUIDLabel, "uid", velerov1api.PVCNamespaceNameLabel, "velero.testPVC", velerov1api.ResourceUsageLabel, label.GetValidName(string(velerov1api.VeleroResourceUsageDataUploadResult)))).Result(),
+			kubeClientObj: []runtime.Object{
+				builder.ForPersistentVolumeClaim("restore", "testPVC").VolumeName("testPV").Phase(corev1api.ClaimBound).ObjectMeta(builder.WithAnnotations(velerov1api.VolumeSnapshotLabel, "vsName", velerov1api.VolumeSnapshotRestoreSize, "10Gi", velerov1api.DataUploadNameAnnotation, "velero/")).Result(),
+				builder.ForPersistentVolume("testPV").ReclaimPolicy(corev1api.PersistentVolumeReclaimRetain).Result(),
+			},
+			crObjects: []runtime.Object{
+				builder.ForPersistentVolumeClaim("restore", "testPVC").VolumeName("testPV").Phase(corev1api.ClaimBound).ObjectMeta(builder.WithAnnotations(velerov1api.VolumeSnapshotLabel, "vsName", velerov1api.VolumeSnapshotRestoreSize, "10Gi", velerov1api.DataUploadNameAnnotation, "velero/")).Result(),
+			},
+			expectedErr: "the existing PVC restore/testPVC should be created from a VolumeSnapshot when triggering the in-place incremental restore with block data mover and namespace-mapping set, fail the restore",
+		},
+		{
+			name:             "PVC exists and in-place incremental restore set with namespace mapping, existing PVC is not bound",
+			backup:           builder.ForBackup("velero", "testBackup").SnapshotMoveData(true).Result(),
+			restore:          builder.ForRestore("velero", "testRestore").Backup("testBackup").NamespaceMappings("velero", "restore").ExistingVolumeDataPolicy(string(velerov1api.VolumeDataPolicyTypeIncremental)).ItemOperationTimeout(time.Minute * 10).ObjectMeta(builder.WithUID("uid")).Result(),
+			pvc:              builder.ForPersistentVolumeClaim("velero", "testPVC").ObjectMeta(builder.WithAnnotations(velerov1api.VolumeSnapshotLabel, "vsName", velerov1api.VolumeSnapshotRestoreSize, "10Gi", velerov1api.DataUploadNameAnnotation, "velero/")).Result(),
+			pv:               builder.ForPersistentVolume("testPV").ReclaimPolicy(corev1api.PersistentVolumeReclaimRetain).Result(),
+			dataUploadResult: builder.ForConfigMap("velero", "testCM").Data("uid", "{\"DataMover\":\"velero-block\", \"SnapshotClass\":\"test-snapclass\"}").ObjectMeta(builder.WithLabels(velerov1api.RestoreUIDLabel, "uid", velerov1api.PVCNamespaceNameLabel, "velero.testPVC", velerov1api.ResourceUsageLabel, label.GetValidName(string(velerov1api.VeleroResourceUsageDataUploadResult)))).Result(),
+			kubeClientObj: []runtime.Object{
+				builder.ForPersistentVolumeClaim("restore", "testPVC").Phase(corev1api.ClaimBound).ObjectMeta(builder.WithAnnotations(velerov1api.VolumeSnapshotLabel, "vsName", velerov1api.VolumeSnapshotRestoreSize, "10Gi", velerov1api.DataUploadNameAnnotation, "velero/")).DataSource(&corev1api.TypedLocalObjectReference{APIGroup: ptr.To(snapshotv1api.SchemeGroupVersion.Group), Kind: "VolumeSnapshot", Name: "source-snap"}).Result(),
+				builder.ForPersistentVolume("testPV").ReclaimPolicy(corev1api.PersistentVolumeReclaimRetain).Result(),
+			},
+			crObjects: []runtime.Object{
+				builder.ForPersistentVolumeClaim("restore", "testPVC").Phase(corev1api.ClaimBound).ObjectMeta(builder.WithAnnotations(velerov1api.VolumeSnapshotLabel, "vsName", velerov1api.VolumeSnapshotRestoreSize, "10Gi", velerov1api.DataUploadNameAnnotation, "velero/")).DataSource(&corev1api.TypedLocalObjectReference{APIGroup: ptr.To(snapshotv1api.SchemeGroupVersion.Group), Kind: "VolumeSnapshot", Name: "source-snap"}).Result(),
+				builder.ForVolumeSnapshot("restore", "source-snap").Result(),
+			},
+			expectedErr: "the existing PVC restore/testPVC should be bound before triggering the in-place incremental restore with block data mover and namespace-mapping set, fail the restore",
+		},
+		{
+			name:             "PVC exists and in-place incremental restore set with namespace mapping, VolumeSnapshot get fails",
+			backup:           builder.ForBackup("velero", "testBackup").SnapshotMoveData(true).Result(),
+			restore:          builder.ForRestore("velero", "testRestore").Backup("testBackup").NamespaceMappings("velero", "restore").ExistingVolumeDataPolicy(string(velerov1api.VolumeDataPolicyTypeIncremental)).ItemOperationTimeout(time.Minute * 10).ObjectMeta(builder.WithUID("uid")).Result(),
+			pvc:              builder.ForPersistentVolumeClaim("velero", "testPVC").ObjectMeta(builder.WithAnnotations(velerov1api.VolumeSnapshotLabel, "vsName", velerov1api.VolumeSnapshotRestoreSize, "10Gi", velerov1api.DataUploadNameAnnotation, "velero/")).Result(),
+			pv:               builder.ForPersistentVolume("testPV").ReclaimPolicy(corev1api.PersistentVolumeReclaimRetain).Result(),
+			dataUploadResult: builder.ForConfigMap("velero", "testCM").Data("uid", "{\"DataMover\":\"velero-block\", \"SnapshotClass\":\"test-snapclass\"}").ObjectMeta(builder.WithLabels(velerov1api.RestoreUIDLabel, "uid", velerov1api.PVCNamespaceNameLabel, "velero.testPVC", velerov1api.ResourceUsageLabel, label.GetValidName(string(velerov1api.VeleroResourceUsageDataUploadResult)))).Result(),
+			kubeClientObj: []runtime.Object{
+				builder.ForPersistentVolumeClaim("restore", "testPVC").VolumeName("testPV").Phase(corev1api.ClaimBound).ObjectMeta(builder.WithAnnotations(velerov1api.VolumeSnapshotLabel, "vsName", velerov1api.VolumeSnapshotRestoreSize, "10Gi", velerov1api.DataUploadNameAnnotation, "velero/")).DataSource(&corev1api.TypedLocalObjectReference{APIGroup: ptr.To(snapshotv1api.SchemeGroupVersion.Group), Kind: "VolumeSnapshot", Name: "missing-vs"}).Result(),
+				builder.ForPersistentVolume("testPV").ReclaimPolicy(corev1api.PersistentVolumeReclaimRetain).Result(),
+			},
+			crObjects: []runtime.Object{
+				builder.ForPersistentVolumeClaim("restore", "testPVC").VolumeName("testPV").Phase(corev1api.ClaimBound).ObjectMeta(builder.WithAnnotations(velerov1api.VolumeSnapshotLabel, "vsName", velerov1api.VolumeSnapshotRestoreSize, "10Gi", velerov1api.DataUploadNameAnnotation, "velero/")).DataSource(&corev1api.TypedLocalObjectReference{APIGroup: ptr.To(snapshotv1api.SchemeGroupVersion.Group), Kind: "VolumeSnapshot", Name: "missing-vs"}).Result(),
+			},
+			expectedErr: "fail to get VolumeSnapshot restore/missing-vs: volumesnapshots.snapshot.storage.k8s.io \"missing-vs\" not found",
+		},
 	}
 
 	for _, tc := range tests {
@@ -573,11 +660,13 @@ func TestExecute(t *testing.T) {
 			if tc.dataUploadResult != nil {
 				object = append(object, tc.dataUploadResult)
 			}
+			object = append(object, tc.crObjects...)
 
 			pvcRIA := pvcRestoreItemAction{
-				log:        logrus.New(),
-				crClient:   velerotest.NewFakeControllerRuntimeClient(t, object...),
-				kubeClient: fake.NewSimpleClientset(tc.kubeClientObj...),
+				log:               logrus.New(),
+				crClient:          velerotest.NewFakeControllerRuntimeClient(t, object...),
+				kubeClient:        fake.NewSimpleClientset(tc.kubeClientObj...),
+				csiSnapshotClient: snapshotFake.NewSimpleClientset(tc.snapshotClientObj...).SnapshotV1(),
 			}
 
 			output, err := pvcRIA.Execute(input)
@@ -868,6 +957,137 @@ func TestExecuteInplaceRestorePreflight(t *testing.T) {
 		})
 	}
 }
+func TestExecuteInplaceIncrementalRestoreWithNamespaceMapping(t *testing.T) {
+	existingPVC := builder.ForPersistentVolumeClaim("restore-ns", "testPVC").
+		ObjectMeta(builder.WithAnnotations(AnnSelectedNode, "node-1")).
+		VolumeName("testPV").
+		Phase(corev1api.ClaimBound).
+		DataSource(&corev1api.TypedLocalObjectReference{
+			APIGroup: ptr.To(snapshotv1api.SchemeGroupVersion.Group),
+			Kind:     "VolumeSnapshot",
+			Name:     "existing-snap",
+		}).Result()
+	existingPV := builder.ForPersistentVolume("testPV").Result()
+	existingVS := builder.ForVolumeSnapshot("restore-ns", "existing-snap").
+		Status().BoundVolumeSnapshotContentName("existing-vsc").ReadyToUse(true).Result()
+	existingVSC := builder.ForVolumeSnapshotContent("existing-vsc").
+		Status(&snapshotv1api.VolumeSnapshotContentStatus{SnapshotHandle: ptr.To("snap-handle-1")}).Result()
+
+	backup := builder.ForBackup("velero", "testBackup").SnapshotMoveData(true).Result()
+	restore := builder.ForRestore("velero", "testRestore").Backup("testBackup").
+		NamespaceMappings("velero", "restore-ns").
+		ExistingVolumeDataPolicy("incremental").
+		ObjectMeta(builder.WithUID("uid")).Result()
+	pvcFromBackup := builder.ForPersistentVolumeClaim("velero", "testPVC").
+		ObjectMeta(builder.WithAnnotations(
+			velerov1api.VolumeSnapshotLabel, "vsName",
+			velerov1api.DataUploadNameAnnotation, "velero/testDU",
+		)).Result()
+	dataUploadResult := builder.ForConfigMap("velero", "testCM").
+		Data("uid", "{\"DataMover\":\"velero-block\", \"SnapshotClass\":\"test-snapclass\"}").
+		ObjectMeta(builder.WithLabels(
+			velerov1api.RestoreUIDLabel, "uid",
+			velerov1api.PVCNamespaceNameLabel, "velero.testPVC",
+			velerov1api.ResourceUsageLabel, label.GetValidName(string(velerov1api.VeleroResourceUsageDataUploadResult)),
+		)).Result()
+
+	pvcRIA := pvcRestoreItemAction{
+		log:               logrus.New(),
+		crClient:          velerotest.NewFakeControllerRuntimeClient(t, existingPVC, existingPV, existingVS, existingVSC, backup, dataUploadResult),
+		kubeClient:        fake.NewSimpleClientset(existingPVC, existingPV),
+		csiSnapshotClient: snapshotFake.NewSimpleClientset(existingVS).SnapshotV1(),
+	}
+
+	pvcMap, err := runtime.DefaultUnstructuredConverter.ToUnstructured(pvcFromBackup.DeepCopy())
+	require.NoError(t, err)
+	pvcFromBackupMap, err := runtime.DefaultUnstructuredConverter.ToUnstructured(pvcFromBackup)
+	require.NoError(t, err)
+
+	output, err := pvcRIA.Execute(&velero.RestoreItemActionExecuteInput{
+		Item:           &unstructured.Unstructured{Object: pvcMap},
+		ItemFromBackup: &unstructured.Unstructured{Object: pvcFromBackupMap},
+		Restore:        restore,
+	})
+	require.NoError(t, err)
+
+	updatedPVC := new(corev1api.PersistentVolumeClaim)
+	require.NoError(t, runtime.DefaultUnstructuredConverter.FromUnstructured(
+		output.UpdatedItem.UnstructuredContent(), updatedPVC))
+
+	// Carrier annotation carries the captured value; the Kubernetes annotation is not set by this RIA.
+	require.Equal(t, "node-1", updatedPVC.Annotations[velerov1api.InplaceRestoreSelectedNodeAnnotation])
+	require.NotContains(t, updatedPVC.Annotations, AnnSelectedNode)
+
+	// The existing PVC in the mapped namespace is deleted so the exposer can bind a temporary PVC to the PV.
+	_, err = pvcRIA.kubeClient.CoreV1().PersistentVolumeClaims("restore-ns").Get(t.Context(), "testPVC", metav1.GetOptions{})
+	require.True(t, apierrors.IsNotFound(err))
+
+	// A DataDownload with the incremental restoreType, velero-block data mover, and cleanUp=false is created.
+	dataDownloadList := new(velerov2alpha1.DataDownloadList)
+	require.NoError(t, pvcRIA.crClient.List(t.Context(), dataDownloadList, &crclient.ListOptions{}))
+	require.Len(t, dataDownloadList.Items, 1)
+	require.Equal(t, "incremental", dataDownloadList.Items[0].Spec.RestoreType)
+	require.Equal(t, "velero-block", dataDownloadList.Items[0].Spec.DataMover)
+	require.Equal(t, "testPV", dataDownloadList.Items[0].Spec.TargetVolume.PV)
+	require.Equal(t, "restore-ns", dataDownloadList.Items[0].Spec.TargetVolume.Namespace)
+	require.NotNil(t, dataDownloadList.Items[0].Spec.CSISnapshot)
+	require.Equal(t, "existing-snap", dataDownloadList.Items[0].Spec.CSISnapshot.VolumeSnapshot)
+	require.Equal(t, "restore-ns", dataDownloadList.Items[0].Spec.CSISnapshot.VolumeSnapshotNamespace)
+	require.False(t, dataDownloadList.Items[0].Spec.CSISnapshot.CleanUp)
+}
+
+func TestNewDataDownload(t *testing.T) {
+	restore := builder.ForRestore("velero", "testRestore").ObjectMeta(builder.WithUID("uid")).Result()
+	backup := builder.ForBackup("velero", "testBackup").CSISnapshotTimeout(10 * time.Minute).Result()
+	dataUploadResult := &velerov2alpha1.DataUploadResult{
+		BackupStorageLocation: "bsl",
+		DataMover:             "velero-block",
+		SnapshotID:            "snap-id",
+		SnapshotSize:          1024,
+		SourceNamespace:       "source-ns",
+		NodeOS:                "linux",
+		FSType:                "ext4",
+	}
+	pvc := builder.ForPersistentVolumeClaim("velero", "testPVC").Result()
+	pv := builder.ForPersistentVolume("testPV").Result()
+	vs := builder.ForVolumeSnapshot("velero", "testVS").Result()
+
+	t.Run("volumeSnapshot is nil", func(t *testing.T) {
+		dd := newDataDownload(restore, backup, dataUploadResult, pvc, pv, "restore-ns", "op-id", "full", nil, false)
+		require.NotNil(t, dd)
+		assert.Equal(t, "restore-ns", dd.Spec.TargetVolume.Namespace)
+		assert.Equal(t, "testPVC", dd.Spec.TargetVolume.PVC)
+		assert.Equal(t, "testPV", dd.Spec.TargetVolume.PV)
+		assert.Equal(t, "full", dd.Spec.RestoreType)
+		assert.Nil(t, dd.Spec.CSISnapshot)
+	})
+
+	t.Run("volumeSnapshot with cleanUp false", func(t *testing.T) {
+		dd := newDataDownload(restore, backup, dataUploadResult, pvc, pv, "restore-ns", "op-id", "incremental", vs, false)
+		require.NotNil(t, dd)
+		assert.Equal(t, "incremental", dd.Spec.RestoreType)
+		require.NotNil(t, dd.Spec.CSISnapshot)
+		assert.Equal(t, "testVS", dd.Spec.CSISnapshot.VolumeSnapshot)
+		assert.Equal(t, "velero", dd.Spec.CSISnapshot.VolumeSnapshotNamespace)
+		assert.False(t, dd.Spec.CSISnapshot.CleanUp)
+	})
+
+	t.Run("volumeSnapshot with cleanUp true", func(t *testing.T) {
+		dd := newDataDownload(restore, backup, dataUploadResult, pvc, pv, "restore-ns", "op-id", "incremental", vs, true)
+		require.NotNil(t, dd)
+		assert.Equal(t, "incremental", dd.Spec.RestoreType)
+		require.NotNil(t, dd.Spec.CSISnapshot)
+		assert.Equal(t, "testVS", dd.Spec.CSISnapshot.VolumeSnapshot)
+		assert.Equal(t, "velero", dd.Spec.CSISnapshot.VolumeSnapshotNamespace)
+		assert.True(t, dd.Spec.CSISnapshot.CleanUp)
+	})
+
+	t.Run("pv is nil", func(t *testing.T) {
+		dd := newDataDownload(restore, backup, dataUploadResult, pvc, nil, "restore-ns", "op-id", "full", nil, false)
+		require.NotNil(t, dd)
+		assert.Empty(t, dd.Spec.TargetVolume.PV)
+	})
+}
 
 func TestPVCAppliesTo(t *testing.T) {
 	p := pvcRestoreItemAction{
@@ -923,3 +1143,247 @@ func TestDeleteExistingPVCFailure(t *testing.T) {
 	assert.Nil(t, returnedPV)
 	assert.Contains(t, err.Error(), "failed to get PV non-existent-pv")
 }
+
+func TestIsCreatedFromSnapshot(t *testing.T) {
+	wrongGroup := "other.group.io"
+	crossNS := "cross-ns"
+
+	tests := []struct {
+		name              string
+		pvc               *corev1api.PersistentVolumeClaim
+		crObjects         []runtime.Object
+		expectedFound     bool
+		expectedVSName    string
+		expectedVSNS      string
+		expectedErrSubstr string
+	}{
+		{
+			name: "dataSource and dataSourceRef are nil",
+			pvc: &corev1api.PersistentVolumeClaim{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "test-pvc",
+					Namespace: "test-ns",
+				},
+			},
+			expectedFound: false,
+		},
+		{
+			name: "dataSource is not VolumeSnapshot Kind",
+			pvc: &corev1api.PersistentVolumeClaim{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "test-pvc",
+					Namespace: "test-ns",
+				},
+				Spec: corev1api.PersistentVolumeClaimSpec{
+					DataSource: &corev1api.TypedLocalObjectReference{
+						APIGroup: ptr.To(snapshotv1api.SchemeGroupVersion.Group),
+						Kind:     "PersistentVolumeClaim",
+						Name:     "source-pvc",
+					},
+				},
+			},
+			expectedFound: false,
+		},
+		{
+			name: "dataSource has nil APIGroup",
+			pvc: &corev1api.PersistentVolumeClaim{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "test-pvc",
+					Namespace: "test-ns",
+				},
+				Spec: corev1api.PersistentVolumeClaimSpec{
+					DataSource: &corev1api.TypedLocalObjectReference{
+						APIGroup: nil,
+						Kind:     "VolumeSnapshot",
+						Name:     "source-vs",
+					},
+				},
+			},
+			expectedFound: false,
+		},
+		{
+			name: "dataSource has wrong APIGroup",
+			pvc: &corev1api.PersistentVolumeClaim{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "test-pvc",
+					Namespace: "test-ns",
+				},
+				Spec: corev1api.PersistentVolumeClaimSpec{
+					DataSource: &corev1api.TypedLocalObjectReference{
+						APIGroup: &wrongGroup,
+						Kind:     "VolumeSnapshot",
+						Name:     "source-vs",
+					},
+				},
+			},
+			expectedFound: false,
+		},
+		{
+			name: "dataSourceRef is not VolumeSnapshot Kind",
+			pvc: &corev1api.PersistentVolumeClaim{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "test-pvc",
+					Namespace: "test-ns",
+				},
+				Spec: corev1api.PersistentVolumeClaimSpec{
+					DataSourceRef: &corev1api.TypedObjectReference{
+						APIGroup: ptr.To(snapshotv1api.SchemeGroupVersion.Group),
+						Kind:     "PersistentVolumeClaim",
+						Name:     "source-pvc",
+					},
+				},
+			},
+			expectedFound: false,
+		},
+		{
+			name: "dataSourceRef has nil APIGroup",
+			pvc: &corev1api.PersistentVolumeClaim{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "test-pvc",
+					Namespace: "test-ns",
+				},
+				Spec: corev1api.PersistentVolumeClaimSpec{
+					DataSourceRef: &corev1api.TypedObjectReference{
+						APIGroup: nil,
+						Kind:     "VolumeSnapshot",
+						Name:     "source-vs",
+					},
+				},
+			},
+			expectedFound: false,
+		},
+		{
+			name: "dataSourceRef has wrong APIGroup",
+			pvc: &corev1api.PersistentVolumeClaim{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "test-pvc",
+					Namespace: "test-ns",
+				},
+				Spec: corev1api.PersistentVolumeClaimSpec{
+					DataSourceRef: &corev1api.TypedObjectReference{
+						APIGroup: &wrongGroup,
+						Kind:     "VolumeSnapshot",
+						Name:     "source-vs",
+					},
+				},
+			},
+			expectedFound: false,
+		},
+		{
+			name: "dataSource VolumeSnapshot not found in crClient",
+			pvc: &corev1api.PersistentVolumeClaim{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "test-pvc",
+					Namespace: "test-ns",
+				},
+				Spec: corev1api.PersistentVolumeClaimSpec{
+					DataSource: &corev1api.TypedLocalObjectReference{
+						APIGroup: ptr.To(snapshotv1api.SchemeGroupVersion.Group),
+						Kind:     "VolumeSnapshot",
+						Name:     "missing-vs",
+					},
+				},
+			},
+			expectedFound:     false,
+			expectedErrSubstr: "fail to get VolumeSnapshot test-ns/missing-vs",
+		},
+		{
+			name: "dataSource VolumeSnapshot found (same namespace)",
+			pvc: &corev1api.PersistentVolumeClaim{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "test-pvc",
+					Namespace: "test-ns",
+				},
+				Spec: corev1api.PersistentVolumeClaimSpec{
+					DataSource: &corev1api.TypedLocalObjectReference{
+						APIGroup: ptr.To(snapshotv1api.SchemeGroupVersion.Group),
+						Kind:     "VolumeSnapshot",
+						Name:     "ready-vs",
+					},
+				},
+			},
+			crObjects: []runtime.Object{
+				builder.ForVolumeSnapshot("test-ns", "ready-vs").Result(),
+			},
+			expectedFound:  true,
+			expectedVSName: "ready-vs",
+			expectedVSNS:   "test-ns",
+		},
+		{
+			name: "dataSourceRef VolumeSnapshot found (cross namespace)",
+			pvc: &corev1api.PersistentVolumeClaim{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "test-pvc",
+					Namespace: "test-ns",
+				},
+				Spec: corev1api.PersistentVolumeClaimSpec{
+					DataSourceRef: &corev1api.TypedObjectReference{
+						APIGroup:  ptr.To(snapshotv1api.SchemeGroupVersion.Group),
+						Kind:      "VolumeSnapshot",
+						Name:      "cross-vs",
+						Namespace: &crossNS,
+					},
+				},
+			},
+			crObjects: []runtime.Object{
+				builder.ForVolumeSnapshot("cross-ns", "cross-vs").Result(),
+			},
+			expectedFound:  true,
+			expectedVSName: "cross-vs",
+			expectedVSNS:   "cross-ns",
+		},
+		{
+			name: "dataSourceRef VolumeSnapshot found (nil namespace falls back to pvc namespace)",
+			pvc: &corev1api.PersistentVolumeClaim{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "test-pvc",
+					Namespace: "test-ns",
+				},
+				Spec: corev1api.PersistentVolumeClaimSpec{
+					DataSourceRef: &corev1api.TypedObjectReference{
+						APIGroup:  ptr.To(snapshotv1api.SchemeGroupVersion.Group),
+						Kind:      "VolumeSnapshot",
+						Name:      "same-ns-vs",
+						Namespace: nil,
+					},
+				},
+			},
+			crObjects: []runtime.Object{
+				builder.ForVolumeSnapshot("test-ns", "same-ns-vs").Result(),
+			},
+			expectedFound:  true,
+			expectedVSName: "same-ns-vs",
+			expectedVSNS:   "test-ns",
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			logger := logrus.New()
+			p := &pvcRestoreItemAction{
+				log:      logger,
+				crClient: velerotest.NewFakeControllerRuntimeClient(t, tc.crObjects...),
+			}
+
+			vs, ok, err := p.isCreatedFromSnapshot(t.Context(), tc.pvc)
+
+			if tc.expectedErrSubstr != "" {
+				require.Error(t, err)
+				assert.Contains(t, err.Error(), tc.expectedErrSubstr)
+				assert.False(t, ok)
+				assert.Nil(t, vs)
+				return
+			}
+
+			require.NoError(t, err)
+			assert.Equal(t, tc.expectedFound, ok)
+			if tc.expectedFound {
+				require.NotNil(t, vs)
+				assert.Equal(t, tc.expectedVSName, vs.Name)
+				assert.Equal(t, tc.expectedVSNS, vs.Namespace)
+			} else {
+				assert.Nil(t, vs)
+			}
+		})
+	}
+}