Fetch ClickHouse MergeTree limits dynamically and update backpressure configuration. Adjust schema defaults for compatibility.

Author: kanwarujjaval

src/config/schema.ts

@@ -84,10 +84,10 @@ export const configSchema = z.object({
         enabled: booleanFromEnv.default(true),
         partsToThrowInsert: intFromEnv.default(300),
         maxPartsInTotal: intFromEnv.default(500),
-        partitionPctHigh: numberFromEnv.default(0.50).pipe(z.number().min(0).max(1)),
-        partitionPctLow: numberFromEnv.default(0.35).pipe(z.number().min(0).max(1)),
-        totalPctHigh: numberFromEnv.default(0.50).pipe(z.number().min(0).max(1)),
-        totalPctLow: numberFromEnv.default(0.40).pipe(z.number().min(0).max(1)),
+        partitionPctHigh: numberFromEnv.default(0.70).pipe(z.number().min(0).max(1)),
+        partitionPctLow: numberFromEnv.default(0.55).pipe(z.number().min(0).max(1)),
+        totalPctHigh: numberFromEnv.default(0.70).pipe(z.number().min(0).max(1)),
+        totalPctLow: numberFromEnv.default(0.55).pipe(z.number().min(0).max(1)),
         pollIntervalMs: intFromEnv.default(15_000),
         maxPauseEpisodeMs: intFromEnv.default(180_000),
     }),

src/main.ts

@@ -87,6 +87,11 @@ async function main(): Promise<void> {
     request_timeout: config.target.queryTimeoutMs,
   });
 
+  // Fetch actual ClickHouse MergeTree settings and override config defaults
+  const serverLimits = await ClickHousePressure.fetchServerLimits(pressureClient, logger);
+  config.backpressure.partsToThrowInsert = serverLimits.partsToThrowInsert;
+  config.backpressure.maxPartsInTotal = serverLimits.maxPartsInTotal;
+
   const chPressure = new ClickHousePressure(pressureClient, config.backpressure, logger);
 
   // Map config.memory to GcConfig shape

src/source/discover-collections.ts

@@ -20,6 +20,13 @@ export async function discoverCollections(
 
     const names = collections.map((c) => c.name).sort();
 
+    // Ensure the base collection (exact prefix match) is always processed first
+    const baseIdx = names.indexOf(prefix);
+    if (baseIdx > 0) {
+        names.splice(baseIdx, 1);
+        names.unshift(prefix);
+    }
+
     if (names.length === 0) {
         throw new Error(
             `No collections found matching prefix "${prefix}" in database "${db.databaseName}"`,

src/target/clickhouse-pressure.ts

@@ -30,6 +30,14 @@ export interface PressureState {
   pauseReason: string | null;
 }
 
+export interface ServerMergeTreeLimits {
+  partsToThrowInsert: number;
+  partsToDelayInsert: number;
+  maxPartsInTotal: number;
+  inactivePartsToThrowInsert: number;
+  inactivePartsToDelayInsert: number;
+}
+
 export class ClickHousePressure {
   private readonly client: ClickHouseClient;
   private readonly config: BackpressureConfig;
@@ -41,6 +49,47 @@ export class ClickHousePressure {
     this.logger = logger.child({ component: 'clickhouse-pressure' });
   }
 
+  /**
+   * Query the actual MergeTree settings from the ClickHouse server.
+   * Returns the server-side limits that govern when inserts are rejected or delayed.
+   */
+  static async fetchServerLimits(
+    client: ClickHouseClient,
+    logger: Logger,
+  ): Promise<ServerMergeTreeLimits> {
+    const result = await client.query({
+      query: `
+        SELECT name, value
+        FROM system.merge_tree_settings
+        WHERE name IN (
+          'parts_to_throw_insert',
+          'parts_to_delay_insert',
+          'max_parts_in_total',
+          'inactive_parts_to_throw_insert',
+          'inactive_parts_to_delay_insert'
+        )
+      `,
+      format: 'JSONEachRow',
+    });
+
+    const rows = await result.json<{ name: string; value: string }>();
+    const map: Record<string, number> = {};
+    for (const row of rows) {
+      map[row.name] = Number(row.value);
+    }
+
+    const limits: ServerMergeTreeLimits = {
+      partsToThrowInsert: map['parts_to_throw_insert'] ?? 300,
+      partsToDelayInsert: map['parts_to_delay_insert'] ?? 150,
+      maxPartsInTotal: map['max_parts_in_total'] ?? 100000,
+      inactivePartsToThrowInsert: map['inactive_parts_to_throw_insert'] ?? 0,
+      inactivePartsToDelayInsert: map['inactive_parts_to_delay_insert'] ?? 0,
+    };
+
+    logger.info({ limits }, 'Fetched ClickHouse merge_tree_settings');
+    return limits;
+  }
+
   async sample(database: string, table: string): Promise<PressureState> {
     const results = await Promise.allSettled([
       this.queryActivePartsTotal(database, table),