add: max_partitions_per_insert_block clickhouse sink (#82)

pipelane · Feb 3, 2025 · eb5f78d · eb5f78d
1 parent f52525c
commit eb5f78d
Show file tree

Hide file tree

Showing 4 changed files with 14 additions and 8 deletions.
diff --git a/gen/source/sink/Clickhouse.pkl.go b/gen/source/sink/Clickhouse.pkl.go
diff --git a/pkl/PklProject b/pkl/PklProject
@@ -2,7 +2,7 @@ amends "pkl:Project"
 
 package {
   name = "pipelaner"
-  version = "1.0.5"
+  version = "1.0.6"
   baseUri = "package://github.com/pipelane/pipelaner"
   sourceCode = "https://github.com/pipelane/pipelaner"
   packageZipUrl = "https://github.com/pipelane/pipelaner/releases/download/\(name)@\(version)/\(name)@\(version).zip"

diff --git a/pkl/source/sink/Sinks.pkl b/pkl/source/sink/Sinks.pkl
@@ -40,6 +40,7 @@ class Clickhouse extends Sink {
 
   asyncInsert: String = "1"
   waitForAsyncInsert: String = "1"
+  maxPartitionsPerInsertBlock: Int = 1000
 }
 
 typealias Method = "PATCH"|"POST"|"PUT"|"DELETE"|"GET"
@@ -48,11 +49,4 @@ class Http extends Sink {
   url: String(!isEmpty)
   method: Method
   headers: Mapping<String, String>?
-}
-
-
-hh = new Mapping {
-  ["Authorization"] {
-    "Bearer \(read("env:"))"
-  }
 }
diff --git a/sources/sink/clickhouse/clickhouse.go b/sources/sink/clickhouse/clickhouse.go
@@ -272,6 +272,10 @@ func (c *Clickhouse) write(ctx context.Context, chData chan any) error {
 				Value:     c.clickConfig.GetWaitForAsyncInsert(),
 				Important: true,
 			},
+			{
+				Key:   "max_partitions_per_insert_block",
+				Value: fmt.Sprintf("%d", c.clickConfig.GetMaxPartitionsPerInsertBlock()),
+			},
 		},
 		OnInput: func(_ context.Context) error {
 			input.Reset()