filesink: support castToString field config for parquet

Adds support for the `castToString` field configuration option in `filesink`,
allowing fields to be cast to string format when writing to parquet files. The
current motivator is that sometimes people want to read UUID values as strings
rather than the UUID logical type due to downstream compatibility issues.

It didn't make much sense to me to add this to the CSV writer also, since CSV
files are always strings for everything.

Author: williamhbaker

filesink/filesink.go

@@ -55,13 +55,21 @@ func (r resource) Validate() error {
 	return nil
 }
 
+type fieldConfig struct {
+	CastToString bool `json:"castToString"`
+}
+
+func (fc fieldConfig) Validate() error {
+	return nil
+}
+
 var _ boilerplate.Connector = &FileDriver{}
 
 // FileDriver contains the behaviors particular to a destination system and file format.
 type FileDriver struct {
 	NewConfig        func(raw json.RawMessage) (Config, error)
 	NewStore         func(ctx context.Context, config Config) (Store, error)
-	NewEncoder       func(config Config, b *pf.MaterializationSpec_Binding, w io.WriteCloser) StreamEncoder
+	NewEncoder       func(config Config, b *pf.MaterializationSpec_Binding, w io.WriteCloser) (StreamEncoder, error)
 	NewConstraints   func(p *pf.Projection) *pm.Response_Validated_Constraint
 	DocumentationURL func() string
 	ConfigSchema     func() ([]byte, error)
@@ -131,8 +139,6 @@ func (d FileDriver) NewTransactor(ctx context.Context, open pm.Request_Open, _ *
 	bindings := make([]binding, 0, len(open.Materialization.Bindings))
 
 	for _, b := range open.Materialization.Bindings {
-		b := b // for the newEncoder closure
-
 		var res resource
 		if err := pf.UnmarshalStrict(b.ResourceConfigJson, &res); err != nil {
 			return nil, nil, nil, err
@@ -143,7 +149,7 @@ func (d FileDriver) NewTransactor(ctx context.Context, open pm.Request_Open, _ *
 			backfill:   b.Backfill,
 			path:       res.Path,
 			includeDoc: b.FieldSelection.Document != "",
-			newEncoder: func(w io.WriteCloser) StreamEncoder {
+			newEncoder: func(w io.WriteCloser) (StreamEncoder, error) {
 				// Partial application of the driverCfg and b arguments to the FileDriver's NewEncoder
 				// function, for convenience.
 				return d.NewEncoder(driverCfg, b, w)
@@ -184,7 +190,7 @@ type binding struct {
 	backfill   uint32
 	path       string
 	includeDoc bool
-	newEncoder func(w io.WriteCloser) StreamEncoder
+	newEncoder func(w io.WriteCloser) (StreamEncoder, error)
 }
 
 // File keys are the full "path" to a file, usually applied as a key for an object in an object
@@ -244,7 +250,7 @@ func (t *transactor) Store(it *m.StoreIterator) (m.StartCommitFunc, error) {
 	var encoder StreamEncoder
 	var group errgroup.Group
 
-	startFile := func(b binding) {
+	startFile := func(b binding) error {
 		// Start a new file upload. This may be called multiple times in a single transaction if
 		// there is more data than can fit in a single file.
 		r, w := io.Pipe()
@@ -261,7 +267,9 @@ func (t *transactor) Store(it *m.StoreIterator) (m.StartCommitFunc, error) {
 			return nil
 		})
 
-		encoder = b.newEncoder(w)
+		var err error
+		encoder, err = b.newEncoder(w)
+		return err
 	}
 
 	finishFile := func() error {
@@ -290,7 +298,9 @@ func (t *transactor) Store(it *m.StoreIterator) (m.StartCommitFunc, error) {
 		lastBinding = it.Binding
 
 		if encoder == nil {
-			startFile(b)
+			if err := startFile(b); err != nil {
+				return nil, fmt.Errorf("startFile for binding %d: %w", it.Binding, err)
+			}
 		}
 
 		row := make([]any, 0, len(it.Key)+len(it.Values)+1)

filesink/format.go

@@ -25,8 +25,20 @@ func (c ParquetConfig) Validate() error {
 	return nil
 }
 
-func NewParquetStreamEncoder(cfg ParquetConfig, b *pf.MaterializationSpec_Binding, w io.WriteCloser) StreamEncoder {
-	sch := enc.FieldsToParquetSchema(b.FieldSelection.AllFields(), b.Collection)
+func NewParquetStreamEncoder(cfg ParquetConfig, b *pf.MaterializationSpec_Binding, w io.WriteCloser) (StreamEncoder, error) {
+	sch := make(enc.ParquetSchema, 0, len(b.FieldSelection.AllFields()))
+	for _, f := range b.FieldSelection.AllFields() {
+		p := b.Collection.GetProjection(f)
+
+		var fc fieldConfig
+		if raw, ok := b.FieldSelection.FieldConfigJsonMap[f]; ok {
+			if err := pf.UnmarshalStrict(raw, &fc); err != nil {
+				return nil, fmt.Errorf("unmarshaling field config for field %q: %w", f, err)
+			}
+		}
+
+		sch = append(sch, enc.ProjectionToParquetSchemaElement(*p, fc.CastToString))
+	}
 
 	var opts []enc.ParquetOption
 
@@ -41,7 +53,7 @@ func NewParquetStreamEncoder(cfg ParquetConfig, b *pf.MaterializationSpec_Bindin
 	// compression.
 	opts = append(opts, enc.WithParquetCompression(enc.Snappy))
 
-	return enc.NewParquetEncoder(w, sch, opts...)
+	return enc.NewParquetEncoder(w, sch, opts...), nil
 }
 
 type CsvConfig struct {

materialize-azure-blob-parquet/main.go

@@ -47,7 +47,7 @@ var driver = filesink.FileDriver{
 	NewStore: func(ctx context.Context, c filesink.Config) (filesink.Store, error) {
 		return filesink.NewAzureBlob(ctx, c.(config).AzureBlobConfig)
 	},
-	NewEncoder: func(c filesink.Config, b *pf.MaterializationSpec_Binding, w io.WriteCloser) filesink.StreamEncoder {
+	NewEncoder: func(c filesink.Config, b *pf.MaterializationSpec_Binding, w io.WriteCloser) (filesink.StreamEncoder, error) {
 		return filesink.NewParquetStreamEncoder(c.(config).ParquetConfig, b, w)
 	},
 	NewConstraints: func(p *pf.Projection) *materialize.Response_Validated_Constraint {

materialize-boilerplate/stream-encode/parquet_schema.go

@@ -217,18 +217,7 @@ func WithParquetUUIDAsString() ParquetSchemaOption {
 	}
 }
 
-func FieldsToParquetSchema(fields []string, collection pf.CollectionSpec, opts ...ParquetSchemaOption) ParquetSchema {
-	out := make(ParquetSchema, 0, len(fields))
-
-	for _, f := range fields {
-		p := collection.GetProjection(f)
-		out = append(out, ProjectionToParquetSchemaElement(*p, opts...))
-	}
-
-	return out
-}
-
-func ProjectionToParquetSchemaElement(p pf.Projection, opts ...ParquetSchemaOption) ParquetSchemaElement {
+func ProjectionToParquetSchemaElement(p pf.Projection, castToString bool, opts ...ParquetSchemaOption) ParquetSchemaElement {
 	cfg := parquetSchemaConfig{}
 	for _, o := range opts {
 		o(&cfg)
@@ -239,24 +228,24 @@ func ProjectionToParquetSchemaElement(p pf.Projection, opts ...ParquetSchemaOpti
 		Required: !slices.Contains(p.Inference.Types, "null") && (p.Inference.Exists == pf.Inference_MUST || p.Inference.DefaultJson != nil),
 	}
 
+	if castToString {
+		out.DataType = LogicalTypeString
+		return out
+	}
+
 	if numFormat, ok := boilerplate.AsFormattedNumeric(&p); ok {
 		if numFormat == boilerplate.StringFormatInteger {
 			out.DataType = PrimitiveTypeInteger
 		} else {
 			out.DataType = PrimitiveTypeNumber
 		}
 
-		if slices.Contains(p.Inference.Types, "null") {
-			out.Required = false
-		}
-
 		return out
 	}
 
 	hadType := false
 	for _, t := range p.Inference.Types {
 		if t == "null" {
-			out.Required = false
 			continue
 		}
 

materialize-gcs-csv/main.go

@@ -47,8 +47,8 @@ var driver = filesink.FileDriver{
 	NewStore: func(ctx context.Context, c filesink.Config) (filesink.Store, error) {
 		return filesink.NewGCSStore(ctx, c.(config).GCSStoreConfig)
 	},
-	NewEncoder: func(c filesink.Config, b *pf.MaterializationSpec_Binding, w io.WriteCloser) filesink.StreamEncoder {
-		return filesink.NewCsvStreamEncoder(c.(config).CsvConfig, b, w)
+	NewEncoder: func(c filesink.Config, b *pf.MaterializationSpec_Binding, w io.WriteCloser) (filesink.StreamEncoder, error) {
+		return filesink.NewCsvStreamEncoder(c.(config).CsvConfig, b, w), nil
 	},
 	NewConstraints: func(p *pf.Projection) *materialize.Response_Validated_Constraint {
 		return filesink.StdConstraints(p)

materialize-gcs-parquet/main.go

@@ -47,7 +47,7 @@ var driver = filesink.FileDriver{
 	NewStore: func(ctx context.Context, c filesink.Config) (filesink.Store, error) {
 		return filesink.NewGCSStore(ctx, c.(config).GCSStoreConfig)
 	},
-	NewEncoder: func(c filesink.Config, b *pf.MaterializationSpec_Binding, w io.WriteCloser) filesink.StreamEncoder {
+	NewEncoder: func(c filesink.Config, b *pf.MaterializationSpec_Binding, w io.WriteCloser) (filesink.StreamEncoder, error) {
 		return filesink.NewParquetStreamEncoder(c.(config).ParquetConfig, b, w)
 	},
 	NewConstraints: func(p *pf.Projection) *materialize.Response_Validated_Constraint {

materialize-s3-csv/main.go

@@ -47,8 +47,8 @@ var driver = filesink.FileDriver{
 	NewStore: func(ctx context.Context, c filesink.Config) (filesink.Store, error) {
 		return filesink.NewS3Store(ctx, c.(config).S3StoreConfig)
 	},
-	NewEncoder: func(c filesink.Config, b *pf.MaterializationSpec_Binding, w io.WriteCloser) filesink.StreamEncoder {
-		return filesink.NewCsvStreamEncoder(c.(config).CsvConfig, b, w)
+	NewEncoder: func(c filesink.Config, b *pf.MaterializationSpec_Binding, w io.WriteCloser) (filesink.StreamEncoder, error) {
+		return filesink.NewCsvStreamEncoder(c.(config).CsvConfig, b, w), nil
 	},
 	NewConstraints: func(p *pf.Projection) *materialize.Response_Validated_Constraint {
 		return filesink.StdConstraints(p)

materialize-s3-iceberg/type_mapping.go

@@ -82,7 +82,7 @@ func projectionToParquetSchemaElement(p pf.Projection, fc fieldConfig) (enc.Parq
 		p.Inference.String_.Format = ""
 	}
 
-	return enc.ProjectionToParquetSchemaElement(p, schemaOptions...), nil
+	return enc.ProjectionToParquetSchemaElement(p, false, schemaOptions...), nil
 }
 
 func parquetTypeToIcebergType(pqt enc.ParquetDataType) icebergType {

materialize-s3-parquet/main.go

@@ -47,7 +47,7 @@ var driver = filesink.FileDriver{
 	NewStore: func(ctx context.Context, c filesink.Config) (filesink.Store, error) {
 		return filesink.NewS3Store(ctx, c.(config).S3StoreConfig)
 	},
-	NewEncoder: func(c filesink.Config, b *pf.MaterializationSpec_Binding, w io.WriteCloser) filesink.StreamEncoder {
+	NewEncoder: func(c filesink.Config, b *pf.MaterializationSpec_Binding, w io.WriteCloser) (filesink.StreamEncoder, error) {
 		return filesink.NewParquetStreamEncoder(c.(config).ParquetConfig, b, w)
 	},
 	NewConstraints: func(p *pf.Projection) *materialize.Response_Validated_Constraint {