materialize-snowflake: use SQL nulls instead of JSON nulls for null variants

When querying data from staged JSON files, Snowflake interprets a present JSON
`null` value as a JSON null rather than a SQL null, and we end up with values in
variant columns that aren't SQL null, but are actually JSON null instead.

The most expedient way to fix this it seems to is to do some casting in the
queries that interact with the staged JSON files, replacing JSON nulls with SQL
nulls where needed.

Author: williamhbaker

materialize-snowflake/.snapshots/TestSQLGeneration

@@ -95,7 +95,7 @@ ALTER TABLE "a-schema".key_value ALTER COLUMN
 --- Begin "a-schema".key_value mergeInto ---
 MERGE INTO "a-schema".key_value AS l
 USING (
-	SELECT $1[0] AS key1, $1[1] AS key2, $1[2] AS "key!binary", $1[3] AS array, $1[4] AS binary, $1[5] AS boolean, $1[6] AS flow_published_at, $1[7] AS integer, $1[8] AS integerGt64Bit, $1[9] AS integerWithUserDDL, $1[10] AS multiple, $1[11] AS number, $1[12] AS numberCastToString, $1[13] AS object, $1[14] AS string, $1[15] AS stringInteger, $1[16] AS stringInteger39Chars, $1[17] AS stringInteger66Chars, $1[18] AS stringNumber, $1[19] AS flow_document
+	SELECT $1[0] AS key1, $1[1] AS key2, $1[2] AS "key!binary", NULLIF($1[3], PARSE_JSON('null')) AS array, $1[4] AS binary, $1[5] AS boolean, $1[6] AS flow_published_at, $1[7] AS integer, $1[8] AS integerGt64Bit, $1[9] AS integerWithUserDDL, NULLIF($1[10], PARSE_JSON('null')) AS multiple, $1[11] AS number, $1[12] AS numberCastToString, NULLIF($1[13], PARSE_JSON('null')) AS object, $1[14] AS string, $1[15] AS stringInteger, $1[16] AS stringInteger39Chars, $1[17] AS stringInteger66Chars, $1[18] AS stringNumber, $1[19] AS flow_document
 	FROM test-file
 ) AS r
 ON 
@@ -125,7 +125,7 @@ SELECT 0, "a-schema".key_value.flow_document
 COPY INTO "a-schema".key_value (
 	key1, key2, "key!binary", array, binary, boolean, flow_published_at, integer, integerGt64Bit, integerWithUserDDL, multiple, number, numberCastToString, object, string, stringInteger, stringInteger39Chars, stringInteger66Chars, stringNumber, flow_document
 ) FROM (
-	SELECT $1[0] AS key1, $1[1] AS key2, $1[2] AS "key!binary", $1[3] AS array, $1[4] AS binary, $1[5] AS boolean, $1[6] AS flow_published_at, $1[7] AS integer, $1[8] AS integerGt64Bit, $1[9] AS integerWithUserDDL, $1[10] AS multiple, $1[11] AS number, $1[12] AS numberCastToString, $1[13] AS object, $1[14] AS string, $1[15] AS stringInteger, $1[16] AS stringInteger39Chars, $1[17] AS stringInteger66Chars, $1[18] AS stringNumber, $1[19] AS flow_document
+	SELECT $1[0] AS key1, $1[1] AS key2, $1[2] AS "key!binary", NULLIF($1[3], PARSE_JSON('null')) AS array, $1[4] AS binary, $1[5] AS boolean, $1[6] AS flow_published_at, $1[7] AS integer, $1[8] AS integerGt64Bit, $1[9] AS integerWithUserDDL, NULLIF($1[10], PARSE_JSON('null')) AS multiple, $1[11] AS number, $1[12] AS numberCastToString, NULLIF($1[13], PARSE_JSON('null')) AS object, $1[14] AS string, $1[15] AS stringInteger, $1[16] AS stringInteger39Chars, $1[17] AS stringInteger66Chars, $1[18] AS stringNumber, $1[19] AS flow_document
 	FROM test-file
 );
 --- End "a-schema".key_value copyInto ---

materialize-snowflake/sqlgen.go

@@ -249,7 +249,7 @@ CREATE PIPE {{ template "pipe_name" . }}
 ) FROM (
 	SELECT {{ range $ind, $key := $.Table.Columns }}
 	{{- if $ind }}, {{ end -}}
-	$1[{{$ind}}] AS {{$key.Identifier -}}
+	{{ if eq $key.DDL "VARIANT" }}NULLIF($1[{{$ind}}], PARSE_JSON('null')){{ else }}$1[{{$ind}}]{{ end }} AS {{$key.Identifier -}}
 	{{- end }}
 	FROM @flow_v1
 );
@@ -264,7 +264,7 @@ COPY INTO {{ $.Table.Identifier }} (
 ) FROM (
 	SELECT {{ range $ind, $key := $.Table.Columns }}
 	{{- if $ind }}, {{ end -}}
-	$1[{{$ind}}] AS {{$key.Identifier -}}
+	{{ if eq $key.DDL "VARIANT" }}NULLIF($1[{{$ind}}], PARSE_JSON('null')){{ else }}$1[{{$ind}}]{{ end }} AS {{$key.Identifier -}}
 	{{- end }}
 	FROM {{ $.File }}
 );
@@ -276,7 +276,7 @@ MERGE INTO {{ $.Table.Identifier }} AS l
 USING (
 	SELECT {{ range $ind, $key := $.Table.Columns }}
 		{{- if $ind }}, {{ end -}}
-		$1[{{$ind}}] AS {{$key.Identifier -}}
+		{{ if eq $key.DDL "VARIANT" }}NULLIF($1[{{$ind}}], PARSE_JSON('null')){{ else }}$1[{{$ind}}]{{ end }} AS {{$key.Identifier -}}
 	{{- end }}
 	FROM {{ $.File }}
 ) AS r

tests/materialize/materialize-snowflake/snapshot.json

@@ -58,7 +58,7 @@
       },
       "multiple_types": {
         "Table": "multiple_types",
-        "Query": "\nCOPY INTO multiple_types (\n\tid, array_int, binary_field, bool_field, float_field, flow_published_at, multiple, nested, nullable_int, str_field, flow_document\n) FROM (\n\tSELECT $1[0] AS id, $1[1] AS array_int, $1[2] AS binary_field, $1[3] AS bool_field, $1[4] AS float_field, $1[5] AS flow_published_at, $1[6] AS multiple, $1[7] AS nested, $1[8] AS nullable_int, $1[9] AS str_field, $1[10] AS flow_document\n\tFROM <uuid>\n);\n",
+        "Query": "\nCOPY INTO multiple_types (\n\tid, array_int, binary_field, bool_field, float_field, flow_published_at, multiple, nested, nullable_int, str_field, flow_document\n) FROM (\n\tSELECT $1[0] AS id, NULLIF($1[1], PARSE_JSON('null')) AS array_int, $1[2] AS binary_field, $1[3] AS bool_field, $1[4] AS float_field, $1[5] AS flow_published_at, NULLIF($1[6], PARSE_JSON('null')) AS multiple, NULLIF($1[7], PARSE_JSON('null')) AS nested, $1[8] AS nullable_int, $1[9] AS str_field, $1[10] AS flow_document\n\tFROM <uuid>\n);\n",
         "StagedDir": "<uuid>",
         "PipeName": "",
         "PipeFiles": null,
@@ -148,7 +148,7 @@
       },
       "multiple_types": {
         "Table": "multiple_types",
-        "Query": "\nMERGE INTO multiple_types AS l\nUSING (\n\tSELECT $1[0] AS id, $1[1] AS array_int, $1[2] AS binary_field, $1[3] AS bool_field, $1[4] AS float_field, $1[5] AS flow_published_at, $1[6] AS multiple, $1[7] AS nested, $1[8] AS nullable_int, $1[9] AS str_field, $1[10] AS flow_document\n\tFROM <uuid>\n) AS r\nON \n\tl.id = r.id AND l.id >= 1 AND l.id <= 10\nWHEN MATCHED AND r.flow_document='delete' THEN\n\tDELETE\nWHEN MATCHED THEN\n\tUPDATE SET l.array_int = r.array_int, l.binary_field = r.binary_field, l.bool_field = r.bool_field, l.float_field = r.float_field, l.flow_published_at = r.flow_published_at, l.multiple = r.multiple, l.nested = r.nested, l.nullable_int = r.nullable_int, l.str_field = r.str_field, l.flow_document = r.flow_document\nWHEN NOT MATCHED and r.flow_document!='delete' THEN\n\tINSERT (id, array_int, binary_field, bool_field, float_field, flow_published_at, multiple, nested, nullable_int, str_field, flow_document)\n\tVALUES (r.id, r.array_int, r.binary_field, r.bool_field, r.float_field, r.flow_published_at, r.multiple, r.nested, r.nullable_int, r.str_field, r.flow_document);\n",
+        "Query": "\nMERGE INTO multiple_types AS l\nUSING (\n\tSELECT $1[0] AS id, NULLIF($1[1], PARSE_JSON('null')) AS array_int, $1[2] AS binary_field, $1[3] AS bool_field, $1[4] AS float_field, $1[5] AS flow_published_at, NULLIF($1[6], PARSE_JSON('null')) AS multiple, NULLIF($1[7], PARSE_JSON('null')) AS nested, $1[8] AS nullable_int, $1[9] AS str_field, $1[10] AS flow_document\n\tFROM <uuid>\n) AS r\nON \n\tl.id = r.id AND l.id >= 1 AND l.id <= 10\nWHEN MATCHED AND r.flow_document='delete' THEN\n\tDELETE\nWHEN MATCHED THEN\n\tUPDATE SET l.array_int = r.array_int, l.binary_field = r.binary_field, l.bool_field = r.bool_field, l.float_field = r.float_field, l.flow_published_at = r.flow_published_at, l.multiple = r.multiple, l.nested = r.nested, l.nullable_int = r.nullable_int, l.str_field = r.str_field, l.flow_document = r.flow_document\nWHEN NOT MATCHED and r.flow_document!='delete' THEN\n\tINSERT (id, array_int, binary_field, bool_field, float_field, flow_published_at, multiple, nested, nullable_int, str_field, flow_document)\n\tVALUES (r.id, r.array_int, r.binary_field, r.bool_field, r.float_field, r.flow_published_at, r.multiple, r.nested, r.nullable_int, r.str_field, r.flow_document);\n",
         "StagedDir": "<uuid>",
         "PipeName": "",
         "PipeFiles": null,
@@ -219,7 +219,7 @@
       },
       "multiple_types": {
         "PipeName": "",
-        "Query": "\nMERGE INTO multiple_types AS l\nUSING (\n\tSELECT $1[0] AS id, $1[1] AS array_int, $1[2] AS binary_field, $1[3] AS bool_field, $1[4] AS float_field, $1[5] AS flow_published_at, $1[6] AS multiple, $1[7] AS nested, $1[8] AS nullable_int, $1[9] AS str_field, $1[10] AS flow_document\n\tFROM <uuid>\n) AS r\nON \n\tl.id = r.id AND l.id >= 1 AND l.id <= 10\nWHEN MATCHED AND r.flow_document='delete' THEN\n\tDELETE\nWHEN MATCHED THEN\n\tUPDATE SET l.array_int = r.array_int, l.binary_field = r.binary_field, l.bool_field = r.bool_field, l.float_field = r.float_field, l.flow_published_at = r.flow_published_at, l.multiple = r.multiple, l.nested = r.nested, l.nullable_int = r.nullable_int, l.str_field = r.str_field, l.flow_document = r.flow_document\nWHEN NOT MATCHED and r.flow_document!='delete' THEN\n\tINSERT (id, array_int, binary_field, bool_field, float_field, flow_published_at, multiple, nested, nullable_int, str_field, flow_document)\n\tVALUES (r.id, r.array_int, r.binary_field, r.bool_field, r.float_field, r.flow_published_at, r.multiple, r.nested, r.nullable_int, r.str_field, r.flow_document);\n",
+        "Query": "\nMERGE INTO multiple_types AS l\nUSING (\n\tSELECT $1[0] AS id, NULLIF($1[1], PARSE_JSON('null')) AS array_int, $1[2] AS binary_field, $1[3] AS bool_field, $1[4] AS float_field, $1[5] AS flow_published_at, NULLIF($1[6], PARSE_JSON('null')) AS multiple, NULLIF($1[7], PARSE_JSON('null')) AS nested, $1[8] AS nullable_int, $1[9] AS str_field, $1[10] AS flow_document\n\tFROM <uuid>\n) AS r\nON \n\tl.id = r.id AND l.id >= 1 AND l.id <= 10\nWHEN MATCHED AND r.flow_document='delete' THEN\n\tDELETE\nWHEN MATCHED THEN\n\tUPDATE SET l.array_int = r.array_int, l.binary_field = r.binary_field, l.bool_field = r.bool_field, l.float_field = r.float_field, l.flow_published_at = r.flow_published_at, l.multiple = r.multiple, l.nested = r.nested, l.nullable_int = r.nullable_int, l.str_field = r.str_field, l.flow_document = r.flow_document\nWHEN NOT MATCHED and r.flow_document!='delete' THEN\n\tINSERT (id, array_int, binary_field, bool_field, float_field, flow_published_at, multiple, nested, nullable_int, str_field, flow_document)\n\tVALUES (r.id, r.array_int, r.binary_field, r.bool_field, r.float_field, r.flow_published_at, r.multiple, r.nested, r.nullable_int, r.str_field, r.flow_document);\n",
         "StagedDir": "<uuid>",
         "Table": "multiple_types",
         "Version": "ffffffffffffffff"
@@ -548,7 +548,7 @@
       "FLOAT_FIELD": 88.88,
       "FLOW_PUBLISHED_AT": "1970-01-01T01:00:21Z",
       "ID": "8",
-      "MULTIPLE": "null",
+      "MULTIPLE": null,
       "NESTED": "{\n  \"id\": \"i8\"\n}",
       "NULLABLE_INT": "8",
       "STR_FIELD": "str8 v2"
@@ -560,7 +560,7 @@
       "FLOAT_FIELD": 99.99,
       "FLOW_PUBLISHED_AT": "1970-01-01T01:00:22Z",
       "ID": "9",
-      "MULTIPLE": "null",
+      "MULTIPLE": null,
       "NESTED": "{\n  \"id\": \"i9\"\n}",
       "NULLABLE_INT": null,
       "STR_FIELD": "str9 v2"
@@ -572,7 +572,7 @@
       "FLOAT_FIELD": 1010.101,
       "FLOW_PUBLISHED_AT": "1970-01-01T01:00:23Z",
       "ID": "10",
-      "MULTIPLE": "null",
+      "MULTIPLE": null,
       "NESTED": "{\n  \"id\": \"i10\"\n}",
       "NULLABLE_INT": "10",
       "STR_FIELD": "str10 v2"