make ch parallel view processing configurable (#223)

iuwqyir · web-flow · commit 15d07b7696bf · 2025-06-10T14:08:01.000+03:00
### TL;DR

Added configuration option to enable parallel view processing for ClickHouse connections.

### What changed?

- Added a new configuration flag `enableParallelViewProcessing` for all ClickHouse storage types (orchestrator, main, and staging)
- Modified the ClickHouse connection setup to make parallel view processing optional instead of always enabled
- Added the corresponding configuration binding in the root command

### How to test?

1. Run the application with the new flag to enable parallel view processing:
   ```
   --storage-orchestrator-clickhouse-enableParallelViewProcessing=true
   --storage-main-clickhouse-enableParallelViewProcessing=true
   --storage-staging-clickhouse-enableParallelViewProcessing=true
   ```
2. Verify that the ClickHouse connections are established with the `parallel_view_processing` setting when the flag is enabled
3. Verify that the setting is not applied when the flag is disabled (default behavior)

### Why make this change?

Parallel view processing in ClickHouse can improve query performance for certain workloads, but it's not always desirable for all environments or use cases. Making this setting configurable allows users to enable it only when beneficial for their specific workload patterns, providing more flexibility in optimizing ClickHouse performance.

&lt;!-- This is an auto-generated comment: release notes by coderabbit.ai --&gt;

## Summary by CodeRabbit

- **New Features**
  - Added new configuration options to enable or disable parallel view processing for Clickhouse storage in orchestrator, main, and staging environments. These can be set via command-line flags or configuration files.

- **Improvements**
  - Parallel view processing in Clickhouse is now configurable, allowing greater flexibility in storage settings.

&lt;!-- end of auto-generated comment: release notes by coderabbit.ai --&gt;
diff --git a/cmd/root.go b/cmd/root.go
@@ -82,6 +82,7 @@ func init() {
 	rootCmd.PersistentFlags().Int("storage-orchestrator-clickhouse-maxOpenConns", 30, "Clickhouse max open connections for orchestrator storage")
 	rootCmd.PersistentFlags().Int("storage-orchestrator-clickhouse-maxIdleConns", 30, "Clickhouse max idle connections for orchestrator storage")
 	rootCmd.PersistentFlags().Bool("storage-orchestrator-clickhouse-disableTLS", false, "Clickhouse disableTLS for orchestrator storage")
+	rootCmd.PersistentFlags().Bool("storage-orchestrator-clickhouse-enableParallelViewProcessing", false, "Clickhouse enableParallelViewProcessing for orchestrator storage")
 	rootCmd.PersistentFlags().String("storage-staging-clickhouse-host", "", "Clickhouse host for staging storage")
 	rootCmd.PersistentFlags().String("storage-main-clickhouse-host", "", "Clickhouse host for main storage")
 	rootCmd.PersistentFlags().String("storage-main-clickhouse-username", "", "Clickhouse username for main storage")
@@ -91,13 +92,15 @@ func init() {
 	rootCmd.PersistentFlags().Int("storage-main-clickhouse-maxOpenConns", 30, "Clickhouse max open connections for main storage")
 	rootCmd.PersistentFlags().Int("storage-main-clickhouse-maxIdleConns", 30, "Clickhouse max idle connections for main storage")
 	rootCmd.PersistentFlags().Bool("storage-main-clickhouse-disableTLS", false, "Clickhouse disableTLS for main storage")
+	rootCmd.PersistentFlags().Bool("storage-main-clickhouse-enableParallelViewProcessing", false, "Clickhouse enableParallelViewProcessing for main storage")
 	rootCmd.PersistentFlags().String("storage-staging-clickhouse-username", "", "Clickhouse username for staging storage")
 	rootCmd.PersistentFlags().String("storage-staging-clickhouse-password", "", "Clickhouse password for staging storage")
 	rootCmd.PersistentFlags().Bool("storage-staging-clickhouse-asyncInsert", false, "Clickhouse async insert for staging storage")
 	rootCmd.PersistentFlags().Int("storage-staging-clickhouse-maxRowsPerInsert", 100000, "Clickhouse max rows per insert for staging storage")
 	rootCmd.PersistentFlags().Int("storage-staging-clickhouse-maxOpenConns", 30, "Clickhouse max open connections for staging storage")
 	rootCmd.PersistentFlags().Int("storage-staging-clickhouse-maxIdleConns", 30, "Clickhouse max idle connections for staging storage")
 	rootCmd.PersistentFlags().Bool("storage-staging-clickhouse-disableTLS", false, "Clickhouse disableTLS for staging storage")
+	rootCmd.PersistentFlags().Bool("storage-staging-clickhouse-enableParallelViewProcessing", false, "Clickhouse enableParallelViewProcessing for staging storage")
 	rootCmd.PersistentFlags().String("api-host", "localhost:3000", "API host")
 	rootCmd.PersistentFlags().String("api-basicAuth-username", "", "API basic auth username")
 	rootCmd.PersistentFlags().String("api-basicAuth-password", "", "API basic auth password")
@@ -164,6 +167,7 @@ func init() {
 	viper.BindPFlag("storage.staging.clickhouse.maxOpenConns", rootCmd.PersistentFlags().Lookup("storage-staging-clickhouse-maxOpenConns"))
 	viper.BindPFlag("storage.staging.clickhouse.maxIdleConns", rootCmd.PersistentFlags().Lookup("storage-staging-clickhouse-maxIdleConns"))
 	viper.BindPFlag("storage.staging.clickhouse.disableTLS", rootCmd.PersistentFlags().Lookup("storage-staging-clickhouse-disableTLS"))
+	viper.BindPFlag("storage.staging.clickhouse.enableParallelViewProcessing", rootCmd.PersistentFlags().Lookup("storage-staging-clickhouse-enableParallelViewProcessing"))
 	viper.BindPFlag("storage.main.clickhouse.database", rootCmd.PersistentFlags().Lookup("storage-main-clickhouse-database"))
 	viper.BindPFlag("storage.main.clickhouse.host", rootCmd.PersistentFlags().Lookup("storage-main-clickhouse-host"))
 	viper.BindPFlag("storage.main.clickhouse.port", rootCmd.PersistentFlags().Lookup("storage-main-clickhouse-port"))
@@ -174,6 +178,7 @@ func init() {
 	viper.BindPFlag("storage.main.clickhouse.maxOpenConns", rootCmd.PersistentFlags().Lookup("storage-main-clickhouse-maxOpenConns"))
 	viper.BindPFlag("storage.main.clickhouse.maxIdleConns", rootCmd.PersistentFlags().Lookup("storage-main-clickhouse-maxIdleConns"))
 	viper.BindPFlag("storage.main.clickhouse.disableTLS", rootCmd.PersistentFlags().Lookup("storage-main-clickhouse-disableTLS"))
+	viper.BindPFlag("storage.main.clickhouse.enableParallelViewProcessing", rootCmd.PersistentFlags().Lookup("storage-main-clickhouse-enableParallelViewProcessing"))
 	viper.BindPFlag("storage.orchestrator.clickhouse.database", rootCmd.PersistentFlags().Lookup("storage-orchestrator-clickhouse-database"))
 	viper.BindPFlag("storage.orchestrator.clickhouse.host", rootCmd.PersistentFlags().Lookup("storage-orchestrator-clickhouse-host"))
 	viper.BindPFlag("storage.orchestrator.clickhouse.port", rootCmd.PersistentFlags().Lookup("storage-orchestrator-clickhouse-port"))
@@ -184,6 +189,7 @@ func init() {
 	viper.BindPFlag("storage.orchestrator.clickhouse.maxOpenConns", rootCmd.PersistentFlags().Lookup("storage-orchestrator-clickhouse-maxOpenConns"))
 	viper.BindPFlag("storage.orchestrator.clickhouse.maxIdleConns", rootCmd.PersistentFlags().Lookup("storage-orchestrator-clickhouse-maxIdleConns"))
 	viper.BindPFlag("storage.orchestrator.clickhouse.disableTLS", rootCmd.PersistentFlags().Lookup("storage-orchestrator-clickhouse-disableTLS"))
+	viper.BindPFlag("storage.orchestrator.clickhouse.enableParallelViewProcessing", rootCmd.PersistentFlags().Lookup("storage-orchestrator-clickhouse-enableParallelViewProcessing"))
 	viper.BindPFlag("api.host", rootCmd.PersistentFlags().Lookup("api-host"))
 	viper.BindPFlag("api.basicAuth.username", rootCmd.PersistentFlags().Lookup("api-basicAuth-username"))
 	viper.BindPFlag("api.basicAuth.password", rootCmd.PersistentFlags().Lookup("api-basicAuth-password"))
diff --git a/configs/config.go b/configs/config.go
@@ -71,17 +71,18 @@ type TableConfig struct {
 type TableOverrideConfig map[string]TableConfig
 
 type ClickhouseConfig struct {
-	Host             string                         `mapstructure:"host"`
-	Port             int                            `mapstructure:"port"`
-	Username         string                         `mapstructure:"username"`
-	Password         string                         `mapstructure:"password"`
-	Database         string                         `mapstructure:"database"`
-	DisableTLS       bool                           `mapstructure:"disableTLS"`
-	AsyncInsert      bool                           `mapstructure:"asyncInsert"`
-	MaxRowsPerInsert int                            `mapstructure:"maxRowsPerInsert"`
-	MaxOpenConns     int                            `mapstructure:"maxOpenConns"`
-	MaxIdleConns     int                            `mapstructure:"maxIdleConns"`
-	ChainBasedConfig map[string]TableOverrideConfig `mapstructure:"chainBasedConfig"`
+	Host                         string                         `mapstructure:"host"`
+	Port                         int                            `mapstructure:"port"`
+	Username                     string                         `mapstructure:"username"`
+	Password                     string                         `mapstructure:"password"`
+	Database                     string                         `mapstructure:"database"`
+	DisableTLS                   bool                           `mapstructure:"disableTLS"`
+	AsyncInsert                  bool                           `mapstructure:"asyncInsert"`
+	MaxRowsPerInsert             int                            `mapstructure:"maxRowsPerInsert"`
+	MaxOpenConns                 int                            `mapstructure:"maxOpenConns"`
+	MaxIdleConns                 int                            `mapstructure:"maxIdleConns"`
+	ChainBasedConfig             map[string]TableOverrideConfig `mapstructure:"chainBasedConfig"`
+	EnableParallelViewProcessing bool                           `mapstructure:"enableParallelViewProcessing"`
 }
 
 type RPCBatchRequestConfig struct {
diff --git a/internal/storage/clickhouse.go b/internal/storage/clickhouse.go
@@ -103,7 +103,9 @@ func connectDB(cfg *config.ClickhouseConfig) (clickhouse.Conn, error) {
 				"do_not_merge_across_partitions_select_final": "1",
 				"use_skip_indexes_if_final":                   "1",
 				"optimize_move_to_prewhere_if_final":          "1",
-				"parallel_view_processing":                    "1",
+			}
+			if cfg.EnableParallelViewProcessing {
+				settings["parallel_view_processing"] = "1"
 			}
 			if cfg.AsyncInsert {
 				settings["async_insert"] = "1"
