Repository: parquet-go/parquet-go
Pull request #306 (2025-06-18 22:31:32)

add MergeNodes function to merge parquet nodes
This change adds a new `MergeNodes` function which combines multiple parquet nodes in a greedy way to retain the most important properties of each node.

I ended up expanding the PR to add `EqualTypes` and tests for `EqualTypes` and `EqualNodes`, as well as fixing an inconsistency in `EqualNodes` (it was checking logical types of leaf nodes but node the logical types of group nodes).


Patch:
diff --git a/convert.go b/convert.go
index 85f2995..fcb31b9 100644
--- a/convert.go
+++ b/convert.go
@@ -76,6 +76,7 @@ type conversionBuffer struct {
 type conversionColumn struct {
 	sourceIndex   int
 	convertValues conversionFunc
+	targetKind    Kind // Target column kind for creating proper null values
 }
 
 type conversionFunc func([]Value) error
@@ -193,12 +194,15 @@ func (c *conversion) Convert(rows []Row) (int, error) {
 				// placeholder in the column. This is a condition where the
 				// target contained a column which did not exist at had not
 				// other columns existing at that same level.
-				row = append(row, Value{})
+				// Create a properly typed null value for the target column
+				nullValue := ZeroValue(conv.targetKind)
+				row = append(row, nullValue)
 			} else {
+				sourceValues := source.columns[conv.sourceIndex]
 				// We must copy to the output row first and not mutate the
 				// source columns because multiple target columns may map to
 				// the same source column.
-				row = append(row, source.columns[conv.sourceIndex]...)
+				row = append(row, sourceValues...)
 			}
 			columnValues := row[columnOffset:]
 
@@ -210,6 +214,11 @@ func (c *conversion) Convert(rows []Row) (int, error) {
 			// taget columns we ensure that the right value is always written
 			// to the output row.
 			for i := range columnValues {
+				// Fix: If we have a zero Value{}, convert it to a properly typed null value
+				if columnValues[i].Kind() == Kind(0) {
+					columnValues[i] = ZeroValue(conv.targetKind)
+				}
+
 				columnValues[i].columnIndex = ^int16(columnIndex)
 			}
 		}
@@ -255,6 +264,7 @@ func Convert(to, from Node) (conv Conversion, err error) {
 
 	targetMapping, targetColumns := columnMappingOf(to)
 	sourceMapping, sourceColumns := columnMappingOf(from)
+
 	columns := make([]conversionColumn, len(targetColumns))
 
 	for i, path := range targetColumns {
@@ -265,7 +275,7 @@ func Convert(to, from Node) (conv Conversion, err error) {
 		if sourceColumn.node != nil {
 			targetType := targetColumn.node.Type()
 			sourceType := sourceColumn.node.Type()
-			if !typesAreEqual(targetType, sourceType) {
+			if !EqualTypes(targetType, sourceType) {
 				conversions = append(conversions,
 					convertToType(targetType, sourceType),
 				)
@@ -323,9 +333,13 @@ func Convert(to, from Node) (conv Conversion, err error) {
 			}
 		}
 
+		// Store target column type for creating proper null values
+		targetType := targetColumn.node.Type()
+
 		columns[i] = conversionColumn{
 			sourceIndex:   int(sourceColumn.columnIndex),
 			convertValues: multiConversionFunc(conversions),
+			targetKind:    targetType.Kind(), // Store target kind for null value creation
 		}
 	}
 
diff --git a/merge.go b/merge.go
index 1655a83..d08b5ff 100644
--- a/merge.go
+++ b/merge.go
@@ -1,11 +1,16 @@
 package parquet
 
 import (
+	"cmp"
 	"container/heap"
 	"errors"
 	"fmt"
 	"io"
+	"slices"
 	"sync"
+
+	"github.com/parquet-go/parquet-go/encoding"
+	"github.com/parquet-go/parquet-go/format"
 )
 
 // MergeRowGroups constructs a row group which is a merged view of rowGroups. If
@@ -30,39 +35,55 @@ func MergeRowGroups(rowGroups []RowGroup, options ...RowGroupOption) (RowGroup,
 		return newEmptyRowGroup(schema), nil
 	}
 	if schema == nil {
-		schema = rowGroups[0].Schema()
-
-		for _, rowGroup := range rowGroups[1:] {
-			if !EqualNodes(schema, rowGroup.Schema()) {
-				return nil, ErrRowGroupSchemaMismatch
-			}
+		schemas := make([]Node, len(rowGroups))
+		for i, rowGroup := range rowGroups {
+			schemas[i] = rowGroup.Schema()
 		}
+		schema = NewSchema(rowGroups[0].Schema().Name(), MergeNodes(schemas...))
 	}
 
-	mergedRowGroups := make([]RowGroup, len(rowGroups))
-	copy(mergedRowGroups, rowGroups)
-
+	mergedRowGroups := slices.Clone(rowGroups)
 	for i, rowGroup := range mergedRowGroups {
-		if rowGroupSchema := rowGroup.Schema(); !EqualNodes(schema, rowGroupSchema) {
-			conv, err := Convert(schema, rowGroupSchema)
-			if err != nil {
-				return nil, fmt.Errorf("cannot merge row groups: %w", err)
-			}
-			mergedRowGroups[i] = ConvertRowGroup(rowGroup, conv)
+		rowGroupSchema := rowGroup.Schema()
+		// Always apply conversion when merging multiple row groups to ensure
+		// column indices match the merged schema layout. The merge process can
+		// reorder fields even when schemas are otherwise identical.
+		conv, err := Convert(schema, rowGroupSchema)
+		if err != nil {
+			return nil, fmt.Errorf("cannot merge row groups: %w", err)
 		}
+		mergedRowGroups[i] = ConvertRowGroup(rowGroup, conv)
 	}
 
-	m := &mergedRowGroup{sorting: config.Sorting.SortingColumns}
-	m.init(schema, mergedRowGroups)
+	// Determine the effective sorting columns for the merge
+	mergedSortingColumns := slices.Clone(config.Sorting.SortingColumns)
+	if len(mergedSortingColumns) == 0 {
+		// Auto-detect common sorting columns from input row groups
+		sortingColumns := make([][]SortingColumn, len(mergedRowGroups))
+		for i, rowGroup := range mergedRowGroups {
+			sortingColumns[i] = rowGroup.SortingColumns()
+		}
+		mergedSortingColumns = MergeSortingColumns(sortingColumns...)
+	}
 
-	if len(m.sorting) == 0 {
-		// When the row group has no ordering, use a simpler version of the
+	if len(mergedSortingColumns) == 0 {
+		// When there are no effective sorting columns, use a simpler version of the
 		// merger which simply concatenates rows from each of the row groups.
 		// This is preferable because it makes the output deterministic, the
 		// heap merge may otherwise reorder rows across groups.
-		return &m.multiRowGroup, nil
+		//
+		// IMPORTANT: We need to ensure conversions are applied even in the simple
+		// concatenation path. Instead of returning the multiRowGroup directly
+		// (which bypasses row-level conversion), we create a simple concatenating
+		// row reader that preserves the conversion logic.
+		c := new(concatRowGroup)
+		c.init(schema, mergedRowGroups)
+		return c, nil
 	}
 
+	m := &mergedRowGroup{sorting: mergedSortingColumns}
+	m.init(schema, mergedRowGroups)
+
 	for _, rowGroup := range m.rowGroups {
 		if !sortingColumnsHavePrefix(rowGroup.SortingColumns(), m.sorting) {
 			return nil, ErrRowGroupSortingColumnsMismatch
@@ -73,6 +94,74 @@ func MergeRowGroups(rowGroups []RowGroup, options ...RowGroupOption) (RowGroup,
 	return m, nil
 }
 
+// concatRowGroup is used when there's no sorting requirement.
+// It provides the same interface as multiRowGroup but ensures that row-level
+// conversions are applied by reading through the converted row groups instead
+// of directly from column chunks.
+type concatRowGroup struct {
+	multiRowGroup
+	sorting []SortingColumn
+}
+
+func (s *concatRowGroup) SortingColumns() []SortingColumn {
+	return s.sorting
+}
+
+func (s *concatRowGroup) Rows() Rows {
+	rowReaders := make([]Rows, len(s.rowGroups))
+	for i, rowGroup := range s.rowGroups {
+		rowReaders[i] = rowGroup.Rows()
+	}
+	return &concatRows{readers: rowReaders}
+}
+
+// concatRows reads rows sequentially from multiple converted row groups
+type concatRows struct {
+	readers []Rows
+	current int
+}
+
+func (s *concatRows) ReadRows(rows []Row) (int, error) {
+	for s.current < len(s.readers) {
+		n, err := s.readers[s.current].ReadRows(rows)
+		if n > 0 {
+			return n, nil
+		}
+		if err != io.EOF {
+			return 0, err
+		}
+		s.current++
+	}
+	return 0, io.EOF
+}
+
+func (s *concatRows) Close() error {
+	var lastErr error
+	for _, reader := range s.readers {
+		if err := reader.Close(); err != nil {
+			lastErr = err
+		}
+	}
+	return lastErr
+}
+
+func (s *concatRows) Schema() *Schema {
+	if len(s.readers) > 0 {
+		return s.readers[0].Schema()
+	}
+	return nil
+}
+
+func (s *concatRows) SeekToRow(rowIndex int64) error {
+	// Simple implementation: just reset current reader
+	// More sophisticated seeking would require tracking row counts
+	s.current = 0
+	if len(s.readers) > 0 {
+		return s.readers[0].SeekToRow(rowIndex)
+	}
+	return io.EOF
+}
+
 type mergedRowGroup struct {
 	multiRowGroup
 	sorting []SortingColumn
@@ -499,3 +588,166 @@ var (
 	_ heap.Interface      = (*mergeBuffer)(nil)
 	_ RowWriterTo         = (*mergeBuffer)(nil)
 )
+
+// MergeNodes takes a list of nodes and greedily retains properties of the schemas:
+// - keeps last compression that is not nil
+// - keeps last non-plain encoding that is not nil
+// - keeps last non-zero field id
+// - union of all columns for group nodes
+// - retains the most permissive repetition (required < optional < repeated)
+func MergeNodes(nodes ...Node) Node {
+	switch len(nodes) {
+	case 0:
+		return nil
+	case 1:
+		return nodes[0]
+	default:
+		merged := nodes[0]
+		for _, node := range nodes[1:] {
+			merged = mergeTwoNodes(merged, node)
+		}
+		return merged
+	}
+}
+
+// mergeTwoNodes merges two nodes using greedy property retention
+func mergeTwoNodes(a, b Node) Node {
+	leaf1 := a.Leaf()
+	leaf2 := b.Leaf()
+	// Both must be either leaf or group nodes
+	if leaf1 != leaf2 {
+		// Cannot merge leaf with group - return the last one
+		return b
+	}
+
+	var merged Node
+	if leaf1 {
+		merged = Leaf(b.Type())
+
+		// Apply compression (keep last non-nil)
+		compression1 := a.Compression()
+		compression2 := b.Compression()
+		compression := cmp.Or(compression2, compression1)
+		if compression != nil {
+			merged = Compressed(merged, compression)
+		}
+
+		// Apply encoding (keep last non-plain, non-nil)
+		encoding := encoding.Encoding(&Plain)
+		encoding1 := a.Encoding()
+		encoding2 := b.Encoding()
+		if !isPlainEncoding(encoding1) {
+			encoding = encoding1
+		}
+		if !isPlainEncoding(encoding2) {
+			encoding = encoding2
+		}
+		if encoding != nil {
+			merged = Encoded(merged, encoding)
+		}
+	} else {
+		fields1 := slices.Clone(a.Fields())
+		fields2 := slices.Clone(b.Fields())
+		sortFields(fields1)
+		sortFields(fields2)
+
+		group := make(Group, len(fields1))
+		i1 := 0
+		i2 := 0
+		for i1 < len(fields1) && i2 < len(fields2) {
+			name1 := fields1[i1].Name()
+			name2 := fields2[i2].Name()
+			switch {
+			case name1 < name2:
+				group[name1] = nullable(fields1[i1])
+				i1++
+			case name1 > name2:
+				group[name2] = nullable(fields2[i2])
+				i2++
+			default:
+				group[name1] = mergeTwoNodes(fields1[i1], fields2[i2])
+				i1++
+				i2++
+			}
+		}
+
+		for _, field := range fields1[i1:] {
+			group[field.Name()] = nullable(field)
+		}
+
+		for _, field := range fields2[i2:] {
+			group[field.Name()] = nullable(field)
+		}
+
+		merged = group
+
+		if logicalType := b.Type().LogicalType(); logicalType != nil {
+			switch {
+			case logicalType.List != nil:
+				merged = &listNode{group}
+			case logicalType.Map != nil:
+				merged = &mapNode{group}
+			}
+		}
+	}
+
+	// Apply repetition (most permissive: required < optional < repeated)
+	if a.Repeated() || b.Repeated() {
+		merged = Repeated(merged)
+	} else if a.Optional() || b.Optional() {
+		merged = Optional(merged)
+	} else {
+		merged = Required(merged)
+	}
+
+	// Apply field ID (keep last non-zero)
+	return FieldID(merged, cmp.Or(b.ID(), a.ID()))
+}
+
+// isPlainEncoding checks if the encoding is plain encoding
+func isPlainEncoding(enc encoding.Encoding) bool {
+	return enc == nil || enc.Encoding() == format.Plain
+}
+
+func nullable(n Node) Node {
+	if !n.Repeated() {
+		return Optional(n)
+	}
+	return n
+}
+
+// MergeSortingColumns returns the common prefix of all sorting columns passed as arguments.
+// This function is used to determine the resulting sorting columns when merging multiple
+// row groups that each have their own sorting columns.
+//
+// The function returns the longest common prefix where all sorting columns match exactly
+// (same path, same descending flag, same nulls first flag). If any row group has no
+// sorting columns, or if there's no common prefix, an empty slice is returned.
+//
+// Example:
+//
+//	columns1 := []SortingColumn{Ascending("A"), Ascending("B"), Descending("C")}
+//	columns2 := []SortingColumn{Ascending("A"), Ascending("B"), Ascending("D")}
+//	result := MergeSortingColumns(columns1, columns2)
+//	// result will be []SortingColumn{Ascending("A"), Ascending("B")}
+func MergeSortingColumns(sortingColumns ...[]SortingColumn) []SortingColumn {
+	if len(sortingColumns) == 0 {
+		return nil
+	}
+	merged := slices.Clone(sortingColumns[0])
+	for _, columns := range sortingColumns[1:] {
+		merged = commonSortingPrefix(merged, columns)
+	}
+	return merged
+}
+
+// commonSortingPrefix returns the common prefix of two sorting column slices
+func commonSortingPrefix(a, b []SortingColumn) []SortingColumn {
+	minLen := min(len(a), len(b))
+	for i := range minLen {
+		if !equalSortingColumn(a[i], b[i]) {
+			return a[:i]
+		}
+	}
+	return a[:minLen]
+}
diff --git a/node.go b/node.go
index 14ad672..8abf0fa 100644
--- a/node.go
+++ b/node.go
@@ -493,12 +493,17 @@ func fieldByName(node Node, name string) Field {
 
 // EqualNodes returns true if node1 and node2 are equal.
 //
-// Nodes that are not of the same repetition type (optional, required, repeated) or
-// of the same hierarchical type (leaf, group) are considered not equal.
+// Nodes that are not of the same repetition type (optional, required, repeated)
+// or of the same hierarchical type (leaf, group) are considered not equal.
 // Leaf nodes are considered equal if they are of the same data type.
-// Group nodes are considered equal if their fields have the same names and are recursively equal.
 //
-// Note that the encoding and compression of the nodes are not considered by this function.
+// Groups are compared recursively, taking the order of fields into account
+// (because it influences the column index of each leaf node), and comparing
+// their logical types: for example, a MAP node is not equal to a GROUP node
+// with the same fields, because MAP nodes have a specific logical type.
+//
+// Note that the encoding and compression of the nodes are not considered by this
+// function.
 func EqualNodes(node1, node2 Node) bool {
 	if node1.Leaf() {
 		return node2.Leaf() && leafNodesAreEqual(node1, node2)
@@ -507,10 +512,20 @@ func EqualNodes(node1, node2 Node) bool {
 	}
 }
 
-func typesAreEqual(type1, type2 Type) bool {
-	return type1.Kind() == type2.Kind() &&
-		type1.Length() == type2.Length() &&
-		reflect.DeepEqual(type1.LogicalType(), type2.LogicalType())
+// SameNodes returns true if node1 and node2 are equivalent, ignoring field order.
+//
+// Unlike EqualNodes, this function considers nodes with the same fields in different
+// orders as equivalent. This is useful when comparing schemas that may have been
+// reordered by operations like MergeNodes.
+//
+// For leaf nodes, this behaves identically to EqualNodes.
+// For group nodes, this compares fields by name rather than position.
+func SameNodes(node1, node2 Node) bool {
+	if node1.Leaf() {
+		return node2.Leaf() && leafNodesAreEqual(node1, node2)
+	} else {
+		return !node2.Leaf() && groupNodesAreSame(node1, node2)
+	}
 }
 
 func repetitionsAreEqual(node1, node2 Node) bool {
@@ -518,33 +533,72 @@ func repetitionsAreEqual(node1, node2 Node) bool {
 }
 
 func leafNodesAreEqual(node1, node2 Node) bool {
-	return typesAreEqual(node1.Type(), node2.Type()) && repetitionsAreEqual(node1, node2)
+	return EqualTypes(node1.Type(), node2.Type()) && repetitionsAreEqual(node1, node2)
 }
 
 func groupNodesAreEqual(node1, node2 Node) bool {
 	fields1 := node1.Fields()
 	fields2 := node2.Fields()
-
 	if len(fields1) != len(fields2) {
 		return false
 	}
+	if !repetitionsAreEqual(node1, node2) {
+		return false
+	}
+	if !fieldsAreEqual(fields1, fields2, EqualNodes) {
+		return false
+	}
+	return equalLogicalTypes(node1.Type(), node2.Type())
+}
 
+func groupNodesAreSame(node1, node2 Node) bool {
+	fields1 := node1.Fields()
+	fields2 := node2.Fields()
+	if len(fields1) != len(fields2) {
+		return false
+	}
 	if !repetitionsAreEqual(node1, node2) {
 		return false
 	}
+	if !fieldsAreSorted(fields1) {
+		fields1 = slices.Clone(fields1)
+		sortFields(fields1)
+	}
+	if !fieldsAreSorted(fields2) {
+		fields2 = slices.Clone(fields2)
+		sortFields(fields2)
+	}
+	if !fieldsAreEqual(fields1, fields2, SameNodes) {
+		return false
+	}
+	return equalLogicalTypes(node1.Type(), node2.Type())
+}
 
+func fieldsAreEqual(fields1, fields2 []Field, equal func(Node, Node) bool) bool {
+	if len(fields1) != len(fields2) {
+		return false
+	}
 	for i := range fields1 {
-		f1 := fields1[i]
-		f2 := fields2[i]
-
-		if f1.Name() != f2.Name() {
+		if fields1[i].Name() != fields2[i].Name() {
 			return false
 		}
-
-		if !EqualNodes(f1, f2) {
+	}
+	for i := range fields1 {
+		if !equal(fields1[i], fields2[i]) {
 			return false
 		}
 	}
-
 	return true
 }
+
+func fieldsAreSorted(fields []Field) bool {
+	return slices.IsSortedFunc(fields, compareFields)
+}
+
+func sortFields(fields []Field) {
+	slices.SortFunc(fields, compareFields)
+}
+
+func compareFields(a, b Field) int {
+	return strings.Compare(a.Name(), b.Name())
+}
diff --git a/sorting.go b/sorting.go
index 4f213de..e4087b6 100644
--- a/sorting.go
+++ b/sorting.go
@@ -2,6 +2,7 @@ package parquet
 
 import (
 	"io"
+	"slices"
 	"sort"
 )
 
@@ -229,3 +230,41 @@ func (w *SortingWriter[T]) sortAndWriteBufferedRows() error {
 func (w *SortingWriter[T]) File() FileView {
 	return w.output.File()
 }
+
+// EqualSortingColumns compares two slices of sorting columns for equality.
+//
+// Two sorting column slices are considered equal if they have the same length
+// and each corresponding pair of sorting columns is equal. Two sorting columns
+// are equal if they have:
+//   - The same column path (including nested field paths)
+//   - The same sort direction (ascending or descending)
+//   - The same nulls handling (nulls first or nulls last)
+//
+// The comparison is order-sensitive, meaning that [A, B] is not equal to [B, A].
+// Both nil and empty slices are considered equal.
+//
+// This function is useful for:
+//   - Validating that merged row groups maintain expected sorting
+//   - Comparing sorting configurations between different row groups
+//   - Testing sorting column propagation in merge operations
+//
+// Example:
+//
+//	cols1 := []SortingColumn{Ascending("name"), Descending("age")}
+//	cols2 := []SortingColumn{Ascending("name"), Descending("age")}
+//	equal := EqualSortingColumns(cols1, cols2) // returns true
+//
+//	cols3 := []SortingColumn{Descending("age"), Ascending("name")}
+//	equal = EqualSortingColumns(cols1, cols3) // returns false (different order)
+//
+//	cols4 := []SortingColumn{Ascending("name"), Ascending("age")}
+//	equal = EqualSortingColumns(cols1, cols4) // returns false (different direction)
+func EqualSortingColumns(a, b []SortingColumn) bool {
+	return len(a) == len(b) && slices.EqualFunc(a, b, equalSortingColumn)
+}
+
+// equalSortingColumn compares two individual sorting columns for equality.
+// Two sorting columns are equal if they have the same path, direction, and nulls handling.
+func equalSortingColumn(a, b SortingColumn) bool {
+	return slices.Equal(a.Path(), b.Path()) && a.Descending() == b.Descending() && a.NullsFirst() == b.NullsFirst()
+}
diff --git a/type.go b/type.go
index fc3c6c7..a253a44 100644
--- a/type.go
+++ b/type.go
@@ -213,6 +213,30 @@ type Type interface {
 	ConvertValue(val Value, typ Type) (Value, error)
 }
 
+// EqualTypes returns true if type1 and type2 are equal.
+//
+// Types are considered equal if they have the same Kind, Length, and LogicalType.
+// The comparison uses reflect.DeepEqual for LogicalType comparison.
+//
+// Note: This function is designed for leaf types. For complex group types like
+// MAP and LIST, use EqualNodes instead, as those types require structural comparison
+// of their nested fields.
+func EqualTypes(type1, type2 Type) bool {
+	return equalKind(type1, type2) && equalLength(type1, type2) && equalLogicalTypes(type1, type2)
+}
+
+func equalKind(type1, type2 Type) bool {
+	return type1.Kind() == type2.Kind()
+}
+
+func equalLength(type1, type2 Type) bool {
+	return type1.Length() == type2.Length()
+}
+
+func equalLogicalTypes(type1, type2 Type) bool {
+	return reflect.DeepEqual(type1.LogicalType(), type2.LogicalType())
+}
+
 var (
 	BooleanType   Type = booleanType{}
 	Int32Type     Type = int32Type{}


Test patch:
diff --git a/merge_test.go b/merge_test.go
index f2f2088..8e59acb 100644
--- a/merge_test.go
+++ b/merge_test.go
@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"io"
 	"math/rand"
+	"slices"
 	"sort"
 	"testing"
 
@@ -55,6 +56,101 @@ func TestMergeRowGroups(t *testing.T) {
 			),
 		},
 
+		{
+			scenario: "two sorted row groups with common sorting",
+			options: []parquet.RowGroupOption{
+				parquet.SortingRowGroupConfig(
+					parquet.SortingColumns(
+						parquet.Ascending("FirstName"),
+						parquet.Ascending("LastName"),
+					),
+				),
+			},
+			input: []parquet.RowGroup{
+				sortedRowGroup(
+					[]parquet.RowGroupOption{
+						parquet.SortingRowGroupConfig(
+							parquet.SortingColumns(
+								parquet.Ascending("FirstName"),
+								parquet.Ascending("LastName"),
+							),
+						),
+					},
+					Person{FirstName: "Alice", LastName: "Brown", Age: 25},
+					Person{FirstName: "Bob", LastName: "Smith", Age: 30},
+				),
+				sortedRowGroup(
+					[]parquet.RowGroupOption{
+						parquet.SortingRowGroupConfig(
+							parquet.SortingColumns(
+								parquet.Ascending("FirstName"),
+								parquet.Ascending("LastName"),
+							),
+						),
+					},
+					Person{FirstName: "Charlie", LastName: "Johnson", Age: 35},
+					Person{FirstName: "David", LastName: "Wilson", Age: 40},
+				),
+			},
+			output: sortedRowGroup(
+				[]parquet.RowGroupOption{
+					parquet.SortingRowGroupConfig(
+						parquet.SortingColumns(
+							parquet.Ascending("FirstName"),
+							parquet.Ascending("LastName"),
+						),
+					),
+				},
+				Person{FirstName: "Alice", LastName: "Brown", Age: 25},
+				Person{FirstName: "Bob", LastName: "Smith", Age: 30},
+				Person{FirstName: "Charlie", LastName: "Johnson", Age: 35},
+				Person{FirstName: "David", LastName: "Wilson", Age: 40},
+			),
+		},
+
+		{
+			scenario: "two sorted row groups with partial common sorting",
+			input: []parquet.RowGroup{
+				sortedRowGroup(
+					[]parquet.RowGroupOption{
+						parquet.SortingRowGroupConfig(
+							parquet.SortingColumns(
+								parquet.Ascending("FirstName"),
+								parquet.Ascending("LastName"),
+							),
+						),
+					},
+					Person{FirstName: "Alice", LastName: "Brown", Age: 25},
+					Person{FirstName: "Bob", LastName: "Smith", Age: 30},
+				),
+				sortedRowGroup(
+					[]parquet.RowGroupOption{
+						parquet.SortingRowGroupConfig(
+							parquet.SortingColumns(
+								parquet.Ascending("FirstName"),
+								parquet.Descending("Age"), // Different second column
+							),
+						),
+					},
+					Person{FirstName: "Charlie", LastName: "Johnson", Age: 35},
+					Person{FirstName: "David", LastName: "Wilson", Age: 40},
+				),
+			},
+			output: sortedRowGroup(
+				[]parquet.RowGroupOption{
+					parquet.SortingRowGroupConfig(
+						parquet.SortingColumns(
+							parquet.Ascending("FirstName"), // Only FirstName should be preserved
+						),
+					),
+				},
+				Person{FirstName: "Alice", LastName: "Brown", Age: 25},
+				Person{FirstName: "Bob", LastName: "Smith", Age: 30},
+				Person{FirstName: "Charlie", LastName: "Johnson", Age: 35},
+				Person{FirstName: "David", LastName: "Wilson", Age: 40},
+			),
+		},
+
 		{
 			scenario: "a single row group",
 			input: []parquet.RowGroup{
@@ -315,6 +411,103 @@ func TestMergeRowGroups(t *testing.T) {
 		},
 	}
 
+	// Additional tests for different field ordering scenarios
+	fieldOrderTests := []struct {
+		scenario string
+		input    []parquet.RowGroup
+		output   parquet.RowGroup
+	}{
+		{
+			scenario: "two row groups with same fields in different order",
+			input: []parquet.RowGroup{
+				createRowGroupWithFieldOrder([]string{"Age", "FirstName", "LastName"},
+					Person{Age: 25, FirstName: "John", LastName: "Doe"},
+					Person{Age: 30, FirstName: "Jane", LastName: "Smith"},
+				),
+				createRowGroupWithFieldOrder([]string{"LastName", "Age", "FirstName"},
+					Person{Age: 35, FirstName: "Bob", LastName: "Johnson"},
+					Person{Age: 40, FirstName: "Alice", LastName: "Brown"},
+				),
+			},
+			output: createExpectedMergedRowGroup(
+				Person{Age: 25, FirstName: "John", LastName: "Doe"},
+				Person{Age: 30, FirstName: "Jane", LastName: "Smith"},
+				Person{Age: 35, FirstName: "Bob", LastName: "Johnson"},
+				Person{Age: 40, FirstName: "Alice", LastName: "Brown"},
+			),
+		},
+		{
+			scenario: "three row groups with mixed field ordering",
+			input: []parquet.RowGroup{
+				createRowGroupWithFieldOrder([]string{"FirstName", "LastName", "Age"},
+					Person{Age: 20, FirstName: "Charlie", LastName: "Wilson"},
+				),
+				createRowGroupWithFieldOrder([]string{"Age", "LastName", "FirstName"},
+					Person{Age: 25, FirstName: "David", LastName: "Taylor"},
+				),
+				createRowGroupWithFieldOrder([]string{"LastName", "FirstName", "Age"},
+					Person{Age: 30, FirstName: "Eve", LastName: "Anderson"},
+				),
+			},
+			output: createExpectedMergedRowGroup(
+				Person{Age: 20, FirstName: "Charlie", LastName: "Wilson"},
+				Person{Age: 25, FirstName: "David", LastName: "Taylor"},
+				Person{Age: 30, FirstName: "Eve", LastName: "Anderson"},
+			),
+		},
+		{
+			scenario: "nested groups with field reordering",
+			input: []parquet.RowGroup{
+				createNestedRowGroupWithFieldOrder(
+					map[string][]string{
+						"":         {"user", "metadata"},
+						"user":     {"id", "name"},
+						"metadata": {"created", "updated"},
+					},
+					map[string]any{
+						"user.id":          int64(1),
+						"user.name":        "Alice",
+						"metadata.created": "2023-01-01",
+						"metadata.updated": "2023-01-02",
+					},
+				),
+				createNestedRowGroupWithFieldOrder(
+					map[string][]string{
+						"":         {"metadata", "user"},
+						"user":     {"name", "id"},
+						"metadata": {"updated", "created"},
+					},
+					map[string]any{
+						"user.id":          int64(2),
+						"user.name":        "Bob",
+						"metadata.created": "2023-02-01",
+						"metadata.updated": "2023-02-02",
+					},
+				),
+			},
+			output: createNestedRowGroupExpected(),
+		},
+		{
+			scenario: "row groups with different field sets - missing fields become optional",
+			input: []parquet.RowGroup{
+				createPersonSubsetRowGroup([]string{"Age", "FirstName"}, // Missing LastName
+					PersonSubset1{Age: 25, FirstName: "John"},
+					PersonSubset1{Age: 30, FirstName: "Jane"},
+				),
+				createPersonSubsetRowGroup([]string{"FirstName", "LastName"}, // Missing Age
+					PersonSubset2{FirstName: "Bob", LastName: "Johnson"},
+					PersonSubset2{FirstName: "Alice", LastName: "Brown"},
+				),
+			},
+			output: createExpectedMergedWithOptionalFields(
+				Person{Age: 25, FirstName: "John", LastName: ""}, // Missing fields will be zero values
+				Person{Age: 30, FirstName: "Jane", LastName: ""},
+				Person{Age: 0, FirstName: "Bob", LastName: "Johnson"},
+				Person{Age: 0, FirstName: "Alice", LastName: "Brown"},
+			),
+		},
+	}
+
 	for _, adapter := range []struct {
 		scenario string
 		function func(parquet.RowGroup) parquet.RowGroup
@@ -337,10 +530,18 @@ func TestMergeRowGroups(t *testing.T) {
 					if merged.NumRows() != test.output.NumRows() {
 						t.Fatalf("the number of rows mismatch: want=%d got=%d", merged.NumRows(), test.output.NumRows())
 					}
-					if merged.Schema() != test.output.Schema() {
+					if !parquet.SameNodes(merged.Schema(), test.output.Schema()) {
 						t.Fatalf("the row group schemas mismatch:\n%v\n%v", test.output.Schema(), merged.Schema())
 					}
 
+					// Validate sorting columns are properly propagated
+					expectedSortingColumns := determineMergedSortingColumns(test.input, test.options)
+					actualSortingColumns := merged.SortingColumns()
+					if !equalSortingColumns(expectedSortingColumns, actualSortingColumns) {
+						t.Errorf("sorting columns not properly propagated:\nexpected: %v\nactual: %v",
+							expectedSortingColumns, actualSortingColumns)
+					}
+
 					options := []parquet.RowGroupOption{parquet.SchemaOf(Person{})}
 					options = append(options, test.options...)
 					// We test two views of the resulting row group: the one originally
@@ -416,6 +617,194 @@ func TestMergeRowGroups(t *testing.T) {
 			}
 		})
 	}
+
+	// Test field ordering scenarios
+	for _, adapter := range []struct {
+		scenario string
+		function func(parquet.RowGroup) parquet.RowGroup
+	}{
+		{scenario: "buffer", function: selfRowGroup},
+		{scenario: "file", function: fileRowGroup},
+	} {
+		t.Run("field-order-"+adapter.scenario, func(t *testing.T) {
+			for _, test := range fieldOrderTests {
+				t.Run(test.scenario, func(t *testing.T) {
+					input := make([]parquet.RowGroup, len(test.input))
+					for i := range test.input {
+						if test.input[i] != nil {
+							input[i] = adapter.function(test.input[i])
+						}
+					}
+
+					merged, err := parquet.MergeRowGroups(input)
+					if err != nil {
+						t.Fatal(err)
+					}
+
+					if merged.NumRows() != test.output.NumRows() {
+						t.Fatalf("the number of rows mismatch: want=%d got=%d", test.output.NumRows(), merged.NumRows())
+					}
+
+					// For field ordering tests, we just verify that:
+					// 1. The merge succeeded
+					// 2. We got the expected number of rows
+					// 3. The schema has the same fields (using SameNodes which ignores order)
+					// 4. We can read the data correctly
+
+					// For missing fields tests, verify schema structure manually
+					if test.scenario == "row groups with different field sets - missing fields become optional" {
+						// Verify the merged schema has correct field optionality
+						schema := merged.Schema()
+						fields := schema.Fields()
+
+						// Check that we have the expected fields
+						fieldMap := make(map[string]parquet.Field)
+						for _, field := range fields {
+							fieldMap[field.Name()] = field
+						}
+
+						// Age should be optional (missing from second input)
+						if ageField, exists := fieldMap["Age"]; !exists {
+							t.Error("Age field missing from merged schema")
+						} else if !ageField.Optional() {
+							t.Error("Age field should be optional in merged schema")
+						}
+
+						// FirstName should be required (present in both inputs)
+						if firstNameField, exists := fieldMap["FirstName"]; !exists {
+							t.Error("FirstName field missing from merged schema")
+						} else if !firstNameField.Required() {
+							t.Error("FirstName field should be required in merged schema")
+						}
+
+						// LastName should be optional (missing from first input)
+						if lastNameField, exists := fieldMap["LastName"]; !exists {
+							t.Error("LastName field missing from merged schema")
+						} else if !lastNameField.Optional() {
+							t.Error("LastName field should be optional in merged schema")
+						}
+					} else {
+						// Verify schema compatibility (field order independent)
+						if !parquet.SameNodes(merged.Schema(), test.output.Schema()) {
+							t.Fatalf("the row group schemas are not equivalent:\nmerged=%v\nexpected=%v", merged.Schema(), test.output.Schema())
+						}
+					}
+
+					// Verify we can read all rows without error
+					var mergedRows = merged.Rows()
+					var rowBuf = make([]parquet.Row, 1)
+					var readRows int64
+
+					defer mergedRows.Close()
+
+					for {
+						n, err := mergedRows.ReadRows(rowBuf)
+
+						// Process any data we got first
+						if n > 0 {
+							readRows++
+
+							// For nested tests, expect more columns
+							expectedCols := 3
+							if test.scenario == "nested groups with field reordering" {
+								expectedCols = 4 // nested structure has more columns
+							}
+
+							// Verify the row has the expected number of values
+							if len(rowBuf[0]) != expectedCols {
+								t.Errorf("row %d has wrong number of values: want=%d got=%d", readRows-1, expectedCols, len(rowBuf[0]))
+							}
+						}
+
+						// Then check for errors or end conditions
+						if err != nil {
+							if err == io.EOF {
+								break
+							}
+							t.Fatalf("error reading merged rows at index %d: %v", readRows, err)
+						}
+						if n == 0 {
+							break
+						}
+					}
+
+					if readRows != test.output.NumRows() {
+						t.Errorf("expected to read %d rows but %d were found", test.output.NumRows(), readRows)
+					}
+				})
+			}
+		})
+	}
+}
+
+// determineMergedSortingColumns calculates what the sorting columns should be
+// after merging row groups with given options
+func determineMergedSortingColumns(input []parquet.RowGroup, options []parquet.RowGroupOption) []parquet.SortingColumn {
+	// If explicit sorting is specified in options, use that
+	for _, option := range options {
+		if config, ok := option.(interface {
+			SortingColumns() []parquet.SortingColumn
+		}); ok {
+			sortingCols := config.SortingColumns()
+			if len(sortingCols) > 0 {
+				return sortingCols
+			}
+		}
+	}
+
+	// Otherwise, determine the common prefix of sorting columns from input row groups
+	if len(input) == 0 {
+		return nil
+	}
+
+	// Start with the sorting columns of the first row group
+	commonSorting := input[0].SortingColumns()
+
+	// Find the common prefix with all other row groups
+	for _, rowGroup := range input[1:] {
+		rowGroupSorting := rowGroup.SortingColumns()
+		commonSorting = commonSortingPrefix(commonSorting, rowGroupSorting)
+		if len(commonSorting) == 0 {
+			break // No common sorting
+		}
+	}
+
+	return commonSorting
+}
+
+// commonSortingPrefix returns the common prefix of two sorting column slices
+func commonSortingPrefix(a, b []parquet.SortingColumn) []parquet.SortingColumn {
+	minLen := min(len(b), len(a))
+
+	for i := range minLen {
+		if !equalSortingColumn(a[i], b[i]) {
+			return a[:i]
+		}
+	}
+
+	return a[:minLen]
+}
+
+// equalSortingColumn compares two sorting columns for equality
+func equalSortingColumn(a, b parquet.SortingColumn) bool {
+	return slices.Equal(a.Path(), b.Path()) &&
+		a.Descending() == b.Descending() &&
+		a.NullsFirst() == b.NullsFirst()
+}
+
+// equalSortingColumns compares two slices of sorting columns for equality
+func equalSortingColumns(a, b []parquet.SortingColumn) bool {
+	if len(a) != len(b) {
+		return false
+	}
+
+	for i := range a {
+		if !equalSortingColumn(a[i], b[i]) {
+			return false
+		}
+	}
+
+	return true
 }
 
 func TestMergeRowGroupsCursorsAreClosed(t *testing.T) {
@@ -591,54 +980,1023 @@ func BenchmarkMergeRowGroups(b *testing.B) {
 	}
 }
 
-func BenchmarkMergeFiles(b *testing.B) {
-	rowGroupBuffers := make([]bytes.Buffer, numRowGroups)
-
-	for _, test := range readerTests {
-		b.Run(test.scenario, func(b *testing.B) {
-			schema := parquet.SchemaOf(test.model)
-
-			sortingOptions := []parquet.SortingOption{
-				parquet.SortingColumns(
-					parquet.Ascending(schema.Columns()[0]...),
+func TestMergeNodes(t *testing.T) {
+	tests := []struct {
+		name     string
+		nodes    []parquet.Node
+		expected parquet.Node
+	}{
+		{
+			name:     "empty input",
+			nodes:    []parquet.Node{},
+			expected: nil,
+		},
+		{
+			name:     "single node",
+			nodes:    []parquet.Node{parquet.Leaf(parquet.Int32Type)},
+			expected: parquet.Required(parquet.Leaf(parquet.Int32Type)),
+		},
+		{
+			name: "merge two simple leaf nodes",
+			nodes: []parquet.Node{
+				parquet.Leaf(parquet.Int32Type),
+				parquet.Leaf(parquet.Int64Type),
+			},
+			expected: parquet.Required(parquet.Leaf(parquet.Int64Type)),
+		},
+		{
+			name: "merge nodes with compression - keep last",
+			nodes: []parquet.Node{
+				parquet.Compressed(parquet.Leaf(parquet.Int32Type), &parquet.Snappy),
+				parquet.Compressed(parquet.Leaf(parquet.Int32Type), &parquet.Gzip),
+			},
+			expected: parquet.Required(
+				parquet.Compressed(
+					parquet.Leaf(parquet.Int32Type),
+					&parquet.Gzip,
 				),
-			}
-
-			options := []parquet.RowGroupOption{
-				schema,
-				parquet.SortingRowGroupConfig(
-					sortingOptions...,
+			),
+		},
+		{
+			name: "merge nodes with encoding - keep last non-plain",
+			nodes: []parquet.Node{
+				parquet.Encoded(parquet.Leaf(parquet.Int32Type), &parquet.DeltaBinaryPacked),
+				parquet.Encoded(parquet.Leaf(parquet.Int32Type), &parquet.RLEDictionary),
+			},
+			expected: parquet.Required(
+				parquet.Encoded(
+					parquet.Leaf(parquet.Int32Type),
+					&parquet.RLEDictionary,
 				),
-			}
-
-			buffer := parquet.NewBuffer(options...)
-
-			prng := rand.New(rand.NewSource(0))
-			files := make([]*parquet.File, numRowGroups)
-			rowGroups := make([]parquet.RowGroup, numRowGroups)
-
-			for i := range files {
-				for _, row := range randomRowsOf(prng, rowsPerGroup, test.model) {
-					buffer.Write(row)
-				}
-				sort.Sort(buffer)
-				rowGroupBuffers[i].Reset()
-				writer := parquet.NewWriter(&rowGroupBuffers[i],
-					schema,
-					parquet.SortingWriterConfig(
-						sortingOptions...,
-					),
-				)
-				_, err := copyRowsAndClose(writer, buffer.Rows())
-				if err != nil {
-					b.Fatal(err)
-				}
-				if err := writer.Close(); err != nil {
-					b.Fatal(err)
-				}
-				r := bytes.NewReader(rowGroupBuffers[i].Bytes())
-				f, err := parquet.OpenFile(r, r.Size())
-				if err != nil {
+			),
+		},
+		{
+			name: "merge nodes with field IDs - keep last non-zero",
+			nodes: []parquet.Node{
+				parquet.FieldID(parquet.Leaf(parquet.Int32Type), 1),
+				parquet.FieldID(parquet.Leaf(parquet.Int32Type), 2),
+			},
+			expected: parquet.FieldID(
+				parquet.Required(parquet.Leaf(parquet.Int32Type)),
+				2,
+			),
+		},
+		{
+			name: "merge repetition types - most permissive (repeated)",
+			nodes: []parquet.Node{
+				parquet.Required(parquet.Leaf(parquet.Int32Type)),
+				parquet.Repeated(parquet.Leaf(parquet.Int32Type)),
+			},
+			expected: parquet.Repeated(parquet.Leaf(parquet.Int32Type)),
+		},
+		{
+			name: "merge repetition types - optional over required",
+			nodes: []parquet.Node{
+				parquet.Required(parquet.Leaf(parquet.Int32Type)),
+				parquet.Optional(parquet.Leaf(parquet.Int32Type)),
+			},
+			expected: parquet.Optional(parquet.Leaf(parquet.Int32Type)),
+		},
+		{
+			name: "merge complex nodes with all properties",
+			nodes: []parquet.Node{
+				parquet.FieldID(
+					parquet.Compressed(
+						parquet.Encoded(
+							parquet.Optional(parquet.Leaf(parquet.Int32Type)),
+							&parquet.DeltaBinaryPacked,
+						),
+						&parquet.Snappy,
+					),
+					1,
+				),
+				parquet.FieldID(
+					parquet.Compressed(
+						parquet.Encoded(
+							parquet.Repeated(parquet.Leaf(parquet.Int64Type)),
+							&parquet.RLEDictionary,
+						),
+						&parquet.Gzip,
+					),
+					2,
+				),
+			},
+			expected: parquet.FieldID(
+				parquet.Repeated(
+					parquet.Compressed(
+						parquet.Encoded(
+							parquet.Leaf(parquet.Int64Type),
+							&parquet.RLEDictionary,
+						),
+						&parquet.Gzip,
+					),
+				),
+				2,
+			),
+		},
+		{
+			name: "merge group nodes - union of fields",
+			nodes: []parquet.Node{
+				parquet.Group{
+					"field1": parquet.Leaf(parquet.Int32Type),
+					"field2": parquet.Leaf(parquet.ByteArrayType),
+				},
+				parquet.Group{
+					"field2": parquet.Leaf(parquet.Int64Type), // Will override
+					"field3": parquet.Leaf(parquet.FloatType),
+				},
+			},
+			expected: parquet.Required(parquet.Group{
+				"field1": parquet.Optional(parquet.Leaf(parquet.Int32Type)), // Missing from second schema
+				"field2": parquet.Required(parquet.Leaf(parquet.Int64Type)), // Present in both, overridden
+				"field3": parquet.Optional(parquet.Leaf(parquet.FloatType)), // Missing from first schema
+			}),
+		},
+		{
+			name: "merge nested group nodes",
+			nodes: []parquet.Node{
+				parquet.Group{
+					"group1": parquet.Group{
+						"nested1": parquet.Leaf(parquet.Int32Type),
+					},
+				},
+				parquet.Group{
+					"group1": parquet.Group{
+						"nested1": parquet.Leaf(parquet.Int64Type), // Will override
+						"nested2": parquet.Leaf(parquet.ByteArrayType),
+					},
+					"group2": parquet.Group{
+						"nested3": parquet.Leaf(parquet.FloatType),
+					},
+				},
+			},
+			expected: parquet.Required(parquet.Group{
+				"group1": parquet.Required(parquet.Group{
+					"nested1": parquet.Required(parquet.Leaf(parquet.Int64Type)),     // Present in both, overridden
+					"nested2": parquet.Optional(parquet.Leaf(parquet.ByteArrayType)), // Missing from first schema
+				}),
+				"group2": parquet.Optional(parquet.Group{ // Missing from first schema
+					"nested3": parquet.Required(parquet.Leaf(parquet.FloatType)),
+				}),
+			}),
+		},
+		{
+			name: "merge leaf with group - returns last",
+			nodes: []parquet.Node{
+				parquet.Leaf(parquet.Int32Type),
+				parquet.Group{
+					"field1": parquet.Leaf(parquet.ByteArrayType),
+				},
+			},
+			expected: parquet.Required(parquet.Group{
+				"field1": parquet.Required(parquet.Leaf(parquet.ByteArrayType)),
+			}),
+		},
+		{
+			name: "merge with plain encoding - should prefer non-plain",
+			nodes: []parquet.Node{
+				parquet.Encoded(parquet.Leaf(parquet.Int32Type), &parquet.Plain),
+				parquet.Encoded(parquet.Leaf(parquet.Int32Type), &parquet.DeltaBinaryPacked),
+			},
+			expected: parquet.Required(
+				parquet.Encoded(
+					parquet.Leaf(parquet.Int32Type),
+					&parquet.DeltaBinaryPacked,
+				),
+			),
+		},
+		{
+			name: "merge with mixed properties on groups",
+			nodes: []parquet.Node{
+				parquet.FieldID(
+					parquet.Optional(
+						parquet.Group{
+							"field1": parquet.Leaf(parquet.Int32Type),
+						},
+					),
+					1,
+				),
+				parquet.FieldID(
+					parquet.Repeated(
+						parquet.Group{
+							"field2": parquet.Leaf(parquet.ByteArrayType),
+						},
+					),
+					2,
+				),
+			},
+			expected: parquet.FieldID(
+				parquet.Repeated(
+					parquet.Group{
+						"field1": parquet.Optional(parquet.Leaf(parquet.Int32Type)),     // Missing from second schema
+						"field2": parquet.Optional(parquet.Leaf(parquet.ByteArrayType)), // Missing from first schema
+					},
+				),
+				2,
+			),
+		},
+		{
+			name: "merge logical map types directly",
+			nodes: []parquet.Node{
+				parquet.Map(parquet.String(), parquet.Int(32)),
+				parquet.Map(parquet.String(), parquet.Int(64)),
+			},
+			expected: parquet.Map(parquet.String(), parquet.Int(64)),
+		},
+		{
+			name: "merge complex map with field properties",
+			nodes: []parquet.Node{
+				parquet.FieldID(
+					parquet.Map(parquet.String(), parquet.Group{
+						"name": parquet.String(),
+						"age":  parquet.Int(32),
+					}),
+					10,
+				),
+				parquet.FieldID(
+					parquet.Optional(
+						parquet.Map(parquet.String(), parquet.Group{
+							"name":   parquet.String(),
+							"age":    parquet.Int(64),                   // Override age type
+							"active": parquet.Leaf(parquet.BooleanType), // Add new field
+						}),
+					),
+					20,
+				),
+			},
+			expected: parquet.FieldID(
+				parquet.Optional(
+					parquet.Map(parquet.String(), parquet.Group{
+						"name":   parquet.Required(parquet.String()),                  // Present in both schemas
+						"age":    parquet.Required(parquet.Int(64)),                   // Present in both, overridden to Int64
+						"active": parquet.Optional(parquet.Leaf(parquet.BooleanType)), // Missing from first schema
+					}),
+				),
+				20,
+			),
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			result := parquet.MergeNodes(test.nodes...)
+
+			// Handle nil cases
+			if test.expected == nil {
+				if result != nil {
+					t.Errorf("Expected nil result, got: %s", result.String())
+				}
+				return
+			}
+
+			if result == nil {
+				t.Errorf("Expected non-nil result, got nil")
+				return
+			}
+
+			if !parquet.EqualNodes(result, test.expected) {
+				t.Errorf("MergeNodes result did not match expected node")
+				t.Logf("Expected: %s", test.expected.String())
+				t.Logf("Got:      %s", result.String())
+
+				// Additional debugging for group nodes
+				if !result.Leaf() {
+					t.Logf("Result fields: %d", len(result.Fields()))
+					for _, field := range result.Fields() {
+						t.Logf("  %s: %s", field.Name(), field.String())
+					}
+				}
+				if !test.expected.Leaf() {
+					t.Logf("Expected fields: %d", len(test.expected.Fields()))
+					for _, field := range test.expected.Fields() {
+						t.Logf("  %s: %s", field.Name(), field.String())
+					}
+				}
+			}
+		})
+	}
+}
+
+// TestMergeRowGroupsWithOverlappingAndMissingFields tests merging row groups where:
+// - Each row group has one overlapping field (shared between both)
+// - Each row group has one unique field (missing from the other)
+func TestMergeRowGroupsWithOverlappingAndMissingFields(t *testing.T) {
+	// Define structs for testing overlapping and missing fields
+	type UserProfile struct {
+		UserID   int        // Overlapping field - present in both schemas
+		Username utf8string // Unique to first schema - missing from second
+	}
+
+	type UserStats struct {
+		UserID     int // Overlapping field - present in both schemas
+		LoginCount int // Unique to second schema - missing from first
+	}
+
+	type FullUser struct {
+		UserID     int
+		Username   utf8string
+		LoginCount int
+	}
+
+	// Create row groups with overlapping and missing fields
+	profileBuffer := parquet.NewBuffer()
+	profileBuffer.Write(UserProfile{UserID: 1, Username: "alice"})
+	profileBuffer.Write(UserProfile{UserID: 2, Username: "bob"})
+	profileRowGroup := profileBuffer
+
+	statsBuffer := parquet.NewBuffer()
+	statsBuffer.Write(UserStats{UserID: 3, LoginCount: 10})
+	statsBuffer.Write(UserStats{UserID: 4, LoginCount: 25})
+	statsRowGroup := statsBuffer
+
+	// Merge the row groups
+	merged, err := parquet.MergeRowGroups([]parquet.RowGroup{profileRowGroup, statsRowGroup})
+	if err != nil {
+		t.Fatalf("Failed to merge row groups: %v", err)
+	}
+
+	// Validate the merged schema
+	schema := merged.Schema()
+	fields := schema.Fields()
+	fieldMap := make(map[string]parquet.Field)
+	for _, field := range fields {
+		fieldMap[field.Name()] = field
+	}
+
+	// UserID should be required (present in both schemas)
+	if userIDField, exists := fieldMap["UserID"]; !exists {
+		t.Error("UserID field missing from merged schema")
+	} else if !userIDField.Required() {
+		t.Error("UserID field should be required in merged schema (present in both)")
+	}
+
+	// Username should be optional (missing from second schema)
+	if usernameField, exists := fieldMap["Username"]; !exists {
+		t.Error("Username field missing from merged schema")
+	} else if !usernameField.Optional() {
+		t.Error("Username field should be optional in merged schema (missing from UserStats)")
+	}
+
+	// LoginCount should be optional (missing from first schema)
+	if loginCountField, exists := fieldMap["LoginCount"]; !exists {
+		t.Error("LoginCount field missing from merged schema")
+	} else if !loginCountField.Optional() {
+		t.Error("LoginCount field should be optional in merged schema (missing from UserProfile)")
+	}
+
+	// Read and validate the merged data
+	rows := merged.Rows()
+	defer rows.Close()
+
+	readRows := make([]parquet.Row, 0, 4)
+	readBuffer := make([]parquet.Row, 10)
+	for {
+		n, err := rows.ReadRows(readBuffer)
+		for _, row := range readBuffer[:n] {
+			readRows = append(readRows, row.Clone())
+		}
+		if err == io.EOF {
+			break
+		}
+		if err != nil {
+			t.Fatalf("Failed to read merged rows: %v", err)
+		}
+	}
+	if len(readRows) != 4 {
+		t.Fatalf("Expected 4 rows, got %d", len(readRows))
+	}
+
+	// Validate the merged data by examining row structure
+	// We expect 4 rows total from merging 2+2 rows
+
+	// Validate that all rows have values for the merged schema's fields
+	schemaFields := schema.Fields()
+	expectedFieldCount := len(schemaFields)
+
+	for i, row := range readRows {
+		// Each row should have values for all fields in the merged schema
+		if len(row) != expectedFieldCount {
+			t.Errorf("Row %d has %d values, expected %d for merged schema", i, len(row), expectedFieldCount)
+		}
+
+		// Log the row structure for debugging
+		t.Logf("Row %d: %d columns", i, len(row))
+		for j, value := range row {
+			if j < len(schemaFields) {
+				fieldName := schemaFields[j].Name()
+				t.Logf("  %s: %s (column %d)", fieldName, value.String(), j)
+			}
+		}
+	}
+
+	// The key validation is that the merge succeeded and we got the right number of rows
+	// The detailed field-level validation is covered by the schema structure tests above
+
+	// Also validate that the merge process correctly handled the overlapping and missing fields:
+	// - UserProfile rows (0,1) should have UserID and Username, with LoginCount as zero/default
+	// - UserStats rows (2,3) should have UserID and LoginCount, with Username as zero/default
+	// The exact value handling depends on how parquet processes missing fields
+
+	t.Logf("Successfully merged %d rows with overlapping and missing fields", len(readRows))
+	t.Logf("Schema validation passed: UserID=required, Username=optional, LoginCount=optional")
+	t.Logf("Merge behavior: Missing fields are handled through schema conversion during merge")
+}
+
+// TestMergeFixedSizeByteArrayMinimalReproduction creates a minimal test case
+// to reproduce the FIXED_LEN_BYTE_ARRAY validation issue with different field orders
+func TestMergeFixedSizeByteArrayMinimalReproduction(t *testing.T) {
+	// Link struct for repeated field testing
+	type Link struct {
+		SpanID [8]byte `parquet:"span_id"` // Fixed[8] in repeated group
+	}
+
+	// First struct with span_id first, parent_span_id second, links third
+	type SpanRecord1 struct {
+		SpanID       [8]byte  `parquet:"span_id"`        // Required fixed[8]
+		ParentSpanID *[8]byte `parquet:"parent_span_id"` // Optional fixed[8]
+		Links        []Link   `parquet:"links"`          // Repeated group with fixed[8]
+		Name         string   `parquet:"name"`
+	}
+
+	// Second struct with different field order: parent_span_id first, span_id second, links third
+	type SpanRecord2 struct {
+		ParentSpanID *[8]byte `parquet:"parent_span_id"` // Optional fixed[8]
+		SpanID       [8]byte  `parquet:"span_id"`        // Required fixed[8]
+		Links        []Link   `parquet:"links"`          // Repeated group with fixed[8]
+		Name         string   `parquet:"name"`
+	}
+
+	// Create test data with repeated links to exercise the third reordering condition
+	spanID1 := [8]byte{0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07, 0x08}
+	parentSpanID1 := [8]byte{0x11, 0x12, 0x13, 0x14, 0x15, 0x16, 0x17, 0x18}
+	linkSpanID1 := [8]byte{0x31, 0x32, 0x33, 0x34, 0x35, 0x36, 0x37, 0x38}
+
+	spanID2 := [8]byte{0x21, 0x22, 0x23, 0x24, 0x25, 0x26, 0x27, 0x28}
+	linkSpanID2 := [8]byte{0x41, 0x42, 0x43, 0x44, 0x45, 0x46, 0x47, 0x48}
+	// parentSpanID2 is nil (optional field)
+
+	records1 := []SpanRecord1{
+		{
+			SpanID:       spanID1,
+			ParentSpanID: &parentSpanID1,
+			Links:        []Link{{SpanID: linkSpanID1}}, // Non-empty links
+			Name:         "span1",
+		},
+	}
+
+	records2 := []SpanRecord2{
+		{
+			SpanID:       spanID2,
+			ParentSpanID: nil,                           // nil parent_span_id
+			Links:        []Link{{SpanID: linkSpanID2}}, // Non-empty links
+			Name:         "span2",
+		},
+	}
+
+	// Create first row group
+	buffer1 := &bytes.Buffer{}
+	writer1 := parquet.NewGenericWriter[SpanRecord1](buffer1)
+	_, err := writer1.Write(records1)
+	if err != nil {
+		t.Fatalf("Failed to write records1: %v", err)
+	}
+	writer1.Close()
+
+	// Create second row group
+	buffer2 := &bytes.Buffer{}
+	writer2 := parquet.NewGenericWriter[SpanRecord2](buffer2)
+	_, err = writer2.Write(records2)
+	if err != nil {
+		t.Fatalf("Failed to write records2: %v", err)
+	}
+	writer2.Close()
+
+	// Read back the row groups
+	reader1 := parquet.NewReader(bytes.NewReader(buffer1.Bytes()))
+	reader2 := parquet.NewReader(bytes.NewReader(buffer2.Bytes()))
+
+	file1 := reader1.File()
+	file2 := reader2.File()
+
+	schema1 := file1.Schema()
+	schema2 := file2.Schema()
+
+	fmt.Printf("\nMinimal reproduction test:\n")
+	fmt.Printf("Schema 1 columns: %d\n", len(schema1.Columns()))
+	fmt.Printf("Schema 2 columns: %d\n", len(schema2.Columns()))
+
+	// Print field order for both schemas
+	for i, columnPath := range schema1.Columns() {
+		leaf, found := schema1.Lookup(columnPath...)
+		if found {
+			fmt.Printf("Schema1 col %d: %v, type: %s\n", i, columnPath, leaf.Node.Type())
+		}
+	}
+
+	for i, columnPath := range schema2.Columns() {
+		leaf, found := schema2.Lookup(columnPath...)
+		if found {
+			fmt.Printf("Schema2 col %d: %v, type: %s\n", i, columnPath, leaf.Node.Type())
+		}
+	}
+
+	// Create row groups and merge them - this will automatically merge schemas
+	rowGroup1 := file1.RowGroups()[0]
+	rowGroup2 := file2.RowGroups()[0]
+
+	fmt.Printf("\nAttempting to merge row groups...\n")
+	mergedRowGroup, err := parquet.MergeRowGroups([]parquet.RowGroup{rowGroup1, rowGroup2})
+	if err != nil {
+		t.Fatalf("Failed to merge row groups: %v", err)
+	}
+
+	// Get the merged schema from the merged row group
+	mergedSchema := mergedRowGroup.Schema()
+	fmt.Printf("Merged schema columns: %d\n", len(mergedSchema.Columns()))
+
+	// Print merged field order
+	for i, columnPath := range mergedSchema.Columns() {
+		leaf, found := mergedSchema.Lookup(columnPath...)
+		if found {
+			fmt.Printf("Merged col %d: %v, type: %s\n", i, columnPath, leaf.Node.Type())
+		}
+	}
+
+	// Try to write the merged result
+	buffer3 := &bytes.Buffer{}
+	writer3 := parquet.NewWriter(buffer3, mergedSchema)
+
+	rows := mergedRowGroup.Rows()
+	defer rows.Close()
+
+	// Use CopyRows for efficient copying
+	fmt.Printf("\nCopying rows from merged row group to writer...\n")
+
+	// Debug: Let's manually read a few rows to see what's wrong
+	rowBuf := make([]parquet.Row, 1)
+	for i := range 3 {
+		n, err := rows.ReadRows(rowBuf)
+		if err != nil && err != io.EOF {
+			t.Fatalf("Failed to read debug row %d: %v", i, err)
+		}
+		if n == 0 {
+			break
+		}
+
+		row := rowBuf[0]
+		fmt.Printf("Debug row %d has %d values:\n", i, len(row))
+		for j, val := range row {
+			fmt.Printf("  [%d] col=%d, kind=%s, bytes=%d, value=%q\n",
+				j, val.Column(), val.Kind(), len(val.ByteArray()), val.String())
+			if j >= 10 { // Limit output
+				break
+			}
+		}
+	}
+	rows.Close()
+
+	// Now try the actual copy
+	numCopied, err := parquet.CopyRows(writer3, mergedRowGroup.Rows())
+	if err != nil {
+		t.Fatalf("Failed to copy rows: %v", err)
+	}
+
+	writer3.Close()
+	fmt.Printf("Successfully merged %d rows\n", numCopied)
+}
+
+// TestMergeFixedSizeByteArrayNullValues tests merging with null fixed-size byte arrays
+// This reproduces the specific issue where all parent_span_id and links.span_id are null
+func TestMergeFixedSizeByteArrayNullValues(t *testing.T) {
+	// Link struct for repeated field testing
+	type Link struct {
+		SpanID [8]byte `parquet:"span_id"` // Fixed[8] in repeated group - will be null
+	}
+
+	// First struct with span_id first, parent_span_id second, links third
+	type SpanRecord1 struct {
+		SpanID       [8]byte  `parquet:"span_id"`        // Required fixed[8] - NOT null
+		ParentSpanID *[8]byte `parquet:"parent_span_id"` // Optional fixed[8] - NULL
+		Links        []Link   `parquet:"links"`          // Repeated group with fixed[8] - EMPTY (so span_id is null)
+		Name         string   `parquet:"name"`
+	}
+
+	// Second struct with different field order
+	type SpanRecord2 struct {
+		ParentSpanID *[8]byte `parquet:"parent_span_id"` // Optional fixed[8] - NULL
+		SpanID       [8]byte  `parquet:"span_id"`        // Required fixed[8] - NOT null
+		Links        []Link   `parquet:"links"`          // Repeated group with fixed[8] - EMPTY (so span_id is null)
+		Name         string   `parquet:"name"`
+	}
+
+	// Create test data with NULL values matching the problematic parquet files
+	spanID1 := [8]byte{0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07, 0x08}
+	spanID2 := [8]byte{0x21, 0x22, 0x23, 0x24, 0x25, 0x26, 0x27, 0x28}
+
+	records1 := []SpanRecord1{
+		{
+			SpanID:       spanID1,
+			ParentSpanID: nil,      // NULL parent_span_id (like in the problematic files)
+			Links:        []Link{}, // EMPTY links (so all link span_ids are null)
+			Name:         "span1",
+		},
+	}
+
+	records2 := []SpanRecord2{
+		{
+			SpanID:       spanID2,
+			ParentSpanID: nil,      // NULL parent_span_id (like in the problematic files)
+			Links:        []Link{}, // EMPTY links (so all link span_ids are null)
+			Name:         "span2",
+		},
+	}
+
+	// Create first row group
+	buffer1 := &bytes.Buffer{}
+	writer1 := parquet.NewGenericWriter[SpanRecord1](buffer1)
+	_, err := writer1.Write(records1)
+	if err != nil {
+		t.Fatalf("Failed to write records1: %v", err)
+	}
+	writer1.Close()
+
+	// Create second row group
+	buffer2 := &bytes.Buffer{}
+	writer2 := parquet.NewGenericWriter[SpanRecord2](buffer2)
+	_, err = writer2.Write(records2)
+	if err != nil {
+		t.Fatalf("Failed to write records2: %v", err)
+	}
+	writer2.Close()
+
+	// Read back the row groups
+	reader1 := parquet.NewReader(bytes.NewReader(buffer1.Bytes()))
+	reader2 := parquet.NewReader(bytes.NewReader(buffer2.Bytes()))
+
+	file1 := reader1.File()
+	file2 := reader2.File()
+
+	schema1 := file1.Schema()
+	schema2 := file2.Schema()
+
+	fmt.Printf("\nNull values test:\n")
+	fmt.Printf("Schema 1 columns: %d\n", len(schema1.Columns()))
+	fmt.Printf("Schema 2 columns: %d\n", len(schema2.Columns()))
+
+	// Print field order for both schemas
+	for i, columnPath := range schema1.Columns() {
+		leaf, found := schema1.Lookup(columnPath...)
+		if found {
+			fmt.Printf("Schema1 col %d: %v, type: %s\n", i, columnPath, leaf.Node.Type())
+		}
+	}
+
+	for i, columnPath := range schema2.Columns() {
+		leaf, found := schema2.Lookup(columnPath...)
+		if found {
+			fmt.Printf("Schema2 col %d: %v, type: %s\n", i, columnPath, leaf.Node.Type())
+		}
+	}
+
+	// Create row groups and merge them - this should trigger the null value issue
+	rowGroup1 := file1.RowGroups()[0]
+	rowGroup2 := file2.RowGroups()[0]
+
+	fmt.Printf("\nAttempting to merge row groups with null fixed-size byte arrays...\n")
+	mergedRowGroup, err := parquet.MergeRowGroups([]parquet.RowGroup{rowGroup1, rowGroup2})
+	if err != nil {
+		t.Fatalf("Failed to merge row groups: %v", err)
+	}
+
+	// Get the merged schema from the merged row group
+	mergedSchema := mergedRowGroup.Schema()
+	fmt.Printf("Merged schema columns: %d\n", len(mergedSchema.Columns()))
+
+	// Print merged field order
+	for i, columnPath := range mergedSchema.Columns() {
+		leaf, found := mergedSchema.Lookup(columnPath...)
+		if found {
+			fmt.Printf("Merged col %d: %v, type: %s\n", i, columnPath, leaf.Node.Type())
+		}
+	}
+
+	// Try to copy rows - this should reproduce the index out of range panic
+	buffer3 := &bytes.Buffer{}
+	writer3 := parquet.NewWriter(buffer3, mergedSchema)
+
+	rows := mergedRowGroup.Rows()
+	defer rows.Close()
+
+	// This is where we expect the panic due to null value handling issues
+	fmt.Printf("\nCopying rows from merged row group to writer...\n")
+	numCopied, err := parquet.CopyRows(writer3, rows)
+	if err != nil {
+		t.Fatalf("Failed to copy rows: %v", err)
+	}
+
+	writer3.Close()
+	fmt.Printf("Successfully merged %d rows\n", numCopied)
+}
+
+func BenchmarkMergeNodes(b *testing.B) {
+	// Create test nodes for benchmarking
+	nodes := []parquet.Node{
+		parquet.FieldID(
+			parquet.Compressed(
+				parquet.Encoded(
+					parquet.Optional(parquet.Leaf(parquet.Int32Type)),
+					&parquet.DeltaBinaryPacked,
+				),
+				&parquet.Snappy,
+			),
+			1,
+		),
+		parquet.FieldID(
+			parquet.Compressed(
+				parquet.Encoded(
+					parquet.Repeated(parquet.Leaf(parquet.Int64Type)),
+					&parquet.RLEDictionary,
+				),
+				&parquet.Gzip,
+			),
+			2,
+		),
+		parquet.Group{
+			"field1": parquet.Leaf(parquet.ByteArrayType),
+			"field2": parquet.Leaf(parquet.FloatType),
+		},
+	}
+
+	b.ResetTimer()
+	for i := 0; i < b.N; i++ {
+		_ = parquet.MergeNodes(nodes...)
+	}
+}
+
+// Helper functions for field ordering tests
+
+// Define struct types with same fields but different declaration order
+type PersonAgeFirst struct {
+	Age       int
+	FirstName utf8string
+	LastName  utf8string
+}
+
+type PersonLastNameFirst struct {
+	LastName  utf8string
+	Age       int
+	FirstName utf8string
+}
+
+type PersonFirstNameFirst struct {
+	FirstName utf8string
+	LastName  utf8string
+	Age       int
+}
+
+// Struct types for testing missing fields
+type PersonSubset1 struct {
+	Age       int
+	FirstName utf8string
+	// Missing LastName
+}
+
+type PersonSubset2 struct {
+	FirstName utf8string
+	LastName  utf8string
+	// Missing Age
+}
+
+// createRowGroupWithFieldOrder creates a row group using different struct orders
+func createRowGroupWithFieldOrder(fieldOrder []string, rows ...Person) parquet.RowGroup {
+	switch {
+	case len(fieldOrder) >= 1 && fieldOrder[0] == "Age":
+		// Age first
+		buf := parquet.NewBuffer()
+		for _, row := range rows {
+			buf.Write(PersonAgeFirst{
+				Age:       row.Age,
+				FirstName: row.FirstName,
+				LastName:  row.LastName,
+			})
+		}
+		return buf
+	case len(fieldOrder) >= 1 && fieldOrder[0] == "LastName":
+		// LastName first
+		buf := parquet.NewBuffer()
+		for _, row := range rows {
+			buf.Write(PersonLastNameFirst{
+				LastName:  row.LastName,
+				Age:       row.Age,
+				FirstName: row.FirstName,
+			})
+		}
+		return buf
+	case len(fieldOrder) >= 1 && fieldOrder[0] == "FirstName":
+		// FirstName first
+		buf := parquet.NewBuffer()
+		for _, row := range rows {
+			buf.Write(PersonFirstNameFirst{
+				FirstName: row.FirstName,
+				LastName:  row.LastName,
+				Age:       row.Age,
+			})
+		}
+		return buf
+	default:
+		// Default order
+		buf := parquet.NewBuffer()
+		for _, row := range rows {
+			buf.Write(row)
+		}
+		return buf
+	}
+}
+
+// createPersonSubsetRowGroup creates a row group with only a subset of Person fields
+func createPersonSubsetRowGroup(fields []string, rows ...any) parquet.RowGroup {
+	buf := parquet.NewBuffer()
+	for _, row := range rows {
+		buf.Write(row)
+	}
+	return buf
+}
+
+// Define nested struct types with different field orders
+type UserType1 struct {
+	ID   int64
+	Name utf8string
+}
+
+type UserType2 struct {
+	Name utf8string
+	ID   int64
+}
+
+type MetadataType1 struct {
+	Created utf8string
+	Updated utf8string
+}
+
+type MetadataType2 struct {
+	Updated utf8string
+	Created utf8string
+}
+
+type NestedType1 struct {
+	User     UserType1
+	Metadata MetadataType1
+}
+
+type NestedType2 struct {
+	Metadata MetadataType2
+	User     UserType2
+}
+
+// createNestedRowGroupWithFieldOrder creates a nested row group using different struct orders
+func createNestedRowGroupWithFieldOrder(fieldOrders map[string][]string, values map[string]any) parquet.RowGroup {
+	userID := values["user.id"].(int64)
+	userName := utf8string(values["user.name"].(string))
+	created := utf8string(values["metadata.created"].(string))
+	updated := utf8string(values["metadata.updated"].(string))
+
+	// Check root field order to determine which struct type to use
+	if rootOrder, exists := fieldOrders[""]; exists && len(rootOrder) > 0 && rootOrder[0] == "user" {
+		// User first
+		buf := parquet.NewBuffer()
+		buf.Write(NestedType1{
+			User: UserType1{
+				ID:   userID,
+				Name: userName,
+			},
+			Metadata: MetadataType1{
+				Created: created,
+				Updated: updated,
+			},
+		})
+		return buf
+	} else {
+		// Metadata first
+		buf := parquet.NewBuffer()
+		buf.Write(NestedType2{
+			Metadata: MetadataType2{
+				Updated: updated,
+				Created: created,
+			},
+			User: UserType2{
+				Name: userName,
+				ID:   userID,
+			},
+		})
+		return buf
+	}
+}
+
+// createNestedRowGroupExpected creates the expected result for nested field ordering test
+func createNestedRowGroupExpected() parquet.RowGroup {
+	// Use one of the struct types as the expected format - after merge,
+	// the schemas should be equivalent regardless of original field order
+	buf := parquet.NewBuffer()
+
+	// Add expected rows using consistent struct type
+	buf.Write(NestedType1{
+		User: UserType1{
+			ID:   int64(1),
+			Name: utf8string("Alice"),
+		},
+		Metadata: MetadataType1{
+			Created: utf8string("2023-01-01"),
+			Updated: utf8string("2023-01-02"),
+		},
+	})
+	buf.Write(NestedType1{
+		User: UserType1{
+			ID:   int64(2),
+			Name: utf8string("Bob"),
+		},
+		Metadata: MetadataType1{
+			Created: utf8string("2023-02-01"),
+			Updated: utf8string("2023-02-02"),
+		},
+	})
+
+	return buf
+}
+
+// createExpectedMergedRowGroup creates a simple expected result for comparison
+// For field ordering tests, we just need to verify the merge works and produces the right number of rows
+// The actual field order will be handled by SameNodes comparison
+func createExpectedMergedRowGroup(rows ...Person) parquet.RowGroup {
+	// Just create a simple row group with the same data
+	// The test will use SameNodes for schema comparison which handles field order differences
+	buf := parquet.NewBuffer()
+	for _, row := range rows {
+		buf.Write(row)
+	}
+	return buf
+}
+
+// createExpectedMergedWithOptionalFields creates expected result with optional fields
+// For this test case, we just create a simple reference that will be compared using SameNodes
+// The actual validation is that the merge succeeds and uses the nullable() function correctly
+func createExpectedMergedWithOptionalFields(rows ...Person) parquet.RowGroup {
+	// Just create a simple row group - the test will validate the actual schema using SameNodes
+	// and check that the merge operation succeeds with the correct optional field handling
+	buf := parquet.NewBuffer()
+	for _, row := range rows {
+		buf.Write(row)
+	}
+	return buf
+}
+
+func BenchmarkMergeFiles(b *testing.B) {
+	rowGroupBuffers := make([]bytes.Buffer, numRowGroups)
+
+	for _, test := range readerTests {
+		b.Run(test.scenario, func(b *testing.B) {
+			schema := parquet.SchemaOf(test.model)
+
+			sortingOptions := []parquet.SortingOption{
+				parquet.SortingColumns(
+					parquet.Ascending(schema.Columns()[0]...),
+				),
+			}
+
+			options := []parquet.RowGroupOption{
+				schema,
+				parquet.SortingRowGroupConfig(
+					sortingOptions...,
+				),
+			}
+
+			buffer := parquet.NewBuffer(options...)
+
+			prng := rand.New(rand.NewSource(0))
+			files := make([]*parquet.File, numRowGroups)
+			rowGroups := make([]parquet.RowGroup, numRowGroups)
+
+			for i := range files {
+				for _, row := range randomRowsOf(prng, rowsPerGroup, test.model) {
+					buffer.Write(row)
+				}
+				sort.Sort(buffer)
+				rowGroupBuffers[i].Reset()
+				writer := parquet.NewWriter(&rowGroupBuffers[i],
+					schema,
+					parquet.SortingWriterConfig(
+						sortingOptions...,
+					),
+				)
+				_, err := copyRowsAndClose(writer, buffer.Rows())
+				if err != nil {
+					b.Fatal(err)
+				}
+				if err := writer.Close(); err != nil {
+					b.Fatal(err)
+				}
+				r := bytes.NewReader(rowGroupBuffers[i].Bytes())
+				f, err := parquet.OpenFile(r, r.Size())
+				if err != nil {
 					b.Fatal(err)
 				}
 				files[i], rowGroups[i] = f, f.RowGroups()[0]
@@ -676,3 +2034,151 @@ func BenchmarkMergeFiles(b *testing.B) {
 		})
 	}
 }
+
+func TestMergeSortingColumns(t *testing.T) {
+	tests := []struct {
+		name     string
+		input    [][]parquet.SortingColumn
+		expected []parquet.SortingColumn
+	}{
+		{
+			name:     "empty input",
+			input:    [][]parquet.SortingColumn{},
+			expected: nil,
+		},
+		{
+			name:     "single empty slice",
+			input:    [][]parquet.SortingColumn{{}},
+			expected: nil,
+		},
+		{
+			name:     "one slice with empty slice",
+			input:    [][]parquet.SortingColumn{{parquet.Ascending("A")}, {}},
+			expected: nil,
+		},
+		{
+			name: "identical single column",
+			input: [][]parquet.SortingColumn{
+				{parquet.Ascending("A")},
+				{parquet.Ascending("A")},
+			},
+			expected: []parquet.SortingColumn{parquet.Ascending("A")},
+		},
+		{
+			name: "identical multi-column",
+			input: [][]parquet.SortingColumn{
+				{parquet.Ascending("A"), parquet.Descending("B")},
+				{parquet.Ascending("A"), parquet.Descending("B")},
+			},
+			expected: []parquet.SortingColumn{parquet.Ascending("A"), parquet.Descending("B")},
+		},
+		{
+			name: "common prefix - partial match",
+			input: [][]parquet.SortingColumn{
+				{parquet.Ascending("A"), parquet.Ascending("B"), parquet.Descending("C")},
+				{parquet.Ascending("A"), parquet.Ascending("B"), parquet.Ascending("D")},
+			},
+			expected: []parquet.SortingColumn{parquet.Ascending("A"), parquet.Ascending("B")},
+		},
+		{
+			name: "no common prefix - different first column",
+			input: [][]parquet.SortingColumn{
+				{parquet.Ascending("A"), parquet.Ascending("B")},
+				{parquet.Ascending("X"), parquet.Ascending("B")},
+			},
+			expected: []parquet.SortingColumn{},
+		},
+		{
+			name: "different direction on same column",
+			input: [][]parquet.SortingColumn{
+				{parquet.Ascending("A"), parquet.Ascending("B")},
+				{parquet.Descending("A"), parquet.Ascending("B")},
+			},
+			expected: []parquet.SortingColumn{},
+		},
+		{
+			name: "different nulls first setting",
+			input: [][]parquet.SortingColumn{
+				{parquet.NullsFirst(parquet.Ascending("A"))},
+				{parquet.Ascending("A")}, // defaults to nulls last
+			},
+			expected: []parquet.SortingColumn{},
+		},
+		{
+			name: "three inputs with common prefix",
+			input: [][]parquet.SortingColumn{
+				{parquet.Ascending("A"), parquet.Ascending("B"), parquet.Descending("C")},
+				{parquet.Ascending("A"), parquet.Ascending("B"), parquet.Ascending("D")},
+				{parquet.Ascending("A"), parquet.Ascending("B"), parquet.Descending("E")},
+			},
+			expected: []parquet.SortingColumn{parquet.Ascending("A"), parquet.Ascending("B")},
+		},
+		{
+			name: "unequal length inputs - shorter wins",
+			input: [][]parquet.SortingColumn{
+				{parquet.Ascending("A"), parquet.Ascending("B")},
+				{parquet.Ascending("A")},
+				{parquet.Ascending("A"), parquet.Ascending("B"), parquet.Descending("C")},
+			},
+			expected: []parquet.SortingColumn{parquet.Ascending("A")},
+		},
+		{
+			name: "complex paths",
+			input: [][]parquet.SortingColumn{
+				{parquet.Ascending("nested.field.a")},
+				{parquet.Ascending("nested.field.a")},
+			},
+			expected: []parquet.SortingColumn{parquet.Ascending("nested.field.a")},
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			result := parquet.MergeSortingColumns(test.input...)
+
+			if !equalSortingColumnsSlices(result, test.expected) {
+				t.Errorf("MergeSortingColumns() = %v, expected %v", result, test.expected)
+				t.Logf("Result length: %d, Expected length: %d", len(result), len(test.expected))
+				for i, col := range result {
+					t.Logf("  Result[%d]: Path=%v, Desc=%v, NullsFirst=%v", i, col.Path(), col.Descending(), col.NullsFirst())
+				}
+				for i, col := range test.expected {
+					t.Logf("  Expected[%d]: Path=%v, Desc=%v, NullsFirst=%v", i, col.Path(), col.Descending(), col.NullsFirst())
+				}
+			}
+		})
+	}
+}
+
+// Helper function to compare slices of sorting columns
+func equalSortingColumnsSlices(a, b []parquet.SortingColumn) bool {
+	if len(a) != len(b) {
+		return false
+	}
+
+	for i := range a {
+		if !equalSortingColumnsTest(a[i], b[i]) {
+			return false
+		}
+	}
+
+	return true
+}
+
+// Helper function to compare two sorting columns
+func equalSortingColumnsTest(a, b parquet.SortingColumn) bool {
+	aPath := a.Path()
+	bPath := b.Path()
+
+	if len(aPath) != len(bPath) {
+		return false
+	}
+
+	for i, pathElement := range aPath {
+		if pathElement != bPath[i] {
+			return false
+		}
+	}
+
+	return a.Descending() == b.Descending() && a.NullsFirst() == b.NullsFirst()
+}
diff --git a/node_test.go b/node_test.go
index 1b395d2..2bc58db 100644
--- a/node_test.go
+++ b/node_test.go
@@ -5,6 +5,8 @@ import (
 	"testing"
 
 	"github.com/parquet-go/parquet-go/compress"
+	"github.com/parquet-go/parquet-go/compress/gzip"
+	"github.com/parquet-go/parquet-go/compress/snappy"
 	"github.com/parquet-go/parquet-go/encoding"
 )
 
@@ -34,6 +36,665 @@ func (m *mockNode) Encoding() encoding.Encoding { return m.enc() }
 func (m *mockNode) Compression() compress.Codec { return m.compression() }
 func (m *mockNode) GoType() reflect.Type        { return m.goType() }
 
+func TestEqualNodes(t *testing.T) {
+	tests := []struct {
+		name     string
+		node1    Node
+		node2    Node
+		expected bool
+	}{
+		// Basic leaf node equality
+		{
+			name:     "same leaf nodes - boolean",
+			node1:    Leaf(BooleanType),
+			node2:    Leaf(BooleanType),
+			expected: true,
+		},
+		{
+			name:     "same leaf nodes - int32",
+			node1:    Leaf(Int32Type),
+			node2:    Leaf(Int32Type),
+			expected: true,
+		},
+		{
+			name:     "same leaf nodes - string",
+			node1:    String(),
+			node2:    String(),
+			expected: true,
+		},
+		{
+			name:     "different leaf types - int32 vs int64",
+			node1:    Leaf(Int32Type),
+			node2:    Leaf(Int64Type),
+			expected: false,
+		},
+		{
+			name:     "different leaf types - string vs json",
+			node1:    String(),
+			node2:    JSON(),
+			expected: false,
+		},
+
+		// Repetition type differences
+		{
+			name:     "same types, different repetition - required vs optional",
+			node1:    Required(String()),
+			node2:    Optional(String()),
+			expected: false,
+		},
+		{
+			name:     "same types, different repetition - optional vs repeated",
+			node1:    Optional(String()),
+			node2:    Repeated(String()),
+			expected: false,
+		},
+		{
+			name:     "same types, same repetition - both optional",
+			node1:    Optional(String()),
+			node2:    Optional(String()),
+			expected: true,
+		},
+		{
+			name:     "same types, same repetition - both repeated",
+			node1:    Repeated(Int(32)),
+			node2:    Repeated(Int(32)),
+			expected: true,
+		},
+
+		// Logical types
+		{
+			name:     "same logical types - int32",
+			node1:    Int(32),
+			node2:    Int(32),
+			expected: true,
+		},
+		{
+			name:     "same logical types - uint64",
+			node1:    Uint(64),
+			node2:    Uint(64),
+			expected: true,
+		},
+		{
+			name:     "different logical types - int32 vs uint32",
+			node1:    Int(32),
+			node2:    Uint(32),
+			expected: false,
+		},
+		{
+			name:     "same temporal types - date",
+			node1:    Date(),
+			node2:    Date(),
+			expected: true,
+		},
+		{
+			name:     "same temporal types - timestamp millis",
+			node1:    Timestamp(Millisecond),
+			node2:    Timestamp(Millisecond),
+			expected: true,
+		},
+		{
+			name:     "different temporal types - timestamp units",
+			node1:    Timestamp(Millisecond),
+			node2:    Timestamp(Microsecond),
+			expected: false,
+		},
+		{
+			name:     "same decimal types",
+			node1:    Decimal(10, 2, Int32Type),
+			node2:    Decimal(10, 2, Int32Type),
+			expected: true,
+		},
+		{
+			name:     "different decimal precision",
+			node1:    Decimal(10, 2, Int32Type),
+			node2:    Decimal(12, 2, Int32Type),
+			expected: false,
+		},
+
+		// Encoding and compression should be ignored
+		{
+			name:     "same types, different encoding (should be equal)",
+			node1:    Encoded(String(), &DeltaLengthByteArray),
+			node2:    Encoded(String(), &Plain),
+			expected: true,
+		},
+		{
+			name:     "same types, different compression (should be equal)",
+			node1:    Compressed(String(), &gzip.Codec{}),
+			node2:    Compressed(String(), &snappy.Codec{}),
+			expected: true,
+		},
+
+		// Field IDs should be ignored
+		{
+			name:     "same types, different field IDs (should be equal)",
+			node1:    FieldID(String(), 1),
+			node2:    FieldID(String(), 2),
+			expected: true,
+		},
+
+		// Group nodes - empty groups
+		{
+			name:     "empty groups",
+			node1:    Group{},
+			node2:    Group{},
+			expected: true,
+		},
+
+		// Group nodes - single field
+		{
+			name: "groups with same single field",
+			node1: Group{
+				"name": String(),
+			},
+			node2: Group{
+				"name": String(),
+			},
+			expected: true,
+		},
+		{
+			name: "groups with different field names",
+			node1: Group{
+				"name": String(),
+			},
+			node2: Group{
+				"title": String(),
+			},
+			expected: false,
+		},
+		{
+			name: "groups with same field name, different types",
+			node1: Group{
+				"name": String(),
+			},
+			node2: Group{
+				"name": Int(32),
+			},
+			expected: false,
+		},
+
+		// Group nodes - multiple fields
+		{
+			name: "groups with same multiple fields",
+			node1: Group{
+				"id":   Int(64),
+				"name": String(),
+				"age":  Int(32),
+			},
+			node2: Group{
+				"id":   Int(64),
+				"name": String(),
+				"age":  Int(32),
+			},
+			expected: true,
+		},
+		{
+			name: "groups with different field count",
+			node1: Group{
+				"id":   Int(64),
+				"name": String(),
+			},
+			node2: Group{
+				"id":   Int(64),
+				"name": String(),
+				"age":  Int(32),
+			},
+			expected: false,
+		},
+
+		// Group nodes with repetition
+		{
+			name: "groups with same repetition",
+			node1: Optional(Group{
+				"name": String(),
+			}),
+			node2: Optional(Group{
+				"name": String(),
+			}),
+			expected: true,
+		},
+		{
+			name: "groups with different repetition",
+			node1: Required(Group{
+				"name": String(),
+			}),
+			node2: Optional(Group{
+				"name": String(),
+			}),
+			expected: false,
+		},
+
+		// Nested groups
+		{
+			name: "nested groups - same structure",
+			node1: Group{
+				"person": Group{
+					"name": String(),
+					"age":  Int(32),
+				},
+			},
+			node2: Group{
+				"person": Group{
+					"name": String(),
+					"age":  Int(32),
+				},
+			},
+			expected: true,
+		},
+		{
+			name: "nested groups - different inner structure",
+			node1: Group{
+				"person": Group{
+					"name": String(),
+					"age":  Int(32),
+				},
+			},
+			node2: Group{
+				"person": Group{
+					"name": String(),
+					"age":  Int(64),
+				},
+			},
+			expected: false,
+		},
+
+		// Mixed leaf and group comparisons
+		{
+			name:     "leaf vs group",
+			node1:    String(),
+			node2:    Group{"name": String()},
+			expected: false,
+		},
+		{
+			name:     "group vs leaf",
+			node1:    Group{"name": String()},
+			node2:    String(),
+			expected: false,
+		},
+
+		// Complex nested structures
+		{
+			name: "complex nested structure - same",
+			node1: Group{
+				"user": Group{
+					"id": Int(64),
+					"profile": Group{
+						"name":  String(),
+						"email": String(),
+						"settings": Group{
+							"theme":         String(),
+							"notifications": Repeated(String()),
+						},
+					},
+				},
+				"metadata": Optional(Group{
+					"created_at": Timestamp(Millisecond),
+					"tags":       Repeated(String()),
+				}),
+			},
+			node2: Group{
+				"user": Group{
+					"id": Int(64),
+					"profile": Group{
+						"name":  String(),
+						"email": String(),
+						"settings": Group{
+							"theme":         String(),
+							"notifications": Repeated(String()),
+						},
+					},
+				},
+				"metadata": Optional(Group{
+					"created_at": Timestamp(Millisecond),
+					"tags":       Repeated(String()),
+				}),
+			},
+			expected: true,
+		},
+		{
+			name: "complex nested structure - different deep field",
+			node1: Group{
+				"user": Group{
+					"id": Int(64),
+					"profile": Group{
+						"name":  String(),
+						"email": String(),
+						"settings": Group{
+							"theme":         String(),
+							"notifications": Repeated(String()),
+						},
+					},
+				},
+			},
+			node2: Group{
+				"user": Group{
+					"id": Int(64),
+					"profile": Group{
+						"name":  String(),
+						"email": String(),
+						"settings": Group{
+							"theme":         String(),
+							"notifications": Repeated(Int(32)), // Different type here
+						},
+					},
+				},
+			},
+			expected: false,
+		},
+
+		// Map types (group types with MAP logical type)
+		{
+			name:     "same map types",
+			node1:    Map(String(), Int(32)),
+			node2:    Map(String(), Int(32)),
+			expected: true,
+		},
+		{
+			name:     "different map key types",
+			node1:    Map(String(), Int(32)),
+			node2:    Map(Int(32), Int(32)),
+			expected: false,
+		},
+		{
+			name:     "different map value types",
+			node1:    Map(String(), Int(32)),
+			node2:    Map(String(), String()),
+			expected: false,
+		},
+
+		// List types (group types with LIST logical type)
+		{
+			name:     "same list types",
+			node1:    List(String()),
+			node2:    List(String()),
+			expected: true,
+		},
+		{
+			name:     "different list element types",
+			node1:    List(String()),
+			node2:    List(Int(32)),
+			expected: false,
+		},
+
+		// Logical types vs non-logical groups with same structure
+		{
+			name:  "logical map vs regular group with same structure",
+			node1: Map(String(), Int(32)),
+			node2: Group{
+				"key_value": Repeated(Group{
+					"key":   String(),
+					"value": Int(32),
+				}),
+			},
+			expected: false,
+		},
+		{
+			name:  "logical list vs regular group with same structure",
+			node1: List(String()),
+			node2: Group{
+				"list": Repeated(Group{
+					"element": String(),
+				}),
+			},
+			expected: false,
+		},
+		{
+			name: "regular group vs logical map with same structure",
+			node1: Group{
+				"key_value": Repeated(Group{
+					"key":   String(),
+					"value": Int(32),
+				}),
+			},
+			node2:    Map(String(), Int(32)),
+			expected: false,
+		},
+		{
+			name: "regular group vs logical list with same structure",
+			node1: Group{
+				"list": Repeated(Group{
+					"element": String(),
+				}),
+			},
+			node2:    List(String()),
+			expected: false,
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			result := EqualNodes(test.node1, test.node2)
+			if result != test.expected {
+				t.Errorf("EqualNodes(%v, %v) = %v, expected %v",
+					test.node1, test.node2, result, test.expected)
+			}
+		})
+	}
+}
+
+func TestSameNodes(t *testing.T) {
+	tests := []struct {
+		name     string
+		node1    Node
+		node2    Node
+		expected bool
+	}{
+		// Same tests as EqualNodes for leaf nodes (should behave identically)
+		{
+			name:     "same leaf nodes - boolean",
+			node1:    Leaf(BooleanType),
+			node2:    Leaf(BooleanType),
+			expected: true,
+		},
+		{
+			name:     "different leaf types - int32 vs int64",
+			node1:    Leaf(Int32Type),
+			node2:    Leaf(Int64Type),
+			expected: false,
+		},
+
+		// Group nodes - same fields in same order
+		{
+			name: "groups with same fields in same order",
+			node1: Group{
+				"id":   Int(64),
+				"name": String(),
+				"age":  Int(32),
+			},
+			node2: Group{
+				"id":   Int(64),
+				"name": String(),
+				"age":  Int(32),
+			},
+			expected: true,
+		},
+
+		// Group nodes - same fields in different order (key difference from EqualNodes)
+		{
+			name: "groups with same fields in different order",
+			node1: Group{
+				"id":   Int(64),
+				"name": String(),
+				"age":  Int(32),
+			},
+			node2: Group{
+				"age":  Int(32),
+				"id":   Int(64),
+				"name": String(),
+			},
+			expected: true, // This should be true for SameNodes but false for EqualNodes
+		},
+
+		// Group nodes - different fields
+		{
+			name: "groups with different field names",
+			node1: Group{
+				"id":   Int(64),
+				"name": String(),
+			},
+			node2: Group{
+				"id":    Int(64),
+				"title": String(),
+			},
+			expected: false,
+		},
+		{
+			name: "groups with different field types",
+			node1: Group{
+				"id":   Int(64),
+				"name": String(),
+			},
+			node2: Group{
+				"id":   Int(32), // Different type
+				"name": String(),
+			},
+			expected: false,
+		},
+
+		// Nested groups - same structure, different order
+		{
+			name: "nested groups with field reordering",
+			node1: Group{
+				"user": Group{
+					"id":   Int(64),
+					"name": String(),
+				},
+				"metadata": Group{
+					"created": Date(),
+					"updated": Date(),
+				},
+			},
+			node2: Group{
+				"metadata": Group{
+					"updated": Date(), // Different order
+					"created": Date(),
+				},
+				"user": Group{
+					"name": String(), // Different order
+					"id":   Int(64),
+				},
+			},
+			expected: true,
+		},
+
+		// Mixed cases
+		{
+			name: "groups with different repetition",
+			node1: Required(Group{
+				"name": String(),
+			}),
+			node2: Optional(Group{
+				"name": String(),
+			}),
+			expected: false,
+		},
+
+		// MAP and LIST types - logical type preserved
+		{
+			name:     "same map types",
+			node1:    Map(String(), Int(32)),
+			node2:    Map(String(), Int(32)),
+			expected: true,
+		},
+		{
+			name:  "map vs regular group with same structure",
+			node1: Map(String(), Int(32)),
+			node2: Group{
+				"key_value": Repeated(Group{
+					"key":   String(),
+					"value": Int(32),
+				}),
+			},
+			expected: false, // Different logical types
+		},
+
+		// Edge cases
+		{
+			name:     "empty groups",
+			node1:    Group{},
+			node2:    Group{},
+			expected: true,
+		},
+		{
+			name:     "leaf vs group",
+			node1:    String(),
+			node2:    Group{"name": String()},
+			expected: false,
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			result := SameNodes(test.node1, test.node2)
+			if result != test.expected {
+				t.Errorf("SameNodes(%v, %v) = %v, expected %v",
+					test.node1, test.node2, result, test.expected)
+			}
+
+			// Also verify that for leaf nodes, SameNodes behaves identically to EqualNodes
+			if test.node1.Leaf() && test.node2.Leaf() {
+				equalResult := EqualNodes(test.node1, test.node2)
+				if result != equalResult {
+					t.Errorf("SameNodes and EqualNodes should be identical for leaf nodes, got SameNodes=%v EqualNodes=%v",
+						result, equalResult)
+				}
+			}
+		})
+	}
+}
+
+func TestSameNodesVsEqualNodes(t *testing.T) {
+	// Test case that shows the difference between SameNodes and EqualNodes
+	// Create groups dynamically to ensure different field orders
+
+	// Create first group by building it step by step
+	group1 := make(Group)
+	group1["FieldA"] = String()
+	group1["FieldB"] = Int(32)
+	group1["FieldC"] = Date()
+
+	// Create second group in different order
+	group2 := make(Group)
+	group2["FieldC"] = Date() // Different order
+	group2["FieldA"] = String()
+	group2["FieldB"] = Int(32)
+
+	// Verify the fields are actually in different order by checking field iteration
+	fields1 := group1.Fields()
+	fields2 := group2.Fields()
+
+	// Check if any field is in a different position
+	differentOrder := false
+	for i := range fields1 {
+		if fields1[i].Name() != fields2[i].Name() {
+			differentOrder = true
+			break
+		}
+	}
+
+	// Only run the EqualNodes test if they're actually in different order
+	if differentOrder {
+		// SameNodes should be true (order-independent)
+		if !SameNodes(group1, group2) {
+			t.Error("SameNodes should return true for same fields in different order")
+		}
+
+		// EqualNodes should be false (order-dependent)
+		if EqualNodes(group1, group2) {
+			t.Error("EqualNodes should return false for same fields in different order")
+		}
+	} else {
+		// If they happen to be in the same order, both should be true
+		t.Logf("Groups ended up in same order, both comparisons should be true")
+		if !SameNodes(group1, group2) {
+			t.Error("SameNodes should return true for same fields")
+		}
+		if !EqualNodes(group1, group2) {
+			t.Error("EqualNodes should return true for same fields in same order")
+		}
+	}
+}
+
 func TestEncodingOf(t *testing.T) {
 	testCases := []struct {
 		name             string
diff --git a/row_group_test.go b/row_group_test.go
index 39e3347..9753950 100644
--- a/row_group_test.go
+++ b/row_group_test.go
@@ -23,9 +23,9 @@ func sortedRowGroup(options []parquet.RowGroupOption, rows ...any) parquet.RowGr
 }
 
 type Person struct {
+	Age       int
 	FirstName utf8string
 	LastName  utf8string
-	Age       int
 }
 
 type LastNameOnly struct {
diff --git a/sorting_test.go b/sorting_test.go
index 7486706..e56614d 100644
--- a/sorting_test.go
+++ b/sorting_test.go
@@ -439,3 +439,220 @@ func TestIssue293(t *testing.T) {
 		}
 	}
 }
+
+func TestEqualSortingColumns(t *testing.T) {
+	tests := []struct {
+		name     string
+		a        []parquet.SortingColumn
+		b        []parquet.SortingColumn
+		expected bool
+	}{
+		{
+			name:     "both empty",
+			a:        []parquet.SortingColumn{},
+			b:        []parquet.SortingColumn{},
+			expected: true,
+		},
+		{
+			name:     "both nil",
+			a:        nil,
+			b:        nil,
+			expected: true,
+		},
+		{
+			name:     "one empty, one nil",
+			a:        []parquet.SortingColumn{},
+			b:        nil,
+			expected: true,
+		},
+		{
+			name:     "different lengths",
+			a:        []parquet.SortingColumn{parquet.Ascending("A")},
+			b:        []parquet.SortingColumn{parquet.Ascending("A"), parquet.Descending("B")},
+			expected: false,
+		},
+		{
+			name:     "one empty, one non-empty",
+			a:        []parquet.SortingColumn{},
+			b:        []parquet.SortingColumn{parquet.Ascending("A")},
+			expected: false,
+		},
+		{
+			name:     "identical single column",
+			a:        []parquet.SortingColumn{parquet.Ascending("A")},
+			b:        []parquet.SortingColumn{parquet.Ascending("A")},
+			expected: true,
+		},
+		{
+			name: "identical multiple columns",
+			a: []parquet.SortingColumn{
+				parquet.Ascending("A"),
+				parquet.Descending("B"),
+				parquet.Ascending("C"),
+			},
+			b: []parquet.SortingColumn{
+				parquet.Ascending("A"),
+				parquet.Descending("B"),
+				parquet.Ascending("C"),
+			},
+			expected: true,
+		},
+		{
+			name:     "different column names",
+			a:        []parquet.SortingColumn{parquet.Ascending("A")},
+			b:        []parquet.SortingColumn{parquet.Ascending("B")},
+			expected: false,
+		},
+		{
+			name:     "different directions",
+			a:        []parquet.SortingColumn{parquet.Ascending("A")},
+			b:        []parquet.SortingColumn{parquet.Descending("A")},
+			expected: false,
+		},
+		{
+			name:     "different nulls first setting",
+			a:        []parquet.SortingColumn{parquet.NullsFirst(parquet.Ascending("A"))},
+			b:        []parquet.SortingColumn{parquet.Ascending("A")}, // defaults to nulls last
+			expected: false,
+		},
+		{
+			name:     "same nulls first setting",
+			a:        []parquet.SortingColumn{parquet.NullsFirst(parquet.Ascending("A"))},
+			b:        []parquet.SortingColumn{parquet.NullsFirst(parquet.Ascending("A"))},
+			expected: true,
+		},
+		{
+			name:     "complex paths identical",
+			a:        []parquet.SortingColumn{parquet.Ascending("nested.field.a")},
+			b:        []parquet.SortingColumn{parquet.Ascending("nested.field.a")},
+			expected: true,
+		},
+		{
+			name:     "complex paths different",
+			a:        []parquet.SortingColumn{parquet.Ascending("nested.field.a")},
+			b:        []parquet.SortingColumn{parquet.Ascending("nested.field.b")},
+			expected: false,
+		},
+		{
+			name: "multiple columns with difference in middle",
+			a: []parquet.SortingColumn{
+				parquet.Ascending("A"),
+				parquet.Ascending("B"), // different from b
+				parquet.Descending("C"),
+			},
+			b: []parquet.SortingColumn{
+				parquet.Ascending("A"),
+				parquet.Descending("B"), // different from a
+				parquet.Descending("C"),
+			},
+			expected: false,
+		},
+		{
+			name: "multiple columns with difference at end",
+			a: []parquet.SortingColumn{
+				parquet.Ascending("A"),
+				parquet.Descending("B"),
+				parquet.Ascending("C"), // different from b
+			},
+			b: []parquet.SortingColumn{
+				parquet.Ascending("A"),
+				parquet.Descending("B"),
+				parquet.Descending("C"), // different from a
+			},
+			expected: false,
+		},
+		{
+			name: "mixed properties - all same",
+			a: []parquet.SortingColumn{
+				parquet.NullsFirst(parquet.Ascending("A")),
+				parquet.Descending("B"),
+				parquet.NullsFirst(parquet.Descending("C")),
+			},
+			b: []parquet.SortingColumn{
+				parquet.NullsFirst(parquet.Ascending("A")),
+				parquet.Descending("B"),
+				parquet.NullsFirst(parquet.Descending("C")),
+			},
+			expected: true,
+		},
+		{
+			name: "mixed properties - different nulls handling",
+			a: []parquet.SortingColumn{
+				parquet.NullsFirst(parquet.Ascending("A")),
+				parquet.Descending("B"),
+			},
+			b: []parquet.SortingColumn{
+				parquet.Ascending("A"), // nulls last by default
+				parquet.Descending("B"),
+			},
+			expected: false,
+		},
+		{
+			name: "same columns in different order",
+			a: []parquet.SortingColumn{
+				parquet.Ascending("A"),
+				parquet.Descending("B"),
+			},
+			b: []parquet.SortingColumn{
+				parquet.Descending("B"),
+				parquet.Ascending("A"),
+			},
+			expected: false, // Order matters
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			result := parquet.EqualSortingColumns(test.a, test.b)
+			if result != test.expected {
+				t.Errorf("EqualSortingColumns(%v, %v) = %v, expected %v",
+					formatSortingColumns(test.a), formatSortingColumns(test.b), result, test.expected)
+			}
+
+			// Test symmetry - EqualSortingColumns should be commutative
+			resultReverse := parquet.EqualSortingColumns(test.b, test.a)
+			if result != resultReverse {
+				t.Errorf("EqualSortingColumns is not symmetric: EqualSortingColumns(a, b) = %v, but EqualSortingColumns(b, a) = %v",
+					result, resultReverse)
+			}
+		})
+	}
+}
+
+// formatSortingColumns formats a slice of sorting columns for test output
+func formatSortingColumns(cols []parquet.SortingColumn) string {
+	if len(cols) == 0 {
+		return "[]"
+	}
+
+	result := "["
+	for i, col := range cols {
+		if i > 0 {
+			result += ", "
+		}
+
+		// Format each column with its properties
+		path := col.Path()
+		pathStr := ""
+		if len(path) > 0 {
+			pathStr = path[0]
+			for _, p := range path[1:] {
+				pathStr += "." + p
+			}
+		}
+
+		direction := "asc"
+		if col.Descending() {
+			direction = "desc"
+		}
+
+		nulls := ""
+		if col.NullsFirst() {
+			nulls = "+nulls_first"
+		}
+
+		result += direction + "(" + pathStr + ")" + nulls
+	}
+	result += "]"
+	return result
+}
diff --git a/type_test.go b/type_test.go
index 0716459..8aadc62 100644
--- a/type_test.go
+++ b/type_test.go
@@ -21,3 +21,312 @@ func TestLogicalTypesEqual(t *testing.T) {
 		}
 	}
 }
+
+func TestEqualTypes(t *testing.T) {
+	tests := []struct {
+		name     string
+		type1    parquet.Type
+		type2    parquet.Type
+		expected bool
+	}{
+		// Basic physical types - should be equal
+		{
+			name:     "same boolean types",
+			type1:    parquet.BooleanType,
+			type2:    parquet.BooleanType,
+			expected: true,
+		},
+		{
+			name:     "same int32 types",
+			type1:    parquet.Int32Type,
+			type2:    parquet.Int32Type,
+			expected: true,
+		},
+		{
+			name:     "same int64 types",
+			type1:    parquet.Int64Type,
+			type2:    parquet.Int64Type,
+			expected: true,
+		},
+		{
+			name:     "same float types",
+			type1:    parquet.FloatType,
+			type2:    parquet.FloatType,
+			expected: true,
+		},
+		{
+			name:     "same double types",
+			type1:    parquet.DoubleType,
+			type2:    parquet.DoubleType,
+			expected: true,
+		},
+		{
+			name:     "same byte array types",
+			type1:    parquet.ByteArrayType,
+			type2:    parquet.ByteArrayType,
+			expected: true,
+		},
+		{
+			name:     "same int96 types",
+			type1:    parquet.Int96Type,
+			type2:    parquet.Int96Type,
+			expected: true,
+		},
+
+		// Different physical types - should not be equal
+		{
+			name:     "different kinds - int32 vs int64",
+			type1:    parquet.Int32Type,
+			type2:    parquet.Int64Type,
+			expected: false,
+		},
+		{
+			name:     "different kinds - boolean vs int32",
+			type1:    parquet.BooleanType,
+			type2:    parquet.Int32Type,
+			expected: false,
+		},
+		{
+			name:     "different kinds - float vs double",
+			type1:    parquet.FloatType,
+			type2:    parquet.DoubleType,
+			expected: false,
+		},
+		{
+			name:     "different kinds - byte array vs fixed len byte array",
+			type1:    parquet.ByteArrayType,
+			type2:    parquet.FixedLenByteArrayType(10),
+			expected: false,
+		},
+
+		// Fixed length byte arrays with different lengths
+		{
+			name:     "same fixed len byte array types",
+			type1:    parquet.FixedLenByteArrayType(16),
+			type2:    parquet.FixedLenByteArrayType(16),
+			expected: true,
+		},
+		{
+			name:     "different fixed len byte array lengths",
+			type1:    parquet.FixedLenByteArrayType(16),
+			type2:    parquet.FixedLenByteArrayType(32),
+			expected: false,
+		},
+
+		// Logical types - same underlying physical type
+		{
+			name:     "same string logical types",
+			type1:    parquet.String().Type(),
+			type2:    parquet.String().Type(),
+			expected: true,
+		},
+		{
+			name:     "same int32 logical types",
+			type1:    parquet.Int(32).Type(),
+			type2:    parquet.Int(32).Type(),
+			expected: true,
+		},
+		{
+			name:     "same int64 logical types",
+			type1:    parquet.Int(64).Type(),
+			type2:    parquet.Int(64).Type(),
+			expected: true,
+		},
+		{
+			name:     "same uint32 logical types",
+			type1:    parquet.Uint(32).Type(),
+			type2:    parquet.Uint(32).Type(),
+			expected: true,
+		},
+		{
+			name:     "same uint64 logical types",
+			type1:    parquet.Uint(64).Type(),
+			type2:    parquet.Uint(64).Type(),
+			expected: true,
+		},
+		{
+			name:     "same date logical types",
+			type1:    parquet.Date().Type(),
+			type2:    parquet.Date().Type(),
+			expected: true,
+		},
+		{
+			name:     "same json logical types",
+			type1:    parquet.JSON().Type(),
+			type2:    parquet.JSON().Type(),
+			expected: true,
+		},
+		{
+			name:     "same bson logical types",
+			type1:    parquet.BSON().Type(),
+			type2:    parquet.BSON().Type(),
+			expected: true,
+		},
+
+		// Different logical types with same physical type
+		{
+			name:     "string vs json (both byte array)",
+			type1:    parquet.String().Type(),
+			type2:    parquet.JSON().Type(),
+			expected: false,
+		},
+		{
+			name:     "string vs bson (both byte array)",
+			type1:    parquet.String().Type(),
+			type2:    parquet.BSON().Type(),
+			expected: false,
+		},
+		{
+			name:     "json vs bson (both byte array)",
+			type1:    parquet.JSON().Type(),
+			type2:    parquet.BSON().Type(),
+			expected: false,
+		},
+		{
+			name:     "int32 vs uint32 (same physical kind)",
+			type1:    parquet.Int(32).Type(),
+			type2:    parquet.Uint(32).Type(),
+			expected: false,
+		},
+		{
+			name:     "int64 vs uint64 (same physical kind)",
+			type1:    parquet.Int(64).Type(),
+			type2:    parquet.Uint(64).Type(),
+			expected: false,
+		},
+
+		// Different bit widths for same logical type
+		{
+			name:     "int32 vs int64 logical types",
+			type1:    parquet.Int(32).Type(),
+			type2:    parquet.Int(64).Type(),
+			expected: false,
+		},
+		{
+			name:     "uint32 vs uint64 logical types",
+			type1:    parquet.Uint(32).Type(),
+			type2:    parquet.Uint(64).Type(),
+			expected: false,
+		},
+
+		// Timestamp logical types with different units
+		{
+			name:     "same timestamp millis",
+			type1:    parquet.Timestamp(parquet.Millisecond).Type(),
+			type2:    parquet.Timestamp(parquet.Millisecond).Type(),
+			expected: true,
+		},
+		{
+			name:     "same timestamp micros",
+			type1:    parquet.Timestamp(parquet.Microsecond).Type(),
+			type2:    parquet.Timestamp(parquet.Microsecond).Type(),
+			expected: true,
+		},
+		{
+			name:     "same timestamp nanos",
+			type1:    parquet.Timestamp(parquet.Nanosecond).Type(),
+			type2:    parquet.Timestamp(parquet.Nanosecond).Type(),
+			expected: true,
+		},
+		{
+			name:     "different timestamp units - millis vs micros",
+			type1:    parquet.Timestamp(parquet.Millisecond).Type(),
+			type2:    parquet.Timestamp(parquet.Microsecond).Type(),
+			expected: false,
+		},
+		{
+			name:     "different timestamp units - micros vs nanos",
+			type1:    parquet.Timestamp(parquet.Microsecond).Type(),
+			type2:    parquet.Timestamp(parquet.Nanosecond).Type(),
+			expected: false,
+		},
+
+		// Time logical types with different units
+		{
+			name:     "same time millis",
+			type1:    parquet.Time(parquet.Millisecond).Type(),
+			type2:    parquet.Time(parquet.Millisecond).Type(),
+			expected: true,
+		},
+		{
+			name:     "same time micros",
+			type1:    parquet.Time(parquet.Microsecond).Type(),
+			type2:    parquet.Time(parquet.Microsecond).Type(),
+			expected: true,
+		},
+		{
+			name:     "same time nanos",
+			type1:    parquet.Time(parquet.Nanosecond).Type(),
+			type2:    parquet.Time(parquet.Nanosecond).Type(),
+			expected: true,
+		},
+		{
+			name:     "different time units - millis vs micros",
+			type1:    parquet.Time(parquet.Millisecond).Type(),
+			type2:    parquet.Time(parquet.Microsecond).Type(),
+			expected: false,
+		},
+
+		// Logical type vs physical type
+		{
+			name:     "string logical vs byte array physical",
+			type1:    parquet.String().Type(),
+			type2:    parquet.ByteArrayType,
+			expected: false,
+		},
+		{
+			name:     "int32 logical vs int32 with same logical type",
+			type1:    parquet.Int(32).Type(),
+			type2:    parquet.Int32Type,
+			expected: true, // Both have the same Integer logical type
+		},
+		{
+			name:     "date logical vs int32 physical",
+			type1:    parquet.Date().Type(),
+			type2:    parquet.Int32Type,
+			expected: false,
+		},
+
+		// Decimal logical types with different precision/scale
+		{
+			name:     "same decimal(10,2)",
+			type1:    parquet.Decimal(10, 2, parquet.Int32Type).Type(),
+			type2:    parquet.Decimal(10, 2, parquet.Int32Type).Type(),
+			expected: true,
+		},
+		{
+			name:     "different decimal precision",
+			type1:    parquet.Decimal(10, 2, parquet.Int32Type).Type(),
+			type2:    parquet.Decimal(12, 2, parquet.Int32Type).Type(),
+			expected: false,
+		},
+		{
+			name:     "different decimal scale",
+			type1:    parquet.Decimal(10, 2, parquet.Int32Type).Type(),
+			type2:    parquet.Decimal(10, 3, parquet.Int32Type).Type(),
+			expected: false,
+		},
+		{
+			name:     "same decimal different physical type",
+			type1:    parquet.Decimal(10, 2, parquet.Int32Type).Type(),
+			type2:    parquet.Decimal(10, 2, parquet.Int64Type).Type(),
+			expected: false,
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			result := parquet.EqualTypes(test.type1, test.type2)
+			if result != test.expected {
+				t.Errorf("EqualTypes(%v, %v) = %v, expected %v",
+					test.type1, test.type2, result, test.expected)
+
+				// Additional debugging info
+				t.Logf("Type1: Kind=%v, Length=%v, LogicalType=%v",
+					test.type1.Kind(), test.type1.Length(), test.type1.LogicalType())
+				t.Logf("Type2: Kind=%v, Length=%v, LogicalType=%v",
+					test.type2.Kind(), test.type2.Length(), test.type2.LogicalType())
+			}
+		})
+	}
+}