[Snowflake] Distinguishing NULL and empty strings (#112)

clients/snowflake/cast.go

@@ -18,7 +18,8 @@ import (
 // This is necessary because CSV writers require values to in `string`.
 func CastColValStaging(colVal interface{}, colKind typing.Column) (string, error) {
 	if colVal == nil {
-		return "", fmt.Errorf("colVal is nil")
+		// \\N needs to match NULL_IF(...) from ddl.go
+		return `\\N`, nil
 	}
 
 	colValString := fmt.Sprint(colVal)

clients/snowflake/cast_test.go

@@ -35,6 +35,24 @@ func evaluateTestCase(t *testing.T, testCase _testCase) {
 
 func (s *SnowflakeTestSuite) TestCastColValStaging_Basic() {
 	testCases := []_testCase{
+		{
+			name:   "empty string",
+			colVal: "",
+			colKind: typing.Column{
+				KindDetails: typing.String,
+			},
+
+			expectedString: "",
+		},
+		{
+			name:   "null value (string, not that it matters)",
+			colVal: nil,
+			colKind: typing.Column{
+				KindDetails: typing.String,
+			},
+
+			expectedString: `\\N`,
+		},
 		{
 			name:   "string",
 			colVal: "foo",

clients/snowflake/staging.go

@@ -81,17 +81,13 @@ func (s *Store) loadTemporaryTable(tableData *optimization.TableData, newTableNa
 		for _, col := range tableData.ReadOnlyInMemoryCols().GetColumnsToUpdate() {
 			colKind, _ := tableData.ReadOnlyInMemoryCols().GetColumn(col)
 			colVal := value[col]
-			if colVal != nil {
-				// Check
-				castedValue, castErr := CastColValStaging(colVal, colKind)
-				if castErr != nil {
-					return "", castErr
-				}
-
-				row = append(row, castedValue)
-			} else {
-				row = append(row, "")
+			// Check
+			castedValue, castErr := CastColValStaging(colVal, colKind)
+			if castErr != nil {
+				return "", castErr
 			}
+
+			row = append(row, castedValue)
 		}
 
 		if err = writer.Write(row); err != nil {

clients/snowflake/staging_test.go

@@ -55,7 +55,7 @@ func (s *SnowflakeTestSuite) TestPrepareTempTable() {
 	createQuery, _ := s.fakeStageStore.ExecArgsForCall(0)
 
 	prefixQuery := fmt.Sprintf(
-		`CREATE TABLE IF NOT EXISTS %s (user_id string,first_name string,last_name string) STAGE_COPY_OPTIONS = ( PURGE = TRUE ) STAGE_FILE_FORMAT = ( TYPE = 'csv' FIELD_DELIMITER= '\t' FIELD_OPTIONALLY_ENCLOSED_BY='"') COMMENT=`, tempTableName)
+		`CREATE TABLE IF NOT EXISTS %s (user_id string,first_name string,last_name string) STAGE_COPY_OPTIONS = ( PURGE = TRUE ) STAGE_FILE_FORMAT = ( TYPE = 'csv' FIELD_DELIMITER= '\t' FIELD_OPTIONALLY_ENCLOSED_BY='"' NULL_IF='\\N' EMPTY_FIELD_AS_NULL=FALSE) COMMENT=`, tempTableName)
 	containsPrefix := strings.HasPrefix(createQuery, prefixQuery)
 	assert.True(s.T(), containsPrefix, fmt.Sprintf("createQuery:%v, prefixQuery:%s", createQuery, prefixQuery))
 	resourceName := addPrefixToTableName(tempTableName, "%")

clients/snowflake/sweep_test.go

@@ -101,7 +101,6 @@ func (s *SnowflakeTestSuite) TestSweep() {
 	})
 
 	assert.NoError(s.T(), s.stageStore.Sweep(s.ctx))
-	fmt.Println(s.fakeStageStore.QueryCallCount())
 	query, _ := s.fakeStageStore.QueryArgsForCall(0)
 	assert.Equal(s.T(), `SELECT table_name, comment FROM db.information_schema.tables where table_name ILIKE '%__artie%' AND table_schema = UPPER('schema')`, query)
 }

lib/dwh/ddl/ddl.go

@@ -110,7 +110,7 @@ func AlterTable(_ context.Context, args AlterTableArgs, cols ...typing.Column) e
 				// TEMPORARY Table syntax - https://docs.snowflake.com/en/sql-reference/sql/create-table
 				// PURGE syntax - https://docs.snowflake.com/en/sql-reference/sql/copy-into-table#purging-files-after-loading
 				// FIELD_OPTIONALLY_ENCLOSED_BY - is needed because CSV will try to escape any values that have `"`
-				sqlQuery = fmt.Sprintf(`CREATE TABLE IF NOT EXISTS %s (%s) STAGE_COPY_OPTIONS = ( PURGE = TRUE ) STAGE_FILE_FORMAT = ( TYPE = 'csv' FIELD_DELIMITER= '\t' FIELD_OPTIONALLY_ENCLOSED_BY='"') COMMENT='%s'`,
+				sqlQuery = fmt.Sprintf(`CREATE TABLE IF NOT EXISTS %s (%s) STAGE_COPY_OPTIONS = ( PURGE = TRUE ) STAGE_FILE_FORMAT = ( TYPE = 'csv' FIELD_DELIMITER= '\t' FIELD_OPTIONALLY_ENCLOSED_BY='"' NULL_IF='\\N' EMPTY_FIELD_AS_NULL=FALSE) COMMENT='%s'`,
 					args.FqTableName, strings.Join(colSQLParts, ","),
 					// Comment on the table
 					fmt.Sprintf("%s%s", constants.SnowflakeExpireCommentPrefix, expiryString))

lib/dwh/ddl/ddl_temp_test.go

@@ -82,9 +82,10 @@ func (d *DDLTestSuite) TestCreateTemporaryTable() {
 	assert.NoError(d.T(), err)
 	assert.Equal(d.T(), 1, d.fakeSnowflakeStagesStore.ExecCallCount())
 	query, _ := d.fakeSnowflakeStagesStore.ExecArgsForCall(0)
+
 	assert.Contains(d.T(),
 		query,
-		`CREATE TABLE IF NOT EXISTS db.schema.tempTableName (foo string,bar float) STAGE_COPY_OPTIONS = ( PURGE = TRUE ) STAGE_FILE_FORMAT = ( TYPE = 'csv' FIELD_DELIMITER= '\t' FIELD_OPTIONALLY_ENCLOSED_BY='"') COMMENT=`,
+		`CREATE TABLE IF NOT EXISTS db.schema.tempTableName (foo string,bar float) STAGE_COPY_OPTIONS = ( PURGE = TRUE ) STAGE_FILE_FORMAT = ( TYPE = 'csv' FIELD_DELIMITER= '\t' FIELD_OPTIONALLY_ENCLOSED_BY='"' NULL_IF='\\N' EMPTY_FIELD_AS_NULL=FALSE) COMMENT=`,
 		query)
 
 	// BigQuery

models/event/event.go

@@ -23,18 +23,17 @@ type Event struct {
 	Table          string
 	PrimaryKeyMap  map[string]interface{}
 	Data           map[string]interface{} // json serialized column data
-	OptiomalSchema map[string]typing.KindDetails
+	OptionalSchema map[string]typing.KindDetails
 	Columns        *typing.Columns
 	ExecutionTime  time.Time // When the SQL command was executed
-
 }
 
 func ToMemoryEvent(ctx context.Context, event cdc.Event, pkMap map[string]interface{}, tc *kafkalib.TopicConfig) Event {
 	return Event{
 		Table:          tc.TableName,
 		PrimaryKeyMap:  pkMap,
 		ExecutionTime:  event.GetExecutionTime(),
-		OptiomalSchema: event.GetOptionalSchema(ctx),
+		OptionalSchema: event.GetOptionalSchema(ctx),
 		Columns:        event.GetColumns(),
 		Data:           event.GetData(ctx, pkMap, tc),
 	}
@@ -147,15 +146,15 @@ func (e *Event) Save(ctx context.Context, topicConfig *kafkalib.TopicConfig, mes
 				// This would only happen if the columns did not get passed in initially.
 				inMemoryColumns.AddColumn(typing.Column{
 					Name:        newColName,
-					KindDetails: typing.ParseValue(_col, e.OptiomalSchema, val),
+					KindDetails: typing.ParseValue(_col, e.OptionalSchema, val),
 				})
 			} else {
 				if retrievedColumn.KindDetails == typing.Invalid {
 					// If colType is Invalid, let's see if we can update it to a better type
 					// If everything is nil, we don't need to add a column
 					// However, it's important to create a column even if it's nil.
 					// This is because we don't want to think that it's okay to drop a column in DWH
-					if kindDetails := typing.ParseValue(_col, e.OptiomalSchema, val); kindDetails.Kind != typing.Invalid.Kind {
+					if kindDetails := typing.ParseValue(_col, e.OptionalSchema, val); kindDetails.Kind != typing.Invalid.Kind {
 						retrievedColumn.KindDetails = kindDetails
 						inMemoryColumns.UpdateColumn(retrievedColumn)
 					}

models/event/event_save_test.go

@@ -120,7 +120,7 @@ func (e *EventsTestSuite) TestEventSaveOptionalSchema() {
 			"json_object_string":         `{"foo": "bar"}`,
 			"json_object_no_schema":      `{"foo": "bar"}`,
 		},
-		OptiomalSchema: map[string]typing.KindDetails{
+		OptionalSchema: map[string]typing.KindDetails{
 			// Explicitly casting this as a string.
 			"created_at_date_string": typing.String,
 			"json_object_string":     typing.String,