Repository: apecloud/kubeblocks

Issue:
[Features] support to scale out replicas from backup


Hints from issue comments:

Solution patch:
diff --git a/apis/operations/v1alpha1/opsrequest_types.go b/apis/operations/v1alpha1/opsrequest_types.go
index 1eb8451d7..e022284cf 100644
--- a/apis/operations/v1alpha1/opsrequest_types.go
+++ b/apis/operations/v1alpha1/opsrequest_types.go
@@ -460,6 +460,12 @@ type HorizontalScaling struct {
 // ScaleOut defines the configuration for a scale-out operation.
 type ScaleOut struct {
 
+	// FromBackup specifies the configuration for creating new instances from an existing backup.
+	// This is only effective for non-sharding components.
+	// When specified, new instances will be created using data from the specified backup.
+	// +optional
+	FromBackup *FromBackup `json:"fromBackup,omitempty"`
+
 	// Modifies the replicas of the component and instance templates.
 	ReplicaChanger `json:",inline"`
 
@@ -504,6 +510,41 @@ type ReplicaChanger struct {
 	Instances []InstanceReplicasTemplate `json:"instances,omitempty" patchStrategy:"merge,retainKeys" patchMergeKey:"name"`
 }
 
+type FromBackup struct {
+	// Specifies the name of the Backup name.
+	// +kubebuilder:validation:Required
+	Name string `json:"name"`
+
+	// Specifies the namespace of the Backup namespace.
+	// If not specified, the namespace of the OpsRequest will be used.
+	// +kubebuilder:validation:Required
+	Namespace string `json:"namespace,omitempty"`
+
+	// Defines container environment variables for the restore process.
+	// merged with the ones specified in the Backup and ActionSet resources.
+	//
+	// Merge priority: Restore env > Backup env > ActionSet env.
+	//
+	// Purpose: Some databases require different configurations when being restored as a standby
+	// compared to being restored as a primary.
+	// For example, when restoring MySQL as a replica, you need to set `skip_slave_start="ON"` for 5.7
+	// or `skip_replica_start="ON"` for 8.0.
+	// Allowing environment variables to be passed in makes it more convenient to control these behavioral differences
+	// during the restore process.
+	//
+	// +kubebuilder:pruning:PreserveUnknownFields
+	// +optional
+	RestoreEnv []corev1.EnvVar `json:"restoreEnv,omitempty" patchStrategy:"merge" patchMergeKey:"name"`
+
+	// Specifies the point in time to which the restore should be performed.
+	// Supported time formats:
+	//
+	// - RFC3339 format, e.g. "2023-11-25T18:52:53Z"
+	// - A human-readable date-time format, e.g. "Jul 25,2023 18:52:53 UTC+0800"
+	//
+	RestorePointInTime string `json:"restorePointInTime,omitempty"`
+}
+
 // InstanceReplicasTemplate defines the template for instance replicas.
 type InstanceReplicasTemplate struct {
 	// Specifies the name of the instance template.
@@ -764,32 +805,6 @@ type OpsService struct {
 	IPFamilyPolicy *corev1.IPFamilyPolicy `json:"ipFamilyPolicy,omitempty" protobuf:"bytes,17,opt,name=ipFamilyPolicy,casttype=IPFamilyPolicy"`
 }
 
-type RefNamespaceName struct {
-	// Refers to the specific name of the resource.
-	// +optional
-	Name string `json:"name,omitempty"`
-
-	// Refers to the specific namespace of the resource.
-	// +optional
-	Namespace string `json:"namespace,omitempty"`
-}
-
-type BackupRefSpec struct {
-	// Refers to a reference backup that needs to be restored.
-	// +optional
-	Ref RefNamespaceName `json:"ref,omitempty"`
-}
-
-type PointInTimeRefSpec struct {
-	// Refers to the specific time point for restoration, with UTC as the time zone.
-	// +optional
-	Time *metav1.Time `json:"time,omitempty"`
-
-	// Refers to a reference source cluster that needs to be restored.
-	// +optional
-	Ref RefNamespaceName `json:"ref,omitempty"`
-}
-
 type Backup struct {
 	// Specifies the name of the Backup custom resource.
 	//
diff --git a/apis/operations/v1alpha1/opsrequest_validation.go b/apis/operations/v1alpha1/opsrequest_validation.go
index 673d185f2..b5ce78aed 100644
--- a/apis/operations/v1alpha1/opsrequest_validation.go
+++ b/apis/operations/v1alpha1/opsrequest_validation.go
@@ -298,10 +298,12 @@ func (r *OpsRequest) validateHorizontalScaling(ctx context.Context, cli client.C
 				return err
 			}
 		}
-
 	}
 	for _, spec := range cluster.Spec.Shardings {
 		if hScale, ok := hScaleMap[spec.Name]; ok {
+			if hScale.ScaleOut != nil && hScale.ScaleOut.FromBackup != nil {
+				return fmt.Errorf("sharding component %s does not support scaling out from backup", hScale.ComponentName)
+			}
 			// Default values if no limit is found
 			minNum, maxNum := 1, 2048
 			if spec.ShardingDef != "" {
diff --git a/apis/operations/v1alpha1/zz_generated.deepcopy.go b/apis/operations/v1alpha1/zz_generated.deepcopy.go
index 7a8a647c6..a3d0ea12e 100644
--- a/apis/operations/v1alpha1/zz_generated.deepcopy.go
+++ b/apis/operations/v1alpha1/zz_generated.deepcopy.go
@@ -66,22 +66,6 @@ func (in *Backup) DeepCopy() *Backup {
 	return out
 }
 
-// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
-func (in *BackupRefSpec) DeepCopyInto(out *BackupRefSpec) {
-	*out = *in
-	out.Ref = in.Ref
-}
-
-// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new BackupRefSpec.
-func (in *BackupRefSpec) DeepCopy() *BackupRefSpec {
-	if in == nil {
-		return nil
-	}
-	out := new(BackupRefSpec)
-	in.DeepCopyInto(out)
-	return out
-}
-
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *CompletionProbe) DeepCopyInto(out *CompletionProbe) {
 	*out = *in
@@ -223,6 +207,28 @@ func (in *Expose) DeepCopy() *Expose {
 	return out
 }
 
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *FromBackup) DeepCopyInto(out *FromBackup) {
+	*out = *in
+	if in.RestoreEnv != nil {
+		in, out := &in.RestoreEnv, &out.RestoreEnv
+		*out = make([]v1.EnvVar, len(*in))
+		for i := range *in {
+			(*in)[i].DeepCopyInto(&(*out)[i])
+		}
+	}
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new FromBackup.
+func (in *FromBackup) DeepCopy() *FromBackup {
+	if in == nil {
+		return nil
+	}
+	out := new(FromBackup)
+	in.DeepCopyInto(out)
+	return out
+}
+
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *HorizontalScaling) DeepCopyInto(out *HorizontalScaling) {
 	*out = *in
@@ -1103,26 +1109,6 @@ func (in *PodSelector) DeepCopy() *PodSelector {
 	return out
 }
 
-// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
-func (in *PointInTimeRefSpec) DeepCopyInto(out *PointInTimeRefSpec) {
-	*out = *in
-	if in.Time != nil {
-		in, out := &in.Time, &out.Time
-		*out = (*in).DeepCopy()
-	}
-	out.Ref = in.Ref
-}
-
-// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new PointInTimeRefSpec.
-func (in *PointInTimeRefSpec) DeepCopy() *PointInTimeRefSpec {
-	if in == nil {
-		return nil
-	}
-	out := new(PointInTimeRefSpec)
-	in.DeepCopyInto(out)
-	return out
-}
-
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *PreCheckResult) DeepCopyInto(out *PreCheckResult) {
 	*out = *in
@@ -1231,21 +1217,6 @@ func (in *Reconfigure) DeepCopy() *Reconfigure {
 	return out
 }
 
-// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
-func (in *RefNamespaceName) DeepCopyInto(out *RefNamespaceName) {
-	*out = *in
-}
-
-// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new RefNamespaceName.
-func (in *RefNamespaceName) DeepCopy() *RefNamespaceName {
-	if in == nil {
-		return nil
-	}
-	out := new(RefNamespaceName)
-	in.DeepCopyInto(out)
-	return out
-}
-
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *ReplicaChanger) DeepCopyInto(out *ReplicaChanger) {
 	*out = *in
@@ -1337,6 +1308,11 @@ func (in *ScaleIn) DeepCopy() *ScaleIn {
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *ScaleOut) DeepCopyInto(out *ScaleOut) {
 	*out = *in
+	if in.FromBackup != nil {
+		in, out := &in.FromBackup, &out.FromBackup
+		*out = new(FromBackup)
+		(*in).DeepCopyInto(*out)
+	}
 	in.ReplicaChanger.DeepCopyInto(&out.ReplicaChanger)
 	if in.NewInstances != nil {
 		in, out := &in.NewInstances, &out.NewInstances
diff --git a/config/crd/bases/operations.kubeblocks.io_opsrequests.yaml b/config/crd/bases/operations.kubeblocks.io_opsrequests.yaml
index 66c3104d2..c2627f8b7 100644
--- a/config/crd/bases/operations.kubeblocks.io_opsrequests.yaml
+++ b/config/crd/bases/operations.kubeblocks.io_opsrequests.yaml
@@ -615,6 +615,163 @@ spec:
                         and brings offline instances back online. Can be used in conjunction with the "scaleIn" operation.
                         Note: Any configuration that deletes instances is considered invalid.
                       properties:
+                        fromBackup:
+                          description: |-
+                            FromBackup specifies the configuration for creating new instances from an existing backup.
+                            This is only effective for non-sharding components.
+                            When specified, new instances will be created using data from the specified backup.
+                          properties:
+                            name:
+                              description: Specifies the name of the Backup name.
+                              type: string
+                            namespace:
+                              description: |-
+                                Specifies the namespace of the Backup namespace.
+                                If not specified, the namespace of the OpsRequest will be used.
+                              type: string
+                            restoreEnv:
+                              description: |-
+                                Defines container environment variables for the restore process.
+                                merged with the ones specified in the Backup and ActionSet resources.
+
+
+                                Merge priority: Restore env > Backup env > ActionSet env.
+
+
+                                Purpose: Some databases require different configurations when being restored as a standby
+                                compared to being restored as a primary.
+                                For example, when restoring MySQL as a replica, you need to set `skip_slave_start="ON"` for 5.7
+                                or `skip_replica_start="ON"` for 8.0.
+                                Allowing environment variables to be passed in makes it more convenient to control these behavioral differences
+                                during the restore process.
+                              items:
+                                description: EnvVar represents an environment variable
+                                  present in a Container.
+                                properties:
+                                  name:
+                                    description: Name of the environment variable.
+                                      Must be a C_IDENTIFIER.
+                                    type: string
+                                  value:
+                                    description: |-
+                                      Variable references $(VAR_NAME) are expanded
+                                      using the previously defined environment variables in the container and
+                                      any service environment variables. If a variable cannot be resolved,
+                                      the reference in the input string will be unchanged. Double $$ are reduced
+                                      to a single $, which allows for escaping the $(VAR_NAME) syntax: i.e.
+                                      "$$(VAR_NAME)" will produce the string literal "$(VAR_NAME)".
+                                      Escaped references will never be expanded, regardless of whether the variable
+                                      exists or not.
+                                      Defaults to "".
+                                    type: string
+                                  valueFrom:
+                                    description: Source for the environment variable's
+                                      value. Cannot be used if value is not empty.
+                                    properties:
+                                      configMapKeyRef:
+                                        description: Selects a key of a ConfigMap.
+                                        properties:
+                                          key:
+                                            description: The key to select.
+                                            type: string
+                                          name:
+                                            description: |-
+                                              Name of the referent.
+                                              More info: https://kubernetes.io/docs/concepts/overview/working-with-objects/names/#names
+                                              TODO: Add other useful fields. apiVersion, kind, uid?
+                                            type: string
+                                          optional:
+                                            description: Specify whether the ConfigMap
+                                              or its key must be defined
+                                            type: boolean
+                                        required:
+                                        - key
+                                        type: object
+                                        x-kubernetes-map-type: atomic
+                                      fieldRef:
+                                        description: |-
+                                          Selects a field of the pod: supports metadata.name, metadata.namespace, `metadata.labels['<KEY>']`, `metadata.annotations['<KEY>']`,
+                                          spec.nodeName, spec.serviceAccountName, status.hostIP, status.podIP, status.podIPs.
+                                        properties:
+                                          apiVersion:
+                                            description: Version of the schema the
+                                              FieldPath is written in terms of, defaults
+                                              to "v1".
+                                            type: string
+                                          fieldPath:
+                                            description: Path of the field to select
+                                              in the specified API version.
+                                            type: string
+                                        required:
+                                        - fieldPath
+                                        type: object
+                                        x-kubernetes-map-type: atomic
+                                      resourceFieldRef:
+                                        description: |-
+                                          Selects a resource of the container: only resources limits and requests
+                                          (limits.cpu, limits.memory, limits.ephemeral-storage, requests.cpu, requests.memory and requests.ephemeral-storage) are currently supported.
+                                        properties:
+                                          containerName:
+                                            description: 'Container name: required
+                                              for volumes, optional for env vars'
+                                            type: string
+                                          divisor:
+                                            anyOf:
+                                            - type: integer
+                                            - type: string
+                                            description: Specifies the output format
+                                              of the exposed resources, defaults to
+                                              "1"
+                                            pattern: ^(\+|-)?(([0-9]+(\.[0-9]*)?)|(\.[0-9]+))(([KMGTPE]i)|[numkMGTPE]|([eE](\+|-)?(([0-9]+(\.[0-9]*)?)|(\.[0-9]+))))?$
+                                            x-kubernetes-int-or-string: true
+                                          resource:
+                                            description: 'Required: resource to select'
+                                            type: string
+                                        required:
+                                        - resource
+                                        type: object
+                                        x-kubernetes-map-type: atomic
+                                      secretKeyRef:
+                                        description: Selects a key of a secret in
+                                          the pod's namespace
+                                        properties:
+                                          key:
+                                            description: The key of the secret to
+                                              select from.  Must be a valid secret
+                                              key.
+                                            type: string
+                                          name:
+                                            description: |-
+                                              Name of the referent.
+                                              More info: https://kubernetes.io/docs/concepts/overview/working-with-objects/names/#names
+                                              TODO: Add other useful fields. apiVersion, kind, uid?
+                                            type: string
+                                          optional:
+                                            description: Specify whether the Secret
+                                              or its key must be defined
+                                            type: boolean
+                                        required:
+                                        - key
+                                        type: object
+                                        x-kubernetes-map-type: atomic
+                                    type: object
+                                required:
+                                - name
+                                type: object
+                              type: array
+                              x-kubernetes-preserve-unknown-fields: true
+                            restorePointInTime:
+                              description: |-
+                                Specifies the point in time to which the restore should be performed.
+                                Supported time formats:
+
+
+                                - RFC3339 format, e.g. "2023-11-25T18:52:53Z"
+                                - A human-readable date-time format, e.g. "Jul 25,2023 18:52:53 UTC+0800"
+                              type: string
+                          required:
+                          - name
+                          type: object
                         instances:
                           description: |-
                             Modifies the desired replicas count for existing InstanceTemplate.
diff --git a/controllers/operations/opsrequest_controller.go b/controllers/operations/opsrequest_controller.go
index 3b45d87cc..20780a631 100644
--- a/controllers/operations/opsrequest_controller.go
+++ b/controllers/operations/opsrequest_controller.go
@@ -431,6 +431,23 @@ func (r *OpsRequestReconciler) getRunningOpsRequestsFromCluster(cluster *appsv1.
 
 func (r *OpsRequestReconciler) parseRunningOpsRequests(ctx context.Context, object client.Object) []reconcile.Request {
 	cluster := object.(*appsv1.Cluster)
+	if cluster.IsDeleting() {
+		// if the cluster is deleting, we should enqueue all the opsRequests in this cluster.
+		opsList := &opsv1alpha1.OpsRequestList{}
+		_ = r.Client.List(ctx, opsList, client.InNamespace(cluster.Namespace), client.MatchingFields{
+			constant.AppInstanceLabelKey: cluster.Name,
+		})
+		var requests []reconcile.Request
+		for i := range opsList.Items {
+			requests = append(requests, reconcile.Request{
+				NamespacedName: types.NamespacedName{
+					Namespace: cluster.Namespace,
+					Name:      opsList.Items[i].Name,
+				},
+			})
+		}
+		return requests
+	}
 	return r.getRunningOpsRequestsFromCluster(cluster)
 }
 
diff --git a/deploy/helm/crds/operations.kubeblocks.io_opsrequests.yaml b/deploy/helm/crds/operations.kubeblocks.io_opsrequests.yaml
index 66c3104d2..c2627f8b7 100755
--- a/deploy/helm/crds/operations.kubeblocks.io_opsrequests.yaml
+++ b/deploy/helm/crds/operations.kubeblocks.io_opsrequests.yaml
@@ -615,6 +615,163 @@ spec:
                         and brings offline instances back online. Can be used in conjunction with the "scaleIn" operation.
                         Note: Any configuration that deletes instances is considered invalid.
                       properties:
+                        fromBackup:
+                          description: |-
+                            FromBackup specifies the configuration for creating new instances from an existing backup.
+                            This is only effective for non-sharding components.
+                            When specified, new instances will be created using data from the specified backup.
+                          properties:
+                            name:
+                              description: Specifies the name of the Backup name.
+                              type: string
+                            namespace:
+                              description: |-
+                                Specifies the namespace of the Backup namespace.
+                                If not specified, the namespace of the OpsRequest will be used.
+                              type: string
+                            restoreEnv:
+                              description: |-
+                                Defines container environment variables for the restore process.
+                                merged with the ones specified in the Backup and ActionSet resources.
+
+
+                                Merge priority: Restore env > Backup env > ActionSet env.
+
+
+                                Purpose: Some databases require different configurations when being restored as a standby
+                                compared to being restored as a primary.
+                                For example, when restoring MySQL as a replica, you need to set `skip_slave_start="ON"` for 5.7
+                                or `skip_replica_start="ON"` for 8.0.
+                                Allowing environment variables to be passed in makes it more convenient to control these behavioral differences
+                                during the restore process.
+                              items:
+                                description: EnvVar represents an environment variable
+                                  present in a Container.
+                                properties:
+                                  name:
+                                    description: Name of the environment variable.
+                                      Must be a C_IDENTIFIER.
+                                    type: string
+                                  value:
+                                    description: |-
+                                      Variable references $(VAR_NAME) are expanded
+                                      using the previously defined environment variables in the container and
+                                      any service environment variables. If a variable cannot be resolved,
+                                      the reference in the input string will be unchanged. Double $$ are reduced
+                                      to a single $, which allows for escaping the $(VAR_NAME) syntax: i.e.
+                                      "$$(VAR_NAME)" will produce the string literal "$(VAR_NAME)".
+                                      Escaped references will never be expanded, regardless of whether the variable
+                                      exists or not.
+                                      Defaults to "".
+                                    type: string
+                                  valueFrom:
+                                    description: Source for the environment variable's
+                                      value. Cannot be used if value is not empty.
+                                    properties:
+                                      configMapKeyRef:
+                                        description: Selects a key of a ConfigMap.
+                                        properties:
+                                          key:
+                                            description: The key to select.
+                                            type: string
+                                          name:
+                                            description: |-
+                                              Name of the referent.
+                                              More info: https://kubernetes.io/docs/concepts/overview/working-with-objects/names/#names
+                                              TODO: Add other useful fields. apiVersion, kind, uid?
+                                            type: string
+                                          optional:
+                                            description: Specify whether the ConfigMap
+                                              or its key must be defined
+                                            type: boolean
+                                        required:
+                                        - key
+                                        type: object
+                                        x-kubernetes-map-type: atomic
+                                      fieldRef:
+                                        description: |-
+                                          Selects a field of the pod: supports metadata.name, metadata.namespace, `metadata.labels['<KEY>']`, `metadata.annotations['<KEY>']`,
+                                          spec.nodeName, spec.serviceAccountName, status.hostIP, status.podIP, status.podIPs.
+                                        properties:
+                                          apiVersion:
+                                            description: Version of the schema the
+                                              FieldPath is written in terms of, defaults
+                                              to "v1".
+                                            type: string
+                                          fieldPath:
+                                            description: Path of the field to select
+                                              in the specified API version.
+                                            type: string
+                                        required:
+                                        - fieldPath
+                                        type: object
+                                        x-kubernetes-map-type: atomic
+                                      resourceFieldRef:
+                                        description: |-
+                                          Selects a resource of the container: only resources limits and requests
+                                          (limits.cpu, limits.memory, limits.ephemeral-storage, requests.cpu, requests.memory and requests.ephemeral-storage) are currently supported.
+                                        properties:
+                                          containerName:
+                                            description: 'Container name: required
+                                              for volumes, optional for env vars'
+                                            type: string
+                                          divisor:
+                                            anyOf:
+                                            - type: integer
+                                            - type: string
+                                            description: Specifies the output format
+                                              of the exposed resources, defaults to
+                                              "1"
+                                            pattern: ^(\+|-)?(([0-9]+(\.[0-9]*)?)|(\.[0-9]+))(([KMGTPE]i)|[numkMGTPE]|([eE](\+|-)?(([0-9]+(\.[0-9]*)?)|(\.[0-9]+))))?$
+                                            x-kubernetes-int-or-string: true
+                                          resource:
+                                            description: 'Required: resource to select'
+                                            type: string
+                                        required:
+                                        - resource
+                                        type: object
+                                        x-kubernetes-map-type: atomic
+                                      secretKeyRef:
+                                        description: Selects a key of a secret in
+                                          the pod's namespace
+                                        properties:
+                                          key:
+                                            description: The key of the secret to
+                                              select from.  Must be a valid secret
+                                              key.
+                                            type: string
+                                          name:
+                                            description: |-
+                                              Name of the referent.
+                                              More info: https://kubernetes.io/docs/concepts/overview/working-with-objects/names/#names
+                                              TODO: Add other useful fields. apiVersion, kind, uid?
+                                            type: string
+                                          optional:
+                                            description: Specify whether the Secret
+                                              or its key must be defined
+                                            type: boolean
+                                        required:
+                                        - key
+                                        type: object
+                                        x-kubernetes-map-type: atomic
+                                    type: object
+                                required:
+                                - name
+                                type: object
+                              type: array
+                              x-kubernetes-preserve-unknown-fields: true
+                            restorePointInTime:
+                              description: |-
+                                Specifies the point in time to which the restore should be performed.
+                                Supported time formats:
+
+
+                                - RFC3339 format, e.g. "2023-11-25T18:52:53Z"
+                                - A human-readable date-time format, e.g. "Jul 25,2023 18:52:53 UTC+0800"
+                              type: string
+                          required:
+                          - name
+                          type: object
                         instances:
                           description: |-
                             Modifies the desired replicas count for existing InstanceTemplate.
diff --git a/docs/developer_docs/api-reference/operations.md b/docs/developer_docs/api-reference/operations.md
index 8b3f21a10..37b089581 100644
--- a/docs/developer_docs/api-reference/operations.md
+++ b/docs/developer_docs/api-reference/operations.md
@@ -610,34 +610,6 @@ Parameters match the schema specified in the <code>actionset.spec.parametersSche
 </tr>
 </tbody>
 </table>
-<h3 id="operations.kubeblocks.io/v1alpha1.BackupRefSpec">BackupRefSpec
-</h3>
-<div>
-</div>
-<table>
-<thead>
-<tr>
-<th>Field</th>
-<th>Description</th>
-</tr>
-</thead>
-<tbody>
-<tr>
-<td>
-<code>ref</code><br/>
-<em>
-<a href="#operations.kubeblocks.io/v1alpha1.RefNamespaceName">
-RefNamespaceName
-</a>
-</em>
-</td>
-<td>
-<em>(Optional)</em>
-<p>Refers to a reference backup that needs to be restored.</p>
-</td>
-</tr>
-</tbody>
-</table>
 <h3 id="operations.kubeblocks.io/v1alpha1.CompletionProbe">CompletionProbe
 </h3>
 <p>
@@ -1080,6 +1052,84 @@ in each OpsService definition.</p>
 </td>
 </tr></tbody>
 </table>
+<h3 id="operations.kubeblocks.io/v1alpha1.FromBackup">FromBackup
+</h3>
+<p>
+(<em>Appears on:</em><a href="#operations.kubeblocks.io/v1alpha1.ScaleOut">ScaleOut</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>Specifies the name of the Backup name.</p>
+</td>
+</tr>
+<tr>
+<td>
+<code>namespace</code><br/>
+<em>
+string
+</em>
+</td>
+<td>
+<p>Specifies the namespace of the Backup namespace.
+If not specified, the namespace of the OpsRequest will be used.</p>
+</td>
+</tr>
+<tr>
+<td>
+<code>restoreEnv</code><br/>
+<em>
+<a href="https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.25/#envvar-v1-core">
+[]Kubernetes core/v1.EnvVar
+</a>
+</em>
+</td>
+<td>
+<em>(Optional)</em>
+<p>Defines container environment variables for the restore process.
+merged with the ones specified in the Backup and ActionSet resources.</p>
+<p>Merge priority: Restore env &gt; Backup env &gt; ActionSet env.</p>
+<p>Purpose: Some databases require different configurations when being restored as a standby
+compared to being restored as a primary.
+For example, when restoring MySQL as a replica, you need to set <code>skip_slave_start=&quot;ON&quot;</code> for 5.7
+or <code>skip_replica_start=&quot;ON&quot;</code> for 8.0.
+Allowing environment variables to be passed in makes it more convenient to control these behavioral differences
+during the restore process.</p>
+</td>
+</tr>
+<tr>
+<td>
+<code>restorePointInTime</code><br/>
+<em>
+string
+</em>
+</td>
+<td>
+<p>Specifies the point in time to which the restore should be performed.
+Supported time formats:</p>
+<ul>
+<li>RFC3339 format, e.g. &ldquo;2023-11-25T18:52:53Z&rdquo;</li>
+<li>A human-readable date-time format, e.g. &ldquo;Jul 25,2023 18:52:53 UTC+0800&rdquo;</li>
+</ul>
+</td>
+</tr>
+</tbody>
+</table>
 <h3 id="operations.kubeblocks.io/v1alpha1.HorizontalScaling">HorizontalScaling
 </h3>
 <p>
@@ -3352,48 +3402,6 @@ or &lsquo;All&rsquo; (select all pods that match the podSelector).</p>
 </tr>
 </tbody>
 </table>
-<h3 id="operations.kubeblocks.io/v1alpha1.PointInTimeRefSpec">PointInTimeRefSpec
-</h3>
-<div>
-</div>
-<table>
-<thead>
-<tr>
-<th>Field</th>
-<th>Description</th>
-</tr>
-</thead>
-<tbody>
-<tr>
-<td>
-<code>time</code><br/>
-<em>
-<a href="https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.25/#time-v1-meta">
-Kubernetes meta/v1.Time
-</a>
-</em>
-</td>
-<td>
-<em>(Optional)</em>
-<p>Refers to the specific time point for restoration, with UTC as the time zone.</p>
-</td>
-</tr>
-<tr>
-<td>
-<code>ref</code><br/>
-<em>
-<a href="#operations.kubeblocks.io/v1alpha1.RefNamespaceName">
-RefNamespaceName
-</a>
-</em>
-</td>
-<td>
-<em>(Optional)</em>
-<p>Refers to a reference source cluster that needs to be restored.</p>
-</td>
-</tr>
-</tbody>
-</table>
 <h3 id="operations.kubeblocks.io/v1alpha1.PreCheckResult">PreCheckResult
 </h3>
 <p>
@@ -3770,47 +3778,6 @@ This field is used to override or set the values of parameters without modifying
 </tr>
 </tbody>
 </table>
-<h3 id="operations.kubeblocks.io/v1alpha1.RefNamespaceName">RefNamespaceName
-</h3>
-<p>
-(<em>Appears on:</em><a href="#operations.kubeblocks.io/v1alpha1.BackupRefSpec">BackupRefSpec</a>, <a href="#operations.kubeblocks.io/v1alpha1.PointInTimeRefSpec">PointInTimeRefSpec</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>
-<em>(Optional)</em>
-<p>Refers to the specific name of the resource.</p>
-</td>
-</tr>
-<tr>
-<td>
-<code>namespace</code><br/>
-<em>
-string
-</em>
-</td>
-<td>
-<em>(Optional)</em>
-<p>Refers to the specific namespace of the resource.</p>
-</td>
-</tr>
-</tbody>
-</table>
 <h3 id="operations.kubeblocks.io/v1alpha1.ReplicaChanger">ReplicaChanger
 </h3>
 <p>
@@ -4080,6 +4047,22 @@ ReplicaChanger
 <tbody>
 <tr>
 <td>
+<code>fromBackup</code><br/>
+<em>
+<a href="#operations.kubeblocks.io/v1alpha1.FromBackup">
+FromBackup
+</a>
+</em>
+</td>
+<td>
+<em>(Optional)</em>
+<p>FromBackup specifies the configuration for creating new instances from an existing backup.
+This is only effective for non-sharding components.
+When specified, new instances will be created using data from the specified backup.</p>
+</td>
+</tr>
+<tr>
+<td>
 <code>ReplicaChanger</code><br/>
 <em>
 <a href="#operations.kubeblocks.io/v1alpha1.ReplicaChanger">
diff --git a/pkg/controller/configuration/value_transformer.go b/pkg/controller/configuration/value_transformer.go
index ca91efec6..ce6cbd0ca 100644
--- a/pkg/controller/configuration/value_transformer.go
+++ b/pkg/controller/configuration/value_transformer.go
@@ -70,7 +70,8 @@ func (v *valueManager) BuildValueTransformer(key string) core.ValueTransformerFu
 	index := generics.FindFirstFunc(v.paramsDefs, func(paramDef *parametersv1alpha1.ParametersDefinition) bool {
 		return paramDef.Spec.FileName == key
 	})
-	if index < 0 || v.paramsDefs[index].Spec.ParametersSchema == nil {
+	if index < 0 || v.paramsDefs[index].Spec.ParametersSchema == nil ||
+		v.paramsDefs[index].Spec.ParametersSchema.SchemaInJSON == nil {
 		return nil
 	}
 	schema := v.paramsDefs[index].Spec.ParametersSchema.SchemaInJSON
@@ -80,9 +81,7 @@ func (v *valueManager) BuildValueTransformer(key string) core.ValueTransformerFu
 	defaultTransformer := &defaultValueTransformer{
 		openapi.FlattenSchema(schema.Properties[openapi.DefaultSchemaName]),
 	}
-	return func(value string, fieldName string) (any, error) {
-		return defaultTransformer.resolveValueWithType(value, fieldName)
-	}
+	return defaultTransformer.resolveValueWithType
 }
 
 func NewValueManager(paramsDefs []*parametersv1alpha1.ParametersDefinition, configs []parametersv1alpha1.ComponentConfigDescription) *valueManager {
diff --git a/pkg/controller/plan/restore.go b/pkg/controller/plan/restore.go
index 85d12f0c6..d8ac0de1b 100644
--- a/pkg/controller/plan/restore.go
+++ b/pkg/controller/plan/restore.go
@@ -54,7 +54,7 @@ type RestoreManager struct {
 
 	// private
 	namespace                         string
-	restoreTime                       string
+	RestoreTime                       string
 	env                               []corev1.EnvVar
 	parameters                        []dpv1alpha1.ParameterPair
 	volumeRestorePolicy               dpv1alpha1.VolumeClaimRestorePolicy
@@ -62,6 +62,7 @@ type RestoreManager struct {
 	startingIndex                     int32
 	replicas                          int32
 	restoreLabels                     map[string]string
+	RestoreNamePrefix                 string
 }
 
 func NewRestoreManager(ctx context.Context,
@@ -218,7 +219,7 @@ func (r *RestoreManager) BuildPrepareDataRestore(comp *component.SynthesizedComp
 				Namespace:        r.namespace,
 				SourceTargetName: sourceTargetName,
 			},
-			RestoreTime: r.restoreTime,
+			RestoreTime: r.RestoreTime,
 			Env:         r.env,
 			Parameters:  r.parameters,
 			PrepareDataConfig: &dpv1alpha1.PrepareDataConfig{
@@ -280,7 +281,7 @@ func (r *RestoreManager) DoPostReady(comp *component.SynthesizedComponent,
 				Namespace:        r.namespace,
 				SourceTargetName: sourceTargetName,
 			},
-			RestoreTime: r.restoreTime,
+			RestoreTime: r.RestoreTime,
 			Env:         r.env,
 			Parameters:  r.parameters,
 			ReadyConfig: &dpv1alpha1.ReadyConfig{
@@ -362,6 +363,9 @@ func (r *RestoreManager) GetRestoreObjectMeta(comp *component.SynthesizedCompone
 	if r.startingIndex != 0 {
 		name = fmt.Sprintf("%s-%d", name, r.startingIndex)
 	}
+	if r.RestoreNamePrefix != "" {
+		name = fmt.Sprintf("%s-%s", r.RestoreNamePrefix, name)
+	}
 	if len(r.restoreLabels) == 0 {
 		r.restoreLabels = constant.GetCompLabels(r.Cluster.Name, comp.Name)
 	}
@@ -396,7 +400,7 @@ func (r *RestoreManager) initFromAnnotation(comp *component.SynthesizedComponent
 	if volumeRestorePolicy := backupSource[constant.VolumeRestorePolicyKeyForRestore]; volumeRestorePolicy != "" {
 		r.volumeRestorePolicy = dpv1alpha1.VolumeClaimRestorePolicy(volumeRestorePolicy)
 	}
-	r.restoreTime = backupSource[constant.RestoreTimeKeyForRestore]
+	r.RestoreTime = backupSource[constant.RestoreTimeKeyForRestore]
 	doReadyRestoreAfterClusterRunning := backupSource[constant.DoReadyRestoreAfterClusterRunning]
 	if doReadyRestoreAfterClusterRunning == "true" {
 		r.doReadyRestoreAfterClusterRunning = true
diff --git a/pkg/operations/horizontal_scaling.go b/pkg/operations/horizontal_scaling.go
index f1561b940..f4712bc11 100644
--- a/pkg/operations/horizontal_scaling.go
+++ b/pkg/operations/horizontal_scaling.go
@@ -22,17 +22,25 @@ package operations
 import (
 	"fmt"
 	"slices"
+	"strconv"
+	"strings"
 	"time"
 
+	corev1 "k8s.io/api/core/v1"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/util/sets"
 	"k8s.io/utils/pointer"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
 	appsv1 "github.com/apecloud/kubeblocks/apis/apps/v1"
+	dpv1alpha1 "github.com/apecloud/kubeblocks/apis/dataprotection/v1alpha1"
 	opsv1alpha1 "github.com/apecloud/kubeblocks/apis/operations/v1alpha1"
 	"github.com/apecloud/kubeblocks/pkg/constant"
 	intctrlcomp "github.com/apecloud/kubeblocks/pkg/controller/component"
+	"github.com/apecloud/kubeblocks/pkg/controller/model"
+	"github.com/apecloud/kubeblocks/pkg/controller/plan"
 	intctrlutil "github.com/apecloud/kubeblocks/pkg/controllerutil"
 )
 
@@ -129,6 +137,10 @@ func (hs horizontalScalingOpsHandler) Action(reqCtx intctrlutil.RequestCtx, cli
 				horizontalScaling.ComponentName)
 			return intctrlutil.NewFatalError(errMsg)
 		}
+		if horizontalScaling.ScaleOut != nil && horizontalScaling.ScaleOut.FromBackup != nil {
+			// Wait for the persistent volume to be restored from backup before proceeding.
+			return nil
+		}
 		compSpec.Replicas = replicas
 		compSpec.Instances = instances
 		compSpec.OfflineInstances = offlineInstances
@@ -155,7 +167,13 @@ func (hs horizontalScalingOpsHandler) ReconcileAction(reqCtx intctrlutil.Request
 		}
 		var err error
 		lastCompConfiguration := opsRes.OpsRequest.Status.LastConfiguration.Components[pgRes.compOps.GetComponentName()]
-		pgRes.createdPodSet, pgRes.deletedPodSet, err = hs.getCreateAndDeletePodSet(opsRes, lastCompConfiguration, *pgRes.clusterComponent, horizontalScaling, pgRes.fullComponentName)
+		clusterComponentSpec := pgRes.clusterComponent.DeepCopy()
+		if horizontalScaling.ScaleOut != nil && horizontalScaling.ScaleOut.FromBackup != nil {
+			if err := hs.restoreDataFromBackup(reqCtx, cli, opsRes, pgRes, clusterComponentSpec, horizontalScaling, lastCompConfiguration, compStatus); err != nil {
+				return 0, 0, err
+			}
+		}
+		pgRes.createdPodSet, pgRes.deletedPodSet, err = hs.getCreateAndDeletePodSet(opsRes, lastCompConfiguration, *clusterComponentSpec, horizontalScaling, pgRes.fullComponentName)
 		if err != nil {
 			return 0, 0, err
 		}
@@ -166,6 +184,157 @@ func (hs horizontalScalingOpsHandler) ReconcileAction(reqCtx intctrlutil.Request
 	return compOpsHelper.reconcileActionWithComponentOps(reqCtx, cli, opsRes, "", handleComponentProgress)
 }
 
+func (hs horizontalScalingOpsHandler) getBackupObj(reqCtx intctrlutil.RequestCtx,
+	cli client.Client,
+	opsRes *OpsResource,
+	fromBackup opsv1alpha1.FromBackup) (*dpv1alpha1.Backup, error) {
+	backupNamespace := opsRes.Cluster.Namespace
+	if fromBackup.Namespace != "" {
+		backupNamespace = fromBackup.Namespace
+	}
+	backupObj := &dpv1alpha1.Backup{}
+	if err := cli.Get(reqCtx.Ctx, client.ObjectKey{Namespace: backupNamespace, Name: fromBackup.Name}, backupObj); err != nil {
+		if apierrors.IsNotFound(err) {
+			return nil, intctrlutil.NewFatalError(fmt.Sprintf("backup %s not found", fromBackup.Name))
+		}
+		return nil, err
+	}
+	if backupObj.Status.Phase != dpv1alpha1.BackupPhaseCompleted {
+		return nil, intctrlutil.NewFatalError(fmt.Sprintf("backup %s phase is not completed", fromBackup.Name))
+	}
+	return backupObj, nil
+}
+
+func (hs horizontalScalingOpsHandler) createRestore(reqCtx intctrlutil.RequestCtx,
+	cli client.Client,
+	opsRes *OpsResource,
+	synthesizedComponent *intctrlcomp.SynthesizedComponent,
+	restoreMGR *plan.RestoreManager,
+	compSpecDeepyCopy *appsv1.ClusterComponentSpec,
+	backupObj *dpv1alpha1.Backup,
+	templateName string) error {
+	getTemplate := func(templateName string) *appsv1.InstanceTemplate {
+		if templateName == "" {
+			return nil
+		}
+		for _, template := range compSpecDeepyCopy.Instances {
+			if template.Name == templateName {
+				return &template
+			}
+		}
+		return nil
+	}
+	// create restore
+	restore, err := restoreMGR.BuildPrepareDataRestore(synthesizedComponent, backupObj, getTemplate(templateName))
+	if err != nil {
+		return err
+	}
+	scheme, _ := opsv1alpha1.SchemeBuilder.Build()
+	if err := intctrlutil.SetOwnership(opsRes.OpsRequest, restore, scheme, ""); err != nil {
+		return err
+	}
+	if err := cli.Create(reqCtx.Ctx, restore); err != nil {
+		return err
+	}
+	reqCtx.Recorder.Eventf(opsRes.OpsRequest, corev1.EventTypeNormal, "RestoreCreated", "Restore %s created", restore.Name)
+	return nil
+}
+
+func (hs horizontalScalingOpsHandler) restoreDataFromBackup(reqCtx intctrlutil.RequestCtx,
+	cli client.Client,
+	opsRes *OpsResource,
+	pgRes *progressResource,
+	compSpecDeepyCopy *appsv1.ClusterComponentSpec,
+	horizontalScaling opsv1alpha1.HorizontalScaling,
+	lastCompConfiguration opsv1alpha1.LastComponentConfiguration,
+	compStatus *opsv1alpha1.OpsRequestComponentStatus) error {
+	restoreCompletedMsg := "Restore Data Completed"
+	// check if restore completed
+	if compStatus.Message == restoreCompletedMsg {
+		return nil
+	}
+	// get and check backup
+	fromBackup := horizontalScaling.ScaleOut.FromBackup
+	backupObj, err := hs.getBackupObj(reqCtx, cli, opsRes, *fromBackup)
+	if err != nil {
+		return err
+	}
+	replicas, instances, offlineInstances, err := hs.getExpectedCompValues(opsRes, compSpecDeepyCopy,
+		lastCompConfiguration, horizontalScaling)
+	if err != nil {
+		return err
+	}
+	compSpecDeepyCopy.Replicas = replicas
+	compSpecDeepyCopy.Instances = instances
+	compSpecDeepyCopy.OfflineInstances = offlineInstances
+	createdPodSet, _, err := hs.getCreateAndDeletePodSet(opsRes, lastCompConfiguration, *compSpecDeepyCopy, horizontalScaling, pgRes.fullComponentName)
+	if err != nil {
+		return err
+	}
+	comp, compDef, err := intctrlcomp.GetCompNCompDefByName(reqCtx.Ctx, cli, opsRes.Cluster.Namespace, constant.GenerateClusterComponentName(opsRes.Cluster.Name, pgRes.fullComponentName))
+	if err != nil {
+		return err
+	}
+	synthesizedComponent, err := intctrlcomp.BuildSynthesizedComponent(reqCtx.Ctx, cli, compDef, comp)
+	if err != nil {
+		return err
+	}
+	allRestoreCompleted := true
+	for podName, templateName := range createdPodSet {
+		idx := strings.LastIndex(podName, "-")
+		podIndex := podName[idx+1:]
+		podIndexInt, _ := strconv.Atoi(podIndex)
+		restoreMGR := plan.NewRestoreManager(reqCtx.Ctx, cli, opsRes.Cluster, model.GetScheme(), map[string]string{
+			constant.OpsRequestNameLabelKey: opsRes.OpsRequest.Name,
+		}, 1, int32(podIndexInt))
+		restoreMGR.RestoreTime = fromBackup.RestorePointInTime
+		restoreMGR.RestoreNamePrefix = string(opsRes.OpsRequest.UID[:8])
+		// check restore status
+		restoreMeta := restoreMGR.GetRestoreObjectMeta(synthesizedComponent, dpv1alpha1.PrepareData, templateName)
+		restore := &dpv1alpha1.Restore{}
+		if err := cli.Get(reqCtx.Ctx, types.NamespacedName{Namespace: opsRes.Cluster.Namespace, Name: restoreMeta.Name}, restore); err != nil {
+			if apierrors.IsNotFound(err) {
+				allRestoreCompleted = false
+				if err = hs.createRestore(reqCtx, cli, opsRes, synthesizedComponent, restoreMGR, compSpecDeepyCopy, backupObj, templateName); err != nil {
+					return err
+				}
+				continue
+			}
+			return err
+		}
+		if restore.Status.Phase == dpv1alpha1.RestorePhaseFailed {
+			return intctrlutil.NewFatalError(fmt.Sprintf("restore for horizontalScaling failed: you can describe the restore resource \"%s\"", restore.Name))
+		}
+		if restore.Status.Phase != dpv1alpha1.RestorePhaseCompleted {
+			allRestoreCompleted = false
+		}
+	}
+	compStatus.Message = "Restore Data In Progress"
+	if allRestoreCompleted {
+		if err := hs.scaleOutComponentAfterRestoreData(reqCtx, cli, opsRes, compSpecDeepyCopy); err != nil {
+			return err
+		}
+		// set restore completed message
+		compStatus.Message = restoreCompletedMsg
+	}
+	return nil
+}
+
+func (hs horizontalScalingOpsHandler) scaleOutComponentAfterRestoreData(reqCtx intctrlutil.RequestCtx,
+	cli client.Client,
+	opsRes *OpsResource,
+	compSpecDeepyCopy *appsv1.ClusterComponentSpec) error {
+	for i := range opsRes.Cluster.Spec.ComponentSpecs {
+		compSpec := &opsRes.Cluster.Spec.ComponentSpecs[i]
+		if compSpec.Name == compSpecDeepyCopy.Name {
+			compSpec.Replicas = compSpecDeepyCopy.Replicas
+			compSpec.OfflineInstances = compSpecDeepyCopy.OfflineInstances
+			compSpec.Instances = compSpecDeepyCopy.Instances
+		}
+	}
+	return cli.Update(reqCtx.Ctx, opsRes.Cluster)
+}
+
 // SaveLastConfiguration records last configuration to the OpsRequest.status.lastConfiguration
 func (hs horizontalScalingOpsHandler) SaveLastConfiguration(reqCtx intctrlutil.RequestCtx, cli client.Client, opsRes *OpsResource) error {
 	shardsMap := make(map[string]int32, len(opsRes.Cluster.Spec.Shardings))



Test patch:
diff --git a/pkg/operations/horizontal_scaling_test.go b/pkg/operations/horizontal_scaling_test.go
index a7b5194e6..4c3c0fbed 100644
--- a/pkg/operations/horizontal_scaling_test.go
+++ b/pkg/operations/horizontal_scaling_test.go
@@ -28,17 +28,23 @@ import (
 
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/util/sets"
 	"k8s.io/utils/pointer"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
 	appsv1 "github.com/apecloud/kubeblocks/apis/apps/v1"
+	dpv1alpha1 "github.com/apecloud/kubeblocks/apis/dataprotection/v1alpha1"
 	opsv1alpha1 "github.com/apecloud/kubeblocks/apis/operations/v1alpha1"
 	"github.com/apecloud/kubeblocks/pkg/constant"
+	"github.com/apecloud/kubeblocks/pkg/controller/component"
+	"github.com/apecloud/kubeblocks/pkg/controller/model"
+	"github.com/apecloud/kubeblocks/pkg/controller/plan"
 	intctrlutil "github.com/apecloud/kubeblocks/pkg/controllerutil"
 	"github.com/apecloud/kubeblocks/pkg/generics"
 	opsutil "github.com/apecloud/kubeblocks/pkg/operations/util"
 	testapps "github.com/apecloud/kubeblocks/pkg/testutil/apps"
+	testdp "github.com/apecloud/kubeblocks/pkg/testutil/dataprotection"
 	testk8s "github.com/apecloud/kubeblocks/pkg/testutil/k8s"
 	testops "github.com/apecloud/kubeblocks/pkg/testutil/operations"
 )
@@ -71,6 +77,9 @@ var _ = Describe("HorizontalScaling OpsRequest", func() {
 		testapps.ClearResourcesWithRemoveFinalizerOption(&testCtx, generics.InstanceSetSignature, true, inNS, ml)
 		// default GracePeriod is 30s
 		testapps.ClearResources(&testCtx, generics.PodSignature, inNS, ml, client.GracePeriodSeconds(0))
+		testapps.ClearResourcesWithRemoveFinalizerOption(&testCtx, generics.BackupSignature, true, inNS)
+		testapps.ClearResourcesWithRemoveFinalizerOption(&testCtx, generics.RestoreSignature, true, inNS)
+		testapps.ClearResourcesWithRemoveFinalizerOption(&testCtx, generics.ComponentSignature, true, inNS)
 	}
 
 	BeforeEach(cleanEnv)
@@ -88,9 +97,14 @@ var _ = Describe("HorizontalScaling OpsRequest", func() {
 		commonHScaleConsensusCompTest := func(reqCtx intctrlutil.RequestCtx,
 			changeClusterSpec func(cluster *appsv1.Cluster),
 			horizontalScaling opsv1alpha1.HorizontalScaling,
-			ignoreHscaleStrictValidate bool) (*OpsResource, []*corev1.Pod) {
+			ignoreHscaleStrictValidate,
+			skipCompReplicasCheck bool) (*OpsResource, []*corev1.Pod) {
 			By("init operations resources with CLusterDefinition/Hybrid components Cluster/consensus Pods")
 			opsRes, compDef, _ := initOperationsResources(compDefName, clusterName)
+			// mock component
+			comp, err := component.BuildComponent(opsRes.Cluster, &opsRes.Cluster.Spec.ComponentSpecs[0], nil, nil)
+			Expect(err).Should(BeNil())
+			Expect(k8sClient.Create(testCtx.Ctx, comp)).Should(Succeed())
 			Expect(testapps.ChangeObj(&testCtx, compDef, func(compDef *appsv1.ComponentDefinition) {
 				compDef.Spec.ReplicasLimit = &appsv1.ReplicasLimit{
 					MinReplicas: 0,
@@ -113,39 +127,41 @@ var _ = Describe("HorizontalScaling OpsRequest", func() {
 			mockComponentIsOperating(opsRes.Cluster, appsv1.UpdatingComponentPhase, defaultCompName)
 
 			By("expect for opsRequest phase is Creating after doing action")
-			_, err := GetOpsManager().Do(reqCtx, k8sClient, opsRes)
+			_, err = GetOpsManager().Do(reqCtx, k8sClient, opsRes)
 			Expect(err).ShouldNot(HaveOccurred())
 			Eventually(testops.GetOpsRequestPhase(&testCtx, client.ObjectKeyFromObject(opsRes.OpsRequest))).Should(Equal(opsv1alpha1.OpsCreatingPhase))
 
 			By("check for the replicas of consensus component after doing action again when opsRequest phase is Creating")
 			_, err = GetOpsManager().Do(reqCtx, k8sClient, opsRes)
 			Expect(err).ShouldNot(HaveOccurred())
-			Eventually(testapps.CheckObj(&testCtx, client.ObjectKeyFromObject(opsRes.Cluster), func(g Gomega, tmpCluster *appsv1.Cluster) {
-				lastCompConfiguration := opsRes.OpsRequest.Status.LastConfiguration.Components[defaultCompName]
-				expectedCompReplicas := *lastCompConfiguration.Replicas
-				scaleIn := horizontalScaling.ScaleIn
-				if scaleIn != nil {
-					if scaleIn.ReplicaChanges != nil {
-						expectedCompReplicas -= *scaleIn.ReplicaChanges
-					} else {
-						expectedCompReplicas -= int32(len(scaleIn.OnlineInstancesToOffline))
+			if !skipCompReplicasCheck {
+				Eventually(testapps.CheckObj(&testCtx, client.ObjectKeyFromObject(opsRes.Cluster), func(g Gomega, tmpCluster *appsv1.Cluster) {
+					lastCompConfiguration := opsRes.OpsRequest.Status.LastConfiguration.Components[defaultCompName]
+					expectedCompReplicas := *lastCompConfiguration.Replicas
+					scaleIn := horizontalScaling.ScaleIn
+					if scaleIn != nil {
+						if scaleIn.ReplicaChanges != nil {
+							expectedCompReplicas -= *scaleIn.ReplicaChanges
+						} else {
+							expectedCompReplicas -= int32(len(scaleIn.OnlineInstancesToOffline))
+						}
 					}
-				}
-				scaleOut := horizontalScaling.ScaleOut
-				if scaleOut != nil {
-					switch {
-					case scaleOut.ReplicaChanges != nil:
-						expectedCompReplicas += *scaleOut.ReplicaChanges
-					case len(scaleOut.NewInstances) > 0:
-						for _, v := range scaleOut.NewInstances {
-							expectedCompReplicas += *v.Replicas
+					scaleOut := horizontalScaling.ScaleOut
+					if scaleOut != nil {
+						switch {
+						case scaleOut.ReplicaChanges != nil:
+							expectedCompReplicas += *scaleOut.ReplicaChanges
+						case len(scaleOut.NewInstances) > 0:
+							for _, v := range scaleOut.NewInstances {
+								expectedCompReplicas += *v.Replicas
+							}
+						default:
+							expectedCompReplicas += int32(len(scaleOut.OfflineInstancesToOnline))
 						}
-					default:
-						expectedCompReplicas += int32(len(scaleOut.OfflineInstancesToOnline))
 					}
-				}
-				g.Expect(tmpCluster.Spec.GetComponentByName(defaultCompName).Replicas).Should(BeEquivalentTo(expectedCompReplicas))
-			})).Should(Succeed())
+					g.Expect(tmpCluster.Spec.GetComponentByName(defaultCompName).Replicas).Should(BeEquivalentTo(expectedCompReplicas))
+				})).Should(Succeed())
+			}
 
 			By("Test OpsManager.Reconcile function when horizontal scaling OpsRequest is Running")
 			opsRes.OpsRequest.Status.Phase = opsv1alpha1.OpsRunningPhase
@@ -198,7 +214,7 @@ var _ = Describe("HorizontalScaling OpsRequest", func() {
 			horizontalScaling opsv1alpha1.HorizontalScaling,
 			mockHScale func(podList []*corev1.Pod)) {
 			reqCtx := intctrlutil.RequestCtx{Ctx: testCtx.Ctx}
-			opsRes, podList := commonHScaleConsensusCompTest(reqCtx, changeClusterSpec, horizontalScaling, false)
+			opsRes, podList := commonHScaleConsensusCompTest(reqCtx, changeClusterSpec, horizontalScaling, false, false)
 			mockHScale(podList)
 			testapps.MockInstanceSetStatus(testCtx, opsRes.Cluster, defaultCompName)
 			checkOpsRequestPhaseIsSucceed(reqCtx, opsRes)
@@ -208,7 +224,7 @@ var _ = Describe("HorizontalScaling OpsRequest", func() {
 			horizontalScaling opsv1alpha1.HorizontalScaling,
 			isScaleDown bool) {
 			reqCtx := intctrlutil.RequestCtx{Ctx: testCtx.Ctx}
-			opsRes, podList := commonHScaleConsensusCompTest(reqCtx, nil, horizontalScaling, false)
+			opsRes, podList := commonHScaleConsensusCompTest(reqCtx, nil, horizontalScaling, false, false)
 			var pod *corev1.Pod
 			if isScaleDown {
 				By("delete the pod")
@@ -246,6 +262,67 @@ var _ = Describe("HorizontalScaling OpsRequest", func() {
 			})
 		})
 
+		It("test to scale out replicas from a full backup", func() {
+			By("create Backup")
+			backupName := "backup-for-ops-" + randomStr
+			backup := testdp.NewBackupFactory(testCtx.DefaultNamespace, backupName).
+				SetBackupPolicyName(testdp.BackupPolicyName).
+				SetBackupMethod(testdp.VSBackupMethodName).
+				Create(&testCtx).GetObject()
+
+			Expect(testapps.ChangeObjStatus(&testCtx, backup, func() {
+				backup.Status.Phase = dpv1alpha1.BackupPhaseCompleted
+				backup.Status.BackupMethod = &dpv1alpha1.BackupMethod{
+					Name: testdp.VSBackupMethodName,
+					TargetVolumes: &dpv1alpha1.TargetVolumeInfo{
+						Volumes: []string{"data"},
+					},
+				}
+			})).Should(Succeed())
+
+			By("scale out replicas from a full backup")
+			horizontalScaling := opsv1alpha1.HorizontalScaling{ScaleOut: &opsv1alpha1.ScaleOut{
+				FromBackup: &opsv1alpha1.FromBackup{
+					Name: backupName,
+				},
+			}}
+
+			horizontalScaling.ScaleOut.ReplicaChanges = pointer.Int32(2)
+			reqCtx := intctrlutil.RequestCtx{Ctx: testCtx.Ctx, Recorder: eventRecorder}
+			opsRes, _ := commonHScaleConsensusCompTest(reqCtx, nil, horizontalScaling, false, true)
+
+			By("mock restore phase to completed")
+			comp, compDef, err := component.GetCompNCompDefByName(reqCtx.Ctx, k8sClient, opsRes.Cluster.Namespace, constant.GenerateClusterComponentName(opsRes.Cluster.Name, defaultCompName))
+			Expect(err).Should(BeNil())
+			synthesizedComponent, err := component.BuildSynthesizedComponent(reqCtx.Ctx, k8sClient, compDef, comp)
+			Expect(err).Should(BeNil())
+			changeRestorePhaseToComplete := func(index int32) {
+				restoreMGR := plan.NewRestoreManager(reqCtx.Ctx, k8sClient, opsRes.Cluster, model.GetScheme(), map[string]string{
+					constant.OpsRequestNameLabelKey: opsRes.OpsRequest.Name,
+				}, 1, index)
+				// check restore status
+				restoreMeta := restoreMGR.GetRestoreObjectMeta(synthesizedComponent, dpv1alpha1.PrepareData, "")
+				testapps.GetAndChangeObjStatus(&testCtx, types.NamespacedName{
+					Namespace: restoreMeta.Namespace,
+					Name:      restoreMeta.Name,
+				}, func(restore *dpv1alpha1.Restore) {
+					restore.Status.Phase = dpv1alpha1.RestorePhaseCompleted
+				})
+			}
+			changeRestorePhaseToComplete(3)
+			changeRestorePhaseToComplete(4)
+
+			By("expect component replicas to 5")
+			Eventually(testapps.CheckObj(&testCtx, client.ObjectKeyFromObject(opsRes.Cluster), func(g Gomega, cluster *appsv1.Cluster) {
+				g.Expect(cluster.Spec.GetComponentByName(defaultCompName).Replicas).Should(Equal(pointer.Int32(5)))
+			}))
+
+			By("mock pods to created and expect opsRequest phase to Succeed")
+			createPods("", 3, 4)
+			testapps.MockInstanceSetStatus(testCtx, opsRes.Cluster, defaultCompName)
+			checkOpsRequestPhaseIsSucceed(reqCtx, opsRes)
+		})
+
 		It("test to scale in replicas with `scaleIn`", func() {
 			By("scale in replicas from 3 to 1")
 			horizontalScaling := opsv1alpha1.HorizontalScaling{ScaleIn: &opsv1alpha1.ScaleIn{}}
@@ -300,7 +377,7 @@ var _ = Describe("HorizontalScaling OpsRequest", func() {
 			mockHScale func(podList []*corev1.Pod),
 			ignoreHscalingStrictValidate bool) *OpsResource {
 			reqCtx := intctrlutil.RequestCtx{Ctx: testCtx.Ctx}
-			opsRes, podList := commonHScaleConsensusCompTest(reqCtx, changeClusterSpec, horizontalScaling, ignoreHscalingStrictValidate)
+			opsRes, podList := commonHScaleConsensusCompTest(reqCtx, changeClusterSpec, horizontalScaling, ignoreHscalingStrictValidate, false)
 			By("verify cluster spec is correct")
 			targetSpec := opsRes.Cluster.Spec.GetComponentByName(defaultCompName)
 			Expect(targetSpec.OfflineInstances).Should(HaveLen(len(expectOfflineInstances)))
@@ -472,7 +549,7 @@ var _ = Describe("HorizontalScaling OpsRequest", func() {
 				ScaleIn: &opsv1alpha1.ScaleIn{
 					ReplicaChanger: opsv1alpha1.ReplicaChanger{ReplicaChanges: pointer.Int32(3)},
 				},
-			}, false)
+			}, false, false)
 			By("verify cluster spec is correct")
 			var targetSpec *appsv1.ClusterComponentSpec
 			for i := range opsRes.Cluster.Spec.ComponentSpecs {
@@ -880,6 +957,7 @@ var _ = Describe("HorizontalScaling OpsRequest", func() {
 			}
 
 		})
+
 	})
 })