Merge pull request #14 from pinax-network/feature/handle_pointers_dates

handle pointer and dates in clickhouse dialect

.gitignore

@@ -7,3 +7,4 @@ dist/
 devel/data*
 build/
 *.spkg
+/substreams-sink-sql

cmd/substreams-sink-sql/main.go

@@ -1,7 +1,6 @@
 package main
 
 import (
-	"github.com/spf13/viper"
 	"net/http"
 	_ "net/http/pprof"
 	"time"
@@ -19,9 +18,6 @@ import (
 var version = "dev"
 
 func main() {
-
-	viper.AutomaticEnv()
-
 	Run("substreams-sink-sql", "Substreams SQL Sink",
 		sinkRunCmd,
 		sinkSetupCmd,

db/dialect_clickhouse.go

@@ -233,6 +233,8 @@ func convertToType(value string, valueType reflect.Type) (any, error) {
 			var err error
 			if strings.Contains(value, "T") && strings.HasSuffix(value, "Z") {
 				v, err = time.Parse("2006-01-02T15:04:05Z", value)
+			} else if dateRegex.MatchString(value) {
+				v, err = time.Parse("2006-01-02", value)
 			} else {
 				v, err = time.Parse("2006-01-02 15:04:05", value)
 			}
@@ -250,7 +252,60 @@ func convertToType(value string, valueType reflect.Type) (any, error) {
 			newInt.SetString(value, 10)
 			return newInt, nil
 		}
-		return "", fmt.Errorf("unsupported pointer type %s", valueType)
+
+		switch valueType.Elem().Kind() {
+		case reflect.String:
+			return &value, nil
+		case reflect.Bool:
+			res, err := strconv.ParseBool(value)
+			if err != nil {
+				return nil, err
+			}
+			return &res, err
+		case reflect.Int:
+			v, err := strconv.ParseInt(value, 10, 0)
+			res := int(v)
+			return &res, err
+		case reflect.Int8:
+			v, err := strconv.ParseInt(value, 10, 8)
+			res := int8(v)
+			return &res, err
+		case reflect.Int16:
+			v, err := strconv.ParseInt(value, 10, 16)
+			res := int16(v)
+			return &res, err
+		case reflect.Int32:
+			v, err := strconv.ParseInt(value, 10, 32)
+			res := int32(v)
+			return &res, err
+		case reflect.Int64:
+			v, err := strconv.ParseInt(value, 10, 64)
+			return &v, err
+		case reflect.Uint:
+			v, err := strconv.ParseUint(value, 10, 0)
+			res := uint(v)
+			return &res, err
+		case reflect.Uint8:
+			v, err := strconv.ParseUint(value, 10, 8)
+			res := uint8(v)
+			return &res, err
+		case reflect.Uint16:
+			v, err := strconv.ParseUint(value, 10, 16)
+			res := uint16(v)
+			return &res, err
+		case reflect.Uint32:
+			v, err := strconv.ParseUint(value, 10, 32)
+			res := uint32(v)
+			return &res, err
+		case reflect.Uint64:
+			v, err := strconv.ParseUint(value, 10, 0)
+			return &v, err
+		case reflect.Float32, reflect.Float64:
+			v, err := strconv.ParseFloat(value, 10)
+			return &v, err
+		default:
+			return "", fmt.Errorf("unsupported pointer type %s", valueType)
+		}
 	default:
 		return value, nil
 	}

db/operations.go

@@ -76,6 +76,7 @@ func (o *Operation) mergeData(newData map[string]string) error {
 }
 
 var integerRegex = regexp.MustCompile(`^\d+$`)
+var dateRegex = regexp.MustCompile(`^\d{4}-\d{2}-\d{2}$`)
 var reflectTypeTime = reflect.TypeOf(time.Time{})
 
 func EscapeIdentifier(valueToEscape string) string {