Repository: cloudnative-pg/cloudnative-pg

Issue:
Declarative support for PostgreSQL major in-place upgrades
Currently, Major upgrades of PostgreSQL can be done in an imperative way, by relying on the native logical replication capabilities and the `cnpg` plugin. See: https://www.gabrielebartolini.it/articles/2024/03/cloudnativepg-recipe-5-how-to-migrate-your-postgresql-database-in-kubernetes-with-~0-downtime-from-anywhere/

However, we should provide a way to support major upgrades just by changing the `major` field like in the following example:

```yaml
apiVersion: postgresql.cnpg.io/v1
kind: Cluster
metadata:
  name: cluster-example
spec:
  instances: 3
  imageCatalogRef:
    apiGroup: postgresql.cnpg.io
    kind: ImageCatalog
    name: postgresql
    major: 16
  storage:
    size: 1Gi
```

To:

```yaml
apiVersion: postgresql.cnpg.io/v1
kind: Cluster
metadata:
  name: cluster-example
spec:
  instances: 3
  imageCatalogRef:
    apiGroup: postgresql.cnpg.io
    kind: ImageCatalog
    name: postgresql
    major: 17
  storage:
    size: 1Gi
```

Alternative approaches are valid too, for example through CRDs that evoke imperative names (like upgrades) and that are useful too as they prepare the underlying PVCs for the new major.

Discussion:
@maxpain: Any updates on this?
I can't believe that CNPG doesn't have major pg upgrades.@bo0tzz: It's on the roadmap for 1.26.0 https://github.com/orgs/cloudnative-pg/projects/1/views/1 - as is also noted by the events listed above your comment ;)@gbartolini: Version 1.25 introduced support for logical subscriptions, which simplifies the process of performing major upgrades using logical replication. Additionally, we've added the ability to perform [parallel dump and restore](https://cloudnative-pg.io/documentation/current/database_import/#customizing-pg_dump-and-pg_restore-behavior), providing more flexibility for handling large-scale database operations. These features already cover many everyday use cases—logical online upgrades were, in fact, our preferred approach for Postgres upgrades during our time at 2ndQuadrant.

I've been reflecting on this feature and considering how the direction we're taking with operand container images and extensions might influence its evolution. Specifically, we're working towards separating PostgreSQL operand images from extensions starting with PostgreSQL 18, with a proposed patch by EDB for the `extension_control_path` in PostgreSQL (see [RFC: Extension Packaging Lookup](https://justatheory.com/2024/11/rfc-extension-packaging-lookup/). It is hard to say without a prototype, but I remain concerned about the potential for extension upgrades to lead to unpredictable or unrecoverable scenarios. 

However, given the strong demand for this feature, one approach for an initial implementation could be an **offline update**. If the update fails, the system would revert to the previous PostgreSQL version, resulting in an inconsistent state and requiring manual intervention to change the cluster definition. If successful, the process would promote the primary, destroy the standbys, and recreate them in sequence. Thoughts?

Given our current commitments, I believe we can start working on this feature for version 1.26. However, due to the uncertainties mentioned earlier, there is a possibility that its delivery might be delayed.@pschichtel: > one approach for an initial implementation could be an offline update. If the update fails, the system would revert to the previous PostgreSQL version, resulting in an inconsistent state and requiring manual intervention to change the cluster definition.

The main point of concern is extensions, right? How is the risk of failure/inconsistency different when performing offline updates?@gbartolini: > > one approach for an initial implementation could be an offline update. If the update fails, the system would revert to the previous PostgreSQL version, resulting in an inconsistent state and requiring manual intervention to change the cluster definition.
> 
> The main point of concern is extensions, right? How is the risk of failure/inconsistency different when performing offline updates?

You work on a new cluster. You have time to prepare it in advance and test it, especially with applications. That's always been the main advantage of migrations to a different cluster, at least in my experience. You will have time to test upgrades to a different extension version as well.@pschichtel: Ah ok that's what you mean, makes sense. The declarative in-place upgrade approach wouldn't prevent people from using the imperative approach to have more safety if preferred, right? Even clusters that don't use extensions could hit a bug in a newer PG version and (subtly) break, so  cluster upgrades would need to go through the usual QA processes anyway@mnencia: I’ve submitted a possible implementation for declarative support for PostgreSQL major in-place upgrades in #6664. I would greatly appreciate it if you could review and test my approach and provide feedback or suggestions for improvement.@egkristi: Will in-place major upgrades be included in version 1.26, scheduled for February 20th?

We are currently evaluating CNPG, and this feature is critical for our decision. With over a thousand databases to manage, performing major upgrades imperatively is far too time-consuming. Declarative support for in-place upgrades would be a game-changer for us.

We’d greatly appreciate any updates on whether this will make it into the next release.@gbartolini: This feature will be in 1.26.0.

Solution patch:
diff --git a/.github/workflows/continuous-delivery.yml b/.github/workflows/continuous-delivery.yml
index f296d9fe..9f6d940f 100644
--- a/.github/workflows/continuous-delivery.yml
+++ b/.github/workflows/continuous-delivery.yml
@@ -22,8 +22,8 @@ on:
         description: >
           Feature Type (backup-restore, basic, cluster-metadata, declarative-databases, disruptive,
           importing-databases, maintenance, no-openshift, observability, operator, performance, plugin,
-          pod-scheduling, postgres-configuration, publication-subscription, recovery, replication, 
-          security, self-healing, service-connectivity, smoke, snapshot, storage, tablespaces, upgrade)
+          pod-scheduling, postgres-configuration, postgres-major-upgrade, publication-subscription, recovery,
+          replication, security, self-healing, service-connectivity, smoke, snapshot, storage, tablespaces, upgrade)
         required: false
       log_level:
         description: 'Log level for operator (error, warning, info, debug(default), trace)'
diff --git a/.wordlist-en-custom.txt b/.wordlist-en-custom.txt
index 40b69b52..268856ea 100644
--- a/.wordlist-en-custom.txt
+++ b/.wordlist-en-custom.txt
@@ -961,6 +961,7 @@ lsn
 lt
 lz
 macOS
+majorVersionUpgradeFromImage
 malcolm
 mallocs
 managedRoleSecretVersion
diff --git a/api/v1/cluster_funcs.go b/api/v1/cluster_funcs.go
index ab08c6ea..0f36bcdb 100644
--- a/api/v1/cluster_funcs.go
+++ b/api/v1/cluster_funcs.go
@@ -394,26 +394,6 @@ func (cluster *Cluster) SetInContext(ctx context.Context) context.Context {
 	return context.WithValue(ctx, utils.ContextKeyCluster, cluster)
 }
 
-// GetImageName get the name of the image that should be used
-// to create the pods
-func (cluster *Cluster) GetImageName() string {
-	// If the image is specified in the status, use that one
-	// It should be there since the first reconciliation
-	if len(cluster.Status.Image) > 0 {
-		return cluster.Status.Image
-	}
-
-	// Fallback to the information we have in the spec
-	if len(cluster.Spec.ImageName) > 0 {
-		return cluster.Spec.ImageName
-	}
-
-	// TODO: check: does a scenario exists in which we do have an imageCatalog
-	//   and no status.image? In that case this should probably error out, not
-	//   returning the default image name.
-	return configuration.Current.PostgresImageName
-}
-
 // GetPostgresqlVersion gets the PostgreSQL image version detecting it from the
 // image name or from the ImageCatalogRef.
 // Example:
@@ -421,13 +401,20 @@ func (cluster *Cluster) GetImageName() string {
 // ghcr.io/cloudnative-pg/postgresql:14.0 corresponds to version (14,0)
 // ghcr.io/cloudnative-pg/postgresql:13.2 corresponds to version (13,2)
 func (cluster *Cluster) GetPostgresqlVersion() (version.Data, error) {
+	if cluster.Status.Image != "" {
+		return version.FromTag(reference.New(cluster.Status.Image).Tag)
+	}
+
+	if cluster.Spec.ImageName != "" {
+		return version.FromTag(reference.New(cluster.Spec.ImageName).Tag)
+	}
+
 	if cluster.Spec.ImageCatalogRef != nil {
 		return version.FromTag(strconv.Itoa(cluster.Spec.ImageCatalogRef.Major))
 	}
 
-	image := cluster.GetImageName()
-	tag := reference.New(image).Tag
-	return version.FromTag(tag)
+	// Fallback for unit tests where a cluster is created without status or defaults
+	return version.FromTag(reference.New(configuration.Current.PostgresImageName).Tag)
 }
 
 // GetImagePullSecret get the name of the pull secret to use
diff --git a/api/v1/cluster_types.go b/api/v1/cluster_types.go
index 55a0d7cf..c570c9b6 100644
--- a/api/v1/cluster_types.go
+++ b/api/v1/cluster_types.go
@@ -196,7 +196,6 @@ type ImageCatalogRef struct {
 	// +kubebuilder:validation:XValidation:rule="self.kind == 'ImageCatalog' || self.kind == 'ClusterImageCatalog'",message="Only image catalogs are supported"
 	// +kubebuilder:validation:XValidation:rule="self.apiGroup == 'postgresql.cnpg.io'",message="Only image catalogs are supported"
 	corev1.TypedLocalObjectReference `json:",inline"`
-	// +kubebuilder:validation:XValidation:rule="self == oldSelf",message="Major is immutable"
 	// The major version of PostgreSQL we want to use from the ImageCatalog
 	Major int `json:"major"`
 }
@@ -587,7 +586,10 @@ const (
 	// PhaseUpgrade upgrade in process
 	PhaseUpgrade = "Upgrading cluster"
 
-	// PhaseUpgradeDelayed is set when a cluster need to be upgraded
+	// PhaseMajorUpgrade major version upgrade in process
+	PhaseMajorUpgrade = "Upgrading Postgres major version"
+
+	// PhaseUpgradeDelayed is set when a cluster needs to be upgraded,
 	// but the operation is being delayed by the operator configuration
 	PhaseUpgradeDelayed = "Cluster upgrade delayed"
 
@@ -942,6 +944,11 @@ type ClusterStatus struct {
 	// +optional
 	Image string `json:"image,omitempty"`
 
+	// MajorVersionUpgradeFromImage contains the image that was
+	// running before the major version upgrade started.
+	// +optional
+	MajorVersionUpgradeFromImage *string `json:"majorVersionUpgradeFromImage,omitempty"`
+
 	// PluginStatus is the status of the loaded plugins
 	// +optional
 	PluginStatus []PluginStatus `json:"pluginStatus,omitempty"`
diff --git a/api/v1/zz_generated.deepcopy.go b/api/v1/zz_generated.deepcopy.go
index b19b17b6..5943b911 100644
--- a/api/v1/zz_generated.deepcopy.go
+++ b/api/v1/zz_generated.deepcopy.go
@@ -958,6 +958,11 @@ func (in *ClusterStatus) DeepCopyInto(out *ClusterStatus) {
 		*out = make([]string, len(*in))
 		copy(*out, *in)
 	}
+	if in.MajorVersionUpgradeFromImage != nil {
+		in, out := &in.MajorVersionUpgradeFromImage, &out.MajorVersionUpgradeFromImage
+		*out = new(string)
+		**out = **in
+	}
 	if in.PluginStatus != nil {
 		in, out := &in.PluginStatus, &out.PluginStatus
 		*out = make([]PluginStatus, len(*in))
diff --git a/config/crd/bases/postgresql.cnpg.io_clusters.yaml b/config/crd/bases/postgresql.cnpg.io_clusters.yaml
index 38de43f0..611ca6b9 100644
--- a/config/crd/bases/postgresql.cnpg.io_clusters.yaml
+++ b/config/crd/bases/postgresql.cnpg.io_clusters.yaml
@@ -3074,9 +3074,6 @@ spec:
                     description: The major version of PostgreSQL we want to use from
                       the ImageCatalog
                     type: integer
-                    x-kubernetes-validations:
-                    - message: Major is immutable
-                      rule: self == oldSelf
                   name:
                     description: Name is the name of resource being referenced
                     type: string
@@ -6180,6 +6177,11 @@ spec:
                 description: ID of the latest generated node (used to avoid node name
                   clashing)
                 type: integer
+              majorVersionUpgradeFromImage:
+                description: |-
+                  MajorVersionUpgradeFromImage contains the image that was
+                  running before the major version upgrade started.
+                type: string
               managedRolesStatus:
                 description: ManagedRolesStatus reports the state of the managed roles
                   in the cluster
diff --git a/contribute/e2e_testing_environment/README.md b/contribute/e2e_testing_environment/README.md
index 53d0ee74..45ed0e59 100644
--- a/contribute/e2e_testing_environment/README.md
+++ b/contribute/e2e_testing_environment/README.md
@@ -185,6 +185,7 @@ exported, it will select all medium test cases from the feature type provided.
 | `tablespaces`                     |
 | `publication-subscription`        |
 | `declarative-databases`           |
+| `postgres-major-upgrade`          |
 
 ex:
 ```shell
diff --git a/docs/mkdocs.yml b/docs/mkdocs.yml
index 7375d2cd..2ef9d1cc 100644
--- a/docs/mkdocs.yml
+++ b/docs/mkdocs.yml
@@ -55,6 +55,7 @@ nav:
   - connection_pooling.md
   - replica_cluster.md
   - kubernetes_upgrade.md
+  - postgres_upgrades.md
   - kubectl-plugin.md
   - failover.md
   - troubleshooting.md
diff --git a/docs/src/cloudnative-pg.v1.md b/docs/src/cloudnative-pg.v1.md
index afd71dee..22a0a1d8 100644
--- a/docs/src/cloudnative-pg.v1.md
+++ b/docs/src/cloudnative-pg.v1.md
@@ -2265,6 +2265,14 @@ This field is reported when <code>.spec.failoverDelay</code> is populated or dur
    <p>Image contains the image name used by the pods</p>
 </td>
 </tr>
+<tr><td><code>majorVersionUpgradeFromImage</code><br/>
+<i>string</i>
+</td>
+<td>
+   <p>MajorVersionUpgradeFromImage contains the image that was
+running before the major version upgrade started.</p>
+</td>
+</tr>
 <tr><td><code>pluginStatus</code><br/>
 <a href="#postgresql-cnpg-io-v1-PluginStatus"><i>[]PluginStatus</i></a>
 </td>
diff --git a/docs/src/e2e.md b/docs/src/e2e.md
index be32a8a1..b2410fe4 100644
--- a/docs/src/e2e.md
+++ b/docs/src/e2e.md
@@ -133,3 +133,6 @@ and the following suite of E2E tests are performed on that cluster:
 * **Declarative databases**
   * Declarative creation of databases with default (retain) reclaim policy
   * Declarative creation of databases with delete reclaim policy
+
+* **Major version upgrade**
+  * Upgrade to the latest major version
diff --git a/docs/src/index.md b/docs/src/index.md
index 5225000f..c1f8ef98 100644
--- a/docs/src/index.md
+++ b/docs/src/index.md
@@ -118,6 +118,7 @@ Additionally, the community provides images for the [PostGIS extension](postgis.
     - Backups using volume snapshots (where supported by storage classes).
     - Full and Point-In-Time recovery from volume snapshots or object stores (via Barman Cloud plugin).
     - Backup from standby replicas to reduce primary workload impact.
+- Offline in-place major upgrades of PostgreSQL
 - Offline and online import of PostgreSQL databases, including major upgrades:
     - *Offline Import*: Direct restore from existing databases.
     - *Online Import*: PostgreSQL native logical replication via the `Subscription` resource.
diff --git a/docs/src/operator_capability_levels.md b/docs/src/operator_capability_levels.md
index b2613784..b7eeb208 100644
--- a/docs/src/operator_capability_levels.md
+++ b/docs/src/operator_capability_levels.md
@@ -308,11 +308,8 @@ or a subsequent switchover of the cluster.
 ### Upgrade of the managed workload
 
 The operand can be upgraded using a declarative configuration approach as
-part of changing the CR and, in particular, the `imageName` parameter. The
-operator prevents major upgrades of PostgreSQL while making it possible to go
-in both directions in terms of minor PostgreSQL releases within a major
-version, enabling updates and rollbacks.
-
+part of changing the CR and, in particular, the `imageName` parameter.
+This is normally initiated by security updates or Postgres minor version updates.
 In the presence of standby servers, the operator performs rolling updates
 starting from the replicas. It does this by dropping the existing pod and creating a new
 one with the new requested operand image that reuses the underlying storage.
@@ -324,11 +321,24 @@ The setting to use depends on the business requirements, as the operation
 might generate some downtime for the applications. This downtime can range from a few seconds to
 minutes, based on the actual database workload.
 
+### Offline In-Place Major Upgrades of PostgreSQL
+
+CloudNativePG supports declarative offline in-place major upgrades when a new
+operand container image with a higher PostgreSQL major version is applied to a
+cluster. The upgrade can be triggered by updating the image tag via the
+`.spec.imageName` option or by using an image catalog to manage version
+changes. During the upgrade, all cluster pods are shut down to ensure data
+consistency. A new job is then created to validate the upgrade conditions,
+execute `pg_upgrade`, and create new directories for `PGDATA`, WAL files, and
+tablespaces if needed. Once the upgrade is complete, replicas are re-created.
+Failed upgrades can be rolled back.
+
 ### Display cluster availability status during upgrade
 
 At any time, convey the cluster's high availability status, for example,
 `Setting up primary`, `Creating a new replica`, `Cluster in healthy state`,
-`Switchover in progress`, `Failing over`, and `Upgrading cluster`.
+`Switchover in progress`, `Failing over`, `Upgrading cluster`, and `Upgrading
+Postgres major version`.
 
 ## Level 3: Full lifecycle
 
diff --git a/docs/src/postgres_upgrades.md b/docs/src/postgres_upgrades.md
new file mode 100644
index 00000000..dac7a534
--- /dev/null
+++ b/docs/src/postgres_upgrades.md
@@ -0,0 +1,204 @@
+# PostgreSQL Upgrades
+<!-- SPDX-License-Identifier: CC-BY-4.0 -->
+
+PostgreSQL upgrades fall into two categories:
+
+- [Minor version upgrades](#minor-version-upgrades) (e.g., from 17.0 to 17.1)
+- [Major version upgrades](#major-version-upgrades) (e.g., from 16.x to 17.0)
+
+## Minor Version Upgrades
+
+PostgreSQL version numbers follow a `major.minor` format. For instance, in
+version 17.1:
+
+- `17` is the major version
+- `1` is the minor version
+
+Minor releases are fully compatible with earlier and later minor releases of
+the same major version. They include bug fixes and security updates but do not
+introduce changes to the internal storage format.
+For example, PostgreSQL 17.1 is compatible with 17.0 and 17.4.
+
+### Upgrading a Minor Version in CloudNativePG
+
+To upgrade to a newer minor version, simply update the PostgreSQL container
+image reference in your cluster definition, either directly or via image catalogs.
+CloudNativePG will trigger a [rolling update of the cluster](rolling_update.md),
+replacing each instance one by one, starting with the replicas. Once all
+replicas have been updated, it will perform either a switchover or a restart of
+the primary to complete the process.
+
+## Major Version Upgrades
+
+Major PostgreSQL releases introduce changes to the internal data storage
+format, requiring a more structured upgrade process.
+
+CloudNativePG supports three methods for performing major upgrades:
+
+1. [Logical dump/restore](database_import.md) – Blue/green deployment, offline.
+2. [Native logical replication](logical_replication.md#example-of-live-migration-and-major-postgres-upgrade-with-logical-replication) – Blue/green deployment, online.
+3. Physical with `pg_upgrade` – In-place upgrade, offline (covered in the
+   ["Offline In-Place Major Upgrades" section](#offline-in-place-major-upgrades) below).
+
+Each method has trade-offs in terms of downtime, complexity, and data volume
+handling. The best approach depends on your upgrade strategy and operational
+constraints.
+
+!!! Important
+    We strongly recommend testing all methods in a controlled environment
+    before proceeding with a production upgrade.
+
+## Offline In-Place Major Upgrades
+
+CloudNativePG performs an **offline in-place major upgrade** when a new operand
+container image with a higher PostgreSQL major version is declaratively
+requested for a cluster.
+
+You can trigger the upgrade in one of two ways:
+
+- By updating the major version in the image tag via the `.spec.imageName`
+  option.
+- Using an [image catalog](image_catalog.md) to manage version changes.
+
+For details on supported image tags, see
+["Image Tag Requirements"](container_images.md#image-tag-requirements).
+
+!!! Warning
+    CloudNativePG is not responsible for PostgreSQL extensions. You must ensure
+    that extensions in the source PostgreSQL image are compatible with those in the
+    target image and that upgrade paths are supported. Thoroughly test the upgrade
+    process in advance to avoid unexpected issues.
+    The [extensions management feature](declarative_database_management.md#managing-extensions-in-a-database)
+    can help manage extension upgrades declaratively.
+
+### Upgrade Process
+
+1. Shuts down all cluster pods to ensure data consistency.
+2. Records the previous PostgreSQL version in the cluster’s status under
+   `.status.majorVersionUpgradeFromImage`.
+3. Initiates a new upgrade job, which:
+   - Verifies that the binaries in the image and the data files align with a
+     major upgrade request.
+   - Creates new directories for `PGDATA`, and where applicable, WAL files and
+     tablespaces.
+   - Performs the upgrade using `pg_upgrade` with the `--link` option.
+   - Upon successful completion, replaces the original directories with their
+     upgraded counterparts.
+
+!!! Warning
+    During the upgrade process, the entire PostgreSQL cluster, including
+    replicas, is unavailable to applications. Ensure that your system can
+    tolerate this downtime before proceeding.
+
+!!! Warning
+    Performing an in-place upgrade is an exceptional operation that carries inherent
+    risks. It is strongly recommended to take a full backup of the cluster before
+    initiating the upgrade process.
+
+!!! Info
+    For detailed guidance on `pg_upgrade`, refer to the official
+    [PostgreSQL documentation](https://www.postgresql.org/docs/current/pgupgrade.html).
+
+### Post-Upgrade Actions
+
+If the upgrade is successful, CloudNativePG:
+
+- Destroys the PVCs of replicas (if available).
+- Scales up replicas as required.
+
+!!! Warning
+    Re-cloning replicas can be time-consuming, especially for very large
+    databases. Plan accordingly to accommodate potential delays. After completing
+    the upgrade, it is strongly recommended to take a full backup. Existing backup
+    data (namely base backups and WAL files) is only available for the previous
+    minor PostgreSQL release.
+
+!!! Warning
+    `pg_upgrade` doesn't transfer optimizer statistics. After the upgrade, you
+    may want to run `ANALYZE` on your databases to update them.
+
+If the upgrade fails, you must manually revert the major version change in the
+cluster's configuration and delete the upgrade job, as CloudNativePG cannot
+automatically decide the rollback.
+
+!!! Important
+    This process **protects your existing database from data loss**, as no data
+    is modified during the upgrade. If the upgrade fails, a rollback is
+    usually possible, without having to perform a full recovery from a backup.
+    Ensure you monitor the process closely and take corrective action if needed.
+
+### Example: Performing a Major Upgrade
+
+Consider the following PostgreSQL cluster running version 16:
+
+```yaml
+apiVersion: postgresql.cnpg.io/v1
+kind: Cluster
+metadata:
+  name: cluster-example
+spec:
+  imageName: ghcr.io/cloudnative-pg/postgresql:16-minimal-bookworm
+  instances: 3
+  storage:
+    size: 1Gi
+```
+
+You can check the current PostgreSQL version using the following command:
+
+```sh
+kubectl cnpg psql cluster-example -- -qAt -c 'SELECT version()'
+```
+
+This will return output similar to:
+
+```console
+PostgreSQL 16.x ...
+```
+
+To upgrade the cluster to version 17, update the `imageName` field by changing
+the major version tag from `16` to `17`:
+
+```yaml
+apiVersion: postgresql.cnpg.io/v1
+kind: Cluster
+metadata:
+  name: cluster-example
+spec:
+  imageName: ghcr.io/cloudnative-pg/postgresql:17-minimal-bookworm
+  instances: 3
+  storage:
+    size: 1Gi
+```
+
+### Upgrade Process
+
+1. Cluster shutdown – All cluster pods are terminated to ensure a consistent
+   upgrade.
+2. Upgrade job execution – A new job is created with the name of the primary
+   pod, appended with the suffix `-major-upgrade`. This job runs `pg_upgrade`
+   on the primary’s persistent volume group.
+3. Post-upgrade steps:
+   - The PVC groups of the replicas (`cluster-example-2` and
+     `cluster-example-3`) are removed.
+   - The primary pod is restarted.
+   - Two new replicas (`cluster-example-4` and `cluster-example-5`) are
+     re-cloned from the upgraded primary.
+
+Once the upgrade is complete, you can verify the new major version by running
+the same command:
+
+```sh
+kubectl cnpg psql cluster-example -- -qAt -c 'SELECT version()'
+```
+
+This should now return output similar to:
+
+```console
+PostgreSQL 17.x ...
+```
+
+You can now update the statistics by running `ANALYZE` on the `app` database:
+
+```sh
+kubectl cnpg psql cluster-example -- app -c 'ANALYZE'
+```
diff --git a/docs/src/release_notes/v1.26.md b/docs/src/release_notes/v1.26.md
index 69e24971..426cb56c 100644
--- a/docs/src/release_notes/v1.26.md
+++ b/docs/src/release_notes/v1.26.md
@@ -23,6 +23,15 @@ on the release branch in GitHub.
 
 ### Features:
 
+- **Declarative Offline In-Place Major Upgrades of PostgreSQL**: Introduced
+  support for offline in-place major upgrades when a new operand container
+  image with a higher PostgreSQL major version is applied to a cluster. During
+  the upgrade, all cluster pods are shut down to ensure data consistency. A new
+  job is created to validate upgrade conditions, run `pg_upgrade`, and set up new
+  directories for `PGDATA`, WAL files, and tablespaces as needed. Once the
+  upgrade is complete, replicas are re-created. Failed upgrades can be rolled
+  back declaratively. (#6664)
+
 - **Improved Startup and Readiness Probes for Replicas**: Enhanced support for
   Kubernetes startup and readiness probes in PostgreSQL instances, providing
   greater control over replicas based on the streaming lag. (#6623)
@@ -31,9 +40,6 @@ on the release branch in GitHub.
   `extensions` and `schemas` stanzas in the Database resource to declaratively
   create, modify, and drop PostgreSQL extensions and schemas within a database. (#7062)
 
-- **MAIN FEATURE #1**: short description
-- **MAIN FEATURE #2**: short description
-
 ### Enhancements:
 
 - Implemented the `cnpg.io/validation` annotation, allowing users to disable
diff --git a/docs/src/samples/cluster-example.yaml b/docs/src/samples/cluster-example.yaml
index fb331362..924f62bf 100644
--- a/docs/src/samples/cluster-example.yaml
+++ b/docs/src/samples/cluster-example.yaml
@@ -4,6 +4,8 @@ metadata:
   name: cluster-example
 spec:
   instances: 3
-
+  imageName: ghcr.io/cloudnative-pg/postgresql:13
+  
   storage:
     size: 1Gi
+
diff --git a/internal/cmd/manager/instance/cmd.go b/internal/cmd/manager/instance/cmd.go
index 3a45f490..62ac1e06 100644
--- a/internal/cmd/manager/instance/cmd.go
+++ b/internal/cmd/manager/instance/cmd.go
@@ -33,6 +33,7 @@ import (
 	"github.com/cloudnative-pg/cloudnative-pg/internal/cmd/manager/instance/restoresnapshot"
 	"github.com/cloudnative-pg/cloudnative-pg/internal/cmd/manager/instance/run"
 	"github.com/cloudnative-pg/cloudnative-pg/internal/cmd/manager/instance/status"
+	"github.com/cloudnative-pg/cloudnative-pg/internal/cmd/manager/instance/upgrade"
 	"github.com/cloudnative-pg/cloudnative-pg/pkg/postgres"
 )
 
@@ -56,6 +57,7 @@ func NewCmd() *cobra.Command {
 	cmd.AddCommand(pgbasebackup.NewCmd())
 	cmd.AddCommand(restore.NewCmd())
 	cmd.AddCommand(restoresnapshot.NewCmd())
+	cmd.AddCommand(upgrade.NewCmd())
 
 	return cmd
 }
diff --git a/internal/cmd/manager/instance/pgbasebackup/cmd.go b/internal/cmd/manager/instance/pgbasebackup/cmd.go
index 8b699cf4..d3f74879 100644
--- a/internal/cmd/manager/instance/pgbasebackup/cmd.go
+++ b/internal/cmd/manager/instance/pgbasebackup/cmd.go
@@ -138,7 +138,7 @@ func (env *CloneInfo) bootstrapUsingPgbasebackup(ctx context.Context) error {
 	if err != nil {
 		contextLogger.Warning(
 			"Error while parsing PostgreSQL server version to define connection options, defaulting to PostgreSQL 11",
-			"imageName", cluster.GetImageName(),
+			"imageName", cluster.Status.Image,
 			"err", err)
 	} else if pgVersion.Major() >= 12 {
 		// We explicitly disable wal_sender_timeout for join-related pg_basebackup executions.
diff --git a/internal/cmd/manager/instance/upgrade/cmd.go b/internal/cmd/manager/instance/upgrade/cmd.go
new file mode 100644
index 00000000..6275e014
--- /dev/null
+++ b/internal/cmd/manager/instance/upgrade/cmd.go
@@ -0,0 +1,40 @@
+/*
+Copyright © contributors to CloudNativePG, established as
+CloudNativePG a Series of LF Projects, LLC.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+
+SPDX-License-Identifier: Apache-2.0
+*/
+
+// Package upgrade implements the "instance upgrade" subcommand of the operator
+package upgrade
+
+import (
+	"github.com/spf13/cobra"
+
+	"github.com/cloudnative-pg/cloudnative-pg/internal/cmd/manager/instance/upgrade/execute"
+	"github.com/cloudnative-pg/cloudnative-pg/internal/cmd/manager/instance/upgrade/prepare"
+)
+
+// NewCmd creates the "instance upgrade" subcommand
+func NewCmd() *cobra.Command {
+	cmd := &cobra.Command{
+		Use: "upgrade",
+	}
+
+	cmd.AddCommand(prepare.NewCmd())
+	cmd.AddCommand(execute.NewCmd())
+
+	return cmd
+}
diff --git a/internal/cmd/manager/instance/upgrade/execute/cmd.go b/internal/cmd/manager/instance/upgrade/execute/cmd.go
new file mode 100644
index 00000000..40c192a7
--- /dev/null
+++ b/internal/cmd/manager/instance/upgrade/execute/cmd.go
@@ -0,0 +1,434 @@
+/*
+Copyright © contributors to CloudNativePG, established as
+CloudNativePG a Series of LF Projects, LLC.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+
+SPDX-License-Identifier: Apache-2.0
+*/
+
+// Package execute implements the "instance upgrade execute" subcommand
+package execute
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"os/exec"
+	"path"
+	"path/filepath"
+	"strings"
+	"time"
+
+	"github.com/blang/semver"
+	"github.com/cloudnative-pg/machinery/pkg/env"
+	"github.com/cloudnative-pg/machinery/pkg/execlog"
+	"github.com/cloudnative-pg/machinery/pkg/fileutils"
+	"github.com/cloudnative-pg/machinery/pkg/fileutils/compatibility"
+	"github.com/cloudnative-pg/machinery/pkg/log"
+	"github.com/spf13/cobra"
+	"k8s.io/utils/ptr"
+	ctrl "sigs.k8s.io/controller-runtime/pkg/client"
+
+	apiv1 "github.com/cloudnative-pg/cloudnative-pg/api/v1"
+	"github.com/cloudnative-pg/cloudnative-pg/internal/management/controller"
+	"github.com/cloudnative-pg/cloudnative-pg/internal/management/istio"
+	"github.com/cloudnative-pg/cloudnative-pg/internal/management/linkerd"
+	"github.com/cloudnative-pg/cloudnative-pg/pkg/configfile"
+	"github.com/cloudnative-pg/cloudnative-pg/pkg/management"
+	"github.com/cloudnative-pg/cloudnative-pg/pkg/management/postgres"
+	"github.com/cloudnative-pg/cloudnative-pg/pkg/management/postgres/constants"
+	"github.com/cloudnative-pg/cloudnative-pg/pkg/management/postgres/utils"
+	"github.com/cloudnative-pg/cloudnative-pg/pkg/management/postgres/webserver/metricserver"
+	"github.com/cloudnative-pg/cloudnative-pg/pkg/specs"
+)
+
+// NewCmd creates the cobra command
+func NewCmd() *cobra.Command {
+	var pgData string
+	var podName string
+	var clusterName string
+	var namespace string
+	var pgUpgrade string
+
+	cmd := &cobra.Command{
+		Use:  "execute [options]",
+		Args: cobra.ExactArgs(1),
+		RunE: func(cmd *cobra.Command, args []string) error {
+			oldBinDirFile := args[0]
+			ctx := cmd.Context()
+
+			// The fields in the instance are needed to correctly
+			// download the secret containing the TLS
+			// certificates
+			instance := postgres.NewInstance().
+				WithNamespace(namespace).
+				WithPodName(podName).
+				WithClusterName(clusterName)
+
+			// Read the old bindir from the passed file
+			oldBinDirBytes, err := fileutils.ReadFile(oldBinDirFile)
+			if err != nil {
+				return fmt.Errorf("error while reading the old bindir: %w", err)
+			}
+
+			oldBinDir := strings.TrimSpace(string(oldBinDirBytes))
+			return upgradeSubCommand(ctx, instance, pgData, oldBinDir, pgUpgrade)
+		},
+		PostRunE: func(cmd *cobra.Command, _ []string) error {
+			if err := istio.TryInvokeQuitEndpoint(cmd.Context()); err != nil {
+				return err
+			}
+
+			return linkerd.TryInvokeShutdownEndpoint(cmd.Context())
+		},
+	}
+
+	cmd.Flags().StringVar(&pgData, "pg-data", os.Getenv("PGDATA"), "The PGDATA to be created")
+	cmd.Flags().StringVar(&podName, "pod-name", os.Getenv("POD_NAME"), "The name of this pod, to "+
+		"be checked against the cluster state")
+	cmd.Flags().StringVar(&namespace, "namespace", os.Getenv("NAMESPACE"), "The namespace of "+
+		"the cluster and of the Pod in k8s")
+	cmd.Flags().StringVar(&clusterName, "cluster-name", os.Getenv("CLUSTER_NAME"), "The name of "+
+		"the current cluster in k8s, used to download TLS certificates")
+	cmd.Flags().StringVar(&pgUpgrade, "pg-upgrade", env.GetOrDefault("PG_UPGRADE", "pg_upgrade"),
+		`The path of "pg_upgrade" executable. Defaults to "pg_upgrade".`)
+
+	return cmd
+}
+
+func upgradeSubCommand(
+	ctx context.Context,
+	instance *postgres.Instance,
+	pgData string,
+	oldBinDir string,
+	pgUpgrade string,
+) error {
+	contextLogger := log.FromContext(ctx)
+
+	client, err := management.NewControllerRuntimeClient()
+	if err != nil {
+		contextLogger.Error(err, "Error creating Kubernetes client")
+		return err
+	}
+
+	clusterObjectKey := ctrl.ObjectKey{Name: instance.GetClusterName(), Namespace: instance.GetNamespaceName()}
+	if err = management.WaitForGetClusterWithClient(ctx, client, clusterObjectKey); err != nil {
+		return err
+	}
+
+	// Create a fake reconciler just to download the secrets and
+	// the cluster definition
+	metricExporter := metricserver.NewExporter(instance)
+	reconciler := controller.NewInstanceReconciler(instance, client, metricExporter)
+
+	// Download the cluster definition from the API server
+	var cluster apiv1.Cluster
+	if err := reconciler.GetClient().Get(ctx, clusterObjectKey, &cluster); err != nil {
+		contextLogger.Error(err, "Error while getting cluster")
+		return err
+	}
+
+	// Since we're directly using the reconciler here, we cannot
+	// tell if the secrets were correctly downloaded or not.
+	// If they were the following "pg_upgrade" command will work, if
+	// they don't "pg_upgrade" with fail, complaining that the
+	// cryptographic material is not available.
+	reconciler.RefreshSecrets(ctx, &cluster)
+
+	if err := reconciler.ReconcileWalStorage(ctx); err != nil {
+		return fmt.Errorf("error while reconciling the WAL storage: %w", err)
+	}
+
+	if err := fileutils.EnsureDirectoryExists(postgres.GetSocketDir()); err != nil {
+		return fmt.Errorf("while creating socket directory: %w", err)
+	}
+
+	contextLogger.Info("Searching for failed upgrades")
+
+	var failedDirs []string
+	for _, dir := range []string{specs.PgDataPath, specs.PgWalVolumePgWalPath} {
+		matches, err := filepath.Glob(dir + "*.failed_*")
+		if err != nil {
+			return fmt.Errorf("error matching paths: %w", err)
+		}
+		failedDirs = append(failedDirs, matches...)
+	}
+	if len(failedDirs) > 0 {
+		return fmt.Errorf("found failed upgrade directories: %v", failedDirs)
+	}
+
+	contextLogger.Info("Starting the upgrade process")
+
+	newDataDir := fmt.Sprintf("%s-new", specs.PgDataPath)
+	var newWalDir *string
+	if cluster.ShouldCreateWalArchiveVolume() {
+		newWalDir = ptr.To(fmt.Sprintf("%s-new", specs.PgWalVolumePgWalPath))
+	}
+
+	contextLogger.Info("Ensuring the new data directory does not exist", "directory", newDataDir)
+
+	if err := os.RemoveAll(newDataDir); err != nil {
+		return fmt.Errorf("failed to remove the directory: %w", err)
+	}
+
+	if newWalDir != nil {
+		contextLogger.Info("Ensuring the new pg_wal directory does not exist", "directory", *newWalDir)
+		if err := os.RemoveAll(*newWalDir); err != nil {
+			return fmt.Errorf("failed to remove the directory: %w", err)
+		}
+	}
+
+	contextLogger.Info("Creating data directory", "directory", newDataDir)
+	if err := runInitDB(newDataDir, newWalDir); err != nil {
+		return fmt.Errorf("error while creating the data directory: %w", err)
+	}
+
+	contextLogger.Info("Preparing configuration files", "directory", newDataDir)
+	if err := prepareConfigurationFiles(ctx, cluster, newDataDir); err != nil {
+		return err
+	}
+
+	contextLogger.Info("Checking if we have anything to update")
+	// Read pg_version from both the old and new data directories
+	oldVersion, err := utils.GetPgdataVersion(pgData)
+	if err != nil {
+		return fmt.Errorf("error while reading the old version: %w", err)
+	}
+
+	newVersion, err := utils.GetPgdataVersion(newDataDir)
+	if err != nil {
+		return fmt.Errorf("error while reading the new version: %w", err)
+	}
+
+	if oldVersion.Equals(newVersion) {
+		contextLogger.Info("Versions are the same, no need to upgrade")
+		if err := os.RemoveAll(newDataDir); err != nil {
+			return fmt.Errorf("failed to remove the directory: %w", err)
+		}
+		return nil
+	}
+
+	// We need to make sure that the permissions are the right ones
+	// in some systems they may be messed up even if we fix them before
+	_ = fileutils.EnsurePgDataPerms(pgData)
+	_ = fileutils.EnsurePgDataPerms(newDataDir)
+
+	contextLogger.Info("Running pg_upgrade")
+
+	if err := runPgUpgrade(pgData, pgUpgrade, newDataDir, oldBinDir); err != nil {
+		// TODO: in case of failures we should dump the content of the pg_upgrade logs
+		return fmt.Errorf("error while running pg_upgrade: %w", err)
+	}
+
+	err = moveDataInPlace(ctx, pgData, oldVersion, newDataDir, newWalDir)
+	if err != nil {
+		contextLogger.Error(err,
+			"Error while moving the data in place, saving the new data directory to avoid data loss")
+
+		suffixTimestamp := fileutils.FormatFriendlyTimestamp(time.Now())
+
+		dirToBeSaved := []string{
+			newDataDir,
+			pgData + ".old",
+		}
+		if newWalDir != nil {
+			dirToBeSaved = append(dirToBeSaved,
+				*newWalDir,
+				specs.PgWalVolumePgWalPath+".old",
+			)
+		}
+
+		for _, dir := range dirToBeSaved {
+			failedPgDataName := fmt.Sprintf("%s.failed_%s", dir, suffixTimestamp)
+			if errInner := moveDirIfExists(ctx, dir, failedPgDataName); errInner != nil {
+				contextLogger.Error(errInner, "Error while saving a directory after a failure", "dir", dir)
+			}
+		}
+
+		return err
+	}
+
+	contextLogger.Info("Upgrade completed successfully")
+
+	return nil
+}
+
+func runInitDB(destDir string, walDir *string) error {
+	// Invoke initdb to generate a data directory
+	options := []string{
+		"--username",
+		"postgres",
+		"-D",
+		destDir,
+	}
+
+	if walDir != nil {
+		options = append(options, "--waldir", *walDir)
+	}
+
+	// Certain CSI drivers may add setgid permissions on newly created folders.
+	// A default umask is set to attempt to avoid this, by revoking group/other
+	// permission bits on the PGDATA
+	_ = compatibility.Umask(0o077)
+
+	initdbCmd := exec.Command(constants.InitdbName, options...) // #nosec
+	if err := execlog.RunStreaming(initdbCmd, constants.InitdbName); err != nil {
+		return err
+	}
+
+	return nil
+}
+
+func prepareConfigurationFiles(ctx context.Context, cluster apiv1.Cluster, destDir string) error {
+	// Always read the custom and override configuration files created by the operator
+	_, err := configfile.EnsureIncludes(path.Join(destDir, "postgresql.conf"),
+		constants.PostgresqlCustomConfigurationFile,
+		constants.PostgresqlOverrideConfigurationFile,
+	)
+	if err != nil {
+		return fmt.Errorf("appending inclusion directives to postgresql.conf file resulted in an error: %w", err)
+	}
+
+	newInstance := postgres.Instance{PgData: destDir}
+	if _, err := newInstance.RefreshConfigurationFilesFromCluster(ctx, &cluster, false); err != nil {
+		return fmt.Errorf("error while creating the configuration files for new datadir %q: %w", destDir, err)
+	}
+
+	if _, err := newInstance.RefreshPGIdent(ctx, nil); err != nil {
+		return fmt.Errorf("error while creating the pg_ident.conf file for new datadir %q: %w", destDir, err)
+	}
+
+	// Create a stub for the configuration file
+	// to be filled during the real startup of this instance
+	err = fileutils.CreateEmptyFile(path.Join(destDir, constants.PostgresqlOverrideConfigurationFile))
+	if err != nil {
+		return fmt.Errorf("creating the operator managed configuration file '%v' resulted in an error: %w",
+			constants.PostgresqlOverrideConfigurationFile, err)
+	}
+
+	return nil
+}
+
+func runPgUpgrade(oldDataDir string, pgUpgrade string, newDataDir string, oldBinDir string) error {
+	// Run the pg_upgrade command
+	cmd := exec.Command(pgUpgrade,
+		"--link",
+		"--username", "postgres",
+		"--old-bindir", oldBinDir,
+		"--old-datadir", oldDataDir,
+		"--new-datadir", newDataDir,
+	) // #nosec
+	cmd.Dir = newDataDir
+	if err := execlog.RunStreaming(cmd, path.Base(pgUpgrade)); err != nil {
+		return fmt.Errorf("error while running %q: %w", cmd, err)
+	}
+
+	return nil
+}
+
+func moveDataInPlace(
+	ctx context.Context,
+	pgData string,
+	oldVersion semver.Version,
+	newDataDir string,
+	newWalDir *string,
+) error {
+	contextLogger := log.FromContext(ctx)
+
+	contextLogger.Info("Cleaning up the new data directory")
+	if err := os.RemoveAll(path.Join(newDataDir, "delete_old_cluster.sh")); err != nil {
+		return fmt.Errorf("error while removing the delete_old_cluster.sh script: %w", err)
+	}
+
+	contextLogger.Info("Moving the old data directory")
+	if err := os.Rename(pgData, pgData+".old"); err != nil {
+		return fmt.Errorf("error while moving the old data directory: %w", err)
+	}
+
+	if newWalDir != nil {
+		contextLogger.Info("Moving the old pg_wal directory")
+		if err := os.Rename(specs.PgWalVolumePgWalPath, specs.PgWalVolumePgWalPath+".old"); err != nil {
+			return fmt.Errorf("error while moving the old pg_wal directory: %w", err)
+		}
+	}
+
+	contextLogger.Info("Moving the new data directory in place")
+	if err := os.Rename(newDataDir, pgData); err != nil {
+		return fmt.Errorf("error while moving the new data directory: %w", err)
+	}
+
+	if newWalDir != nil {
+		contextLogger.Info("Moving the new pg_wal directory in place")
+		if err := os.Rename(*newWalDir, specs.PgWalVolumePgWalPath); err != nil {
+			return fmt.Errorf("error while moving the pg_wal directory content: %w", err)
+		}
+		if err := fileutils.RemoveFile(specs.PgWalPath); err != nil {
+			return fmt.Errorf("error while removing the symlink to pg_wal: %w", err)
+		}
+		if err := os.Symlink(specs.PgWalVolumePgWalPath, specs.PgWalPath); err != nil {
+			return fmt.Errorf("error while creating the symlink to pg_wal: %w", err)
+		}
+	}
+
+	contextLogger.Info("Removing the old data directory and pg_wal directory")
+	if err := os.RemoveAll(pgData + ".old"); err != nil {
+		return fmt.Errorf("error while removing the old data directory: %w", err)
+	}
+	if err := os.RemoveAll(specs.PgWalVolumePgWalPath + ".old"); err != nil {
+		return fmt.Errorf("error while removing the old pg_wal directory: %w", err)
+	}
+
+	contextLogger.Info("Cleaning up the previous version directory from tablespaces")
+	if err := removeMatchingPaths(ctx,
+		path.Join(pgData, "pg_tblspc", "*", fmt.Sprintf("PG_%v_*", oldVersion.Major))); err != nil {
+		return fmt.Errorf("error while removing the old tablespaces directories: %w", err)
+	}
+
+	return nil
+}
+
+func removeMatchingPaths(ctx context.Context, pattern string) error {
+	contextLogger := log.FromContext(ctx)
+	contextLogger.Info("Removing matching paths", "pattern", pattern)
+
+	// Find all matching paths
+	matches, err := filepath.Glob(pattern)
+	if err != nil {
+		return fmt.Errorf("error matching paths: %w", err)
+	}
+
+	// Iterate through the matches and remove each
+	for _, match := range matches {
+		contextLogger.Info("Removing path", "path", match)
+		err := os.RemoveAll(match)
+		if err != nil {
+			return fmt.Errorf("failed to remove %s: %w", match, err)
+		}
+	}
+
+	return nil
+}
+
+func moveDirIfExists(ctx context.Context, oldPath string, newPath string) error {
+	contextLogger := log.FromContext(ctx)
+	if _, errExists := os.Stat(oldPath); !os.IsNotExist(errExists) {
+		contextLogger.Info("Moving directory", "oldPath", oldPath, "newPath", newPath)
+		err := os.Rename(oldPath, newPath)
+		if err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
diff --git a/internal/cmd/manager/instance/upgrade/prepare/cmd.go b/internal/cmd/manager/instance/upgrade/prepare/cmd.go
new file mode 100644
index 00000000..b5a102a9
--- /dev/null
+++ b/internal/cmd/manager/instance/upgrade/prepare/cmd.go
@@ -0,0 +1,126 @@
+/*
+Copyright © contributors to CloudNativePG, established as
+CloudNativePG a Series of LF Projects, LLC.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+
+SPDX-License-Identifier: Apache-2.0
+*/
+
+// Package prepare implement the "instance upgrade prepare" subcommand
+package prepare
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"os/exec"
+	"path"
+
+	"github.com/cloudnative-pg/machinery/pkg/env"
+	"github.com/cloudnative-pg/machinery/pkg/fileutils"
+	"github.com/cloudnative-pg/machinery/pkg/log"
+	"github.com/cloudnative-pg/machinery/pkg/postgres/pgconfig"
+	"github.com/spf13/cobra"
+)
+
+// NewCmd create the cobra command
+func NewCmd() *cobra.Command {
+	var pgConfig string
+
+	cmd := cobra.Command{
+		Use:  "prepare [target]",
+		Args: cobra.ExactArgs(1),
+		RunE: func(cmd *cobra.Command, args []string) error {
+			contextLogger := log.FromContext(cmd.Context())
+			dest := args[0]
+
+			if err := copyPostgresInstallation(cmd.Context(), pgConfig, dest); err != nil {
+				contextLogger.Error(err, "Failed to copy the PostgreSQL installation")
+				return err
+			}
+
+			return nil
+		},
+	}
+
+	cmd.Flags().StringVar(&pgConfig, "pg-config", env.GetOrDefault("PG_CONFIG", "pg_config"),
+		`The path of "pg_config" executable. Defaults to "pg_config".`)
+
+	return &cmd
+}
+
+// copyPostgresInstallation replicates the PostgreSQL installation to the specified destination directory
+// for use by the pg_upgrade command as the old binary directory.
+//
+// Steps performed:
+// 1. Removes the existing destination directory if it exists.
+// 2. Retrieves the PostgreSQL binary, library, and shared directories using pg_config.
+// 3. Creates the corresponding directories in the destination path.
+// 4. Copies the contents of the PostgreSQL directories to the destination.
+// 5. Creates a bindir.txt file in the destination directory with the path to the binary directory.
+func copyPostgresInstallation(ctx context.Context, pgConfig string, dest string) error {
+	contextLogger := log.FromContext(ctx)
+
+	dest = path.Clean(dest)
+
+	contextLogger.Info("Copying the PostgreSQL installation to the destination", "destination", dest)
+
+	contextLogger.Info("Removing the destination directory", "directory", dest)
+	if err := os.RemoveAll(dest); err != nil {
+		return fmt.Errorf("failed to remove the directory: %w", err)
+	}
+
+	contextLogger.Info("Creating the destination directory", "directory", dest)
+	if err := os.MkdirAll(dest, 0o750); err != nil {
+		return fmt.Errorf("failed to create the directory: %w", err)
+	}
+
+	copyLocations := []pgconfig.ConfigurationParameter{pgconfig.BinDir, pgconfig.PkgLibDir, pgconfig.ShareDir}
+	for _, config := range copyLocations {
+		sourceDir, err := pgconfig.GetConfigurationParameter(pgConfig, config)
+		if err != nil {
+			return err
+		}
+		sourceDir = path.Clean(sourceDir)
+		destDir := path.Clean(path.Join(dest, sourceDir))
+
+		if config == pgconfig.BinDir {
+			destFile := path.Join(dest, "bindir.txt")
+			contextLogger.Info("Creating the bindir.txt file", "file", destFile)
+			if _, err := fileutils.WriteStringToFile(destFile, fmt.Sprintf("%s\n", destDir)); err != nil {
+				return fmt.Errorf("failed to write the %q file: %w", destFile, err)
+			}
+		}
+
+		contextLogger.Info("Creating the directory", "directory", destDir)
+		if err := os.MkdirAll(destDir, 0o750); err != nil {
+			return fmt.Errorf("failed to create the directory: %w", err)
+		}
+
+		contextLogger.Info("Copying the files", "source", sourceDir, "destination", destDir)
+
+		// We use "cp" instead of os.CopyFS because the latter doesn't
+		// support symbolic links as of Go 1.24 and we don't want to
+		// include any other dependencies in the project nor
+		// re-implementing the wheel.
+		//
+		// This should be re-evaluated in the future and the
+		// requirement to have "cp" in the image should be removed.
+		if err := exec.Command("cp", "-a", sourceDir+"/.", destDir).Run(); err != nil { //nolint:gosec
+			return fmt.Errorf("failed to copy the files: %w", err)
+		}
+	}
+
+	return nil
+}
diff --git a/internal/cmd/plugin/pgbench/pgbench.go b/internal/cmd/plugin/pgbench/pgbench.go
index 6b5fbb81..a9ec47f5 100644
--- a/internal/cmd/plugin/pgbench/pgbench.go
+++ b/internal/cmd/plugin/pgbench/pgbench.go
@@ -154,7 +154,7 @@ func (cmd *pgBenchRun) buildJob(cluster *apiv1.Cluster) *batchv1.Job {
 					Containers: []corev1.Container{
 						{
 							Name:            "pgbench",
-							Image:           cluster.GetImageName(),
+							Image:           cluster.Status.Image,
 							ImagePullPolicy: corev1.PullAlways,
 							Env:             cmd.buildEnvVariables(),
 							Command:         []string{pgBenchKeyWord},
diff --git a/internal/cmd/plugin/promote/promote.go b/internal/cmd/plugin/promote/promote.go
index fde8a61d..4f181d71 100644
--- a/internal/cmd/plugin/promote/promote.go
+++ b/internal/cmd/plugin/promote/promote.go
@@ -67,7 +67,7 @@ func Promote(ctx context.Context, cli client.Client,
 	}
 	if err := status.PatchWithOptimisticLock(ctx, cli, &cluster,
 		reconcileTargetPrimaryFunc,
-		status.SetClusterReadyConditionTX,
+		status.SetClusterReadyCondition,
 	); err != nil {
 		return err
 	}
diff --git a/internal/cmd/plugin/restart/restart.go b/internal/cmd/plugin/restart/restart.go
index da50c19a..63934941 100644
--- a/internal/cmd/plugin/restart/restart.go
+++ b/internal/cmd/plugin/restart/restart.go
@@ -75,8 +75,8 @@ func instanceRestart(ctx context.Context, clusterName, node string) error {
 			ctx,
 			plugin.Client,
 			&cluster,
-			status.SetPhaseTX(apiv1.PhaseInplacePrimaryRestart, "Requested by the user"),
-			status.SetClusterReadyConditionTX,
+			status.SetPhase(apiv1.PhaseInplacePrimaryRestart, "Requested by the user"),
+			status.SetClusterReadyCondition,
 		); err != nil {
 			return fmt.Errorf("while requesting restart on primary POD for cluster %v: %w", clusterName, err)
 		}
diff --git a/internal/cmd/plugin/status/status.go b/internal/cmd/plugin/status/status.go
index 3c83061e..4f8300d6 100644
--- a/internal/cmd/plugin/status/status.go
+++ b/internal/cmd/plugin/status/status.go
@@ -263,7 +263,7 @@ func (fullStatus *PostgresqlStatus) printBasicInfo(ctx context.Context, k8sClien
 	if primaryInstanceStatus != nil {
 		summary.AddLine("System ID:", primaryInstanceStatus.SystemID)
 	}
-	summary.AddLine("PostgreSQL Image:", cluster.GetImageName())
+	summary.AddLine("PostgreSQL Image:", cluster.Status.Image)
 	if cluster.IsReplica() {
 		summary.AddLine("Designated primary:", primaryInstance)
 		summary.AddLine("Source cluster: ", cluster.Spec.ReplicaCluster.Source)
diff --git a/internal/controller/cluster_controller.go b/internal/controller/cluster_controller.go
index c5613d33..6427391f 100644
--- a/internal/controller/cluster_controller.go
+++ b/internal/controller/cluster_controller.go
@@ -57,6 +57,7 @@ import (
 	"github.com/cloudnative-pg/cloudnative-pg/pkg/postgres"
 	"github.com/cloudnative-pg/cloudnative-pg/pkg/reconciler/hibernation"
 	instanceReconciler "github.com/cloudnative-pg/cloudnative-pg/pkg/reconciler/instance"
+	"github.com/cloudnative-pg/cloudnative-pg/pkg/reconciler/majorupgrade"
 	"github.com/cloudnative-pg/cloudnative-pg/pkg/reconciler/persistentvolumeclaim"
 	"github.com/cloudnative-pg/cloudnative-pg/pkg/reconciler/replicaclusterswitch"
 	"github.com/cloudnative-pg/cloudnative-pg/pkg/specs"
@@ -732,6 +733,20 @@ func (r *ClusterReconciler) reconcileResources(
 		return res, err
 	}
 
+	// In-place Postgres major version upgrades
+	if result, err := majorupgrade.Reconcile(
+		ctx,
+		r.Client,
+		cluster,
+		resources.instances.Items,
+		resources.pvcs.Items,
+		resources.jobs.Items,
+	); err != nil {
+		return ctrl.Result{}, fmt.Errorf("cannot reconcile in-place major version upgrades: %w", err)
+	} else if result != nil {
+		return *result, err
+	}
+
 	// Reconcile Pods
 	if res, err := r.reconcilePods(ctx, cluster, resources, instancesStatus); !res.IsZero() || err != nil {
 		return res, err
@@ -1099,6 +1114,19 @@ func (r *ClusterReconciler) SetupWithManager(ctx context.Context, mgr ctrl.Manag
 		Complete(r)
 }
 
+// jobOwnerIndexFunc maps a job definition to its owning cluster and
+// is used as an index function to speed up the lookup of jobs
+// created by the operator.
+func jobOwnerIndexFunc(rawObj client.Object) []string {
+	job := rawObj.(*batchv1.Job)
+
+	if ownerName, ok := IsOwnedByCluster(job); ok {
+		return []string{ownerName}
+	}
+
+	return nil
+}
+
 // createFieldIndexes creates the indexes needed by this controller
 func (r *ClusterReconciler) createFieldIndexes(ctx context.Context, mgr ctrl.Manager) error {
 	// Create a new indexed field on Pods. This field will be used to easily
@@ -1203,15 +1231,7 @@ func (r *ClusterReconciler) createFieldIndexes(ctx context.Context, mgr ctrl.Man
 	return mgr.GetFieldIndexer().IndexField(
 		ctx,
 		&batchv1.Job{},
-		jobOwnerKey, func(rawObj client.Object) []string {
-			job := rawObj.(*batchv1.Job)
-
-			if ownerName, ok := IsOwnedByCluster(job); ok {
-				return []string{ownerName}
-			}
-
-			return nil
-		})
+		jobOwnerKey, jobOwnerIndexFunc)
 }
 
 // IsOwnedByCluster checks that an object is owned by a Cluster and returns
diff --git a/internal/controller/cluster_image.go b/internal/controller/cluster_image.go
index 63c16b85..d838adb7 100644
--- a/internal/controller/cluster_image.go
+++ b/internal/controller/cluster_image.go
@@ -23,7 +23,9 @@ import (
 	"context"
 	"fmt"
 
+	"github.com/cloudnative-pg/machinery/pkg/image/reference"
 	"github.com/cloudnative-pg/machinery/pkg/log"
+	"github.com/cloudnative-pg/machinery/pkg/postgres/version"
 	apierrs "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/fields"
@@ -34,32 +36,89 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
 
 	apiv1 "github.com/cloudnative-pg/cloudnative-pg/api/v1"
+	"github.com/cloudnative-pg/cloudnative-pg/pkg/resources/status"
 )
 
-// reconcileImage sets the image inside the status, to be used by the following
-// functions of the reconciler loop
+// reconcileImage processes the image request, executes it, and stores
+// the result in the .status.image field. If the user requested a
+// major version upgrade, the current image is saved in the
+// .status.majorVersionUpgradeFromImage field. This allows for
+// reverting the upgrade if it doesn't complete successfully.
 func (r *ClusterReconciler) reconcileImage(ctx context.Context, cluster *apiv1.Cluster) (*ctrl.Result, error) {
 	contextLogger := log.FromContext(ctx)
 
-	oldCluster := cluster.DeepCopy()
+	image, err := r.getConfiguredImage(ctx, cluster)
+	if err != nil {
+		return &ctrl.Result{}, r.RegisterPhase(ctx, cluster, apiv1.PhaseImageCatalogError, err.Error())
+	}
+
+	currentDataImage := cluster.Status.Image
+	if cluster.Status.MajorVersionUpgradeFromImage != nil {
+		currentDataImage = *cluster.Status.MajorVersionUpgradeFromImage
+	}
+
+	// Case 1: the cluster is being initialized and there is still no
+	// running image. In this case, we should simply apply the image selected by the user.
+	if currentDataImage == "" {
+		return nil, status.PatchWithOptimisticLock(
+			ctx,
+			r.Client,
+			cluster,
+			status.SetImage(image),
+			status.SetMajorVersionUpgradeFromImage(nil),
+		)
+	}
+
+	// Case 2: there's a running image. The code checks if the user selected
+	// an image of the same major version or if a change in the major
+	// version has been requested.
+	var majorVersionUpgradeFromImage *string
+	currentVersion, err := version.FromTag(reference.New(currentDataImage).Tag)
+	if err != nil {
+		contextLogger.Error(err, "While parsing current major versions")
+		return nil, err
+	}
+
+	requestedVersion, err := version.FromTag(reference.New(image).Tag)
+	if err != nil {
+		contextLogger.Error(err, "While parsing requested major versions")
+		return nil, err
+	}
+
+	switch {
+	case currentVersion.Major() < requestedVersion.Major():
+		// The current major version is older than the requested one
+		majorVersionUpgradeFromImage = &currentDataImage
+	case currentVersion.Major() == requestedVersion.Major():
+		// The major versions are the same, cancel the update
+		majorVersionUpgradeFromImage = nil
+	default:
+		contextLogger.Info(
+			"Cannot downgrade the PostgreSQL major version. Forcing the current image.",
+			"currentImage", currentDataImage,
+			"requestedImage", image)
+		image = currentDataImage
+	}
+
+	return nil, status.PatchWithOptimisticLock(
+		ctx,
+		r.Client,
+		cluster,
+		status.SetImage(image),
+		status.SetMajorVersionUpgradeFromImage(majorVersionUpgradeFromImage),
+	)
+}
+
+func (r *ClusterReconciler) getConfiguredImage(ctx context.Context, cluster *apiv1.Cluster) (string, error) {
+	contextLogger := log.FromContext(ctx)
 
 	// If ImageName is defined and different from the current image in the status, we update the status
-	if cluster.Spec.ImageName != "" && cluster.Status.Image != cluster.Spec.ImageName {
-		cluster.Status.Image = cluster.Spec.ImageName
-		if err := r.Status().Patch(ctx, cluster, client.MergeFrom(oldCluster)); err != nil {
-			contextLogger.Error(
-				err,
-				"While patching cluster status to set the image name from the cluster Spec",
-				"imageName", cluster.Status.Image,
-			)
-			return nil, err
-		}
-		return nil, nil
+	if cluster.Spec.ImageName != "" {
+		return cluster.Spec.ImageName, nil
 	}
 
-	// If ImageName was defined, we rely on what the user requested
 	if cluster.Spec.ImageCatalogRef == nil {
-		return nil, nil
+		return "", fmt.Errorf("ImageName is not defined and no catalog is referenced")
 	}
 
 	contextLogger = contextLogger.WithValues("catalogRef", cluster.Spec.ImageCatalogRef)
@@ -74,15 +133,13 @@ func (r *ClusterReconciler) reconcileImage(ctx context.Context, cluster *apiv1.C
 		catalog = &apiv1.ImageCatalog{}
 	default:
 		contextLogger.Info("Unknown catalog kind")
-		return &ctrl.Result{}, r.RegisterPhase(ctx, cluster, apiv1.PhaseImageCatalogError,
-			"Invalid image catalog type")
+		return "", fmt.Errorf("invalid image catalog type")
 	}
 
 	apiGroup := cluster.Spec.ImageCatalogRef.APIGroup
 	if apiGroup == nil || *apiGroup != apiv1.SchemeGroupVersion.Group {
 		contextLogger.Info("Unknown catalog group")
-		return &ctrl.Result{}, r.RegisterPhase(ctx, cluster, apiv1.PhaseImageCatalogError,
-			"Invalid image catalog group")
+		return "", fmt.Errorf("invalid image catalog group")
 	}
 
 	// Get the referenced catalog
@@ -92,10 +149,11 @@ func (r *ClusterReconciler) reconcileImage(ctx context.Context, cluster *apiv1.C
 		if apierrs.IsNotFound(err) {
 			r.Recorder.Eventf(cluster, "Warning", "DiscoverImage", "Cannot get %v/%v",
 				catalogKind, catalogName)
-			return &ctrl.Result{}, nil
+			contextLogger.Info("catalog not found", "catalogKind", catalogKind, "catalogName", catalogName)
+			return "", fmt.Errorf("catalog %s/%s not found", catalogKind, catalogName)
 		}
 
-		return nil, err
+		return "", err
 	}
 
 	// Catalog found, we try to find the image for the major version
@@ -111,25 +169,10 @@ func (r *ClusterReconciler) reconcileImage(ctx context.Context, cluster *apiv1.C
 			catalogName)
 		contextLogger.Info("cannot find requested major version",
 			"requestedMajorVersion", requestedMajorVersion)
-		return &ctrl.Result{}, r.RegisterPhase(ctx, cluster, apiv1.PhaseImageCatalogError,
-			"Selected major version is not available in the catalog")
-	}
-
-	// If the image is different, we set it into the cluster status
-	if cluster.Status.Image != catalogImage {
-		cluster.Status.Image = catalogImage
-		patch := client.MergeFrom(oldCluster)
-		if err := r.Status().Patch(ctx, cluster, patch); err != nil {
-			patchBytes, _ := patch.Data(cluster)
-			contextLogger.Error(
-				err,
-				"While patching cluster status to set the image name from the catalog",
-				"patch", string(patchBytes))
-			return nil, err
-		}
+		return "", fmt.Errorf("selected major version is not available in the catalog")
 	}
 
-	return nil, nil
+	return catalogImage, nil
 }
 
 func (r *ClusterReconciler) getClustersForImageCatalogsToClustersMapper(
diff --git a/internal/controller/cluster_scale.go b/internal/controller/cluster_scale.go
index 0fc21229..ae7889a9 100644
--- a/internal/controller/cluster_scale.go
+++ b/internal/controller/cluster_scale.go
@@ -32,7 +32,7 @@ import (
 
 	apiv1 "github.com/cloudnative-pg/cloudnative-pg/api/v1"
 	"github.com/cloudnative-pg/cloudnative-pg/pkg/reconciler/persistentvolumeclaim"
-	"github.com/cloudnative-pg/cloudnative-pg/pkg/specs"
+	"github.com/cloudnative-pg/cloudnative-pg/pkg/utils"
 )
 
 // scaleDownCluster handles the scaling down operations of a PostgreSQL cluster.
@@ -101,13 +101,25 @@ func (r *ClusterReconciler) ensureInstanceJobAreDeleted(
 ) error {
 	contextLogger := log.FromContext(ctx)
 
-	for _, jobName := range specs.GetPossibleJobNames(instanceName) {
-		job := &batchv1.Job{
-			ObjectMeta: metav1.ObjectMeta{
-				Name:      jobName,
-				Namespace: cluster.Namespace,
-			},
-		}
+	var jobList batchv1.JobList
+	if err := r.List(
+		ctx,
+		&jobList,
+		client.InNamespace(cluster.Namespace),
+		client.MatchingFields{jobOwnerKey: cluster.Name},
+		client.MatchingLabels{
+			utils.InstanceNameLabelName: instanceName,
+			utils.ClusterLabelName:      cluster.Name,
+		},
+		client.HasLabels{
+			utils.JobRoleLabelName,
+		},
+	); err != nil {
+		return fmt.Errorf("while looking for stale jobs of instance %s: %w", instanceName, err)
+	}
+
+	for i := range jobList.Items {
+		job := &jobList.Items[i]
 
 		// This job was working against the PVC of this Pod,
 		// let's remove it
diff --git a/internal/controller/cluster_status.go b/internal/controller/cluster_status.go
index 5703b18b..62d0aa56 100644
--- a/internal/controller/cluster_status.go
+++ b/internal/controller/cluster_status.go
@@ -739,8 +739,8 @@ func (r *ClusterReconciler) RegisterPhase(ctx context.Context,
 		ctx,
 		r.Client,
 		cluster,
-		status.SetPhaseTX(phase, reason),
-		status.SetClusterReadyConditionTX,
+		status.SetPhase(phase, reason),
+		status.SetClusterReadyCondition,
 	)
 }
 
diff --git a/internal/controller/cluster_upgrade.go b/internal/controller/cluster_upgrade.go
index b13c94f3..c8be1840 100644
--- a/internal/controller/cluster_upgrade.go
+++ b/internal/controller/cluster_upgrade.go
@@ -495,7 +495,7 @@ func getProjectedVolumeConfigurationFromPod(pod corev1.Pod) *corev1.ProjectedVol
 }
 
 func checkPodImageIsOutdated(_ context.Context, pod *corev1.Pod, cluster *apiv1.Cluster) (rollout, error) {
-	targetImageName := cluster.GetImageName()
+	targetImageName := cluster.Status.Image
 
 	pgCurrentImageName, err := specs.GetPostgresImageName(*pod)
 	if err != nil {
@@ -680,7 +680,7 @@ func (r *ClusterReconciler) upgradePod(
 ) error {
 	log.FromContext(ctx).Info("Recreating instance pod",
 		"pod", pod.Name,
-		"to", cluster.GetImageName(),
+		"to", cluster.Status.Image,
 		"reason", reason,
 	)
 
diff --git a/internal/management/controller/instance_controller.go b/internal/management/controller/instance_controller.go
index 78ee1b6b..b8c8aebe 100644
--- a/internal/management/controller/instance_controller.go
+++ b/internal/management/controller/instance_controller.go
@@ -338,8 +338,8 @@ func (r *InstanceReconciler) restartPrimaryInplaceIfRequested(
 			ctx,
 			r.client,
 			cluster,
-			clusterstatus.SetPhaseTX(apiv1.PhaseHealthy, "Primary instance restarted in-place"),
-			clusterstatus.SetClusterReadyConditionTX,
+			clusterstatus.SetPhase(apiv1.PhaseHealthy, "Primary instance restarted in-place"),
+			clusterstatus.SetClusterReadyCondition,
 		)
 	}
 	return false, nil
@@ -1074,8 +1074,8 @@ func (r *InstanceReconciler) processConfigReloadAndManageRestart(ctx context.Con
 		ctx,
 		r.client,
 		cluster,
-		clusterstatus.SetPhaseTX(phase, phaseReason),
-		clusterstatus.SetClusterReadyConditionTX,
+		clusterstatus.SetPhase(phase, phaseReason),
+		clusterstatus.SetClusterReadyCondition,
 	)
 }
 
@@ -1098,8 +1098,8 @@ func (r *InstanceReconciler) triggerRestartForDecrease(ctx context.Context, clus
 		ctx,
 		r.client,
 		cluster,
-		clusterstatus.SetPhaseTX(phase, phaseReason),
-		clusterstatus.SetClusterReadyConditionTX,
+		clusterstatus.SetPhase(phase, phaseReason),
+		clusterstatus.SetClusterReadyCondition,
 	)
 }
 
diff --git a/internal/webhook/v1/cluster_webhook.go b/internal/webhook/v1/cluster_webhook.go
index 1671ce99..3de01012 100644
--- a/internal/webhook/v1/cluster_webhook.go
+++ b/internal/webhook/v1/cluster_webhook.go
@@ -1231,22 +1231,26 @@ func (v *ClusterCustomValidator) validateImageChange(r, old *apiv1.Cluster) fiel
 	var result field.ErrorList
 	var newVersion, oldVersion version.Data
 	var err error
-	var newImagePath *field.Path
+	var fieldPath *field.Path
 	if r.Spec.ImageCatalogRef != nil {
-		newImagePath = field.NewPath("spec", "imageCatalogRef")
+		fieldPath = field.NewPath("spec", "imageCatalogRef", "major")
 	} else {
-		newImagePath = field.NewPath("spec", "imageName")
+		fieldPath = field.NewPath("spec", "imageName")
 	}
 
-	r.Status.Image = ""
-	newVersion, err = r.GetPostgresqlVersion()
+	newCluster := r.DeepCopy()
+	newCluster.Status.Image = ""
+	newVersion, err = newCluster.GetPostgresqlVersion()
 	if err != nil {
 		// The validation error will be already raised by the
 		// validateImageName function
 		return result
 	}
 
-	old.Status.Image = ""
+	old = old.DeepCopy()
+	if old.Status.MajorVersionUpgradeFromImage != nil {
+		old.Status.Image = *old.Status.MajorVersionUpgradeFromImage
+	}
 	oldVersion, err = old.GetPostgresqlVersion()
 	if err != nil {
 		// The validation error will be already raised by the
@@ -1254,18 +1258,27 @@ func (v *ClusterCustomValidator) validateImageChange(r, old *apiv1.Cluster) fiel
 		return result
 	}
 
-	status := version.IsUpgradePossible(oldVersion, newVersion)
-
-	if !status {
+	if oldVersion.Major() > newVersion.Major() {
 		result = append(
 			result,
 			field.Invalid(
-				newImagePath,
-				newVersion,
-				fmt.Sprintf("can't upgrade between majors %v and %v",
-					oldVersion, newVersion)))
+				fieldPath,
+				fmt.Sprintf("%v", newVersion.Major()),
+				fmt.Sprintf("can't downgrade from majors %v to %v",
+					oldVersion.Major(), newVersion.Major())))
 	}
 
+	// TODO: Upgrading to versions 14 and 15 would require carrying information around about the collation used.
+	//   See https://git.postgresql.org/gitweb/?p=postgresql.git;a=commitdiff;h=9637badd9.
+	//   This is not implemented yet, and users should not upgrade to old versions anyway, so we are blocking it.
+	if oldVersion.Major() < newVersion.Major() && newVersion.Major() < 16 {
+		result = append(
+			result,
+			field.Invalid(
+				fieldPath,
+				fmt.Sprintf("%v", newVersion.Major()),
+				"major upgrades are only supported to version 16 or higher"))
+	}
 	return result
 }
 
diff --git a/pkg/management/postgres/initdb.go b/pkg/management/postgres/initdb.go
index 2675271a..a2c6ad2e 100644
--- a/pkg/management/postgres/initdb.go
+++ b/pkg/management/postgres/initdb.go
@@ -506,7 +506,7 @@ func (info InitInfo) Bootstrap(ctx context.Context) error {
 
 	// In case of import bootstrap, we restore the standard configuration file content
 	if isImportBootstrap {
-		/// Write standard replication configuration
+		// Write standard replication configuration
 		if _, err = configurePostgresOverrideConfFile(info.PgData, primaryConnInfo, ""); err != nil {
 			return fmt.Errorf("while configuring Postgres for replication: %w", err)
 		}
diff --git a/pkg/management/postgres/join.go b/pkg/management/postgres/join.go
index a341e54e..fda8bdcd 100644
--- a/pkg/management/postgres/join.go
+++ b/pkg/management/postgres/join.go
@@ -81,7 +81,7 @@ func (info InitInfo) Join(ctx context.Context, cluster *apiv1.Cluster) error {
 	if err != nil {
 		log.Warning(
 			"Error while parsing PostgreSQL server version to define connection options, defaulting to PostgreSQL 11",
-			"imageName", cluster.GetImageName(),
+			"image", cluster.Status.Image,
 			"err", err)
 	} else if pgVersion.Major() >= 12 {
 		// We explicitly disable wal_sender_timeout for join-related pg_basebackup executions.
diff --git a/pkg/reconciler/majorupgrade/doc.go b/pkg/reconciler/majorupgrade/doc.go
new file mode 100644
index 00000000..f75d3082
--- /dev/null
+++ b/pkg/reconciler/majorupgrade/doc.go
@@ -0,0 +1,30 @@
+/*
+Copyright © contributors to CloudNativePG, established as
+CloudNativePG a Series of LF Projects, LLC.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+
+SPDX-License-Identifier: Apache-2.0
+*/
+
+// Package majorupgrade provides the logic for upgrading a PostgreSQL cluster
+// to a new major version.
+//
+// The upgrade process consists of the following steps:
+//
+//  1. Delete all Pods in the cluster.
+//  2. Create and initiate the major upgrade job.
+//  3. Wait for the job to complete.
+//  4. If the upgrade job completes successfully, start new Pods for the upgraded version.
+//     Otherwise, stop and wait for input by the user.
+package majorupgrade
diff --git a/pkg/reconciler/majorupgrade/job.go b/pkg/reconciler/majorupgrade/job.go
new file mode 100644
index 00000000..f72690ab
--- /dev/null
+++ b/pkg/reconciler/majorupgrade/job.go
@@ -0,0 +1,87 @@
+/*
+Copyright © contributors to CloudNativePG, established as
+CloudNativePG a Series of LF Projects, LLC.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+
+SPDX-License-Identifier: Apache-2.0
+*/
+
+package majorupgrade
+
+import (
+	batchv1 "k8s.io/api/batch/v1"
+	corev1 "k8s.io/api/core/v1"
+
+	apiv1 "github.com/cloudnative-pg/cloudnative-pg/api/v1"
+	"github.com/cloudnative-pg/cloudnative-pg/pkg/specs"
+	"github.com/cloudnative-pg/cloudnative-pg/pkg/utils"
+)
+
+const jobMajorUpgrade = "major-upgrade"
+
+// isMajorUpgradeJob tells if the passed Job definition corresponds to
+// the job handling the major upgrade
+func isMajorUpgradeJob(job *batchv1.Job) bool {
+	return job.GetLabels()[utils.JobRoleLabelName] == string(jobMajorUpgrade)
+}
+
+// getTargetImageFromMajorUpgradeJob gets the image that is being used as
+// target of the major upgrade process.
+func getTargetImageFromMajorUpgradeJob(job *batchv1.Job) (string, bool) {
+	if !isMajorUpgradeJob(job) {
+		return "", false
+	}
+
+	for _, container := range job.Spec.Template.Spec.Containers {
+		if container.Name == string(jobMajorUpgrade) {
+			return container.Image, true
+		}
+	}
+
+	return "", false
+}
+
+// createMajorUpgradeJobDefinition creates a job to upgrade the primary node to a new Postgres major version
+func createMajorUpgradeJobDefinition(cluster *apiv1.Cluster, nodeSerial int) *batchv1.Job {
+	oldImage := *cluster.Status.MajorVersionUpgradeFromImage
+
+	prepareCommand := []string{
+		"/controller/manager",
+		"instance",
+		"upgrade",
+		"prepare",
+		"/controller/old",
+	}
+	oldVersionInitContainer := corev1.Container{
+		Name:            "prepare",
+		Image:           oldImage,
+		ImagePullPolicy: cluster.Spec.ImagePullPolicy,
+		Command:         prepareCommand,
+		VolumeMounts:    specs.CreatePostgresVolumeMounts(*cluster),
+		Resources:       cluster.Spec.Resources,
+		SecurityContext: specs.CreateContainerSecurityContext(cluster.GetSeccompProfile()),
+	}
+
+	majorUpgradeCommand := []string{
+		"/controller/manager",
+		"instance",
+		"upgrade",
+		"execute",
+		"/controller/old/bindir.txt",
+	}
+	job := specs.CreatePrimaryJob(*cluster, nodeSerial, jobMajorUpgrade, majorUpgradeCommand)
+	job.Spec.Template.Spec.InitContainers = append(job.Spec.Template.Spec.InitContainers, oldVersionInitContainer)
+
+	return job
+}
diff --git a/pkg/reconciler/majorupgrade/reconciler.go b/pkg/reconciler/majorupgrade/reconciler.go
new file mode 100644
index 00000000..ed150dc8
--- /dev/null
+++ b/pkg/reconciler/majorupgrade/reconciler.go
@@ -0,0 +1,286 @@
+/*
+Copyright © contributors to CloudNativePG, established as
+CloudNativePG a Series of LF Projects, LLC.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+
+SPDX-License-Identifier: Apache-2.0
+*/
+
+package majorupgrade
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"github.com/cloudnative-pg/machinery/pkg/log"
+	batchv1 "k8s.io/api/batch/v1"
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/utils/ptr"
+	ctrl "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+
+	apiv1 "github.com/cloudnative-pg/cloudnative-pg/api/v1"
+	"github.com/cloudnative-pg/cloudnative-pg/internal/configuration"
+	"github.com/cloudnative-pg/cloudnative-pg/pkg/resources/status"
+	"github.com/cloudnative-pg/cloudnative-pg/pkg/specs"
+	"github.com/cloudnative-pg/cloudnative-pg/pkg/utils"
+	"github.com/cloudnative-pg/cloudnative-pg/pkg/versions"
+)
+
+// ErrIncoherentMajorUpgradeJob is raised when the major upgrade job
+// is missing the target image
+var ErrIncoherentMajorUpgradeJob = fmt.Errorf("major upgrade job is missing the target image")
+
+// ErrNoPrimaryPVCFound is raised when the list of PVCs doesn't
+// include any primary instance.
+var ErrNoPrimaryPVCFound = fmt.Errorf("no primary PVC found")
+
+// Reconcile implements the major version upgrade logic.
+func Reconcile(
+	ctx context.Context,
+	c client.Client,
+	cluster *apiv1.Cluster,
+	instances []corev1.Pod,
+	pvcs []corev1.PersistentVolumeClaim,
+	jobs []batchv1.Job,
+) (*ctrl.Result, error) {
+	contextLogger := log.FromContext(ctx)
+
+	if majorUpgradeJob := getMajorUpdateJob(jobs); majorUpgradeJob != nil {
+		return majorVersionUpgradeHandleCompletion(ctx, c, cluster, majorUpgradeJob, pvcs)
+	}
+
+	if cluster.Status.MajorVersionUpgradeFromImage == nil {
+		return nil, nil
+	}
+
+	desiredVersion, err := cluster.GetPostgresqlVersion()
+	if err != nil {
+		contextLogger.Error(err, "Unable to retrieve the new PostgreSQL version")
+		return nil, err
+	}
+
+	primaryNodeSerial, err := getPrimarySerial(pvcs)
+	if err != nil || primaryNodeSerial == 0 {
+		contextLogger.Error(err, "Unable to retrieve the primary node serial")
+		return nil, err
+	}
+
+	contextLogger.Info("Reconciling in-place major version upgrades",
+		"primaryNodeSerial", primaryNodeSerial, "desiredVersion", desiredVersion.Major())
+
+	err = registerPhase(ctx, c, cluster, apiv1.PhaseMajorUpgrade,
+		fmt.Sprintf("Upgrading cluster to major version %v", desiredVersion.Major()))
+	if err != nil {
+		return nil, err
+	}
+
+	if result, err := deleteAllPodsInMajorUpgradePreparation(ctx, c, instances, jobs); err != nil {
+		contextLogger.Error(err, "Unable to delete pods and jobs in preparation for major upgrade")
+		return nil, err
+	} else if result != nil {
+		return result, err
+	}
+
+	if result, err := createMajorUpgradeJob(ctx, c, cluster, primaryNodeSerial); err != nil {
+		contextLogger.Error(err, "Unable to create major upgrade job")
+		return nil, err
+	} else if result != nil {
+		return result, err
+	}
+
+	return &ctrl.Result{RequeueAfter: 30 * time.Second}, nil
+}
+
+func getMajorUpdateJob(items []batchv1.Job) *batchv1.Job {
+	for _, job := range items {
+		if isMajorUpgradeJob(&job) {
+			return &job
+		}
+	}
+
+	return nil
+}
+
+func deleteAllPodsInMajorUpgradePreparation(
+	ctx context.Context,
+	c client.Client,
+	instances []corev1.Pod,
+	jobs []batchv1.Job,
+) (*ctrl.Result, error) {
+	foundSomethingToDelete := false
+
+	for _, pod := range instances {
+		if pod.GetDeletionTimestamp() != nil {
+			continue
+		}
+
+		foundSomethingToDelete = true
+		if err := c.Delete(ctx, &pod); err != nil {
+			return nil, err
+		}
+	}
+
+	for _, job := range jobs {
+		if job.GetDeletionTimestamp() != nil {
+			continue
+		}
+
+		foundSomethingToDelete = true
+		if err := c.Delete(ctx, &job, &client.DeleteOptions{
+			PropagationPolicy: ptr.To(metav1.DeletePropagationForeground),
+		}); err != nil {
+			return nil, err
+		}
+	}
+
+	if foundSomethingToDelete {
+		return &ctrl.Result{RequeueAfter: 30 * time.Second}, nil
+	}
+
+	return nil, nil
+}
+
+func createMajorUpgradeJob(
+	ctx context.Context,
+	c client.Client,
+	cluster *apiv1.Cluster,
+	primaryNodeSerial int,
+) (*ctrl.Result, error) {
+	contextLogger := log.FromContext(ctx)
+
+	job := createMajorUpgradeJobDefinition(cluster, primaryNodeSerial)
+
+	if err := ctrl.SetControllerReference(cluster, job, c.Scheme()); err != nil {
+		contextLogger.Error(err, "Unable to set the owner reference for major upgrade job")
+		return nil, err
+	}
+
+	utils.SetOperatorVersion(&job.ObjectMeta, versions.Version)
+	utils.InheritAnnotations(&job.ObjectMeta, cluster.Annotations,
+		cluster.GetFixedInheritedAnnotations(), configuration.Current)
+	utils.InheritAnnotations(&job.Spec.Template.ObjectMeta, cluster.Annotations,
+		cluster.GetFixedInheritedAnnotations(), configuration.Current)
+	utils.InheritLabels(&job.ObjectMeta, cluster.Labels,
+		cluster.GetFixedInheritedLabels(), configuration.Current)
+	utils.InheritLabels(&job.Spec.Template.ObjectMeta, cluster.Labels,
+		cluster.GetFixedInheritedLabels(), configuration.Current)
+	utils.SetInstanceRole(job.Spec.Template.ObjectMeta, specs.ClusterRoleLabelPrimary)
+
+	contextLogger.Info("Creating new major upgrade Job",
+		"jobName", job.Name,
+		"primary", true)
+
+	if err := c.Create(ctx, job); err != nil {
+		if errors.IsAlreadyExists(err) {
+			// This Job was already created, maybe the cache is stale.
+			return &ctrl.Result{RequeueAfter: 30 * time.Second}, nil
+		}
+		return nil, err
+	}
+
+	return nil, nil
+}
+
+func majorVersionUpgradeHandleCompletion(
+	ctx context.Context,
+	c client.Client,
+	cluster *apiv1.Cluster,
+	job *batchv1.Job,
+	pvcs []corev1.PersistentVolumeClaim,
+) (*ctrl.Result, error) {
+	contextLogger := log.FromContext(ctx)
+
+	if !utils.JobHasOneCompletion(*job) {
+		contextLogger.Info("Major upgrade job not completed.")
+		return &ctrl.Result{RequeueAfter: 30 * time.Second}, nil
+	}
+
+	for _, pvc := range pvcs {
+		if pvc.GetDeletionTimestamp() != nil {
+			continue
+		}
+
+		if specs.IsPrimary(pvc.ObjectMeta) {
+			continue
+		}
+
+		if err := c.Delete(ctx, &pvc); err != nil {
+			// Ignore if NotFound, otherwise report the error
+			if !errors.IsNotFound(err) {
+				return nil, err
+			}
+		}
+	}
+
+	jobImage, ok := getTargetImageFromMajorUpgradeJob(job)
+	if !ok {
+		return nil, ErrIncoherentMajorUpgradeJob
+	}
+
+	if err := status.PatchWithOptimisticLock(
+		ctx,
+		c,
+		cluster,
+		status.SetMajorVersionUpgradeFromImage(&jobImage),
+	); err != nil {
+		contextLogger.Error(err, "Unable to update cluster status after major upgrade completed.")
+		return nil, err
+	}
+
+	if err := c.Delete(ctx, job, &client.DeleteOptions{
+		PropagationPolicy: ptr.To(metav1.DeletePropagationForeground),
+	}); err != nil {
+		contextLogger.Error(err, "Unable to delete major upgrade job.")
+		return nil, err
+	}
+
+	return &ctrl.Result{Requeue: true}, nil
+}
+
+// registerPhase sets a phase into the cluster
+func registerPhase(
+	ctx context.Context,
+	c client.Client,
+	cluster *apiv1.Cluster,
+	phase string,
+	reason string,
+) error {
+	return status.PatchWithOptimisticLock(
+		ctx,
+		c,
+		cluster,
+		status.SetPhase(phase, reason),
+		status.SetClusterReadyCondition,
+	)
+}
+
+// getPrimarySerial tries to obtain the primary serial from a group of PVCs
+func getPrimarySerial(
+	pvcs []corev1.PersistentVolumeClaim,
+) (int, error) {
+	for _, pvc := range pvcs {
+		instanceRole, _ := utils.GetInstanceRole(pvc.ObjectMeta.Labels)
+		if instanceRole != specs.ClusterRoleLabelPrimary {
+			continue
+		}
+
+		return specs.GetNodeSerial(pvc.ObjectMeta)
+	}
+
+	return 0, ErrNoPrimaryPVCFound
+}
diff --git a/pkg/resources/status/patch.go b/pkg/resources/status/patch.go
index 51f8847e..5bc7e594 100644
--- a/pkg/resources/status/patch.go
+++ b/pkg/resources/status/patch.go
@@ -39,7 +39,7 @@ func PatchWithOptimisticLock(
 	ctx context.Context,
 	c client.Client,
 	cluster *apiv1.Cluster,
-	txs ...func(cluster *apiv1.Cluster),
+	txs ...Transaction,
 ) error {
 	if cluster == nil {
 		return nil
diff --git a/pkg/resources/status/transactions.go b/pkg/resources/status/transactions.go
index 590352b8..4c9f678f 100644
--- a/pkg/resources/status/transactions.go
+++ b/pkg/resources/status/transactions.go
@@ -26,9 +26,12 @@ import (
 	apiv1 "github.com/cloudnative-pg/cloudnative-pg/api/v1"
 )
 
-// SetClusterReadyConditionTX updates the cluster's readiness condition
+// Transaction is a function that modifies a cluster
+type Transaction func(cluster *apiv1.Cluster)
+
+// SetClusterReadyCondition updates the cluster's readiness condition
 // according to the cluster phase
-func SetClusterReadyConditionTX(cluster *apiv1.Cluster) {
+func SetClusterReadyCondition(cluster *apiv1.Cluster) {
 	if cluster.Status.Conditions == nil {
 		cluster.Status.Conditions = []metav1.Condition{}
 	}
@@ -52,10 +55,25 @@ func SetClusterReadyConditionTX(cluster *apiv1.Cluster) {
 	meta.SetStatusCondition(&cluster.Status.Conditions, condition)
 }
 
-// SetPhaseTX is a transaction that sets the cluster phase and reason
-func SetPhaseTX(phase string, reason string) func(cluster *apiv1.Cluster) {
+// SetPhase is a transaction that sets the cluster phase and reason
+func SetPhase(phase string, reason string) Transaction {
 	return func(cluster *apiv1.Cluster) {
 		cluster.Status.Phase = phase
 		cluster.Status.PhaseReason = reason
 	}
 }
+
+// SetImage is a transaction that sets the cluster image
+func SetImage(image string) Transaction {
+	return func(cluster *apiv1.Cluster) {
+		cluster.Status.Image = image
+	}
+}
+
+// SetMajorVersionUpgradeFromImage is a transaction that sets the cluster as upgrading to a newer major version
+// starting from the provided image
+func SetMajorVersionUpgradeFromImage(image *string) Transaction {
+	return func(cluster *apiv1.Cluster) {
+		cluster.Status.MajorVersionUpgradeFromImage = image
+	}
+}
diff --git a/pkg/specs/containers.go b/pkg/specs/containers.go
index ed18f04a..27ebfe3f 100644
--- a/pkg/specs/containers.go
+++ b/pkg/specs/containers.go
@@ -41,7 +41,7 @@ func createBootstrapContainer(cluster apiv1.Cluster) corev1.Container {
 			"bootstrap",
 			"/controller/manager",
 		},
-		VolumeMounts:    createPostgresVolumeMounts(cluster),
+		VolumeMounts:    CreatePostgresVolumeMounts(cluster),
 		Resources:       cluster.Spec.Resources,
 		SecurityContext: CreateContainerSecurityContext(cluster.GetSeccompProfile()),
 	}
diff --git a/pkg/specs/jobs.go b/pkg/specs/jobs.go
index b651d5cf..12ae77f2 100644
--- a/pkg/specs/jobs.go
+++ b/pkg/specs/jobs.go
@@ -89,7 +89,7 @@ func CreatePrimaryJobViaInitdb(cluster apiv1.Cluster, nodeSerial int) *batchv1.J
 	initCommand = append(initCommand, buildCommonInitJobFlags(cluster)...)
 
 	if cluster.Spec.Bootstrap.InitDB.Import != nil {
-		return createPrimaryJob(cluster, nodeSerial, jobRoleImport, initCommand)
+		return CreatePrimaryJob(cluster, nodeSerial, jobRoleImport, initCommand)
 	}
 
 	if cluster.ShouldInitDBRunPostInitApplicationSQLRefs() {
@@ -107,7 +107,7 @@ func CreatePrimaryJobViaInitdb(cluster apiv1.Cluster, nodeSerial int) *batchv1.J
 			"--post-init-sql-refs-folder", postInitSQLRefsFolder.toString())
 	}
 
-	return createPrimaryJob(cluster, nodeSerial, jobRoleInitDB, initCommand)
+	return CreatePrimaryJob(cluster, nodeSerial, jobRoleInitDB, initCommand)
 }
 
 func buildInitDBFlags(cluster apiv1.Cluster) (initCommand []string) {
@@ -196,7 +196,7 @@ func CreatePrimaryJobViaRestoreSnapshot(
 
 	initCommand = append(initCommand, buildCommonInitJobFlags(cluster)...)
 
-	job := createPrimaryJob(cluster, nodeSerial, jobRoleSnapshotRecovery, initCommand)
+	job := CreatePrimaryJob(cluster, nodeSerial, jobRoleSnapshotRecovery, initCommand)
 
 	addBarmanEndpointCAToJobFromCluster(cluster, backup, job)
 
@@ -213,7 +213,7 @@ func CreatePrimaryJobViaRecovery(cluster apiv1.Cluster, nodeSerial int, backup *
 
 	initCommand = append(initCommand, buildCommonInitJobFlags(cluster)...)
 
-	job := createPrimaryJob(cluster, nodeSerial, jobRoleFullRecovery, initCommand)
+	job := CreatePrimaryJob(cluster, nodeSerial, jobRoleFullRecovery, initCommand)
 
 	addBarmanEndpointCAToJobFromCluster(cluster, backup, job)
 
@@ -254,7 +254,7 @@ func CreatePrimaryJobViaPgBaseBackup(cluster apiv1.Cluster, nodeSerial int) *bat
 
 	initCommand = append(initCommand, buildCommonInitJobFlags(cluster)...)
 
-	return createPrimaryJob(cluster, nodeSerial, jobRolePGBaseBackup, initCommand)
+	return CreatePrimaryJob(cluster, nodeSerial, jobRolePGBaseBackup, initCommand)
 }
 
 // JoinReplicaInstance create a new PostgreSQL node, copying the contents from another Pod
@@ -268,7 +268,7 @@ func JoinReplicaInstance(cluster apiv1.Cluster, nodeSerial int) *batchv1.Job {
 
 	initCommand = append(initCommand, buildCommonInitJobFlags(cluster)...)
 
-	return createPrimaryJob(cluster, nodeSerial, jobRoleJoin, initCommand)
+	return CreatePrimaryJob(cluster, nodeSerial, jobRoleJoin, initCommand)
 }
 
 // RestoreReplicaInstance creates a new PostgreSQL replica starting from a volume snapshot backup
@@ -282,7 +282,7 @@ func RestoreReplicaInstance(cluster apiv1.Cluster, nodeSerial int) *batchv1.Job
 
 	initCommand = append(initCommand, buildCommonInitJobFlags(cluster)...)
 
-	job := createPrimaryJob(cluster, nodeSerial, jobRoleSnapshotRecovery, initCommand)
+	job := CreatePrimaryJob(cluster, nodeSerial, jobRoleSnapshotRecovery, initCommand)
 	return job
 }
 
@@ -308,25 +308,14 @@ const (
 	jobRoleSnapshotRecovery jobRole = "snapshot-recovery"
 )
 
-var jobRoleList = []jobRole{jobRoleImport, jobRoleInitDB, jobRolePGBaseBackup, jobRoleFullRecovery, jobRoleJoin}
-
 // getJobName returns a string indicating the job name
 func (role jobRole) getJobName(instanceName string) string {
 	return fmt.Sprintf("%s-%s", instanceName, role)
 }
 
-// GetPossibleJobNames get all the possible job names for a given instance
-func GetPossibleJobNames(instanceName string) []string {
-	res := make([]string, len(jobRoleList))
-	for idx, role := range jobRoleList {
-		res[idx] = role.getJobName(instanceName)
-	}
-	return res
-}
-
-// createPrimaryJob create a job that executes the provided command.
+// CreatePrimaryJob create a job that executes the provided command.
 // The role should describe the purpose of the executed job
-func createPrimaryJob(cluster apiv1.Cluster, nodeSerial int, role jobRole, initCommand []string) *batchv1.Job {
+func CreatePrimaryJob(cluster apiv1.Cluster, nodeSerial int, role jobRole, initCommand []string) *batchv1.Job {
 	instanceName := GetInstanceName(cluster.Name, nodeSerial)
 	jobName := role.getJobName(instanceName)
 
@@ -339,6 +328,7 @@ func createPrimaryJob(cluster apiv1.Cluster, nodeSerial int, role jobRole, initC
 			Labels: map[string]string{
 				utils.InstanceNameLabelName: instanceName,
 				utils.ClusterLabelName:      cluster.Name,
+				utils.JobRoleLabelName:      string(role),
 			},
 		},
 		Spec: batchv1.JobSpec{
@@ -359,12 +349,12 @@ func createPrimaryJob(cluster apiv1.Cluster, nodeSerial int, role jobRole, initC
 					Containers: []corev1.Container{
 						{
 							Name:            string(role),
-							Image:           cluster.GetImageName(),
+							Image:           cluster.Status.Image,
 							ImagePullPolicy: cluster.Spec.ImagePullPolicy,
 							Env:             envConfig.EnvVars,
 							EnvFrom:         envConfig.EnvFrom,
 							Command:         initCommand,
-							VolumeMounts:    createPostgresVolumeMounts(cluster),
+							VolumeMounts:    CreatePostgresVolumeMounts(cluster),
 							Resources:       cluster.Spec.Resources,
 							SecurityContext: CreateContainerSecurityContext(cluster.GetSeccompProfile()),
 						},
diff --git a/pkg/specs/pods.go b/pkg/specs/pods.go
index 01468628..85a4b673 100644
--- a/pkg/specs/pods.go
+++ b/pkg/specs/pods.go
@@ -211,11 +211,11 @@ func createPostgresContainers(cluster apiv1.Cluster, envConfig EnvConfig, enable
 	containers := []corev1.Container{
 		{
 			Name:            PostgresContainerName,
-			Image:           cluster.GetImageName(),
+			Image:           cluster.Status.Image,
 			ImagePullPolicy: cluster.Spec.ImagePullPolicy,
 			Env:             envConfig.EnvVars,
 			EnvFrom:         envConfig.EnvFrom,
-			VolumeMounts:    createPostgresVolumeMounts(cluster),
+			VolumeMounts:    CreatePostgresVolumeMounts(cluster),
 			// This is the default startup probe, and can be overridden
 			// the user configuration in cluster.spec.probes.startup
 			StartupProbe: &corev1.Probe{
diff --git a/pkg/specs/volumes.go b/pkg/specs/volumes.go
index 186040e2..4ac7dbdd 100644
--- a/pkg/specs/volumes.go
+++ b/pkg/specs/volumes.go
@@ -222,7 +222,9 @@ func createVolumesAndVolumeMountsForSQLRefs(
 	return volumes, volumeMounts
 }
 
-func createPostgresVolumeMounts(cluster apiv1.Cluster) []corev1.VolumeMount {
+// CreatePostgresVolumeMounts creates the volume mounts that are used
+// by PostgreSQL Pods
+func CreatePostgresVolumeMounts(cluster apiv1.Cluster) []corev1.VolumeMount {
 	volumeMounts := []corev1.VolumeMount{
 		{
 			Name:      "pgdata",



Test patch:
diff --git a/internal/controller/cluster_image_test.go b/internal/controller/cluster_image_test.go
new file mode 100644
index 00000000..ec69797b
--- /dev/null
+++ b/internal/controller/cluster_image_test.go
@@ -0,0 +1,214 @@
+/*
+Copyright © contributors to CloudNativePG, established as
+CloudNativePG a Series of LF Projects, LLC.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+
+SPDX-License-Identifier: Apache-2.0
+*/
+
+package controller
+
+import (
+	"context"
+
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/tools/record"
+	"sigs.k8s.io/controller-runtime/pkg/client/fake"
+
+	apiv1 "github.com/cloudnative-pg/cloudnative-pg/api/v1"
+	schemeBuilder "github.com/cloudnative-pg/cloudnative-pg/internal/scheme"
+
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+)
+
+func newFakeReconcilerFor(cluster *apiv1.Cluster, catalog *apiv1.ImageCatalog) *ClusterReconciler {
+	fakeClient := fake.NewClientBuilder().
+		WithScheme(schemeBuilder.BuildWithAllKnownScheme()).
+		WithRuntimeObjects(cluster).
+		WithStatusSubresource(cluster).
+		Build()
+
+	if catalog != nil {
+		_ = fakeClient.Create(context.Background(), catalog)
+	}
+
+	return &ClusterReconciler{
+		Client:   fakeClient,
+		Recorder: record.NewFakeRecorder(10),
+	}
+}
+
+var _ = Describe("Cluster image detection", func() {
+	It("gets the image from .spec.imageName", func(ctx SpecContext) {
+		// This is a simple situation, having a cluster with an
+		// explicit image. The image should be directly set into the
+		// status and the reconciliation loop can proceed.
+		// No major version upgrade have been requested.
+		cluster := &apiv1.Cluster{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      "cluster-example",
+				Namespace: "default",
+			},
+			Spec: apiv1.ClusterSpec{
+				ImageName: "postgres:15.2",
+			},
+		}
+		r := newFakeReconcilerFor(cluster, nil)
+
+		result, err := r.reconcileImage(ctx, cluster)
+		Expect(err).Error().ShouldNot(HaveOccurred())
+		Expect(result).To(BeNil())
+
+		Expect(cluster.Status.Image).To(Equal("postgres:15.2"))
+		Expect(cluster.Status.MajorVersionUpgradeFromImage).To(BeNil())
+	})
+
+	It("gets the image from an image catalog", func(ctx SpecContext) {
+		// This is slightly more complex, having an image catalog reference
+		// instead of an explicit image name. No major version upgrade have
+		// been requested, the reconciliation loop can proceed correctly
+		cluster := &apiv1.Cluster{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      "cluster-example",
+				Namespace: "default",
+			},
+			Spec: apiv1.ClusterSpec{
+				ImageCatalogRef: &apiv1.ImageCatalogRef{
+					TypedLocalObjectReference: corev1.TypedLocalObjectReference{
+						Name:     "catalog",
+						Kind:     "ImageCatalog",
+						APIGroup: &apiv1.SchemeGroupVersion.Group,
+					},
+					Major: 15,
+				},
+			},
+		}
+		catalog := &apiv1.ImageCatalog{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      "catalog",
+				Namespace: "default",
+			},
+			Spec: apiv1.ImageCatalogSpec{
+				Images: []apiv1.CatalogImage{
+					{
+						Image: "postgres:15.2",
+						Major: 15,
+					},
+				},
+			},
+		}
+
+		r := newFakeReconcilerFor(cluster, catalog)
+		result, err := r.reconcileImage(ctx, cluster)
+		Expect(err).Error().ShouldNot(HaveOccurred())
+		Expect(result).To(BeNil())
+
+		Expect(cluster.Status.Image).To(Equal("postgres:15.2"))
+		Expect(cluster.Status.MajorVersionUpgradeFromImage).To(BeNil())
+	})
+
+	It("gets the name from the image catalog, but the catalog is incomplete", func(ctx SpecContext) {
+		// As a variant of the previous case, the catalog may be
+		// incomplete and have no image for the selected major.  When
+		// this happens, the reconciliation loop should be stopped and
+		// the proper phase should be set.
+		cluster := &apiv1.Cluster{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      "cluster-example",
+				Namespace: "default",
+			},
+			Spec: apiv1.ClusterSpec{
+				ImageCatalogRef: &apiv1.ImageCatalogRef{
+					TypedLocalObjectReference: corev1.TypedLocalObjectReference{
+						Name:     "catalog",
+						Kind:     "ImageCatalog",
+						APIGroup: &apiv1.SchemeGroupVersion.Group,
+					},
+					Major: 15,
+				},
+			},
+		}
+		catalog := &apiv1.ImageCatalog{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      "catalog",
+				Namespace: "default",
+			},
+			Spec: apiv1.ImageCatalogSpec{
+				Images: []apiv1.CatalogImage{
+					{
+						Image: "postgres:11.2",
+						Major: 11,
+					},
+				},
+			},
+		}
+
+		r := newFakeReconcilerFor(cluster, catalog)
+		result, err := r.reconcileImage(ctx, cluster)
+		Expect(err).Error().ShouldNot(HaveOccurred())
+		Expect(result).ToNot(BeNil())
+
+		Expect(cluster.Status.Phase).To(Equal(apiv1.PhaseImageCatalogError))
+	})
+
+	It("skips major version downgrades", func(ctx SpecContext) {
+		cluster := &apiv1.Cluster{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      "cluster-example",
+				Namespace: "default",
+			},
+			Spec: apiv1.ClusterSpec{
+				ImageName: "postgres:15.2",
+			},
+			Status: apiv1.ClusterStatus{
+				Image: "postgres:16.2",
+			},
+		}
+
+		r := newFakeReconcilerFor(cluster, nil)
+
+		result, err := r.reconcileImage(ctx, cluster)
+		Expect(err).Error().ShouldNot(HaveOccurred())
+		Expect(result).To(BeNil())
+
+		Expect(cluster.Status.Image).To(Equal("postgres:16.2"))
+	})
+
+	It("process major version upgrades", func(ctx SpecContext) {
+		cluster := &apiv1.Cluster{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      "cluster-example",
+				Namespace: "default",
+			},
+			Spec: apiv1.ClusterSpec{
+				ImageName: "postgres:17.2",
+			},
+			Status: apiv1.ClusterStatus{
+				Image: "postgres:16.2",
+			},
+		}
+
+		r := newFakeReconcilerFor(cluster, nil)
+
+		result, err := r.reconcileImage(ctx, cluster)
+		Expect(err).Error().ShouldNot(HaveOccurred())
+		Expect(result).To(BeNil())
+
+		Expect(cluster.Status.Image).To(Equal("postgres:17.2"))
+		Expect(cluster.Status.MajorVersionUpgradeFromImage).ToNot(BeNil())
+		Expect(*cluster.Status.MajorVersionUpgradeFromImage).To(Equal("postgres:16.2"))
+	})
+})
diff --git a/internal/controller/cluster_scale_test.go b/internal/controller/cluster_scale_test.go
index 2c9f24e5..7ee3ed3b 100644
--- a/internal/controller/cluster_scale_test.go
+++ b/internal/controller/cluster_scale_test.go
@@ -33,7 +33,7 @@ import (
 	apiv1 "github.com/cloudnative-pg/cloudnative-pg/api/v1"
 	schemeBuilder "github.com/cloudnative-pg/cloudnative-pg/internal/scheme"
 	"github.com/cloudnative-pg/cloudnative-pg/pkg/reconciler/persistentvolumeclaim"
-	"github.com/cloudnative-pg/cloudnative-pg/pkg/specs"
+	"github.com/cloudnative-pg/cloudnative-pg/pkg/utils"
 
 	. "github.com/onsi/ginkgo/v2"
 	. "github.com/onsi/gomega"
@@ -172,35 +172,58 @@ var _ = Describe("cluster scale pod and job deletion logic", func() {
 	)
 
 	BeforeEach(func() {
-		fakeClientSet = fake.NewClientBuilder().WithScheme(schemeBuilder.BuildWithAllKnownScheme()).Build()
+		fakeClientSet = fake.
+			NewClientBuilder().
+			WithScheme(schemeBuilder.BuildWithAllKnownScheme()).
+			WithIndex(&batchv1.Job{}, jobOwnerKey, jobOwnerIndexFunc).
+			Build()
 		ctx, cancel = context.WithCancel(context.Background())
 
 		reconciler = &ClusterReconciler{
 			Client: fakeClientSet,
 		}
 
+		instanceName = "test-instance"
+
 		cluster = &apiv1.Cluster{
 			ObjectMeta: metav1.ObjectMeta{
 				Name:      "test-cluster",
 				Namespace: "default",
 			},
 		}
-
-		instanceName = "test-instance"
+		cluster.TypeMeta = metav1.TypeMeta{
+			Kind:       apiv1.ClusterKind,
+			APIVersion: apiv1.SchemeGroupVersion.String(),
+		}
 	})
 
 	AfterEach(func() {
 		cancel()
 	})
 
+	It("creates the cluster", func(ctx SpecContext) {
+		err := fakeClientSet.Create(ctx, cluster)
+		Expect(err).ToNot(HaveOccurred())
+	})
+
 	It("should delete all the jobs", func(ctx SpecContext) {
-		for _, jobName := range specs.GetPossibleJobNames(instanceName) {
+		jobNames := []string{
+			cluster.Name + "-initdb",
+			cluster.Name + "-pgbasebackup",
+		}
+		for _, jobName := range jobNames {
 			job := &batchv1.Job{
 				ObjectMeta: metav1.ObjectMeta{
 					Name:      jobName,
 					Namespace: cluster.Namespace,
+					Labels: map[string]string{
+						utils.InstanceNameLabelName: instanceName,
+						utils.ClusterLabelName:      cluster.Name,
+						utils.JobRoleLabelName:      "test",
+					},
 				},
 			}
+			cluster.SetInheritedDataAndOwnership(&job.ObjectMeta)
 			err := fakeClientSet.Create(ctx, job)
 			Expect(err).NotTo(HaveOccurred())
 		}
@@ -208,7 +231,7 @@ var _ = Describe("cluster scale pod and job deletion logic", func() {
 		err := reconciler.ensureInstanceJobAreDeleted(ctx, cluster, instanceName)
 		Expect(err).NotTo(HaveOccurred())
 
-		for _, jobName := range specs.GetPossibleJobNames(instanceName) {
+		for _, jobName := range jobNames {
 			var expectedJob batchv1.Job
 			err = fakeClientSet.Get(context.Background(),
 				types.NamespacedName{Name: jobName, Namespace: cluster.Namespace},
diff --git a/internal/controller/cluster_upgrade_test.go b/internal/controller/cluster_upgrade_test.go
index 8737fb1a..cda4e66f 100644
--- a/internal/controller/cluster_upgrade_test.go
+++ b/internal/controller/cluster_upgrade_test.go
@@ -55,6 +55,9 @@ var _ = Describe("Pod upgrade", Ordered, func() {
 			Spec: apiv1.ClusterSpec{
 				ImageName: "postgres:13.11",
 			},
+			Status: apiv1.ClusterStatus{
+				Image: "postgres:13.11",
+			},
 		}
 		configuration.Current = configuration.NewConfiguration()
 	})
diff --git a/internal/controller/suite_test.go b/internal/controller/suite_test.go
index 8315875b..d567ba4d 100644
--- a/internal/controller/suite_test.go
+++ b/internal/controller/suite_test.go
@@ -77,6 +77,7 @@ func buildTestEnvironment() *testingEnvironment {
 	k8sClient := fake.NewClientBuilder().WithScheme(scheme).
 		WithStatusSubresource(&apiv1.Cluster{}, &apiv1.Backup{}, &apiv1.Pooler{}, &corev1.Service{},
 			&corev1.ConfigMap{}, &corev1.Secret{}).
+		WithIndex(&batchv1.Job{}, jobOwnerKey, jobOwnerIndexFunc).
 		Build()
 	Expect(err).ToNot(HaveOccurred())
 
diff --git a/internal/webhook/v1/cluster_webhook_test.go b/internal/webhook/v1/cluster_webhook_test.go
index 6edc5a16..ae22f9d0 100644
--- a/internal/webhook/v1/cluster_webhook_test.go
+++ b/internal/webhook/v1/cluster_webhook_test.go
@@ -1245,7 +1245,7 @@ var _ = Describe("validate image name change", func() {
 		})
 	})
 	Context("using image catalog", func() {
-		It("complains on major upgrades", func() {
+		It("complains on major downgrades", func() {
 			clusterOld := &apiv1.Cluster{
 				Spec: apiv1.ClusterSpec{
 					ImageCatalogRef: &apiv1.ImageCatalogRef{
@@ -1253,7 +1253,7 @@ var _ = Describe("validate image name change", func() {
 							Name: "test",
 							Kind: "ImageCatalog",
 						},
-						Major: 15,
+						Major: 16,
 					},
 				},
 			}
@@ -1264,7 +1264,7 @@ var _ = Describe("validate image name change", func() {
 							Name: "test",
 							Kind: "ImageCatalog",
 						},
-						Major: 16,
+						Major: 15,
 					},
 				},
 			}
@@ -1291,10 +1291,10 @@ var _ = Describe("validate image name change", func() {
 			}
 			Expect(v.validateImageChange(clusterNew, clusterOld)).To(BeEmpty())
 		})
-		It("complains on major upgrades", func() {
+		It("complains on major downgrades", func() {
 			clusterOld := &apiv1.Cluster{
 				Spec: apiv1.ClusterSpec{
-					ImageName: "postgres:16.1",
+					ImageName: "postgres:17.1",
 				},
 			}
 			clusterNew := &apiv1.Cluster{
@@ -1304,7 +1304,7 @@ var _ = Describe("validate image name change", func() {
 							Name: "test",
 							Kind: "ImageCatalog",
 						},
-						Major: 17,
+						Major: 16,
 					},
 				},
 			}
@@ -1369,7 +1369,7 @@ var _ = Describe("validate image name change", func() {
 			}
 			Expect(v.validateImageChange(clusterNew, clusterOld)).To(BeEmpty())
 		})
-		It("complains on major upgrades", func() {
+		It("complains on major downgrades", func() {
 			clusterOld := &apiv1.Cluster{
 				Spec: apiv1.ClusterSpec{
 					ImageCatalogRef: &apiv1.ImageCatalogRef{
@@ -1377,18 +1377,18 @@ var _ = Describe("validate image name change", func() {
 							Name: "test",
 							Kind: "ImageCatalog",
 						},
-						Major: 16,
+						Major: 17,
 					},
 				},
 			}
 			clusterNew := &apiv1.Cluster{
 				Spec: apiv1.ClusterSpec{
-					ImageName: "postgres:17.1",
+					ImageName: "postgres:16.1",
 				},
 			}
 			Expect(v.validateImageChange(clusterNew, clusterOld)).To(HaveLen(1))
 		})
-		It("complains going from imageCatalogRef to different major default imageName", func() {
+		It("complains going from imageCatalogRef to lower major default imageName", func() {
 			clusterOld := &apiv1.Cluster{
 				Spec: apiv1.ClusterSpec{
 					ImageCatalogRef: &apiv1.ImageCatalogRef{
@@ -1396,7 +1396,7 @@ var _ = Describe("validate image name change", func() {
 							Name: "test",
 							Kind: "ImageCatalog",
 						},
-						Major: 16,
+						Major: 18,
 					},
 				},
 			}
diff --git a/pkg/reconciler/majorupgrade/job_test.go b/pkg/reconciler/majorupgrade/job_test.go
new file mode 100644
index 00000000..6f15aee5
--- /dev/null
+++ b/pkg/reconciler/majorupgrade/job_test.go
@@ -0,0 +1,72 @@
+/*
+Copyright © contributors to CloudNativePG, established as
+CloudNativePG a Series of LF Projects, LLC.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+
+SPDX-License-Identifier: Apache-2.0
+*/
+
+package majorupgrade
+
+import (
+	batchv1 "k8s.io/api/batch/v1"
+
+	apiv1 "github.com/cloudnative-pg/cloudnative-pg/api/v1"
+	"github.com/cloudnative-pg/cloudnative-pg/pkg/specs"
+
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+)
+
+var _ = Describe("Major upgrade Job generation", func() {
+	oldImageName := "postgres:16"
+	newImageName := "postgres:17"
+
+	cluster := apiv1.Cluster{
+		Spec: apiv1.ClusterSpec{
+			ImageName: newImageName,
+			Bootstrap: &apiv1.BootstrapConfiguration{
+				InitDB: &apiv1.BootstrapInitDB{},
+			},
+		},
+		Status: apiv1.ClusterStatus{
+			Image:                        newImageName,
+			MajorVersionUpgradeFromImage: &oldImageName,
+		},
+	}
+
+	It("creates major upgrade jobs", func() {
+		majorUpgradeJob := createMajorUpgradeJobDefinition(&cluster, 1)
+		Expect(majorUpgradeJob).ToNot(BeNil())
+		Expect(majorUpgradeJob.Spec.Template.Spec.Containers[0].Image).To(Equal(newImageName))
+	})
+
+	It("is able to discover which target image was used", func() {
+		majorUpgradeJob := createMajorUpgradeJobDefinition(&cluster, 1)
+		Expect(majorUpgradeJob).ToNot(BeNil())
+
+		imgName, found := getTargetImageFromMajorUpgradeJob(majorUpgradeJob)
+		Expect(found).To(BeTrue())
+		Expect(imgName).To(Equal(newImageName))
+	})
+
+	DescribeTable(
+		"Tells major upgrade jobs apart from jobs of other types",
+		func(job *batchv1.Job, isMajorUpgrade bool) {
+			Expect(isMajorUpgradeJob(job)).To(Equal(isMajorUpgrade))
+		},
+		Entry("initdb jobs are not major upgrades", specs.CreatePrimaryJobViaInitdb(cluster, 1), false),
+		Entry("major-upgrade jobs are major upgrades", createMajorUpgradeJobDefinition(&cluster, 1), true),
+	)
+})
diff --git a/pkg/reconciler/majorupgrade/reconciler_test.go b/pkg/reconciler/majorupgrade/reconciler_test.go
new file mode 100644
index 00000000..ea7cdfb8
--- /dev/null
+++ b/pkg/reconciler/majorupgrade/reconciler_test.go
@@ -0,0 +1,229 @@
+/*
+Copyright © contributors to CloudNativePG, established as
+CloudNativePG a Series of LF Projects, LLC.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+
+SPDX-License-Identifier: Apache-2.0
+*/
+
+package majorupgrade
+
+import (
+	"fmt"
+
+	batchv1 "k8s.io/api/batch/v1"
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/utils/ptr"
+	ctrl "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/client/fake"
+
+	apiv1 "github.com/cloudnative-pg/cloudnative-pg/api/v1"
+	schemeBuilder "github.com/cloudnative-pg/cloudnative-pg/internal/scheme"
+	"github.com/cloudnative-pg/cloudnative-pg/pkg/specs"
+	"github.com/cloudnative-pg/cloudnative-pg/pkg/utils"
+
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+)
+
+var _ = Describe("Major upgrade job status reconciliation", func() {
+	It("waits until the job completed", func(ctx SpecContext) {
+		job := buildRunningUpgradeJob()
+		cluster := &apiv1.Cluster{}
+		fakeClient := fake.NewClientBuilder().
+			WithScheme(schemeBuilder.BuildWithAllKnownScheme()).
+			WithRuntimeObjects(job, cluster).
+			WithStatusSubresource(cluster).
+			Build()
+
+		result, err := majorVersionUpgradeHandleCompletion(ctx, fakeClient, cluster, job, nil)
+		Expect(err).ToNot(HaveOccurred())
+		Expect(result).ToNot(BeNil())
+
+		// the job has not been deleted
+		Expect(job.ObjectMeta.DeletionTimestamp).To(BeNil())
+	})
+
+	It("deletes the replica PVCs when and makes the cluster use the new image", func(ctx SpecContext) {
+		job := buildCompletedUpgradeJob()
+		cluster := &apiv1.Cluster{
+			ObjectMeta: metav1.ObjectMeta{
+				Name: "cluster-example",
+			},
+		}
+		pvcs := []corev1.PersistentVolumeClaim{
+			buildPrimaryPVC(1),
+			buildReplicaPVC(2),
+			buildReplicaPVC(3),
+		}
+
+		objects := []runtime.Object{
+			job,
+			cluster,
+		}
+		for i := range pvcs {
+			objects = append(objects, &pvcs[i])
+		}
+		fakeClient := fake.NewClientBuilder().
+			WithScheme(schemeBuilder.BuildWithAllKnownScheme()).
+			WithRuntimeObjects(objects...).
+			WithStatusSubresource(cluster).
+			Build()
+
+		result, err := majorVersionUpgradeHandleCompletion(ctx, fakeClient, cluster, job, pvcs)
+		Expect(err).ToNot(HaveOccurred())
+		Expect(result).ToNot(BeNil())
+		Expect(*result).To(Equal(ctrl.Result{Requeue: true}))
+
+		// the replica PVCs have been deleted
+		for i := range pvcs {
+			if !specs.IsPrimary(pvcs[i].ObjectMeta) {
+				var pvc corev1.PersistentVolumeClaim
+				err := fakeClient.Get(ctx, client.ObjectKeyFromObject(&pvcs[i]), &pvc)
+				Expect(err).To(MatchError(errors.IsNotFound, "is not found"))
+			}
+		}
+
+		// the upgrade has been marked as done
+		Expect(cluster.Status.MajorVersionUpgradeFromImage).ToNot(BeNil())
+		Expect(*cluster.Status.MajorVersionUpgradeFromImage).To(Equal("postgres:16"))
+
+		// the job has been deleted
+		var tempJob batchv1.Job
+		err = fakeClient.Get(ctx, client.ObjectKeyFromObject(job), &tempJob)
+		Expect(err).To(MatchError(errors.IsNotFound, "is not found"))
+	})
+})
+
+var _ = Describe("Major upgrade job decoding", func() {
+	It("is able to find the target image", func() {
+		job := buildCompletedUpgradeJob()
+		imageName, ok := getTargetImageFromMajorUpgradeJob(job)
+		Expect(ok).To(BeTrue())
+		Expect(imageName).To(Equal("postgres:16"))
+	})
+})
+
+var _ = Describe("PVC metadata decoding", func() {
+	It("is able to find the serial number of the primary server", func() {
+		pvcs := []corev1.PersistentVolumeClaim{
+			buildReplicaPVC(1),
+			buildPrimaryPVC(2),
+		}
+
+		Expect(getPrimarySerial(pvcs)).To(Equal(2))
+	})
+
+	It("raises an error if no primary PVC is found", func() {
+		pvcs := []corev1.PersistentVolumeClaim{
+			buildReplicaPVC(1),
+			buildReplicaPVC(2),
+		}
+
+		Expect(getPrimarySerial(pvcs)).Error().To(BeEquivalentTo(ErrNoPrimaryPVCFound))
+	})
+
+	It("raises an error if the primary PVC has an invalid serial", func() {
+		pvcs := []corev1.PersistentVolumeClaim{
+			buildReplicaPVC(1),
+			buildInvalidPrimaryPVC(2),
+		}
+
+		Expect(getPrimarySerial(pvcs)).Error().To(HaveOccurred())
+	})
+})
+
+func buildPrimaryPVC(serial int) corev1.PersistentVolumeClaim {
+	return corev1.PersistentVolumeClaim{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: fmt.Sprintf("cluster-example-%d", serial),
+			Labels: map[string]string{
+				utils.ClusterRoleLabelName: specs.ClusterRoleLabelPrimary,
+			},
+			Annotations: map[string]string{
+				utils.ClusterSerialAnnotationName: fmt.Sprintf("%v", serial),
+			},
+		},
+	}
+}
+
+func buildInvalidPrimaryPVC(serial int) corev1.PersistentVolumeClaim {
+	return corev1.PersistentVolumeClaim{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: fmt.Sprintf("cluster-example-%d", serial),
+			Labels: map[string]string{
+				utils.ClusterRoleLabelName: specs.ClusterRoleLabelPrimary,
+			},
+			Annotations: map[string]string{
+				utils.ClusterSerialAnnotationName: fmt.Sprintf("%v - this is a test", serial),
+			},
+		},
+	}
+}
+
+func buildReplicaPVC(serial int) corev1.PersistentVolumeClaim {
+	return corev1.PersistentVolumeClaim{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: fmt.Sprintf("cluster-example-%d", serial),
+			Labels: map[string]string{
+				utils.ClusterRoleLabelName: specs.ClusterRoleLabelReplica,
+			},
+			Annotations: map[string]string{
+				utils.ClusterSerialAnnotationName: fmt.Sprintf("%v", serial),
+			},
+		},
+	}
+}
+
+func buildCompletedUpgradeJob() *batchv1.Job {
+	return &batchv1.Job{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "cluster-example-major-upgrade",
+			Labels: map[string]string{
+				utils.JobRoleLabelName: jobMajorUpgrade,
+			},
+		},
+		Spec: batchv1.JobSpec{
+			Completions: ptr.To[int32](1),
+			Template: corev1.PodTemplateSpec{
+				Spec: corev1.PodSpec{
+					Containers: []corev1.Container{
+						{
+							Name:  jobMajorUpgrade,
+							Image: "postgres:16",
+						},
+					},
+				},
+			},
+		},
+		Status: batchv1.JobStatus{
+			Succeeded: 1,
+		},
+	}
+}
+
+func buildRunningUpgradeJob() *batchv1.Job {
+	return &batchv1.Job{
+		Spec: batchv1.JobSpec{
+			Completions: ptr.To[int32](1),
+		},
+		Status: batchv1.JobStatus{
+			Succeeded: 0,
+		},
+	}
+}
diff --git a/pkg/reconciler/majorupgrade/suite_test.go b/pkg/reconciler/majorupgrade/suite_test.go
new file mode 100644
index 00000000..a4f8479d
--- /dev/null
+++ b/pkg/reconciler/majorupgrade/suite_test.go
@@ -0,0 +1,32 @@
+/*
+Copyright © contributors to CloudNativePG, established as
+CloudNativePG a Series of LF Projects, LLC.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+
+SPDX-License-Identifier: Apache-2.0
+*/
+
+package majorupgrade
+
+import (
+	"testing"
+
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+)
+
+func TestMajorUpgrade(t *testing.T) {
+	RegisterFailHandler(Fail)
+	RunSpecs(t, "Major upgrade reconciler")
+}
diff --git a/pkg/specs/pg_pods_test.go b/pkg/specs/pg_pods_test.go
index b41a3833..982be12a 100644
--- a/pkg/specs/pg_pods_test.go
+++ b/pkg/specs/pg_pods_test.go
@@ -37,6 +37,9 @@ var _ = Describe("Extract the used image name", func() {
 			Name:      "clusterName",
 			Namespace: "default",
 		},
+		Status: apiv1.ClusterStatus{
+			Image: configuration.Current.PostgresImageName,
+		},
 	}
 	pod, err := NewInstance(context.TODO(), cluster, 1, true)
 	Expect(err).ToNot(HaveOccurred())
diff --git a/pkg/specs/pods_test.go b/pkg/specs/pods_test.go
index dcb69dc1..ac2eba31 100644
--- a/pkg/specs/pods_test.go
+++ b/pkg/specs/pods_test.go
@@ -941,6 +941,9 @@ var _ = Describe("NewInstance", func() {
 					utils.PodPatchAnnotationName: `[{"op": "replace", "path": "/spec/containers/0/image", "value": "new-image:latest"}]`, // nolint: lll
 				},
 			},
+			Status: v1.ClusterStatus{
+				Image: "test",
+			},
 		}
 
 		pod, err := NewInstance(ctx, cluster, 1, true)
diff --git a/pkg/specs/volumes_test.go b/pkg/specs/volumes_test.go
index ce2b59dd..6370da93 100644
--- a/pkg/specs/volumes_test.go
+++ b/pkg/specs/volumes_test.go
@@ -298,7 +298,7 @@ var _ = Describe("test createVolumesAndVolumeMountsForSQLRefs", func() {
 
 var _ = DescribeTable("test creation of volume mounts",
 	func(cluster apiv1.Cluster, mounts []corev1.VolumeMount) {
-		mts := createPostgresVolumeMounts(cluster)
+		mts := CreatePostgresVolumeMounts(cluster)
 		Expect(mts).NotTo(BeEmpty())
 		for _, mt := range mounts {
 			Expect(mts).To(ContainElement(mt))
diff --git a/tests/e2e/cluster_major_upgrade_test.go b/tests/e2e/cluster_major_upgrade_test.go
new file mode 100644
index 00000000..2386e40a
--- /dev/null
+++ b/tests/e2e/cluster_major_upgrade_test.go
@@ -0,0 +1,356 @@
+/*
+Copyright © contributors to CloudNativePG, established as
+CloudNativePG a Series of LF Projects, LLC.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+
+SPDX-License-Identifier: Apache-2.0
+*/
+
+package e2e
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"strconv"
+	"time"
+
+	"github.com/cloudnative-pg/machinery/pkg/image/reference"
+	"github.com/cloudnative-pg/machinery/pkg/postgres/version"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/apimachinery/pkg/util/rand"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+
+	v1 "github.com/cloudnative-pg/cloudnative-pg/api/v1"
+	"github.com/cloudnative-pg/cloudnative-pg/pkg/specs"
+	"github.com/cloudnative-pg/cloudnative-pg/pkg/utils"
+	"github.com/cloudnative-pg/cloudnative-pg/pkg/versions"
+	"github.com/cloudnative-pg/cloudnative-pg/tests"
+	"github.com/cloudnative-pg/cloudnative-pg/tests/utils/clusterutils"
+	"github.com/cloudnative-pg/cloudnative-pg/tests/utils/environment"
+	"github.com/cloudnative-pg/cloudnative-pg/tests/utils/exec"
+	"github.com/cloudnative-pg/cloudnative-pg/tests/utils/objects"
+	"github.com/cloudnative-pg/cloudnative-pg/tests/utils/postgres"
+	"github.com/cloudnative-pg/cloudnative-pg/tests/utils/storage"
+	"github.com/cloudnative-pg/cloudnative-pg/tests/utils/timeouts"
+
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+)
+
+var _ = Describe("Postgres Major Upgrade", Label(tests.LabelPostgresMajorUpgrade), func() {
+	const (
+		level                  = tests.Medium
+		namespacePrefix        = "cluster-major-upgrade"
+		postgisEntry           = "postgis"
+		postgresqlEntry        = "postgresql"
+		postgresqlMinimalEntry = "postgresql-minimal"
+	)
+
+	var namespace string
+
+	type scenario struct {
+		startingCluster *v1.Cluster
+		startingMajor   int
+		targetImage     string
+		targetMajor     int
+	}
+	scenarios := map[string]*scenario{}
+
+	generateBaseCluster := func(namespace string, storageClass string) *v1.Cluster {
+		return &v1.Cluster{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      "pg-major-upgrade",
+				Namespace: namespace,
+			},
+			Spec: v1.ClusterSpec{
+				Instances: 3,
+				StorageConfiguration: v1.StorageConfiguration{
+					StorageClass: &storageClass,
+					Size:         "1Gi",
+				},
+				WalStorage: &v1.StorageConfiguration{
+					StorageClass: &storageClass,
+					Size:         "1Gi",
+				},
+				PostgresConfiguration: v1.PostgresConfiguration{
+					Parameters: map[string]string{
+						"log_checkpoints":             "on",
+						"log_lock_waits":              "on",
+						"log_min_duration_statement":  "1000",
+						"log_statement":               "ddl",
+						"log_temp_files":              "1024",
+						"log_autovacuum_min_duration": "1000",
+						"log_replication_commands":    "on",
+					},
+				},
+			},
+		}
+	}
+
+	generatePostgreSQLCluster := func(namespace string, storageClass string, majorVersion int) *v1.Cluster {
+		cluster := generateBaseCluster(namespace, storageClass)
+		cluster.Spec.ImageName = "ghcr.io/cloudnative-pg/postgresql:" + strconv.Itoa(majorVersion)
+		cluster.Spec.Bootstrap = &v1.BootstrapConfiguration{
+			InitDB: &v1.BootstrapInitDB{
+				PostInitSQL: []string{
+					"CREATE EXTENSION IF NOT EXISTS pg_stat_statements;",
+					"CREATE EXTENSION IF NOT EXISTS pg_trgm;",
+				},
+			},
+		}
+		cluster.Spec.PostgresConfiguration.Parameters["pg_stat_statements.track"] = "top"
+		return cluster
+	}
+	generatePostgreSQLMinimalCluster := func(namespace string, storageClass string, majorVersion int) *v1.Cluster {
+		cluster := generatePostgreSQLCluster(namespace, storageClass, majorVersion)
+		cluster.Spec.ImageName = fmt.Sprintf("ghcr.io/cloudnative-pg/postgresql:%d-minimal-bookworm", majorVersion)
+		return cluster
+	}
+
+	generatePostGISCluster := func(namespace string, storageClass string, majorVersion int) *v1.Cluster {
+		cluster := generateBaseCluster(namespace, storageClass)
+		cluster.Spec.ImageName = "ghcr.io/cloudnative-pg/postgis:" + strconv.Itoa(majorVersion)
+		cluster.Spec.Bootstrap = &v1.BootstrapConfiguration{
+			InitDB: &v1.BootstrapInitDB{
+				PostInitApplicationSQL: []string{
+					"CREATE EXTENSION postgis",
+					"CREATE EXTENSION postgis_raster",
+					"CREATE EXTENSION postgis_sfcgal",
+					"CREATE EXTENSION fuzzystrmatch",
+					"CREATE EXTENSION address_standardizer",
+					"CREATE EXTENSION address_standardizer_data_us",
+					"CREATE EXTENSION postgis_tiger_geocoder",
+					"CREATE EXTENSION postgis_topology",
+					"CREATE TABLE geometries (name varchar, geom geometry)",
+					"INSERT INTO geometries VALUES" +
+						" ('Point', 'POINT(0 0)')," +
+						" ('Linestring', 'LINESTRING(0 0, 1 1, 2 1, 2 2)')," +
+						" ('Polygon', 'POLYGON((0 0, 1 0, 1 1, 0 1, 0 0))')," +
+						" ('PolygonWithHole', 'POLYGON((0 0, 10 0, 10 10, 0 10, 0 0),(1 1, 1 2, 2 2, 2 1, 1 1))')," +
+						" ('Collection', 'GEOMETRYCOLLECTION(POINT(2 0),POLYGON((0 0, 1 0, 1 1, 0 1, 0 0)))');",
+				},
+			},
+		}
+		return cluster
+	}
+
+	determineVersionsForTesting := func() (uint64, uint64) {
+		currentImage := os.Getenv("POSTGRES_IMG")
+		Expect(currentImage).ToNot(BeEmpty())
+
+		currentVersion, err := version.FromTag(reference.New(currentImage).Tag)
+		Expect(err).NotTo(HaveOccurred())
+		currentMajor := currentVersion.Major()
+
+		targetVersion, err := version.FromTag(reference.New(versions.DefaultImageName).Tag)
+		Expect(err).ToNot(HaveOccurred())
+		targetMajor := targetVersion.Major()
+
+		// If same version, choose a previous one for testing
+		if currentMajor == targetMajor {
+			currentMajor = targetMajor - (uint64(rand.Int() % 4)) - 1
+			GinkgoWriter.Printf("Using %v as the current major version instead.\n", currentMajor)
+		}
+
+		return currentMajor, targetMajor
+	}
+
+	buildScenarios := func(
+		namespace string, storageClass string, currentMajor, targetMajor uint64,
+	) map[string]*scenario {
+		return map[string]*scenario{
+			postgisEntry: {
+				startingCluster: generatePostGISCluster(namespace, storageClass, int(currentMajor)),
+				startingMajor:   int(currentMajor),
+				targetImage:     fmt.Sprintf("ghcr.io/cloudnative-pg/postgis:%v", targetMajor),
+				targetMajor:     int(targetMajor),
+			},
+			postgresqlEntry: {
+				startingCluster: generatePostgreSQLCluster(namespace, storageClass, int(currentMajor)),
+				startingMajor:   int(currentMajor),
+				targetImage:     fmt.Sprintf("ghcr.io/cloudnative-pg/postgresql:%v", targetMajor),
+				targetMajor:     int(targetMajor),
+			},
+			postgresqlMinimalEntry: {
+				startingCluster: generatePostgreSQLMinimalCluster(namespace, storageClass, int(currentMajor)),
+				startingMajor:   int(currentMajor),
+				targetImage:     fmt.Sprintf("ghcr.io/cloudnative-pg/postgresql:%v-minimal-bookworm", targetMajor),
+				targetMajor:     int(targetMajor),
+			},
+		}
+	}
+
+	verifyPodsChanged := func(
+		ctx context.Context, client client.Client, cluster *v1.Cluster, oldPodsUUIDs []types.UID,
+	) {
+		Eventually(func(g Gomega) {
+			podList, err := clusterutils.ListPods(ctx, client, cluster.Name, cluster.Namespace)
+			g.Expect(err).ToNot(HaveOccurred())
+			g.Expect(podList.Items).To(HaveLen(len(oldPodsUUIDs)))
+			for _, pod := range podList.Items {
+				g.Expect(oldPodsUUIDs).NotTo(ContainElement(pod.UID))
+			}
+		}).WithTimeout(5 * time.Minute).WithPolling(10 * time.Second).Should(Succeed())
+	}
+
+	verifyPVCsChanged := func(
+		ctx context.Context, client client.Client, cluster *v1.Cluster, oldPVCsUUIDs []types.UID,
+	) {
+		Eventually(func(g Gomega) {
+			pvcList, err := storage.GetPVCList(ctx, client, cluster.Namespace)
+			g.Expect(err).ToNot(HaveOccurred())
+			g.Expect(pvcList.Items).To(HaveLen(len(oldPVCsUUIDs)))
+			for _, pvc := range pvcList.Items {
+				if pvc.Labels[utils.ClusterInstanceRoleLabelName] == specs.ClusterRoleLabelReplica {
+					g.Expect(oldPVCsUUIDs).NotTo(ContainElement(pvc.UID))
+				} else {
+					g.Expect(oldPVCsUUIDs).To(ContainElement(pvc.UID))
+				}
+			}
+		}).WithTimeout(5 * time.Minute).WithPolling(10 * time.Second).Should(Succeed())
+	}
+
+	verifyPostgresVersion := func(
+		env *environment.TestingEnvironment, primary *corev1.Pod, oldStdOut string, targetMajor int,
+	) {
+		Eventually(func(g Gomega) {
+			stdOut, stdErr, err := exec.EventuallyExecQueryInInstancePod(env.Ctx, env.Client, env.Interface,
+				env.RestClientConfig,
+				exec.PodLocator{Namespace: primary.GetNamespace(), PodName: primary.GetName()}, postgres.AppDBName,
+				"SELECT version();", 60, objects.PollingTime)
+			g.Expect(err).ToNot(HaveOccurred(), "failed to execute version query")
+			g.Expect(stdErr).To(BeEmpty(), "unexpected stderr output when checking version")
+			g.Expect(stdOut).ToNot(Equal(oldStdOut), "postgres version did not change")
+			g.Expect(stdOut).To(ContainSubstring(strconv.Itoa(targetMajor)),
+				fmt.Sprintf("version string doesn't contain expected major version %d: %s", targetMajor, stdOut))
+		}).WithTimeout(5 * time.Minute).WithPolling(10 * time.Second).Should(Succeed())
+	}
+
+	verifyCleanupAfterUpgrade := func(ctx context.Context, client client.Client, primary *corev1.Pod) {
+		shouldHaveBeenDeleted := []string{
+			"/var/lib/postgresql/data/pgdata/pg_upgrade_output.d",
+			"/var/lib/postgresql/data/pgdata-new",
+			"/var/lib/postgresql/data/pgwal-new",
+		}
+		timeout := time.Second * 20
+		for _, path := range shouldHaveBeenDeleted {
+			_, stdErr, err := exec.CommandInInstancePod(ctx, client, env.Interface, env.RestClientConfig,
+				exec.PodLocator{Namespace: primary.GetNamespace(), PodName: primary.GetName()}, &timeout,
+				"stat", path)
+			Expect(err).To(HaveOccurred(), "path: %s", path)
+			Expect(stdErr).To(ContainSubstring("No such file or directory"), "path: %s", path)
+		}
+	}
+
+	BeforeEach(func() {
+		if testLevelEnv.Depth < int(level) {
+			Skip("Test depth is lower than the amount requested for this test")
+		}
+
+		currentMajor, targetMajor := determineVersionsForTesting()
+		var err error
+		namespace, err = env.CreateUniqueTestNamespace(env.Ctx, env.Client, namespacePrefix)
+		Expect(err).ToNot(HaveOccurred())
+
+		storageClass := os.Getenv("E2E_DEFAULT_STORAGE_CLASS")
+		Expect(storageClass).ToNot(BeEmpty())
+
+		// We cannot use generated entries in the DescribeTable, so we use the scenario key as a constant, but
+		// define the actual content here.
+		// See https://onsi.github.io/ginkgo/#mental-model-table-specs-are-just-syntactic-sugar
+		scenarios = buildScenarios(namespace, storageClass, currentMajor, targetMajor)
+	})
+
+	DescribeTable("can upgrade a Cluster to a newer major version", func(scenarioName string) {
+		By("Creating the starting cluster")
+		scenario := scenarios[scenarioName]
+		cluster := scenario.startingCluster
+		err := env.Client.Create(env.Ctx, cluster)
+		Expect(err).NotTo(HaveOccurred())
+		AssertClusterIsReady(cluster.Namespace, cluster.Name, testTimeouts[timeouts.ClusterIsReady],
+			env)
+
+		By("Collecting the pods UUIDs")
+		podList, err := clusterutils.ListPods(env.Ctx, env.Client, cluster.Name, cluster.Namespace)
+		Expect(err).ToNot(HaveOccurred())
+		oldPodsUUIDs := make([]types.UID, len(podList.Items))
+		for i, pod := range podList.Items {
+			oldPodsUUIDs[i] = pod.UID
+		}
+
+		By("Collecting the PVCs UUIDs")
+		pvcList, err := storage.GetPVCList(env.Ctx, env.Client, cluster.Namespace)
+		Expect(err).ToNot(HaveOccurred())
+		oldPVCsUUIDs := make([]types.UID, len(pvcList.Items))
+		for i, pvc := range pvcList.Items {
+			oldPVCsUUIDs[i] = pvc.UID
+		}
+
+		By("Checking the starting version of the cluster")
+		primary, err := clusterutils.GetPrimary(env.Ctx, env.Client, cluster.Namespace, cluster.Name)
+		Expect(err).ToNot(HaveOccurred())
+
+		oldStdOut, stdErr, err := exec.EventuallyExecQueryInInstancePod(env.Ctx, env.Client, env.Interface,
+			env.RestClientConfig,
+			exec.PodLocator{Namespace: primary.GetNamespace(), PodName: primary.GetName()}, postgres.AppDBName,
+			"SELECT version();", 60, objects.PollingTime)
+		Expect(err).ToNot(HaveOccurred())
+		Expect(stdErr).To(BeEmpty())
+		Expect(oldStdOut).To(ContainSubstring(strconv.Itoa(scenario.startingMajor)))
+
+		By("Updating the major")
+		cluster, err = clusterutils.Get(env.Ctx, env.Client, cluster.Namespace, cluster.Name)
+		Expect(err).ToNot(HaveOccurred())
+		cluster.Spec.ImageName = scenario.targetImage
+		// We wrap this in an Eventually to avoid possible failures if the cluster changes
+		Eventually(func() error {
+			return env.Client.Update(env.Ctx, cluster)
+		}).WithTimeout(1 * time.Minute).WithPolling(10 * time.Second).Should(Succeed())
+
+		By("Waiting for the cluster to be in the major upgrade phase")
+		Eventually(func(g Gomega) {
+			cluster, err = clusterutils.Get(env.Ctx, env.Client, cluster.Namespace, cluster.Name)
+			g.Expect(err).ToNot(HaveOccurred())
+			g.Expect(cluster.Status.Phase).To(Equal(v1.PhaseMajorUpgrade))
+		}).WithTimeout(1 * time.Minute).WithPolling(10 * time.Second).Should(Succeed())
+
+		AssertClusterIsReady(cluster.Namespace, cluster.Name, testTimeouts[timeouts.ClusterIsReady], env)
+
+		// The upgrade destroys all the original pods and creates new ones. We want to make sure that we have
+		// the same amount of pods as before, but with different UUIDs.
+		By("Verifying the pods UUIDs have changed")
+		verifyPodsChanged(env.Ctx, env.Client, cluster, oldPodsUUIDs)
+
+		// The upgrade destroys all the original PVCs and creates new ones, except for the ones associated to the
+		// primary. We want to make sure that we have the same amount of PVCs as before, but with different UUIDs,
+		// which should be the same instead for the primary PVCs.
+		By("Verifying the replicas' PVCs have changed")
+		verifyPVCsChanged(env.Ctx, env.Client, cluster, oldPVCsUUIDs)
+
+		// Check that the version has been updated
+		By("Verifying the cluster is running the target version")
+		verifyPostgresVersion(env, primary, oldStdOut, scenario.targetMajor)
+
+		// Expect temporary files to be deleted
+		By("Checking no leftovers exist from the upgrade")
+		verifyCleanupAfterUpgrade(env.Ctx, env.Client, primary)
+	},
+		Entry("PostGIS", postgisEntry),
+		Entry("PostgreSQL", postgresqlEntry),
+		Entry("PostgreSQL minimal", postgresqlMinimalEntry),
+	)
+})
diff --git a/tests/e2e/cluster_microservice_test.go b/tests/e2e/cluster_microservice_test.go
index 198f3278..dbea8c78 100644
--- a/tests/e2e/cluster_microservice_test.go
+++ b/tests/e2e/cluster_microservice_test.go
@@ -24,13 +24,10 @@ import (
 	"os"
 	"strings"
 
-	"github.com/cloudnative-pg/machinery/pkg/image/reference"
-	"github.com/cloudnative-pg/machinery/pkg/postgres/version"
 	batchv1 "k8s.io/api/batch/v1"
 	"k8s.io/apimachinery/pkg/types"
 
 	apiv1 "github.com/cloudnative-pg/cloudnative-pg/api/v1"
-	"github.com/cloudnative-pg/cloudnative-pg/pkg/versions"
 	"github.com/cloudnative-pg/cloudnative-pg/tests"
 	"github.com/cloudnative-pg/cloudnative-pg/tests/utils/clusterutils"
 	"github.com/cloudnative-pg/cloudnative-pg/tests/utils/exec"
@@ -174,7 +171,7 @@ var _ = Describe("Imports with Microservice Approach", Label(tests.LabelImportin
 		Expect(postgresImage).ShouldNot(BeEmpty(), "POSTGRES_IMG env should not be empty")
 
 		// this test case is only applicable if we are not already on the latest major
-		if shouldSkip(postgresImage) {
+		if postgres.IsLatestMajor(postgresImage) {
 			Skip("Already running on the latest major. This test is not applicable for PostgreSQL " + postgresImage)
 		}
 
@@ -194,20 +191,6 @@ var _ = Describe("Imports with Microservice Approach", Label(tests.LabelImportin
 	})
 })
 
-// shouldSkip skip this test if the current POSTGRES_IMG is already the latest major
-func shouldSkip(postgresImage string) bool {
-	// Get the current tag
-	currentImageReference := reference.New(postgresImage)
-	currentImageVersion, err := version.FromTag(currentImageReference.Tag)
-	Expect(err).ToNot(HaveOccurred())
-	// Get the default tag
-	defaultImageReference := reference.New(versions.DefaultImageName)
-	defaultImageVersion, err := version.FromTag(defaultImageReference.Tag)
-	Expect(err).ToNot(HaveOccurred())
-
-	return currentImageVersion.Major() >= defaultImageVersion.Major()
-}
-
 // assertCreateTableWithDataOnSourceCluster will create on the source Cluster, as postgres superUser:
 // 1. a new user `micro`
 // 2. a new table with 2 records owned by `micro` in the `app` database
diff --git a/tests/labels.go b/tests/labels.go
index 50f7698d..15d5461f 100644
--- a/tests/labels.go
+++ b/tests/labels.go
@@ -97,4 +97,7 @@ const (
 
 	// LabelUpgrade is a label for upgrade tests
 	LabelUpgrade = "upgrade"
+
+	// LabelPostgresMajorUpgrade is a label for Cluster major version upgrade tests
+	LabelPostgresMajorUpgrade = "postgres-major-upgrade"
 )
diff --git a/tests/utils/postgres/postgres.go b/tests/utils/postgres/postgres.go
index 54f0327b..3c081fcc 100644
--- a/tests/utils/postgres/postgres.go
+++ b/tests/utils/postgres/postgres.go
@@ -134,3 +134,21 @@ func BumpPostgresImageMajorVersion(postgresImage string) (string, error) {
 
 	return imageReference.GetNormalizedName(), nil
 }
+
+// IsLatestMajor returns true if the given postgresImage is using latest Postgres major version
+func IsLatestMajor(postgresImage string) bool {
+	// Get the current tag
+	currentImageReference := reference.New(postgresImage)
+	currentImageVersion, err := version.FromTag(currentImageReference.Tag)
+	if err != nil {
+		return false
+	}
+	// Get the default tag
+	defaultImageReference := reference.New(versions.DefaultImageName)
+	defaultImageVersion, err := version.FromTag(defaultImageReference.Tag)
+	if err != nil {
+		return false
+	}
+
+	return currentImageVersion.Major() >= defaultImageVersion.Major()
+}