add clickhouse buffer (#93)

Author: alexeyxo

gen/source/sink/Clickhouse.pkl.go

@@ -37,6 +37,7 @@ class Clickhouse extends Sink {
   asyncInsert: String = "1"
   waitForAsyncInsert: String = "1"
   maxPartitionsPerInsertBlock: Int = 1000
+  channelBuffersize: UInt = 10_000
 }
 
 typealias Method = "PATCH"|"POST"|"PUT"|"DELETE"|"GET"

pkl/source/sink/Sinks.pkl

@@ -10,6 +10,7 @@ import (
 	"errors"
 	"fmt"
 	"io"
+	"sync"
 	"time"
 
 	"github.com/ClickHouse/ch-go"
@@ -335,7 +336,29 @@ func (c *Clickhouse) Sink(val any) {
 		chData <- v
 		close(chData)
 	case chan any:
-		chData = v
+		newCh := make(chan any, c.clickConfig.GetChannelBuffersize())
+		wg := sync.WaitGroup{}
+		for vals := range v {
+			wg.Add(1)
+			switch item := vals.(type) {
+			case []map[string]interface{}:
+				go func() {
+					for _, chV := range item {
+						newCh <- chV
+					}
+					wg.Done()
+				}()
+				continue
+			default:
+				newCh <- vals
+				wg.Done()
+			}
+		}
+		go func() {
+			wg.Wait()
+			close(newCh)
+		}()
+		chData = newCh
 	default:
 		c.Log().Error().Err(errors.New("unknown type val")).Msg("failed write clickhouse")
 		return