From 0e2fdfc75085c3da3d5f7abc64284064cd564d4b Mon Sep 17 00:00:00 2001
From: wanglong <46259301+gnolong@users.noreply.github.com>
Date: Fri, 29 Nov 2024 11:29:15 +0800
Subject: [PATCH] feat: support backup and restore parameters (#8472)

(cherry picked from commit a68fb12f9c28c353717484a8b5c4ac5d1a71cb07)
---
 .../v1alpha1/actionset_types.go               |  15 ++
 apis/dataprotection/v1alpha1/backup_types.go  |  11 +
 .../v1alpha1/backupschedule_types.go          |  26 +++
 apis/dataprotection/v1alpha1/restore_types.go |  11 +
 apis/dataprotection/v1alpha1/types.go         |  26 +++
 .../v1alpha1/zz_generated.deepcopy.go         |  66 +++++-
 apis/operations/v1alpha1/opsrequest_types.go  |  24 ++
 .../v1alpha1/zz_generated.deepcopy.go         |  13 +-
 ...taprotection.kubeblocks.io_actionsets.yaml |  25 ++
 ...n.kubeblocks.io_backuppolicytemplates.yaml |  26 +++
 .../dataprotection.kubeblocks.io_backups.yaml |  27 +++
 ...tection.kubeblocks.io_backupschedules.yaml |  26 +++
 ...dataprotection.kubeblocks.io_restores.yaml |  27 +++
 .../operations.kubeblocks.io_opsrequests.yaml |  54 +++++
 .../apps/transformer_cluster_backup_policy.go |  19 +-
 .../dataprotection/actionset_controller.go    |  45 +++-
 .../actionset_controller_test.go              |  49 +++-
 .../dataprotection/backup_controller.go       |   6 +-
 .../dataprotection/backup_controller_test.go  |  55 ++++-
 .../backuppolicytemplate_controller.go        |  24 +-
 .../backuppolicytemplate_controller_test.go   |  38 +++-
 .../backupschedule_controller_test.go         |  65 +++++-
 .../dataprotection/restore_controller_test.go |  63 ++++-
 ...taprotection.kubeblocks.io_actionsets.yaml |  25 ++
 ...n.kubeblocks.io_backuppolicytemplates.yaml |  26 +++
 .../dataprotection.kubeblocks.io_backups.yaml |  27 +++
 ...tection.kubeblocks.io_backupschedules.yaml |  26 +++
 ...dataprotection.kubeblocks.io_restores.yaml |  27 +++
 .../operations.kubeblocks.io_opsrequests.yaml |  54 +++++
 docs/developer_docs/api-reference/backup.md   | 215 ++++++++++++++++++
 pkg/common/openapiv3schema.go                 |   2 +-
 pkg/constant/backup.go                        |   1 +
 pkg/controller/builder/builder_component.go   |   3 +-
 pkg/controller/plan/restore.go                |   8 +
 pkg/dataprotection/backup/request.go          |   1 +
 pkg/dataprotection/backup/scheduler.go        |  58 +++--
 pkg/dataprotection/backup/utils.go            |  26 ++-
 pkg/dataprotection/restore/builder.go         |   4 +
 pkg/dataprotection/restore/utils.go           |  24 +-
 pkg/dataprotection/utils/backup.go            |  18 ++
 pkg/dataprotection/utils/envvar.go            |  11 +-
 pkg/dataprotection/utils/utils.go             |  46 ++++
 pkg/operations/backup.go                      |   1 +
 pkg/operations/custom.go                      |   2 +-
 pkg/operations/restore.go                     |   3 +-
 pkg/testutil/dataprotection/backup_utils.go   |   5 +-
 .../backuppolicytemplate_factory.go           |   9 +-
 pkg/testutil/dataprotection/constant.go       |  29 +++
 .../dataprotection/restore_factory.go         |   5 +
 pkg/testutil/dataprotection/utils.go          |  38 +++-
 50 files changed, 1373 insertions(+), 62 deletions(-)

diff --git a/apis/dataprotection/v1alpha1/actionset_types.go b/apis/dataprotection/v1alpha1/actionset_types.go
index f693f38c715..42cd81a6bf5 100644
--- a/apis/dataprotection/v1alpha1/actionset_types.go
+++ b/apis/dataprotection/v1alpha1/actionset_types.go
@@ -37,6 +37,11 @@ type ActionSetSpec struct {
 	// +kubebuilder:validation:Required
 	BackupType BackupType `json:"backupType"`
 
+	// Specifies the schema of parameters in backups and restores before their usage.
+	//
+	// +optional
+	ParametersSchema *ActionSetParametersSchema `json:"parametersSchema,omitempty"`
+
 	// Specifies a list of environment variables to be set in the container.
 	//
 	// +kubebuilder:pruning:PreserveUnknownFields
@@ -117,6 +122,11 @@ type BackupActionSpec struct {
 	//
 	// +optional
 	PreDeleteBackup *BaseJobActionSpec `json:"preDelete,omitempty"`
+
+	// Specifies the parameters used by the backup action
+	//
+	// +optional
+	WithParameters []string `json:"withParameters,omitempty"`
 }
 
 // BackupDataActionSpec defines how to back up data.
@@ -162,6 +172,11 @@ type RestoreActionSpec struct {
 	// +optional
 	// +kubebuilder:default=true
 	BaseBackupRequired *bool `json:"baseBackupRequired,omitempty"`
+
+	// Specifies the parameters used by the restore action
+	//
+	// +optional
+	WithParameters []string `json:"withParameters,omitempty"`
 }
 
 // ActionSpec defines an action that should be executed. Only one of the fields may be set.
diff --git a/apis/dataprotection/v1alpha1/backup_types.go b/apis/dataprotection/v1alpha1/backup_types.go
index 584ca1f8f04..25f93afe762 100644
--- a/apis/dataprotection/v1alpha1/backup_types.go
+++ b/apis/dataprotection/v1alpha1/backup_types.go
@@ -22,6 +22,7 @@ import (
 )
 
 // BackupSpec defines the desired state of Backup.
+// +kubebuilder:validation:XValidation:rule="has(oldSelf.parameters) == has(self.parameters)",message="forbidden to update spec.parameters"
 type BackupSpec struct {
 	// Specifies the backup policy to be applied for this backup.
 	//
@@ -74,6 +75,16 @@ type BackupSpec struct {
 	// +optional
 	// +kubebuilder:validation:XValidation:rule="self == oldSelf",message="forbidden to update spec.parentBackupName"
 	ParentBackupName string `json:"parentBackupName,omitempty"`
+
+	// Specifies a list of name-value pairs representing parameters and their corresponding values.
+	// Parameters match the schema specified in the `actionset.spec.parametersSchema`
+	//
+	// +listType=map
+	// +listMapKey=name
+	// +kubebuilder:validation:MaxItems=128
+	// +kubebuilder:validation:XValidation:rule="self == oldSelf",message="forbidden to update spec.parameters"
+	// +optional
+	Parameters []ParameterPair `json:"parameters,omitempty"`
 }
 
 // BackupStatus defines the observed state of Backup.
diff --git a/apis/dataprotection/v1alpha1/backupschedule_types.go b/apis/dataprotection/v1alpha1/backupschedule_types.go
index 37ff4c1c215..ba0476cc4a6 100644
--- a/apis/dataprotection/v1alpha1/backupschedule_types.go
+++ b/apis/dataprotection/v1alpha1/backupschedule_types.go
@@ -49,6 +49,12 @@ type SchedulePolicy struct {
 	// +optional
 	Enabled *bool `json:"enabled,omitempty"`
 
+	// Specifies the name of the schedule. Names cannot be duplicated.
+	// If the name is empty, it will be considered the same as the value of the backupMethod below.
+	//
+	// +optional
+	Name string `json:"name,omitempty"`
+
 	// Specifies the backup method name that is defined in backupPolicy.
 	//
 	// +kubebuilder:validation:Required
@@ -76,6 +82,17 @@ type SchedulePolicy struct {
 	// +optional
 	// +kubebuilder:default="7d"
 	RetentionPeriod RetentionPeriod `json:"retentionPeriod,omitempty"`
+
+	// Specifies a list of name-value pairs representing parameters and their corresponding values.
+	// Parameters match the schema specified in the `actionset.spec.parametersSchema`
+	//
+	// +patchMergeKey=name
+	// +patchStrategy=merge,retainKeys
+	// +listType=map
+	// +listMapKey=name
+	// +kubebuilder:validation:MaxItems=128
+	// +optional
+	Parameters []ParameterPair `json:"parameters,omitempty"`
 }
 
 // BackupScheduleStatus defines the observed state of BackupSchedule.
@@ -174,3 +191,12 @@ type BackupScheduleList struct {
 func init() {
 	SchemeBuilder.Register(&BackupSchedule{}, &BackupScheduleList{})
 }
+
+// GetScheduleName gets the name of schedulePolicy.
+// If name is empty, return backupMethod.
+func (s *SchedulePolicy) GetScheduleName() string {
+	if len(s.Name) > 0 {
+		return s.Name
+	}
+	return s.BackupMethod
+}
diff --git a/apis/dataprotection/v1alpha1/restore_types.go b/apis/dataprotection/v1alpha1/restore_types.go
index 7794ccca108..0c72d7b354c 100644
--- a/apis/dataprotection/v1alpha1/restore_types.go
+++ b/apis/dataprotection/v1alpha1/restore_types.go
@@ -22,6 +22,7 @@ import (
 )
 
 // RestoreSpec defines the desired state of Restore
+// +kubebuilder:validation:XValidation:rule="has(oldSelf.parameters) == has(self.parameters)",message="forbidden to update spec.parameters"
 type RestoreSpec struct {
 	// Specifies the backup to be restored. The restore behavior is based on the backup type:
 	//
@@ -84,6 +85,16 @@ type RestoreSpec struct {
 	// +kubebuilder:validation:Minimum=0
 	// +kubebuilder:validation:Maximum=10
 	BackoffLimit *int32 `json:"backoffLimit,omitempty"`
+
+	// Specifies a list of name-value pairs representing parameters and their corresponding values.
+	// Parameters match the schema specified in the `actionset.spec.parametersSchema`
+	//
+	// +listType=map
+	// +listMapKey=name
+	// +kubebuilder:validation:MaxItems=128
+	// +kubebuilder:validation:XValidation:rule="self == oldSelf",message="forbidden to update spec.parameters"
+	// +optional
+	Parameters []ParameterPair `json:"parameters,omitempty"`
 }
 
 // BackupRef describes the backup info.
diff --git a/apis/dataprotection/v1alpha1/types.go b/apis/dataprotection/v1alpha1/types.go
index 26c4b8f42bb..5b4512e5434 100644
--- a/apis/dataprotection/v1alpha1/types.go
+++ b/apis/dataprotection/v1alpha1/types.go
@@ -24,6 +24,7 @@ import (
 	"unicode"
 
 	corev1 "k8s.io/api/core/v1"
+	apiextensionsv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
 )
 
 // Phase defines the BackupPolicy and ActionSet CR .status.phase
@@ -261,3 +262,28 @@ type EncryptionConfig struct {
 	// +kubebuilder:validation:Required
 	PassPhraseSecretKeyRef *corev1.SecretKeySelector `json:"passPhraseSecretKeyRef"`
 }
+
+type ActionSetParametersSchema struct {
+	// Defines the schema for parameters using the OpenAPI v3.
+	// The supported property types include:
+	// - string
+	// - number
+	// - integer
+	// - array: Note that only items of string type are supported.
+	// +kubebuilder:validation:Schemaless
+	// +kubebuilder:validation:Type=object
+	// +kubebuilder:pruning:PreserveUnknownFields
+	// +k8s:conversion-gen=false
+	// +optional
+	OpenAPIV3Schema *apiextensionsv1.JSONSchemaProps `json:"openAPIV3Schema,omitempty"`
+}
+
+type ParameterPair struct {
+	// Represents the name of the parameter.
+	// +kubebuilder:validation:Required
+	Name string `json:"name"`
+
+	// Represents the parameter values.
+	// +kubebuilder:validation:Required
+	Value string `json:"value"`
+}
diff --git a/apis/dataprotection/v1alpha1/zz_generated.deepcopy.go b/apis/dataprotection/v1alpha1/zz_generated.deepcopy.go
index e17f4620e1e..d9ec61e6d6c 100644
--- a/apis/dataprotection/v1alpha1/zz_generated.deepcopy.go
+++ b/apis/dataprotection/v1alpha1/zz_generated.deepcopy.go
@@ -88,9 +88,33 @@ func (in *ActionSetList) DeepCopyObject() runtime.Object {
 	return nil
 }
 
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *ActionSetParametersSchema) DeepCopyInto(out *ActionSetParametersSchema) {
+	*out = *in
+	if in.OpenAPIV3Schema != nil {
+		in, out := &in.OpenAPIV3Schema, &out.OpenAPIV3Schema
+		*out = (*in).DeepCopy()
+	}
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new ActionSetParametersSchema.
+func (in *ActionSetParametersSchema) DeepCopy() *ActionSetParametersSchema {
+	if in == nil {
+		return nil
+	}
+	out := new(ActionSetParametersSchema)
+	in.DeepCopyInto(out)
+	return out
+}
+
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *ActionSetSpec) DeepCopyInto(out *ActionSetSpec) {
 	*out = *in
+	if in.ParametersSchema != nil {
+		in, out := &in.ParametersSchema, &out.ParametersSchema
+		*out = new(ActionSetParametersSchema)
+		(*in).DeepCopyInto(*out)
+	}
 	if in.Env != nil {
 		in, out := &in.Env, &out.Env
 		*out = make([]v1.EnvVar, len(*in))
@@ -215,7 +239,7 @@ func (in *Backup) DeepCopyInto(out *Backup) {
 	*out = *in
 	out.TypeMeta = in.TypeMeta
 	in.ObjectMeta.DeepCopyInto(&out.ObjectMeta)
-	out.Spec = in.Spec
+	in.Spec.DeepCopyInto(&out.Spec)
 	in.Status.DeepCopyInto(&out.Status)
 }
 
@@ -264,6 +288,11 @@ func (in *BackupActionSpec) DeepCopyInto(out *BackupActionSpec) {
 		*out = new(BaseJobActionSpec)
 		(*in).DeepCopyInto(*out)
 	}
+	if in.WithParameters != nil {
+		in, out := &in.WithParameters, &out.WithParameters
+		*out = make([]string, len(*in))
+		copy(*out, *in)
+	}
 }
 
 // DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new BackupActionSpec.
@@ -897,6 +926,11 @@ func (in *BackupScheduleStatus) DeepCopy() *BackupScheduleStatus {
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *BackupSpec) DeepCopyInto(out *BackupSpec) {
 	*out = *in
+	if in.Parameters != nil {
+		in, out := &in.Parameters, &out.Parameters
+		*out = make([]ParameterPair, len(*in))
+		copy(*out, *in)
+	}
 }
 
 // DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new BackupSpec.
@@ -1331,6 +1365,21 @@ func (in *KubeResources) DeepCopy() *KubeResources {
 	return out
 }
 
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *ParameterPair) DeepCopyInto(out *ParameterPair) {
+	*out = *in
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new ParameterPair.
+func (in *ParameterPair) DeepCopy() *ParameterPair {
+	if in == nil {
+		return nil
+	}
+	out := new(ParameterPair)
+	in.DeepCopyInto(out)
+	return out
+}
+
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *ParametersSchema) DeepCopyInto(out *ParametersSchema) {
 	*out = *in
@@ -1556,6 +1605,11 @@ func (in *RestoreActionSpec) DeepCopyInto(out *RestoreActionSpec) {
 		*out = new(bool)
 		**out = **in
 	}
+	if in.WithParameters != nil {
+		in, out := &in.WithParameters, &out.WithParameters
+		*out = make([]string, len(*in))
+		copy(*out, *in)
+	}
 }
 
 // DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new RestoreActionSpec.
@@ -1654,6 +1708,11 @@ func (in *RestoreSpec) DeepCopyInto(out *RestoreSpec) {
 		*out = new(int32)
 		**out = **in
 	}
+	if in.Parameters != nil {
+		in, out := &in.Parameters, &out.Parameters
+		*out = make([]ParameterPair, len(*in))
+		copy(*out, *in)
+	}
 }
 
 // DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new RestoreSpec.
@@ -1812,6 +1871,11 @@ func (in *SchedulePolicy) DeepCopyInto(out *SchedulePolicy) {
 		*out = new(bool)
 		**out = **in
 	}
+	if in.Parameters != nil {
+		in, out := &in.Parameters, &out.Parameters
+		*out = make([]ParameterPair, len(*in))
+		copy(*out, *in)
+	}
 }
 
 // DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new SchedulePolicy.
diff --git a/apis/operations/v1alpha1/opsrequest_types.go b/apis/operations/v1alpha1/opsrequest_types.go
index b99d8fd08a4..656540374ab 100644
--- a/apis/operations/v1alpha1/opsrequest_types.go
+++ b/apis/operations/v1alpha1/opsrequest_types.go
@@ -24,6 +24,7 @@ import (
 
 	appsv1 "github.com/apecloud/kubeblocks/apis/apps/v1"
 	appsv1alpha1 "github.com/apecloud/kubeblocks/apis/apps/v1alpha1"
+	dpv1alpha1 "github.com/apecloud/kubeblocks/apis/dataprotection/v1alpha1"
 )
 
 // OpsRequestSpec defines the desired state of OpsRequest
@@ -205,12 +206,15 @@ type SpecificOpsRequest struct {
 	ExposeList []Expose `json:"expose,omitempty"`
 
 	// Specifies the parameters to back up a Cluster.
+	//
+	// +kubebuilder:validation:XValidation:rule="has(oldSelf.parameters) == has(self.parameters)",message="forbidden to update backup.parameters"
 	// +optional
 	Backup *Backup `json:"backup,omitempty"`
 
 	// Specifies the parameters to restore a Cluster.
 	// Note that this restore operation will roll back cluster services.
 	//
+	// +kubebuilder:validation:XValidation:rule="has(oldSelf.parameters) == has(self.parameters)",message="forbidden to update restore.parameters"
 	// +optional
 	Restore *Restore `json:"restore,omitempty"`
 
@@ -912,6 +916,16 @@ type Backup struct {
 	//
 	// +optional
 	ParentBackupName string `json:"parentBackupName,omitempty"`
+
+	// Specifies a list of name-value pairs representing parameters and their corresponding values.
+	// Parameters match the schema specified in the `actionset.spec.parametersSchema`
+	//
+	// +listType=map
+	// +listMapKey=name
+	// +kubebuilder:validation:MaxItems=128
+	// +kubebuilder:validation:XValidation:rule="self == oldSelf",message="forbidden to update parameters"
+	// +optional
+	Parameters []dpv1alpha1.ParameterPair `json:"parameters,omitempty"`
 }
 
 type Restore struct {
@@ -953,6 +967,16 @@ type Restore struct {
 	//
 	// This setting is useful for coordinating PostReady operations across the Cluster for optimal cluster conditions.
 	DeferPostReadyUntilClusterRunning bool `json:"deferPostReadyUntilClusterRunning,omitempty"`
+
+	// Specifies a list of name-value pairs representing parameters and their corresponding values.
+	// Parameters match the schema specified in the `actionset.spec.parametersSchema`
+	//
+	// +listType=map
+	// +listMapKey=name
+	// +kubebuilder:validation:MaxItems=128
+	// +kubebuilder:validation:XValidation:rule="self == oldSelf",message="forbidden to update parameters"
+	// +optional
+	Parameters []dpv1alpha1.ParameterPair `json:"parameters,omitempty"`
 }
 
 // OpsRequestStatus represents the observed state of an OpsRequest.
diff --git a/apis/operations/v1alpha1/zz_generated.deepcopy.go b/apis/operations/v1alpha1/zz_generated.deepcopy.go
index 1c7fc224c16..a6e0f7bb3c6 100644
--- a/apis/operations/v1alpha1/zz_generated.deepcopy.go
+++ b/apis/operations/v1alpha1/zz_generated.deepcopy.go
@@ -26,6 +26,7 @@ package v1alpha1
 import (
 	appsv1 "github.com/apecloud/kubeblocks/apis/apps/v1"
 	appsv1alpha1 "github.com/apecloud/kubeblocks/apis/apps/v1alpha1"
+	dataprotectionv1alpha1 "github.com/apecloud/kubeblocks/apis/dataprotection/v1alpha1"
 	"k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	runtime "k8s.io/apimachinery/pkg/runtime"
@@ -49,6 +50,11 @@ func (in *ActionTask) DeepCopy() *ActionTask {
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *Backup) DeepCopyInto(out *Backup) {
 	*out = *in
+	if in.Parameters != nil {
+		in, out := &in.Parameters, &out.Parameters
+		*out = make([]dataprotectionv1alpha1.ParameterPair, len(*in))
+		copy(*out, *in)
+	}
 }
 
 // DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new Backup.
@@ -1352,6 +1358,11 @@ func (in *Restore) DeepCopyInto(out *Restore) {
 			(*in)[i].DeepCopyInto(&(*out)[i])
 		}
 	}
+	if in.Parameters != nil {
+		in, out := &in.Parameters, &out.Parameters
+		*out = make([]dataprotectionv1alpha1.ParameterPair, len(*in))
+		copy(*out, *in)
+	}
 }
 
 // DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new Restore.
@@ -1494,7 +1505,7 @@ func (in *SpecificOpsRequest) DeepCopyInto(out *SpecificOpsRequest) {
 	if in.Backup != nil {
 		in, out := &in.Backup, &out.Backup
 		*out = new(Backup)
-		**out = **in
+		(*in).DeepCopyInto(*out)
 	}
 	if in.Restore != nil {
 		in, out := &in.Restore, &out.Restore
diff --git a/config/crd/bases/dataprotection.kubeblocks.io_actionsets.yaml b/config/crd/bases/dataprotection.kubeblocks.io_actionsets.yaml
index 40a85d26a67..71b5f38182c 100644
--- a/config/crd/bases/dataprotection.kubeblocks.io_actionsets.yaml
+++ b/config/crd/bases/dataprotection.kubeblocks.io_actionsets.yaml
@@ -269,6 +269,11 @@ spec:
                     - command
                     - image
                     type: object
+                  withParameters:
+                    description: Specifies the parameters used by the backup action
+                    items:
+                      type: string
+                    type: array
                 type: object
               backupType:
                 allOf:
@@ -457,6 +462,21 @@ spec:
                   type: object
                 type: array
                 x-kubernetes-preserve-unknown-fields: true
+              parametersSchema:
+                description: Specifies the schema of parameters in backups and restores
+                  before their usage.
+                properties:
+                  openAPIV3Schema:
+                    description: |-
+                      Defines the schema for parameters using the OpenAPI v3.
+                      The supported property types include:
+                      - string
+                      - number
+                      - integer
+                      - array: Note that only items of string type are supported.
+                    type: object
+                    x-kubernetes-preserve-unknown-fields: true
+                type: object
               restore:
                 description: Specifies the restore action.
                 properties:
@@ -568,6 +588,11 @@ spec:
                     - command
                     - image
                     type: object
+                  withParameters:
+                    description: Specifies the parameters used by the restore action
+                    items:
+                      type: string
+                    type: array
                 type: object
             required:
             - backupType
diff --git a/config/crd/bases/dataprotection.kubeblocks.io_backuppolicytemplates.yaml b/config/crd/bases/dataprotection.kubeblocks.io_backuppolicytemplates.yaml
index 7cd65a409c2..ba430dc6ac7 100644
--- a/config/crd/bases/dataprotection.kubeblocks.io_backuppolicytemplates.yaml
+++ b/config/crd/bases/dataprotection.kubeblocks.io_backuppolicytemplates.yaml
@@ -359,6 +359,32 @@ spec:
                       description: Specifies whether the backup schedule is enabled
                         or not.
                       type: boolean
+                    name:
+                      description: |-
+                        Specifies the name of the schedule. Names cannot be duplicated.
+                        If the name is empty, it will be considered the same as the value of the backupMethod below.
+                      type: string
+                    parameters:
+                      description: |-
+                        Specifies a list of name-value pairs representing parameters and their corresponding values.
+                        Parameters match the schema specified in the `actionset.spec.parametersSchema`
+                      items:
+                        properties:
+                          name:
+                            description: Represents the name of the parameter.
+                            type: string
+                          value:
+                            description: Represents the parameter values.
+                            type: string
+                        required:
+                        - name
+                        - value
+                        type: object
+                      maxItems: 128
+                      type: array
+                      x-kubernetes-list-map-keys:
+                      - name
+                      x-kubernetes-list-type: map
                     retentionPeriod:
                       default: 7d
                       description: "Determines the duration for which the backup should
diff --git a/config/crd/bases/dataprotection.kubeblocks.io_backups.yaml b/config/crd/bases/dataprotection.kubeblocks.io_backups.yaml
index a2d196267d1..23060c13a49 100644
--- a/config/crd/bases/dataprotection.kubeblocks.io_backups.yaml
+++ b/config/crd/bases/dataprotection.kubeblocks.io_backups.yaml
@@ -110,6 +110,30 @@ spec:
                     the backup CR but retaining the backup contents in backup repository.
                     The current implementation only prevent accidental deletion of backup data.
                 type: string
+              parameters:
+                description: |-
+                  Specifies a list of name-value pairs representing parameters and their corresponding values.
+                  Parameters match the schema specified in the `actionset.spec.parametersSchema`
+                items:
+                  properties:
+                    name:
+                      description: Represents the name of the parameter.
+                      type: string
+                    value:
+                      description: Represents the parameter values.
+                      type: string
+                  required:
+                  - name
+                  - value
+                  type: object
+                maxItems: 128
+                type: array
+                x-kubernetes-list-map-keys:
+                - name
+                x-kubernetes-list-type: map
+                x-kubernetes-validations:
+                - message: forbidden to update spec.parameters
+                  rule: self == oldSelf
               parentBackupName:
                 description: Determines the parent backup name for incremental or
                   differential backup.
@@ -130,6 +154,9 @@ spec:
             - backupMethod
             - backupPolicyName
             type: object
+            x-kubernetes-validations:
+            - message: forbidden to update spec.parameters
+              rule: has(oldSelf.parameters) == has(self.parameters)
           status:
             description: BackupStatus defines the observed state of Backup.
             properties:
diff --git a/config/crd/bases/dataprotection.kubeblocks.io_backupschedules.yaml b/config/crd/bases/dataprotection.kubeblocks.io_backupschedules.yaml
index f702df13a54..bd81a15826b 100644
--- a/config/crd/bases/dataprotection.kubeblocks.io_backupschedules.yaml
+++ b/config/crd/bases/dataprotection.kubeblocks.io_backupschedules.yaml
@@ -72,6 +72,32 @@ spec:
                       description: Specifies whether the backup schedule is enabled
                         or not.
                       type: boolean
+                    name:
+                      description: |-
+                        Specifies the name of the schedule. Names cannot be duplicated.
+                        If the name is empty, it will be considered the same as the value of the backupMethod below.
+                      type: string
+                    parameters:
+                      description: |-
+                        Specifies a list of name-value pairs representing parameters and their corresponding values.
+                        Parameters match the schema specified in the `actionset.spec.parametersSchema`
+                      items:
+                        properties:
+                          name:
+                            description: Represents the name of the parameter.
+                            type: string
+                          value:
+                            description: Represents the parameter values.
+                            type: string
+                        required:
+                        - name
+                        - value
+                        type: object
+                      maxItems: 128
+                      type: array
+                      x-kubernetes-list-map-keys:
+                      - name
+                      x-kubernetes-list-type: map
                     retentionPeriod:
                       default: 7d
                       description: "Determines the duration for which the backup should
diff --git a/config/crd/bases/dataprotection.kubeblocks.io_restores.yaml b/config/crd/bases/dataprotection.kubeblocks.io_restores.yaml
index a4d6823c12a..199ea02288d 100644
--- a/config/crd/bases/dataprotection.kubeblocks.io_restores.yaml
+++ b/config/crd/bases/dataprotection.kubeblocks.io_restores.yaml
@@ -270,6 +270,30 @@ spec:
                   type: object
                 type: array
                 x-kubernetes-preserve-unknown-fields: true
+              parameters:
+                description: |-
+                  Specifies a list of name-value pairs representing parameters and their corresponding values.
+                  Parameters match the schema specified in the `actionset.spec.parametersSchema`
+                items:
+                  properties:
+                    name:
+                      description: Represents the name of the parameter.
+                      type: string
+                    value:
+                      description: Represents the parameter values.
+                      type: string
+                  required:
+                  - name
+                  - value
+                  type: object
+                maxItems: 128
+                type: array
+                x-kubernetes-list-map-keys:
+                - name
+                x-kubernetes-list-type: map
+                x-kubernetes-validations:
+                - message: forbidden to update spec.parameters
+                  rule: self == oldSelf
               prepareDataConfig:
                 description: |-
                   Configuration for the action of "prepareData" phase, including the persistent volume claims
@@ -2414,6 +2438,9 @@ spec:
             required:
             - backup
             type: object
+            x-kubernetes-validations:
+            - message: forbidden to update spec.parameters
+              rule: has(oldSelf.parameters) == has(self.parameters)
           status:
             description: RestoreStatus defines the observed state of Restore
             properties:
diff --git a/config/crd/bases/operations.kubeblocks.io_opsrequests.yaml b/config/crd/bases/operations.kubeblocks.io_opsrequests.yaml
index 40dfb75b258..49398da5161 100644
--- a/config/crd/bases/operations.kubeblocks.io_opsrequests.yaml
+++ b/config/crd/bases/operations.kubeblocks.io_opsrequests.yaml
@@ -91,6 +91,30 @@ spec:
                     - Delete
                     - Retain
                     type: string
+                  parameters:
+                    description: |-
+                      Specifies a list of name-value pairs representing parameters and their corresponding values.
+                      Parameters match the schema specified in the `actionset.spec.parametersSchema`
+                    items:
+                      properties:
+                        name:
+                          description: Represents the name of the parameter.
+                          type: string
+                        value:
+                          description: Represents the parameter values.
+                          type: string
+                      required:
+                      - name
+                      - value
+                      type: object
+                    maxItems: 128
+                    type: array
+                    x-kubernetes-list-map-keys:
+                    - name
+                    x-kubernetes-list-type: map
+                    x-kubernetes-validations:
+                    - message: forbidden to update parameters
+                      rule: self == oldSelf
                   parentBackupName:
                     description: If the specified BackupMethod is incremental, `parentBackupName`
                       is required.
@@ -121,6 +145,9 @@ spec:
                       Otherwise, only the Backup custom resource will be deleted.
                     type: string
                 type: object
+                x-kubernetes-validations:
+                - message: forbidden to update backup.parameters
+                  rule: has(oldSelf.parameters) == has(self.parameters)
               cancel:
                 description: |-
                   Indicates whether the current operation should be canceled and terminated gracefully if it's in the
@@ -4407,6 +4434,30 @@ spec:
                       type: object
                     type: array
                     x-kubernetes-preserve-unknown-fields: true
+                  parameters:
+                    description: |-
+                      Specifies a list of name-value pairs representing parameters and their corresponding values.
+                      Parameters match the schema specified in the `actionset.spec.parametersSchema`
+                    items:
+                      properties:
+                        name:
+                          description: Represents the name of the parameter.
+                          type: string
+                        value:
+                          description: Represents the parameter values.
+                          type: string
+                      required:
+                      - name
+                      - value
+                      type: object
+                    maxItems: 128
+                    type: array
+                    x-kubernetes-list-map-keys:
+                    - name
+                    x-kubernetes-list-type: map
+                    x-kubernetes-validations:
+                    - message: forbidden to update parameters
+                      rule: self == oldSelf
                   restorePointInTime:
                     description: |-
                       Specifies the point in time to which the restore should be performed.
@@ -4433,6 +4484,9 @@ spec:
                 required:
                 - backupName
                 type: object
+                x-kubernetes-validations:
+                - message: forbidden to update restore.parameters
+                  rule: has(oldSelf.parameters) == has(self.parameters)
               start:
                 description: Lists Components to be started. If empty, all components
                   will be started.
diff --git a/controllers/apps/transformer_cluster_backup_policy.go b/controllers/apps/transformer_cluster_backup_policy.go
index 2ce9c3090ed..f6867accd71 100644
--- a/controllers/apps/transformer_cluster_backup_policy.go
+++ b/controllers/apps/transformer_cluster_backup_policy.go
@@ -283,11 +283,14 @@ func (r *backupPolicyBuilder) buildBackupSchedule(
 
 	var schedules []dpv1alpha1.SchedulePolicy
 	for _, s := range r.backupPolicyTPL.Spec.Schedules {
+		name := s.GetScheduleName()
 		schedules = append(schedules, dpv1alpha1.SchedulePolicy{
 			BackupMethod:    s.BackupMethod,
 			CronExpression:  s.CronExpression,
 			Enabled:         s.Enabled,
 			RetentionPeriod: s.RetentionPeriod,
+			Name:            name,
+			Parameters:      s.Parameters,
 		})
 	}
 	backupSchedule.Spec.Schedules = schedules
@@ -295,16 +298,22 @@ func (r *backupPolicyBuilder) buildBackupSchedule(
 }
 
 func (r *backupPolicyBuilder) syncBackupSchedule(backupSchedule *dpv1alpha1.BackupSchedule) {
-	scheduleMethodMap := map[string]struct{}{}
-	for _, s := range backupSchedule.Spec.Schedules {
-		scheduleMethodMap[s.BackupMethod] = struct{}{}
+	scheduleNameMap := map[string]struct{}{}
+	for i := range backupSchedule.Spec.Schedules {
+		s := &backupSchedule.Spec.Schedules[i]
+		if len(s.Name) == 0 {
+			// assign to backupMethod if name is empty.
+			s.Name = s.BackupMethod
+		}
+		scheduleNameMap[s.Name] = struct{}{}
 	}
 	mergeMap(backupSchedule.Annotations, r.buildAnnotations())
 	// update backupSchedule annotation to reconcile it.
 	backupSchedule.Annotations[constant.ReconcileAnnotationKey] = r.Cluster.ResourceVersion
 	// sync the newly added schedule policies.
 	for _, s := range r.backupPolicyTPL.Spec.Schedules {
-		if _, ok := scheduleMethodMap[s.BackupMethod]; ok {
+		name := s.GetScheduleName()
+		if _, ok := scheduleNameMap[name]; ok {
 			continue
 		}
 		backupSchedule.Spec.Schedules = append(backupSchedule.Spec.Schedules, dpv1alpha1.SchedulePolicy{
@@ -312,6 +321,8 @@ func (r *backupPolicyBuilder) syncBackupSchedule(backupSchedule *dpv1alpha1.Back
 			CronExpression:  s.CronExpression,
 			Enabled:         s.Enabled,
 			RetentionPeriod: s.RetentionPeriod,
+			Name:            name,
+			Parameters:      s.Parameters,
 		})
 	}
 }
diff --git a/controllers/dataprotection/actionset_controller.go b/controllers/dataprotection/actionset_controller.go
index 787977e47c0..43f963fe967 100644
--- a/controllers/dataprotection/actionset_controller.go
+++ b/controllers/dataprotection/actionset_controller.go
@@ -21,6 +21,7 @@ package dataprotection
 
 import (
 	"context"
+	"fmt"
 
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/client-go/tools/record"
@@ -68,8 +69,7 @@ func (r *ActionSetReconciler) Reconcile(ctx context.Context, req ctrl.Request) (
 		return *res, err
 	}
 
-	if actionSet.Status.ObservedGeneration == actionSet.Generation &&
-		actionSet.Status.Phase.IsAvailable() {
+	if actionSet.Status.ObservedGeneration == actionSet.Generation && actionSet.Status.Phase.IsAvailable() {
 		return ctrl.Result{}, nil
 	}
 
@@ -81,9 +81,14 @@ func (r *ActionSetReconciler) Reconcile(ctx context.Context, req ctrl.Request) (
 		return r.Client.Status().Patch(reqCtx.Ctx, actionSet, patch)
 	}
 
-	// TODO(ldm): validate actionSet
+	phase := dpv1alpha1.AvailablePhase
+	var message string
+	if err = r.validate(actionSet); err != nil {
+		phase = dpv1alpha1.UnavailablePhase
+		message = err.Error()
+	}
 
-	if err = patchStatus(dpv1alpha1.AvailablePhase, ""); err != nil {
+	if err = patchStatus(phase, message); err != nil {
 		return intctrlutil.CheckedRequeueWithError(err, reqCtx.Log, "")
 	}
 	intctrlutil.RecordCreatedEvent(r.Recorder, actionSet)
@@ -102,3 +107,35 @@ func (r *ActionSetReconciler) deleteExternalResources(
 	_ *dpv1alpha1.ActionSet) error {
 	return nil
 }
+
+func (r *ActionSetReconciler) validate(actionset *dpv1alpha1.ActionSet) error {
+	validateWithParameters := func(withParameters []string) error {
+		if len(withParameters) == 0 {
+			return nil
+		}
+		schema := actionset.Spec.ParametersSchema
+		if schema == nil || schema.OpenAPIV3Schema == nil || len(schema.OpenAPIV3Schema.Properties) == 0 {
+			return fmt.Errorf("the parametersSchema is invalid")
+		}
+		properties := schema.OpenAPIV3Schema.Properties
+		for _, parameter := range withParameters {
+			if _, ok := properties[parameter]; !ok {
+				return fmt.Errorf("parameter %s is not defined in the parametersSchema", parameter)
+			}
+		}
+		return nil
+	}
+
+	// validate withParameters
+	if actionset.Spec.Backup != nil {
+		if err := validateWithParameters(actionset.Spec.Backup.WithParameters); err != nil {
+			return fmt.Errorf("fails to validate backup withParameters: %v", err)
+		}
+	}
+	if actionset.Spec.Restore != nil {
+		if err := validateWithParameters(actionset.Spec.Restore.WithParameters); err != nil {
+			return fmt.Errorf("fails to validate restore withParameters: %v", err)
+		}
+	}
+	return nil
+}
diff --git a/controllers/dataprotection/actionset_controller_test.go b/controllers/dataprotection/actionset_controller_test.go
index e53f9df1f5b..e3495edaf52 100644
--- a/controllers/dataprotection/actionset_controller_test.go
+++ b/controllers/dataprotection/actionset_controller_test.go
@@ -22,9 +22,10 @@ package dataprotection
 import (
 	. "github.com/onsi/ginkgo/v2"
 	. "github.com/onsi/gomega"
-
+	apiextensionsv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
+	dpv1alpha1 "github.com/apecloud/kubeblocks/apis/dataprotection/v1alpha1"
 	intctrlutil "github.com/apecloud/kubeblocks/pkg/generics"
 	testapps "github.com/apecloud/kubeblocks/pkg/testutil/apps"
 	testdp "github.com/apecloud/kubeblocks/pkg/testutil/dataprotection"
@@ -58,4 +59,50 @@ var _ = Describe("ActionSet Controller test", func() {
 			Expect(as).ShouldNot(BeNil())
 		})
 	})
+
+	Context("validate a actionSet", func() {
+		It("validate withParameters", func() {
+			as := testdp.NewFakeActionSet(&testCtx)
+			Expect(as).ShouldNot(BeNil())
+			By("set invalid withParameters and schema")
+			Expect(testapps.ChangeObj(&testCtx, as, func(action *dpv1alpha1.ActionSet) {
+				as.Spec.ParametersSchema = &dpv1alpha1.ActionSetParametersSchema{
+					OpenAPIV3Schema: &apiextensionsv1.JSONSchemaProps{
+						Properties: map[string]apiextensionsv1.JSONSchemaProps{
+							testdp.ParameterString: {
+								Type: testdp.ParameterStringType,
+							},
+							testdp.ParameterArray: {
+								Type: testdp.ParameterArrayType,
+								Items: &apiextensionsv1.JSONSchemaPropsOrArray{
+									Schema: &apiextensionsv1.JSONSchemaProps{
+										Type: testdp.ParameterStringType,
+									},
+								},
+							},
+						},
+					},
+				}
+				as.Spec.Backup.WithParameters = []string{testdp.InvalidParameter}
+			})).Should(Succeed())
+			By("should be unavailable with invalid withParameters")
+			Eventually(testapps.CheckObj(&testCtx, client.ObjectKeyFromObject(as),
+				func(g Gomega, as *dpv1alpha1.ActionSet) {
+					g.Expect(as.Status.ObservedGeneration).Should(Equal(as.Generation))
+					g.Expect(as.Status.Phase).Should(BeEquivalentTo(dpv1alpha1.UnavailablePhase))
+					g.Expect(as.Status.Message).ShouldNot(BeEmpty())
+				})).Should(Succeed())
+			By("set valid parameters")
+			Expect(testapps.ChangeObj(&testCtx, as, func(action *dpv1alpha1.ActionSet) {
+				as.Spec.Backup.WithParameters = []string{testdp.ParameterString, testdp.ParameterArray}
+			})).Should(Succeed())
+			By("should be available")
+			Eventually(testapps.CheckObj(&testCtx, client.ObjectKeyFromObject(as),
+				func(g Gomega, as *dpv1alpha1.ActionSet) {
+					g.Expect(as.Status.ObservedGeneration).Should(Equal(as.Generation))
+					g.Expect(as.Status.Phase).Should(BeEquivalentTo(dpv1alpha1.AvailablePhase))
+					g.Expect(as.Status.Message).Should(BeEmpty())
+				})).Should(Succeed())
+		})
+	})
 })
diff --git a/controllers/dataprotection/backup_controller.go b/controllers/dataprotection/backup_controller.go
index 3595876b8e0..cabe2f844ab 100644
--- a/controllers/dataprotection/backup_controller.go
+++ b/controllers/dataprotection/backup_controller.go
@@ -358,7 +358,6 @@ func (r *BackupReconciler) prepareBackupRequest(
 		RequestCtx: reqCtx,
 		Client:     r.Client,
 	}
-
 	if request.Annotations == nil {
 		request.Annotations = make(map[string]string)
 	}
@@ -413,6 +412,11 @@ func (r *BackupReconciler) prepareBackupRequest(
 					backupSchedule.Namespace, backupSchedule.Name)
 			}
 		}
+
+		// validate parameters
+		if err := dputils.ValidateParameters(actionSet, backup.Spec.Parameters, true); err != nil {
+			return nil, fmt.Errorf("fails to validate parameters with actionset %s: %v", actionSet.Name, err)
+		}
 	}
 
 	// check encryption config
diff --git a/controllers/dataprotection/backup_controller_test.go b/controllers/dataprotection/backup_controller_test.go
index e138d6609ab..f9ceb194bb9 100644
--- a/controllers/dataprotection/backup_controller_test.go
+++ b/controllers/dataprotection/backup_controller_test.go
@@ -26,11 +26,10 @@ import (
 	"strconv"
 	"time"
 
+	vsv1 "github.com/kubernetes-csi/external-snapshotter/client/v6/apis/volumesnapshot/v1"
 	. "github.com/onsi/ginkgo/v2"
 	. "github.com/onsi/gomega"
 	appsv1 "k8s.io/api/apps/v1"
-
-	vsv1 "github.com/kubernetes-csi/external-snapshotter/client/v6/apis/volumesnapshot/v1"
 	batchv1 "k8s.io/api/batch/v1"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -100,6 +99,7 @@ var _ = Describe("Backup Controller test", func() {
 
 	When("with default settings", func() {
 		var (
+			actionSet    *dpv1alpha1.ActionSet
 			backupPolicy *dpv1alpha1.BackupPolicy
 			repoPVCName  string
 			cluster      *kbappsv1.Cluster
@@ -109,7 +109,7 @@ var _ = Describe("Backup Controller test", func() {
 
 		BeforeEach(func() {
 			By("creating an actionSet")
-			actionSet := testdp.NewFakeActionSet(&testCtx)
+			actionSet = testdp.NewFakeActionSet(&testCtx)
 
 			By("creating storage provider")
 			_ = testdp.NewFakeStorageProvider(&testCtx, nil)
@@ -483,6 +483,55 @@ var _ = Describe("Backup Controller test", func() {
 				g.Expect(getDPDBPortEnv(&fetched.Spec.Template.Spec.Containers[0]).Value).Should(Equal(strconv.Itoa(testdp.PortNum)))
 			})).Should(Succeed())
 		})
+		Context("creates backups with parameters", func() {
+			BeforeEach(func() {
+				By("set backup parameters and schema in acitionSet")
+				testdp.MockActionSetWithSchema(&testCtx, actionSet)
+			})
+			It("should fail if parameters are invalid", func() {
+				By("create a backup with invalid parameters")
+				backup := testdp.NewFakeBackup(&testCtx, func(bp *dpv1alpha1.Backup) {
+					bp.Spec.Parameters = testdp.InvalidParameters
+				})
+				By("check the backup")
+				Eventually(testapps.CheckObj(&testCtx, client.ObjectKeyFromObject(backup), func(g Gomega, fetched *dpv1alpha1.Backup) {
+					g.Expect(fetched.Status.Phase).To(Equal(dpv1alpha1.BackupPhaseFailed))
+				})).Should(Succeed())
+
+			})
+			It("should succeed if parameters are valid", func() {
+				By("create a backup with parameters")
+				backup := testdp.NewFakeBackup(&testCtx, func(bp *dpv1alpha1.Backup) {
+					bp.Spec.Parameters = testdp.TestParameters
+				})
+				By("check the backup")
+				Eventually(testapps.CheckObj(&testCtx, client.ObjectKeyFromObject(backup), func(g Gomega, fetched *dpv1alpha1.Backup) {
+					g.Expect(fetched.Status.Phase).To(Equal(dpv1alpha1.BackupPhaseRunning))
+				})).Should(Succeed())
+
+				By("check the backup job and env")
+				getJobKey := func(index int) client.ObjectKey {
+					return client.ObjectKey{
+						Name:      dpbackup.GenerateBackupJobName(backup, fmt.Sprintf("%s-%d", dpbackup.BackupDataJobNamePrefix, index)),
+						Namespace: backup.Namespace,
+					}
+				}
+				Eventually(testapps.CheckObj(&testCtx, getJobKey(0), func(g Gomega, job *batchv1.Job) {
+					g.Expect(len(job.Spec.Template.Spec.Containers)).ShouldNot(BeZero())
+					for _, c := range job.Spec.Template.Spec.Containers {
+						count := 0
+						for _, env := range c.Env {
+							for _, param := range testdp.TestParameters {
+								if param.Name == env.Name && param.Value == env.Value {
+									count++
+								}
+							}
+						}
+						g.Expect(count).To(Equal(len(testdp.TestParameters)))
+					}
+				})).Should(Succeed())
+			})
+		})
 		Context("creates a backup with encryption", func() {
 			const (
 				encryptionKeySecretName = "backup-encryption"
diff --git a/controllers/dataprotection/backuppolicytemplate_controller.go b/controllers/dataprotection/backuppolicytemplate_controller.go
index 40d46cedcc4..fafcde67158 100644
--- a/controllers/dataprotection/backuppolicytemplate_controller.go
+++ b/controllers/dataprotection/backuppolicytemplate_controller.go
@@ -24,7 +24,6 @@ import (
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	k8sruntime "k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/types"
-	"k8s.io/apimachinery/pkg/util/sets"
 	"k8s.io/client-go/tools/record"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
@@ -36,6 +35,7 @@ import (
 	dpv1alpha1 "github.com/apecloud/kubeblocks/apis/dataprotection/v1alpha1"
 	"github.com/apecloud/kubeblocks/pkg/controller/component"
 	intctrlutil "github.com/apecloud/kubeblocks/pkg/controllerutil"
+	dputils "github.com/apecloud/kubeblocks/pkg/dataprotection/utils"
 	"github.com/apecloud/kubeblocks/pkg/dataprotection/utils/boolptr"
 )
 
@@ -95,11 +95,12 @@ func (r *BackupPolicyTemplateReconciler) setComponentDefLabels(reqCtx intctrluti
 
 func (r *BackupPolicyTemplateReconciler) validateAvailable(reqCtx intctrlutil.RequestCtx, oldBPT, bpt *dpv1alpha1.BackupPolicyTemplate) error {
 	message := ""
-	backupMethodMap := map[string]sets.Empty{}
+	backupMethodMap := map[string]*dpv1alpha1.ActionSet{}
 	actionSetNotFound := false
 	// validate the referred actionSetName of the backupMethod
 	for _, v := range bpt.Spec.BackupMethods {
-		backupMethodMap[v.Name] = sets.Empty{}
+		// confirm the method exists
+		backupMethodMap[v.Name] = nil
 		if boolptr.IsSetToFalse(v.SnapshotVolumes) && v.ActionSetName == "" {
 			message += fmt.Sprintf(`backupMethod "%s" is missing an ActionSet name;`, v.Name)
 			continue
@@ -116,13 +117,28 @@ func (r *BackupPolicyTemplateReconciler) validateAvailable(reqCtx intctrlutil.Re
 			}
 			return err
 		}
+		// record found actionSets
+		backupMethodMap[v.Name] = actionSet
+	}
+	// validate the schedule names
+	if err := dputils.ValidateScheduleNames(bpt.Spec.Schedules); err != nil {
+		message += fmt.Sprintf(`fails to validate schedule name: %v;`, err)
 	}
 	// validate the schedules
 	for _, v := range bpt.Spec.Schedules {
-		if _, ok := backupMethodMap[v.BackupMethod]; !ok {
+		actionSet, ok := backupMethodMap[v.BackupMethod]
+		if !ok {
 			message += fmt.Sprintf(`backupMethod "%s" not found in the spec.backupMethods;`, v.BackupMethod)
+			continue
+		}
+		// validate schedule parameters
+		if actionSet != nil {
+			if err := dputils.ValidateParameters(actionSet, v.Parameters, true); err != nil {
+				message += fmt.Sprintf(`fails to validate parameters of backupMethod "%s": %v;`, v.BackupMethod, err)
+			}
 		}
 	}
+
 	bpt.Status.ObservedGeneration = bpt.Generation
 	bpt.Status.Message = message
 	if len(message) > 0 {
diff --git a/controllers/dataprotection/backuppolicytemplate_controller_test.go b/controllers/dataprotection/backuppolicytemplate_controller_test.go
index 7f958e55e29..8a1f419127c 100644
--- a/controllers/dataprotection/backuppolicytemplate_controller_test.go
+++ b/controllers/dataprotection/backuppolicytemplate_controller_test.go
@@ -21,7 +21,6 @@ import (
 
 	. "github.com/onsi/ginkgo/v2"
 	. "github.com/onsi/gomega"
-
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
 	dpv1alpha1 "github.com/apecloud/kubeblocks/apis/dataprotection/v1alpha1"
@@ -47,6 +46,7 @@ var _ = Describe("", func() {
 
 		ml := client.HasLabels{testCtx.TestObjLabelKey}
 		testapps.ClearResourcesWithRemoveFinalizerOption(&testCtx, intctrlutil.BackupPolicyTemplateSignature, true, ml)
+		testapps.ClearResourcesWithRemoveFinalizerOption(&testCtx, intctrlutil.ActionSetSignature, true, ml)
 	}
 
 	BeforeEach(func() {
@@ -77,8 +77,8 @@ var _ = Describe("", func() {
 				SetBackupMethodVolumeMounts("data", "/data").
 				AddBackupMethod(VsBackupMethodName, true, "").
 				SetBackupMethodVolumeMounts("data", "/data").
-				AddSchedule(BackupMethod, "0 0 * * *", ttl, true).
-				AddSchedule(VsBackupMethodName, "0 0 * * *", ttl, true).
+				AddSchedule(BackupMethod, "0 0 * * *", ttl, true, "", nil).
+				AddSchedule(VsBackupMethodName, "0 0 * * *", ttl, true, "", nil).
 				Create(&testCtx).GetObject()
 			key := client.ObjectKeyFromObject(bpt)
 
@@ -103,6 +103,38 @@ var _ = Describe("", func() {
 				g.Expect(pobj.Status.Message).To(BeEmpty())
 			})).Should(Succeed())
 		})
+		It("test BackupPolicyTemplate schedule parameters", func() {
+			const (
+				scheduleName1 = "test1"
+				scheduleName2 = "test2"
+			)
+			By("set backup parameters and schema in acitionSet")
+			actionSet := testdp.NewFakeActionSet(&testCtx)
+			testdp.MockActionSetWithSchema(&testCtx, actionSet)
+			bpt := testdp.NewBackupPolicyTemplateFactory(BackupPolicyTemplateName).
+				AddBackupMethod(BackupMethod, false, testdp.ActionSetName).
+				SetBackupMethodVolumeMounts("data", "/data").
+				AddSchedule(BackupMethod, "0 0 * * *", ttl, true, scheduleName1, testdp.InvalidParameters).
+				AddSchedule(BackupMethod, "0 0 * * *", ttl, true, scheduleName2, testdp.TestParameters).
+				AddSchedule(BackupMethod, "0 0 * * *", ttl, true, "", nil).
+				Create(&testCtx).GetObject()
+			key := client.ObjectKeyFromObject(bpt)
+			By("should be unavailable")
+			Eventually(testapps.CheckObj(&testCtx, key, func(g Gomega, pobj *dpv1alpha1.BackupPolicyTemplate) {
+				g.Expect(pobj.Status.ObservedGeneration).To(Equal(bpt.Generation))
+				g.Expect(pobj.Status.Phase).To(Equal(dpv1alpha1.UnavailablePhase))
+				g.Expect(pobj.Status.Message).To(ContainSubstring(fmt.Sprintf(`fails to validate parameters of backupMethod "%s"`, BackupMethod)))
+			})).Should(Succeed())
+			By("should be available")
+			Expect(testapps.ChangeObj(&testCtx, bpt, func(pobj *dpv1alpha1.BackupPolicyTemplate) {
+				bpt.Spec.Schedules[0].Parameters = testdp.TestParameters
+			})).Should(Succeed())
+			Eventually(testapps.CheckObj(&testCtx, key, func(g Gomega, pobj *dpv1alpha1.BackupPolicyTemplate) {
+				g.Expect(pobj.Status.ObservedGeneration).To(Equal(bpt.Generation))
+				g.Expect(pobj.Status.Phase).To(Equal(dpv1alpha1.AvailablePhase))
+				g.Expect(pobj.Status.Message).To(BeEmpty())
+			})).Should(Succeed())
+		})
 	})
 
 })
diff --git a/controllers/dataprotection/backupschedule_controller_test.go b/controllers/dataprotection/backupschedule_controller_test.go
index 3557f62cf31..25c4a82ab04 100644
--- a/controllers/dataprotection/backupschedule_controller_test.go
+++ b/controllers/dataprotection/backupschedule_controller_test.go
@@ -24,7 +24,6 @@ import (
 
 	. "github.com/onsi/ginkgo/v2"
 	. "github.com/onsi/gomega"
-
 	batchv1 "k8s.io/api/batch/v1"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
@@ -85,19 +84,20 @@ var _ = Describe("Backup Schedule Controller", func() {
 	When("creating backup schedule with default settings", func() {
 		var (
 			backupPolicy *dpv1alpha1.BackupPolicy
+			actionSet    *dpv1alpha1.ActionSet
 		)
 
 		getCronjobKey := func(backupSchedule *dpv1alpha1.BackupSchedule,
-			method string) client.ObjectKey {
+			method string, name string) client.ObjectKey {
 			return client.ObjectKey{
-				Name:      dpbackup.GenerateCRNameByBackupSchedule(backupSchedule, method),
+				Name:      dpbackup.GenerateCRNameByScheduleNameAndMethod(backupSchedule, method, name),
 				Namespace: backupPolicy.Namespace,
 			}
 		}
 
 		BeforeEach(func() {
 			By("creating an actionSet")
-			actionSet := testdp.NewFakeActionSet(&testCtx)
+			actionSet = testdp.NewFakeActionSet(&testCtx)
 
 			By("creating storage provider")
 			_ = testdp.NewFakeStorageProvider(&testCtx, nil)
@@ -130,16 +130,16 @@ var _ = Describe("Backup Schedule Controller", func() {
 				})).Should(Succeed())
 
 				By("checking cronjob, should not exist because all schedule policies of methods are disabled")
-				Eventually(testapps.CheckObjExists(&testCtx, getCronjobKey(backupSchedule, testdp.BackupMethodName),
+				Eventually(testapps.CheckObjExists(&testCtx, getCronjobKey(backupSchedule, testdp.BackupMethodName, ""),
 					&batchv1.CronJob{}, false)).Should(Succeed())
-				Eventually(testapps.CheckObjExists(&testCtx, getCronjobKey(backupSchedule, testdp.VSBackupMethodName),
+				Eventually(testapps.CheckObjExists(&testCtx, getCronjobKey(backupSchedule, testdp.VSBackupMethodName, ""),
 					&batchv1.CronJob{}, false)).Should(Succeed())
 
 				By(fmt.Sprintf("enabling %s method schedule", testdp.BackupMethodName))
 				testdp.EnableBackupSchedule(&testCtx, backupSchedule, testdp.BackupMethodName)
 
 				By("checking cronjob, should exist one cronjob to create backup")
-				Eventually(testapps.CheckObj(&testCtx, getCronjobKey(backupSchedule, testdp.BackupMethodName), func(g Gomega, fetched *batchv1.CronJob) {
+				Eventually(testapps.CheckObj(&testCtx, getCronjobKey(backupSchedule, testdp.BackupMethodName, ""), func(g Gomega, fetched *batchv1.CronJob) {
 					schedulePolicy := dpbackup.GetSchedulePolicyByMethod(backupSchedule, testdp.BackupMethodName)
 					timeZone, cronExpr := dpbackup.BuildCronJobSchedule(schedulePolicy.CronExpression)
 					g.Expect(fetched.Labels[constant.AppManagedByLabelKey]).Should(Equal(dptypes.AppName))
@@ -192,6 +192,57 @@ var _ = Describe("Backup Schedule Controller", func() {
 				})).Should(Succeed())
 			})
 		})
+
+		Context("create a backup schedule with parameters", func() {
+			const (
+				scheduleName = "test"
+			)
+			BeforeEach(func() {
+				By("set backup parameters and schema in acitionSet")
+				testdp.MockActionSetWithSchema(&testCtx, actionSet)
+			})
+			It("with parameters", func() {
+				By("creating a backupSchedule with invalid parameters")
+				backupSchedule := testdp.NewFakeBackupSchedule(&testCtx, func(schedule *dpv1alpha1.BackupSchedule) {
+					schedule.Spec.Schedules[1].BackupMethod = testdp.BackupMethodName
+					schedule.Spec.Schedules[1].Name = scheduleName
+					schedule.Spec.Schedules[1].Parameters = testdp.InvalidParameters
+				})
+				backupScheduleKey := client.ObjectKeyFromObject(backupSchedule)
+				By("the backupSchedule should fail ")
+				Eventually(testapps.CheckObj(&testCtx, backupScheduleKey, func(g Gomega, fetched *dpv1alpha1.BackupSchedule) {
+					g.Expect(fetched.Status.Phase).NotTo(Equal(dpv1alpha1.BackupSchedulePhaseAvailable))
+				})).Should(Succeed())
+				By("set valid parameters")
+				Expect(testapps.ChangeObj(&testCtx, backupSchedule, func(bs *dpv1alpha1.BackupSchedule) {
+					backupSchedule.Spec.Schedules[1].Parameters = testdp.TestParameters
+				})).Should(Succeed())
+				By("checking backupSchedule status, should be available")
+				Eventually(testapps.CheckObj(&testCtx, backupScheduleKey, func(g Gomega, fetched *dpv1alpha1.BackupSchedule) {
+					g.Expect(fetched.Status.Phase).To(Equal(dpv1alpha1.BackupSchedulePhaseAvailable))
+				})).Should(Succeed())
+
+				By("checking cronjob, should not exist because all schedule policies of methods are disabled")
+				Eventually(testapps.CheckObjExists(&testCtx, getCronjobKey(backupSchedule, testdp.BackupMethodName, ""),
+					&batchv1.CronJob{}, false)).Should(Succeed())
+				Eventually(testapps.CheckObjExists(&testCtx, getCronjobKey(backupSchedule, testdp.BackupMethodName, scheduleName),
+					&batchv1.CronJob{}, false)).Should(Succeed())
+				By(fmt.Sprintf("enabling %s method schedule", testdp.BackupMethodName))
+				testdp.EnableBackupSchedule(&testCtx, backupSchedule, testdp.BackupMethodName)
+
+				By("checking cronjob, should exist one cronjob to create backup")
+				Eventually(testapps.CheckObj(&testCtx, getCronjobKey(backupSchedule, testdp.BackupMethodName, scheduleName), func(g Gomega, fetched *batchv1.CronJob) {
+					g.Expect(fetched.Spec.StartingDeadlineSeconds).ShouldNot(BeNil())
+					g.Expect(*fetched.Spec.StartingDeadlineSeconds).To(Equal(getStartingDeadlineSeconds(backupSchedule)))
+					g.Expect(fetched.Spec.JobTemplate.Spec.Template.Spec.ServiceAccountName).To(Equal(viper.GetString(dptypes.CfgKeyWorkerServiceAccountName)))
+					By("check parameters manifest")
+					g.Expect(fetched.Spec.JobTemplate.Spec.Template.Spec.Containers).Should(HaveLen(1))
+					args := fetched.Spec.JobTemplate.Spec.Template.Spec.Containers[0].Args
+					g.Expect(args).Should(HaveLen(1))
+					g.Expect(args[0]).Should(ContainSubstring(`  parameters: [{"name":"testString","value":"stringValue"},{"name":"testArray","value":"v1,v2"}]`))
+				})).Should(Succeed())
+			})
+		})
 	})
 })
 
diff --git a/controllers/dataprotection/restore_controller_test.go b/controllers/dataprotection/restore_controller_test.go
index b0359897ff7..e5cd5c9e944 100644
--- a/controllers/dataprotection/restore_controller_test.go
+++ b/controllers/dataprotection/restore_controller_test.go
@@ -26,7 +26,6 @@ import (
 
 	. "github.com/onsi/ginkgo/v2"
 	. "github.com/onsi/gomega"
-
 	batchv1 "k8s.io/api/batch/v1"
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/meta"
@@ -235,7 +234,28 @@ var _ = Describe("Restore Controller test", func() {
 			By("mock jobs are completed and wait for restore is completed")
 			mockAndCheckRestoreCompleted(restore)
 		}
+		checkJobParametersEnv := func(restore *dpv1alpha1.Restore) {
 
+			By("check parameters env in restore jobs")
+			jobList := &batchv1.JobList{}
+			Expect(k8sClient.List(ctx, jobList,
+				client.MatchingLabels{dprestore.DataProtectionRestoreLabelKey: restore.Name},
+				client.InNamespace(testCtx.DefaultNamespace))).Should(Succeed())
+			for _, job := range jobList.Items {
+				Expect(len(job.Spec.Template.Spec.Containers)).ShouldNot(BeZero())
+				for _, c := range job.Spec.Template.Spec.Containers {
+					count := 0
+					for _, env := range c.Env {
+						for _, param := range testdp.TestParameters {
+							if param.Name == env.Name && param.Value == env.Value {
+								count++
+							}
+						}
+					}
+					Expect(count).To(Equal(len(testdp.TestParameters)))
+				}
+			}
+		}
 		Context("with restore fails", func() {
 			It("test restore is Failed when backup is not completed", func() {
 				By("expect for restore is Failed ")
@@ -307,7 +327,25 @@ var _ = Describe("Restore Controller test", func() {
 			It("test volumeClaimsTemplate when startingIndex is 1", func() {
 				testRestoreWithVolumeClaimsTemplate(2, 1)
 			})
+			It("test restore parameters", func() {
+				By("set schema and parameters in actionSet")
+				testdp.MockActionSetWithSchema(&testCtx, actionSet)
+				replicas := 3
+				startingIndex := 0
+				restore := initResourcesAndWaitRestore(true, false, false, dpv1alpha1.RestorePhaseRunning,
+					func(f *testdp.MockRestoreFactory) {
+						f.SetVolumeClaimsTemplate(testdp.MysqlTemplateName, testdp.DataVolumeName,
+							testdp.DataVolumeMountPath, "", int32(replicas), int32(startingIndex), nil)
+						// Note: should ignore this policy when podSelectionStrategy is Any of the source target.
+						f.SetPrepareDataRequiredPolicy(dpv1alpha1.OneToOneRestorePolicy, "")
+						f.SetParameters(testdp.TestParameters)
+					}, nil)
 
+				By("expect restore jobs and pvcs are created")
+				checkJobAndPVCSCount(restore, replicas, replicas, startingIndex)
+				By("expect parameters env in restore jobs")
+				checkJobParametersEnv(restore)
+			})
 			It("test volumeClaimsTemplate when volumeClaimRestorePolicy is Serial", func() {
 				replicas := 2
 				startingIndex := 1
@@ -509,7 +547,6 @@ var _ = Describe("Restore Controller test", func() {
 					}
 					return corev1.EnvVar{}
 				}
-
 				By("wait for creating two exec jobs with the matchLabels")
 				a := testapps.List(&testCtx, generics.JobSignature,
 					client.MatchingLabels{dprestore.DataProtectionRestoreLabelKey: restore.Name},
@@ -532,6 +569,28 @@ var _ = Describe("Restore Controller test", func() {
 				})).Should(Succeed())
 
 			})
+			It("test parameters env", func() {
+				By("set schema and parameters in actionSet")
+				testdp.MockActionSetWithSchema(&testCtx, actionSet)
+				By("remove the prepareData stage for testing post ready actions")
+				Expect(testapps.ChangeObj(&testCtx, actionSet, func(set *dpv1alpha1.ActionSet) {
+					set.Spec.Restore.PrepareData = nil
+				})).Should(Succeed())
+
+				matchLabels := map[string]string{
+					constant.AppInstanceLabelKey: testdp.ClusterName,
+				}
+
+				restore := initResourcesAndWaitRestore(true, false, false, dpv1alpha1.RestorePhaseRunning,
+					func(f *testdp.MockRestoreFactory) {
+						f.SetJobActionConfig(matchLabels).SetExecActionConfig(matchLabels)
+						f.SetParameters(testdp.TestParameters)
+					}, func(b *dpv1alpha1.Backup) {
+						b.Status.Target.ConnectionCredential = nil
+					})
+				By("expect parameters env in restore jobs")
+				checkJobParametersEnv(restore)
+			})
 		})
 
 		Context("test cross namespace", func() {
diff --git a/deploy/helm/crds/dataprotection.kubeblocks.io_actionsets.yaml b/deploy/helm/crds/dataprotection.kubeblocks.io_actionsets.yaml
index 40a85d26a67..71b5f38182c 100644
--- a/deploy/helm/crds/dataprotection.kubeblocks.io_actionsets.yaml
+++ b/deploy/helm/crds/dataprotection.kubeblocks.io_actionsets.yaml
@@ -269,6 +269,11 @@ spec:
                     - command
                     - image
                     type: object
+                  withParameters:
+                    description: Specifies the parameters used by the backup action
+                    items:
+                      type: string
+                    type: array
                 type: object
               backupType:
                 allOf:
@@ -457,6 +462,21 @@ spec:
                   type: object
                 type: array
                 x-kubernetes-preserve-unknown-fields: true
+              parametersSchema:
+                description: Specifies the schema of parameters in backups and restores
+                  before their usage.
+                properties:
+                  openAPIV3Schema:
+                    description: |-
+                      Defines the schema for parameters using the OpenAPI v3.
+                      The supported property types include:
+                      - string
+                      - number
+                      - integer
+                      - array: Note that only items of string type are supported.
+                    type: object
+                    x-kubernetes-preserve-unknown-fields: true
+                type: object
               restore:
                 description: Specifies the restore action.
                 properties:
@@ -568,6 +588,11 @@ spec:
                     - command
                     - image
                     type: object
+                  withParameters:
+                    description: Specifies the parameters used by the restore action
+                    items:
+                      type: string
+                    type: array
                 type: object
             required:
             - backupType
diff --git a/deploy/helm/crds/dataprotection.kubeblocks.io_backuppolicytemplates.yaml b/deploy/helm/crds/dataprotection.kubeblocks.io_backuppolicytemplates.yaml
index 7cd65a409c2..ba430dc6ac7 100644
--- a/deploy/helm/crds/dataprotection.kubeblocks.io_backuppolicytemplates.yaml
+++ b/deploy/helm/crds/dataprotection.kubeblocks.io_backuppolicytemplates.yaml
@@ -359,6 +359,32 @@ spec:
                       description: Specifies whether the backup schedule is enabled
                         or not.
                       type: boolean
+                    name:
+                      description: |-
+                        Specifies the name of the schedule. Names cannot be duplicated.
+                        If the name is empty, it will be considered the same as the value of the backupMethod below.
+                      type: string
+                    parameters:
+                      description: |-
+                        Specifies a list of name-value pairs representing parameters and their corresponding values.
+                        Parameters match the schema specified in the `actionset.spec.parametersSchema`
+                      items:
+                        properties:
+                          name:
+                            description: Represents the name of the parameter.
+                            type: string
+                          value:
+                            description: Represents the parameter values.
+                            type: string
+                        required:
+                        - name
+                        - value
+                        type: object
+                      maxItems: 128
+                      type: array
+                      x-kubernetes-list-map-keys:
+                      - name
+                      x-kubernetes-list-type: map
                     retentionPeriod:
                       default: 7d
                       description: "Determines the duration for which the backup should
diff --git a/deploy/helm/crds/dataprotection.kubeblocks.io_backups.yaml b/deploy/helm/crds/dataprotection.kubeblocks.io_backups.yaml
index a2d196267d1..23060c13a49 100644
--- a/deploy/helm/crds/dataprotection.kubeblocks.io_backups.yaml
+++ b/deploy/helm/crds/dataprotection.kubeblocks.io_backups.yaml
@@ -110,6 +110,30 @@ spec:
                     the backup CR but retaining the backup contents in backup repository.
                     The current implementation only prevent accidental deletion of backup data.
                 type: string
+              parameters:
+                description: |-
+                  Specifies a list of name-value pairs representing parameters and their corresponding values.
+                  Parameters match the schema specified in the `actionset.spec.parametersSchema`
+                items:
+                  properties:
+                    name:
+                      description: Represents the name of the parameter.
+                      type: string
+                    value:
+                      description: Represents the parameter values.
+                      type: string
+                  required:
+                  - name
+                  - value
+                  type: object
+                maxItems: 128
+                type: array
+                x-kubernetes-list-map-keys:
+                - name
+                x-kubernetes-list-type: map
+                x-kubernetes-validations:
+                - message: forbidden to update spec.parameters
+                  rule: self == oldSelf
               parentBackupName:
                 description: Determines the parent backup name for incremental or
                   differential backup.
@@ -130,6 +154,9 @@ spec:
             - backupMethod
             - backupPolicyName
             type: object
+            x-kubernetes-validations:
+            - message: forbidden to update spec.parameters
+              rule: has(oldSelf.parameters) == has(self.parameters)
           status:
             description: BackupStatus defines the observed state of Backup.
             properties:
diff --git a/deploy/helm/crds/dataprotection.kubeblocks.io_backupschedules.yaml b/deploy/helm/crds/dataprotection.kubeblocks.io_backupschedules.yaml
index f702df13a54..bd81a15826b 100644
--- a/deploy/helm/crds/dataprotection.kubeblocks.io_backupschedules.yaml
+++ b/deploy/helm/crds/dataprotection.kubeblocks.io_backupschedules.yaml
@@ -72,6 +72,32 @@ spec:
                       description: Specifies whether the backup schedule is enabled
                         or not.
                       type: boolean
+                    name:
+                      description: |-
+                        Specifies the name of the schedule. Names cannot be duplicated.
+                        If the name is empty, it will be considered the same as the value of the backupMethod below.
+                      type: string
+                    parameters:
+                      description: |-
+                        Specifies a list of name-value pairs representing parameters and their corresponding values.
+                        Parameters match the schema specified in the `actionset.spec.parametersSchema`
+                      items:
+                        properties:
+                          name:
+                            description: Represents the name of the parameter.
+                            type: string
+                          value:
+                            description: Represents the parameter values.
+                            type: string
+                        required:
+                        - name
+                        - value
+                        type: object
+                      maxItems: 128
+                      type: array
+                      x-kubernetes-list-map-keys:
+                      - name
+                      x-kubernetes-list-type: map
                     retentionPeriod:
                       default: 7d
                       description: "Determines the duration for which the backup should
diff --git a/deploy/helm/crds/dataprotection.kubeblocks.io_restores.yaml b/deploy/helm/crds/dataprotection.kubeblocks.io_restores.yaml
index a4d6823c12a..199ea02288d 100644
--- a/deploy/helm/crds/dataprotection.kubeblocks.io_restores.yaml
+++ b/deploy/helm/crds/dataprotection.kubeblocks.io_restores.yaml
@@ -270,6 +270,30 @@ spec:
                   type: object
                 type: array
                 x-kubernetes-preserve-unknown-fields: true
+              parameters:
+                description: |-
+                  Specifies a list of name-value pairs representing parameters and their corresponding values.
+                  Parameters match the schema specified in the `actionset.spec.parametersSchema`
+                items:
+                  properties:
+                    name:
+                      description: Represents the name of the parameter.
+                      type: string
+                    value:
+                      description: Represents the parameter values.
+                      type: string
+                  required:
+                  - name
+                  - value
+                  type: object
+                maxItems: 128
+                type: array
+                x-kubernetes-list-map-keys:
+                - name
+                x-kubernetes-list-type: map
+                x-kubernetes-validations:
+                - message: forbidden to update spec.parameters
+                  rule: self == oldSelf
               prepareDataConfig:
                 description: |-
                   Configuration for the action of "prepareData" phase, including the persistent volume claims
@@ -2414,6 +2438,9 @@ spec:
             required:
             - backup
             type: object
+            x-kubernetes-validations:
+            - message: forbidden to update spec.parameters
+              rule: has(oldSelf.parameters) == has(self.parameters)
           status:
             description: RestoreStatus defines the observed state of Restore
             properties:
diff --git a/deploy/helm/crds/operations.kubeblocks.io_opsrequests.yaml b/deploy/helm/crds/operations.kubeblocks.io_opsrequests.yaml
index 40dfb75b258..49398da5161 100755
--- a/deploy/helm/crds/operations.kubeblocks.io_opsrequests.yaml
+++ b/deploy/helm/crds/operations.kubeblocks.io_opsrequests.yaml
@@ -91,6 +91,30 @@ spec:
                     - Delete
                     - Retain
                     type: string
+                  parameters:
+                    description: |-
+                      Specifies a list of name-value pairs representing parameters and their corresponding values.
+                      Parameters match the schema specified in the `actionset.spec.parametersSchema`
+                    items:
+                      properties:
+                        name:
+                          description: Represents the name of the parameter.
+                          type: string
+                        value:
+                          description: Represents the parameter values.
+                          type: string
+                      required:
+                      - name
+                      - value
+                      type: object
+                    maxItems: 128
+                    type: array
+                    x-kubernetes-list-map-keys:
+                    - name
+                    x-kubernetes-list-type: map
+                    x-kubernetes-validations:
+                    - message: forbidden to update parameters
+                      rule: self == oldSelf
                   parentBackupName:
                     description: If the specified BackupMethod is incremental, `parentBackupName`
                       is required.
@@ -121,6 +145,9 @@ spec:
                       Otherwise, only the Backup custom resource will be deleted.
                     type: string
                 type: object
+                x-kubernetes-validations:
+                - message: forbidden to update backup.parameters
+                  rule: has(oldSelf.parameters) == has(self.parameters)
               cancel:
                 description: |-
                   Indicates whether the current operation should be canceled and terminated gracefully if it's in the
@@ -4407,6 +4434,30 @@ spec:
                       type: object
                     type: array
                     x-kubernetes-preserve-unknown-fields: true
+                  parameters:
+                    description: |-
+                      Specifies a list of name-value pairs representing parameters and their corresponding values.
+                      Parameters match the schema specified in the `actionset.spec.parametersSchema`
+                    items:
+                      properties:
+                        name:
+                          description: Represents the name of the parameter.
+                          type: string
+                        value:
+                          description: Represents the parameter values.
+                          type: string
+                      required:
+                      - name
+                      - value
+                      type: object
+                    maxItems: 128
+                    type: array
+                    x-kubernetes-list-map-keys:
+                    - name
+                    x-kubernetes-list-type: map
+                    x-kubernetes-validations:
+                    - message: forbidden to update parameters
+                      rule: self == oldSelf
                   restorePointInTime:
                     description: |-
                       Specifies the point in time to which the restore should be performed.
@@ -4433,6 +4484,9 @@ spec:
                 required:
                 - backupName
                 type: object
+                x-kubernetes-validations:
+                - message: forbidden to update restore.parameters
+                  rule: has(oldSelf.parameters) == has(self.parameters)
               start:
                 description: Lists Components to be started. If empty, all components
                   will be started.
diff --git a/docs/developer_docs/api-reference/backup.md b/docs/developer_docs/api-reference/backup.md
index 908897949d3..e4a3d1c1773 100644
--- a/docs/developer_docs/api-reference/backup.md
+++ b/docs/developer_docs/api-reference/backup.md
@@ -108,6 +108,20 @@ BackupType
 </tr>
 <tr>
 <td>
+<code>parametersSchema</code><br/>
+<em>
+<a href="#dataprotection.kubeblocks.io/v1alpha1.ActionSetParametersSchema">
+ActionSetParametersSchema
+</a>
+</em>
+</td>
+<td>
+<em>(Optional)</em>
+<p>Specifies the schema of parameters in backups and restores before their usage.</p>
+</td>
+</tr>
+<tr>
+<td>
 <code>env</code><br/>
 <em>
 <a href="https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.25/#envvar-v1-core">
@@ -320,6 +334,21 @@ string
 <p>Determines the parent backup name for incremental or differential backup.</p>
 </td>
 </tr>
+<tr>
+<td>
+<code>parameters</code><br/>
+<em>
+<a href="#dataprotection.kubeblocks.io/v1alpha1.ParameterPair">
+[]ParameterPair
+</a>
+</em>
+</td>
+<td>
+<em>(Optional)</em>
+<p>Specifies a list of name-value pairs representing parameters and their corresponding values.
+Parameters match the schema specified in the <code>actionset.spec.parametersSchema</code></p>
+</td>
+</tr>
 </table>
 </td>
 </tr>
@@ -976,6 +1005,21 @@ int32
 <p>Specifies the number of retries before marking the restore failed.</p>
 </td>
 </tr>
+<tr>
+<td>
+<code>parameters</code><br/>
+<em>
+<a href="#dataprotection.kubeblocks.io/v1alpha1.ParameterPair">
+[]ParameterPair
+</a>
+</em>
+</td>
+<td>
+<em>(Optional)</em>
+<p>Specifies a list of name-value pairs representing parameters and their corresponding values.
+Parameters match the schema specified in the <code>actionset.spec.parametersSchema</code></p>
+</td>
+</tr>
 </table>
 </td>
 </tr>
@@ -1236,6 +1280,42 @@ the BackupController.</p>
 </td>
 </tr></tbody>
 </table>
+<h3 id="dataprotection.kubeblocks.io/v1alpha1.ActionSetParametersSchema">ActionSetParametersSchema
+</h3>
+<p>
+(<em>Appears on:</em><a href="#dataprotection.kubeblocks.io/v1alpha1.ActionSetSpec">ActionSetSpec</a>)
+</p>
+<div>
+</div>
+<table>
+<thead>
+<tr>
+<th>Field</th>
+<th>Description</th>
+</tr>
+</thead>
+<tbody>
+<tr>
+<td>
+<code>openAPIV3Schema</code><br/>
+<em>
+<a href="https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.25/#jsonschemaprops-v1-apiextensions-k8s-io">
+Kubernetes api extensions v1.JSONSchemaProps
+</a>
+</em>
+</td>
+<td>
+<em>(Optional)</em>
+<p>Defines the schema for parameters using the OpenAPI v3.
+The supported property types include:
+- string
+- number
+- integer
+- array: Note that only items of string type are supported.</p>
+</td>
+</tr>
+</tbody>
+</table>
 <h3 id="dataprotection.kubeblocks.io/v1alpha1.ActionSetSpec">ActionSetSpec
 </h3>
 <p>
@@ -1274,6 +1354,20 @@ BackupType
 </tr>
 <tr>
 <td>
+<code>parametersSchema</code><br/>
+<em>
+<a href="#dataprotection.kubeblocks.io/v1alpha1.ActionSetParametersSchema">
+ActionSetParametersSchema
+</a>
+</em>
+</td>
+<td>
+<em>(Optional)</em>
+<p>Specifies the schema of parameters in backups and restores before their usage.</p>
+</td>
+</tr>
+<tr>
+<td>
 <code>env</code><br/>
 <em>
 <a href="https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.25/#envvar-v1-core">
@@ -1707,6 +1801,18 @@ BaseJobActionSpec
 Note: The preDelete action job will ignore the env/envFrom.</p>
 </td>
 </tr>
+<tr>
+<td>
+<code>withParameters</code><br/>
+<em>
+[]string
+</em>
+</td>
+<td>
+<em>(Optional)</em>
+<p>Specifies the parameters used by the backup action</p>
+</td>
+</tr>
 </tbody>
 </table>
 <h3 id="dataprotection.kubeblocks.io/v1alpha1.BackupDataActionSpec">BackupDataActionSpec
@@ -3137,6 +3243,21 @@ string
 <p>Determines the parent backup name for incremental or differential backup.</p>
 </td>
 </tr>
+<tr>
+<td>
+<code>parameters</code><br/>
+<em>
+<a href="#dataprotection.kubeblocks.io/v1alpha1.ParameterPair">
+[]ParameterPair
+</a>
+</em>
+</td>
+<td>
+<em>(Optional)</em>
+<p>Specifies a list of name-value pairs representing parameters and their corresponding values.
+Parameters match the schema specified in the <code>actionset.spec.parametersSchema</code></p>
+</td>
+</tr>
 </tbody>
 </table>
 <h3 id="dataprotection.kubeblocks.io/v1alpha1.BackupStatus">BackupStatus
@@ -4320,6 +4441,45 @@ The default value is empty.</p>
 </tr>
 </tbody>
 </table>
+<h3 id="dataprotection.kubeblocks.io/v1alpha1.ParameterPair">ParameterPair
+</h3>
+<p>
+(<em>Appears on:</em><a href="#dataprotection.kubeblocks.io/v1alpha1.BackupSpec">BackupSpec</a>, <a href="#dataprotection.kubeblocks.io/v1alpha1.RestoreSpec">RestoreSpec</a>, <a href="#dataprotection.kubeblocks.io/v1alpha1.SchedulePolicy">SchedulePolicy</a>)
+</p>
+<div>
+</div>
+<table>
+<thead>
+<tr>
+<th>Field</th>
+<th>Description</th>
+</tr>
+</thead>
+<tbody>
+<tr>
+<td>
+<code>name</code><br/>
+<em>
+string
+</em>
+</td>
+<td>
+<p>Represents the name of the parameter.</p>
+</td>
+</tr>
+<tr>
+<td>
+<code>value</code><br/>
+<em>
+string
+</em>
+</td>
+<td>
+<p>Represents the parameter values.</p>
+</td>
+</tr>
+</tbody>
+</table>
 <h3 id="dataprotection.kubeblocks.io/v1alpha1.ParametersSchema">ParametersSchema
 </h3>
 <p>
@@ -4871,6 +5031,18 @@ bool
 <p>Determines if a base backup is required during restoration.</p>
 </td>
 </tr>
+<tr>
+<td>
+<code>withParameters</code><br/>
+<em>
+[]string
+</em>
+</td>
+<td>
+<em>(Optional)</em>
+<p>Specifies the parameters used by the restore action</p>
+</td>
+</tr>
 </tbody>
 </table>
 <h3 id="dataprotection.kubeblocks.io/v1alpha1.RestoreActionStatus">RestoreActionStatus
@@ -5096,6 +5268,21 @@ int32
 <p>Specifies the number of retries before marking the restore failed.</p>
 </td>
 </tr>
+<tr>
+<td>
+<code>parameters</code><br/>
+<em>
+<a href="#dataprotection.kubeblocks.io/v1alpha1.ParameterPair">
+[]ParameterPair
+</a>
+</em>
+</td>
+<td>
+<em>(Optional)</em>
+<p>Specifies a list of name-value pairs representing parameters and their corresponding values.
+Parameters match the schema specified in the <code>actionset.spec.parametersSchema</code></p>
+</td>
+</tr>
 </tbody>
 </table>
 <h3 id="dataprotection.kubeblocks.io/v1alpha1.RestoreStage">RestoreStage
@@ -5574,6 +5761,19 @@ bool
 </tr>
 <tr>
 <td>
+<code>name</code><br/>
+<em>
+string
+</em>
+</td>
+<td>
+<em>(Optional)</em>
+<p>Specifies the name of the schedule. Names cannot be duplicated.
+If the name is empty, it will be considered the same as the value of the backupMethod below.</p>
+</td>
+</tr>
+<tr>
+<td>
 <code>backupMethod</code><br/>
 <em>
 string
@@ -5620,6 +5820,21 @@ Sample duration format:</p>
 <p>You can also combine the above durations. For example: 30d12h30m</p>
 </td>
 </tr>
+<tr>
+<td>
+<code>parameters</code><br/>
+<em>
+<a href="#dataprotection.kubeblocks.io/v1alpha1.ParameterPair">
+[]ParameterPair
+</a>
+</em>
+</td>
+<td>
+<em>(Optional)</em>
+<p>Specifies a list of name-value pairs representing parameters and their corresponding values.
+Parameters match the schema specified in the <code>actionset.spec.parametersSchema</code></p>
+</td>
+</tr>
 </tbody>
 </table>
 <h3 id="dataprotection.kubeblocks.io/v1alpha1.ScheduleStatus">ScheduleStatus
diff --git a/pkg/common/openapiv3schema.go b/pkg/common/openapiv3schema.go
index 53d1c50a3de..f684c63739f 100644
--- a/pkg/common/openapiv3schema.go
+++ b/pkg/common/openapiv3schema.go
@@ -54,7 +54,7 @@ func ValidateDataWithSchema(openAPIV3Schema *apiextensionsv1.JSONSchemaProps, da
 	return nil
 }
 
-func CoverStringToInterfaceBySchemaType(openAPIV3Schema *apiextensionsv1.JSONSchemaProps, input map[string]string) (map[string]interface{}, error) {
+func ConvertStringToInterfaceBySchemaType(openAPIV3Schema *apiextensionsv1.JSONSchemaProps, input map[string]string) (map[string]interface{}, error) {
 	out := map[string]interface{}{}
 	properties := openAPIV3Schema.Properties
 	covertError := func(key string, err error) error {
diff --git a/pkg/constant/backup.go b/pkg/constant/backup.go
index 7fdf155df9b..9ded2714778 100755
--- a/pkg/constant/backup.go
+++ b/pkg/constant/backup.go
@@ -26,6 +26,7 @@ const (
 	DoReadyRestoreAfterClusterRunning = "doReadyRestoreAfterClusterRunning"
 	RestoreTimeKeyForRestore          = "restoreTime"
 	EnvForRestore                     = "restoreEnv"
+	ParametersForRestore              = "restoreParameters"
 	ConnectionPassword                = "connectionPassword"
 	EncryptedSystemAccounts           = "encryptedSystemAccounts"
 )
diff --git a/pkg/controller/builder/builder_component.go b/pkg/controller/builder/builder_component.go
index 78d44f6440d..ce41b00d655 100644
--- a/pkg/controller/builder/builder_component.go
+++ b/pkg/controller/builder/builder_component.go
@@ -20,9 +20,10 @@ along with this program.  If not, see <http://www.gnu.org/licenses/>.
 package builder
 
 import (
-	appsv1 "github.com/apecloud/kubeblocks/apis/apps/v1"
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/util/intstr"
+
+	appsv1 "github.com/apecloud/kubeblocks/apis/apps/v1"
 )
 
 type ComponentBuilder struct {
diff --git a/pkg/controller/plan/restore.go b/pkg/controller/plan/restore.go
index 813ecab3db4..de8861bc352 100644
--- a/pkg/controller/plan/restore.go
+++ b/pkg/controller/plan/restore.go
@@ -57,6 +57,7 @@ type RestoreManager struct {
 	namespace                         string
 	restoreTime                       string
 	env                               []corev1.EnvVar
+	parameters                        []dpv1alpha1.ParameterPair
 	volumeRestorePolicy               dpv1alpha1.VolumeClaimRestorePolicy
 	doReadyRestoreAfterClusterRunning bool
 	startingIndex                     int32
@@ -212,6 +213,7 @@ func (r *RestoreManager) BuildPrepareDataRestore(comp *component.SynthesizedComp
 			},
 			RestoreTime: r.restoreTime,
 			Env:         r.env,
+			Parameters:  r.parameters,
 			PrepareDataConfig: &dpv1alpha1.PrepareDataConfig{
 				RequiredPolicyForAllPodSelection: r.buildRequiredPolicy(sourceTarget),
 				SchedulingSpec:                   schedulingSpec,
@@ -245,6 +247,7 @@ func (r *RestoreManager) DoPostReady(comp *component.SynthesizedComponent,
 			},
 			RestoreTime: r.restoreTime,
 			Env:         r.env,
+			Parameters:  r.parameters,
 			ReadyConfig: &dpv1alpha1.ReadyConfig{
 				ExecAction: &dpv1alpha1.ExecAction{
 					Target: dpv1alpha1.ExecActionTarget{
@@ -360,6 +363,11 @@ func (r *RestoreManager) initFromAnnotation(synthesizedComponent *component.Synt
 			return nil, err
 		}
 	}
+	if parameters := backupSource[constant.ParametersForRestore]; len(parameters) > 0 {
+		if err = json.Unmarshal([]byte(parameters), &r.parameters); err != nil {
+			return nil, err
+		}
+	}
 	return GetBackupFromClusterAnnotation(r.Ctx, r.Client, backupSource, synthesizedComponent.Name, r.Cluster.Namespace)
 }
 
diff --git a/pkg/dataprotection/backup/request.go b/pkg/dataprotection/backup/request.go
index 47ed59fd25d..93b72b4cfcd 100644
--- a/pkg/dataprotection/backup/request.go
+++ b/pkg/dataprotection/backup/request.go
@@ -352,6 +352,7 @@ func (r *Request) BuildJobActionPodSpec(targetPod *corev1.Pod,
 		envVars = append(envVars, envFromTarget...)
 		if r.ActionSet != nil {
 			envVars = append(envVars, r.ActionSet.Spec.Env...)
+			envVars = append(envVars, utils.BuildEnvByParameters(r.Backup.Spec.Parameters)...)
 		}
 		// build envs for kb cluster
 		setKBClusterEnv := func(labelKey, envName string) {
diff --git a/pkg/dataprotection/backup/scheduler.go b/pkg/dataprotection/backup/scheduler.go
index d94e10bf066..f2971d94fb7 100644
--- a/pkg/dataprotection/backup/scheduler.go
+++ b/pkg/dataprotection/backup/scheduler.go
@@ -69,23 +69,39 @@ func (s *Scheduler) Schedule() error {
 
 // validate validates the backup schedule.
 func (s *Scheduler) validate() error {
-	methodInBackupPolicy := func(name string) bool {
-		for _, method := range s.BackupPolicy.Spec.BackupMethods {
+	methodInBackupPolicy := func(name string) *dpv1alpha1.BackupMethod {
+		for i, method := range s.BackupPolicy.Spec.BackupMethods {
 			if method.Name == name {
-				return true
+				return &s.BackupPolicy.Spec.BackupMethods[i]
 			}
 		}
-		return false
+		return nil
+	}
+
+	// validate schedule names
+	if err := dputils.ValidateScheduleNames(s.BackupSchedule.Spec.Schedules); err != nil {
+		return err
 	}
 
 	for _, sp := range s.BackupSchedule.Spec.Schedules {
-		if methodInBackupPolicy(sp.BackupMethod) {
-			continue
+		method := methodInBackupPolicy(sp.BackupMethod)
+		if method == nil {
+			// backup method name is not in backup policy
+			return fmt.Errorf("backup method %s is not in backup policy %s/%s",
+				sp.BackupMethod, s.BackupPolicy.Namespace, s.BackupPolicy.Name)
+		}
+		// validate schedule parameters
+		if len(method.ActionSetName) > 0 && len(sp.Parameters) > 0 {
+			actionSet, err := dputils.GetActionSetByName(s.RequestCtx, s.Client, method.ActionSetName)
+			if err != nil {
+				return err
+			}
+			if err := dputils.ValidateParameters(actionSet, sp.Parameters, true); err != nil {
+				return fmt.Errorf("fails to validate parameters of backupMethod %s: %v", sp.BackupMethod, err)
+			}
 		}
-		// backup method name is not in backup policy
-		return fmt.Errorf("backup method %s is not in backup policy %s/%s",
-			sp.BackupMethod, s.BackupPolicy.Namespace, s.BackupPolicy.Name)
 	}
+
 	return nil
 }
 
@@ -125,7 +141,7 @@ func (s *Scheduler) buildCronJob(schedulePolicy *dpv1alpha1.SchedulePolicy, cron
 	)
 
 	if cronJobName == "" {
-		cronJobName = GenerateCRNameByBackupSchedule(s.BackupSchedule, schedulePolicy.BackupMethod)
+		cronJobName = GenerateCRNameByScheduleNameAndMethod(s.BackupSchedule, schedulePolicy.BackupMethod, schedulePolicy.Name)
 	}
 
 	podSpec, err := s.buildPodSpec(schedulePolicy)
@@ -175,6 +191,10 @@ func (s *Scheduler) buildCronJob(schedulePolicy *dpv1alpha1.SchedulePolicy, cron
 
 func (s *Scheduler) buildPodSpec(schedulePolicy *dpv1alpha1.SchedulePolicy) (*corev1.PodSpec, error) {
 	// TODO(ldm): add backup deletionPolicy
+	parameters, err := BuildParametersManifest(schedulePolicy.Parameters)
+	if err != nil {
+		return nil, err
+	}
 	createBackupCmd := fmt.Sprintf(`
 kubectl create -f - <<EOF
 apiVersion: dataprotection.kubeblocks.io/v1alpha1
@@ -188,11 +208,11 @@ metadata:
 spec:
   backupPolicyName: %s
   backupMethod: %s
-  retentionPeriod: %s
+  retentionPeriod: %s%s
 EOF
 `, s.BackupSchedule.Name, s.generateBackupName(schedulePolicy), s.BackupSchedule.Namespace,
 		s.BackupPolicy.Name, schedulePolicy.BackupMethod,
-		schedulePolicy.RetentionPeriod)
+		schedulePolicy.RetentionPeriod, parameters)
 
 	container := corev1.Container{
 		Name:            "backup-schedule",
@@ -228,7 +248,14 @@ func (s *Scheduler) reconcileCronJob(schedulePolicy *dpv1alpha1.SchedulePolicy)
 	); err != nil {
 		return err
 	} else if len(cronJobList.Items) > 0 {
-		cronJob = &cronJobList.Items[0]
+		// the schedulePolicy name can be empty
+		targetCronJobName := GenerateCRNameByScheduleNameAndMethod(s.BackupSchedule, schedulePolicy.BackupMethod, schedulePolicy.Name)
+		for i, item := range cronJobList.Items {
+			if item.Name == targetCronJobName {
+				cronJob = &cronJobList.Items[i]
+				break
+			}
+		}
 	}
 
 	// schedule is disabled, delete cronjob if exists
@@ -285,6 +312,11 @@ func (s *Scheduler) generateBackupName(schedulePolicy *dpv1alpha1.SchedulePolicy
 	if backupNamePrefix == "" {
 		backupNamePrefix = s.BackupSchedule.Name
 	}
+	// use schedule name to distinguish different schedule policies
+	name := schedulePolicy.GetScheduleName()
+	if len(name) > 0 {
+		backupNamePrefix = fmt.Sprintf("%s-%s", backupNamePrefix, name)
+	}
 	return backupNamePrefix + "-$(date -u +'%Y%m%d%H%M%S')"
 }
 
diff --git a/pkg/dataprotection/backup/utils.go b/pkg/dataprotection/backup/utils.go
index 5735ff2453e..b62a6b9773a 100644
--- a/pkg/dataprotection/backup/utils.go
+++ b/pkg/dataprotection/backup/utils.go
@@ -21,6 +21,7 @@ package backup
 
 import (
 	"context"
+	"encoding/json"
 	"fmt"
 	"path/filepath"
 	"slices"
@@ -177,7 +178,7 @@ func generateBaseCRNameByBackupSchedule(uniqueNameWithBackupSchedule, backupSche
 	return fmt.Sprintf("%s-%s", name, method)
 }
 
-// GenerateCRNameByBackupSchedule generate a CR name which is created by BackupSchedule, such as CronJob Backup.
+// GenerateCRNameByBackupSchedule generate a CR name which is created by BackupSchedule, such as Continuous Backup.
 func GenerateCRNameByBackupSchedule(backupSchedule *dpv1alpha1.BackupSchedule, method string) string {
 	uid := backupSchedule.UID[:8]
 	if len(backupSchedule.OwnerReferences) > 0 {
@@ -187,7 +188,16 @@ func GenerateCRNameByBackupSchedule(backupSchedule *dpv1alpha1.BackupSchedule, m
 	return generateBaseCRNameByBackupSchedule(uniqueNameWithBackupSchedule, backupSchedule.Namespace, method)
 }
 
-// GenerateLegacyCRNameByBackupSchedule generate a legacy CR name which is created by BackupSchedule, such as CronJob Backup.
+// GenerateCRNameByScheduleNameAndMethod generate a CR name which is created by BackupSchedule, such as CronJob.
+func GenerateCRNameByScheduleNameAndMethod(backupSchedule *dpv1alpha1.BackupSchedule, method string, name string) string {
+	suffix := name
+	if len(suffix) == 0 {
+		suffix = method
+	}
+	return GenerateCRNameByBackupSchedule(backupSchedule, suffix)
+}
+
+// GenerateLegacyCRNameByBackupSchedule generate a legacy CR name which is created by BackupSchedule, such as CronJob.
 func GenerateLegacyCRNameByBackupSchedule(backupSchedule *dpv1alpha1.BackupSchedule, method string) string {
 	uniqueNameWithBackupSchedule := fmt.Sprintf("%s-%s", backupSchedule.UID[:8], backupSchedule.Name)
 	return generateBaseCRNameByBackupSchedule(uniqueNameWithBackupSchedule, backupSchedule.Namespace, method)
@@ -302,3 +312,15 @@ func StopStatefulSetsWhenFailed(ctx context.Context, cli client.Client, backup *
 	sts.Spec.Replicas = pointer.Int32(0)
 	return cli.Update(ctx, sts)
 }
+
+func BuildParametersManifest(parameters []dpv1alpha1.ParameterPair) (string, error) {
+	if len(parameters) == 0 {
+		return "", nil
+	}
+	bytes, err := json.Marshal(parameters)
+	if err != nil {
+		return "", err
+	}
+	res := fmt.Sprintf("\n  parameters: %s", string(bytes))
+	return res, nil
+}
diff --git a/pkg/dataprotection/restore/builder.go b/pkg/dataprotection/restore/builder.go
index 54f09d39343..b1c69536cb4 100644
--- a/pkg/dataprotection/restore/builder.go
+++ b/pkg/dataprotection/restore/builder.go
@@ -226,6 +226,10 @@ func (r *restoreJobBuilder) addCommonEnv(sourceTargetPodName string) *restoreJob
 		restoreTime, _ := time.Parse(time.RFC3339, r.restore.Spec.RestoreTime)
 		appendTimeEnv(DPRestoreTime, DPRestoreTimestamp, backup.GetTimeZone(), &metav1.Time{Time: restoreTime})
 	}
+	// append restore parameters env
+	if r.restore != nil {
+		r.env = append(r.env, utils.BuildEnvByParameters(r.restore.Spec.Parameters)...)
+	}
 	// append actionSet env
 	r.env = append(r.env, actionSetEnv...)
 	backupMethod := r.backupSet.Backup.Status.BackupMethod
diff --git a/pkg/dataprotection/restore/utils.go b/pkg/dataprotection/restore/utils.go
index 500f2b1e837..0e9281f023e 100644
--- a/pkg/dataprotection/restore/utils.go
+++ b/pkg/dataprotection/restore/utils.go
@@ -257,6 +257,13 @@ func ValidateAndInitRestoreMGR(reqCtx intctrlutil.RequestCtx,
 		return err
 	}
 
+	// validate restore parameters
+	if backupSet.ActionSet != nil {
+		if err := utils.ValidateParameters(backupSet.ActionSet, restoreMgr.Restore.Spec.Parameters, false); err != nil {
+			return fmt.Errorf("fails to validate parameters with actionset %s: %v", backupSet.ActionSet.Name, err)
+		}
+	}
+
 	// TODO: check if there is permission for cross namespace recovery.
 
 	// check if the backup is completed exclude continuous backup.
@@ -322,7 +329,14 @@ func isTimeInRange(t time.Time, start time.Time, end time.Time) bool {
 	return !t.Before(start) && !t.After(end)
 }
 
-func GetRestoreFromBackupAnnotation(backup *dpv1alpha1.Backup, volumeRestorePolicy, restoreTime string, env []corev1.EnvVar, doReadyRestoreAfterClusterRunning bool) (string, error) {
+func GetRestoreFromBackupAnnotation(
+	backup *dpv1alpha1.Backup,
+	volumeRestorePolicy string,
+	restoreTime string,
+	env []corev1.EnvVar,
+	doReadyRestoreAfterClusterRunning bool,
+	parameters []dpv1alpha1.ParameterPair,
+) (string, error) {
 	componentName := component.GetComponentNameFromObj(backup)
 	if len(componentName) == 0 {
 		return "", intctrlutil.NewFatalError("unable to obtain the name of the component to be recovered, please ensure that Backup.status.componentName exists")
@@ -342,7 +356,13 @@ func GetRestoreFromBackupAnnotation(backup *dpv1alpha1.Backup, volumeRestorePoli
 		}
 		restoreInfoMap[constant.EnvForRestore] = string(bytes)
 	}
-
+	if len(parameters) > 0 {
+		bytes, err := json.Marshal(parameters)
+		if err != nil {
+			return "", err
+		}
+		restoreInfoMap[constant.ParametersForRestore] = string(bytes)
+	}
 	connectionPassword := backup.Annotations[dptypes.ConnectionPasswordAnnotationKey]
 	if connectionPassword != "" {
 		restoreInfoMap[constant.ConnectionPassword] = connectionPassword
diff --git a/pkg/dataprotection/utils/backup.go b/pkg/dataprotection/utils/backup.go
index cef336bee40..3c5e7202b60 100644
--- a/pkg/dataprotection/utils/backup.go
+++ b/pkg/dataprotection/utils/backup.go
@@ -20,6 +20,8 @@ along with this program.  If not, see <http://www.gnu.org/licenses/>.
 package utils
 
 import (
+	"fmt"
+
 	dpv1alpha1 "github.com/apecloud/kubeblocks/apis/dataprotection/v1alpha1"
 	dptypes "github.com/apecloud/kubeblocks/pkg/dataprotection/types"
 	"github.com/apecloud/kubeblocks/pkg/dataprotection/utils/boolptr"
@@ -53,3 +55,19 @@ func GetBackupMethodsFromBackupPolicy(backupPolicyList *dpv1alpha1.BackupPolicyL
 	}
 	return defaultBackupMethod, backupMethodsMap
 }
+
+func ValidateScheduleNames(schedules []dpv1alpha1.SchedulePolicy) error {
+	if len(schedules) == 0 {
+		return nil
+	}
+	nameMap := map[string]struct{}{}
+	for _, sp := range schedules {
+		name := sp.GetScheduleName()
+		// names cannot be duplicated
+		if _, ok := nameMap[name]; ok {
+			return fmt.Errorf("schedule name %s is duplicated", name)
+		}
+		nameMap[name] = struct{}{}
+	}
+	return nil
+}
diff --git a/pkg/dataprotection/utils/envvar.go b/pkg/dataprotection/utils/envvar.go
index e6e990cebe4..f43070a3d54 100644
--- a/pkg/dataprotection/utils/envvar.go
+++ b/pkg/dataprotection/utils/envvar.go
@@ -20,10 +20,11 @@ along with this program.  If not, see <http://www.gnu.org/licenses/>.
 package utils
 
 import (
+	corev1 "k8s.io/api/core/v1"
+
 	dpv1alpha1 "github.com/apecloud/kubeblocks/apis/dataprotection/v1alpha1"
 	intctrlutil "github.com/apecloud/kubeblocks/pkg/controllerutil"
 	dptypes "github.com/apecloud/kubeblocks/pkg/dataprotection/types"
-	corev1 "k8s.io/api/core/v1"
 )
 
 func BuildEnvByTarget(pod *corev1.Pod, credential *dpv1alpha1.ConnectionCredential, containerPort *dpv1alpha1.ContainerPort) ([]corev1.EnvVar, error) {
@@ -72,3 +73,11 @@ func buildEnvBySecretKey(name, secretName, key string) corev1.EnvVar {
 		},
 	}
 }
+
+func BuildEnvByParameters(parameters []dpv1alpha1.ParameterPair) []corev1.EnvVar {
+	env := []corev1.EnvVar{}
+	for _, pair := range parameters {
+		env = append(env, corev1.EnvVar{Name: pair.Name, Value: pair.Value})
+	}
+	return env
+}
diff --git a/pkg/dataprotection/utils/utils.go b/pkg/dataprotection/utils/utils.go
index 3e2703db588..6ebe93e2e3d 100644
--- a/pkg/dataprotection/utils/utils.go
+++ b/pkg/dataprotection/utils/utils.go
@@ -39,6 +39,7 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
 
 	dpv1alpha1 "github.com/apecloud/kubeblocks/apis/dataprotection/v1alpha1"
+	"github.com/apecloud/kubeblocks/pkg/common"
 	"github.com/apecloud/kubeblocks/pkg/constant"
 	intctrlutil "github.com/apecloud/kubeblocks/pkg/controllerutil"
 	dptypes "github.com/apecloud/kubeblocks/pkg/dataprotection/types"
@@ -393,3 +394,48 @@ func GetBackupStatusTarget(backupObj *dpv1alpha1.Backup, sourceTargetName string
 	}
 	return nil
 }
+
+func ValidateParameters(actionSet *dpv1alpha1.ActionSet, parameters []dpv1alpha1.ParameterPair, isBackup bool) error {
+	if len(parameters) == 0 {
+		return nil
+	}
+	if actionSet == nil {
+		return fmt.Errorf("actionSet is empty")
+	}
+	var withParameters []string
+	if isBackup && actionSet.Spec.Backup != nil {
+		withParameters = actionSet.Spec.Backup.WithParameters
+	} else if !isBackup && actionSet.Spec.Restore != nil {
+		withParameters = actionSet.Spec.Restore.WithParameters
+	}
+	if len(withParameters) < len(parameters) {
+		return fmt.Errorf("some parameters are undeclared in withParameters of actionSet %s", actionSet.Name)
+	}
+	// check whether the parameter is declared in withParameters
+	parametersMap := map[string]string{}
+	for _, pair := range parameters {
+		parametersMap[pair.Name] = pair.Value
+	}
+	withParametersMap := map[string]struct{}{}
+	for _, v := range withParameters {
+		withParametersMap[v] = struct{}{}
+	}
+	for k := range parametersMap {
+		if _, ok := withParametersMap[k]; !ok {
+			return fmt.Errorf("parameter %s is undeclared in withParameters of actionSet %s", k, actionSet.Name)
+		}
+	}
+	schema := actionSet.Spec.ParametersSchema
+	if schema == nil || schema.OpenAPIV3Schema == nil || len(schema.OpenAPIV3Schema.Properties) == 0 {
+		return fmt.Errorf("the parametersSchema is invalid in actionSet %s", actionSet.Name)
+	}
+	// convert to type map[string]interface{} and validate the schema
+	params, err := common.ConvertStringToInterfaceBySchemaType(schema.OpenAPIV3Schema, parametersMap)
+	if err != nil {
+		return intctrlutil.NewFatalError(err.Error())
+	}
+	if err = common.ValidateDataWithSchema(schema.OpenAPIV3Schema, params); err != nil {
+		return intctrlutil.NewFatalError(err.Error())
+	}
+	return nil
+}
diff --git a/pkg/operations/backup.go b/pkg/operations/backup.go
index cf3c2fd2e25..0ce281eb221 100644
--- a/pkg/operations/backup.go
+++ b/pkg/operations/backup.go
@@ -149,6 +149,7 @@ func buildBackup(reqCtx intctrlutil.RequestCtx, cli client.Client, opsRequest *o
 		Spec: dpv1alpha1.BackupSpec{
 			BackupPolicyName: backupSpec.BackupPolicyName,
 			BackupMethod:     backupSpec.BackupMethod,
+			Parameters:       backupSpec.Parameters,
 		},
 	}
 
diff --git a/pkg/operations/custom.go b/pkg/operations/custom.go
index e8aab5e30a0..29fc7849b23 100644
--- a/pkg/operations/custom.go
+++ b/pkg/operations/custom.go
@@ -330,7 +330,7 @@ func initOpsDefAndValidate(reqCtx intctrlutil.RequestCtx,
 				return err
 			}
 			// covert to type map[string]interface{}
-			params, err := common.CoverStringToInterfaceBySchemaType(parametersSchema.OpenAPIV3Schema, paramsMap)
+			params, err := common.ConvertStringToInterfaceBySchemaType(parametersSchema.OpenAPIV3Schema, paramsMap)
 			if err != nil {
 				return intctrlutil.NewFatalError(err.Error())
 			}
diff --git a/pkg/operations/restore.go b/pkg/operations/restore.go
index 4efeba60e17..8e597446c87 100644
--- a/pkg/operations/restore.go
+++ b/pkg/operations/restore.go
@@ -193,7 +193,8 @@ func (r RestoreOpsHandler) getClusterObjFromBackup(backup *dpv1alpha1.Backup, op
 	}
 	restoreSpec := opsRequest.Spec.GetRestore()
 	// set the restore annotation to cluster
-	restoreAnnotation, err := restore.GetRestoreFromBackupAnnotation(backup, restoreSpec.VolumeRestorePolicy, restoreSpec.RestorePointInTime, restoreSpec.Env, restoreSpec.DeferPostReadyUntilClusterRunning)
+	restoreAnnotation, err := restore.GetRestoreFromBackupAnnotation(backup, restoreSpec.VolumeRestorePolicy, restoreSpec.RestorePointInTime,
+		restoreSpec.Env, restoreSpec.DeferPostReadyUntilClusterRunning, restoreSpec.Parameters)
 	if err != nil {
 		return nil, err
 	}
diff --git a/pkg/testutil/dataprotection/backup_utils.go b/pkg/testutil/dataprotection/backup_utils.go
index d6db8db3a5f..d2bd90d292b 100644
--- a/pkg/testutil/dataprotection/backup_utils.go
+++ b/pkg/testutil/dataprotection/backup_utils.go
@@ -22,13 +22,11 @@ package dataprotection
 import (
 	"fmt"
 
-	"k8s.io/apimachinery/pkg/api/meta"
-
 	. "github.com/onsi/ginkgo/v2"
 	. "github.com/onsi/gomega"
-
 	batchv1 "k8s.io/api/batch/v1"
 	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
 	"sigs.k8s.io/controller-runtime/pkg/client"
@@ -256,7 +254,6 @@ func EnableBackupSchedule(testCtx *testutil.TestContext,
 		for i := range schedule.Spec.Schedules {
 			if schedule.Spec.Schedules[i].BackupMethod == method {
 				schedule.Spec.Schedules[i].Enabled = boolptr.True()
-				break
 			}
 		}
 	})).Should(Succeed())
diff --git a/pkg/testutil/dataprotection/backuppolicytemplate_factory.go b/pkg/testutil/dataprotection/backuppolicytemplate_factory.go
index 50989006ded..39983a4e789 100644
--- a/pkg/testutil/dataprotection/backuppolicytemplate_factory.go
+++ b/pkg/testutil/dataprotection/backuppolicytemplate_factory.go
@@ -52,12 +52,19 @@ func (f *MockBackupPolicyTemplateFactory) getLastBackupMethod() *dpv1alpha1.Back
 	return &backupMethods[l-1]
 }
 
-func (f *MockBackupPolicyTemplateFactory) AddSchedule(method, schedule, retentionPeriod string, enable bool) *MockBackupPolicyTemplateFactory {
+func (f *MockBackupPolicyTemplateFactory) AddSchedule(
+	method, schedule, retentionPeriod string,
+	enable bool,
+	name string,
+	parameters []dpv1alpha1.ParameterPair,
+) *MockBackupPolicyTemplateFactory {
 	schedulePolicy := dpv1alpha1.SchedulePolicy{
 		Enabled:         &enable,
 		CronExpression:  schedule,
 		BackupMethod:    method,
 		RetentionPeriod: dpv1alpha1.RetentionPeriod(retentionPeriod),
+		Parameters:      parameters,
+		Name:            name,
 	}
 	f.Get().Spec.Schedules = append(f.Get().Spec.Schedules, schedulePolicy)
 	return f
diff --git a/pkg/testutil/dataprotection/constant.go b/pkg/testutil/dataprotection/constant.go
index 08b605da934..66f01692a22 100644
--- a/pkg/testutil/dataprotection/constant.go
+++ b/pkg/testutil/dataprotection/constant.go
@@ -19,6 +19,8 @@ along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
 package dataprotection
 
+import dpv1alpha1 "github.com/apecloud/kubeblocks/apis/dataprotection/v1alpha1"
+
 const (
 	ClusterName   = "test-cluster"
 	ComponentName = "test-comp"
@@ -59,3 +61,30 @@ const (
 	RestoreName       = "test-restore"
 	MysqlTemplateName = "data-mysql-mysql"
 )
+
+const (
+	InvalidParameter    = "invalid"
+	ParameterString     = "testString"
+	ParameterStringType = "string"
+	ParameterArray      = "testArray"
+	ParameterArrayType  = "array"
+)
+
+var (
+	TestParameters = []dpv1alpha1.ParameterPair{
+		{
+			Name:  ParameterString,
+			Value: "stringValue",
+		},
+		{
+			Name:  ParameterArray,
+			Value: "v1,v2",
+		},
+	}
+	InvalidParameters = []dpv1alpha1.ParameterPair{
+		{
+			Name:  "invalid",
+			Value: "invalid",
+		},
+	}
+)
diff --git a/pkg/testutil/dataprotection/restore_factory.go b/pkg/testutil/dataprotection/restore_factory.go
index 80a5098e94c..853afff19b1 100644
--- a/pkg/testutil/dataprotection/restore_factory.go
+++ b/pkg/testutil/dataprotection/restore_factory.go
@@ -112,6 +112,11 @@ func (f *MockRestoreFactory) buildRestoreVolumeClaim(name, volumeSource, mountPa
 	}
 }
 
+func (f *MockRestoreFactory) SetParameters(parameters []dpv1alpha1.ParameterPair) *MockRestoreFactory {
+	f.Get().Spec.Parameters = parameters
+	return f
+}
+
 func (f *MockRestoreFactory) SetVolumeClaimRestorePolicy(policy dpv1alpha1.VolumeClaimRestorePolicy) *MockRestoreFactory {
 	f.initPrepareDataConfig()
 	f.Get().Spec.PrepareDataConfig.VolumeClaimRestorePolicy = policy
diff --git a/pkg/testutil/dataprotection/utils.go b/pkg/testutil/dataprotection/utils.go
index a629eb06590..386aefeccb6 100644
--- a/pkg/testutil/dataprotection/utils.go
+++ b/pkg/testutil/dataprotection/utils.go
@@ -20,12 +20,12 @@ along with this program.  If not, see <http://www.gnu.org/licenses/>.
 package dataprotection
 
 import (
+	vsv1 "github.com/kubernetes-csi/external-snapshotter/client/v6/apis/volumesnapshot/v1"
 	. "github.com/onsi/ginkgo/v2"
 	. "github.com/onsi/gomega"
-
-	vsv1 "github.com/kubernetes-csi/external-snapshotter/client/v6/apis/volumesnapshot/v1"
 	batchv1 "k8s.io/api/batch/v1"
 	corev1 "k8s.io/api/core/v1"
+	apiextensionsv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
 	"sigs.k8s.io/controller-runtime/pkg/client"
@@ -120,8 +120,8 @@ func CreateBackupPolicyTpl(testCtx *testutil.TestContext, compDef string) *dpv1a
 		SetCompDefs(compDef).
 		AddBackupMethod(VSBackupMethodName, true, "").
 		SetBackupMethodVolumes([]string{"data"}).
-		AddSchedule(BackupMethodName, "0 0 * * *", ttl, true).
-		AddSchedule(VSBackupMethodName, "0 0 * * *", ttl, true).
+		AddSchedule(BackupMethodName, "0 0 * * *", ttl, true, "", nil).
+		AddSchedule(VSBackupMethodName, "0 0 * * *", ttl, true, "", nil).
 		Create(testCtx).Get()
 }
 
@@ -149,3 +149,33 @@ func MockRestoreCompleted(testCtx *testutil.TestContext, ml client.MatchingLabel
 		Expect(err).ShouldNot(HaveOccurred())
 	}
 }
+
+func MockActionSetWithSchema(testCtx *testutil.TestContext, actionSet *dpv1alpha1.ActionSet) {
+	Expect(testapps.ChangeObj(testCtx, actionSet, func(as *dpv1alpha1.ActionSet) {
+		as.Spec.ParametersSchema = &dpv1alpha1.ActionSetParametersSchema{
+			OpenAPIV3Schema: &apiextensionsv1.JSONSchemaProps{
+				Properties: map[string]apiextensionsv1.JSONSchemaProps{
+					ParameterString: {
+						Type: ParameterStringType,
+					},
+					ParameterArray: {
+						Type: ParameterArrayType,
+						Items: &apiextensionsv1.JSONSchemaPropsOrArray{
+							Schema: &apiextensionsv1.JSONSchemaProps{
+								Type: ParameterStringType,
+							},
+						},
+					},
+				},
+			},
+		}
+		as.Spec.Backup.WithParameters = []string{ParameterString, ParameterArray}
+		as.Spec.Restore.WithParameters = []string{ParameterString, ParameterArray}
+	})).Should(Succeed())
+	By("the actionSet should be available")
+	Eventually(testapps.CheckObj(testCtx, client.ObjectKeyFromObject(actionSet),
+		func(g Gomega, as *dpv1alpha1.ActionSet) {
+			g.Expect(as.Status.Phase).Should(BeEquivalentTo(dpv1alpha1.AvailablePhase))
+			g.Expect(as.Status.Message).Should(BeEmpty())
+		})).Should(Succeed())
+}