fix: Cannot convert NULL value to non-Nullable type issue

Author: YenchangChan

model/metric.go

@@ -53,9 +53,10 @@ type DimMetrics struct {
 
 // ColumnWithType
 type ColumnWithType struct {
-	Name       string
-	Type       *TypeInfo
-	SourceName string
+	Name        string
+	Type        *TypeInfo
+	SourceName  string
+	NotNullable bool
 }
 
 // struct for ingesting a clickhouse Map type value

output/clickhouse.go

@@ -37,9 +37,25 @@ import (
 )
 
 var (
-	ErrTblNotExist    = errors.Newf("table doesn't exist")
-	selectSQLTemplate = `select name, type, default_kind from system.columns where database = '%s' and table = '%s'`
-
+	ErrTblNotExist     = errors.Newf("table doesn't exist")
+	selectSQLTemplate  = `select name, type, default_kind from system.columns where database = '%s' and table = '%s'`
+	referedSQLTemplate = `SELECT 
+    current_col.default_expression,
+    referenced_col.type AS referenced_col_type,
+    current_col.name,
+    current_col.type
+FROM 
+    system.columns AS current_col
+JOIN 
+    system.columns AS referenced_col 
+ON 
+    current_col.database = referenced_col.database 
+    AND current_col.table = referenced_col.table 
+    AND current_col.default_expression = referenced_col.name 
+WHERE 
+    current_col.database = '%s' 
+    AND
+    current_col.table = '%s';`
 	wrSeriesQuota int = 16384
 
 	SeriesQuotas sync.Map

output/clickhouse_util.go

@@ -2,6 +2,7 @@ package output
 
 import (
 	"fmt"
+	"strings"
 
 	"github.com/housepower/clickhouse_sinker/model"
 	"github.com/housepower/clickhouse_sinker/pool"
@@ -15,6 +16,23 @@ func writeRows(prepareSQL string, rows model.Rows, idxBegin, idxEnd int, conn *p
 
 func getDims(database, table string, excludedColumns []string, parser string, conn *pool.Conn) (dims []*model.ColumnWithType, err error) {
 	var rs *pool.Rows
+	notNullable := make(map[string]bool)
+	if rs, err = conn.Query(fmt.Sprintf(referedSQLTemplate, database, table)); err != nil {
+		err = errors.Wrapf(err, "")
+		return
+	}
+	var default_expression, referenced_col_type, col_name, ori_type string
+	for rs.Next() {
+		if err = rs.Scan(&default_expression, &referenced_col_type, &col_name, &ori_type); err != nil {
+			err = errors.Wrapf(err, "")
+			return
+		}
+		if strings.HasPrefix(referenced_col_type, "Nullable(") && !strings.HasSuffix(ori_type, "Nullable(") {
+			notNullable[default_expression] = true
+		}
+	}
+
+	rs.Close()
 	if rs, err = conn.Query(fmt.Sprintf(selectSQLTemplate, database, table)); err != nil {
 		err = errors.Wrapf(err, "")
 		return
@@ -29,7 +47,16 @@ func getDims(database, table string, excludedColumns []string, parser string, co
 			return
 		}
 		if !util.StringContains(excludedColumns, name) && defaultKind != "MATERIALIZED" {
-			dims = append(dims, &model.ColumnWithType{Name: name, Type: model.WhichType(typ), SourceName: util.GetSourceName(parser, name)})
+			nnull, ok := notNullable[name]
+			if !ok {
+				nnull = false
+			}
+			dims = append(dims, &model.ColumnWithType{
+				Name:        name,
+				Type:        model.WhichType(typ),
+				SourceName:  util.GetSourceName(parser, name),
+				NotNullable: nnull,
+			})
 		}
 	}
 	if len(dims) == 0 {

task/task.go

@@ -172,6 +172,9 @@ func (service *Service) Put(msg *model.InputMessage, traceId string, flushFn fun
 		return nil
 	} else {
 		row = service.metric2Row(metric, msg)
+		if row == nil {
+			return nil
+		}
 		if taskCfg.DynamicSchema.Enable {
 			foundNewKeys = metric.GetNewKeys(&service.knownKeys, &service.newKeys, &service.warnKeys, service.whiteList, service.blackList, msg.Partition, msg.Offset)
 		}
@@ -277,6 +280,18 @@ func (service *Service) metric2Row(metric model.Metric, msg *model.InputMessage)
 				}
 			} else {
 				val := model.GetValueByType(metric, dim)
+				if dim.NotNullable && val == nil {
+					// null 不能插入到非 nullbale字段中
+					util.Logger.Warn("null value detected, throw this message",
+						zap.String("dimension", dim.Name),
+						zap.String("task", service.taskCfg.Name),
+						zap.String("topic", msg.Topic),
+						zap.Int("partition", msg.Partition),
+						zap.Int64("offset", msg.Offset),
+						zap.String("key", string(msg.Key)),
+						zap.Time("timestamp", *msg.Timestamp))
+					return nil
+				}
 				row = append(row, val)
 			}
 		}