diff --git a/api/v1/postgres_types.go b/api/v1/postgres_types.go
index 48b472fb..9d709840 100644
--- a/api/v1/postgres_types.go
+++ b/api/v1/postgres_types.go
@@ -602,7 +602,7 @@ func (p *Postgres) ToPeripheralResourceLookupKey() types.NamespacedName {
 	}
 }
 
-func (p *Postgres) ToUnstructuredZalandoPostgresql(z *zalando.Postgresql, c *corev1.ConfigMap, sc string, pgParamBlockList map[string]bool, rbs *BackupConfig, srcDB *Postgres, patroniTTL, patroniLoopWait, patroniRetryTimeout uint32) (*unstructured.Unstructured, error) {
+func (p *Postgres) ToUnstructuredZalandoPostgresql(z *zalando.Postgresql, c *corev1.ConfigMap, sc string, pgParamBlockList map[string]bool, rbs *BackupConfig, srcDB *Postgres, patroniTTL, patroniLoopWait, patroniRetryTimeout uint32, dboIsSuperuser bool) (*unstructured.Unstructured, error) {
 	if z == nil {
 		z = &zalando.Postgresql{}
 	}
@@ -657,6 +657,9 @@ func (p *Postgres) ToUnstructuredZalandoPostgresql(z *zalando.Postgresql, c *cor
 	// Create database owner
 	z.Spec.Users = make(map[string]zalando.UserFlags)
 	z.Spec.Users[ownerName] = zalando.UserFlags{"createdb", "createrole"}
+	if dboIsSuperuser {
+		z.Spec.Users[ownerName] = zalando.UserFlags{"createdb", "createrole", "superuser"}
+	}
 	// Add auditor user
 	z.Spec.Users["auditor"] = zalando.UserFlags{"nologin"}
 
diff --git a/api/v1/postgres_types_test.go b/api/v1/postgres_types_test.go
index d5b352c5..0b196a8e 100644
--- a/api/v1/postgres_types_test.go
+++ b/api/v1/postgres_types_test.go
@@ -349,7 +349,7 @@ func TestPostgresRestoreTimestamp_ToUnstructuredZalandoPostgresql(t *testing.T)
 			p := &Postgres{
 				Spec: tt.spec,
 			}
-			got, _ := p.ToUnstructuredZalandoPostgresql(nil, tt.c, tt.sc, tt.pgParamBlockList, tt.rbs, tt.srcDB, 130, 10, 60)
+			got, _ := p.ToUnstructuredZalandoPostgresql(nil, tt.c, tt.sc, tt.pgParamBlockList, tt.rbs, tt.srcDB, 130, 10, 60, false)
 
 			jsonZ, err := runtime.DefaultUnstructuredConverter.ToUnstructured(got)
 			if err != nil {
diff --git a/controllers/postgres_controller.go b/controllers/postgres_controller.go
index f6ce13ca..9a17b5f7 100644
--- a/controllers/postgres_controller.go
+++ b/controllers/postgres_controller.go
@@ -86,6 +86,7 @@ type PostgresReconciler struct {
 	EnableWalGEncryption                bool
 	PostgresletFullname                 string
 	EnableBootstrapStandbyFromS3        bool
+	EnableSuperUserForDBO               bool
 }
 
 // Reconcile is the entry point for postgres reconciliation.
@@ -357,7 +358,7 @@ func (r *PostgresReconciler) createOrUpdateZalandoPostgresql(ctx context.Context
 			return fmt.Errorf("failed to fetch zalando postgresql: %w", err)
 		}
 
-		u, err := instance.ToUnstructuredZalandoPostgresql(nil, sidecarsCM, r.StorageClass, r.PgParamBlockList, restoreBackupConfig, restoreSouceInstance, patroniTTL, patroniLoopWait, patroniRetryTimout)
+		u, err := instance.ToUnstructuredZalandoPostgresql(nil, sidecarsCM, r.StorageClass, r.PgParamBlockList, restoreBackupConfig, restoreSouceInstance, patroniTTL, patroniLoopWait, patroniRetryTimout, r.EnableSuperUserForDBO)
 		if err != nil {
 			return fmt.Errorf("failed to convert to unstructured zalando postgresql: %w", err)
 		}
@@ -373,7 +374,7 @@ func (r *PostgresReconciler) createOrUpdateZalandoPostgresql(ctx context.Context
 	// Update zalando postgresql
 	mergeFrom := client.MergeFrom(rawZ.DeepCopy())
 
-	u, err := instance.ToUnstructuredZalandoPostgresql(rawZ, sidecarsCM, r.StorageClass, r.PgParamBlockList, restoreBackupConfig, restoreSouceInstance, patroniTTL, patroniLoopWait, patroniRetryTimout)
+	u, err := instance.ToUnstructuredZalandoPostgresql(rawZ, sidecarsCM, r.StorageClass, r.PgParamBlockList, restoreBackupConfig, restoreSouceInstance, patroniTTL, patroniLoopWait, patroniRetryTimout, r.EnableSuperUserForDBO)
 	if err != nil {
 		return fmt.Errorf("failed to convert to unstructured zalando postgresql: %w", err)
 	}
diff --git a/main.go b/main.go
index 7dcfbe39..f28cb242 100644
--- a/main.go
+++ b/main.go
@@ -73,6 +73,7 @@ const (
 	enableWalGEncryptionFlg               = "enable-walg-encryption"
 	enableForceSharedIPFlg                = "enable-force-shared-ip"
 	enableBootstrapStandbyFromS3Flg       = "enable-bootsrtap-standby-from-s3"
+	enableSuperUserForDBOFlg              = "enable-superuser-for-dbo"
 )
 
 var (
@@ -126,6 +127,7 @@ func main() {
 		enableWalGEncryption               bool
 		enableForceSharedIP                bool
 		enableBootstrapStandbyFromS3       bool
+		enableSuperUserForDBO              bool
 
 		portRangeStart int
 		portRangeSize  int
@@ -265,6 +267,9 @@ func main() {
 	viper.SetDefault(enableBootstrapStandbyFromS3Flg, true)
 	enableBootstrapStandbyFromS3 = viper.GetBool(enableBootstrapStandbyFromS3Flg)
 
+	viper.SetDefault(enableSuperUserForDBOFlg, false)
+	enableSuperUserForDBO = viper.GetBool(enableSuperUserForDBOFlg)
+
 	ctrl.SetLogger(zap.New(zap.UseDevMode(true)))
 
 	ctrl.Log.Info("flag",
@@ -305,6 +310,7 @@ func main() {
 		enableWalGEncryptionFlg, enableWalGEncryption,
 		enableForceSharedIPFlg, enableForceSharedIP,
 		enableBootstrapStandbyFromS3Flg, enableBootstrapStandbyFromS3,
+		enableSuperUserForDBOFlg, enableSuperUserForDBO,
 	)
 
 	svcClusterConf := ctrl.GetConfigOrDie()
@@ -413,6 +419,7 @@ func main() {
 		EnableWalGEncryption:                enableWalGEncryption,
 		PostgresletFullname:                 postgresletFullname,
 		EnableBootstrapStandbyFromS3:        enableBootstrapStandbyFromS3,
+		EnableSuperUserForDBO:               enableSuperUserForDBO,
 	}).SetupWithManager(ctrlPlaneClusterMgr); err != nil {
 		setupLog.Error(err, "unable to create controller", "controller", "Postgres")
 		os.Exit(1)