Add async insert settings to processors (#67)

- Add configurable asyncInsert and waitForAsyncInsert settings to each
  processor's ClickHouse writes to reduce part creation pressure
- Use *bool pointers to distinguish omitted config (defaults to true)
  from explicit false values
- Remove unused chunkSize and progressLogThreshold config options
- Simplify structlog ProcessTransaction by removing OnInput streaming
  pattern in favor of simple batch insert
- Remove DefaultChunkSize and DefaultProgressLogThreshold from tracker
- Update tests to remove chunk-related assertions

Author: Savid

example_config.yaml

@@ -50,18 +50,21 @@ processors:
     addr: "localhost:9000"
     database: "default"
     table: "canonical_execution_transaction_structlog"
-    # debug: false                     # Enable debug logging for ClickHouse queries
-    # maxPendingBlockRange: 2          # Max distance between oldest incomplete and current block (default: 2)
-    # Channel-based batching configuration for memory-efficient processing
-    # bigTransactionThreshold: 500000  # Transactions with more structlogs are considered "big" (default: 500000)
-    # chunkSize: 10000                 # Number of structlogs per batch (default: 10000)
-    # channelBufferSize: 2             # Number of chunks to buffer in channel (default: 2)
-    # progressLogThreshold: 100000     # Log progress every N structlogs for large transactions (default: 100000)
+    # debug: false                # Enable debug logging for ClickHouse queries
+    # maxPendingBlockRange: 2     # Max distance between oldest incomplete and current block (default: 2)
+    # asyncInsert: true           # Enable async inserts to reduce part creation (default: true)
+    # waitForAsyncInsert: true    # Wait for async insert to complete for durability (default: true)
 
-    # Small transaction batching configuration
-    # batchInsertThreshold: 50000      # Transactions with fewer structlogs than this will be batched (default: 50000)
-    # batchFlushInterval: 5s           # Maximum time to wait before flushing a batch (default: 5s)
-    # batchMaxSize: 100000             # Maximum number of structlogs to accumulate in a batch (default: 100000)
+  # Aggregated structlog processor (call frame level aggregation)
+  transactionStructlogAgg:
+    enabled: false
+    addr: "localhost:9000"
+    database: "default"
+    table: "canonical_execution_transaction_structlog_agg"
+    # debug: false                # Enable debug logging for ClickHouse queries
+    # maxPendingBlockRange: 2     # Max distance between oldest incomplete and current block (default: 2)
+    # asyncInsert: true           # Enable async inserts to reduce part creation (default: true)
+    # waitForAsyncInsert: true    # Wait for async insert to complete for durability (default: true)
 
   # Simple transaction processor (lightweight - no debug traces)
   transactionSimple:
@@ -71,6 +74,8 @@ processors:
     table: "execution_transaction"
     # debug: false                     # Enable debug logging for ClickHouse queries
     # maxPendingBlockRange: 2          # Max distance between oldest incomplete and current block (default: 2)
+    # asyncInsert: true                # Enable async inserts to reduce part creation (default: true)
+    # waitForAsyncInsert: true         # Wait for async insert to complete for durability (default: true)
 
 # Application settings
 shutdownTimeout: 6m

pkg/processor/tracker/processor.go

@@ -65,13 +65,6 @@ const (
 	// incomplete block and the current block before blocking new block processing.
 	DefaultMaxPendingBlockRange = 2
 
-	// DefaultChunkSize is the default number of rows per ClickHouse insert batch.
-	DefaultChunkSize = 10000
-
-	// DefaultProgressLogThreshold is the default threshold for logging progress
-	// on large transactions (structlog processor).
-	DefaultProgressLogThreshold = 100000
-
 	// DefaultClickHouseTimeout is the default timeout for ClickHouse operations.
 	DefaultClickHouseTimeout = 30 * time.Second
 

pkg/processor/transaction/simple/config.go

@@ -12,6 +12,10 @@ type Config struct {
 	Enabled           bool   `yaml:"enabled"`
 	Table             string `yaml:"table"`
 
+	// Async insert settings for ClickHouse (pointers to distinguish omitted from explicit false)
+	AsyncInsert        *bool `yaml:"asyncInsert"`        // Enable async inserts to reduce part creation. Default: true
+	WaitForAsyncInsert *bool `yaml:"waitForAsyncInsert"` // Wait for async insert to complete. Default: true
+
 	// Block completion tracking
 	MaxPendingBlockRange int `yaml:"maxPendingBlockRange"` // Max distance between oldest incomplete and current block. Default: 2
 }

pkg/processor/transaction/simple/handlers.go

@@ -320,6 +320,35 @@ func calculateEffectiveGasPrice(block execution.Block, tx execution.Transaction)
 	return effectiveGasPrice
 }
 
+// getInsertSettings returns ClickHouse settings for async inserts.
+func (p *Processor) getInsertSettings() []ch.Setting {
+	// Default to true if not explicitly configured (nil pointer)
+	asyncInsert := true
+	if p.config.AsyncInsert != nil {
+		asyncInsert = *p.config.AsyncInsert
+	}
+
+	waitForAsync := true
+	if p.config.WaitForAsyncInsert != nil {
+		waitForAsync = *p.config.WaitForAsyncInsert
+	}
+
+	asyncVal := "0"
+	if asyncInsert {
+		asyncVal = "1"
+	}
+
+	waitVal := "0"
+	if waitForAsync {
+		waitVal = "1"
+	}
+
+	return []ch.Setting{
+		{Key: "async_insert", Value: asyncVal},
+		{Key: "wait_for_async_insert", Value: waitVal},
+	}
+}
+
 // insertTransactions inserts transactions into ClickHouse using columnar protocol.
 func (p *Processor) insertTransactions(ctx context.Context, transactions []Transaction) error {
 	if len(transactions) == 0 {
@@ -338,8 +367,9 @@ func (p *Processor) insertTransactions(ctx context.Context, transactions []Trans
 	input := cols.Input()
 
 	if err := p.clickhouse.Do(insertCtx, ch.Query{
-		Body:  input.Into(p.config.Table),
-		Input: input,
+		Body:     input.Into(p.config.Table),
+		Input:    input,
+		Settings: p.getInsertSettings(),
 	}); err != nil {
 		common.ClickHouseInsertsRows.WithLabelValues(
 			p.network.Name,

pkg/processor/transaction/structlog/config.go

@@ -12,9 +12,9 @@ type Config struct {
 	Enabled           bool   `yaml:"enabled"`
 	Table             string `yaml:"table"`
 
-	// Streaming settings
-	ChunkSize            int `yaml:"chunkSize"`            // Default: 10,000 rows per OnInput iteration
-	ProgressLogThreshold int `yaml:"progressLogThreshold"` // Default: 100,000 - log progress for large txs
+	// Async insert settings for ClickHouse (pointers to distinguish omitted from explicit false)
+	AsyncInsert        *bool `yaml:"asyncInsert"`        // Enable async inserts to reduce part creation. Default: true
+	WaitForAsyncInsert *bool `yaml:"waitForAsyncInsert"` // Wait for async insert to complete. Default: true
 
 	// Block completion tracking
 	MaxPendingBlockRange int `yaml:"maxPendingBlockRange"` // Max distance between oldest incomplete and current block. Default: 2

pkg/processor/transaction/structlog/processor.go

@@ -196,6 +196,35 @@ func (p *Processor) getProcessBackwardsQueue() string {
 	return tracker.PrefixedProcessBackwardsQueue(ProcessorName, p.redisPrefix)
 }
 
+// getInsertSettings returns ClickHouse settings for async inserts.
+func (p *Processor) getInsertSettings() []ch.Setting {
+	// Default to true if not explicitly configured (nil pointer)
+	asyncInsert := true
+	if p.config.AsyncInsert != nil {
+		asyncInsert = *p.config.AsyncInsert
+	}
+
+	waitForAsync := true
+	if p.config.WaitForAsyncInsert != nil {
+		waitForAsync = *p.config.WaitForAsyncInsert
+	}
+
+	asyncVal := "0"
+	if asyncInsert {
+		asyncVal = "1"
+	}
+
+	waitVal := "0"
+	if waitForAsync {
+		waitVal = "1"
+	}
+
+	return []ch.Setting{
+		{Key: "async_insert", Value: asyncVal},
+		{Key: "wait_for_async_insert", Value: waitVal},
+	}
+}
+
 // insertStructlogs inserts structlogs into ClickHouse using columnar protocol.
 func (p *Processor) insertStructlogs(ctx context.Context, structlogs []Structlog) error {
 	if len(structlogs) == 0 {
@@ -236,8 +265,9 @@ func (p *Processor) insertStructlogs(ctx context.Context, structlogs []Structlog
 	input := cols.Input()
 
 	if err := p.clickhouse.Do(insertCtx, ch.Query{
-		Body:  input.Into(p.config.Table),
-		Input: input,
+		Body:     input.Into(p.config.Table),
+		Input:    input,
+		Settings: p.getInsertSettings(),
 	}); err != nil {
 		return fmt.Errorf("failed to insert structlogs: %w", err)
 	}

pkg/processor/transaction/structlog/processor_test.go

@@ -22,7 +22,6 @@ func TestProcessor_Creation(t *testing.T) {
 		Config: clickhouse.Config{
 			Addr: "localhost:9000",
 		},
-		ChunkSize: 10000,
 	}
 
 	// Test config validation
@@ -44,7 +43,6 @@ func TestProcessor_ConfigValidation(t *testing.T) {
 				Config: clickhouse.Config{
 					Addr: "localhost:9000",
 				},
-				ChunkSize: 10000,
 			},
 			expectError: false,
 		},
@@ -58,9 +56,8 @@ func TestProcessor_ConfigValidation(t *testing.T) {
 		{
 			name: "missing addr",
 			config: transaction_structlog.Config{
-				Enabled:   true,
-				Table:     "test_table",
-				ChunkSize: 10000,
+				Enabled: true,
+				Table:   "test_table",
 			},
 			expectError: true,
 		},
@@ -71,7 +68,6 @@ func TestProcessor_ConfigValidation(t *testing.T) {
 				Config: clickhouse.Config{
 					Addr: "localhost:9000",
 				},
-				ChunkSize: 10000,
 			},
 			expectError: true,
 		},
@@ -104,7 +100,6 @@ func TestProcessor_ConcurrentConfigValidation(t *testing.T) {
 				Config: clickhouse.Config{
 					Addr: "localhost:9000",
 				},
-				ChunkSize: 10000,
 			}
 			results <- cfg.Validate()
 		}()
@@ -252,33 +247,6 @@ func TestMemoryManagement(t *testing.T) {
 
 	assert.Equal(t, rowCount, cols.Rows(), "Should have correct row count")
 
-	// Test that chunking calculations work properly
-	const chunkSize = 100
-
-	expectedChunks := (rowCount + chunkSize - 1) / chunkSize
-
-	// Verify chunking logic
-	actualChunks := 0
-
-	for i := 0; i < rowCount; i += chunkSize {
-		actualChunks++
-
-		end := i + chunkSize
-		if end > rowCount {
-			end = rowCount
-		}
-
-		// Verify chunk size constraints
-		chunkLen := end - i
-		if chunkLen <= 0 || chunkLen > chunkSize {
-			t.Errorf("Invalid chunk size: %d (expected 1-%d)", chunkLen, chunkSize)
-		}
-	}
-
-	if actualChunks != expectedChunks {
-		t.Errorf("Expected %d chunks, got %d", expectedChunks, actualChunks)
-	}
-
 	// Reset columns to free memory
 	cols.Reset()
 	assert.Equal(t, 0, cols.Rows(), "Reset should clear all rows")
@@ -302,86 +270,6 @@ func TestMemoryManagement(t *testing.T) {
 	}
 }
 
-func TestChunkProcessing(t *testing.T) {
-	tests := []struct {
-		name           string
-		inputSize      int
-		expectedChunks int
-		chunkSize      int
-	}{
-		{
-			name:           "small input",
-			inputSize:      50,
-			expectedChunks: 1,
-			chunkSize:      100,
-		},
-		{
-			name:           "exact chunk size",
-			inputSize:      100,
-			expectedChunks: 1,
-			chunkSize:      100,
-		},
-		{
-			name:           "multiple chunks",
-			inputSize:      250,
-			expectedChunks: 3,
-			chunkSize:      100,
-		},
-		{
-			name:           "large input",
-			inputSize:      1500,
-			expectedChunks: 15,
-			chunkSize:      100,
-		},
-	}
-
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			// Test chunking logic using Columns
-			cols := transaction_structlog.NewColumns()
-			now := time.Now()
-
-			// Fill columns with test data
-			for i := 0; i < tt.inputSize; i++ {
-				cols.Append(
-					now, uint64(i), "0xtest", uint32(0), uint64(21000), false, nil,
-					uint32(i), "PUSH1", uint64(20000), uint64(3), uint64(3), uint64(3), uint64(1),
-					nil, nil, nil, nil, uint32(0), []uint32{}, "test",
-				)
-			}
-
-			assert.Equal(t, tt.inputSize, cols.Rows(), "Should have correct row count")
-
-			// Calculate expected chunks
-			expectedChunks := (tt.inputSize + tt.chunkSize - 1) / tt.chunkSize
-
-			if expectedChunks != tt.expectedChunks {
-				t.Errorf("Expected %d chunks for %d items, got %d", tt.expectedChunks, tt.inputSize, expectedChunks)
-			}
-
-			// Test that the chunking logic would work correctly
-			chunkCount := 0
-
-			for i := 0; i < tt.inputSize; i += tt.chunkSize {
-				chunkCount++
-
-				end := i + tt.chunkSize
-				if end > tt.inputSize {
-					end = tt.inputSize
-				}
-				// Verify chunk boundaries
-				if end <= i {
-					t.Errorf("Invalid chunk boundaries: start=%d, end=%d", i, end)
-				}
-			}
-
-			if chunkCount != tt.expectedChunks {
-				t.Errorf("Chunking produced %d chunks, expected %d", chunkCount, tt.expectedChunks)
-			}
-		})
-	}
-}
-
 func TestColumnsAppendAndReset(t *testing.T) {
 	cols := transaction_structlog.NewColumns()
 	now := time.Now()