@@ -65,6 +65,168 @@ def configure_patroni_on_unit(self):
6565 self .workload .paths .data , mode = POSTGRESQL_STORAGE_PERMISSIONS , exist_ok = True
6666 )
6767
68+ def _calculate_max_worker_processes (self , cpu_cores : int ) -> str | None :
69+ """Calculate cpu_max_worker_processes configuration value."""
70+ if self .state .config .cpu_max_worker_processes == "auto" :
71+ # auto = minimum(8, 2 * vCores)
72+ return str (min (8 , 2 * cpu_cores ))
73+ elif self .state .config .cpu_max_worker_processes is not None :
74+ value = self .state .config .cpu_max_worker_processes
75+ cap = 10 * cpu_cores
76+ if value > cap :
77+ raise ValueError (
78+ f"cpu-max-worker-processes value { value } exceeds maximum allowed "
79+ f"of { cap } (10 * vCores). Please set a value <= { cap } ."
80+ )
81+ return str (value )
82+ return None
83+
84+ def _validate_worker_config_value (self , param_name : str , value : int , cpu_cores : int ) -> str :
85+ """Shared validation logic for worker process parameters.
86+
87+ Args:
88+ param_name: the configuration parameter name (for error messages).
89+ value: the integer value to validate.
90+ cpu_cores: the number of available CPU cores.
91+
92+ Returns:
93+ String representation of the validated value.
94+
95+ Raises:
96+ ValueError: if value exceeds 10 * vCores.
97+ """
98+ cap = 10 * cpu_cores
99+ if value > cap :
100+ raise ValueError (
101+ f"{ param_name } value { value } exceeds maximum allowed "
102+ f"of { cap } (10 * vCores). Please set a value <= { cap } ."
103+ )
104+ return str (value )
105+
106+ def _calculate_max_parallel_workers (self , base_max_workers : int , cpu_cores : int ) -> str | None :
107+ """Calculate cpu_max_parallel_workers configuration value."""
108+ if self .state .config .cpu_max_parallel_workers == "auto" :
109+ return str (base_max_workers )
110+ elif self .state .config .cpu_max_parallel_workers is not None :
111+ validated_value_str = self ._validate_worker_config_value (
112+ "cpu-max-parallel-workers" , self .state .config .cpu_max_parallel_workers , cpu_cores
113+ )
114+ # Apply the min constraint with base_max_workers
115+ return str (min (int (validated_value_str ), base_max_workers ))
116+ return None
117+
118+ def _calculate_max_parallel_maintenance_workers (
119+ self , base_max_workers : int , cpu_cores : int
120+ ) -> str | None :
121+ """Calculate cpu_max_parallel_maintenance_workers configuration value."""
122+ if self .state .config .cpu_max_parallel_maintenance_workers == "auto" :
123+ return str (base_max_workers )
124+ elif self .state .config .cpu_max_parallel_maintenance_workers is not None :
125+ return self ._validate_worker_config_value (
126+ "cpu-max-parallel-maintenance-workers" ,
127+ self .state .config .cpu_max_parallel_maintenance_workers ,
128+ cpu_cores ,
129+ )
130+ return None
131+
132+ def _calculate_max_logical_replication_workers (
133+ self , base_max_workers : int , cpu_cores : int
134+ ) -> str | None :
135+ """Calculate cpu_max_logical_replication_workers configuration value."""
136+ if self .state .config .cpu_max_logical_replication_workers == "auto" :
137+ return str (base_max_workers )
138+ elif self .state .config .cpu_max_logical_replication_workers is not None :
139+ return self ._validate_worker_config_value (
140+ "cpu-max-logical-replication-workers" ,
141+ self .state .config .cpu_max_logical_replication_workers ,
142+ cpu_cores ,
143+ )
144+ return None
145+
146+ def _calculate_max_sync_workers_per_subscription (
147+ self , base_max_workers : int , cpu_cores : int
148+ ) -> str | None :
149+ """Calculate cpu_max_sync_workers_per_subscription configuration value."""
150+ if self .state .config .cpu_max_sync_workers_per_subscription == "auto" :
151+ return str (base_max_workers )
152+ elif self .state .config .cpu_max_sync_workers_per_subscription is not None :
153+ return self ._validate_worker_config_value (
154+ "cpu-max-sync-workers-per-subscription" ,
155+ self .state .config .cpu_max_sync_workers_per_subscription ,
156+ cpu_cores ,
157+ )
158+ return None
159+
160+ def _calculate_max_parallel_apply_workers_per_subscription (
161+ self , base_max_workers : int , cpu_cores : int
162+ ) -> str | None :
163+ """Calculate cpu_max_parallel_apply_workers_per_subscription configuration value."""
164+ if self .state .config .cpu_max_parallel_apply_workers_per_subscription == "auto" :
165+ return str (base_max_workers )
166+ elif self .state .config .cpu_max_parallel_apply_workers_per_subscription is not None :
167+ return self ._validate_worker_config_value (
168+ "cpu-max-parallel-apply-workers-per-subscription" ,
169+ self .state .config .cpu_max_parallel_apply_workers_per_subscription ,
170+ cpu_cores ,
171+ )
172+ return None
173+
174+ def _calculate_worker_process_config (self , cpu_cores : int ) -> dict [str , str ]:
175+ """Calculate worker process configuration values.
176+
177+ Handles 'auto' values and capping logic for worker process parameters.
178+ Returns a dictionary with the calculated values ready for PostgreSQL.
179+ """
180+ result : dict [str , str ] = {}
181+
182+ # Calculate cpu_max_worker_processes (baseline for other worker configs)
183+ cpu_max_worker_processes_value = self ._calculate_max_worker_processes (cpu_cores )
184+ if cpu_max_worker_processes_value is not None :
185+ result ["max_worker_processes" ] = cpu_max_worker_processes_value
186+
187+ # Get the effective cpu_max_worker_processes for dependent configs
188+ # Use the calculated value, or fall back to PostgreSQL default (8)
189+ base_max_workers = int (result .get ("max_worker_processes" , "8" ))
190+
191+ # Calculate other worker parameters
192+ cpu_max_parallel_workers_value = self ._calculate_max_parallel_workers (
193+ base_max_workers , cpu_cores
194+ )
195+ if cpu_max_parallel_workers_value is not None :
196+ result ["max_parallel_workers" ] = cpu_max_parallel_workers_value
197+
198+ cpu_max_parallel_maintenance_workers_value = (
199+ self ._calculate_max_parallel_maintenance_workers (base_max_workers , cpu_cores )
200+ )
201+ if cpu_max_parallel_maintenance_workers_value is not None :
202+ result ["max_parallel_maintenance_workers" ] = cpu_max_parallel_maintenance_workers_value
203+
204+ cpu_max_logical_replication_workers_value = (
205+ self ._calculate_max_logical_replication_workers (base_max_workers , cpu_cores )
206+ )
207+ if cpu_max_logical_replication_workers_value is not None :
208+ result ["max_logical_replication_workers" ] = cpu_max_logical_replication_workers_value
209+
210+ cpu_max_sync_workers_per_subscription_value = (
211+ self ._calculate_max_sync_workers_per_subscription (base_max_workers , cpu_cores )
212+ )
213+ if cpu_max_sync_workers_per_subscription_value is not None :
214+ result ["max_sync_workers_per_subscription" ] = (
215+ cpu_max_sync_workers_per_subscription_value
216+ )
217+
218+ cpu_max_parallel_apply_workers_per_subscription_value = (
219+ self ._calculate_max_parallel_apply_workers_per_subscription (
220+ base_max_workers , cpu_cores
221+ )
222+ )
223+ if cpu_max_parallel_apply_workers_per_subscription_value is not None :
224+ result ["max_parallel_apply_workers_per_subscription" ] = (
225+ cpu_max_parallel_apply_workers_per_subscription_value
226+ )
227+
228+ return result
229+
68230 def _build_postgresql_parameters (
69231 self , postgresql_client : PostgreSQLClient
70232 ) -> dict [str , str ] | None :
@@ -73,17 +235,33 @@ def _build_postgresql_parameters(
73235 Returns:
74236 Dictionary of PostgreSQL parameters or None if base parameters couldn't be built.
75237 """
238+ cpu_cores , available_memory = self .workload .get_available_resources ()
239+
76240 limit_memory = None
77241 if self .state .config .profile_limit_memory :
78242 limit_memory = self .state .config .profile_limit_memory * 10 ** 6
79243
80244 # Build PostgreSQL parameters.
81245 pg_parameters = postgresql_client .build_postgresql_parameters (
82- self .state .model_config , self . workload . get_available_memory () , limit_memory
246+ self .state .model_config , available_memory , limit_memory
83247 )
84248
85249 # Calculate and merge worker process configurations
86- # TODO: Add additional parameters
250+ worker_configs = self ._calculate_worker_process_config (cpu_cores )
251+
252+ # Add cpu_wal_compression configuration (separate from worker processes)
253+ if self .state .config .cpu_wal_compression is not None :
254+ cpu_wal_compression = "on" if self .state .config .cpu_wal_compression else "off"
255+ else :
256+ # Use config.yaml default when unset (default: true)
257+ cpu_wal_compression = "on"
258+
259+ if pg_parameters is not None :
260+ pg_parameters .update (worker_configs )
261+ pg_parameters ["wal_compression" ] = cpu_wal_compression
262+ else :
263+ pg_parameters = dict (worker_configs )
264+ pg_parameters ["wal_compression" ] = cpu_wal_compression
87265
88266 return pg_parameters
89267
0 commit comments