fix(clickhouse): handle nullable JSON columns in normalization

Add support for `Nullable(JSON)` type in ClickHouse connector by:
- Recognizing `Nullable(JSON)` in schema type mapping
- Using `JSONExtract` with proper type casting for both JSON and
  Nullable(JSON) columns
- Casting nullable JSON fields to `Nullable(JSON)` in table functions

Author: aldofunes

flow/connectors/clickhouse/clickhouse.go

@@ -426,7 +426,7 @@ func GetTableSchemaForTable(tm *protos.TableMapping, columns []driver.ColumnType
 			qkind = types.QValueKindArrayBoolean
 		case "Array(Date)":
 			qkind = types.QValueKindArrayDate
-		case "JSON":
+		case "JSON", "Nullable(JSON)":
 			qkind = types.QValueKindJSON
 		default:
 			if strings.Contains(column.DatabaseTypeName(), "Decimal") {

flow/connectors/clickhouse/normalize_query.go

@@ -205,19 +205,25 @@ func (t *NormalizeQueryGenerator) BuildQuery(ctx context.Context) (string, error
 					peerdb_clickhouse.QuoteIdentifier(dstColName),
 				)
 			}
-		case "JSON", "Nullable(JSON)":
-			fmt.Fprintf(&projection,
-				"JSONExtractString(_peerdb_data, %s)::JSON AS %s,",
-				peerdb_clickhouse.QuoteLiteral(colName),
-				peerdb_clickhouse.QuoteIdentifier(dstColName),
-			)
-			if t.enablePrimaryUpdate {
-				fmt.Fprintf(&projectionUpdate,
-					"JSONExtractString(_peerdb_match_data, %s)::JSON AS %s,",
+			case "JSON", "Nullable(JSON)":
+				stringType := strings.Replace(clickHouseType, "JSON", "String", 1)
+
+				fmt.Fprintf(&projection,
+					"JSONExtract(_peerdb_data, %s, '%s')::%s AS %s,",
 					peerdb_clickhouse.QuoteLiteral(colName),
+					stringType,
+					clickHouseType,
 					peerdb_clickhouse.QuoteIdentifier(dstColName),
 				)
-			}
+				if t.enablePrimaryUpdate {
+					fmt.Fprintf(&projectionUpdate,
+						"JSONExtract(_peerdb_match_data, %s, '%s')::%s AS %s,",
+						peerdb_clickhouse.QuoteLiteral(colName),
+						stringType,
+						clickHouseType,
+						peerdb_clickhouse.QuoteIdentifier(dstColName),
+					)
+				}
 
 		default:
 			projLen := projection.Len()

flow/connectors/clickhouse/table_function.go

@@ -51,6 +51,10 @@ func jsonFieldExpressionConverter(
 		return sourceFieldIdentifier, nil
 	}
 
+	if field.Nullable {
+		return fmt.Sprintf("CAST(%s, 'Nullable(JSON)')", sourceFieldIdentifier), nil
+	}
+
 	return fmt.Sprintf("CAST(%s, 'JSON')", sourceFieldIdentifier), nil
 }