feat: DynamicSchema add opt NotNullable

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

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 {