materialize-sql: custom CastSQL function and migration mapping spec

materialize-bigquery/.snapshots/TestValidateAndApplyMigrations

@@ -33,7 +33,7 @@ Migratable Changes Before Apply Schema:
 
 Migratable Changes Before Apply Data:
 key (STRING), _meta_flow_truncated (BOOLEAN), dateValue (DATE), datetimeValue (TIMESTAMP), flow_published_at (TIMESTAMP), int64 (INTEGER), multiple (JSON), nonScalarValue (STRING), numericString (BIGNUMERIC), optional (STRING), scalarValue (STRING), timeValue (STRING), flow_document (STRING)
-1, false, 2024-01-01, 2024-01-01 01:01:01 +0000 UTC, 2024-09-13 01:01:01 +0000 UTC, 1, <nil>, <nil>, 123/1, <nil>, test, 01:01:01, {}
+1, false, 2024-01-01, 2024-01-01 01:01:01.111111 +0000 UTC, 2024-09-13 01:01:01 +0000 UTC, 1, <nil>, <nil>, 123/1, <nil>, test, 01:01:01, {}
 
 Migratable Changes Constraints:
 {"Field":"_meta/flow_truncated","Type":3,"TypeString":"LOCATION_RECOMMENDED","Reason":"This location is part of the current materialization"}
@@ -70,5 +70,5 @@ Migratable Changes Applied Schema:
 
 Migratable Changes Applied Data:
 key (STRING), _meta_flow_truncated (BOOLEAN), flow_published_at (TIMESTAMP), int64 (INTEGER), multiple (JSON), nonScalarValue (STRING), optional (STRING), scalarValue (STRING), timeValue (STRING), flow_document (STRING), dateValue (STRING), datetimeValue (STRING), numericString (STRING)
-1, false, 2024-09-13 01:01:01 +0000 UTC, 1, <nil>, <nil>, <nil>, test, 01:01:01, {}, 2024-01-01, 2024-01-01 01:01:01+00, 123
+1, false, 2024-09-13 01:01:01 +0000 UTC, 1, <nil>, <nil>, <nil>, test, 01:01:01, {}, 2024-01-01, 2024-01-01T01:01:01.111111Z, 123
 

materialize-bigquery/bigquery_test.go

@@ -126,6 +126,15 @@ func TestValidateAndApplyMigrations(t *testing.T) {
 			values = append(values, "'2024-09-13 01:01:01'")
 			keys = append(keys, bqDialect.Identifier("flow_document"))
 			values = append(values, "'{}'")
+
+			// bigquery does not support more than 6 fractional second precision, and will fail if we try
+			// to insert a value with 9
+			for i, _ := range values {
+				if keys[i] == "datetimeValue" {
+					values[i] = "'2024-01-01 01:01:01.111111'"
+				}
+			}
+
 			_, err = client.query(ctx, fmt.Sprintf(
 				"insert into %s (%s) VALUES (%s);",
 				bqDialect.Identifier(cfg.ProjectID, cfg.Dataset, resourceConfig.Table), strings.Join(keys, ","), strings.Join(values, ","),

materialize-bigquery/sqlgen.go

@@ -98,12 +98,12 @@ var bqDialect = func() sql.Dialect {
 	)
 
 	return sql.Dialect{
-		MigratableTypes: map[string][]string{
-			"integer":    {"string"},
-			"bignumeric": {"string"},
-			"float":      {"string"},
-			"date":       {"string"},
-			"timestamp":  {"string"},
+		MigratableTypes: sql.MigrationSpecs{
+			"integer":    {sql.NewMigrationSpec([]string{"string"})},
+			"bignumeric": {sql.NewMigrationSpec([]string{"string"})},
+			"float":      {sql.NewMigrationSpec([]string{"string"})},
+			"date":       {sql.NewMigrationSpec([]string{"string"})},
+			"timestamp":  {sql.NewMigrationSpec([]string{"string"}, sql.WithCastSQL(datetimeToStringCast))},
 		},
 		TableLocatorer: sql.TableLocatorFn(func(path []string) sql.InfoTableLocation {
 			return sql.InfoTableLocation{
@@ -131,6 +131,10 @@ var bqDialect = func() sql.Dialect {
 	}
 }()
 
+func datetimeToStringCast(migration sql.ColumnTypeMigration) string {
+	return fmt.Sprintf(`FORMAT_TIMESTAMP('%%Y-%%m-%%dT%%H:%%M:%%E*SZ', %s, 'UTC') `, migration.Identifier)
+}
+
 var (
 	tplAll = sql.MustParseTemplate(bqDialect, "root", `
 {{ define "tempTableName" -}}

materialize-databricks/.snapshots/TestValidateAndApplyMigrations

@@ -34,7 +34,7 @@ Migratable Changes Before Apply Schema:
 Migratable Changes Before Apply Data:
 key (STRING), _meta/flow_truncated (BOOLEAN), dateValue (DATE), datetimeValue (TIMESTAMP), flow_published_at (TIMESTAMP), int64 (BIGINT), multiple (STRING), nonScalarValue (STRING), numericString (DECIMAL), optional (STRING), scalarValue (STRING), timeValue (STRING), flow_document (STRING)
 
-1, false, 2024-01-01 00:00:00 +0000 UTC, 2024-01-01 01:01:01 +0000 UTC, 2024-09-13 01:01:01 +0000 UTC, 1, <nil>, <nil>, 123, <nil>, test, 01:01:01, {}
+1, false, 2024-01-01 00:00:00 +0000 UTC, 2024-01-01 01:01:01.111111 +0000 UTC, 2024-09-13 01:01:01 +0000 UTC, 1, <nil>, <nil>, 123, <nil>, test, 01:01:01, {}
 
 Migratable Changes Constraints:
 {"Field":"_meta/flow_truncated","Type":3,"TypeString":"LOCATION_RECOMMENDED","Reason":"This location is part of the current materialization"}
@@ -72,5 +72,5 @@ Migratable Changes Applied Schema:
 Migratable Changes Applied Data:
 key (STRING), _meta/flow_truncated (BOOLEAN), flow_published_at (TIMESTAMP), int64 (BIGINT), multiple (STRING), nonScalarValue (STRING), optional (STRING), scalarValue (STRING), timeValue (STRING), flow_document (STRING), dateValue (STRING), datetimeValue (STRING), numericString (STRING)
 
-1, false, 2024-09-13 01:01:01 +0000 UTC, 1, <nil>, <nil>, <nil>, test, 01:01:01, {}, 2024-01-01, 2024-01-01 01:01:01, 123
+1, false, 2024-09-13 01:01:01 +0000 UTC, 1, <nil>, <nil>, <nil>, test, 01:01:01, {}, 2024-01-01, 2024-01-01T01:01:01.111111000Z, 123
 

materialize-databricks/sqlgen.go

@@ -1,6 +1,7 @@
 package main
 
 import (
+	"fmt"
 	"regexp"
 	"slices"
 	"strings"
@@ -59,13 +60,12 @@ var databricksDialect = func() sql.Dialect {
 	)
 
 	return sql.Dialect{
-		MigratableTypes: map[string][]string{
-			"decimal":   {"string"},
-			"long":      {"string"},
-			"double":    {"string"},
-			"string":    {"numeric(38,0)", "double", "long", "date", "timestamp"},
-			"timestamp": {"string"},
-			"date":      {"string"},
+		MigratableTypes: sql.MigrationSpecs{
+			"decimal":   {sql.NewMigrationSpec([]string{"string"})},
+			"long":      {sql.NewMigrationSpec([]string{"string"})},
+			"double":    {sql.NewMigrationSpec([]string{"string"})},
+			"timestamp": {sql.NewMigrationSpec([]string{"string"}, sql.WithCastSQL(datetimeToStringCast))},
+			"date":      {sql.NewMigrationSpec([]string{"string"})},
 		},
 		TableLocatorer: sql.TableLocatorFn(func(path []string) sql.InfoTableLocation {
 			return sql.InfoTableLocation{
@@ -112,6 +112,10 @@ var databricksDialect = func() sql.Dialect {
 	}
 }()
 
+func datetimeToStringCast(migration sql.ColumnTypeMigration) string {
+	return fmt.Sprintf(`date_format(from_utc_timestamp(%s, 'UTC'), "yyyy-MM-dd'T'HH:mm:ss.SSSSSSSSS'Z'")`, migration.Identifier)
+}
+
 var (
 	tplAll = sql.MustParseTemplate(databricksDialect, "root", `
 -- Templated creation of a materialized table definition and comments:

materialize-motherduck/.snapshots/TestValidateAndApplyMigrations

@@ -35,12 +35,12 @@ Migratable Changes Before Apply Schema:
 Migratable Changes Before Apply Data:
 key (VARCHAR), _meta/flow_truncated (BOOLEAN), dateValue (DATE), datetimeValue (TIMESTAMPTZ), flow_published_at (TIMESTAMPTZ), int64 (BIGINT), multiple (VARCHAR), nonScalarValue (VARCHAR), numericString (HUGEINT), optional (VARCHAR), scalarValue (VARCHAR), second_root (VARCHAR), timeValue (TIME), flow_document (VARCHAR)
 
-1, false, 2024-01-01 00:00:00 +0000 UTC, 2024-01-01 01:01:01 +0000 UTC, 2024-09-13 01:01:01 +0000 UTC, 1, <nil>, <nil>, 123, <nil>, test, {}, 1970-01-01 01:01:01 +0000 UTC, {}
+1, false, 2024-01-01 00:00:00 +0000 UTC, 2024-01-01 01:01:01.111111 +0000 UTC, 2024-09-13 01:01:01 +0000 UTC, 1, <nil>, <nil>, 123, <nil>, test, {}, 1970-01-01 01:01:01 +0000 UTC, {}
 
 Migratable Changes Constraints:
 {"Field":"_meta/flow_truncated","Type":3,"TypeString":"LOCATION_RECOMMENDED","Reason":"This location is part of the current materialization"}
 {"Field":"dateValue","Type":3,"TypeString":"LOCATION_RECOMMENDED","Reason":"This location is part of the current materialization"}
-{"Field":"datetimeValue","Type":3,"TypeString":"LOCATION_RECOMMENDED","Reason":"This location is part of the current materialization"}
+{"Field":"datetimeValue","Type":6,"TypeString":"UNSATISFIABLE","Reason":"Field 'datetimeValue' is already being materialized as endpoint type 'TIMESTAMP WITH TIME ZONE' but endpoint type 'VARCHAR' is required by its schema '{ type: [string] }'"}
 {"Field":"flow_document","Type":3,"TypeString":"LOCATION_RECOMMENDED","Reason":"This location is part of the current materialization"}
 {"Field":"flow_published_at","Type":3,"TypeString":"LOCATION_RECOMMENDED","Reason":"This location is part of the current materialization"}
 {"Field":"int64","Type":3,"TypeString":"LOCATION_RECOMMENDED","Reason":"This location is part of the current materialization"}
@@ -57,7 +57,7 @@ Migratable Changes Constraints:
 Migratable Changes Applied Schema:
 {"Name":"_meta/flow_truncated","Nullable":"NO","Type":"BOOLEAN"}
 {"Name":"dateValue","Nullable":"YES","Type":"VARCHAR"}
-{"Name":"datetimeValue","Nullable":"YES","Type":"VARCHAR"}
+{"Name":"datetimeValue","Nullable":"YES","Type":"TIMESTAMP WITH TIME ZONE"}
 {"Name":"flow_document","Nullable":"NO","Type":"JSON"}
 {"Name":"flow_published_at","Nullable":"NO","Type":"TIMESTAMP WITH TIME ZONE"}
 {"Name":"int64","Nullable":"YES","Type":"BIGINT"}
@@ -72,7 +72,7 @@ Migratable Changes Applied Schema:
 
 
 Migratable Changes Applied Data:
-key (VARCHAR), _meta/flow_truncated (BOOLEAN), flow_published_at (TIMESTAMPTZ), int64 (BIGINT), multiple (VARCHAR), nonScalarValue (VARCHAR), optional (VARCHAR), scalarValue (VARCHAR), second_root (VARCHAR), flow_document (VARCHAR), dateValue (VARCHAR), datetimeValue (VARCHAR), numericString (VARCHAR), timeValue (VARCHAR)
+key (VARCHAR), _meta/flow_truncated (BOOLEAN), datetimeValue (TIMESTAMPTZ), flow_published_at (TIMESTAMPTZ), int64 (BIGINT), multiple (VARCHAR), nonScalarValue (VARCHAR), optional (VARCHAR), scalarValue (VARCHAR), second_root (VARCHAR), flow_document (VARCHAR), dateValue (VARCHAR), numericString (VARCHAR), timeValue (VARCHAR)
 
-1, false, 2024-09-13 01:01:01 +0000 UTC, 1, <nil>, <nil>, <nil>, test, {}, {}, 2024-01-01, 2023-12-31 20:01:01-05, 123, 01:01:01
+1, false, 2024-01-01 01:01:01.111111 +0000 UTC, 2024-09-13 01:01:01 +0000 UTC, 1, <nil>, <nil>, <nil>, test, {}, {}, 2024-01-01, 123, 01:01:01
 

materialize-motherduck/sqlgen.go

@@ -44,14 +44,12 @@ var duckDialect = func() sql.Dialect {
 	)
 
 	return sql.Dialect{
-		MigratableTypes: map[string][]string{
-			"double":                   {"varchar"},
-			"bigint":                   {"varchar"},
-			"hugeint":                  {"varchar"},
-			"varchar":                  {"double", "bigint", "hugeint", "date", "timestamp with time zone", "time"},
-			"date":                     {"varchar"},
-			"timestamp with time zone": {"varchar"},
-			"time":                     {"varchar"},
+		MigratableTypes: sql.MigrationSpecs{
+			"double":  {sql.NewMigrationSpec([]string{"varchar"})},
+			"bigint":  {sql.NewMigrationSpec([]string{"varchar"})},
+			"hugeint": {sql.NewMigrationSpec([]string{"varchar"})},
+			"date":    {sql.NewMigrationSpec([]string{"varchar"})},
+			"time":    {sql.NewMigrationSpec([]string{"varchar"})},
 		},
 		TableLocatorer: sql.TableLocatorFn(func(path []string) sql.InfoTableLocation {
 			return sql.InfoTableLocation{TableSchema: path[1], TableName: path[2]}
@@ -74,6 +72,13 @@ var duckDialect = func() sql.Dialect {
 	}
 }()
 
+/* TODO: Unfortunately I have not been able to get this working, I keep getting an error about this
+   function not existing... which is strange
+
+func datetimeToStringCast(migration sql.ColumnTypeMigration) string {
+	return fmt.Sprintf(`strftime(timezone('UTC', %s), '%%Y-%%m-%%dT%%H:%%M:%%S.%%fZ')`, migration.Identifier)
+}*/
+
 type queryParams struct {
 	sql.Table
 	Files []string

materialize-mysql/.snapshots/TestValidateAndApplyMigrations

@@ -34,7 +34,7 @@ Migratable Changes Before Apply Schema:
 Migratable Changes Before Apply Data:
 key (VARCHAR), _meta/flow_truncated (TINYINT), dateValue (DATE), datetimeValue (DATETIME), flow_published_at (DATETIME), int64 (BIGINT), multiple (JSON), nonScalarValue (JSON), numericString (DECIMAL), optional (JSON), scalarValue (TEXT), timeValue (TIME), flow_document (JSON)
 
-1, 0, 2024-01-01, 2024-01-01 01:01:01.000000, 2024-09-13 01:01:01.000000, 1, <nil>, <nil>, 123, <nil>, test, 01:01:01.000000, {}
+1, 0, 2024-01-01, 2024-01-01 01:01:01.111111, 2024-09-13 01:01:01.000000, 1, <nil>, <nil>, 123, <nil>, test, 01:01:01.000000, {}
 
 Migratable Changes Constraints:
 {"Field":"_meta/flow_truncated","Type":3,"TypeString":"LOCATION_RECOMMENDED","Reason":"This location is part of the current materialization"}
@@ -72,5 +72,5 @@ Migratable Changes Applied Schema:
 Migratable Changes Applied Data:
 key (VARCHAR), _meta/flow_truncated (TINYINT), flow_published_at (DATETIME), int64 (BIGINT), multiple (JSON), nonScalarValue (JSON), optional (JSON), scalarValue (TEXT), flow_document (JSON), dateValue (TEXT), datetimeValue (TEXT), numericString (TEXT), timeValue (TEXT)
 
-1, 0, 2024-09-13 01:01:01.000000, 1, <nil>, <nil>, <nil>, test, {}, 2024-01-01, 2024-01-01 01:01:01.000000, 123, 01:01:01.000000
+1, 0, 2024-09-13 01:01:01.000000, 1, <nil>, <nil>, <nil>, test, {}, 2024-01-01, 2024-01-01T01:01:01.111111Z, 123, 01:01:01.000000
 

materialize-mysql/client.go

@@ -116,7 +116,7 @@ func (c *client) columnMigrationSteps(ctx context.Context) []sql.ColumnMigration
 				"UPDATE %s SET %s = %s;",
 				table.Identifier,
 				tempColumnIdentifier,
-				migration.Identifier,
+				migration.CastSQL(migration),
 			), nil
 		},
 		func(dialect sql.Dialect, table sql.Table, migration sql.ColumnTypeMigration, _ string) (string, error) {

materialize-mysql/sqlgen.go

@@ -85,14 +85,16 @@ var mysqlDialect = func(tzLocation *time.Location, database string) sql.Dialect
 		sql.WithNotNullText("NOT NULL"),
 	)
 
+	var nocast = sql.WithCastSQL(migrationIdentifier)
+
 	return sql.Dialect{
-		MigratableTypes: map[string][]string{
-			"decimal":  {"varchar", "longtext"},
-			"bigint":   {"varchar", "longtext"},
-			"double":   {"varchar", "longtext"},
-			"date":     {"longtext", "varchar"},
-			"time":     {"longtext", "varchar"},
-			"datetime": {"longtext", "varchar"},
+		MigratableTypes: sql.MigrationSpecs{
+			"decimal":  {sql.NewMigrationSpec([]string{"varchar", "longtext"}, nocast)},
+			"bigint":   {sql.NewMigrationSpec([]string{"varchar", "longtext"}, nocast)},
+			"double":   {sql.NewMigrationSpec([]string{"varchar", "longtext"}, nocast)},
+			"date":     {sql.NewMigrationSpec([]string{"varchar", "longtext"}, nocast)},
+			"time":     {sql.NewMigrationSpec([]string{"varchar", "longtext"}, nocast)},
+			"datetime": {sql.NewMigrationSpec([]string{"varchar", "longtext"}, sql.WithCastSQL(prepareDatetimeToStringCast(tzLocation)))},
 		},
 		TableLocatorer: sql.TableLocatorFn(func(path []string) sql.InfoTableLocation {
 			// For MySQL, the table_catalog is always "def", and table_schema is the name of the
@@ -118,6 +120,31 @@ var mysqlDialect = func(tzLocation *time.Location, database string) sql.Dialect
 	}
 }
 
+func formatOffset(loc *time.Location) string {
+	now := time.Now().In(loc)
+	_, offset := now.Zone()
+
+	sign := "+"
+	if offset < 0 {
+		sign = "-"
+		offset = -offset
+	}
+	hours := offset / 3600
+	minutes := (offset % 3600) / 60
+	return fmt.Sprintf("%s%02d:%02d", sign, hours, minutes)
+}
+
+// by default we don't want to do `CAST(%s AS %s)` for MySQL
+func migrationIdentifier(migration sql.ColumnTypeMigration) string {
+	return migration.Identifier
+}
+
+func prepareDatetimeToStringCast(loc *time.Location) sql.CastSQLFunc {
+	return func(migration sql.ColumnTypeMigration) string {
+		return fmt.Sprintf(`DATE_FORMAT(CONVERT_TZ(%s, '%s', '+00:00'), '%%Y-%%m-%%dT%%H:%%i:%%s.%%fZ')`, migration.Identifier, formatOffset(loc))
+	}
+}
+
 func rfc3339ToTZ(loc *time.Location) sql.ElementConverter {
 	return sql.StringCastConverter(func(str string) (interface{}, error) {
 		// sanity check, this should not happen

materialize-postgres/.snapshots/TestValidateAndApplyMigrations

@@ -34,7 +34,7 @@ Migratable Changes Before Apply Schema:
 Migratable Changes Before Apply Data:
 key (TEXT), _meta/flow_truncated (BOOL), dateValue (DATE), datetimeValue (TIMESTAMPTZ), flow_published_at (TIMESTAMPTZ), int64 (INT8), multiple (JSON), nonScalarValue (JSON), numericString (NUMERIC), optional (JSON), scalarValue (TEXT), timeValue (TIME), flow_document (JSON)
 
-1, false, 2024-01-01 00:00:00 +0000 UTC, 2024-01-01 01:01:01 +0000 GMT, 2024-09-13 01:01:01 +0000 +0000, 1, <nil>, <nil>, 123, <nil>, test, 01:01:01, {}
+1, false, 2024-01-01 00:00:00 +0000 UTC, 2024-01-01 01:01:01.111111 +0000 GMT, 2024-09-13 01:01:01 +0000 +0000, 1, <nil>, <nil>, 123, <nil>, test, 01:01:01, {}
 
 Migratable Changes Constraints:
 {"Field":"_meta/flow_truncated","Type":3,"TypeString":"LOCATION_RECOMMENDED","Reason":"This location is part of the current materialization"}
@@ -72,5 +72,5 @@ Migratable Changes Applied Schema:
 Migratable Changes Applied Data:
 key (TEXT), _meta/flow_truncated (BOOL), flow_published_at (TIMESTAMPTZ), int64 (INT8), multiple (JSON), nonScalarValue (JSON), optional (JSON), scalarValue (TEXT), flow_document (JSON), dateValue (TEXT), datetimeValue (TEXT), numericString (TEXT), timeValue (TEXT)
 
-1, false, 2024-09-13 01:01:01 +0000 +0000, 1, <nil>, <nil>, <nil>, test, {}, 2024-01-01, 2024-01-01 01:01:01+00, 123, 01:01:01
+1, false, 2024-09-13 01:01:01 +0000 +0000, 1, <nil>, <nil>, <nil>, test, {}, 2024-01-01, 2024-01-01T01:01:01.111111+00, 123, 01:01:01
 

materialize-postgres/sqlgen.go

@@ -54,13 +54,13 @@ var pgDialect = func() sql.Dialect {
 	)
 
 	return sql.Dialect{
-		MigratableTypes: map[string][]string{
-			"numeric":                  {"text"},
-			"integer":                  {"text"},
-			"double precision":         {"text"},
-			"date":                     {"text"},
-			"time without time zone":   {"text"},
-			"timestamp with time zone": {"text"},
+		MigratableTypes: sql.MigrationSpecs{
+			"numeric":                  {sql.NewMigrationSpec([]string{"text"})},
+			"integer":                  {sql.NewMigrationSpec([]string{"text"})},
+			"double precision":         {sql.NewMigrationSpec([]string{"text"})},
+			"date":                     {sql.NewMigrationSpec([]string{"text"})},
+			"time without time zone":   {sql.NewMigrationSpec([]string{"text"})},
+			"timestamp with time zone": {sql.NewMigrationSpec([]string{"text"}, sql.WithCastSQL(datetimeToStringCast))},
 		},
 		TableLocatorer: sql.TableLocatorFn(func(path []string) sql.InfoTableLocation {
 			if len(path) == 1 {
@@ -100,6 +100,10 @@ type loadTableColumns struct {
 	Keys    []loadTableKey
 }
 
+func datetimeToStringCast(migration sql.ColumnTypeMigration) string {
+	return fmt.Sprintf(`to_char(%s AT TIME ZONE 'UTC', 'YYYY-MM-DD"T"HH24:MI:SS.USOF')`, migration.Identifier)
+}
+
 var (
 	tplAll = sql.MustParseTemplate(pgDialect, "root", `
 {{ define "temp_name" -}}

materialize-redshift/.snapshots/TestValidateAndApplyMigrations

@@ -34,7 +34,7 @@ Migratable Changes Before Apply Schema:
 Migratable Changes Before Apply Data:
 key (VARCHAR), _meta/flow_truncated (BOOL), datevalue (DATE), datetimevalue (TIMESTAMPTZ), flow_published_at (TIMESTAMPTZ), int64 (INT8), multiple (4000), nonscalarvalue (4000), numericstring (NUMERIC), optional (4000), scalarvalue (VARCHAR), timevalue (VARCHAR), flow_document (4000)
 
-1, false, 2024-01-01 00:00:00 +0000 UTC, 2024-01-01 01:01:01 +0000 GMT, 2024-09-13 02:01:01 +0100 IST, 1, <nil>, <nil>, 123, <nil>, test, 01:01:01, "{}"
+1, false, 2024-01-01 00:00:00 +0000 UTC, 2024-01-01 01:01:01.111111 +0000 GMT, 2024-09-13 02:01:01 +0100 IST, 1, <nil>, <nil>, 123, <nil>, test, 01:01:01, "{}"
 
 Migratable Changes Constraints:
 {"Field":"_meta/flow_truncated","Type":3,"TypeString":"LOCATION_RECOMMENDED","Reason":"This location is part of the current materialization"}
@@ -72,5 +72,5 @@ Migratable Changes Applied Schema:
 Migratable Changes Applied Data:
 key (VARCHAR), _meta/flow_truncated (BOOL), datevalue (TIMESTAMPTZ), datetimevalue (INT8), flow_published_at (4000), int64 (4000), multiple (4000), nonscalarvalue (VARCHAR), numericstring (VARCHAR), optional (4000), scalarvalue (VARCHAR), timevalue (VARCHAR), flow_document (VARCHAR)
 
-1, false, 2024-09-13 02:01:01 +0100 IST, 1, <nil>, <nil>, <nil>, test, 01:01:01, "{}", 2024-01-01, 2024-01-01 01:01:01+00, 123
+1, false, 2024-09-13 02:01:01 +0100 IST, 1, <nil>, <nil>, <nil>, test, 01:01:01, "{}", 2024-01-01, 2024-01-01T01:01:01.111111+00, 123