diff --git a/e2e/tests/k8s_api.py b/e2e/tests/k8s_api.py
index 1f42ad4bc..12ae97189 100644
--- a/e2e/tests/k8s_api.py
+++ b/e2e/tests/k8s_api.py
@@ -177,6 +177,16 @@ def count_pods_with_env_variable(self, env_variable_key, labels, namespace='defa
 
         return pod_count
 
+    def get_env_variable_value(self, env_variable_key, labels, namespace='default'):
+        """Get the value of an environment variable from the first pod matching the labels."""
+        pods = self.api.core_v1.list_namespaced_pod(namespace, label_selector=labels).items
+        if not pods:
+            return None
+        for env in pods[0].spec.containers[0].env:
+            if env.name == env_variable_key:
+                return env.value
+        return None
+
     def count_pods_with_rolling_update_flag(self, labels, namespace='default'):
         pods = self.api.core_v1.list_namespaced_pod(namespace, label_selector=labels).items
         return len(list(filter(lambda x: "zalando-postgres-operator-rolling-update-required" in x.metadata.annotations, pods)))
diff --git a/e2e/tests/test_e2e.py b/e2e/tests/test_e2e.py
index f473b5cc4..2386b3a64 100644
--- a/e2e/tests/test_e2e.py
+++ b/e2e/tests/test_e2e.py
@@ -2547,6 +2547,69 @@ def assert_distributed_pods(self, target_nodes, cluster_labels='cluster-name=aci
 
         return True
 
+    @timeout_decorator.timeout(TEST_TIMEOUT_SEC)
+    def test_user_env_var_override(self):
+        '''
+        Test that user-provided environment variables from spec.env can override
+        operator-generated ones (like SPILO_CONFIGURATION). This is useful for
+        customizing Patroni DCS configuration such as ignore_slots.
+        '''
+        k8s = self.k8s
+        cluster_label = 'application=spilo,cluster-name=acid-minimal-cluster'
+        
+        # Custom SPILO_CONFIGURATION with ignore_slots for Patroni
+        custom_spilo_config = '{"bootstrap":{"dcs":{"ignore_slots":{"type":"logical"}}}}'
+        
+        try:
+            # Patch the cluster to add custom env var that should override operator defaults
+            pg_patch_env = {
+                "spec": {
+                    "env": [
+                        {
+                            "name": "SPILO_CONFIGURATION",
+                            "value": custom_spilo_config
+                        }
+                    ]
+                }
+            }
+            
+            k8s.api.custom_objects_api.patch_namespaced_custom_object(
+                "acid.zalan.do", "v1", "default", "postgresqls", "acid-minimal-cluster", pg_patch_env)
+            
+            # Wait for operator to process the change
+            self.eventuallyEqual(lambda: k8s.get_operator_state(), {"0": "idle"}, 
+                               "Operator does not get in sync after env var patch")
+            
+            # Wait for pods to be updated with the new env var
+            k8s.wait_for_pod_start(cluster_label)
+            k8s.wait_for_running_pods(cluster_label, 2)
+            
+            # Verify that SPILO_CONFIGURATION env var exists in pods
+            self.eventuallyEqual(lambda: k8s.count_pods_with_env_variable("SPILO_CONFIGURATION", cluster_label), 2,
+                               "SPILO_CONFIGURATION env variable not found in pods")
+            
+            # Verify that the user-provided value overrides the operator-generated one
+            actual_value = k8s.get_env_variable_value("SPILO_CONFIGURATION", cluster_label)
+            self.assertIsNotNone(actual_value, "SPILO_CONFIGURATION value is None")
+            self.assertIn("ignore_slots", actual_value, 
+                        "User-provided SPILO_CONFIGURATION with ignore_slots was not applied")
+            
+            # Clean up: remove the custom env var
+            pg_patch_remove_env = {
+                "spec": {
+                    "env": None
+                }
+            }
+            k8s.api.custom_objects_api.patch_namespaced_custom_object(
+                "acid.zalan.do", "v1", "default", "postgresqls", "acid-minimal-cluster", pg_patch_remove_env)
+            
+            self.eventuallyEqual(lambda: k8s.get_operator_state(), {"0": "idle"}, 
+                               "Operator does not get in sync after removing env var")
+            
+        except timeout_decorator.TimeoutError:
+            print('Operator log: {}'.format(k8s.get_operator_log()))
+            raise
+
     def check_cluster_child_resources_owner_references(self, cluster_name, cluster_namespace='default', inverse=False):
         k8s = self.k8s
 
diff --git a/pkg/cluster/k8sres.go b/pkg/cluster/k8sres.go
index 9bc39a9db..06c50ac6e 100644
--- a/pkg/cluster/k8sres.go
+++ b/pkg/cluster/k8sres.go
@@ -1088,7 +1088,20 @@ func (c *Cluster) generateSpiloPodEnvVars(
 func appendEnvVars(envs []v1.EnvVar, appEnv ...v1.EnvVar) []v1.EnvVar {
 	collectedEnvs := envs
 	for _, env := range appEnv {
-		if !isEnvVarPresent(collectedEnvs, env.Name) {
+		// Check if env var already exists
+		existingIdx := -1
+		for i, existing := range collectedEnvs {
+			if strings.EqualFold(existing.Name, env.Name) {
+				existingIdx = i
+				break
+			}
+		}
+		
+		if existingIdx >= 0 {
+			// Replace existing env var (user override takes precedence)
+			collectedEnvs[existingIdx] = env
+		} else {
+			// Add new env var
 			collectedEnvs = append(collectedEnvs, env)
 		}
 	}