From e21fa865e4c8a6b5261263661f738bdd2853a137 Mon Sep 17 00:00:00 2001
From: exfly <exflyg@gmail.com>
Date: Wed, 25 Oct 2023 18:32:23 +0800
Subject: [PATCH] feat: DynamicSchema add opt NotNullable

---
 config/config.go     |  5 +++--
 output/clickhouse.go | 15 ++++++++++-----
 2 files changed, 13 insertions(+), 7 deletions(-)

diff --git a/config/config.go b/config/config.go
index 689a5646..de79af87 100644
--- a/config/config.go
+++ b/config/config.go
@@ -133,8 +133,9 @@ type TaskConfig struct {
 	} `json:"dims"`
 	// DynamicSchema will add columns present in message to clickhouse. Requires AutoSchema be true.
 	DynamicSchema struct {
-		Enable  bool
-		MaxDims int // the upper limit of dynamic columns number, <=0 means math.MaxInt16. protecting dirty data attack
+		Enable      bool
+		NotNullable bool
+		MaxDims     int // the upper limit of dynamic columns number, <=0 means math.MaxInt16. protecting dirty data attack
 		// A column is added for new key K if all following conditions are true:
 		// - K isn't in ExcludeColumns
 		// - number of existing columns doesn't reach MaxDims-1
diff --git a/output/clickhouse.go b/output/clickhouse.go
index e85aff3f..fd208bf3 100644
--- a/output/clickhouse.go
+++ b/output/clickhouse.go
@@ -512,21 +512,26 @@ func (c *ClickHouse) ChangeSchema(newKeys *sync.Map) (err error) {
 		var strVal string
 		switch intVal {
 		case model.Bool:
-			strVal = "Nullable(Bool)"
+			strVal = "Bool"
 		case model.Int64:
-			strVal = "Nullable(Int64)"
+			strVal = "Int64"
 		case model.Float64:
-			strVal = "Nullable(Float64)"
+			strVal = "Float64"
 		case model.String:
-			strVal = "Nullable(String)"
+			strVal = "String"
 		case model.DateTime:
-			strVal = "Nullable(DateTime64(3))"
+			strVal = "DateTime64(3)"
 		case model.Object:
 			strVal = model.GetTypeName(intVal)
 		default:
 			err = errors.Newf("%s: BUG: unsupported column type %s", taskCfg.Name, model.GetTypeName(intVal))
 			return false
 		}
+
+		if !taskCfg.DynamicSchema.NotNullable {
+			strVal = fmt.Sprintf("Nullable(%v)", strVal)
+		}
+
 		if c.taskCfg.PrometheusSchema && intVal == model.String {
 			alterSeries = append(alterSeries, fmt.Sprintf("ADD COLUMN IF NOT EXISTS `%s` %s", strKey, strVal))
 		} else {