parquet_encode: add default_timestamp_unit field

Adds a new `default_timestamp_unit` configuration field to the
`parquet_encode` processor, accepting `NANOSECOND` (default, preserves
existing behaviour), `MICROSECOND`, or `MILLISECOND`.

The unit is applied to both the static schema path and the dynamic
`schema_metadata` path (used by CDC inputs such as `mysql_cdc`).

`TIMESTAMP(NANOS)` is not readable by Apache Spark / Databricks, AWS
Athena or DuckDB; this field unblocks those consumers without requiring
a pre-encoding transform. MySQL sources additionally cannot exceed
microsecond precision, so `MICROSECOND` is lossless for CDC pipelines.

Resolves the long-standing TODO referenced at
#3570.

Author: binlogreader

CHANGELOG.md

@@ -3,6 +3,12 @@ Changelog
 
 All notable changes to this project will be documented in this file.
 
+## Unreleased
+
+### Added
+
+- parquet_encode: Added `default_timestamp_unit` field (values `NANOSECOND`, `MICROSECOND`, `MILLISECOND`) controlling the precision of TIMESTAMP logical types. Default remains `NANOSECOND` for backwards compatibility. Use `MICROSECOND` when writing files for Apache Spark/Databricks, AWS Athena or DuckDB, which do not support `TIMESTAMP(NANOS)`. ([#3570](https://github.com/redpanda-data/connect/issues/3570))
+
 ## 4.88.0 - 2026-04-16
 
 ### Added

docs/modules/components/pages/processors/parquet_encode.adoc

@@ -56,6 +56,7 @@ parquet_encode:
   schema_metadata: ""
   default_compression: uncompressed
   default_encoding: DELTA_LENGTH_BYTE_ARRAY
+  default_timestamp_unit: NANOSECOND
 ```
 
 --
@@ -216,4 +217,19 @@ Options:
 , `PLAIN`
 .
 
+=== `default_timestamp_unit`
+
+The precision used when encoding TIMESTAMP logical types. The default `NANOSECOND` matches historical behaviour, but `TIMESTAMP(NANOS)` is not readable by Apache Spark (Databricks), AWS Athena or DuckDB; set this to `MICROSECOND` (or `MILLISECOND`) when writing Parquet files intended for consumption by those engines.
+
+
+*Type*: `string`
+
+*Default*: `"NANOSECOND"`
+
+Options:
+`NANOSECOND`
+, `MICROSECOND`
+, `MILLISECOND`
+.
+
 

internal/impl/parquet/processor_encode.go

@@ -49,6 +49,12 @@ func parquetEncodeProcessorConfig() *service.ConfigSpec {
 				Default("DELTA_LENGTH_BYTE_ARRAY").
 				Advanced().
 				Version("4.11.0"),
+			service.NewStringEnumField("default_timestamp_unit",
+				"NANOSECOND", "MICROSECOND", "MILLISECOND",
+			).
+				Description("The precision used when encoding TIMESTAMP logical types. The default `NANOSECOND` matches historical behaviour, but `TIMESTAMP(NANOS)` is not readable by Apache Spark (Databricks), AWS Athena or DuckDB; set this to `MICROSECOND` (or `MILLISECOND`) when writing Parquet files intended for consumption by those engines.").
+				Default("NANOSECOND").
+				Advanced(),
 		).
 		Description(`
 This processor uses https://github.com/parquet-go/parquet-go[https://github.com/parquet-go/parquet-go^], which is itself experimental. Therefore changes could be made into how this processor functions outside of major version releases.
@@ -119,7 +125,19 @@ var plainEncodingFn encodingFn = func(n parquet.Node) parquet.Node {
 	return parquet.Encoded(n, &parquet.Plain)
 }
 
-func parquetGroupFromConfig(columnConfs []*service.ParsedConfig, encodingFn encodingFn) (parquet.Group, error) {
+func parseTimestampUnit(s string) (parquet.TimeUnit, error) {
+	switch s {
+	case "NANOSECOND":
+		return parquet.Nanosecond, nil
+	case "MICROSECOND":
+		return parquet.Microsecond, nil
+	case "MILLISECOND":
+		return parquet.Millisecond, nil
+	}
+	return nil, fmt.Errorf("unknown timestamp unit %q (expected NANOSECOND, MICROSECOND or MILLISECOND)", s)
+}
+
+func parquetGroupFromConfig(columnConfs []*service.ParsedConfig, encodingFn encodingFn, tsUnit parquet.TimeUnit) (parquet.Group, error) {
 	groupNode := parquet.Group{}
 
 	for _, colConf := range columnConfs {
@@ -131,7 +149,7 @@ func parquetGroupFromConfig(columnConfs []*service.ParsedConfig, encodingFn enco
 		}
 
 		if childColumns, _ := colConf.FieldAnyList("fields"); len(childColumns) > 0 {
-			if n, err = parquetGroupFromConfig(childColumns, encodingFn); err != nil {
+			if n, err = parquetGroupFromConfig(childColumns, encodingFn, tsUnit); err != nil {
 				return nil, err
 			}
 		} else {
@@ -155,8 +173,7 @@ func parquetGroupFromConfig(columnConfs []*service.ParsedConfig, encodingFn enco
 			case "UTF8":
 				n = parquet.String()
 			case "TIMESTAMP":
-				// TODO: add field to specify timestamp unit (https://github.com/redpanda-data/connect/issues/3570)
-				n = parquet.Timestamp(parquet.Nanosecond)
+				n = parquet.Timestamp(tsUnit)
 			case "BSON":
 				n = parquet.BSON()
 			case "ENUM":
@@ -193,6 +210,15 @@ func parquetGroupFromConfig(columnConfs []*service.ParsedConfig, encodingFn enco
 //------------------------------------------------------------------------------
 
 func newParquetEncodeProcessorFromConfig(conf *service.ParsedConfig, logger *service.Logger) (*parquetEncodeProcessor, error) {
+	tsUnitStr, err := conf.FieldString("default_timestamp_unit")
+	if err != nil {
+		return nil, err
+	}
+	tsUnit, err := parseTimestampUnit(tsUnitStr)
+	if err != nil {
+		return nil, err
+	}
+
 	var schema *parquet.Schema
 	if conf.Contains("schema") {
 		schemaConfs, err := conf.FieldObjectList("schema")
@@ -212,7 +238,7 @@ func newParquetEncodeProcessorFromConfig(conf *service.ParsedConfig, logger *ser
 			encoding = defaultEncodingFn
 		}
 
-		node, err := parquetGroupFromConfig(schemaConfs, encoding)
+		node, err := parquetGroupFromConfig(schemaConfs, encoding, tsUnit)
 		if err != nil {
 			return nil, err
 		}
@@ -250,22 +276,24 @@ func newParquetEncodeProcessorFromConfig(conf *service.ParsedConfig, logger *ser
 	default:
 		return nil, fmt.Errorf("default_compression type %v not recognised", compressStr)
 	}
-	return newParquetEncodeProcessor(logger, schema, schemaMeta, compressDefault)
+	return newParquetEncodeProcessor(logger, schema, schemaMeta, compressDefault, tsUnit)
 }
 
 type parquetEncodeProcessor struct {
 	logger          *service.Logger
 	schema          *parquet.Schema
 	schemaMeta      string
 	compressionType compress.Codec
+	timestampUnit   parquet.TimeUnit
 }
 
-func newParquetEncodeProcessor(logger *service.Logger, schema *parquet.Schema, schemaMeta string, compressionType compress.Codec) (*parquetEncodeProcessor, error) {
+func newParquetEncodeProcessor(logger *service.Logger, schema *parquet.Schema, schemaMeta string, compressionType compress.Codec, timestampUnit parquet.TimeUnit) (*parquetEncodeProcessor, error) {
 	s := &parquetEncodeProcessor{
 		logger:          logger,
 		schema:          schema,
 		schemaMeta:      schemaMeta,
 		compressionType: compressionType,
+		timestampUnit:   timestampUnit,
 	}
 	return s, nil
 }
@@ -305,7 +333,7 @@ func (s *parquetEncodeProcessor) ProcessBatch(_ context.Context, batch service.M
 		}
 
 		var err error
-		if schema, err = parquetSchemaFromCommon(metaAny); err != nil {
+		if schema, err = parquetSchemaFromCommon(metaAny, s.timestampUnit); err != nil {
 			return nil, err
 		}
 	}
@@ -348,7 +376,7 @@ func (*parquetEncodeProcessor) Close(context.Context) error {
 	return nil
 }
 
-func parquetNodeFromCommonField(field schema.Common) (parquet.Node, error) {
+func parquetNodeFromCommonField(field schema.Common, tsUnit parquet.TimeUnit) (parquet.Node, error) {
 	var n parquet.Node
 
 	switch field.Type {
@@ -365,8 +393,7 @@ func parquetNodeFromCommonField(field schema.Common) (parquet.Node, error) {
 	case schema.String:
 		n = parquet.String()
 	case schema.Timestamp:
-		// TODO: add field to specify timestamp unit (https://github.com/redpanda-data/connect/issues/3570)
-		n = parquet.Timestamp(parquet.Nanosecond)
+		n = parquet.Timestamp(tsUnit)
 	case schema.ByteArray:
 		n = parquet.Leaf(parquet.ByteArrayType)
 	case schema.Array:
@@ -375,7 +402,7 @@ func parquetNodeFromCommonField(field schema.Common) (parquet.Node, error) {
 		}
 
 		var err error
-		if n, err = parquetNodeFromCommonField(field.Children[0]); err != nil {
+		if n, err = parquetNodeFromCommonField(field.Children[0], tsUnit); err != nil {
 			return nil, err
 		}
 		n = parquet.Repeated(n)
@@ -386,7 +413,7 @@ func parquetNodeFromCommonField(field schema.Common) (parquet.Node, error) {
 		}
 
 		var err error
-		if n, err = parquetGroupFromCommonFields(field.Children); err != nil {
+		if n, err = parquetGroupFromCommonFields(field.Children, tsUnit); err != nil {
 			return nil, err
 		}
 
@@ -403,11 +430,11 @@ func parquetNodeFromCommonField(field schema.Common) (parquet.Node, error) {
 	return n, nil
 }
 
-func parquetGroupFromCommonFields(fields []schema.Common) (parquet.Group, error) {
+func parquetGroupFromCommonFields(fields []schema.Common, tsUnit parquet.TimeUnit) (parquet.Group, error) {
 	g := parquet.Group{}
 
 	for _, f := range fields {
-		n, err := parquetNodeFromCommonField(f)
+		n, err := parquetNodeFromCommonField(f, tsUnit)
 		if err != nil {
 			return nil, err
 		}
@@ -417,7 +444,7 @@ func parquetGroupFromCommonFields(fields []schema.Common) (parquet.Group, error)
 	return g, nil
 }
 
-func parquetSchemaFromCommon(a any) (*parquet.Schema, error) {
+func parquetSchemaFromCommon(a any, tsUnit parquet.TimeUnit) (*parquet.Schema, error) {
 	commonSchema, err := schema.ParseFromAny(a)
 	if err != nil {
 		return nil, err
@@ -431,7 +458,7 @@ func parquetSchemaFromCommon(a any) (*parquet.Schema, error) {
 		return nil, fmt.Errorf("source schema must have at least one field, got %v", len(commonSchema.Children))
 	}
 
-	groupNode, err := parquetGroupFromCommonFields(commonSchema.Children)
+	groupNode, err := parquetGroupFromCommonFields(commonSchema.Children, tsUnit)
 	if err != nil {
 		return nil, err
 	}

internal/impl/parquet/processor_encode_test.go

@@ -332,7 +332,7 @@ func TestParquetEncodeProcessor(t *testing.T) {
 			expectedDataBytes, err := json.Marshal(test.input)
 			require.NoError(t, err)
 
-			reader, err := newParquetEncodeProcessor(nil, testPMSchema(), "", &parquet.Uncompressed)
+			reader, err := newParquetEncodeProcessor(nil, testPMSchema(), "", &parquet.Uncompressed, parquet.Nanosecond)
 			require.NoError(t, err)
 
 			readerResBatches, err := reader.ProcessBatch(t.Context(), service.MessageBatch{
@@ -379,7 +379,7 @@ func TestParquetEncodeProcessor(t *testing.T) {
 			inBatch = append(inBatch, service.NewMessage(dataBytes))
 		}
 
-		reader, err := newParquetEncodeProcessor(nil, testPMSchema(), "", &parquet.Uncompressed)
+		reader, err := newParquetEncodeProcessor(nil, testPMSchema(), "", &parquet.Uncompressed, parquet.Nanosecond)
 		require.NoError(t, err)
 
 		readerResBatches, err := reader.ProcessBatch(t.Context(), inBatch)
@@ -578,7 +578,7 @@ func TestParquetEncodeDynamicSchemaProcessor(t *testing.T) {
 
 	inBatch[0].MetaSetMut("foobar", commonSchema.ToAny())
 
-	reader, err := newParquetEncodeProcessor(nil, nil, "foobar", &parquet.Uncompressed)
+	reader, err := newParquetEncodeProcessor(nil, nil, "foobar", &parquet.Uncompressed, parquet.Nanosecond)
 	require.NoError(t, err)
 
 	readerResBatches, err := reader.ProcessBatch(t.Context(), inBatch)
@@ -638,7 +638,7 @@ func TestParquetEncodeDynamicSchemaAnyFieldError(t *testing.T) {
 	}
 	inBatch[0].MetaSetMut("schema", commonSchema.ToAny())
 
-	proc, err := newParquetEncodeProcessor(nil, nil, "schema", &parquet.Uncompressed)
+	proc, err := newParquetEncodeProcessor(nil, nil, "schema", &parquet.Uncompressed, parquet.Nanosecond)
 	require.NoError(t, err)
 
 	_, err = proc.ProcessBatch(t.Context(), inBatch)
@@ -647,6 +647,96 @@ func TestParquetEncodeDynamicSchemaAnyFieldError(t *testing.T) {
 	assert.Contains(t, err.Error(), "ANY")
 }
 
+func TestParquetEncodeTimestampUnit(t *testing.T) {
+	tests := []struct {
+		name           string
+		unitConfig     string
+		expectedUnit   parquet.TimeUnit
+		expectedSchema string
+	}{
+		{name: "default is nanosecond", unitConfig: "", expectedUnit: parquet.Nanosecond, expectedSchema: "unit=NANOS"},
+		{name: "microsecond", unitConfig: "default_timestamp_unit: MICROSECOND", expectedUnit: parquet.Microsecond, expectedSchema: "unit=MICROS"},
+		{name: "millisecond", unitConfig: "default_timestamp_unit: MILLISECOND", expectedUnit: parquet.Millisecond, expectedSchema: "unit=MILLIS"},
+		{name: "explicit nanosecond", unitConfig: "default_timestamp_unit: NANOSECOND", expectedUnit: parquet.Nanosecond, expectedSchema: "unit=NANOS"},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			configYAML := fmt.Sprintf(`
+schema:
+  - { name: id, type: INT64 }
+  - { name: ts, type: TIMESTAMP }
+%s
+`, test.unitConfig)
+			encodeConf, err := parquetEncodeProcessorConfig().ParseYAML(configYAML, nil)
+			require.NoError(t, err)
+
+			encodeProc, err := newParquetEncodeProcessorFromConfig(encodeConf, nil)
+			require.NoError(t, err)
+			require.Equal(t, test.expectedUnit, encodeProc.timestampUnit)
+
+			batches, err := encodeProc.ProcessBatch(t.Context(), service.MessageBatch{
+				service.NewMessage([]byte(`{"id":1,"ts":"2026-04-17T12:00:00Z"}`)),
+			})
+			require.NoError(t, err)
+			require.Len(t, batches, 1)
+			require.Len(t, batches[0], 1)
+
+			pqBytes, err := batches[0][0].AsBytes()
+			require.NoError(t, err)
+
+			pqFile, err := parquet.OpenFile(bytes.NewReader(pqBytes), int64(len(pqBytes)))
+			require.NoError(t, err)
+			assert.Contains(t, pqFile.Schema().String(), test.expectedSchema)
+		})
+	}
+}
+
+func TestParquetEncodeTimestampUnitDynamicSchema(t *testing.T) {
+	encodeConf, err := parquetEncodeProcessorConfig().ParseYAML(`
+schema_metadata: benthos_schema
+default_timestamp_unit: MICROSECOND
+`, nil)
+	require.NoError(t, err)
+
+	encodeProc, err := newParquetEncodeProcessorFromConfig(encodeConf, nil)
+	require.NoError(t, err)
+
+	commonSchema := &schema.Common{
+		Type: schema.Object,
+		Children: []schema.Common{
+			{Name: "id", Type: schema.Int64},
+			{Name: "ts", Type: schema.Timestamp},
+		},
+	}
+
+	msg := service.NewMessage([]byte(`{"id":1,"ts":"2026-04-17T12:00:00Z"}`))
+	msg.MetaSetMut("benthos_schema", commonSchema.ToAny())
+
+	batches, err := encodeProc.ProcessBatch(t.Context(), service.MessageBatch{msg})
+	require.NoError(t, err)
+	require.Len(t, batches, 1)
+	require.Len(t, batches[0], 1)
+
+	pqBytes, err := batches[0][0].AsBytes()
+	require.NoError(t, err)
+
+	pqFile, err := parquet.OpenFile(bytes.NewReader(pqBytes), int64(len(pqBytes)))
+	require.NoError(t, err)
+	assert.Contains(t, pqFile.Schema().String(), "unit=MICROS")
+}
+
+func TestParquetEncodeTimestampUnitInvalid(t *testing.T) {
+	env := service.NewEnvironment()
+	err := env.NewStreamBuilder().AddProcessorYAML(`
+parquet_encode:
+  schema:
+    - { name: id, type: INT64 }
+  default_timestamp_unit: PICOSECOND
+`)
+	require.Error(t, err)
+}
+
 func TestParquetEncodeProcessorConfigLinting(t *testing.T) {
 	configTests := []struct {
 		name        string