diff --git a/clients/bigquery/dialect/dialect.go b/clients/bigquery/dialect/dialect.go
index ee84f179b..d2f19dd8a 100644
--- a/clients/bigquery/dialect/dialect.go
+++ b/clients/bigquery/dialect/dialect.go
@@ -53,6 +53,8 @@ func (BigQueryDialect) DataTypeForKind(kindDetails typing.KindDetails, _ bool) s
 			// 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.DateKindType:
 			return "date"
 		case ext.TimeKindType:
@@ -103,8 +105,10 @@ func (BigQueryDialect) KindForDataType(rawBqType string, _ string) (typing.KindD
 		return typing.Struct, nil
 	case "array":
 		return typing.Array, nil
-	case "datetime", "timestamp":
+	case "timestamp":
 		return typing.NewKindDetailsFromTemplate(typing.ETime, ext.TimestampTzKindType), nil
+	case "datetime":
+		return typing.NewKindDetailsFromTemplate(typing.ETime, ext.TimestampNTZKindType), nil
 	case "time":
 		return typing.NewKindDetailsFromTemplate(typing.ETime, ext.TimeKindType), nil
 	case "date":
diff --git a/clients/databricks/dialect/typing.go b/clients/databricks/dialect/typing.go
index 6e4ba82ee..c3d4bfcef 100644
--- a/clients/databricks/dialect/typing.go
+++ b/clients/databricks/dialect/typing.go
@@ -27,8 +27,11 @@ func (DatabricksDialect) DataTypeForKind(kindDetails typing.KindDetails, _ bool)
 	case typing.ETime.Kind:
 		switch kindDetails.ExtendedTimeDetails.Type {
 		case ext.TimestampTzKindType:
-			// Using datetime2 because it's the recommendation, and it provides more precision: https://stackoverflow.com/a/1884088
 			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.DateKindType:
 			return "DATE"
 		case ext.TimeKindType:
@@ -73,7 +76,7 @@ func (DatabricksDialect) KindForDataType(rawType string, _ string) (typing.KindD
 	case "timestamp":
 		return typing.NewKindDetailsFromTemplate(typing.ETime, ext.TimestampTzKindType), nil
 	case "timestamp_ntz":
-		return typing.NewKindDetailsFromTemplate(typing.ETime, ext.TimestampTzKindType), nil
+		return typing.NewKindDetailsFromTemplate(typing.ETime, ext.TimestampNTZKindType), nil
 	}
 
 	return typing.Invalid, fmt.Errorf("unsupported data type: %q", rawType)
diff --git a/clients/databricks/dialect/typing_test.go b/clients/databricks/dialect/typing_test.go
index ddb0e7ad9..a5e457e4b 100644
--- a/clients/databricks/dialect/typing_test.go
+++ b/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.NewKindDetailsFromTemplate(typing.ETime, ext.TimestampTzKindType), kd)
+		assert.Equal(t, typing.NewKindDetailsFromTemplate(typing.ETime, ext.TimestampNTZKindType), kd)
 	}
 	{
 		// Variant
diff --git a/clients/mssql/dialect/dialect.go b/clients/mssql/dialect/dialect.go
index dc63c049c..ac7b8207d 100644
--- a/clients/mssql/dialect/dialect.go
+++ b/clients/mssql/dialect/dialect.go
@@ -56,6 +56,8 @@ func (MSSQLDialect) DataTypeForKind(kindDetails typing.KindDetails, isPk bool) s
 	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.DateKindType:
@@ -114,6 +116,8 @@ func (MSSQLDialect) KindForDataType(rawType string, stringPrecision string) (typ
 	case
 		"datetime",
 		"datetime2":
+		return typing.NewKindDetailsFromTemplate(typing.ETime, ext.TimestampNTZKindType), nil
+	case "datetimeoffset":
 		return typing.NewKindDetailsFromTemplate(typing.ETime, ext.TimestampTzKindType), nil
 	case "time":
 		return typing.NewKindDetailsFromTemplate(typing.ETime, ext.TimeKindType), nil
diff --git a/clients/redshift/dialect/typing.go b/clients/redshift/dialect/typing.go
index 3394fd2c0..1ba94caf5 100644
--- a/clients/redshift/dialect/typing.go
+++ b/clients/redshift/dialect/typing.go
@@ -49,6 +49,8 @@ func (RedshiftDialect) DataTypeForKind(kd typing.KindDetails, _ bool) string {
 		switch kd.ExtendedTimeDetails.Type {
 		case ext.TimestampTzKindType:
 			return "timestamp with time zone"
+		case ext.TimestampNTZKindType:
+			return "timestamp without time zone"
 		case ext.DateKindType:
 			return "date"
 		case ext.TimeKindType:
@@ -103,7 +105,9 @@ func (RedshiftDialect) KindForDataType(rawType string, stringPrecision string) (
 		}, nil
 	case "double precision":
 		return typing.Float, nil
-	case "timestamp with time zone", "timestamp without time zone":
+	case "timestamp", "timestamp without time zone":
+		return typing.NewKindDetailsFromTemplate(typing.ETime, ext.TimestampNTZKindType), nil
+	case "timestamp with time zone":
 		return typing.NewKindDetailsFromTemplate(typing.ETime, ext.TimestampTzKindType), nil
 	case "time without time zone":
 		return typing.NewKindDetailsFromTemplate(typing.ETime, ext.TimeKindType), nil
diff --git a/clients/redshift/dialect/typing_test.go b/clients/redshift/dialect/typing_test.go
index 784eb76cc..6cc925195 100644
--- a/clients/redshift/dialect/typing_test.go
+++ b/clients/redshift/dialect/typing_test.go
@@ -43,6 +43,17 @@ func TestRedshiftDialect_DataTypeForKind(t *testing.T) {
 			}
 		}
 	}
+	{
+		// Timestamps
+		{
+			// With timezone
+			assert.Equal(t, "timestamp with time zone", RedshiftDialect{}.DataTypeForKind(typing.NewKindDetailsFromTemplate(typing.ETime, ext.TimestampTzKindType), false))
+		}
+		{
+			// Without timezone
+			assert.Equal(t, "timestamp without time zone", RedshiftDialect{}.DataTypeForKind(typing.NewKindDetailsFromTemplate(typing.ETime, ext.TimestampNTZKindType), false))
+		}
+	}
 }
 
 func TestRedshiftDialect_KindForDataType(t *testing.T) {
@@ -130,7 +141,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.TimestampTzKindType, kd.ExtendedTimeDetails.Type)
+			assert.Equal(t, ext.TimestampNTZKindType, kd.ExtendedTimeDetails.Type)
 		}
 		{
 			kd, err := dialect.KindForDataType("time without time zone", "")
diff --git a/clients/snowflake/dialect/dialect.go b/clients/snowflake/dialect/dialect.go
index 92812dc0f..3e81cd95d 100644
--- a/clients/snowflake/dialect/dialect.go
+++ b/clients/snowflake/dialect/dialect.go
@@ -33,11 +33,9 @@ func (SnowflakeDialect) DataTypeForKind(kindDetails typing.KindDetails, _ bool)
 	case typing.ETime.Kind:
 		switch kindDetails.ExtendedTimeDetails.Type {
 		case ext.TimestampTzKindType:
-			// We are not using `TIMESTAMP_NTZ` because Snowflake does not join on this data very well.
-			// It ends up trying to parse this data into a TIMESTAMP_TZ and messes with the join order.
-			// Specifically, if my location is in SF, it'll try to parse TIMESTAMP_NTZ into PST then into UTC.
-			// When it was already stored as UTC.
 			return "timestamp_tz"
+		case ext.TimestampNTZKindType:
+			return "timestamp_ntz"
 		case ext.DateKindType:
 			return "date"
 		case ext.TimeKindType:
@@ -99,8 +97,10 @@ func (SnowflakeDialect) KindForDataType(snowflakeType string, _ string) (typing.
 		return typing.Struct, nil
 	case "array":
 		return typing.Array, nil
-	case "datetime", "timestamp", "timestamp_ltz", "timestamp_ntz", "timestamp_tz":
+	case "timestamp_ltz", "timestamp_tz":
 		return typing.NewKindDetailsFromTemplate(typing.ETime, ext.TimestampTzKindType), nil
+	case "timestamp", "datetime", "timestamp_ntz":
+		return typing.NewKindDetailsFromTemplate(typing.ETime, ext.TimestampNTZKindType), nil
 	case "time":
 		return typing.NewKindDetailsFromTemplate(typing.ETime, ext.TimeKindType), nil
 	case "date":
diff --git a/clients/snowflake/dialect/dialect_test.go b/clients/snowflake/dialect/dialect_test.go
index f2f78ecf2..3c5430654 100644
--- a/clients/snowflake/dialect/dialect_test.go
+++ b/clients/snowflake/dialect/dialect_test.go
@@ -174,11 +174,23 @@ func TestSnowflakeDialect_KindForDataType(t *testing.T) {
 }
 
 func TestSnowflakeDialect_KindForDataType_DateTime(t *testing.T) {
-	expectedDateTimes := []string{"DATETIME", "TIMESTAMP", "TIMESTAMP_LTZ", "TIMESTAMP_NTZ(9)", "TIMESTAMP_TZ"}
-	for _, expectedDateTime := range expectedDateTimes {
-		kd, err := SnowflakeDialect{}.KindForDataType(expectedDateTime, "")
-		assert.NoError(t, err)
-		assert.Equal(t, ext.TimestampTz.Type, kd.ExtendedTimeDetails.Type, expectedDateTime)
+	{
+		// Timestamp with time zone
+		expectedDateTimes := []string{"TIMESTAMP_LTZ", "TIMESTAMP_TZ"}
+		for _, expectedDateTime := range expectedDateTimes {
+			kd, err := SnowflakeDialect{}.KindForDataType(expectedDateTime, "")
+			assert.NoError(t, err)
+			assert.Equal(t, ext.TimestampTz.Type, kd.ExtendedTimeDetails.Type, expectedDateTime)
+		}
+	}
+	{
+		// Timestamp without time zone
+		expectedDateTimes := []string{"TIMESTAMP", "DATETIME", "TIMESTAMP_NTZ(9)"}
+		for _, expectedDateTime := range expectedDateTimes {
+			kd, err := SnowflakeDialect{}.KindForDataType(expectedDateTime, "")
+			assert.NoError(t, err)
+			assert.Equal(t, ext.TimestampNTZ.Type, kd.ExtendedTimeDetails.Type, expectedDateTime)
+		}
 	}
 }
 
diff --git a/lib/cdc/relational/debezium_test.go b/lib/cdc/relational/debezium_test.go
index ab09a3d16..bbfc3ec67 100644
--- a/lib/cdc/relational/debezium_test.go
+++ b/lib/cdc/relational/debezium_test.go
@@ -208,7 +208,7 @@ func (r *RelationTestSuite) TestPostgresEventWithSchemaAndTimestampNoTZ() {
 		r.T(),
 		ext.NewExtendedTime(
 			time.Date(2023, time.February, 2, 17, 51, 35, 175445*1000, time.UTC),
-			ext.TimestampTzKindType, ext.RFC3339Microsecond,
+			ext.TimestampNTZKindType, ext.RFC3339MicrosecondNoTZ,
 		),
 		evtData["ts_no_tz1"],
 	)
diff --git a/lib/debezium/converters/timestamp.go b/lib/debezium/converters/timestamp.go
index a5660ae2e..5caf18750 100644
--- a/lib/debezium/converters/timestamp.go
+++ b/lib/debezium/converters/timestamp.go
@@ -9,48 +9,60 @@ import (
 
 type Timestamp struct{}
 
-func (Timestamp) ToKindDetails() typing.KindDetails {
-	return typing.NewKindDetailsFromTemplate(typing.ETime, ext.TimestampTzKindType)
+func (Timestamp) layout() string {
+	return ext.RFC3339MillisecondNoTZ
 }
 
-func (Timestamp) Convert(value any) (any, error) {
+func (t Timestamp) ToKindDetails() typing.KindDetails {
+	return typing.NewExtendedTimeDetails(typing.ETime, ext.TimestampNTZKindType, t.layout())
+}
+
+func (t Timestamp) Convert(value any) (any, error) {
 	castedValue, err := typing.AssertType[int64](value)
 	if err != nil {
 		return nil, err
 	}
 
 	// Represents the number of milliseconds since the epoch, and does not include timezone information.
-	return ext.NewExtendedTime(time.UnixMilli(castedValue).In(time.UTC), ext.TimestampTzKindType, ext.RFC3339Millisecond), nil
+	return ext.NewExtendedTime(time.UnixMilli(castedValue).In(time.UTC), ext.TimestampNTZKindType, t.layout()), nil
 }
 
 type MicroTimestamp struct{}
 
-func (MicroTimestamp) ToKindDetails() typing.KindDetails {
-	return typing.NewKindDetailsFromTemplate(typing.ETime, ext.TimestampTzKindType)
+func (MicroTimestamp) layout() string {
+	return ext.RFC3339MicrosecondNoTZ
+}
+
+func (mt MicroTimestamp) ToKindDetails() typing.KindDetails {
+	return typing.NewExtendedTimeDetails(typing.ETime, ext.TimestampNTZKindType, mt.layout())
 }
 
-func (MicroTimestamp) Convert(value any) (any, error) {
+func (mt MicroTimestamp) Convert(value any) (any, error) {
 	castedValue, err := typing.AssertType[int64](value)
 	if err != nil {
 		return nil, err
 	}
 
 	// Represents the number of microseconds since the epoch, and does not include timezone information.
-	return ext.NewExtendedTime(time.UnixMicro(castedValue).In(time.UTC), ext.TimestampTzKindType, ext.RFC3339Microsecond), nil
+	return ext.NewExtendedTime(time.UnixMicro(castedValue).In(time.UTC), ext.TimestampNTZKindType, mt.layout()), nil
 }
 
 type NanoTimestamp struct{}
 
-func (NanoTimestamp) ToKindDetails() typing.KindDetails {
-	return typing.NewKindDetailsFromTemplate(typing.ETime, ext.TimestampTzKindType)
+func (nt NanoTimestamp) ToKindDetails() typing.KindDetails {
+	return typing.NewExtendedTimeDetails(typing.ETime, ext.TimestampNTZKindType, nt.layout())
+}
+
+func (NanoTimestamp) layout() string {
+	return ext.RFC3339NanosecondNoTZ
 }
 
-func (NanoTimestamp) Convert(value any) (any, error) {
+func (nt NanoTimestamp) Convert(value any) (any, error) {
 	castedValue, err := typing.AssertType[int64](value)
 	if err != nil {
 		return nil, err
 	}
 
 	// Represents the number of nanoseconds since the epoch, and does not include timezone information.
-	return ext.NewExtendedTime(time.UnixMicro(castedValue/1_000).In(time.UTC), ext.TimestampTzKindType, ext.RFC3339Nanosecond), nil
+	return ext.NewExtendedTime(time.UnixMicro(castedValue/1_000).In(time.UTC), ext.TimestampNTZKindType, nt.layout()), nil
 }
diff --git a/lib/debezium/converters/timestamp_test.go b/lib/debezium/converters/timestamp_test.go
index b54f63e87..cf191929d 100644
--- a/lib/debezium/converters/timestamp_test.go
+++ b/lib/debezium/converters/timestamp_test.go
@@ -9,7 +9,7 @@ import (
 )
 
 func TestTimestamp_Converter(t *testing.T) {
-	assert.Equal(t, typing.NewKindDetailsFromTemplate(typing.ETime, ext.TimestampTzKindType), Timestamp{}.ToKindDetails())
+	assert.Equal(t, typing.NewExtendedTimeDetails(typing.ETime, ext.TimestampNTZKindType, ext.RFC3339MillisecondNoTZ), Timestamp{}.ToKindDetails())
 	{
 		// Invalid conversion
 		_, err := Timestamp{}.Convert("invalid")
@@ -19,18 +19,18 @@ func TestTimestamp_Converter(t *testing.T) {
 		// Valid conversion
 		converted, err := Timestamp{}.Convert(int64(1_725_058_799_089))
 		assert.NoError(t, err)
-		assert.Equal(t, "2024-08-30T22:59:59.089Z", converted.(*ext.ExtendedTime).String(""))
+		assert.Equal(t, "2024-08-30T22:59:59.089", converted.(*ext.ExtendedTime).String(""))
 	}
 	{
 		// ms is preserved despite it being all zeroes.
 		converted, err := Timestamp{}.Convert(int64(1_725_058_799_000))
 		assert.NoError(t, err)
-		assert.Equal(t, "2024-08-30T22:59:59.000Z", converted.(*ext.ExtendedTime).String(""))
+		assert.Equal(t, "2024-08-30T22:59:59.000", converted.(*ext.ExtendedTime).String(""))
 	}
 }
 
 func TestMicroTimestamp_Converter(t *testing.T) {
-	assert.Equal(t, typing.NewKindDetailsFromTemplate(typing.ETime, ext.TimestampTzKindType), MicroTimestamp{}.ToKindDetails())
+	assert.Equal(t, typing.NewExtendedTimeDetails(typing.ETime, ext.TimestampNTZKindType, ext.RFC3339MicrosecondNoTZ), MicroTimestamp{}.ToKindDetails())
 	{
 		// Invalid conversion
 		_, err := MicroTimestamp{}.Convert("invalid")
@@ -40,18 +40,18 @@ func TestMicroTimestamp_Converter(t *testing.T) {
 		// Valid conversion
 		converted, err := MicroTimestamp{}.Convert(int64(1_712_609_795_827_923))
 		assert.NoError(t, err)
-		assert.Equal(t, "2024-04-08T20:56:35.827923Z", converted.(*ext.ExtendedTime).String(""))
+		assert.Equal(t, "2024-04-08T20:56:35.827923", converted.(*ext.ExtendedTime).String(""))
 	}
 	{
 		// micros is preserved despite it being all zeroes.
 		converted, err := MicroTimestamp{}.Convert(int64(1_712_609_795_820_000))
 		assert.NoError(t, err)
-		assert.Equal(t, "2024-04-08T20:56:35.820000Z", converted.(*ext.ExtendedTime).String(""))
+		assert.Equal(t, "2024-04-08T20:56:35.820000", converted.(*ext.ExtendedTime).String(""))
 	}
 }
 
 func TestNanoTimestamp_Converter(t *testing.T) {
-	assert.Equal(t, typing.NewKindDetailsFromTemplate(typing.ETime, ext.TimestampTzKindType), NanoTimestamp{}.ToKindDetails())
+	assert.Equal(t, typing.NewExtendedTimeDetails(typing.ETime, ext.TimestampNTZKindType, ext.RFC3339NanosecondNoTZ), NanoTimestamp{}.ToKindDetails())
 	{
 		// Invalid conversion
 		_, err := NanoTimestamp{}.Convert("invalid")
@@ -61,12 +61,12 @@ func TestNanoTimestamp_Converter(t *testing.T) {
 		// Valid conversion
 		converted, err := NanoTimestamp{}.Convert(int64(1_712_609_795_827_001_000))
 		assert.NoError(t, err)
-		assert.Equal(t, "2024-04-08T20:56:35.827001000Z", converted.(*ext.ExtendedTime).String(""))
+		assert.Equal(t, "2024-04-08T20:56:35.827001000", converted.(*ext.ExtendedTime).String(""))
 	}
 	{
 		// nanos is preserved despite it being all zeroes.
 		converted, err := NanoTimestamp{}.Convert(int64(1_712_609_795_827_000_000))
 		assert.NoError(t, err)
-		assert.Equal(t, "2024-04-08T20:56:35.827000000Z", converted.(*ext.ExtendedTime).String(""))
+		assert.Equal(t, "2024-04-08T20:56:35.827000000", converted.(*ext.ExtendedTime).String(""))
 	}
 }
diff --git a/lib/debezium/schema_test.go b/lib/debezium/schema_test.go
index 25be4285a..1e74371b4 100644
--- a/lib/debezium/schema_test.go
+++ b/lib/debezium/schema_test.go
@@ -229,14 +229,22 @@ func TestField_ToKindDetails(t *testing.T) {
 		assert.Equal(t, typing.Invalid, kd)
 	}
 	{
-		// Timestamp
-		// Datetime (for now)
-		for _, dbzType := range []SupportedDebeziumType{Timestamp, TimestampKafkaConnect, MicroTimestamp, NanoTimestamp, ZonedTimestamp} {
+		// Timestamp with timezone
+		for _, dbzType := range []SupportedDebeziumType{ZonedTimestamp} {
 			kd, err := Field{DebeziumType: dbzType}.ToKindDetails()
 			assert.NoError(t, err)
 			assert.Equal(t, typing.NewKindDetailsFromTemplate(typing.ETime, ext.TimestampTzKindType), kd)
 		}
 	}
+	{
+		// Timestamp without timezone
+		for _, dbzType := range []SupportedDebeziumType{Timestamp, TimestampKafkaConnect, MicroTimestamp, NanoTimestamp} {
+			kd, err := Field{DebeziumType: dbzType}.ToKindDetails()
+			assert.NoError(t, err)
+			assert.Equal(t, ext.TimestampNTZKindType, kd.ExtendedTimeDetails.Type)
+			assert.Equal(t, typing.ETime.Kind, kd.Kind)
+		}
+	}
 	{
 		// Dates
 		for _, dbzType := range []SupportedDebeziumType{Date, DateKafkaConnect} {
diff --git a/lib/debezium/types_test.go b/lib/debezium/types_test.go
index b8093b81b..3ab8ccc42 100644
--- a/lib/debezium/types_test.go
+++ b/lib/debezium/types_test.go
@@ -302,7 +302,7 @@ func TestField_ParseValue(t *testing.T) {
 				field := Field{Type: Int64, DebeziumType: dbzType}
 				value, err := field.ParseValue(int64(1_725_058_799_000))
 				assert.NoError(t, err)
-				assert.Equal(t, "2024-08-30T22:59:59.000Z", value.(*ext.ExtendedTime).String(""))
+				assert.Equal(t, "2024-08-30T22:59:59.000", value.(*ext.ExtendedTime).String(""))
 			}
 		}
 		{
@@ -310,7 +310,7 @@ func TestField_ParseValue(t *testing.T) {
 			field := Field{Type: Int64, DebeziumType: NanoTimestamp}
 			val, err := field.ParseValue(int64(1_712_609_795_827_000_000))
 			assert.NoError(t, err)
-			assert.Equal(t, ext.NewExtendedTime(time.Date(2024, time.April, 8, 20, 56, 35, 827000000, time.UTC), ext.TimestampTzKindType, "2006-01-02T15:04:05.000000000Z07:00"), val.(*ext.ExtendedTime))
+			assert.Equal(t, ext.NewExtendedTime(time.Date(2024, time.April, 8, 20, 56, 35, 827000000, time.UTC), ext.TimestampNTZKindType, "2006-01-02T15:04:05.000000000"), val.(*ext.ExtendedTime))
 		}
 		{
 			// Micro timestamp
@@ -319,13 +319,13 @@ func TestField_ParseValue(t *testing.T) {
 				// Int64
 				val, err := field.ParseValue(int64(1_712_609_795_827_009))
 				assert.NoError(t, err)
-				assert.Equal(t, ext.NewExtendedTime(time.Date(2024, time.April, 8, 20, 56, 35, 827009000, time.UTC), ext.TimestampTzKindType, ext.RFC3339Microsecond), val.(*ext.ExtendedTime))
+				assert.Equal(t, ext.NewExtendedTime(time.Date(2024, time.April, 8, 20, 56, 35, 827009000, time.UTC), ext.TimestampNTZKindType, ext.RFC3339MicrosecondNoTZ), val.(*ext.ExtendedTime))
 			}
 			{
 				// Float64
 				val, err := field.ParseValue(float64(1_712_609_795_827_001))
 				assert.NoError(t, err)
-				assert.Equal(t, ext.NewExtendedTime(time.Date(2024, time.April, 8, 20, 56, 35, 827001000, time.UTC), ext.TimestampTzKindType, ext.RFC3339Microsecond), val.(*ext.ExtendedTime))
+				assert.Equal(t, ext.NewExtendedTime(time.Date(2024, time.April, 8, 20, 56, 35, 827001000, time.UTC), ext.TimestampNTZKindType, ext.RFC3339MicrosecondNoTZ), val.(*ext.ExtendedTime))
 			}
 			{
 				// Invalid (string)
diff --git a/lib/optimization/table_data.go b/lib/optimization/table_data.go
index 0c1da5af4..9a5681c89 100644
--- a/lib/optimization/table_data.go
+++ b/lib/optimization/table_data.go
@@ -293,8 +293,16 @@ func (t *TableData) MergeColumnsFromDestination(destCols ...columns.Column) erro
 					inMemoryCol.KindDetails.ExtendedTimeDetails = &ext.NestedKind{}
 				}
 
+				// If the column in the destination is a timestamp_tz and the in-memory column is a timestamp_ntz, we should update the layout to contain timezone locale.
+				if foundColumn.KindDetails.ExtendedTimeDetails.Type == ext.TimestampTzKindType && inMemoryCol.KindDetails.ExtendedTimeDetails.Type == ext.TimestampNTZKindType {
+					if inMemoryCol.KindDetails.ExtendedTimeDetails.Format != "" {
+						inMemoryCol.KindDetails.ExtendedTimeDetails.Format += ext.TimezoneOffsetFormat
+					}
+				}
+
 				// Just copy over the type since the format wouldn't be present in the destination
 				inMemoryCol.KindDetails.ExtendedTimeDetails.Type = foundColumn.KindDetails.ExtendedTimeDetails.Type
+
 			}
 
 			// Copy over the decimal details
diff --git a/lib/optimization/table_data_merge_columns_test.go b/lib/optimization/table_data_merge_columns_test.go
index 3a1791922..8051213b7 100644
--- a/lib/optimization/table_data_merge_columns_test.go
+++ b/lib/optimization/table_data_merge_columns_test.go
@@ -10,6 +10,33 @@ import (
 	"github.com/stretchr/testify/assert"
 )
 
+func TestTableData_UpdateInMemoryColumnsFromDestination_Tz(t *testing.T) {
+	{
+		// In memory and destination columns are both timestamp_tz
+		tableData := &TableData{inMemoryColumns: &columns.Columns{}}
+		tableData.AddInMemoryCol(columns.NewColumn("foo", typing.NewKindDetailsFromTemplate(typing.ETime, ext.TimestampTzKindType)))
+
+		assert.NoError(t, tableData.MergeColumnsFromDestination(columns.NewColumn("foo", typing.NewKindDetailsFromTemplate(typing.ETime, ext.TimestampTzKindType))))
+		updatedColumn, isOk := tableData.inMemoryColumns.GetColumn("foo")
+		assert.True(t, isOk)
+		assert.Equal(t, ext.TimestampTzKindType, updatedColumn.KindDetails.ExtendedTimeDetails.Type)
+	}
+	{
+		// In memory is timestamp_ntz and destination is timestamp_tz
+		tableData := &TableData{inMemoryColumns: &columns.Columns{}}
+		kd := typing.NewKindDetailsFromTemplate(typing.ETime, ext.TimestampNTZKindType)
+		kd.ExtendedTimeDetails.Format = ext.RFC3339MillisecondNoTZ
+		tableData.AddInMemoryCol(columns.NewColumn("foo", kd))
+
+		assert.NoError(t, tableData.MergeColumnsFromDestination(columns.NewColumn("foo", typing.NewKindDetailsFromTemplate(typing.ETime, ext.TimestampTzKindType))))
+		updatedColumn, isOk := tableData.inMemoryColumns.GetColumn("foo")
+		assert.True(t, isOk)
+		assert.Equal(t, ext.TimestampTzKindType, updatedColumn.KindDetails.ExtendedTimeDetails.Type)
+		assert.Equal(t, ext.RFC3339Millisecond, updatedColumn.KindDetails.ExtendedTimeDetails.Format)
+	}
+
+}
+
 func TestTableData_UpdateInMemoryColumnsFromDestination(t *testing.T) {
 	const strCol = "string"
 	tableDataCols := &columns.Columns{}
diff --git a/lib/typing/ext/time.go b/lib/typing/ext/time.go
index 2b05fc462..ab5499bc2 100644
--- a/lib/typing/ext/time.go
+++ b/lib/typing/ext/time.go
@@ -1,6 +1,7 @@
 package ext
 
 import (
+	"cmp"
 	"encoding/json"
 	"time"
 )
@@ -10,9 +11,10 @@ import (
 type ExtendedTimeKindType string
 
 const (
-	TimestampTzKindType ExtendedTimeKindType = "timestamp_tz"
-	DateKindType        ExtendedTimeKindType = "date"
-	TimeKindType        ExtendedTimeKindType = "time"
+	TimestampTzKindType  ExtendedTimeKindType = "timestamp_tz"
+	TimestampNTZKindType ExtendedTimeKindType = "timestamp_ntz"
+	DateKindType         ExtendedTimeKindType = "date"
+	TimeKindType         ExtendedTimeKindType = "time"
 )
 
 type NestedKind struct {
@@ -21,6 +23,11 @@ type NestedKind struct {
 }
 
 var (
+	TimestampNTZ = NestedKind{
+		Type:   TimestampNTZKindType,
+		Format: RFC3339NanosecondNoTZ,
+	}
+
 	TimestampTz = NestedKind{
 		Type:   TimestampTzKindType,
 		Format: time.RFC3339Nano,
@@ -53,6 +60,8 @@ func NewExtendedTime(t time.Time, kindType ExtendedTimeKindType, originalFormat
 		switch kindType {
 		case TimestampTzKindType:
 			originalFormat = TimestampTz.Format
+		case TimestampNTZKindType:
+			originalFormat = TimestampNTZ.Format
 		case DateKindType:
 			originalFormat = Date.Format
 		case TimeKindType:
@@ -78,9 +87,6 @@ func (e *ExtendedTime) GetNestedKind() NestedKind {
 }
 
 func (e *ExtendedTime) String(overrideFormat string) string {
-	if overrideFormat != "" {
-		return e.ts.Format(overrideFormat)
-	}
-
-	return e.ts.Format(e.nestedKind.Format)
+	format := cmp.Or(overrideFormat, e.nestedKind.Format)
+	return e.ts.Format(format)
 }
diff --git a/lib/typing/ext/variables.go b/lib/typing/ext/variables.go
index 8d4f413ab..defea4453 100644
--- a/lib/typing/ext/variables.go
+++ b/lib/typing/ext/variables.go
@@ -44,12 +44,17 @@ var SupportedTimeFormats = []string{
 	AdditionalTimeFormat,
 }
 
+const TimezoneOffsetFormat = "Z07:00"
+
 // RFC3339 variants
 const (
-	RFC3339MillisecondUTC = "2006-01-02T15:04:05.000Z"
-	RFC3339MicrosecondUTC = "2006-01-02T15:04:05.000000Z"
-	RFC3339NanosecondUTC  = "2006-01-02T15:04:05.000000000Z"
-	RFC3339Millisecond    = "2006-01-02T15:04:05.000Z07:00"
-	RFC3339Microsecond    = "2006-01-02T15:04:05.000000Z07:00"
-	RFC3339Nanosecond     = "2006-01-02T15:04:05.000000000Z07:00"
+	RFC3339MillisecondUTC  = "2006-01-02T15:04:05.000Z"
+	RFC3339MicrosecondUTC  = "2006-01-02T15:04:05.000000Z"
+	RFC3339NanosecondUTC   = "2006-01-02T15:04:05.000000000Z"
+	RFC3339Millisecond     = "2006-01-02T15:04:05.000" + TimezoneOffsetFormat
+	RFC3339MillisecondNoTZ = "2006-01-02T15:04:05.000"
+	RFC3339Microsecond     = "2006-01-02T15:04:05.000000" + TimezoneOffsetFormat
+	RFC3339MicrosecondNoTZ = "2006-01-02T15:04:05.000000"
+	RFC3339Nanosecond      = "2006-01-02T15:04:05.000000000" + TimezoneOffsetFormat
+	RFC3339NanosecondNoTZ  = "2006-01-02T15:04:05.000000000"
 )
diff --git a/lib/typing/typing.go b/lib/typing/typing.go
index 853d50f5b..4abe25528 100644
--- a/lib/typing/typing.go
+++ b/lib/typing/typing.go
@@ -83,6 +83,15 @@ func NewDecimalDetailsFromTemplate(details KindDetails, decimalDetails decimal.D
 	return details
 }
 
+func NewExtendedTimeDetails(details KindDetails, extendedType ext.ExtendedTimeKindType, format string) KindDetails {
+	details.ExtendedTimeDetails = &ext.NestedKind{
+		Type:   extendedType,
+		Format: format,
+	}
+
+	return details
+}
+
 func NewKindDetailsFromTemplate(details KindDetails, extendedType ext.ExtendedTimeKindType) KindDetails {
 	if details.ExtendedTimeDetails == nil {
 		details.ExtendedTimeDetails = &ext.NestedKind{}
diff --git a/lib/typing/values/string.go b/lib/typing/values/string.go
index 37b7df6e5..8c7a485cc 100644
--- a/lib/typing/values/string.go
+++ b/lib/typing/values/string.go
@@ -40,7 +40,6 @@ func ToString(colVal any, colKind typing.KindDetails) (string, error) {
 		if colKind.ExtendedTimeDetails.Type == ext.TimeKindType {
 			return extTime.String(ext.PostgresTimeFormatNoTZ), nil
 		}
-
 		return extTime.String(colKind.ExtendedTimeDetails.Format), nil
 	case typing.String.Kind:
 		isArray := reflect.ValueOf(colVal).Kind() == reflect.Slice