Merge branch 'main' into save-source-size-to-backup

2026-09-13 11:34:54 +00:00 · 2026-09-09 15:50:06 +08:00
parent e1600caab3 193cfdc58f
commit 6b549f35b3
45 changed files with 983 additions and 186 deletions
@@ -194,7 +194,7 @@ jobs:
      - name: Install MinIO
        run: |
          docker run -d --rm -p 9000:9000 -e "MINIO_ROOT_USER=minio" -e "MINIO_ROOT_PASSWORD=minio123" -e "MINIO_DEFAULT_BUCKETS=bucket,additional-bucket" bitnami/minio:local
-      - uses: helm/kind-action@c72b4750145dbfb1c71734c3782a4db35a1c65c0 # v1 + fix: add curl retry flags (https://github.com/helm/kind-action/pull/165)
+      - uses: helm/kind-action@06c1ae10762d3b9c1644e7fe69596ae519e015a2 # v1 + fix: add curl retry flags (https://github.com/helm/kind-action/pull/165)
        with:
          cluster_name: "kind"
          version: "v0.32.0"
@@ -28,7 +28,7 @@ Please use the version selector at the top of the site to ensure you are using t

 ## Troubleshooting

-If you encounter issues, review the [troubleshooting docs][30], [file an issue][4], or talk to us on the [#velero channel][25] on the Kubernetes Slack server.
+If you encounter issues, review the [troubleshooting docs][30], [file an issue][4], or talk to us on the [#velero-users][25] and [#velero-dev][26] channel on the Kubernetes Slack server.

 ## Community

@@ -36,8 +36,7 @@ Velero is an open community and we welcome your participation. The best way to g

 * Join the [Velero community meetings](https://velero.io/community/), held bi-weekly, alternating between Beijing-friendly and US/Europe-friendly time zones.
 * Subscribe to the [project meeting calendar](https://zoom-lfx.platform.linuxfoundation.org/meetings/velero?view=week).
-* Watch previous meetings on our [YouTube channel](https://www.youtube.com/playlist?list=PL7bmigfV0EqQRysvqvqOtRNk4L5S7uqwM).
-* Chat with us on the [Kubernetes Slack][25] `#velero` channel and join the [mailing list][24].
+* Chat with us on the [Kubernetes Slack][25] `#velero-users` channel and join the [mailing list][24].

 See the [community page](https://velero.io/community/) for the full schedule and details.

@@ -98,7 +97,8 @@ For website terms of use, trademark policy and other project policies please see
 [12]: https://github.com/kubernetes/kubernetes/blob/master/cluster/addons/dns/README.md
 [14]: https://github.com/kubernetes/kubernetes
 [24]: https://groups.google.com/forum/#!forum/projectvelero
-[25]: https://kubernetes.slack.com/messages/velero
+[25]: https://kubernetes.slack.com/messages/velero-users
+[26]: https://kubernetes.slack.com/messages/velero-dev
 [29]: https://velero.io/docs/
 [30]: https://velero.io/docs/troubleshooting
 [31]: https://velero.io/docs/start-contributing
@@ -0,0 +1 @@
+Fix generic CSI changeID retrieval and honor snapshot class deletion policy for CBT retention
@@ -0,0 +1 @@
+Fix datamover backup pod arg mismatch for CSI CBT service account name
@@ -0,0 +1 @@
+Add block data mover support for Velero backup/restore describe CLI.
@@ -0,0 +1 @@
+Add ID to repo snapshot and modify block uploader to use snapshot ID in logs
@@ -0,0 +1 @@
+Add MustIncludeAdditionalItemPVCs to help track BIA added PVC's PVB creation.
@@ -49,7 +49,7 @@ require (
 	golang.org/x/sys v0.47.0
 	golang.org/x/text v0.41.0
 	google.golang.org/api v0.283.0
-	google.golang.org/grpc v1.82.1
+	google.golang.org/grpc v1.83.1
 	google.golang.org/protobuf v1.36.12
 	k8s.io/api v0.36.0
 	k8s.io/apiextensions-apiserver v0.36.0
@@ -66,7 +66,7 @@ require (
 )

 require (
-	cel.dev/expr v0.25.1 // indirect
+	cel.dev/expr v0.25.2 // indirect
 	cloud.google.com/go v0.123.0 // indirect
 	cloud.google.com/go/auth v0.20.0 // indirect
 	cloud.google.com/go/auth/oauth2adapt v0.2.8 // indirect
@@ -76,7 +76,7 @@ require (
 	github.com/Azure/azure-sdk-for-go/sdk/internal v1.12.0 // indirect
 	github.com/Azure/go-ansiterm v0.0.0-20230124172434-306776ec8161 // indirect
 	github.com/AzureAD/microsoft-authentication-library-for-go v1.7.2 // indirect
-	github.com/GoogleCloudPlatform/opentelemetry-operations-go/detectors/gcp v1.32.0 // indirect
+	github.com/GoogleCloudPlatform/opentelemetry-operations-go/detectors/gcp v1.33.0 // indirect
 	github.com/GoogleCloudPlatform/opentelemetry-operations-go/exporter/metric v0.55.0 // indirect
 	github.com/GoogleCloudPlatform/opentelemetry-operations-go/internal/resourcemapping v0.55.0 // indirect
 	github.com/Masterminds/semver/v3 v3.5.0 // indirect
@@ -181,7 +181,7 @@ require (
 	github.com/prometheus/procfs v0.20.1 // indirect
 	github.com/rogpeppe/go-internal v1.14.1 // indirect
 	github.com/rs/xid v1.6.0 // indirect
-	github.com/spiffe/go-spiffe/v2 v2.6.0 // indirect
+	github.com/spiffe/go-spiffe/v2 v2.7.0 // indirect
 	github.com/stretchr/objx v0.5.2 // indirect
 	github.com/tinylib/msgp v1.6.1 // indirect
 	github.com/vladimirvivien/gexe v0.4.0 // indirect
@@ -189,7 +189,7 @@ require (
 	github.com/zeebo/blake3 v0.2.4 // indirect
 	github.com/zeebo/xxh3 v1.1.0 // indirect
 	go.opentelemetry.io/auto/sdk v1.2.1 // indirect
-	go.opentelemetry.io/contrib/detectors/gcp v1.43.0 // indirect
+	go.opentelemetry.io/contrib/detectors/gcp v1.44.0 // indirect
 	go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.67.0 // indirect
 	go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.67.0 // indirect
 	go.opentelemetry.io/otel v1.44.0 // indirect
@@ -1,5 +1,5 @@
-cel.dev/expr v0.25.1 h1:1KrZg61W6TWSxuNZ37Xy49ps13NUovb66QLprthtwi4=
-cel.dev/expr v0.25.1/go.mod h1:hrXvqGP6G6gyx8UAHSHJ5RGk//1Oj5nXQ2NI02Nrsg4=
+cel.dev/expr v0.25.2 h1:K6j46C81hXtZQfuX60cVWQFBJahKSE2gfRbNuvr5bFs=
+cel.dev/expr v0.25.2/go.mod h1:hrXvqGP6G6gyx8UAHSHJ5RGk//1Oj5nXQ2NI02Nrsg4=
 cloud.google.com/go v0.123.0 h1:2NAUJwPR47q+E35uaJeYoNhuNEM9kM8SjgRgdeOJUSE=
 cloud.google.com/go v0.123.0/go.mod h1:xBoMV08QcqUGuPW65Qfm1o9Y4zKZBpGS+7bImXLTAZU=
 cloud.google.com/go/auth v0.20.0 h1:kXTssoVb4azsVDoUiF8KvxAqrsQcQtB53DcSgta74CA=
@@ -48,8 +48,8 @@ github.com/AzureAD/microsoft-authentication-library-for-go v1.7.2 h1:RHK7bS+HQMs
 github.com/AzureAD/microsoft-authentication-library-for-go v1.7.2/go.mod h1:HKpQxkWaGLJ+D/5H8QRpyQXA1eKjxkFlOMwck5+33Jk=
 github.com/GehirnInc/crypt v0.0.0-20230320061759-8cc1b52080c5 h1:IEjq88XO4PuBDcvmjQJcQGg+w+UaafSy8G5Kcb5tBhI=
 github.com/GehirnInc/crypt v0.0.0-20230320061759-8cc1b52080c5/go.mod h1:exZ0C/1emQJAw5tHOaUDyY1ycttqBAPcxuzf7QbY6ec=
-github.com/GoogleCloudPlatform/opentelemetry-operations-go/detectors/gcp v1.32.0 h1:rIkQfkCOVKc1OiRCNcSDD8ml5RJlZbH/Xsq7lbpynwc=
-github.com/GoogleCloudPlatform/opentelemetry-operations-go/detectors/gcp v1.32.0/go.mod h1:RD2SsorTmYhF6HkTmDw7KmPYQk8OBYwTkuasChwv7R4=
+github.com/GoogleCloudPlatform/opentelemetry-operations-go/detectors/gcp v1.33.0 h1:l7+6kwRMJNwdCvYdDl7Eax+wzEYHSnNY7zrrfbhDdTA=
+github.com/GoogleCloudPlatform/opentelemetry-operations-go/detectors/gcp v1.33.0/go.mod h1:pJTkW8hEUIIi3Pf65lPZOnn4Y81yCllX6IWk2jNXdkM=
 github.com/GoogleCloudPlatform/opentelemetry-operations-go/exporter/metric v0.55.0 h1:UnDZ/zFfG1JhH/DqxIZYU/1CUAlTUScoXD/LcM2Ykk8=
 github.com/GoogleCloudPlatform/opentelemetry-operations-go/exporter/metric v0.55.0/go.mod h1:IA1C1U7jO/ENqm/vhi7V9YYpBsp+IMyqNrEN94N7tVc=
 github.com/GoogleCloudPlatform/opentelemetry-operations-go/internal/cloudmock v0.55.0 h1:7t/qx5Ost0s0wbA/VDrByOooURhp+ikYwv20i9Y07TQ=
@@ -417,8 +417,8 @@ github.com/spf13/cobra v1.10.2/go.mod h1:7C1pvHqHw5A4vrJfjNwvOdzYu0Gml16OCs2GRiT
 github.com/spf13/pflag v1.0.9/go.mod h1:McXfInJRrz4CZXVZOBLb0bTZqETkiAhM9Iw0y3An2Bg=
 github.com/spf13/pflag v1.0.10 h1:4EBh2KAYBwaONj6b2Ye1GiHfwjqyROoF4RwYO+vPwFk=
 github.com/spf13/pflag v1.0.10/go.mod h1:McXfInJRrz4CZXVZOBLb0bTZqETkiAhM9Iw0y3An2Bg=
-github.com/spiffe/go-spiffe/v2 v2.6.0 h1:l+DolpxNWYgruGQVV0xsfeya3CsC7m8iBzDnMpsbLuo=
-github.com/spiffe/go-spiffe/v2 v2.6.0/go.mod h1:gm2SeUoMZEtpnzPNs2Csc0D/gX33k1xIx7lEzqblHEs=
+github.com/spiffe/go-spiffe/v2 v2.7.0 h1:uXe1MflJoHw58wAUvxVlcM7WpKtijWG7I1UidcGh6g4=
+github.com/spiffe/go-spiffe/v2 v2.7.0/go.mod h1:47Q0Q9/AqGha8QLHp+kxpH4Wca7X7EnOtlIJy3mxZ3U=
 github.com/stoewer/go-strcase v1.3.0 h1:g0eASXYtp+yvN9fK8sH94oCIk0fau9uV1/ZdJ0AVEzs=
 github.com/stoewer/go-strcase v1.3.0/go.mod h1:fAH5hQ5pehh+j3nZfvwdk2RgEgQjAoM8wodgtPmh1xo=
 github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
@@ -466,8 +466,8 @@ github.com/zeebo/xxh3 v1.1.0 h1:s7DLGDK45Dyfg7++yxI0khrfwq9661w9EN78eP/UZVs=
 github.com/zeebo/xxh3 v1.1.0/go.mod h1:IisAie1LELR4xhVinxWS5+zf1lA4p0MW4T+w+W07F5s=
 go.opentelemetry.io/auto/sdk v1.2.1 h1:jXsnJ4Lmnqd11kwkBV2LgLoFMZKizbCi5fNZ/ipaZ64=
 go.opentelemetry.io/auto/sdk v1.2.1/go.mod h1:KRTj+aOaElaLi+wW1kO/DZRXwkF4C5xPbEe3ZiIhN7Y=
-go.opentelemetry.io/contrib/detectors/gcp v1.43.0 h1:62yY3dT7/ShwOxzA0RsKRgshBmfElKI4d/Myu2OxDFU=
-go.opentelemetry.io/contrib/detectors/gcp v1.43.0/go.mod h1:RyaZMFY7yi1kAs45S6mbFGz8O8rqB0dTY14uzvG4LCs=
+go.opentelemetry.io/contrib/detectors/gcp v1.44.0 h1:NmLfL734pJhM0JKaYd2Y28+nY9dPRWYAAbxhRCrKXPw=
+go.opentelemetry.io/contrib/detectors/gcp v1.44.0/go.mod h1:tNAsgd8avTGke1+MndXlU5Cru4PQ9Ai/cCNWQv/ZJ/s=
 go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.67.0 h1:yI1/OhfEPy7J9eoa6Sj051C7n5dvpj0QX8g4sRchg04=
 go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.67.0/go.mod h1:NoUCKYWK+3ecatC4HjkRktREheMeEtrXoQxrqYFeHSc=
 go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.67.0 h1:OyrsyzuttWTSur2qN/Lm0m2a8yqyIjUVBZcxFPuXq2o=
@@ -564,8 +564,8 @@ google.golang.org/genproto/googleapis/api v0.0.0-20260526163538-3dc84a4a5aaa h1:
 google.golang.org/genproto/googleapis/api v0.0.0-20260526163538-3dc84a4a5aaa/go.mod h1:q4lMZS6kskjT5HvCPrnnypcDPVJqT/f4nfxmkE7gryY=
 google.golang.org/genproto/googleapis/rpc v0.0.0-20260526163538-3dc84a4a5aaa h1:mZHHdPZl0dbGHCflZgAq/Q468DWVFcU2whhB2KAo8fk=
 google.golang.org/genproto/googleapis/rpc v0.0.0-20260526163538-3dc84a4a5aaa/go.mod h1:4Hqkh8ycfw05ld/3BWL7rJOSfebL2Q+DVDeRgYgxUU8=
-google.golang.org/grpc v1.82.1 h1:NnAxzGRA0677vCa4BUkOAnO5+FfQqVl9iUXeD0IqcGE=
-google.golang.org/grpc v1.82.1/go.mod h1:yzTZ1TB1Z3SG+LIYaI+WiE8D5+PZ3ArnrSp8zF3+/ZA=
+google.golang.org/grpc v1.83.1 h1:HIO0+BEtBP6soyqvqC8sNUjZ7bTs+0hFQuFF+RAy++Y=
+google.golang.org/grpc v1.83.1/go.mod h1:kDyl6SKsiHKt0uylY5gtn5cEjkrIOhQOGDgIc4JGwzQ=
 google.golang.org/protobuf v1.36.12 h1:pJOKDDOyeXErUroCihFAd5LQuwXBSpVnKGrj5o/fwxc=
 google.golang.org/protobuf v1.36.12/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
@@ -29,13 +29,14 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	kbclient "sigs.k8s.io/controller-runtime/pkg/client"

-	"github.com/vmware-tanzu/velero/pkg/label"
-
+	veleroshared "github.com/vmware-tanzu/velero/pkg/apis/velero/shared"
 	velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
 	velerov2alpha1 "github.com/vmware-tanzu/velero/pkg/apis/velero/v2alpha1"
+	"github.com/vmware-tanzu/velero/pkg/datamover"
 	"github.com/vmware-tanzu/velero/pkg/features"
 	"github.com/vmware-tanzu/velero/pkg/itemoperation"
 	"github.com/vmware-tanzu/velero/pkg/kuberesource"
+	"github.com/vmware-tanzu/velero/pkg/label"
 	"github.com/vmware-tanzu/velero/pkg/util/stringptr"
 )

@@ -89,11 +90,14 @@ type BackupVolumeInfo struct {
 	// Whether the volume data is backed up successfully.
 	Result VolumeResult `json:"result,omitempty"`

-	CSISnapshotInfo          *CSISnapshotInfo          `json:"csiSnapshotInfo,omitempty"`
-	SnapshotDataMovementInfo *SnapshotDataMovementInfo `json:"snapshotDataMovementInfo,omitempty"`
-	NativeSnapshotInfo       *NativeSnapshotInfo       `json:"nativeSnapshotInfo,omitempty"`
-	PVBInfo                  *PodVolumeInfo            `json:"pvbInfo,omitempty"`
-	PVInfo                   *PVInfo                   `json:"pvInfo,omitempty"`
+	// How volume data is backed up, with possible values including Full and Incremental.
+	BackupType velerov1api.BackupType `json:"backupType,omitempty"`
+
+	CSISnapshotInfo          *CSISnapshotInfo                `json:"csiSnapshotInfo,omitempty"`
+	SnapshotDataMovementInfo *BackupSnapshotDataMovementInfo `json:"snapshotDataMovementInfo,omitempty"`
+	NativeSnapshotInfo       *NativeSnapshotInfo             `json:"nativeSnapshotInfo,omitempty"`
+	PVBInfo                  *PodVolumeBackupInfo            `json:"pvbInfo,omitempty"`
+	PVInfo                   *PVInfo                         `json:"pvInfo,omitempty"`
 }

 type VolumeResult string
@@ -122,10 +126,10 @@ type RestoreVolumeInfo struct {
 	// Whether the volume's data are restored via data movement
 	SnapshotDataMoved bool `json:"snapshotDataMoved"`

-	CSISnapshotInfo          *CSISnapshotInfo          `json:"csiSnapshotInfo,omitempty"`
-	SnapshotDataMovementInfo *SnapshotDataMovementInfo `json:"snapshotDataMovementInfo,omitempty"`
-	NativeSnapshotInfo       *NativeSnapshotInfo       `json:"nativeSnapshotInfo,omitempty"`
-	PVRInfo                  *PodVolumeInfo            `json:"pvrInfo,omitempty"`
+	CSISnapshotInfo          *CSISnapshotInfo                 `json:"csiSnapshotInfo,omitempty"`
+	SnapshotDataMovementInfo *RestoreSnapshotDataMovementInfo `json:"snapshotDataMovementInfo,omitempty"`
+	NativeSnapshotInfo       *NativeSnapshotInfo              `json:"nativeSnapshotInfo,omitempty"`
+	PVRInfo                  *PodVolumeRestoreInfo            `json:"pvrInfo,omitempty"`
 }

 // CSISnapshotInfo is used for displaying the CSI snapshot status
@@ -146,15 +150,15 @@ type CSISnapshotInfo struct {
 	OperationID string `json:"operationID,omitempty"`

 	// The VolumeSnapshot's Status.ReadyToUse value
-	ReadyToUse *bool
+	ReadyToUse *bool `json:"readyToUse,omitempty"`

 	// The VolumeGroupSnapshotHandle from VSC status, used to create stub VGSC during restore
 	// for CSI drivers that populate this field (e.g., Ceph RBD).
 	VolumeGroupSnapshotHandle string `json:"volumeGroupSnapshotHandle,omitempty"`
 }

-// SnapshotDataMovementInfo is used for displaying the snapshot data mover status.
-type SnapshotDataMovementInfo struct {
+// BackupSnapshotDataMovementInfo is used for displaying the backup snapshot data mover status.
+type BackupSnapshotDataMovementInfo struct {
 	// The data mover used by the backup. The valid values are `velero` and ``(equals to `velero`).
 	DataMover string `json:"dataMover"`

@@ -181,11 +185,52 @@ type SnapshotDataMovementInfo struct {
 	// incremental and must stay distinguishable from "unknown".
 	IncrementalSize *int64 `json:"incrementalSize,omitempty"`

+	// ParentSnapshot specifies the parent snapshot that current backup is based on.
+	// If its value is "" or "auto", the data mover finds the recent backup of the same volume as parent.
+	// If its value is "none", the data mover will do a full backup
+	// If its value is a specific snapshotID, the data mover finds the specific snapshot as parent.
+	ParentSnapshot string `json:"parentSnapshot,omitempty"`
+
 	// The size of source volume, for backup only
 	SourceSize int64 `json:"sourceSize,omitempty"`

 	// The DataUpload's Status.Phase value
-	Phase velerov2alpha1.DataUploadPhase
+	Phase velerov2alpha1.DataUploadPhase `json:"phase"`
+}
+
+// RestoreSnapshotDataMovementInfo is used for displaying the restore snapshot data mover status.
+type RestoreSnapshotDataMovementInfo struct {
+	// The data mover used by the backup. The valid values are `velero` and ``(equals to `velero`).
+	DataMover string `json:"dataMover"`
+
+	// The type of the uploader that uploads the snapshot data. The valid values are `kopia` and `restic`.
+	UploaderType string `json:"uploaderType"`
+
+	// The name or ID of the snapshot associated object(SAO).
+	// SAO is used to support local snapshots for the snapshot data mover,
+	// e.g. it could be a VolumeSnapshot for CSI snapshot data movement.
+	RetainedSnapshot string `json:"retainedSnapshot,omitempty"`
+
+	// It's the filesystem repository's snapshot ID.
+	SnapshotHandle string `json:"snapshotHandle"`
+
+	// The Async Operation's ID.
+	OperationID string `json:"operationID"`
+
+	// Moved snapshot data size.
+	Size int64 `json:"size"`
+
+	// Moved snapshot incremental size, i.e. the bytes actually transferred. Nil means
+	// the uploader reported no figure (including backups taken before this was
+	// recorded); a pointer to 0 means it transferred nothing, which is the ideal
+	// incremental and must stay distinguishable from "unknown".
+	IncrementalSize *int64 `json:"incrementalSize,omitempty"`
+
+	// The DataDownload's Status.Phase value
+	Phase velerov2alpha1.DataDownloadPhase `json:"phase"`
+
+	// Indicates the type of the restore, incremental or full.
+	RestoreType string `json:"restoreType,omitempty"`
 }

 // NativeSnapshotInfo is used for displaying the Velero native snapshot status.
@@ -206,7 +251,7 @@ type NativeSnapshotInfo struct {
 	IOPS string `json:"iops"`

 	// The NativeSnapshot's Status.Phase value
-	Phase SnapshotPhase
+	Phase SnapshotPhase `json:"phase,omitempty"`
 }

 func newNativeSnapshotInfo(s *Snapshot) *NativeSnapshotInfo {
@@ -223,9 +268,9 @@ func newNativeSnapshotInfo(s *Snapshot) *NativeSnapshotInfo {
 	}
 }

-// PodVolumeInfo is used for displaying the PodVolumeBackup/PodVolumeRestore snapshot status.
-type PodVolumeInfo struct {
-	// It's the file-system uploader's snapshot ID for PodVolumeBackup/PodVolumeRestore.
+// PodVolumeBackupInfo is used for displaying the PodVolumeBackup snapshot status.
+type PodVolumeBackupInfo struct {
+	// It's the file-system uploader's snapshot ID for PodVolumeBackup.
 	SnapshotHandle string `json:"snapshotHandle,omitempty"`

 	// The snapshot corresponding volume size.
@@ -253,11 +298,47 @@ type PodVolumeInfo struct {
 	NodeName string `json:"nodeName,omitempty"`

 	// The PVB's Status.Phase value
-	Phase velerov1api.PodVolumeBackupPhase
+	Phase velerov1api.PodVolumeBackupPhase `json:"phase,omitempty"`
 }

-func newPodVolumeInfoFromPVB(pvb *velerov1api.PodVolumeBackup) *PodVolumeInfo {
-	return &PodVolumeInfo{
+// PodVolumeRestoreInfo is used for displaying the PodVolumeRestore snapshot status.
+type PodVolumeRestoreInfo struct {
+	// It's the file-system uploader's snapshot ID for PodVolumeRestore.
+	SnapshotHandle string `json:"snapshotHandle,omitempty"`
+
+	// The snapshot corresponding volume size.
+	Size int64 `json:"size,omitempty"`
+
+	// The incremental snapshot size, i.e. the bytes actually transferred. Nil means
+	// the uploader reported no figure; a pointer to 0 means it transferred nothing.
+	IncrementalSize *int64 `json:"incrementalSize,omitempty"`
+
+	// The type of the uploader that uploads the data. The valid values are `kopia` and `restic`.
+	UploaderType string `json:"uploaderType"`
+
+	// The PVC's corresponding volume name used by Pod
+	// https://github.com/kubernetes/kubernetes/blob/e4b74dd12fa8cb63c174091d5536a10b8ec19d34/pkg/apis/core/types.go#L48
+	VolumeName string `json:"volumeName"`
+
+	// The Pod name mounting this PVC.
+	PodName string `json:"podName"`
+
+	// The Pod namespace
+	PodNamespace string `json:"podNamespace"`
+
+	// The PVB-taken k8s node's name.
+	// This field will be empty when the struct is used to represent a podvolumerestore.
+	NodeName string `json:"nodeName,omitempty"`
+
+	// The PVR's Status.Phase value
+	Phase velerov1api.PodVolumeRestorePhase `json:"phase,omitempty"`
+
+	// Indicates the type of the restore, incremental or full.
+	RestoreType string `json:"restoreType,omitempty"`
+}
+
+func newPodVolumeInfoFromPVB(pvb *velerov1api.PodVolumeBackup) *PodVolumeBackupInfo {
+	return &PodVolumeBackupInfo{
 		SnapshotHandle:  pvb.Status.SnapshotID,
 		Size:            pvb.Status.Progress.TotalBytes,
 		IncrementalSize: pvb.Status.IncrementalBytes,
@@ -270,14 +351,16 @@ func newPodVolumeInfoFromPVB(pvb *velerov1api.PodVolumeBackup) *PodVolumeInfo {
 	}
 }

-func newPodVolumeInfoFromPVR(pvr *velerov1api.PodVolumeRestore) *PodVolumeInfo {
-	return &PodVolumeInfo{
+func newPodVolumeInfoFromPVR(pvr *velerov1api.PodVolumeRestore) *PodVolumeRestoreInfo {
+	return &PodVolumeRestoreInfo{
 		SnapshotHandle: pvr.Spec.SnapshotID,
 		Size:           pvr.Status.Progress.TotalBytes,
 		UploaderType:   pvr.Spec.UploaderType,
 		VolumeName:     pvr.Spec.Volume,
 		PodName:        pvr.Spec.Pod.Name,
 		PodNamespace:   pvr.Spec.Pod.Namespace,
+		Phase:          pvr.Status.Phase,
+		RestoreType:    pvr.Spec.RestoreType,
 	}
 }

@@ -473,6 +556,11 @@ func (v *BackupVolumesInformation) generateVolumeInfoForCSIVolumeSnapshot() {
 			volumeGroupSnapshotHandle = *volumeSnapshotContent.Status.VolumeGroupSnapshotHandle
 		}
 		if pvcPVInfo := v.pvMap.retrieve("", *volumeSnapshot.Spec.Source.PersistentVolumeClaimName, volumeSnapshot.Namespace); pvcPVInfo != nil {
+			volumeResult := VolumeResultFailed
+			if volumeSnapshot.Status != nil && volumeSnapshot.Status.ReadyToUse != nil && *volumeSnapshot.Status.ReadyToUse {
+				volumeResult = VolumeResultSucceeded
+			}
+
 			volumeInfo := &BackupVolumeInfo{
 				BackupMethod:          CSISnapshot,
 				PVCName:               pvcPVInfo.PVCName,
@@ -481,6 +569,7 @@ func (v *BackupVolumesInformation) generateVolumeInfoForCSIVolumeSnapshot() {
 				Skipped:               false,
 				SnapshotDataMoved:     false,
 				PreserveLocalSnapshot: true,
+				Result:                volumeResult,
 				CSISnapshotInfo: &CSISnapshotInfo{
 					VSCName:                   *volumeSnapshot.Status.BoundVolumeSnapshotContentName,
 					Size:                      size,
@@ -522,6 +611,11 @@ func (v *BackupVolumesInformation) generateVolumeInfoFromPVB() {
 			StartTimestamp:      pvb.Status.StartTimestamp,
 			CompletionTimestamp: pvb.Status.CompletionTimestamp,
 			PVBInfo:             newPodVolumeInfoFromPVB(pvb),
+			BackupType:          velerov1api.BackupTypeIncremental,
+		}
+
+		if pvb.Spec.ParentSnapshot == veleroshared.ParentSnapshotNone {
+			volumeInfo.BackupType = velerov1api.BackupTypeFull
 		}

 		// Only set Succeeded to true when the PVB's phase is Completed.
@@ -625,6 +719,11 @@ func (v *BackupVolumesInformation) generateVolumeInfoFromDataUpload() {
 				dataMover = dataUpload.Spec.DataMover
 			}

+			volumeResult := VolumeResultFailed
+			if dataUpload.Status.Phase == velerov2alpha1.DataUploadPhaseCompleted {
+				volumeResult = VolumeResultSucceeded
+			}
+
 			volumeInfo := &BackupVolumeInfo{
 				BackupMethod:      CSISnapshot,
 				PVCName:           pvcPVInfo.PVCName,
@@ -632,17 +731,22 @@ func (v *BackupVolumesInformation) generateVolumeInfoFromDataUpload() {
 				PVName:            pvcPVInfo.PV.Name,
 				SnapshotDataMoved: true,
 				Skipped:           false,
+				Result:            volumeResult,
+				BackupType:        velerov1api.BackupTypeIncremental,
 				CSISnapshotInfo: &CSISnapshotInfo{
 					SnapshotHandle: FieldValueIsUnknown,
 					VSCName:        FieldValueIsUnknown,
 					OperationID:    FieldValueIsUnknown,
 					Driver:         dataUpload.Spec.CSISnapshot.Driver,
 				},
-				SnapshotDataMovementInfo: &SnapshotDataMovementInfo{
-					DataMover:    dataMover,
-					UploaderType: velerov1api.BackupRepositoryTypeKopia,
-					OperationID:  operation.Spec.OperationID,
-					Phase:        dataUpload.Status.Phase,
+				SnapshotDataMovementInfo: &BackupSnapshotDataMovementInfo{
+					DataMover:      dataMover,
+					UploaderType:   datamover.GetUploaderType(dataMover),
+					OperationID:    operation.Spec.OperationID,
+					Phase:          dataUpload.Status.Phase,
+					ParentSnapshot: dataUpload.Spec.ParentSnapshot,
+					Size:           dataUpload.Status.Progress.TotalBytes,
+					SnapshotHandle: dataUpload.Status.SnapshotID,
 				},
 				PVInfo: &PVInfo{
 					ReclaimPolicy: string(pvcPVInfo.PV.Spec.PersistentVolumeReclaimPolicy),
@@ -650,10 +754,22 @@ func (v *BackupVolumesInformation) generateVolumeInfoFromDataUpload() {
 				},
 			}

+			if dataUpload.Spec.ParentSnapshot == veleroshared.ParentSnapshotNone {
+				volumeInfo.BackupType = velerov1api.BackupTypeFull
+			}
+
 			if dataUpload.Status.StartTimestamp != nil {
 				volumeInfo.StartTimestamp = dataUpload.Status.StartTimestamp
 			}

+			if dataUpload.Status.CompletionTimestamp != nil {
+				volumeInfo.CompletionTimestamp = dataUpload.Status.CompletionTimestamp
+			}
+
+			if dataUpload.Status.IncrementalBytes != nil {
+				volumeInfo.SnapshotDataMovementInfo.IncrementalSize = dataUpload.Status.IncrementalBytes
+			}
+
 			tmpVolumeInfos = append(tmpVolumeInfos, volumeInfo)
 		} else {
 			v.logger.Warnf("Cannot find info for PVC %s/%s", operation.Spec.ResourceIdentifier.Namespace, operation.Spec.ResourceIdentifier.Name)
@@ -881,14 +997,21 @@ func (t *RestoreVolumeInfoTracker) Result() []*RestoreVolumeInfo {
 			// the datadownload was initiated in CSI plugin
 			// For the same reason, no CSI snapshot info will be populated into volumeInfo
 			RestoreMethod: CSISnapshot,
-			SnapshotDataMovementInfo: &SnapshotDataMovementInfo{
+			SnapshotDataMovementInfo: &RestoreSnapshotDataMovementInfo{
 				DataMover:      dataMover,
-				UploaderType:   velerov1api.BackupRepositoryTypeKopia,
+				UploaderType:   datamover.GetUploaderType(dataMover),
 				SnapshotHandle: dd.Spec.SnapshotID,
 				OperationID:    operationID,
+				RestoreType:    dd.Spec.RestoreType,
+				Size:           dd.Status.Progress.TotalBytes,
+				Phase:          dd.Status.Phase,
 			},
 		}

+		if dd.Status.IncrementalBytes != nil {
+			volumeInfo.SnapshotDataMovementInfo.IncrementalSize = dd.Status.IncrementalBytes
+		}
+
 		volumeInfos = append(volumeInfos, volumeInfo)
 	}

@@ -17,6 +17,7 @@ limitations under the License.
 package volume

 import (
+	"encoding/json"
 	"sync"
 	"testing"

@@ -30,7 +31,9 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/utils/ptr"

+	veleroshared "github.com/vmware-tanzu/velero/pkg/apis/velero/shared"
 	velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
 	velerov2alpha1 "github.com/vmware-tanzu/velero/pkg/apis/velero/v2alpha1"
 	"github.com/vmware-tanzu/velero/pkg/builder"
@@ -474,6 +477,7 @@ func TestGenerateVolumeInfoForCSIVolumeSnapshot(t *testing.T) {
 					PVCNamespace:          "velero",
 					PVName:                "testPV",
 					BackupMethod:          CSISnapshot,
+					Result:                VolumeResultSucceeded,
 					StartTimestamp:        &now,
 					PreserveLocalSnapshot: true,
 					CSISnapshotInfo: &CSISnapshotInfo{
@@ -562,8 +566,9 @@ func TestGenerateVolumeInfoFromPVB(t *testing.T) {
 					PVCNamespace: "",
 					PVName:       "",
 					BackupMethod: PodVolumeBackup,
+					BackupType:   velerov1api.BackupTypeIncremental,
 					Result:       VolumeResultFailed,
-					PVBInfo: &PodVolumeInfo{
+					PVBInfo: &PodVolumeBackupInfo{
 						PodName:      "testPod",
 						PodNamespace: "velero",
 					},
@@ -641,10 +646,11 @@ func TestGenerateVolumeInfoFromPVB(t *testing.T) {
 					PVCNamespace:        "velero",
 					PVName:              "testPV",
 					BackupMethod:        PodVolumeBackup,
+					BackupType:          velerov1api.BackupTypeIncremental,
 					StartTimestamp:      &now,
 					CompletionTimestamp: &now,
 					Result:              VolumeResultFailed,
-					PVBInfo: &PodVolumeInfo{
+					PVBInfo: &PodVolumeBackupInfo{
 						PodName:      "testPod",
 						PodNamespace: "velero",
 						Phase:        velerov1api.PodVolumeBackupPhaseFailed,
@@ -704,10 +710,11 @@ func TestGenerateVolumeInfoFromPVB(t *testing.T) {
 					PVCNamespace:        "velero",
 					PVName:              "testPV",
 					BackupMethod:        PodVolumeBackup,
+					BackupType:          velerov1api.BackupTypeIncremental,
 					StartTimestamp:      &now,
 					CompletionTimestamp: &now,
 					Result:              VolumeResultSucceeded,
-					PVBInfo: &PodVolumeInfo{
+					PVBInfo: &PodVolumeBackupInfo{
 						PodName:      "testPod",
 						PodNamespace: "velero",
 						Phase:        velerov1api.PodVolumeBackupPhaseCompleted,
@@ -838,6 +845,9 @@ func TestGenerateVolumeInfoFromDataUpload(t *testing.T) {
 					Driver:         "pd.csi.storage.gke.io",
 				}).SnapshotID("testSnapshotHandle").
 				StartTimestamp(&now).
+				CompletionTimestamp(&now).
+				TotalBytes(1024).
+				IncrementalBytes(512).
 				Phase(velerov2alpha1.DataUploadPhaseCompleted).
 				Result(),
 			vs:  builder.ForVolumeSnapshot(velerov1api.DefaultNamespace, "vs-01").Status().BoundVolumeSnapshotContentName("vsc-01").Result(),
@@ -882,12 +892,15 @@ func TestGenerateVolumeInfoFromDataUpload(t *testing.T) {
 			},
 			expectedVolumeInfos: []*BackupVolumeInfo{
 				{
-					PVCName:           "testPVC",
-					PVCNamespace:      "velero",
-					PVName:            "testPV",
-					BackupMethod:      CSISnapshot,
-					SnapshotDataMoved: true,
-					StartTimestamp:    &now,
+					PVCName:             "testPVC",
+					PVCNamespace:        "velero",
+					PVName:              "testPV",
+					BackupMethod:        CSISnapshot,
+					SnapshotDataMoved:   true,
+					BackupType:          velerov1api.BackupTypeIncremental,
+					Result:              VolumeResultSucceeded,
+					StartTimestamp:      &now,
+					CompletionTimestamp: &now,
 					CSISnapshotInfo: &CSISnapshotInfo{
 						VSCName:        FieldValueIsUnknown,
 						SnapshotHandle: FieldValueIsUnknown,
@@ -895,11 +908,14 @@ func TestGenerateVolumeInfoFromDataUpload(t *testing.T) {
 						Size:           0,
 						Driver:         "pd.csi.storage.gke.io",
 					},
-					SnapshotDataMovementInfo: &SnapshotDataMovementInfo{
-						DataMover:    "velero",
-						UploaderType: "kopia",
-						OperationID:  "testOperation",
-						Phase:        velerov2alpha1.DataUploadPhaseCompleted,
+					SnapshotDataMovementInfo: &BackupSnapshotDataMovementInfo{
+						DataMover:       "velero",
+						UploaderType:    "kopia",
+						OperationID:     "testOperation",
+						Phase:           velerov2alpha1.DataUploadPhaseCompleted,
+						SnapshotHandle:  "testSnapshotHandle",
+						Size:            1024,
+						IncrementalSize: ptr.To(int64(512)),
 					},
 					PVInfo: &PVInfo{
 						ReclaimPolicy: string(corev1api.PersistentVolumeReclaimDelete),
@@ -947,6 +963,19 @@ func TestGenerateVolumeInfoFromDataUpload(t *testing.T) {
 				require.Equal(t, tc.expectedVolumeInfos[0].PVInfo, volumesInfo.volumeInfos[0].PVInfo)
 				require.Equal(t, tc.expectedVolumeInfos[0].SnapshotDataMovementInfo, volumesInfo.volumeInfos[0].SnapshotDataMovementInfo)
 				require.Equal(t, tc.expectedVolumeInfos[0].CSISnapshotInfo, volumesInfo.volumeInfos[0].CSISnapshotInfo)
+				require.Equal(t, tc.expectedVolumeInfos[0].Result, volumesInfo.volumeInfos[0].Result)
+				if tc.expectedVolumeInfos[0].StartTimestamp != nil {
+					require.NotNil(t, volumesInfo.volumeInfos[0].StartTimestamp)
+					require.True(t, tc.expectedVolumeInfos[0].StartTimestamp.Equal(volumesInfo.volumeInfos[0].StartTimestamp))
+				} else {
+					require.Nil(t, volumesInfo.volumeInfos[0].StartTimestamp)
+				}
+				if tc.expectedVolumeInfos[0].CompletionTimestamp != nil {
+					require.NotNil(t, volumesInfo.volumeInfos[0].CompletionTimestamp)
+					require.True(t, tc.expectedVolumeInfos[0].CompletionTimestamp.Equal(volumesInfo.volumeInfos[0].CompletionTimestamp))
+				} else {
+					require.Nil(t, volumesInfo.volumeInfos[0].CompletionTimestamp)
+				}
 			}
 		})
 	}
@@ -1057,7 +1086,7 @@ func TestRestoreVolumeInfoResult(t *testing.T) {
 					PVName:            "testPV2",
 					RestoreMethod:     PodVolumeRestore,
 					SnapshotDataMoved: false,
-					PVRInfo: &PodVolumeInfo{
+					PVRInfo: &PodVolumeRestoreInfo{
 						SnapshotHandle: "pvr-snap-001",
 						PodName:        "testPod",
 						PodNamespace:   "testNS",
@@ -1127,7 +1156,7 @@ func TestRestoreVolumeInfoResult(t *testing.T) {
 					PVName:            "testPV2",
 					RestoreMethod:     PodVolumeRestore,
 					SnapshotDataMoved: false,
-					PVRInfo: &PodVolumeInfo{
+					PVRInfo: &PodVolumeRestoreInfo{
 						SnapshotHandle: "pvr-snap-001",
 						PodName:        "testPod",
 						PodNamespace:   "testNS",
@@ -1182,6 +1211,9 @@ func TestRestoreVolumeInfoResult(t *testing.T) {
 								PVC:       "testPVC",
 								Namespace: "testNS",
 							}).
+							Phase(velerov2alpha1.DataDownloadPhaseCompleted).
+							Progress(veleroshared.DataMoveOperationProgress{TotalBytes: 2048}).
+							RestoreType("Incremental").
 							Result(),
 						*builder.ForDataDownload("velero", "testDataDownload-2").
 							ObjectMeta(builder.WithLabels(velerov1api.AsyncOperationIDLabel, "dd-operation-002")).
@@ -1190,6 +1222,9 @@ func TestRestoreVolumeInfoResult(t *testing.T) {
 								PVC:       "testPVC2",
 								Namespace: "testNS",
 							}).
+							Phase(velerov2alpha1.DataDownloadPhaseCompleted).
+							Progress(veleroshared.DataMoveOperationProgress{TotalBytes: 4096}).
+							RestoreType("Full").
 							Result(),
 					},
 				},
@@ -1202,11 +1237,14 @@ func TestRestoreVolumeInfoResult(t *testing.T) {
 					PVName:            "testPV",
 					RestoreMethod:     CSISnapshot,
 					SnapshotDataMoved: true,
-					SnapshotDataMovementInfo: &SnapshotDataMovementInfo{
+					SnapshotDataMovementInfo: &RestoreSnapshotDataMovementInfo{
 						DataMover:      "velero",
 						UploaderType:   velerov1api.BackupRepositoryTypeKopia,
 						SnapshotHandle: "dd-snap-001",
 						OperationID:    "dd-operation-001",
+						Phase:          velerov2alpha1.DataDownloadPhaseCompleted,
+						Size:           2048,
+						RestoreType:    "Incremental",
 					},
 				},
 				{
@@ -1215,11 +1253,14 @@ func TestRestoreVolumeInfoResult(t *testing.T) {
 					PVName:            "testPV2",
 					RestoreMethod:     CSISnapshot,
 					SnapshotDataMoved: true,
-					SnapshotDataMovementInfo: &SnapshotDataMovementInfo{
+					SnapshotDataMovementInfo: &RestoreSnapshotDataMovementInfo{
 						DataMover:      "velero",
 						UploaderType:   velerov1api.BackupRepositoryTypeKopia,
 						SnapshotHandle: "dd-snap-002",
 						OperationID:    "dd-operation-002",
+						Phase:          velerov2alpha1.DataDownloadPhaseCompleted,
+						Size:           4096,
+						RestoreType:    "Full",
 					},
 				},
 			},
@@ -1262,3 +1303,174 @@ func TestGetVolumeSnapshotClasses(t *testing.T) {
 	require.NoError(t, err)
 	require.Equal(t, []snapshotv1api.VolumeSnapshotClass{*class}, result)
 }
+
+func TestBackupVolumeInfoJSONRoundTrip(t *testing.T) {
+	orig := BackupVolumeInfo{
+		PVCName:               "pvc-1",
+		PVCNamespace:          "ns-1",
+		PVName:                "pv-1",
+		BackupMethod:          CSISnapshot,
+		SnapshotDataMoved:     true,
+		PreserveLocalSnapshot: false,
+		Skipped:               false,
+		Result:                VolumeResultSucceeded,
+		BackupType:            velerov1api.BackupTypeIncremental,
+		CSISnapshotInfo: &CSISnapshotInfo{
+			SnapshotHandle:            "csi-snap-1",
+			Size:                      2000,
+			Driver:                    "csi.driver.com",
+			VSCName:                   "vsc-1",
+			OperationID:               "op-2",
+			VolumeGroupSnapshotHandle: "vgsh-1",
+		},
+		SnapshotDataMovementInfo: &BackupSnapshotDataMovementInfo{
+			DataMover:        "velero",
+			UploaderType:     "kopia",
+			RetainedSnapshot: "retain-1",
+			SnapshotHandle:   "snap-1",
+			OperationID:      "op-1",
+			Size:             1000,
+			IncrementalSize:  int64Ptr(200),
+			ParentSnapshot:   "parent-1",
+			Phase:            velerov2alpha1.DataUploadPhaseCompleted,
+		},
+		NativeSnapshotInfo: &NativeSnapshotInfo{
+			SnapshotHandle: "native-snap-1",
+			VolumeType:     "gp3",
+			VolumeAZ:       "us-west-2a",
+			IOPS:           "3000",
+		},
+		PVBInfo: &PodVolumeBackupInfo{
+			SnapshotHandle:  "pvb-snap-1",
+			Size:            500,
+			IncrementalSize: int64Ptr(50),
+			UploaderType:    "kopia",
+			VolumeName:      "vol-1",
+			PodName:         "pod-1",
+			PodNamespace:    "ns-1",
+			NodeName:        "node-1",
+			Phase:           velerov1api.PodVolumeBackupPhaseCompleted,
+		},
+		PVInfo: &PVInfo{
+			ReclaimPolicy: "Delete",
+			Labels:        map[string]string{"env": "test"},
+		},
+	}
+
+	data, err := json.Marshal(orig)
+	require.NoError(t, err)
+
+	jsonStr := string(data)
+	assert.Contains(t, jsonStr, `"pvcName":"pvc-1"`)
+	assert.Contains(t, jsonStr, `"pvcNamespace":"ns-1"`)
+	assert.Contains(t, jsonStr, `"pvName":"pv-1"`)
+	assert.Contains(t, jsonStr, `"backupMethod":"CSISnapshot"`)
+	assert.Contains(t, jsonStr, `"snapshotDataMoved":true`)
+	assert.Contains(t, jsonStr, `"result":"succeeded"`)
+	assert.Contains(t, jsonStr, `"backupType":"Incremental"`)
+	assert.Contains(t, jsonStr, `"snapshotDataMovementInfo":{`)
+	assert.Contains(t, jsonStr, `"dataMover":"velero"`)
+	assert.Contains(t, jsonStr, `"uploaderType":"kopia"`)
+	assert.Contains(t, jsonStr, `"retainedSnapshot":"retain-1"`)
+	assert.Contains(t, jsonStr, `"snapshotHandle":"snap-1"`)
+	assert.Contains(t, jsonStr, `"operationID":"op-1"`)
+	assert.Contains(t, jsonStr, `"size":1000`)
+	assert.Contains(t, jsonStr, `"incrementalSize":200`)
+	assert.Contains(t, jsonStr, `"parentSnapshot":"parent-1"`)
+	assert.Contains(t, jsonStr, `"phase":"Completed"`)
+	assert.Contains(t, jsonStr, `"pvbInfo":{`)
+	assert.Contains(t, jsonStr, `"podName":"pod-1"`)
+	assert.Contains(t, jsonStr, `"podNamespace":"ns-1"`)
+	assert.Contains(t, jsonStr, `"nodeName":"node-1"`)
+	assert.Contains(t, jsonStr, `"csiSnapshotInfo":{`)
+	assert.Contains(t, jsonStr, `"driver":"csi.driver.com"`)
+	assert.Contains(t, jsonStr, `"vscName":"vsc-1"`)
+	assert.Contains(t, jsonStr, `"volumeGroupSnapshotHandle":"vgsh-1"`)
+	assert.Contains(t, jsonStr, `"nativeSnapshotInfo":{`)
+	assert.Contains(t, jsonStr, `"volumeType":"gp3"`)
+	assert.Contains(t, jsonStr, `"volumeAZ":"us-west-2a"`)
+	assert.Contains(t, jsonStr, `"iops":"3000"`)
+	assert.Contains(t, jsonStr, `"pvInfo":{`)
+	assert.Contains(t, jsonStr, `"reclaimPolicy":"Delete"`)
+
+	var unmarshaled BackupVolumeInfo
+	err = json.Unmarshal(data, &unmarshaled)
+	require.NoError(t, err)
+	assert.Equal(t, orig, unmarshaled)
+}
+
+func TestRestoreVolumeInfoJSONRoundTrip(t *testing.T) {
+	orig := RestoreVolumeInfo{
+		PVCName:           "pvc-2",
+		PVCNamespace:      "ns-2",
+		PVName:            "pv-2",
+		RestoreMethod:     CSISnapshot,
+		SnapshotDataMoved: true,
+		SnapshotDataMovementInfo: &RestoreSnapshotDataMovementInfo{
+			DataMover:        "velero",
+			UploaderType:     "kopia",
+			RetainedSnapshot: "retain-2",
+			SnapshotHandle:   "snap-2",
+			OperationID:      "op-3",
+			Size:             3000,
+			IncrementalSize:  int64Ptr(300),
+			Phase:            velerov2alpha1.DataDownloadPhaseCompleted,
+			RestoreType:      "Incremental",
+		},
+		PVRInfo: &PodVolumeRestoreInfo{
+			SnapshotHandle:  "pvr-snap-1",
+			Size:            600,
+			IncrementalSize: int64Ptr(60),
+			UploaderType:    "kopia",
+			VolumeName:      "vol-2",
+			PodName:         "pod-2",
+			PodNamespace:    "ns-2",
+			NodeName:        "node-2",
+			Phase:           velerov1api.PodVolumeRestorePhaseCompleted,
+			RestoreType:     "Incremental",
+		},
+		CSISnapshotInfo: &CSISnapshotInfo{
+			SnapshotHandle: "csi-snap-2",
+			Size:           4000,
+			Driver:         "csi.driver.com",
+			VSCName:        "vsc-2",
+		},
+		NativeSnapshotInfo: &NativeSnapshotInfo{
+			SnapshotHandle: "native-snap-2",
+			VolumeType:     "ebs",
+			VolumeAZ:       "us-east-1a",
+			IOPS:           "1000",
+		},
+	}
+
+	data, err := json.Marshal(orig)
+	require.NoError(t, err)
+
+	jsonStr := string(data)
+	assert.Contains(t, jsonStr, `"pvcName":"pvc-2"`)
+	assert.Contains(t, jsonStr, `"pvcNamespace":"ns-2"`)
+	assert.Contains(t, jsonStr, `"pvName":"pv-2"`)
+	assert.Contains(t, jsonStr, `"restoreMethod":"CSISnapshot"`)
+	assert.Contains(t, jsonStr, `"snapshotDataMoved":true`)
+	assert.Contains(t, jsonStr, `"snapshotDataMovementInfo":{`)
+	assert.Contains(t, jsonStr, `"dataMover":"velero"`)
+	assert.Contains(t, jsonStr, `"uploaderType":"kopia"`)
+	assert.Contains(t, jsonStr, `"retainedSnapshot":"retain-2"`)
+	assert.Contains(t, jsonStr, `"snapshotHandle":"snap-2"`)
+	assert.Contains(t, jsonStr, `"operationID":"op-3"`)
+	assert.Contains(t, jsonStr, `"size":3000`)
+	assert.Contains(t, jsonStr, `"incrementalSize":300`)
+	assert.Contains(t, jsonStr, `"phase":"Completed"`)
+	assert.Contains(t, jsonStr, `"restoreType":"Incremental"`)
+	assert.Contains(t, jsonStr, `"pvrInfo":{`)
+	assert.Contains(t, jsonStr, `"podName":"pod-2"`)
+	assert.Contains(t, jsonStr, `"podNamespace":"ns-2"`)
+	assert.Contains(t, jsonStr, `"nodeName":"node-2"`)
+	assert.Contains(t, jsonStr, `"csiSnapshotInfo":{`)
+	assert.Contains(t, jsonStr, `"nativeSnapshotInfo":{`)
+
+	var unmarshaled RestoreVolumeInfo
+	err = json.Unmarshal(data, &unmarshaled)
+	require.NoError(t, err)
+	assert.Equal(t, orig, unmarshaled)
+}
@@ -38,6 +38,10 @@ type volumeHelperImpl struct {
 	// pvcPodCache provides cached PVC to Pod mappings for improved performance.
 	// When there are many PVCs and pods, using this cache avoids O(N*M) lookups.
 	pvcPodCache *podvolumeutil.PVCPodCache
+	// pvcMustInclusionTracker provides read-only checks for whether a PVC is included
+	// in the backup as BIA's additionalItems through annotation
+	// backup.velero.io/must-include-additional-items.
+	pvcMustInclusionTracker vhutil.PVCMustInclusionTracker
 }

 // NewVolumeHelperImpl creates a VolumeHelper without PVC-to-Pod caching.
@@ -52,6 +56,7 @@ func NewVolumeHelperImpl(
 	client crclient.Client,
 	defaultVolumesToFSBackup bool,
 	backupExcludePVC bool,
+	pvcMustInclusionTracker vhutil.PVCMustInclusionTracker,
 ) vhutil.VolumeHelper {
 	// Pass nil namespaces - no cache will be built, so this never fails.
 	// This is used by plugins that don't need the cache optimization.
@@ -63,6 +68,7 @@ func NewVolumeHelperImpl(
 		defaultVolumesToFSBackup,
 		backupExcludePVC,
 		nil,
+		pvcMustInclusionTracker,
 	)
 	return vh
 }
@@ -80,6 +86,7 @@ func NewVolumeHelperImplWithNamespaces(
 	defaultVolumesToFSBackup bool,
 	backupExcludePVC bool,
 	namespaces []string,
+	pvcMustInclusionTracker vhutil.PVCMustInclusionTracker,
 ) (vhutil.VolumeHelper, error) {
 	var pvcPodCache *podvolumeutil.PVCPodCache
 	if len(namespaces) > 0 {
@@ -98,6 +105,7 @@ func NewVolumeHelperImplWithNamespaces(
 		defaultVolumesToFSBackup: defaultVolumesToFSBackup,
 		backupExcludePVC:         backupExcludePVC,
 		pvcPodCache:              pvcPodCache,
+		pvcMustInclusionTracker:  pvcMustInclusionTracker,
 	}, nil
 }

@@ -109,6 +117,7 @@ func NewVolumeHelperImplWithCache(
 	client crclient.Client,
 	logger logrus.FieldLogger,
 	pvcPodCache *podvolumeutil.PVCPodCache,
+	pvcMustInclusionTracker vhutil.PVCMustInclusionTracker,
 ) (vhutil.VolumeHelper, error) {
 	resourcePolicies, err := resourcepolicies.GetResourcePoliciesFromBackup(backup, client, logger)
 	if err != nil {
@@ -123,6 +132,7 @@ func NewVolumeHelperImplWithCache(
 		defaultVolumesToFSBackup: boolptr.IsSetToTrue(backup.Spec.DefaultVolumesToFsBackup),
 		backupExcludePVC:         boolptr.IsSetToTrue(backup.Spec.SnapshotMoveData),
 		pvcPodCache:              pvcPodCache,
+		pvcMustInclusionTracker:  pvcMustInclusionTracker,
 	}, nil
 }

@@ -260,7 +270,7 @@ func (v *volumeHelperImpl) ShouldPerformSnapshot(obj runtime.Unstructured, group
 }

 func (v volumeHelperImpl) ShouldPerformFSBackup(volume corev1api.Volume, pod corev1api.Pod) (bool, error) {
-	if !v.shouldIncludeVolumeInBackup(volume) {
+	if !v.shouldIncludeVolumeInBackup(volume, pod) {
 		v.logger.Debugf("skip fs-backup action for pod %s's volume %s, due to not pass volume check.", pod.Namespace+"/"+pod.Name, volume.Name)
 		return false, nil
 	}
@@ -442,7 +452,7 @@ func (v *volumeHelperImpl) GetSnapshotClass(obj runtime.Unstructured, groupResou
 	return action.GetSnapshotClass()
 }

-func (v *volumeHelperImpl) shouldIncludeVolumeInBackup(vol corev1api.Volume) bool {
+func (v *volumeHelperImpl) shouldIncludeVolumeInBackup(vol corev1api.Volume, pod corev1api.Pod) bool {
 	includeVolumeInBackup := true
 	// cannot backup hostpath volumes as they are not mounted into /var/lib/kubelet/pods
 	// and therefore not accessible to the node agent daemon set.
@@ -465,8 +475,12 @@ func (v *volumeHelperImpl) shouldIncludeVolumeInBackup(vol corev1api.Volume) boo
 	if vol.DownwardAPI != nil {
 		includeVolumeInBackup = false
 	}
-	if vol.PersistentVolumeClaim != nil && v.backupExcludePVC {
-		includeVolumeInBackup = false
+	if vol.PersistentVolumeClaim != nil {
+		if v.backupExcludePVC {
+			if v.pvcMustInclusionTracker == nil || !v.pvcMustInclusionTracker.IsPVCIncluded(pod.Namespace, vol.PersistentVolumeClaim.ClaimName) {
+				includeVolumeInBackup = false
+			}
+		}
 	}
 	// don't include volumes that mount the default service account token.
 	if strings.HasPrefix(vol.Name, "default-token") {
@@ -35,6 +35,7 @@ import (
 	"github.com/vmware-tanzu/velero/pkg/kuberesource"
 	velerotest "github.com/vmware-tanzu/velero/pkg/test"
 	podvolumeutil "github.com/vmware-tanzu/velero/pkg/util/podvolume"
+	vhutil "github.com/vmware-tanzu/velero/pkg/util/volumehelper"
 )

 func TestVolumeHelperImpl_ShouldPerformSnapshot(t *testing.T) {
@@ -329,6 +330,7 @@ func TestVolumeHelperImpl_ShouldPerformSnapshot(t *testing.T) {
 				fakeClient,
 				tc.defaultVolumesToFSBackup,
 				false,
+				nil,
 			)

 			obj, err := runtime.DefaultUnstructuredConverter.ToUnstructured(tc.inputObj)
@@ -345,11 +347,23 @@ func TestVolumeHelperImpl_ShouldPerformSnapshot(t *testing.T) {
 	}
 }

+type mockPVCMustInclusionTracker struct {
+	isPVCIncluded func(namespace, pvcName string) bool
+}
+
+func (m *mockPVCMustInclusionTracker) IsPVCIncluded(namespace, pvcName string) bool {
+	if m.isPVCIncluded == nil {
+		return false
+	}
+	return m.isPVCIncluded(namespace, pvcName)
+}
+
 func TestVolumeHelperImpl_ShouldIncludeVolumeInBackup(t *testing.T) {
 	testCases := []struct {
 		name             string
 		vol              corev1api.Volume
 		backupExcludePVC bool
+		isPVCIncluded    func(pvcName string) bool
 		shouldInclude    bool
 	}{
 		{
@@ -445,6 +459,38 @@ func TestVolumeHelperImpl_ShouldIncludeVolumeInBackup(t *testing.T) {
 			backupExcludePVC: true,
 			shouldInclude:    false,
 		},
+		{
+			name: "volume has pvc, backupExcludePVC is true, but isPVCIncluded returns true so include",
+			vol: corev1api.Volume{
+				Name: "sample-volume",
+				VolumeSource: corev1api.VolumeSource{
+					PersistentVolumeClaim: &corev1api.PersistentVolumeClaimVolumeSource{
+						ClaimName: "sample-pvc",
+					},
+				},
+			},
+			backupExcludePVC: true,
+			isPVCIncluded: func(pvcName string) bool {
+				return pvcName == "sample-pvc"
+			},
+			shouldInclude: true,
+		},
+		{
+			name: "volume has pvc, backupExcludePVC is false, isPVCIncluded returns false, but globally included so include",
+			vol: corev1api.Volume{
+				Name: "sample-volume",
+				VolumeSource: corev1api.VolumeSource{
+					PersistentVolumeClaim: &corev1api.PersistentVolumeClaimVolumeSource{
+						ClaimName: "sample-pvc",
+					},
+				},
+			},
+			backupExcludePVC: false,
+			isPVCIncluded: func(pvcName string) bool {
+				return false
+			},
+			shouldInclude: true,
+		},
 		{
 			name: "volume name has prefix default-token so do not include",
 			vol: corev1api.Volume{
@@ -480,13 +526,23 @@ func TestVolumeHelperImpl_ShouldIncludeVolumeInBackup(t *testing.T) {
 			if err != nil {
 				t.Fatalf("failed to build policy with error %v", err)
 			}
-			vh := &volumeHelperImpl{
-				volumePolicy:     p,
-				snapshotVolumes:  ptr.To(true),
-				logger:           velerotest.NewLogger(),
-				backupExcludePVC: tc.backupExcludePVC,
+			var tracker vhutil.PVCMustInclusionTracker
+			if tc.isPVCIncluded != nil {
+				tracker = &mockPVCMustInclusionTracker{
+					isPVCIncluded: func(ns, pvcName string) bool {
+						return tc.isPVCIncluded(pvcName)
+					},
+				}
 			}
-			actualShouldInclude := vh.shouldIncludeVolumeInBackup(tc.vol)
+			vh := &volumeHelperImpl{
+				volumePolicy:            p,
+				snapshotVolumes:         ptr.To(true),
+				logger:                  velerotest.NewLogger(),
+				backupExcludePVC:        tc.backupExcludePVC,
+				pvcMustInclusionTracker: tracker,
+			}
+			pod := corev1api.Pod{ObjectMeta: metav1.ObjectMeta{Namespace: "default"}}
+			actualShouldInclude := vh.shouldIncludeVolumeInBackup(tc.vol, pod)
 			assert.Equalf(t, actualShouldInclude, tc.shouldInclude, "Want shouldInclude as %v; Got actualShouldInclude as %v", tc.shouldInclude, actualShouldInclude)
 		})
 	}
@@ -694,6 +750,7 @@ func TestVolumeHelperImpl_ShouldPerformFSBackup(t *testing.T) {
 				fakeClient,
 				tc.defaultVolumesToFSBackup,
 				false,
+				nil,
 			)

 			actualShouldFSBackup, actualError := vh.ShouldPerformFSBackup(tc.pod.Spec.Volumes[0], *tc.pod)
@@ -889,6 +946,7 @@ func TestVolumeHelperImplWithCache_ShouldPerformSnapshot(t *testing.T) {
 				tc.defaultVolumesToFSBackup,
 				false,
 				namespaces,
+				nil,
 			)
 			require.NoError(t, err)

@@ -1041,6 +1099,7 @@ func TestVolumeHelperImplWithCache_ShouldPerformFSBackup(t *testing.T) {
 				tc.defaultVolumesToFSBackup,
 				false,
 				namespaces,
+				nil,
 			)
 			require.NoError(t, err)

@@ -1166,6 +1225,7 @@ volumePolicies:
 				fakeClient,
 				logrus.StandardLogger(),
 				cache,
+				nil,
 			)

 			if tc.expectError {
@@ -1221,7 +1281,7 @@ func TestNewVolumeHelperImplWithCache_UsesCache(t *testing.T) {
 		},
 	}

-	vh, err := NewVolumeHelperImplWithCache(backup, fakeClient, logrus.StandardLogger(), cache)
+	vh, err := NewVolumeHelperImplWithCache(backup, fakeClient, logrus.StandardLogger(), cache, nil)
 	require.NoError(t, err)

 	// Convert PV to unstructured
@@ -1353,6 +1413,7 @@ func TestVolumeHelperImpl_ShouldPerformSnapshot_UnboundPVC(t *testing.T) {
 				fakeClient,
 				false,
 				false,
+				nil,
 			)

 			obj, err := runtime.DefaultUnstructuredConverter.ToUnstructured(tc.inputPVC)
@@ -1530,6 +1591,7 @@ func TestVolumeHelperImpl_ShouldPerformFSBackup_UnboundPVC(t *testing.T) {
 				fakeClient,
 				false,
 				false,
+				nil,
 			)

 			actualShouldFSBackup, actualError := vh.ShouldPerformFSBackup(tc.pod.Spec.Volumes[0], *tc.pod)
@@ -1669,6 +1731,7 @@ func TestGetDataMoverFromActionParameters(t *testing.T) {
 				fakeClient,
 				false,
 				false,
+				nil,
 			)

 			obj, err := runtime.DefaultUnstructuredConverter.ToUnstructured(tc.inputObj)
@@ -1794,6 +1857,7 @@ func TestGetActionParameters(t *testing.T) {
 				fakeClient,
 				false,
 				false,
+				nil,
 			)

 			obj, err := runtime.DefaultUnstructuredConverter.ToUnstructured(tc.inputObj)
@@ -1980,6 +2044,7 @@ func TestShouldPerformCustomAction(t *testing.T) {
 				fakeClient,
 				false,
 				false,
+				nil,
 			)

 			obj, err := runtime.DefaultUnstructuredConverter.ToUnstructured(tc.inputObj)
@@ -2102,6 +2167,7 @@ func TestGetPVAndMatchAction(t *testing.T) {
 				fakeClient,
 				false,
 				false,
+				nil,
 			)

 			obj, err := runtime.DefaultUnstructuredConverter.ToUnstructured(tc.inputObj)
@@ -136,6 +136,7 @@ func (p *pvcBackupItemAction) getVolumeHelperWithCache(backup *velerov1api.Backu
 		p.crClient,
 		p.log,
 		p.pvcPodCache,
+		nil,
 	)
 	if err != nil {
 		return nil, errors.Wrap(err, "failed to create VolumeHelper")
@@ -485,6 +485,8 @@ func (kb *kubernetesBackupper) BackupWithResolvers(
 		return err
 	}

+	pvcMustInclusionTracker := NewPVCMustInclusionTracker(backupRequest.MustIncludeAdditionalItemPVCs)
+
 	volumeHelperImpl, err := volumehelper.NewVolumeHelperImplWithNamespaces(
 		backupRequest.ResPolicies,
 		backupRequest.Spec.SnapshotVolumes,
@@ -493,6 +495,7 @@ func (kb *kubernetesBackupper) BackupWithResolvers(
 		boolptr.IsSetToTrue(backupRequest.Spec.DefaultVolumesToFsBackup),
 		!backupRequest.ResourceIncludesExcludes.ShouldInclude(kuberesource.PersistentVolumeClaims.String()),
 		namespaces,
+		pvcMustInclusionTracker,
 	)
 	if err != nil {
 		log.WithError(err).Error("Failed to build PVC-to-Pod cache for volume policy lookups")
@@ -5666,7 +5666,7 @@ func TestUpdateVolumeInfos(t *testing.T) {
 					PVCName:             "pvc-1",
 					PVCNamespace:        "ns-1",
 					CompletionTimestamp: &metav1.Time{},
-					SnapshotDataMovementInfo: &volume.SnapshotDataMovementInfo{
+					SnapshotDataMovementInfo: &volume.BackupSnapshotDataMovementInfo{
 						DataMover: "velero",
 					},
 				},
@@ -5677,7 +5677,7 @@ func TestUpdateVolumeInfos(t *testing.T) {
 					PVCNamespace:        "ns-1",
 					CompletionTimestamp: &now,
 					Result:              volume.VolumeResultFailed,
-					SnapshotDataMovementInfo: &volume.SnapshotDataMovementInfo{
+					SnapshotDataMovementInfo: &volume.BackupSnapshotDataMovementInfo{
 						DataMover:        "velero",
 						RetainedSnapshot: "vs-1",
 						SnapshotHandle:   "snapshot-id",
@@ -5706,7 +5706,7 @@ func TestUpdateVolumeInfos(t *testing.T) {
 					PVCName:             "pvc-1",
 					PVCNamespace:        "ns-1",
 					CompletionTimestamp: &metav1.Time{},
-					SnapshotDataMovementInfo: &volume.SnapshotDataMovementInfo{
+					SnapshotDataMovementInfo: &volume.BackupSnapshotDataMovementInfo{
 						DataMover: "velero",
 					},
 				},
@@ -5717,7 +5717,7 @@ func TestUpdateVolumeInfos(t *testing.T) {
 					PVCNamespace:        "ns-1",
 					CompletionTimestamp: &now,
 					Result:              volume.VolumeResultSucceeded,
-					SnapshotDataMovementInfo: &volume.SnapshotDataMovementInfo{
+					SnapshotDataMovementInfo: &volume.BackupSnapshotDataMovementInfo{
 						DataMover:        "velero",
 						RetainedSnapshot: "vs-1",
 						SnapshotHandle:   "snapshot-id",
@@ -245,6 +245,7 @@ func (ib *itemBackupper) backupItemInternal(logger logrus.FieldLogger, obj runti
 			// where it's been backed up from another pod), since we don't need >1 backup per PVC.
 			for _, volume := range pod.Spec.Volumes {
 				shouldDoFSBackup, err := ib.volumeHelperImpl.ShouldPerformFSBackup(volume, *pod)
+
 				if err != nil {
 					backupErrs = append(backupErrs, errors.WithStack(err))
 				}
@@ -480,6 +481,26 @@ func (ib *itemBackupper) executeActions(
 		delete(u.GetAnnotations(), velerov1api.MustIncludeAdditionalItemAnnotation)
 		obj = u

+		// If the BIA specifies that additional items must be included, we track any PVCs returned as additional items.
+		// This tracking is necessary because the FSB (File System Backup) evaluation for a Pod
+		// happens before its PVCs are processed. By tracking these explicitly included PVCs here,
+		// the FSB logic can correctly determine that the PVC will be backed up and therefore
+		// a PodVolumeBackup should be created.
+		// We track this unconditionally when mustInclude is true, because fine-grained backup filters
+		// might exclude a PVC even if it's globally included, but mustInclude overrides those filters.
+		if mustInclude && ib.backupRequest.MustIncludeAdditionalItemPVCs != nil {
+			for _, additionalItem := range additionalItemIdentifiers {
+				if additionalItem.GroupResource == kuberesource.PersistentVolumeClaims {
+					key := itemKey{
+						resource:  additionalItem.GroupResource.String(),
+						namespace: additionalItem.Namespace,
+						name:      additionalItem.Name,
+					}
+					ib.backupRequest.MustIncludeAdditionalItemPVCs.AddItem(key)
+				}
+			}
+		}
+
 		// If async plugin started async operation, add it to the ItemOperations list
 		// ignore during finalize phase
 		if operationID != "" {
@@ -0,0 +1,50 @@
+/*
+Copyright the Velero contributors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package backup
+
+import (
+	"github.com/vmware-tanzu/velero/pkg/kuberesource"
+	vhutil "github.com/vmware-tanzu/velero/pkg/util/volumehelper"
+)
+
+// pvcMustInclusionTracker provides read-only checks for whether a PVC is included
+// in the backup as BIA's additionalItems through annotation
+// backup.velero.io/must-include-additional-items.
+type pvcMustInclusionTracker struct {
+	mustInclude *backedUpItemsMap
+}
+
+func NewPVCMustInclusionTracker(mustInclude *backedUpItemsMap) vhutil.PVCMustInclusionTracker {
+	return &pvcMustInclusionTracker{
+		mustInclude: mustInclude,
+	}
+}
+
+func (p *pvcMustInclusionTracker) IsPVCIncluded(namespace, pvcName string) bool {
+	pvcKey := itemKey{
+		resource:  kuberesource.PersistentVolumeClaims.String(),
+		namespace: namespace,
+		name:      pvcName,
+	}
+
+	// 1. If the PVC was explicitly forced into the backup by a BIA, it will be backed up.
+	if p.mustInclude != nil && p.mustInclude.Has(pvcKey) {
+		return true
+	}
+
+	return false
+}
@@ -0,0 +1,47 @@
+/*
+Copyright the Velero contributors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package backup
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+
+	"github.com/vmware-tanzu/velero/pkg/kuberesource"
+)
+
+func TestPVCMustInclusionTracker_IsPVCIncluded(t *testing.T) {
+	mustIncludeMap := NewBackedUpItemsMap()
+
+	tracker := NewPVCMustInclusionTracker(mustIncludeMap)
+
+	pvcKey1 := itemKey{
+		resource:  kuberesource.PersistentVolumeClaims.String(),
+		namespace: "ns-1",
+		name:      "pvc-1",
+	}
+
+	// Initially neither PVC is included
+	assert.False(t, tracker.IsPVCIncluded("ns-1", "pvc-1"))
+
+	// Add pvc-1 to mustInclude map
+	mustIncludeMap.AddItem(pvcKey1)
+	assert.True(t, tracker.IsPVCIncluded("ns-1", "pvc-1"))
+
+	// Check a PVC not in any map
+	assert.False(t, tracker.IsPVCIncluded("ns-1", "pvc-2"))
+}
@@ -83,11 +83,16 @@ type Request struct {
 	VolumeSnapshots           SynchronizedVSList
 	PodVolumeBackups          []*velerov1api.PodVolumeBackup
 	BackedUpItems             *backedUpItemsMap
-	itemOperationsList        *[]*itemoperation.BackupOperation
-	ResPolicies               *resourcepolicies.Policies
-	SkippedPVTracker          *skipPVTracker
-	VolumesInformation        volume.BackupVolumesInformation
-	WorkerPool                *ItemBlockWorkerPool
+	// MustIncludeAdditionalItemPVCs keeps track of PVCs that are returned as additionalItems
+	// by a BackupItemAction plugin with the must-include annotation. This is specifically
+	// used to ensure PodVolumeBackups (FSB) are created for these PVCs even when PVCs are
+	// excluded by global or fine-grained backup resource filters.
+	MustIncludeAdditionalItemPVCs *backedUpItemsMap
+	itemOperationsList            *[]*itemoperation.BackupOperation
+	ResPolicies                   *resourcepolicies.Policies
+	SkippedPVTracker              *skipPVTracker
+	VolumesInformation            volume.BackupVolumesInformation
+	WorkerPool                    *ItemBlockWorkerPool

 	// ClusterScopedFilterMap holds resolved global filters for cluster-scoped resources.
 	// Key is the resolved group-resource string.
@@ -61,6 +61,12 @@ func (d *DataDownloadBuilder) Phase(phase velerov2alpha1api.DataDownloadPhase) *
 	return d
 }

+// RestoreType sets the DataDownload's RestoreType.
+func (d *DataDownloadBuilder) RestoreType(restoreType string) *DataDownloadBuilder {
+	d.object.Spec.RestoreType = restoreType
+	return d
+}
+
 // SnapshotID sets the DataDownload's SnapshotID.
 func (d *DataDownloadBuilder) SnapshotID(id string) *DataDownloadBuilder {
 	d.object.Spec.SnapshotID = id
@@ -249,6 +249,10 @@ func DescribeBackupSpec(d *Describer, spec velerov1api.BackupSpec) {
 	}
 	d.Printf("Data Mover:\t%s\n", s)

+	if string(spec.BackupType) != "" {
+		d.Printf("Backup Type:\t%s\n", spec.BackupType)
+	}
+
 	d.Println()
 	d.Printf("TTL:\t%s\n", spec.TTL.Duration)

@@ -746,6 +750,10 @@ func describeDataMovement(d *Describer, details bool, info *volume.BackupVolumeI
 		if info.SnapshotDataMovementInfo.IncrementalSize != nil {
 			d.Printf("\t\t\t\tIncremental data Size (bytes): %d\n", *info.SnapshotDataMovementInfo.IncrementalSize)
 		}
+		if info.SnapshotDataMovementInfo.ParentSnapshot != "" {
+			d.Printf("\t\t\t\tParent Snapshot: %s\n", info.SnapshotDataMovementInfo.ParentSnapshot)
+		}
+
 		d.Printf("\t\t\t\tResult: %s\n", info.Result)
 	} else {
 		d.Printf("\t\t\tData Movement: %s\n", "included, specify --details for more information")
@@ -108,6 +108,7 @@ func TestDescribeBackupSpec(t *testing.T) {
 		TTL(72 * time.Hour).
 		CSISnapshotTimeout(10 * time.Minute).
 		DataMover("mover").
+		BackupType(velerov1api.BackupTypeFull).
 		Hooks(velerov1api.BackupHooks{
 			Resources: []velerov1api.BackupResourceHookSpec{
 				{
@@ -156,6 +157,7 @@ Storage Location:  backup-location
 Velero-Native Snapshot PVs:  auto
 Snapshot Move Data:          auto
 Data Mover:                  mover
+Backup Type:                 Full

 TTL:  72h0m0s

@@ -575,7 +577,7 @@ func TestCSISnapshots(t *testing.T) {
 					PVCNamespace:      "pvc-ns-3",
 					PVCName:           "pvc-3",
 					SnapshotDataMoved: true,
-					SnapshotDataMovementInfo: &volume.SnapshotDataMovementInfo{
+					SnapshotDataMovementInfo: &volume.BackupSnapshotDataMovementInfo{
 						DataMover:      "velero",
 						UploaderType:   "fake-uploader",
 						SnapshotHandle: "fake-repo-id-3",
@@ -597,7 +599,7 @@ func TestCSISnapshots(t *testing.T) {
 					PVCName:           "pvc-4",
 					SnapshotDataMoved: true,
 					Result:            volume.VolumeResultSucceeded,
-					SnapshotDataMovementInfo: &volume.SnapshotDataMovementInfo{
+					SnapshotDataMovementInfo: &volume.BackupSnapshotDataMovementInfo{
 						DataMover:      "velero",
 						UploaderType:   "fake-uploader",
 						SnapshotHandle: "fake-repo-id-4",
@@ -625,13 +627,15 @@ func TestCSISnapshots(t *testing.T) {
 					PVCName:           "pvc-5",
 					Result:            volume.VolumeResultFailed,
 					SnapshotDataMoved: true,
-					SnapshotDataMovementInfo: &volume.SnapshotDataMovementInfo{
+					BackupType:        velerov1api.BackupTypeIncremental,
+					SnapshotDataMovementInfo: &volume.BackupSnapshotDataMovementInfo{
 						UploaderType:    "fake-uploader",
 						SnapshotHandle:  "fake-repo-id-5",
 						OperationID:     "fake-operation-5",
 						Size:            100,
 						IncrementalSize: ptr.To(int64(50)),
 						Phase:           velerov2alpha1.DataUploadPhaseFailed,
+						ParentSnapshot:  "fake-parent-snapshot",
 					},
 				},
 			},
@@ -644,6 +648,7 @@ func TestCSISnapshots(t *testing.T) {
        Uploader Type: fake-uploader
        Moved data Size (bytes): 100
        Incremental data Size (bytes): 50
+        Parent Snapshot: fake-parent-snapshot
        Result: failed
 `,
 		},
@@ -136,6 +136,9 @@ func DescribeBackupSpecInSF(d *StructuredDescriber, spec velerov1api.BackupSpec)
 		s = spec.DataMover
 	}
 	backupSpecInfo["dataMover"] = s
+	if string(spec.BackupType) != "" {
+		backupSpecInfo["backupType"] = spec.BackupType
+	}

 	// describe TTL
 	backupSpecInfo["TTL"] = spec.TTL.Duration.String()
@@ -475,6 +478,9 @@ func describeDataMovementInSF(details bool, info *volume.BackupVolumeInfo, snaps
 		if info.SnapshotDataMovementInfo.IncrementalSize != nil {
 			dataMovement["incrementalSize"] = *info.SnapshotDataMovementInfo.IncrementalSize
 		}
+		if info.SnapshotDataMovementInfo.ParentSnapshot != "" {
+			dataMovement["parentSnapshot"] = info.SnapshotDataMovementInfo.ParentSnapshot
+		}

 		snapshotDetail["dataMovement"] = dataMovement
 	} else {
@@ -24,9 +24,11 @@ import (
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	corev1api "k8s.io/api/core/v1"
+	"k8s.io/utils/ptr"

 	"github.com/vmware-tanzu/velero/internal/volume"
 	velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
+	velerov2alpha1 "github.com/vmware-tanzu/velero/pkg/apis/velero/v2alpha1"
 	"github.com/vmware-tanzu/velero/pkg/builder"
 	"github.com/vmware-tanzu/velero/pkg/util/results"
 )
@@ -45,6 +47,7 @@ func TestDescribeBackupInSF(t *testing.T) {
 		TTL(72 * time.Hour).
 		CSISnapshotTimeout(10 * time.Minute).
 		DataMover("mover").
+		BackupType(velerov1api.BackupTypeFull).
 		Hooks(velerov1api.BackupHooks{
 			Resources: []velerov1api.BackupResourceHookSpec{
 				{
@@ -87,6 +90,7 @@ func TestDescribeBackupInSF(t *testing.T) {
 				"clusterScoped": "auto",
 			},
 			"dataMover":               "mover",
+			"backupType":              velerov1api.BackupTypeFull,
 			"labelSelector":           emptyDisplay,
 			"storageLocation":         "backup-location",
 			"veleroNativeSnapshotPVs": "auto",
@@ -517,7 +521,7 @@ func TestDescribeCSISnapshotsInSF(t *testing.T) {
 					PVCNamespace:      "pvc-ns-3",
 					PVCName:           "pvc-3",
 					SnapshotDataMoved: true,
-					SnapshotDataMovementInfo: &volume.SnapshotDataMovementInfo{
+					SnapshotDataMovementInfo: &volume.BackupSnapshotDataMovementInfo{
 						DataMover:      "velero",
 						UploaderType:   "fake-uploader",
 						SnapshotHandle: "fake-repo-id-3",
@@ -542,7 +546,7 @@ func TestDescribeCSISnapshotsInSF(t *testing.T) {
 					PVCName:           "pvc-4",
 					SnapshotDataMoved: true,
 					Result:            volume.VolumeResultSucceeded,
-					SnapshotDataMovementInfo: &volume.SnapshotDataMovementInfo{
+					SnapshotDataMovementInfo: &volume.BackupSnapshotDataMovementInfo{
 						DataMover:      "velero",
 						UploaderType:   "fake-uploader",
 						SnapshotHandle: "fake-repo-id-4",
@@ -573,10 +577,15 @@ func TestDescribeCSISnapshotsInSF(t *testing.T) {
 					Result:            volume.VolumeResultFailed,
 					PVCName:           "pvc-4",
 					SnapshotDataMoved: true,
-					SnapshotDataMovementInfo: &volume.SnapshotDataMovementInfo{
-						UploaderType:   "fake-uploader",
-						SnapshotHandle: "fake-repo-id-4",
-						OperationID:    "fake-operation-4",
+					BackupType:        velerov1api.BackupTypeIncremental,
+					SnapshotDataMovementInfo: &volume.BackupSnapshotDataMovementInfo{
+						UploaderType:    "fake-uploader",
+						SnapshotHandle:  "fake-repo-id-4",
+						OperationID:     "fake-operation-4",
+						Size:            100,
+						IncrementalSize: ptr.To(int64(50)),
+						Phase:           velerov2alpha1.DataUploadPhaseFailed,
+						ParentSnapshot:  "fake-parent-snapshot",
 					},
 				},
 			},
@@ -585,10 +594,13 @@ func TestDescribeCSISnapshotsInSF(t *testing.T) {
 				"csiSnapshots": map[string]any{
 					"pvc-ns-4/pvc-4": map[string]any{
 						"dataMovement": map[string]any{
-							"operationID":  "fake-operation-4",
-							"dataMover":    "velero",
-							"uploaderType": "fake-uploader",
-							"result":       "failed",
+							"operationID":     "fake-operation-4",
+							"dataMover":       "velero",
+							"uploaderType":    "fake-uploader",
+							"size":            int64(100),
+							"incrementalSize": int64(50),
+							"result":          "failed",
+							"parentSnapshot":  "fake-parent-snapshot",
 						},
 					},
 				},
@@ -209,6 +209,11 @@ func DescribeRestore(
 			s = string(restore.Spec.ExistingResourcePolicy)
 		}
 		d.Printf("Existing Resource Policy: \t%s\n", s)
+		s = emptyDisplay
+		if restore.Spec.ExistingVolumeDataPolicy != "" {
+			s = string(restore.Spec.ExistingVolumeDataPolicy)
+		}
+		d.Printf("Existing Volume Data Policy: \t%s\n", s)
 		d.Printf("ItemOperationTimeout:\t%s\n", restore.Spec.ItemOperationTimeout.Duration)

 		d.Println()
@@ -468,6 +473,15 @@ func describeCSISnapshotsRestores(d *Describer, restoreVolInfo []volume.RestoreV
 			d.Printf("\t\t\tOperation ID: %s\n", info.SnapshotDataMovementInfo.OperationID)
 			d.Printf("\t\t\tData Mover: %s\n", info.SnapshotDataMovementInfo.DataMover)
 			d.Printf("\t\t\tUploader Type: %s\n", info.SnapshotDataMovementInfo.UploaderType)
+			if info.SnapshotDataMovementInfo.RestoreType != "" {
+				d.Printf("\t\t\tRestore Type: %s\n", info.SnapshotDataMovementInfo.RestoreType)
+			}
+			if info.SnapshotDataMovementInfo.Size > 0 {
+				d.Printf("\t\t\tRestored data Size (bytes): %d\n", info.SnapshotDataMovementInfo.Size)
+			}
+			if info.SnapshotDataMovementInfo.IncrementalSize != nil {
+				d.Printf("\t\t\tIncremental data Size (bytes): %d\n", *info.SnapshotDataMovementInfo.IncrementalSize)
+			}
 		} else {
 			d.Printf("\t\tData Movement: specify --details for more information\n")
 		}
@@ -2,6 +2,7 @@ package output

 import (
 	"bytes"
+	"context"
 	"fmt"
 	"testing"
 	"text/tabwriter"
@@ -10,11 +11,13 @@ import (
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	corev1api "k8s.io/api/core/v1"
+	"k8s.io/utils/ptr"

 	"github.com/vmware-tanzu/velero/internal/volume"
 	velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
 	"github.com/vmware-tanzu/velero/pkg/builder"
 	"github.com/vmware-tanzu/velero/pkg/itemoperation"
+	velerotest "github.com/vmware-tanzu/velero/pkg/test"
 	"github.com/vmware-tanzu/velero/pkg/util/boolptr"
 	"github.com/vmware-tanzu/velero/pkg/util/results"
 )
@@ -309,11 +312,13 @@ CSI Snapshot Restores:
 					PVName:            "pv-3",
 					RestoreMethod:     volume.CSISnapshot,
 					SnapshotDataMoved: true,
-					SnapshotDataMovementInfo: &volume.SnapshotDataMovementInfo{
-						OperationID:  "op-3",
-						DataMover:    "velero",
-						UploaderType: "kopia",
-						Size:         1234,
+					SnapshotDataMovementInfo: &volume.RestoreSnapshotDataMovementInfo{
+						OperationID:     "op-3",
+						DataMover:       "velero",
+						UploaderType:    "kopia",
+						Size:            1234,
+						IncrementalSize: ptr.To(int64(500)),
+						RestoreType:     "Incremental",
 					},
 				},
 			},
@@ -325,6 +330,9 @@ CSI Snapshot Restores:
      Operation ID: op-3
      Data Mover: velero
      Uploader Type: kopia
+      Restore Type: Incremental
+      Restored data Size (bytes): 1234
+      Incremental data Size (bytes): 500
 `,
 		},
 		{
@@ -336,7 +344,7 @@ CSI Snapshot Restores:
 					PVName:            "pv-3",
 					RestoreMethod:     volume.CSISnapshot,
 					SnapshotDataMoved: true,
-					SnapshotDataMovementInfo: &volume.SnapshotDataMovementInfo{
+					SnapshotDataMovementInfo: &volume.RestoreSnapshotDataMovementInfo{
 						OperationID:  "op-3",
 						DataMover:    "velero",
 						UploaderType: "kopia",
@@ -415,3 +423,16 @@ func TestDescribeResourceModifier(t *testing.T) {
 	fmt.Println(d.buf.String())
 	require.Equal(t, expectOutput, d.buf.String())
 }
+
+func TestDescribeRestore(t *testing.T) {
+	kbClient := velerotest.NewFakeControllerRuntimeClient(t)
+	restore := builder.ForRestore("velero", "test-restore").
+		Backup("test-backup").
+		ExistingResourcePolicy(string(velerov1api.ResourcePolicyTypeUpdate)).
+		ExistingVolumeDataPolicy(string(velerov1api.VolumeDataPolicyTypeFull)).
+		Result()
+
+	out := DescribeRestore(context.Background(), kbClient, restore, nil, false, false, "")
+	assert.Contains(t, out, "Existing Resource Policy:      update")
+	assert.Contains(t, out, "Existing Volume Data Policy:   full")
+}
@@ -196,6 +196,13 @@ func describeRestoreSpecInSF(d *StructuredDescriber, spec velerov1api.RestoreSpe
 		specInfo["existingResourcePolicy"] = emptyDisplay
 	}

+	// existing volume data policy
+	if spec.ExistingVolumeDataPolicy != "" {
+		specInfo["existingVolumeDataPolicy"] = string(spec.ExistingVolumeDataPolicy)
+	} else {
+		specInfo["existingVolumeDataPolicy"] = emptyDisplay
+	}
+
 	specInfo["itemOperationTimeout"] = spec.ItemOperationTimeout.Duration.String()
 	specInfo["preserveNodePorts"] = BoolPointerString(spec.PreserveNodePorts, "false", "true", "auto")

@@ -360,12 +367,22 @@ func describeCSISnapshotsRestoresInSF(d *StructuredDescriber, restoreVolInfo []v
 				}
 				continue
 			}
+			dmInfo := map[string]any{
+				"operationID":  info.SnapshotDataMovementInfo.OperationID,
+				"dataMover":    info.SnapshotDataMovementInfo.DataMover,
+				"uploaderType": info.SnapshotDataMovementInfo.UploaderType,
+			}
+			if info.SnapshotDataMovementInfo.RestoreType != "" {
+				dmInfo["restoreType"] = info.SnapshotDataMovementInfo.RestoreType
+			}
+			if info.SnapshotDataMovementInfo.Size > 0 {
+				dmInfo["size"] = info.SnapshotDataMovementInfo.Size
+			}
+			if info.SnapshotDataMovementInfo.IncrementalSize != nil {
+				dmInfo["incrementalSize"] = *info.SnapshotDataMovementInfo.IncrementalSize
+			}
 			csiRestores[key] = map[string]any{
-				"dataMovement": map[string]any{
-					"operationID":  info.SnapshotDataMovementInfo.OperationID,
-					"dataMover":    info.SnapshotDataMovementInfo.DataMover,
-					"uploaderType": info.SnapshotDataMovementInfo.UploaderType,
-				},
+				"dataMovement": dmInfo,
 			}
 		} else {
 			csiRestores[key] = map[string]any{
@@ -28,6 +28,7 @@ import (
 	"github.com/stretchr/testify/require"
 	corev1api "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/utils/ptr"

 	"github.com/vmware-tanzu/velero/internal/volume"
 	velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
@@ -156,25 +157,27 @@ func TestDescribeRestoreSpecInSF(t *testing.T) {
 						"excluded":      emptyDisplay,
 						"clusterScoped": "auto",
 					},
-					"namespaceMappings":      emptyDisplay,
-					"labelSelector":          emptyDisplay,
-					"orLabelSelectors":       emptyDisplay,
-					"restorePVs":             "auto",
-					"existingResourcePolicy": emptyDisplay,
-					"itemOperationTimeout":   "0s",
-					"preserveNodePorts":      "auto",
+					"namespaceMappings":        emptyDisplay,
+					"labelSelector":            emptyDisplay,
+					"orLabelSelectors":         emptyDisplay,
+					"restorePVs":               "auto",
+					"existingResourcePolicy":   emptyDisplay,
+					"existingVolumeDataPolicy": emptyDisplay,
+					"itemOperationTimeout":     "0s",
+					"preserveNodePorts":        "auto",
 				},
 			},
 		},
 		{
 			name: "included namespaces wildcard treated as all",
 			spec: velerov1api.RestoreSpec{
-				BackupName:             "backup-2",
-				IncludedNamespaces:     []string{"*"},
-				ExcludedNamespaces:     []string{"kube-system"},
-				IncludedResources:      []string{"pods", "configmaps"},
-				ExcludedResources:      []string{"secrets"},
-				ExistingResourcePolicy: velerov1api.ResourcePolicyTypeUpdate,
+				BackupName:               "backup-2",
+				IncludedNamespaces:       []string{"*"},
+				ExcludedNamespaces:       []string{"kube-system"},
+				IncludedResources:        []string{"pods", "configmaps"},
+				ExcludedResources:        []string{"secrets"},
+				ExistingResourcePolicy:   velerov1api.ResourcePolicyTypeUpdate,
+				ExistingVolumeDataPolicy: velerov1api.VolumeDataPolicyTypeFull,
 			},
 			expect: map[string]any{
 				"spec": map[string]any{
@@ -188,13 +191,14 @@ func TestDescribeRestoreSpecInSF(t *testing.T) {
 						"excluded":      "secrets",
 						"clusterScoped": "auto",
 					},
-					"namespaceMappings":      emptyDisplay,
-					"labelSelector":          emptyDisplay,
-					"orLabelSelectors":       emptyDisplay,
-					"restorePVs":             "auto",
-					"existingResourcePolicy": string(velerov1api.ResourcePolicyTypeUpdate),
-					"itemOperationTimeout":   "0s",
-					"preserveNodePorts":      "auto",
+					"namespaceMappings":        emptyDisplay,
+					"labelSelector":            emptyDisplay,
+					"orLabelSelectors":         emptyDisplay,
+					"restorePVs":               "auto",
+					"existingResourcePolicy":   string(velerov1api.ResourcePolicyTypeUpdate),
+					"existingVolumeDataPolicy": string(velerov1api.VolumeDataPolicyTypeFull),
+					"itemOperationTimeout":     "0s",
+					"preserveNodePorts":        "auto",
 				},
 			},
 		},
@@ -223,13 +227,14 @@ func TestDescribeRestoreSpecInSF(t *testing.T) {
 						"excluded":      emptyDisplay,
 						"clusterScoped": "auto",
 					},
-					"namespaceMappings":      emptyDisplay,
-					"labelSelector":          emptyDisplay,
-					"orLabelSelectors":       emptyDisplay,
-					"restorePVs":             "auto",
-					"existingResourcePolicy": emptyDisplay,
-					"itemOperationTimeout":   "0s",
-					"preserveNodePorts":      "auto",
+					"namespaceMappings":        emptyDisplay,
+					"labelSelector":            emptyDisplay,
+					"orLabelSelectors":         emptyDisplay,
+					"restorePVs":               "auto",
+					"existingResourcePolicy":   emptyDisplay,
+					"existingVolumeDataPolicy": emptyDisplay,
+					"itemOperationTimeout":     "0s",
+					"preserveNodePorts":        "auto",
 					"resourceModifier": map[string]any{
 						"type": "ConfigMap",
 						"name": "my-modifier",
@@ -272,13 +277,14 @@ func TestDescribeRestoreSpecInSF(t *testing.T) {
 						"excluded":      emptyDisplay,
 						"clusterScoped": "included",
 					},
-					"namespaceMappings":      map[string]string{"ns-a": "ns-a-new"},
-					"labelSelector":          "app=nginx",
-					"orLabelSelectors":       "env=prod or env=stage",
-					"restorePVs":             "true",
-					"existingResourcePolicy": emptyDisplay,
-					"itemOperationTimeout":   "0s",
-					"preserveNodePorts":      "false",
+					"namespaceMappings":        map[string]string{"ns-a": "ns-a-new"},
+					"labelSelector":            "app=nginx",
+					"orLabelSelectors":         "env=prod or env=stage",
+					"restorePVs":               "true",
+					"existingResourcePolicy":   emptyDisplay,
+					"existingVolumeDataPolicy": emptyDisplay,
+					"itemOperationTimeout":     "0s",
+					"preserveNodePorts":        "false",
 					"resourcePolicy": map[string]any{
 						"type": "configmap",
 						"name": "volume-policy",
@@ -467,10 +473,13 @@ func TestDescribeRestoreCSISnapshotsInSF_NoData(t *testing.T) {
 					SnapshotDataMoved: true,
 					PVCName:           "pvc-3",
 					PVCNamespace:      "ns-3",
-					SnapshotDataMovementInfo: &volume.SnapshotDataMovementInfo{
-						OperationID:  "op-3",
-						DataMover:    "velero",
-						UploaderType: "kopia",
+					SnapshotDataMovementInfo: &volume.RestoreSnapshotDataMovementInfo{
+						OperationID:     "op-3",
+						DataMover:       "velero",
+						UploaderType:    "kopia",
+						Size:            1234,
+						IncrementalSize: ptr.To(int64(500)),
+						RestoreType:     "Incremental",
 					},
 				},
 			},
@@ -479,9 +488,12 @@ func TestDescribeRestoreCSISnapshotsInSF_NoData(t *testing.T) {
 				"csiSnapshotRestores": map[string]any{
 					"ns-3/pvc-3": map[string]any{
 						"dataMovement": map[string]any{
-							"operationID":  "op-3",
-							"dataMover":    "velero",
-							"uploaderType": "kopia",
+							"operationID":     "op-3",
+							"dataMover":       "velero",
+							"uploaderType":    "kopia",
+							"size":            int64(1234),
+							"incrementalSize": int64(500),
+							"restoreType":     "Incremental",
 						},
 					},
 				},
@@ -495,7 +507,7 @@ func TestDescribeRestoreCSISnapshotsInSF_NoData(t *testing.T) {
 					SnapshotDataMoved: true,
 					PVCName:           "pvc-3",
 					PVCNamespace:      "ns-3",
-					SnapshotDataMovementInfo: &volume.SnapshotDataMovementInfo{
+					SnapshotDataMovementInfo: &volume.RestoreSnapshotDataMovementInfo{
 						OperationID:  "op-3",
 						DataMover:    "velero",
 						UploaderType: "kopia",
@@ -395,10 +395,11 @@ func (b *backupReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctr

 func (b *backupReconciler) prepareBackupRequest(ctx context.Context, backup *velerov1api.Backup, logger logrus.FieldLogger) *pkgbackup.Request {
 	request := &pkgbackup.Request{
-		Backup:           backup.DeepCopy(), // don't modify items in the cache
-		SkippedPVTracker: pkgbackup.NewSkipPVTracker(),
-		BackedUpItems:    pkgbackup.NewBackedUpItemsMap(),
-		WorkerPool:       pkgbackup.StartItemBlockWorkerPool(ctx, b.itemBlockWorkerCount, logger),
+		Backup:                        backup.DeepCopy(), // don't modify items in the cache
+		SkippedPVTracker:              pkgbackup.NewSkipPVTracker(),
+		BackedUpItems:                 pkgbackup.NewBackedUpItemsMap(),
+		MustIncludeAdditionalItemPVCs: pkgbackup.NewBackedUpItemsMap(),
+		WorkerPool:                    pkgbackup.StartItemBlockWorkerPool(ctx, b.itemBlockWorkerCount, logger),
 	}
 	request.VolumesInformation.Init()

@@ -158,10 +158,11 @@ func (r *backupFinalizerReconciler) Reconcile(ctx context.Context, req ctrl.Requ
 	}

 	backupRequest := &pkgbackup.Request{
-		Backup:           backup,
-		StorageLocation:  location,
-		SkippedPVTracker: pkgbackup.NewSkipPVTracker(),
-		BackedUpItems:    pkgbackup.NewBackedUpItemsMap(),
+		Backup:                        backup,
+		StorageLocation:               location,
+		SkippedPVTracker:              pkgbackup.NewSkipPVTracker(),
+		BackedUpItems:                 pkgbackup.NewBackedUpItemsMap(),
+		MustIncludeAdditionalItemPVCs: pkgbackup.NewBackedUpItemsMap(),
 	}
 	var outBackupFile *os.File
 	if len(operations) > 0 {
@@ -496,6 +496,7 @@ func (e *csiSnapshotExposer) CleanUp(ctx context.Context, ownerObject corev1api.
 	backupPodName := ownerObject.Name
 	backupPVCName := ownerObject.Name
 	backupVSName := ownerObject.Name
+	backupVSCName := ownerObject.Name

 	kube.DeletePodIfAny(ctx, e.kubeClient.CoreV1(), backupPodName, ownerObject.Namespace, e.log)
 	kube.DeletePVAndPVCIfAny(ctx, e.kubeClient.CoreV1(), backupPVCName, ownerObject.Namespace, cleanUpTimeout, e.log)
@@ -507,6 +508,13 @@ func (e *csiSnapshotExposer) CleanUp(ctx context.Context, ownerObject corev1api.

 	csi.DeleteVolumeSnapshotIfAny(ctx, e.csiSnapshotClient, backupVSName, ownerObject.Namespace, e.log)
 	csi.DeleteVolumeSnapshotIfAny(ctx, e.csiSnapshotClient, vsName, sourceNamespace, e.log)
+
+	// The backup VSC is created by Velero as an internal handle to the source
+	// snapshot. Deleting the backup VS above only cascades to it when its
+	// deletion policy is Delete, so remove it explicitly to avoid leaking the
+	// object under a Retain policy. Deleting a Retain VSC drops only the API
+	// object and leaves the underlying snapshot intact.
+	csi.DeleteVolumeSnapshotContentIfAny(ctx, e.csiSnapshotClient, backupVSCName, e.log)
 }

 func getVolumeModeByAccessMode(accessMode string, dataMover string) (corev1api.PersistentVolumeMode, error) {
@@ -571,7 +579,21 @@ func (e *csiSnapshotExposer) createBackupVSC(ctx context.Context, ownerObject co
 			Source: snapshotv1api.VolumeSnapshotContentSource{
 				SnapshotHandle: snapshotVSC.Status.SnapshotHandle,
 			},
-			DeletionPolicy:          snapshotv1api.VolumeSnapshotContentDelete,
+			// The backup VSC is statically provisioned against the same
+			// snapshot handle as the source VSC, so both objects refer to one
+			// physical snapshot. Inherit the source's deletion policy instead
+			// of forcing Delete, otherwise a user who configured Retain on the
+			// VolumeSnapshotClass still loses the snapshot when the backup VSC
+			// is cleaned up.
+			//
+			// For Case 2 storages per the design (design/block-data-mover/block-data-mover.md,
+			// e.g. Ceph RBD), inheriting Retain is not just an option but a requirement for
+			// incrementals to work at all: rbd snap diff needs the base and target snapshots
+			// in the same clone chain, so Delete destroys the base as soon as this backup
+			// completes. The next incremental's delta query then fails and degrades to an
+			// allocated-blocks backup (see the CBT tier ladder) or, without that fix, a full
+			// whole-device transfer.
+			DeletionPolicy:          snapshotVSC.Spec.DeletionPolicy,
 			Driver:                  snapshotVSC.Spec.Driver,
 			VolumeSnapshotClassName: snapshotVSC.Spec.VolumeSnapshotClassName,
 		},
@@ -736,7 +758,7 @@ func (e *csiSnapshotExposer) createBackupPod(

 	if csiSnapshotMetadataServiceConfigs != nil {
 		if csiSnapshotMetadataServiceConfigs.SAName != "" {
-			args = append(args, fmt.Sprintf("--csi-snapshot-metadata-service-sa=%s", csiSnapshotMetadataServiceConfigs.SAName))
+			args = append(args, fmt.Sprintf("--cbt-sa-name=%s", csiSnapshotMetadataServiceConfigs.SAName))
 		}
 	}

@@ -43,6 +43,7 @@ import (
 	clientFake "sigs.k8s.io/controller-runtime/pkg/client/fake"

 	velerov1 "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
+	datamovercli "github.com/vmware-tanzu/velero/pkg/cmd/cli/datamover"
 	velerotest "github.com/vmware-tanzu/velero/pkg/test"
 	velerotypes "github.com/vmware-tanzu/velero/pkg/types"
 	"github.com/vmware-tanzu/velero/pkg/util"
@@ -2537,3 +2538,91 @@ func TestCleanUp_SecretsAndConfigMaps(t *testing.T) {
 	_, err = fakeKubeClient.CoreV1().Secrets("velero").Get(t.Context(), "other-secret", metav1.GetOptions{})
 	assert.NoError(t, err, "unrelated secret should not be deleted")
 }
+
+// TestBackupPodCBTServiceSAFlagMatchesDatamoverBackupFlags pins the contract between the
+// flag createBackupPod emits for the CSI snapshot metadata service's service account and
+// the flag NewBackupCommand actually registers to consume it. These previously drifted
+// (exposer emitted --csi-snapshot-metadata-service-sa, the datamover backup command only
+// registered --cbt-sa-name), so cobra rejected the unknown flag and the data mover pod
+// exited immediately whenever a dedicated CBT service account was configured. This test
+// fails if either side changes the flag name without the other.
+func TestBackupPodCBTServiceSAFlagMatchesDatamoverBackupFlags(t *testing.T) {
+	const saName = "cbt-service-account"
+
+	// The exact line in createBackupPod (pkg/exposer/csi_snapshot.go) that builds this arg:
+	//   args = append(args, fmt.Sprintf("--cbt-sa-name=%s", csiSnapshotMetadataServiceConfigs.SAName))
+	arg := fmt.Sprintf("--cbt-sa-name=%s", saName)
+
+	cmd := datamovercli.NewBackupCommand(nil)
+	err := cmd.ParseFlags([]string{
+		"--volume-path=/dev/vol",
+		"--volume-mode=Filesystem",
+		"--data-upload=du-test",
+		"--resource-timeout=1m",
+		arg,
+	})
+	require.NoError(t, err, "datamover backup command must accept the flag the exposer emits")
+
+	got, err := cmd.Flags().GetString("cbt-sa-name")
+	require.NoError(t, err)
+	assert.Equal(t, saName, got)
+}
+
+func TestCreateBackupVSCDeletionPolicy(t *testing.T) {
+	tests := []struct {
+		name           string
+		sourcePolicy   snapshotv1api.DeletionPolicy
+		expectedPolicy snapshotv1api.DeletionPolicy
+	}{
+		{
+			name:           "Delete policy is inherited",
+			sourcePolicy:   snapshotv1api.VolumeSnapshotContentDelete,
+			expectedPolicy: snapshotv1api.VolumeSnapshotContentDelete,
+		},
+		{
+			// The backup VSC points at the same snapshot handle as the source
+			// VSC, so forcing Delete here would destroy a snapshot the user
+			// asked to keep.
+			name:           "Retain policy is inherited",
+			sourcePolicy:   snapshotv1api.VolumeSnapshotContentRetain,
+			expectedPolicy: snapshotv1api.VolumeSnapshotContentRetain,
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			handle := "fake-snapshot-handle"
+			className := "fake-snapshot-class"
+
+			sourceVSC := &snapshotv1api.VolumeSnapshotContent{
+				ObjectMeta: metav1.ObjectMeta{Name: "source-vsc"},
+				Spec: snapshotv1api.VolumeSnapshotContentSpec{
+					DeletionPolicy:          test.sourcePolicy,
+					Driver:                  "fake-driver",
+					VolumeSnapshotClassName: &className,
+				},
+				Status: &snapshotv1api.VolumeSnapshotContentStatus{
+					SnapshotHandle: &handle,
+				},
+			}
+
+			exposer := csiSnapshotExposer{
+				csiSnapshotClient: snapshotFake.NewSimpleClientset().SnapshotV1(),
+				log:               velerotest.NewLogger(),
+			}
+
+			ownerObject := corev1api.ObjectReference{
+				Name:      "fake-du",
+				Namespace: "velero",
+			}
+			vs := &snapshotv1api.VolumeSnapshot{
+				ObjectMeta: metav1.ObjectMeta{Name: "fake-du", Namespace: "velero"},
+			}
+
+			backupVSC, err := exposer.createBackupVSC(t.Context(), ownerObject, sourceVSC, vs)
+			require.NoError(t, err)
+			assert.Equal(t, test.expectedPolicy, backupVSC.Spec.DeletionPolicy)
+			assert.Equal(t, handle, *backupVSC.Spec.Source.SnapshotHandle)
+		})
+	}
+}
@@ -93,6 +93,7 @@ func ShouldPerformSnapshotWithVolumeHelper(
 		crClient,
 		boolptr.IsSetToTrue(backup.Spec.DefaultVolumesToFsBackup),
 		true,
+		nil,
 	)

 	return volumeHelperImpl.ShouldPerformSnapshot(unstructured, groupResource)
@@ -111,6 +112,7 @@ func NewVolumeHelperWithNamespaces(
 	defaultVolumesToFSBackup bool,
 	backupExcludePVC bool,
 	namespaces []string,
+	pvcMustInclusionTracker vhutil.PVCMustInclusionTracker,
 ) (vhutil.VolumeHelper, error) {
 	return volumehelper.NewVolumeHelperImplWithNamespaces(
 		volumePolicy,
@@ -120,6 +122,7 @@ func NewVolumeHelperWithNamespaces(
 		defaultVolumesToFSBackup,
 		backupExcludePVC,
 		namespaces,
+		pvcMustInclusionTracker,
 	)
 }

@@ -131,11 +134,13 @@ func NewVolumeHelperWithCache(
 	client crclient.Client,
 	logger logrus.FieldLogger,
 	pvcPodCache *podvolumeutil.PVCPodCache,
+	pvcMustInclusionTracker vhutil.PVCMustInclusionTracker,
 ) (vhutil.VolumeHelper, error) {
 	return volumehelper.NewVolumeHelperImplWithCache(
 		backup,
 		client,
 		logger,
 		pvcPodCache,
+		pvcMustInclusionTracker,
 	)
 }
@@ -300,6 +300,7 @@ func TestShouldPerformSnapshotWithNonNilVolumeHelper(t *testing.T) {
 		false, // defaultVolumesToFSBackup
 		true,  // backupExcludePVC
 		[]string{"default"},
+		nil,
 	)
 	require.NoError(t, err)
 	require.NotNil(t, vh)
@@ -712,6 +712,7 @@ func (kr *kopiaRepository) GetSnapshot(ctx context.Context, id udmrepo.ID) (udmr
 	}

 	return udmrepo.Snapshot{
+		ID:          udmrepo.ID(snap.ID),
 		Source:      snap.Source.Path,
 		Description: snap.Description,
 		StartTime:   snap.StartTime.ToTime(),
@@ -751,6 +752,7 @@ func (kr *kopiaRepository) ListSnapshot(ctx context.Context, source string) ([]u
 	snapshots := []udmrepo.Snapshot{}
 	for _, snap := range mani {
 		snapshots = append(snapshots, udmrepo.Snapshot{
+			ID:          udmrepo.ID(snap.ID),
 			Source:      snap.Source.Path,
 			Description: snap.Description,
 			StartTime:   snap.StartTime.ToTime(),
@@ -1609,6 +1609,7 @@ func TestGetSnapshot(t *testing.T) {
 			snapshotID:  udmrepo.ID("fake-id"),
 			setRepoMock: true,
 			expectedSnap: udmrepo.Snapshot{
+				ID:          "fake-id",
 				Source:      "fake-source",
 				Description: "fake-desc",
 				StartTime:   mockMani.StartTime.ToTime(),
@@ -1805,6 +1806,7 @@ func TestListSnapshot(t *testing.T) {
 			setRepoMock: true,
 			expectedSnaps: []udmrepo.Snapshot{
 				{
+					ID:          "fake-id",
 					Source:      "fake-source",
 					Description: "fake-desc",
 					StartTime:   mockMani.StartTime.ToTime(),
@@ -98,6 +98,7 @@ type Metadata struct {
 }

 type Snapshot struct {
+	ID          ID
 	Source      string
 	Description string
 	StartTime   time.Time
@@ -118,7 +118,7 @@ func TestRestorePVWithVolumeInfo(t *testing.T) {
 				"pv-1": {
 					BackupMethod: volume.PodVolumeBackup,
 					PVName:       "pv-1",
-					PVBInfo: &volume.PodVolumeInfo{
+					PVBInfo: &volume.PodVolumeBackupInfo{
 						SnapshotHandle: "testSnapshotHandle",
 						Size:           100,
 						NodeName:       "testNode",
@@ -173,7 +173,7 @@ func TestRestorePVWithVolumeInfo(t *testing.T) {
 					CSISnapshotInfo: &volume.CSISnapshotInfo{
 						Driver: "pd.csi.storage.gke.io",
 					},
-					SnapshotDataMovementInfo: &volume.SnapshotDataMovementInfo{
+					SnapshotDataMovementInfo: &volume.BackupSnapshotDataMovementInfo{
 						DataMover: "velero",
 					},
 				},
@@ -151,11 +151,6 @@ func snapshotSource(
 func getParentBackupInfo(ctx context.Context, rep udmrepo.BackupRepo, forceFull bool, parentSnapshot string, volumeID string, realSource string, snapshotTags map[string]string, log logrus.FieldLogger) parentBackupInfo {
 	var previous *udmrepo.Snapshot

-	// parentID names whichever snapshot ended up being the parent. On the discovery
-	// branch the parentSnapshot parameter is empty by definition, so logging it there
-	// produces messages that describe a decision without naming the object it was about.
-	parentID := parentSnapshot
-
 	if !forceFull {
 		if parentSnapshot != "" {
 			snap, err := rep.GetSnapshot(ctx, udmrepo.ID(parentSnapshot))
@@ -173,8 +168,7 @@ func getParentBackupInfo(ctx context.Context, rep udmrepo.BackupRepo, forceFull
 				log.WithError(err).Warn("Failed to search previous snapshot, fallback to full backup")
 			} else {
 				previous = &snap
-				parentID = string(snap.RootObject.ID)
-				log.Infof("Using previous snapshot %s", snap.RootObject.ID)
+				log.Infof("Using previous snapshot %s", snap.ID)
 			}
 		}
 	} else {
@@ -184,21 +178,21 @@ func getParentBackupInfo(ctx context.Context, rep udmrepo.BackupRepo, forceFull
 	parentInfo := parentBackupInfo{}
 	if previous != nil {
 		if previous.Tags == nil {
-			log.Warnf("No tag from parent snapshot %s, fallback to full backup", parentID)
+			log.Warnf("No tag from parent snapshot %s, fallback to full backup", previous.ID)
 		} else if previous.Tags[uploader.CBTChangeIDTag] == "" {
-			log.Warnf("No ChangeID tag from parent snapshot %s, fallback to full backup", parentID)
+			log.Warnf("No ChangeID tag from parent snapshot %s, fallback to full backup", previous.ID)
 		} else if previous.Tags[uploader.CBTVolumeIDTag] == "" {
-			log.Warnf("No VolumeID tag from parent snapshot %s, fallback to full backup", parentID)
+			log.Warnf("No VolumeID tag from parent snapshot %s, fallback to full backup", previous.ID)
 		} else if previous.Tags[uploader.CBTVolumeIDTag] != volumeID {
-			log.Warnf("VolumeID %s from parent snapshot %s is not expected as %s, fallback to full backup", previous.Tags[uploader.CBTVolumeIDTag], parentID, volumeID)
+			log.Warnf("VolumeID %s from parent snapshot %s is not expected as %s, fallback to full backup", previous.Tags[uploader.CBTVolumeIDTag], previous.ID, volumeID)
 		} else if obj, err := loadObjectFromSnapshot(ctx, rep, previous); err != nil {
-			log.WithError(err).Warnf("Failed to load object from parent snapshot %s, fallback to full backup", parentID)
+			log.WithError(err).Warnf("Failed to load object from parent snapshot %s, fallback to full backup", previous.ID)
 		} else {
 			parentInfo.parentObject = obj
 			parentInfo.changeID = previous.Tags[uploader.CBTChangeIDTag]
 			parentInfo.volumeID = previous.Tags[uploader.CBTVolumeIDTag]

-			log.Infof("Using parent snapshot %s, start time %v, end time %v, description %s", parentID, previous.StartTime, previous.EndTime, previous.Description)
+			log.Infof("Using parent snapshot %s, start time %v, end time %v, description %s", previous.ID, previous.StartTime, previous.EndTime, previous.Description)
 		}
 	}

@@ -351,6 +351,7 @@ func TestSnapshotSource(t *testing.T) {
 func TestGetParentBackupInfoLogsDiscoveredParentID(t *testing.T) {
 	const volumeID = "vol-123"
 	const realSource = "/test/source"
+	const parentSnapID = "snap-parent-42"
 	const rootObj = "root-obj-42"

 	snapshotTags := map[string]string{
@@ -364,6 +365,7 @@ func TestGetParentBackupInfoLogsDiscoveredParentID(t *testing.T) {
 	repo := udmrepomocks.NewBackupRepo(t)
 	repo.On("ListSnapshot", mock.Anything, realSource).
 		Return([]udmrepo.Snapshot{{
+			ID:         parentSnapID,
 			RootObject: udmrepo.ObjectMetadata{ID: rootObj},
 			Tags: map[string]string{
 				uploader.CBTChangeIDTag:       "cid-abc",
@@ -389,7 +391,7 @@ func TestGetParentBackupInfoLogsDiscoveredParentID(t *testing.T) {
 	for _, entry := range hook.AllEntries() {
 		if strings.HasPrefix(entry.Message, "Using parent snapshot ") {
 			found = true
-			assert.Contains(t, entry.Message, rootObj,
+			assert.Contains(t, entry.Message, parentSnapID,
 				"parent-selection message must name the discovered snapshot, got %q", entry.Message)
 		}
 	}
@@ -155,17 +155,17 @@ func (blkup *blockUploader) Restore(snapshot udmrepo.Snapshot, dest destInfo, bi

 	meta, err := blkup.repoWriter.ReadMetadata(blkup.ctx, snapshot.RootObject.ID)
 	if err != nil {
-		return 0, 0, errors.Wrapf(err, "error reading snapshot metadata for %s", snapshot.Description)
+		return 0, 0, errors.Wrapf(err, "error reading snapshot metadata for %s", snapshot.ID)
 	}

 	if len(meta.SubObjects) != 1 {
-		return 0, 0, errors.Errorf("unexpected number of bdev object (%d) for snapshot %s", len(meta.SubObjects), snapshot.Description)
+		return 0, 0, errors.Errorf("unexpected number of bdev object (%d) for snapshot %s", len(meta.SubObjects), snapshot.ID)
 	}

 	sourceSize, err := getSourceSize(snapshot)
 	if err != nil {
 		sourceSize = meta.SubObjects[0].Size
-		blkup.log.Warnf("Failed to get source size from snapshot %s, use backup size %v", snapshot.Description, sourceSize)
+		blkup.log.Warnf("Failed to get source size from snapshot %s, use backup size %v", snapshot.ID, sourceSize)
 	}

 	if sourceSize > meta.SubObjects[0].Size {
@@ -655,11 +655,11 @@ func loadObjectFromSnapshot(ctx context.Context, rep udmrepo.BackupRepo, snapsho

 	meta, err := rep.ReadMetadata(ctx, snapshot.RootObject.ID)
 	if err != nil {
-		return "", errors.Wrapf(err, "error reading snapshot metadata for %s", snapshot.Description)
+		return "", errors.Wrap(err, "error reading snapshot metadata")
 	}

 	if len(meta.SubObjects) != 1 {
-		return "", errors.Errorf("unexpected number of bdev object (%d) for snapshot %s", len(meta.SubObjects), snapshot.Description)
+		return "", errors.Errorf("unexpected number of bdev object (%d)", len(meta.SubObjects))
 	}

 	return meta.SubObjects[0].ID, nil
@@ -64,6 +64,11 @@ func GetCBTInfo(ctx context.Context, kubeClient kubernetes.Interface, log logrus

 		if vsc.Status != nil && vsc.Status.SnapshotHandle != nil {
 			cbtInfo.ChangeID = *vsc.Status.SnapshotHandle
+		} else if vsc.Spec.Source.SnapshotHandle != nil {
+			// The backup VSC is statically provisioned from the source VSC's
+			// snapshot handle; its status is populated asynchronously and may
+			// not be set yet, but the handle is already in the spec.
+			cbtInfo.ChangeID = *vsc.Spec.Source.SnapshotHandle
 		}

 		if pv.Spec.CSI != nil && pv.Spec.CSI.VolumeHandle != "" {
@@ -22,6 +22,13 @@ import (
 	"k8s.io/apimachinery/pkg/runtime/schema"
 )

+// PVCMustInclusionTracker provides read-only checks for whether a PVC is included
+// in the backup as BIA's additionalItems through annotation
+// backup.velero.io/must-include-additional-items.
+type PVCMustInclusionTracker interface {
+	IsPVCIncluded(namespace, pvcName string) bool
+}
+
 type VolumeHelper interface {
 	ShouldPerformSnapshot(obj runtime.Unstructured, groupResource schema.GroupResource) (bool, error)
 	ShouldPerformFSBackup(volume corev1api.Volume, pod corev1api.Pod) (bool, error)
@@ -65,6 +65,23 @@ order in which item action plugins are invoked. However, if a single binary impl
 they may be invoked in the order in which they are registered but it is best to not depend on this
 implementation. This is not guaranteed officially and the implementation can change at any time.

+### Must-include additional items (Backup Item Actions)
+
+Backup Item Actions may return `AdditionalItems` that Velero backs up as dependencies of the current item.
+By default those additional items must still pass the backup's global resource and namespace include/exclude filters.
+
+To force-backup hard dependencies despite those filters, set the following annotation on the `UpdatedItem` returned from `Execute()`:
+
+```
+backup.velero.io/must-include-additional-items: "true"
+```
+
+Behavior:
+- Only the string value `"true"` enables the bypass.
+- The annotation applies blanket to all `AdditionalItems` from that BIA invocation (not per-item).
+- Velero strips the annotation before saving the item to the backup tarball.
+- **Important Note for File System Backup (FSB):** If your plugin returns both a Pod and its associated PersistentVolumeClaims (PVCs) as `AdditionalItems`, and you expect Velero to create PodVolumeBackups (PVBs) for those PVCs using File System Backup, using this annotation ensures Velero correctly evaluates the PVCs for FSB. Velero explicitly tracks PVCs returned as additional items with this annotation, guaranteeing that PVBs are created even if the PVCs are excluded by global or fine-grained backup filters, regardless of the order they are returned in the `AdditionalItems` slice.
+
 ### Must-include additional items (Restore Item Actions)

 Restore Item Actions may return `AdditionalItems` that Velero restores as dependencies of the current item.
				`@@ -0,0 +1 @@`
				`Fix generic CSI changeID retrieval and honor snapshot class deletion policy for CBT retention`
				`@@ -0,0 +1 @@`
				`Fix datamover backup pod arg mismatch for CSI CBT service account name`
				`@@ -0,0 +1 @@`
				`Add block data mover support for Velero backup/restore describe CLI.`
				`@@ -0,0 +1 @@`
				`Add ID to repo snapshot and modify block uploader to use snapshot ID in logs`
				`@@ -0,0 +1 @@`
				`Add MustIncludeAdditionalItemPVCs to help track BIA added PVC's PVB creation.`