[Extended Time] Pulling TIMESTAMP_NTZ into a separate data type (#992)

clients/bigquery/dialect/dialect.go

@@ -49,14 +49,14 @@ func (BigQueryDialect) DataTypeForKind(kindDetails typing.KindDetails, _ bool) s
 		return "json"
 	case typing.Date.Kind:
 		return "date"
+	case typing.TimestampNTZ.Kind:
+		return "datetime"
 	case typing.ETime.Kind:
 		switch kindDetails.ExtendedTimeDetails.Type {
 		case ext.TimestampTZKindType:
 			// https://cloud.google.com/bigquery/docs/reference/standard-sql/data-types#datetime_type
 			// We should be using TIMESTAMP since it's an absolute point in time.
 			return "timestamp"
-		case ext.TimestampNTZKindType:
-			return "datetime"
 		case ext.TimeKindType:
 			return "time"
 		}
@@ -108,7 +108,7 @@ func (BigQueryDialect) KindForDataType(rawBqType string, _ string) (typing.KindD
 	case "timestamp":
 		return typing.NewExtendedTimeDetails(typing.ETime, ext.TimestampTZKindType, "")
 	case "datetime":
-		return typing.NewExtendedTimeDetails(typing.ETime, ext.TimestampNTZKindType, "")
+		return typing.TimestampNTZ, nil
 	case "time":
 		return typing.NewExtendedTimeDetails(typing.ETime, ext.TimeKindType, "")
 	case "date":

clients/bigquery/dialect/dialect_test.go

@@ -92,7 +92,7 @@ func TestBigQueryDialect_KindForDataType(t *testing.T) {
 		"record":             typing.Struct,
 		"json":               typing.Struct,
 		// Datetime
-		"datetime":  typing.MustNewExtendedTimeDetails(typing.ETime, ext.TimestampTZKindType, ""),
+		"datetime":  typing.TimestampNTZ,
 		"timestamp": typing.MustNewExtendedTimeDetails(typing.ETime, ext.TimestampTZKindType, ""),
 		"time":      typing.MustNewExtendedTimeDetails(typing.ETime, ext.TimeKindType, ""),
 		"date":      typing.Date,

clients/bigquery/storagewrite.go

@@ -40,14 +40,14 @@ func columnToTableFieldSchema(column columns.Column) (*storagepb.TableFieldSchem
 		fieldType = storagepb.TableFieldSchema_STRING
 	case typing.Date.Kind:
 		fieldType = storagepb.TableFieldSchema_DATE
+	case typing.TimestampNTZ.Kind:
+		fieldType = storagepb.TableFieldSchema_DATETIME
 	case typing.ETime.Kind:
 		switch column.KindDetails.ExtendedTimeDetails.Type {
 		case ext.TimeKindType:
 			fieldType = storagepb.TableFieldSchema_TIME
 		case ext.TimestampTZKindType:
 			fieldType = storagepb.TableFieldSchema_TIMESTAMP
-		case ext.TimestampNTZKindType:
-			fieldType = storagepb.TableFieldSchema_DATETIME
 		default:
 			return nil, fmt.Errorf("unsupported extended time details type: %q", column.KindDetails.ExtendedTimeDetails.Type)
 		}
@@ -209,6 +209,13 @@ func rowToMessage(row map[string]any, columns []columns.Column, messageDescripto
 
 			daysSinceEpoch := _time.Unix() / (60 * 60 * 24)
 			message.Set(field, protoreflect.ValueOfInt32(int32(daysSinceEpoch)))
+		case typing.TimestampNTZ.Kind:
+			_time, err := ext.ParseTimestampNTZFromInterface(value)
+			if err != nil {
+				return nil, fmt.Errorf("failed to cast value as time.Time, value: '%v', err: %w", value, err)
+			}
+
+			message.Set(field, protoreflect.ValueOfInt64(encodePacked64DatetimeMicros(_time)))
 		case typing.ETime.Kind:
 			if err := column.KindDetails.EnsureExtendedTimeDetails(); err != nil {
 				return nil, err
@@ -227,8 +234,6 @@ func rowToMessage(row map[string]any, columns []columns.Column, messageDescripto
 					return nil, err
 				}
 				message.Set(field, protoreflect.ValueOfInt64(_time.UnixMicro()))
-			case ext.TimestampNTZKindType:
-				message.Set(field, protoreflect.ValueOfInt64(encodePacked64DatetimeMicros(_time)))
 			default:
 				return nil, fmt.Errorf("unsupported extended time details: %q", column.KindDetails.ExtendedTimeDetails.Type)
 			}

clients/bigquery/storagewrite_test.go

@@ -168,8 +168,9 @@ func TestRowToMessage(t *testing.T) {
 		columns.NewColumn("c_string", typing.String),
 		columns.NewColumn("c_string_decimal", typing.String),
 		columns.NewColumn("c_time", typing.MustNewExtendedTimeDetails(typing.ETime, ext.TimeKindType, "")),
+		columns.NewColumn("c_timestamp", typing.MustNewExtendedTimeDetails(typing.ETime, ext.TimestampTZKindType, "")),
 		columns.NewColumn("c_date", typing.Date),
-		columns.NewColumn("c_datetime", typing.MustNewExtendedTimeDetails(typing.ETime, ext.TimestampTZKindType, "")),
+		columns.NewColumn("c_datetime", typing.TimestampNTZ),
 		columns.NewColumn("c_struct", typing.Struct),
 		columns.NewColumn("c_array", typing.Array),
 	}
@@ -188,8 +189,9 @@ func TestRowToMessage(t *testing.T) {
 		"c_string":         "foo bar",
 		"c_string_decimal": decimal.NewDecimal(numbers.MustParseDecimal("1.61803")),
 		"c_time":           ext.NewExtendedTime(time.Date(0, 0, 0, 4, 5, 6, 7, time.UTC), ext.TimeKindType, ""),
+		"c_timestamp":      ext.NewExtendedTime(time.Date(2001, 2, 3, 4, 5, 6, 7, time.UTC), ext.TimestampTZKindType, ""),
 		"c_date":           time.Date(2001, 2, 3, 0, 0, 0, 0, time.UTC),
-		"c_datetime":       ext.NewExtendedTime(time.Date(2001, 2, 3, 4, 5, 6, 7, time.UTC), ext.TimestampTZKindType, ""),
+		"c_datetime":       time.Date(2001, 2, 3, 4, 5, 6, 7, time.UTC),
 		"c_struct":         map[string]any{"baz": []string{"foo", "bar"}},
 		"c_array":          []string{"foo", "bar"},
 	}
@@ -220,8 +222,9 @@ func TestRowToMessage(t *testing.T) {
 		"cString":        "foo bar",
 		"cStringDecimal": "1.61803",
 		"cTime":          "17521704960",
+		"cTimestamp":     "981173106000000",
 		"cDate":          float64(11356),
-		"cDatetime":      "981173106000000",
+		"cDatetime":      "140817083031093248",
 		"cStruct":        `{"baz":["foo","bar"]}`,
 		"cArray":         []any{"foo", "bar"},
 	}, result)

clients/databricks/dialect/typing.go

@@ -26,14 +26,14 @@ func (DatabricksDialect) DataTypeForKind(kindDetails typing.KindDetails, _ bool)
 		return "BOOLEAN"
 	case typing.Date.Kind:
 		return "DATE"
+	case typing.TimestampNTZ.Kind:
+		// This is currently in public preview, to use this, the customer will need to enable [timestampNtz] in their delta tables.
+		// Ref: https://docs.databricks.com/en/sql/language-manual/data-types/timestamp-ntz-type.html
+		return "TIMESTAMP_NTZ"
 	case typing.ETime.Kind:
 		switch kindDetails.ExtendedTimeDetails.Type {
 		case ext.TimestampTZKindType:
 			return "TIMESTAMP"
-		case ext.TimestampNTZKindType:
-			// This is currently in public preview, to use this, the customer will need to enable [timestampNtz] in their delta tables.
-			// Ref: https://docs.databricks.com/en/sql/language-manual/data-types/timestamp-ntz-type.html
-			return "TIMESTAMP_NTZ"
 		case ext.TimeKindType:
 			return "STRING"
 		}
@@ -76,7 +76,7 @@ func (DatabricksDialect) KindForDataType(rawType string, _ string) (typing.KindD
 	case "timestamp":
 		return typing.NewExtendedTimeDetails(typing.ETime, ext.TimestampTZKindType, "")
 	case "timestamp_ntz":
-		return typing.NewExtendedTimeDetails(typing.ETime, ext.TimestampNTZKindType, "")
+		return typing.TimestampNTZ, nil
 	}
 
 	return typing.Invalid, fmt.Errorf("unsupported data type: %q", rawType)

clients/databricks/dialect/typing_test.go

@@ -151,7 +151,7 @@ func TestDatabricksDialect_KindForDataType(t *testing.T) {
 		// Timestamp NTZ
 		kd, err := DatabricksDialect{}.KindForDataType("TIMESTAMP_NTZ", "")
 		assert.NoError(t, err)
-		assert.Equal(t, typing.MustNewExtendedTimeDetails(typing.ETime, ext.TimestampNTZKindType, ""), kd)
+		assert.Equal(t, typing.TimestampNTZ, kd)
 	}
 	{
 		// Variant

clients/mssql/dialect/dialect.go

@@ -55,13 +55,13 @@ func (MSSQLDialect) DataTypeForKind(kindDetails typing.KindDetails, isPk bool) s
 		return "BIT"
 	case typing.Date.Kind:
 		return "DATE"
+	case typing.TimestampNTZ.Kind:
+		// Using datetime2 because it's the recommendation, and it provides more precision: https://stackoverflow.com/a/1884088
+		return "datetime2"
 	case typing.ETime.Kind:
 		switch kindDetails.ExtendedTimeDetails.Type {
 		case ext.TimestampTZKindType:
 			return "datetimeoffset"
-		case ext.TimestampNTZKindType:
-			// Using datetime2 because it's the recommendation, and it provides more precision: https://stackoverflow.com/a/1884088
-			return "datetime2"
 		case ext.TimeKindType:
 			return "time"
 		}
@@ -116,7 +116,7 @@ func (MSSQLDialect) KindForDataType(rawType string, stringPrecision string) (typ
 	case
 		"datetime",
 		"datetime2":
-		return typing.NewExtendedTimeDetails(typing.ETime, ext.TimestampNTZKindType, "")
+		return typing.TimestampNTZ, nil
 	case "datetimeoffset":
 		return typing.NewExtendedTimeDetails(typing.ETime, ext.TimestampTZKindType, "")
 	case "time":

clients/mssql/dialect/dialect_test.go

@@ -50,22 +50,23 @@ func TestMSSQLDialect_KindForDataType(t *testing.T) {
 	dialect := MSSQLDialect{}
 
 	colToExpectedKind := map[string]typing.KindDetails{
-		"char":      typing.String,
-		"varchar":   typing.String,
-		"nchar":     typing.String,
-		"nvarchar":  typing.String,
-		"ntext":     typing.String,
-		"text":      typing.String,
-		"smallint":  typing.Integer,
-		"tinyint":   typing.Integer,
-		"int":       typing.Integer,
-		"float":     typing.Float,
-		"real":      typing.Float,
-		"bit":       typing.Boolean,
-		"date":      typing.Date,
-		"time":      typing.MustNewExtendedTimeDetails(typing.ETime, ext.TimeKindType, ""),
-		"datetime":  typing.MustNewExtendedTimeDetails(typing.ETime, ext.TimestampTZKindType, ""),
-		"datetime2": typing.MustNewExtendedTimeDetails(typing.ETime, ext.TimestampTZKindType, ""),
+		"char":           typing.String,
+		"varchar":        typing.String,
+		"nchar":          typing.String,
+		"nvarchar":       typing.String,
+		"ntext":          typing.String,
+		"text":           typing.String,
+		"smallint":       typing.Integer,
+		"tinyint":        typing.Integer,
+		"int":            typing.Integer,
+		"float":          typing.Float,
+		"real":           typing.Float,
+		"bit":            typing.Boolean,
+		"date":           typing.Date,
+		"time":           typing.MustNewExtendedTimeDetails(typing.ETime, ext.TimeKindType, ""),
+		"datetime":       typing.TimestampNTZ,
+		"datetime2":      typing.TimestampNTZ,
+		"datetimeoffset": typing.ETime,
 	}
 
 	for col, expectedKind := range colToExpectedKind {

clients/mssql/values.go

@@ -33,6 +33,13 @@ func parseValue(colVal any, colKind columns.Column) (any, error) {
 			return "", fmt.Errorf("failed to cast colVal as time.Time, colVal: '%v', err: %w", colVal, err)
 		}
 
+		return _time, nil
+	case typing.TimestampNTZ.Kind:
+		_time, err := ext.ParseTimestampNTZFromInterface(colVal)
+		if err != nil {
+			return "", fmt.Errorf("failed to cast colVal as time.Time, colVal: '%v', err: %w", colVal, err)
+		}
+
 		return _time, nil
 	case typing.ETime.Kind:
 		if err := colKind.KindDetails.EnsureExtendedTimeDetails(); err != nil {

clients/mssql/values_test.go

@@ -33,6 +33,21 @@ func TestParseValue(t *testing.T) {
 			assert.Equal(t, "2021-01-01", val.(time.Time).Format(ext.PostgresDateFormat))
 		}
 	}
+	{
+		// Timestamp NTZ
+		{
+			// String
+			val, err := parseValue("2021-01-04T09:32:00", columns.NewColumn("timestamp_ntz", typing.TimestampNTZ))
+			assert.NoError(t, err)
+			assert.Equal(t, time.Date(2021, time.January, 4, 9, 32, 0, 0, time.UTC), val.(time.Time))
+		}
+		{
+			// time.Time
+			val, err := parseValue(time.Date(2021, time.January, 4, 9, 32, 0, 0, time.UTC), columns.NewColumn("timestamp_ntz", typing.TimestampNTZ))
+			assert.NoError(t, err)
+			assert.Equal(t, time.Date(2021, time.January, 4, 9, 32, 0, 0, time.UTC), val.(time.Time))
+		}
+	}
 	{
 		val, err := parseValue("string value", columns.NewColumn("foo", typing.String))
 		assert.NoError(t, err)

clients/redshift/dialect/typing.go

@@ -47,12 +47,12 @@ func (RedshiftDialect) DataTypeForKind(kd typing.KindDetails, _ bool) string {
 		return "BOOLEAN NULL"
 	case typing.Date.Kind:
 		return "DATE"
+	case typing.TimestampNTZ.Kind:
+		return "TIMESTAMP WITHOUT TIME ZONE"
 	case typing.ETime.Kind:
 		switch kd.ExtendedTimeDetails.Type {
 		case ext.TimestampTZKindType:
 			return "timestamp with time zone"
-		case ext.TimestampNTZKindType:
-			return "timestamp without time zone"
 		case ext.TimeKindType:
 			return "time"
 		}
@@ -106,7 +106,7 @@ func (RedshiftDialect) KindForDataType(rawType string, stringPrecision string) (
 	case "double precision":
 		return typing.Float, nil
 	case "timestamp", "timestamp without time zone":
-		return typing.NewExtendedTimeDetails(typing.ETime, ext.TimestampNTZKindType, "")
+		return typing.TimestampNTZ, nil
 	case "timestamp with time zone":
 		return typing.NewExtendedTimeDetails(typing.ETime, ext.TimestampTZKindType, "")
 	case "time without time zone":

clients/redshift/dialect/typing_test.go

@@ -51,7 +51,7 @@ func TestRedshiftDialect_DataTypeForKind(t *testing.T) {
 		}
 		{
 			// Without timezone
-			assert.Equal(t, "timestamp without time zone", RedshiftDialect{}.DataTypeForKind(typing.MustNewExtendedTimeDetails(typing.ETime, ext.TimestampNTZKindType, ""), false))
+			assert.Equal(t, "TIMESTAMP WITHOUT TIME ZONE", RedshiftDialect{}.DataTypeForKind(typing.TimestampNTZ, false))
 		}
 	}
 }
@@ -140,8 +140,7 @@ func TestRedshiftDialect_KindForDataType(t *testing.T) {
 		{
 			kd, err := dialect.KindForDataType("timestamp without time zone", "")
 			assert.NoError(t, err)
-			assert.Equal(t, typing.ETime.Kind, kd.Kind)
-			assert.Equal(t, ext.TimestampNTZKindType, kd.ExtendedTimeDetails.Type)
+			assert.Equal(t, typing.TimestampNTZ, kd)
 		}
 		{
 			kd, err := dialect.KindForDataType("time without time zone", "")

clients/shared/default_value.go

@@ -28,6 +28,13 @@ func DefaultValue(column columns.Column, dialect sql.Dialect) (any, error) {
 		}
 
 		return sql.QuoteLiteral(_time.Format(ext.PostgresDateFormat)), nil
+	case typing.TimestampNTZ.Kind:
+		_time, err := ext.ParseTimestampNTZFromInterface(column.DefaultValue())
+		if err != nil {
+			return "", fmt.Errorf("failed to cast colVal as time.Time, colVal: '%v', err: %w", column.DefaultValue(), err)
+		}
+
+		return sql.QuoteLiteral(_time.Format(ext.RFC3339NoTZ)), nil
 	case typing.ETime.Kind:
 		if err := column.KindDetails.EnsureExtendedTimeDetails(); err != nil {
 			return nil, err

clients/shared/default_value_test.go

@@ -82,6 +82,11 @@ func TestColumn_DefaultValue(t *testing.T) {
 			col:           columns.NewColumnWithDefaultValue("", typing.Date, birthdayDateTime),
 			expectedValue: "'2022-09-06'",
 		},
+		{
+			name:          "timestamp_ntz",
+			col:           columns.NewColumnWithDefaultValue("", typing.TimestampNTZ, birthdayDateTime),
+			expectedValue: "'2022-09-06T03:19:24.942'",
+		},
 		{
 			name:          "time",
 			col:           columns.NewColumnWithDefaultValue("", timeKind, birthdayDateTime),

clients/snowflake/dialect/dialect.go

@@ -32,12 +32,12 @@ func (SnowflakeDialect) DataTypeForKind(kindDetails typing.KindDetails, _ bool)
 		return "boolean"
 	case typing.Date.Kind:
 		return "date"
+	case typing.TimestampNTZ.Kind:
+		return "timestamp_ntz"
 	case typing.ETime.Kind:
 		switch kindDetails.ExtendedTimeDetails.Type {
 		case ext.TimestampTZKindType:
 			return "timestamp_tz"
-		case ext.TimestampNTZKindType:
-			return "timestamp_ntz"
 		case ext.TimeKindType:
 			return "time"
 		}
@@ -100,7 +100,7 @@ func (SnowflakeDialect) KindForDataType(snowflakeType string, _ string) (typing.
 	case "timestamp_ltz", "timestamp_tz":
 		return typing.NewExtendedTimeDetails(typing.ETime, ext.TimestampTZKindType, "")
 	case "timestamp", "datetime", "timestamp_ntz":
-		return typing.NewExtendedTimeDetails(typing.ETime, ext.TimestampNTZKindType, "")
+		return typing.TimestampNTZ, nil
 	case "time":
 		return typing.NewExtendedTimeDetails(typing.ETime, ext.TimeKindType, "")
 	case "date":

clients/snowflake/dialect/dialect_test.go

@@ -189,7 +189,7 @@ func TestSnowflakeDialect_KindForDataType_DateTime(t *testing.T) {
 		for _, expectedDateTime := range expectedDateTimes {
 			kd, err := SnowflakeDialect{}.KindForDataType(expectedDateTime, "")
 			assert.NoError(t, err)
-			assert.Equal(t, ext.TimestampNTZKindType, kd.ExtendedTimeDetails.Type, expectedDateTime)
+			assert.Equal(t, typing.TimestampNTZ, kd, expectedDateTime)
 		}
 	}
 }

lib/cdc/relational/debezium_test.go

@@ -206,10 +206,7 @@ func (r *RelationTestSuite) TestPostgresEventWithSchemaAndTimestampNoTZ() {
 	// Datetime without TZ is emitted in microseconds which is 1000x larger than nanoseconds.
 	assert.Equal(
 		r.T(),
-		ext.NewExtendedTime(
-			time.Date(2023, time.February, 2, 17, 51, 35, 175445*1000, time.UTC),
-			ext.TimestampNTZKindType, ext.RFC3339MicrosecondNoTZ,
-		),
+		time.Date(2023, time.February, 2, 17, 51, 35, 175445*1000, time.UTC),
 		evtData["ts_no_tz1"],
 	)
 	assert.Equal(r.T(), time.Date(2023, time.February, 2, 17, 54, 11, 451000000, time.UTC), evt.GetExecutionTime())