Add WriteZeroOptionalFields option for Snowflake compatibility

This commit adds a new WriterConfig option 'WriteZeroOptionalFields' that forces
the thrift encoder to write optional fields even when they have zero values.

This fixes compatibility issues with systems like Snowflake that require explicit
null counts in statistics (NullCount=0) even when columns have no null values.

Changes:
- Added WriteZeroOptionalFields feature flag to thrift protocol
- Made CompactProtocol configurable with SetFeatures method
- Updated thrift encoder to respect writeZeroOptionalFields flag
- Added WriteZeroOptionalFields config option to WriterConfig
- Wired the option through to ColumnWriter protocol initialization

The default behavior remains unchanged (false) for backward compatibility.

Usage:
  writer := parquet.NewWriter(output, schema,
      parquet.WriteZeroOptionalFields(true),
  )

Author: aratz-lasa

config.go

@@ -213,38 +213,40 @@ func (c *ReaderConfig) Validate() error {
 //		CreatedBy: "my test program",
 //	})
 type WriterConfig struct {
-	CreatedBy            string
-	ColumnPageBuffers    BufferPool
-	ColumnIndexSizeLimit int
-	PageBufferSize       int
-	WriteBufferSize      int
-	DataPageVersion      int
-	DataPageStatistics   bool
-	MaxRowsPerRowGroup   int64
-	KeyValueMetadata     map[string]string
-	Schema               *Schema
-	BloomFilters         []BloomFilterColumn
-	Compression          compress.Codec
-	Sorting              SortingConfig
-	SkipPageBounds       [][]string
-	Encodings            map[Kind]encoding.Encoding
-	DictionaryMaxBytes   int64
-	SchemaConfig         *SchemaConfig
+	CreatedBy               string
+	ColumnPageBuffers       BufferPool
+	ColumnIndexSizeLimit    int
+	PageBufferSize          int
+	WriteBufferSize         int
+	DataPageVersion         int
+	DataPageStatistics      bool
+	MaxRowsPerRowGroup      int64
+	KeyValueMetadata        map[string]string
+	Schema                  *Schema
+	BloomFilters            []BloomFilterColumn
+	Compression             compress.Codec
+	Sorting                 SortingConfig
+	SkipPageBounds          [][]string
+	Encodings               map[Kind]encoding.Encoding
+	DictionaryMaxBytes      int64
+	SchemaConfig            *SchemaConfig
+	WriteZeroOptionalFields bool
 }
 
 // DefaultWriterConfig returns a new WriterConfig value initialized with the
 // default writer configuration.
 func DefaultWriterConfig() *WriterConfig {
 	return &WriterConfig{
-		CreatedBy:            defaultCreatedBy(),
-		ColumnPageBuffers:    &defaultColumnBufferPool,
-		ColumnIndexSizeLimit: DefaultColumnIndexSizeLimit,
-		PageBufferSize:       DefaultPageBufferSize,
-		WriteBufferSize:      DefaultWriteBufferSize,
-		DataPageVersion:      DefaultDataPageVersion,
-		DataPageStatistics:   DefaultDataPageStatistics,
-		MaxRowsPerRowGroup:   DefaultMaxRowsPerRowGroup,
-		SchemaConfig:         DefaultSchemaConfig(),
+		CreatedBy:               defaultCreatedBy(),
+		ColumnPageBuffers:       &defaultColumnBufferPool,
+		ColumnIndexSizeLimit:    DefaultColumnIndexSizeLimit,
+		PageBufferSize:          DefaultPageBufferSize,
+		WriteBufferSize:         DefaultWriteBufferSize,
+		DataPageVersion:         DefaultDataPageVersion,
+		DataPageStatistics:      DefaultDataPageStatistics,
+		MaxRowsPerRowGroup:      DefaultMaxRowsPerRowGroup,
+		SchemaConfig:            DefaultSchemaConfig(),
+		WriteZeroOptionalFields: false,
 		Sorting: SortingConfig{
 			SortingBuffers: &defaultSortingBufferPool,
 		},
@@ -288,22 +290,23 @@ func (c *WriterConfig) ConfigureWriter(config *WriterConfig) {
 	}
 
 	*config = WriterConfig{
-		CreatedBy:            coalesceString(c.CreatedBy, config.CreatedBy),
-		ColumnPageBuffers:    coalesceBufferPool(c.ColumnPageBuffers, config.ColumnPageBuffers),
-		ColumnIndexSizeLimit: coalesceInt(c.ColumnIndexSizeLimit, config.ColumnIndexSizeLimit),
-		PageBufferSize:       coalesceInt(c.PageBufferSize, config.PageBufferSize),
-		WriteBufferSize:      coalesceInt(c.WriteBufferSize, config.WriteBufferSize),
-		DataPageVersion:      coalesceInt(c.DataPageVersion, config.DataPageVersion),
-		DataPageStatistics:   coalesceBool(c.DataPageStatistics, config.DataPageStatistics),
-		MaxRowsPerRowGroup:   coalesceInt64(c.MaxRowsPerRowGroup, config.MaxRowsPerRowGroup),
-		KeyValueMetadata:     keyValueMetadata,
-		Schema:               coalesceSchema(c.Schema, config.Schema),
-		BloomFilters:         coalesceBloomFilters(c.BloomFilters, config.BloomFilters),
-		Compression:          coalesceCompression(c.Compression, config.Compression),
-		Sorting:              coalesceSortingConfig(c.Sorting, config.Sorting),
-		SkipPageBounds:       coalesceSkipPageBounds(c.SkipPageBounds, config.SkipPageBounds),
-		Encodings:            encodings,
-		SchemaConfig:         coalesceSchemaConfig(c.SchemaConfig, config.SchemaConfig),
+		CreatedBy:               coalesceString(c.CreatedBy, config.CreatedBy),
+		ColumnPageBuffers:       coalesceBufferPool(c.ColumnPageBuffers, config.ColumnPageBuffers),
+		ColumnIndexSizeLimit:    coalesceInt(c.ColumnIndexSizeLimit, config.ColumnIndexSizeLimit),
+		PageBufferSize:          coalesceInt(c.PageBufferSize, config.PageBufferSize),
+		WriteBufferSize:         coalesceInt(c.WriteBufferSize, config.WriteBufferSize),
+		DataPageVersion:         coalesceInt(c.DataPageVersion, config.DataPageVersion),
+		DataPageStatistics:      coalesceBool(c.DataPageStatistics, config.DataPageStatistics),
+		MaxRowsPerRowGroup:      coalesceInt64(c.MaxRowsPerRowGroup, config.MaxRowsPerRowGroup),
+		KeyValueMetadata:        keyValueMetadata,
+		Schema:                  coalesceSchema(c.Schema, config.Schema),
+		BloomFilters:            coalesceBloomFilters(c.BloomFilters, config.BloomFilters),
+		Compression:             coalesceCompression(c.Compression, config.Compression),
+		Sorting:                 coalesceSortingConfig(c.Sorting, config.Sorting),
+		SkipPageBounds:          coalesceSkipPageBounds(c.SkipPageBounds, config.SkipPageBounds),
+		Encodings:               encodings,
+		SchemaConfig:            coalesceSchemaConfig(c.SchemaConfig, config.SchemaConfig),
+		WriteZeroOptionalFields: coalesceBool(c.WriteZeroOptionalFields, config.WriteZeroOptionalFields),
 	}
 }
 
@@ -688,6 +691,18 @@ func SkipPageBounds(path ...string) WriterOption {
 	return writerOption(func(config *WriterConfig) { config.SkipPageBounds = append(config.SkipPageBounds, path) })
 }
 
+// WriteZeroOptionalFields creates a configuration option which forces the writer
+// to always encode optional thrift fields even when they have zero values.
+// This is useful for compatibility with systems like Snowflake that require
+// explicit null counts in statistics even when they are zero.
+//
+// Defaults to false (zero-valued optional fields are omitted).
+func WriteZeroOptionalFields(enabled bool) WriterOption {
+	return writerOption(func(config *WriterConfig) {
+		config.WriteZeroOptionalFields = enabled
+	})
+}
+
 // DefaultEncodingFor creates a configuration option which sets the default encoding
 // used by a writer for columns with the specified primitive type where none were defined.
 //

encoding/thrift/compact.go

@@ -14,7 +14,9 @@ import (
 // CompactProtocol is a Protocol implementation for the compact thrift protocol.
 //
 // https://github.com/apache/thrift/blob/master/doc/specs/thrift-compact-protocol.md#integer-encoding
-type CompactProtocol struct{}
+type CompactProtocol struct {
+	features Features
+}
 
 func (p *CompactProtocol) NewReader(r io.Reader) Reader {
 	return &compactReader{protocol: p, binary: binaryReader{r: r}}
@@ -25,7 +27,15 @@ func (p *CompactProtocol) NewWriter(w io.Writer) Writer {
 }
 
 func (p *CompactProtocol) Features() Features {
-	return UseDeltaEncoding | CoalesceBoolFields
+	if p.features == 0 {
+		return UseDeltaEncoding | CoalesceBoolFields
+	}
+	return p.features
+}
+
+// SetFeatures configures the protocol features.
+func (p *CompactProtocol) SetFeatures(features Features) {
+	p.features = features
 }
 
 type compactReader struct {

encoding/thrift/encode.go

@@ -278,7 +278,7 @@ encodeFields:
 			}
 		}
 
-		if !f.flags.have(required) && x.IsZero() {
+		if !f.flags.have(required) && x.IsZero() && !flags.have(writeZeroOptionalFields) {
 			continue encodeFields
 		}
 

encoding/thrift/protocol.go

@@ -16,6 +16,10 @@ const (
 	// CoalesceBoolFields is advertised by protocols that allow encoders to
 	// coalesce boolean values into field types.
 	CoalesceBoolFields
+
+	// WriteZeroOptionalFields is advertised by protocols that allow encoders to
+	// write optional fields even when they have zero values.
+	WriteZeroOptionalFields
 )
 
 // The Protocol interface abstracts the creation of low-level thrift readers and

encoding/thrift/struct.go

@@ -16,14 +16,15 @@ const (
 	optional flags = 1 << 3
 	strict   flags = 1 << 4
 
-	featuresBitOffset  = 8
-	useDeltaEncoding   = flags(UseDeltaEncoding) << featuresBitOffset
-	coalesceBoolFields = flags(CoalesceBoolFields) << featuresBitOffset
+	featuresBitOffset       = 8
+	useDeltaEncoding        = flags(UseDeltaEncoding) << featuresBitOffset
+	coalesceBoolFields      = flags(CoalesceBoolFields) << featuresBitOffset
+	writeZeroOptionalFields = flags(WriteZeroOptionalFields) << featuresBitOffset
 
 	structFlags   flags = enum | union | required | optional
 	encodeFlags   flags = strict | protocolFlags
 	decodeFlags   flags = strict | protocolFlags
-	protocolFlags flags = useDeltaEncoding | coalesceBoolFields
+	protocolFlags flags = useDeltaEncoding | coalesceBoolFields | writeZeroOptionalFields
 )
 
 func (f flags) have(x flags) bool {

writer.go

@@ -751,6 +751,14 @@ func newWriterRowGroup(w *writer, config *WriterConfig) *writerRowGroup {
 			dictionaryMaxBytes: config.DictionaryMaxBytes,
 		}
 
+		// Configure protocol features based on writer options
+		c.header.protocol = &thrift.CompactProtocol{}
+		if config.WriteZeroOptionalFields {
+			c.header.protocol.SetFeatures(thrift.UseDeltaEncoding |
+				thrift.CoalesceBoolFields |
+				thrift.WriteZeroOptionalFields)
+		}
+
 		c.header.encoder.Reset(c.header.protocol.NewWriter(&c.buffers.header))
 
 		if leaf.maxDefinitionLevel > 0 {
@@ -1404,7 +1412,7 @@ type ColumnWriter struct {
 	buffers *writerBuffers
 
 	header struct {
-		protocol thrift.CompactProtocol
+		protocol *thrift.CompactProtocol
 		encoder  thrift.Encoder
 	}