From 3816eb3ec98d284ae0f3cc069a96a31ab5db9e39 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Frederik=20Sch=C3=B6ll?= <fschoell@posteo.de>
Date: Tue, 5 Nov 2024 14:24:26 +0100
Subject: [PATCH] add support for Nullable types in Clickhouse

---
 .gitignore               |  1 +
 CHANGELOG.md             |  4 ++++
 db/dialect_clickhouse.go | 15 ++++++++++++++-
 3 files changed, 19 insertions(+), 1 deletion(-)

diff --git a/.gitignore b/.gitignore
index 76ce6c7..43f32f1 100644
--- a/.gitignore
+++ b/.gitignore
@@ -7,3 +7,4 @@ dist/
 devel/data*
 build/
 *.spkg
+/substreams-sink-sql
diff --git a/CHANGELOG.md b/CHANGELOG.md
index e33f0e3..4a473c6 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -5,6 +5,10 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## Unreleased
+
+* Added support for `Nullable` types in Clickhouse.
+
 ## v4.2.1
 
 * Bump substreams to v1.10.3 to support new manifest data like `protobuf:excludePaths`
diff --git a/db/dialect_clickhouse.go b/db/dialect_clickhouse.go
index 6387c1a..48b3d56 100644
--- a/db/dialect_clickhouse.go
+++ b/db/dialect_clickhouse.go
@@ -247,7 +247,20 @@ func convertToType(value string, valueType reflect.Type) (any, error) {
 			newInt.SetString(value, 10)
 			return newInt, nil
 		}
-		return "", fmt.Errorf("unsupported pointer type %s", valueType)
+
+		elemType := valueType.Elem()
+		val, err := convertToType(value, elemType)
+		if err != nil {
+			return nil, fmt.Errorf("invalid pointer type: %w", err)
+		}
+
+		// We cannot just return &val here as this will return an *interface{} that the Clickhouse Go client won't be
+		// able to convert on inserting. Instead, we create a new variable using the type that valueType has been
+		// pointing to, assign the converted value from convertToType to that and then return a pointer to the new variable.
+		result := reflect.New(elemType).Elem()
+		result.Set(reflect.ValueOf(val))
+		return result.Addr().Interface(), nil
+
 	default:
 		return value, nil
 	}