Repository: apache/iceberg-go
Pull request #329 (2025-03-10 20:22:17)

feat(table): Add computation of iceberg stats from parquet files
This one's a big one!

Adding initial implementation of deriving and computing the Iceberg metadata statistics from Parquet files in order to facilitate adding files directly. This includes utilizing the Parquet file metadata retrieve the typed statistics and collecting them based on optional metrics modes and so on. This also includes various unit tests to confirm it is working.


Patch:
diff --git a/exprs.go b/exprs.go
index ed2bc8d..47afe40 100644
--- a/exprs.go
+++ b/exprs.go
@@ -512,7 +512,7 @@ func (b *boundRef[T]) evalToLiteral(st structLike) Optional[Literal] {
 		return Optional[Literal]{}
 	}
 
-	lit := NewLiteral[T](v.Val)
+	lit := NewLiteral(v.Val)
 	if !lit.Type().Equals(b.field.Type) {
 		lit, _ = lit.To(b.field.Type)
 	}
diff --git a/internal/utils.go b/internal/utils.go
index aa3c841..d29ab8a 100644
--- a/internal/utils.go
+++ b/internal/utils.go
@@ -19,6 +19,7 @@ package internal
 
 import (
 	"cmp"
+	"fmt"
 	"io"
 	"slices"
 )
@@ -143,3 +144,14 @@ func (w *CountingWriter) Write(p []byte) (int, error) {
 
 	return n, err
 }
+
+func RecoverError(err *error) {
+	if r := recover(); r != nil {
+		switch e := r.(type) {
+		case string:
+			*err = fmt.Errorf("error encountered during arrow schema visitor: %s", e)
+		case error:
+			*err = fmt.Errorf("error encountered during arrow schema visitor: %w", e)
+		}
+	}
+}
diff --git a/table/name_mapping.go b/name_mapping.go
similarity index 59%
rename from table/name_mapping.go
rename to name_mapping.go
index dae5a37..1418961 100644
--- a/table/name_mapping.go
+++ b/name_mapping.go
@@ -15,7 +15,7 @@
 // specific language governing permissions and limitations
 // under the License.
 
-package table
+package iceberg
 
 import (
 	"fmt"
@@ -23,7 +23,7 @@ import (
 	"strconv"
 	"strings"
 
-	"github.com/apache/iceberg-go"
+	"github.com/apache/iceberg-go/internal"
 )
 
 type MappedField struct {
@@ -35,6 +35,24 @@ type MappedField struct {
 	Fields  []MappedField `json:"fields,omitempty"`
 }
 
+func (m *MappedField) ID() int {
+	if m.FieldID == nil {
+		return -1
+	}
+
+	return *m.FieldID
+}
+
+func (m *MappedField) GetField(field string) *MappedField {
+	for _, f := range m.Fields {
+		if slices.Contains(f.Names, field) {
+			return &f
+		}
+	}
+
+	return nil
+}
+
 func (m *MappedField) Len() int { return len(m.Fields) }
 
 func (m *MappedField) String() string {
@@ -87,18 +105,18 @@ type NameMappingVisitor[S, T any] interface {
 
 func VisitNameMapping[S, T any](obj NameMapping, visitor NameMappingVisitor[S, T]) (res S, err error) {
 	if obj == nil {
-		err = fmt.Errorf("%w: cannot visit nil NameMapping", iceberg.ErrInvalidArgument)
+		err = fmt.Errorf("%w: cannot visit nil NameMapping", ErrInvalidArgument)
 
 		return
 	}
 
-	defer recoverError(&err)
+	defer internal.RecoverError(&err)
 
 	return visitor.Mapping(obj, visitMappedFields([]MappedField(obj), visitor)), err
 }
 
 func VisitMappedFields[S, T any](fields []MappedField, visitor NameMappingVisitor[S, T]) (res S, err error) {
-	defer recoverError(&err)
+	defer internal.RecoverError(&err)
 
 	return visitMappedFields(fields, visitor), err
 }
@@ -118,22 +136,12 @@ func (NameMappingAccessor) SchemaPartner(partner *MappedField) *MappedField {
 	return partner
 }
 
-func (NameMappingAccessor) getField(p *MappedField, field string) *MappedField {
-	for _, f := range p.Fields {
-		if slices.Contains(f.Names, field) {
-			return &f
-		}
-	}
-
-	return nil
-}
-
 func (n NameMappingAccessor) FieldPartner(partnerStruct *MappedField, _ int, fieldName string) *MappedField {
 	if partnerStruct == nil {
 		return nil
 	}
 
-	return n.getField(partnerStruct, fieldName)
+	return partnerStruct.GetField(fieldName)
 }
 
 func (n NameMappingAccessor) ListElementPartner(partnerList *MappedField) *MappedField {
@@ -141,7 +149,7 @@ func (n NameMappingAccessor) ListElementPartner(partnerList *MappedField) *Mappe
 		return nil
 	}
 
-	return n.getField(partnerList, "element")
+	return partnerList.GetField("element")
 }
 
 func (n NameMappingAccessor) MapKeyPartner(partnerMap *MappedField) *MappedField {
@@ -149,7 +157,7 @@ func (n NameMappingAccessor) MapKeyPartner(partnerMap *MappedField) *MappedField
 		return nil
 	}
 
-	return n.getField(partnerMap, "key")
+	return partnerMap.GetField("key")
 }
 
 func (n NameMappingAccessor) MapValuePartner(partnerMap *MappedField) *MappedField {
@@ -157,7 +165,7 @@ func (n NameMappingAccessor) MapValuePartner(partnerMap *MappedField) *MappedFie
 		return nil
 	}
 
-	return n.getField(partnerMap, "value")
+	return partnerMap.GetField("value")
 }
 
 type nameMapProjectVisitor struct {
@@ -168,55 +176,55 @@ func (n *nameMapProjectVisitor) popPath() {
 	n.currentPath = n.currentPath[:len(n.currentPath)-1]
 }
 
-func (n *nameMapProjectVisitor) BeforeField(f iceberg.NestedField, _ *MappedField) {
+func (n *nameMapProjectVisitor) BeforeField(f NestedField, _ *MappedField) {
 	n.currentPath = append(n.currentPath, f.Name)
 }
 
-func (n *nameMapProjectVisitor) AfterField(iceberg.NestedField, *MappedField) {
+func (n *nameMapProjectVisitor) AfterField(NestedField, *MappedField) {
 	n.popPath()
 }
 
-func (n *nameMapProjectVisitor) BeforeListElement(iceberg.NestedField, *MappedField) {
+func (n *nameMapProjectVisitor) BeforeListElement(NestedField, *MappedField) {
 	n.currentPath = append(n.currentPath, "element")
 }
 
-func (n *nameMapProjectVisitor) AfterListElement(iceberg.NestedField, *MappedField) {
+func (n *nameMapProjectVisitor) AfterListElement(NestedField, *MappedField) {
 	n.popPath()
 }
 
-func (n *nameMapProjectVisitor) BeforeMapKey(iceberg.NestedField, *MappedField) {
+func (n *nameMapProjectVisitor) BeforeMapKey(NestedField, *MappedField) {
 	n.currentPath = append(n.currentPath, "key")
 }
 
-func (n *nameMapProjectVisitor) AfterMapKey(iceberg.NestedField, *MappedField) {
+func (n *nameMapProjectVisitor) AfterMapKey(NestedField, *MappedField) {
 	n.popPath()
 }
 
-func (n *nameMapProjectVisitor) BeforeMapValue(iceberg.NestedField, *MappedField) {
+func (n *nameMapProjectVisitor) BeforeMapValue(NestedField, *MappedField) {
 	n.currentPath = append(n.currentPath, "value")
 }
 
-func (n *nameMapProjectVisitor) AfterMapValue(iceberg.NestedField, *MappedField) {
+func (n *nameMapProjectVisitor) AfterMapValue(NestedField, *MappedField) {
 	n.popPath()
 }
 
-func (n *nameMapProjectVisitor) Schema(_ *iceberg.Schema, _ *MappedField, structResult iceberg.NestedField) iceberg.NestedField {
+func (n *nameMapProjectVisitor) Schema(_ *Schema, _ *MappedField, structResult NestedField) NestedField {
 	return structResult
 }
 
-func (n *nameMapProjectVisitor) Struct(_ iceberg.StructType, _ *MappedField, fieldResults []iceberg.NestedField) iceberg.NestedField {
-	return iceberg.NestedField{
-		Type: &iceberg.StructType{FieldList: fieldResults},
+func (n *nameMapProjectVisitor) Struct(_ StructType, _ *MappedField, fieldResults []NestedField) NestedField {
+	return NestedField{
+		Type: &StructType{FieldList: fieldResults},
 	}
 }
 
-func (n *nameMapProjectVisitor) Field(field iceberg.NestedField, fieldPartner *MappedField, fieldResult iceberg.NestedField) iceberg.NestedField {
+func (n *nameMapProjectVisitor) Field(field NestedField, fieldPartner *MappedField, fieldResult NestedField) NestedField {
 	if fieldPartner == nil {
 		panic(fmt.Errorf("%w: field missing from name mapping: %s",
-			iceberg.ErrInvalidArgument, strings.Join(n.currentPath, ".")))
+			ErrInvalidArgument, strings.Join(n.currentPath, ".")))
 	}
 
-	return iceberg.NestedField{
+	return NestedField{
 		ID:             *fieldPartner.FieldID,
 		Name:           field.Name,
 		Type:           fieldResult.Type,
@@ -241,16 +249,16 @@ func (nameMapProjectVisitor) mappedFieldID(mapped *MappedField, name string) int
 	return -1
 }
 
-func (n *nameMapProjectVisitor) List(lt iceberg.ListType, listPartner *MappedField, elemResult iceberg.NestedField) iceberg.NestedField {
+func (n *nameMapProjectVisitor) List(lt ListType, listPartner *MappedField, elemResult NestedField) NestedField {
 	if listPartner == nil {
 		panic(fmt.Errorf("%w: field missing from name mapping: %s",
-			iceberg.ErrInvalidArgument, strings.Join(n.currentPath, ".")))
+			ErrInvalidArgument, strings.Join(n.currentPath, ".")))
 	}
 
 	elementID := n.mappedFieldID(listPartner, "element")
 
-	return iceberg.NestedField{
-		Type: &iceberg.ListType{
+	return NestedField{
+		Type: &ListType{
 			ElementID:       elementID,
 			Element:         elemResult.Type,
 			ElementRequired: lt.ElementRequired,
@@ -258,17 +266,17 @@ func (n *nameMapProjectVisitor) List(lt iceberg.ListType, listPartner *MappedFie
 	}
 }
 
-func (n *nameMapProjectVisitor) Map(m iceberg.MapType, mapPartner *MappedField, keyResult, valResult iceberg.NestedField) iceberg.NestedField {
+func (n *nameMapProjectVisitor) Map(m MapType, mapPartner *MappedField, keyResult, valResult NestedField) NestedField {
 	if mapPartner == nil {
 		panic(fmt.Errorf("%w: field missing from name mapping: %s",
-			iceberg.ErrInvalidArgument, strings.Join(n.currentPath, ".")))
+			ErrInvalidArgument, strings.Join(n.currentPath, ".")))
 	}
 
 	keyID := n.mappedFieldID(mapPartner, "key")
 	valID := n.mappedFieldID(mapPartner, "value")
 
-	return iceberg.NestedField{
-		Type: &iceberg.MapType{
+	return NestedField{
+		Type: &MapType{
 			KeyID:         keyID,
 			KeyType:       keyResult.Type,
 			ValueID:       valID,
@@ -278,17 +286,17 @@ func (n *nameMapProjectVisitor) Map(m iceberg.MapType, mapPartner *MappedField,
 	}
 }
 
-func (n *nameMapProjectVisitor) Primitive(p iceberg.PrimitiveType, primitivePartner *MappedField) iceberg.NestedField {
+func (n *nameMapProjectVisitor) Primitive(p PrimitiveType, primitivePartner *MappedField) NestedField {
 	if primitivePartner == nil {
 		panic(fmt.Errorf("%w: field missing from name mapping: %s",
-			iceberg.ErrInvalidArgument, strings.Join(n.currentPath, ".")))
+			ErrInvalidArgument, strings.Join(n.currentPath, ".")))
 	}
 
-	return iceberg.NestedField{Type: p}
+	return NestedField{Type: p}
 }
 
-func ApplyNameMapping(schemaWithoutIDs *iceberg.Schema, nameMapping NameMapping) (*iceberg.Schema, error) {
-	top, err := iceberg.VisitSchemaWithPartner[iceberg.NestedField, *MappedField](schemaWithoutIDs,
+func ApplyNameMapping(schemaWithoutIDs *Schema, nameMapping NameMapping) (*Schema, error) {
+	top, err := VisitSchemaWithPartner(schemaWithoutIDs,
 		&MappedField{Fields: nameMapping},
 		&nameMapProjectVisitor{currentPath: make([]string, 0, 1)},
 		NameMappingAccessor{})
@@ -296,6 +304,62 @@ func ApplyNameMapping(schemaWithoutIDs *iceberg.Schema, nameMapping NameMapping)
 		return nil, err
 	}
 
-	return iceberg.NewSchema(schemaWithoutIDs.ID,
-		top.Type.(*iceberg.StructType).FieldList...), nil
+	return NewSchema(schemaWithoutIDs.ID,
+		top.Type.(*StructType).FieldList...), nil
+}
+
+type createMapping struct{}
+
+func (createMapping) Schema(_ *Schema, result []MappedField) []MappedField {
+	return result
+}
+
+func (createMapping) Struct(st StructType, result [][]MappedField) []MappedField {
+	output := make([]MappedField, len(st.FieldList))
+	for i, field := range st.FieldList {
+		output[i] = MappedField{
+			Names:   []string{field.Name},
+			FieldID: &field.ID,
+			Fields:  result[i],
+		}
+	}
+
+	return output
+}
+
+func (createMapping) Field(_ NestedField, result []MappedField) []MappedField {
+	return result
+}
+
+func (createMapping) List(listType ListType, elemResult []MappedField) []MappedField {
+	return []MappedField{{
+		Names:   []string{"element"},
+		FieldID: &listType.ElementID,
+		Fields:  elemResult,
+	}}
+}
+
+func (createMapping) Map(mapType MapType, keyResult, valResult []MappedField) []MappedField {
+	return []MappedField{
+		{
+			Names:   []string{"key"},
+			FieldID: &mapType.KeyID,
+			Fields:  keyResult,
+		},
+		{
+			Names:   []string{"value"},
+			FieldID: &mapType.ValueID,
+			Fields:  valResult,
+		},
+	}
+}
+
+func (createMapping) Primitive(_ PrimitiveType) []MappedField {
+	return []MappedField{}
+}
+
+func createMappingFromSchema(schema *Schema) NameMapping {
+	result, _ := Visit(schema, createMapping{})
+
+	return NameMapping(result)
 }
diff --git a/schema.go b/schema.go
index 6fbdf67..d078554 100644
--- a/schema.go
+++ b/schema.go
@@ -46,7 +46,8 @@ type Schema struct {
 	nameToIDLower atomic.Pointer[map[string]int]
 	idToAccessor  atomic.Pointer[map[int]accessor]
 
-	lazyIDToParent func() (map[int]int, error)
+	lazyIDToParent  func() (map[int]int, error)
+	lazyNameMapping func() NameMapping
 }
 
 // NewSchema constructs a new schema with the provided ID
@@ -63,6 +64,9 @@ func NewSchemaWithIdentifiers(id int, identifierIDs []int, fields ...NestedField
 	s.lazyIDToParent = sync.OnceValues(func() (map[int]int, error) {
 		return IndexParents(s)
 	})
+	s.lazyNameMapping = sync.OnceValue(func() NameMapping {
+		return createMappingFromSchema(s)
+	})
 
 	return s
 }
@@ -164,6 +168,8 @@ func (s *Schema) lazyIdToAccessor() (map[int]accessor, error) {
 	return idx, nil
 }
 
+func (s *Schema) NameMapping() NameMapping { return s.lazyNameMapping() }
+
 func (s *Schema) Type() string { return "struct" }
 
 // AsStruct returns a Struct with the same fields as the schema which can
@@ -337,7 +343,7 @@ func (s *Schema) Equals(other *Schema) bool {
 // HighestFieldID returns the value of the numerically highest field ID
 // in this schema.
 func (s *Schema) HighestFieldID() int {
-	id, _ := Visit[int](s, findLastFieldID{})
+	id, _ := Visit(s, findLastFieldID{})
 
 	return id
 }
@@ -685,7 +691,7 @@ func visitFieldPreOrder[T any](f NestedField, visitor PreOrderSchemaVisitor[T])
 // IndexByID performs a post-order traversal of the given schema and
 // returns a mapping from field ID to field.
 func IndexByID(schema *Schema) (map[int]NestedField, error) {
-	return Visit[map[int]NestedField](schema, &indexByID{index: make(map[int]NestedField)})
+	return Visit(schema, &indexByID{index: make(map[int]NestedField)})
 }
 
 type indexByID struct {
@@ -737,7 +743,7 @@ func IndexByName(schema *Schema) (map[string]int, error) {
 			fieldNames:      make([]string, 0),
 			shortFieldNames: make([]string, 0),
 		}
-		if _, err := Visit[map[string]int](schema, indexer); err != nil {
+		if _, err := Visit(schema, indexer); err != nil {
 			return nil, err
 		}
 
@@ -756,7 +762,7 @@ func IndexNameByID(schema *Schema) (map[int]string, error) {
 		fieldNames:      make([]string, 0),
 		shortFieldNames: make([]string, 0),
 	}
-	if _, err := Visit[map[string]int](schema, indexer); err != nil {
+	if _, err := Visit(schema, indexer); err != nil {
 		return nil, err
 	}
 
diff --git a/table/arrow_scanner.go b/table/arrow_scanner.go
index 58285dd..a532c6a 100644
--- a/table/arrow_scanner.go
+++ b/table/arrow_scanner.go
@@ -219,6 +219,8 @@ type arrowScan struct {
 
 	useLargeTypes bool
 	concurrency   int
+
+	nameMapping iceberg.NameMapping
 }
 
 func (as *arrowScan) projectedFieldIDs() (set[int], error) {
@@ -268,14 +270,14 @@ func (as *arrowScan) prepareToRead(ctx context.Context, file iceberg.DataFile) (
 		return nil, nil, nil, err
 	}
 
-	fileSchema, colIndices, err := rdr.PrunedSchema(ids)
+	fileSchema, colIndices, err := rdr.PrunedSchema(ids, as.nameMapping)
 	if err != nil {
 		rdr.Close()
 
 		return nil, nil, nil, err
 	}
 
-	iceSchema, err := ArrowSchemaToIceberg(fileSchema, false, nil)
+	iceSchema, err := ArrowSchemaToIceberg(fileSchema, false, as.nameMapping)
 	if err != nil {
 		rdr.Close()
 
@@ -553,6 +555,7 @@ func createIterator(ctx context.Context, numWorkers uint, records <-chan enumera
 
 func (as *arrowScan) recordBatchesFromTasksAndDeletes(ctx context.Context, tasks []FileScanTask, deletesPerFile perFilePosDeletes) iter.Seq2[arrow.Record, error] {
 	extSet := substrait.NewExtensionSet()
+	as.nameMapping = as.metadata.NameMapping()
 
 	ctx, cancel := context.WithCancelCause(exprs.WithExtensionIDSet(ctx, extSet))
 	taskChan := make(chan internal.Enumerated[FileScanTask], len(tasks))
diff --git a/table/arrow_utils.go b/table/arrow_utils.go
index 4ff4b9f..752f677 100644
--- a/table/arrow_utils.go
+++ b/table/arrow_utils.go
@@ -20,15 +20,20 @@ package table
 import (
 	"context"
 	"fmt"
+	"maps"
+	"regexp"
 	"slices"
 	"strconv"
+	"strings"
 
 	"github.com/apache/arrow-go/v18/arrow"
 	"github.com/apache/arrow-go/v18/arrow/array"
 	"github.com/apache/arrow-go/v18/arrow/compute"
 	"github.com/apache/arrow-go/v18/arrow/extensions"
 	"github.com/apache/arrow-go/v18/arrow/memory"
+	"github.com/apache/arrow-go/v18/parquet/metadata"
 	"github.com/apache/iceberg-go"
+	"github.com/apache/iceberg-go/internal"
 )
 
 // constants to look for as Keys in Arrow field metadata
@@ -51,17 +56,6 @@ type ArrowSchemaVisitor[T any] interface {
 	Primitive(arrow.DataType) T
 }
 
-func recoverError(err *error) {
-	if r := recover(); r != nil {
-		switch e := r.(type) {
-		case string:
-			*err = fmt.Errorf("error encountered during arrow schema visitor: %s", e)
-		case error:
-			*err = fmt.Errorf("error encountered during arrow schema visitor: %w", e)
-		}
-	}
-}
-
 func VisitArrowSchema[T any](sc *arrow.Schema, visitor ArrowSchemaVisitor[T]) (res T, err error) {
 	if sc == nil {
 		err = fmt.Errorf("%w: cannot visit nil arrow schema", iceberg.ErrInvalidArgument)
@@ -69,7 +63,7 @@ func VisitArrowSchema[T any](sc *arrow.Schema, visitor ArrowSchemaVisitor[T]) (r
 		return
 	}
 
-	defer recoverError(&err)
+	defer internal.RecoverError(&err)
 
 	return visitor.Schema(sc, visitArrowStruct(arrow.StructOf(sc.Fields()...), visitor)), err
 }
@@ -199,7 +193,11 @@ func getFieldID(f arrow.Field) *int {
 		return nil
 	}
 
-	return &id
+	if id > 0 {
+		return &id
+	}
+
+	return nil
 }
 
 type hasIDs struct{}
@@ -383,7 +381,7 @@ func ArrowTypeToIceberg(dt arrow.DataType, downcastNsTimestamp bool) (iceberg.Ty
 	return out.Type.(*iceberg.StructType).FieldList[0].Type, nil
 }
 
-func ArrowSchemaToIceberg(sc *arrow.Schema, downcastNsTimestamp bool, nameMapping NameMapping) (*iceberg.Schema, error) {
+func ArrowSchemaToIceberg(sc *arrow.Schema, downcastNsTimestamp bool, nameMapping iceberg.NameMapping) (*iceberg.Schema, error) {
 	hasIDs, _ := VisitArrowSchema(sc, hasIDs{})
 
 	switch {
@@ -405,23 +403,32 @@ func ArrowSchemaToIceberg(sc *arrow.Schema, downcastNsTimestamp bool, nameMappin
 
 		return iceberg.NewSchema(0, out.Type.(*iceberg.StructType).FieldList...), nil
 	case nameMapping != nil:
-		withoutIDs, err := VisitArrowSchema(sc, convertToIceberg{
-			downcastTimestamp: downcastNsTimestamp,
-			fieldID:           func(_ arrow.Field) int { return -1 },
-		})
+		schemaWithoutIDs, err := arrowToSchemaWithoutIDs(sc, downcastNsTimestamp)
 		if err != nil {
 			return nil, err
 		}
 
-		schemaWithoutIDs := iceberg.NewSchema(0, withoutIDs.Type.(*iceberg.StructType).FieldList...)
-
-		return ApplyNameMapping(schemaWithoutIDs, nameMapping)
+		return iceberg.ApplyNameMapping(schemaWithoutIDs, nameMapping)
 	default:
 		return nil, fmt.Errorf("%w: arrow schema does not have field-ids and no name mapping provided",
 			iceberg.ErrInvalidSchema)
 	}
 }
 
+func arrowToSchemaWithoutIDs(sc *arrow.Schema, downcastNsTimestamp bool) (*iceberg.Schema, error) {
+	withoutIDs, err := VisitArrowSchema(sc, convertToIceberg{
+		downcastTimestamp: downcastNsTimestamp,
+		fieldID:           func(_ arrow.Field) int { return -1 },
+	})
+	if err != nil {
+		return nil, err
+	}
+
+	schemaWithoutIDs := iceberg.NewSchema(0, withoutIDs.Type.(*iceberg.StructType).FieldList...)
+
+	return schemaWithoutIDs, nil
+}
+
 type convertToSmallTypes struct{}
 
 func (convertToSmallTypes) Schema(_ *arrow.Schema, structResult arrow.Field) arrow.Field {
@@ -892,3 +899,356 @@ func ToRequestedSchema(ctx context.Context, requested, fileSchema *iceberg.Schem
 
 	return out, nil
 }
+
+func must[T any](v T, err error) T {
+	if err != nil {
+		panic(err)
+	}
+
+	return v
+}
+
+type metricModeType string
+
+const (
+	metricModeTruncate metricModeType = "truncate"
+	metricModeNone     metricModeType = "none"
+	metricModeCounts   metricModeType = "counts"
+	metricModeFull     metricModeType = "full"
+)
+
+type metricsMode struct {
+	typ metricModeType
+	len int
+}
+
+var truncationExpr = regexp.MustCompile(`^truncate\((\d+)\)$`)
+
+func matchMetricsMode(mode string) (metricsMode, error) {
+	sanitized := strings.ToLower(strings.TrimSpace(mode))
+	if strings.HasPrefix(sanitized, string(metricModeTruncate)) {
+		m := truncationExpr.FindStringSubmatch(sanitized)
+		if len(m) < 2 {
+			return metricsMode{}, fmt.Errorf("malformed truncate metrics mode: %s", mode)
+		}
+
+		truncLen, err := strconv.Atoi(m[1])
+		if err != nil {
+			return metricsMode{}, fmt.Errorf("malformed truncate metrics mode: %s", mode)
+		}
+
+		if truncLen <= 0 {
+			return metricsMode{}, fmt.Errorf("invalid truncate length: %d", truncLen)
+		}
+
+		return metricsMode{typ: metricModeTruncate, len: truncLen}, nil
+	}
+
+	switch sanitized {
+	case string(metricModeNone):
+		return metricsMode{typ: metricModeNone}, nil
+	case string(metricModeCounts):
+		return metricsMode{typ: metricModeCounts}, nil
+	case string(metricModeFull):
+		return metricsMode{typ: metricModeFull}, nil
+	default:
+		return metricsMode{}, fmt.Errorf("unsupported metrics mode: %s", mode)
+	}
+}
+
+type statisticsCollector struct {
+	fieldID    int
+	icebergTyp iceberg.PrimitiveType
+	mode       metricsMode
+	colName    string
+}
+
+type arrowStatsCollector struct {
+	fieldID     int
+	schema      *iceberg.Schema
+	props       iceberg.Properties
+	defaultMode string
+}
+
+func (a *arrowStatsCollector) Schema(_ *iceberg.Schema, results func() []statisticsCollector) []statisticsCollector {
+	return results()
+}
+
+func (a *arrowStatsCollector) Struct(_ iceberg.StructType, results []func() []statisticsCollector) []statisticsCollector {
+	result := make([]statisticsCollector, 0, len(results))
+	for _, res := range results {
+		result = append(result, res()...)
+	}
+
+	return result
+}
+
+func (a *arrowStatsCollector) Field(field iceberg.NestedField, fieldRes func() []statisticsCollector) []statisticsCollector {
+	a.fieldID = field.ID
+
+	return fieldRes()
+}
+
+func (a *arrowStatsCollector) List(list iceberg.ListType, elemResult func() []statisticsCollector) []statisticsCollector {
+	a.fieldID = list.ElementID
+
+	return elemResult()
+}
+
+func (a *arrowStatsCollector) Map(m iceberg.MapType, keyResult func() []statisticsCollector, valResult func() []statisticsCollector) []statisticsCollector {
+	a.fieldID = m.KeyID
+	keyRes := keyResult()
+
+	a.fieldID = m.ValueID
+	valRes := valResult()
+
+	return append(keyRes, valRes...)
+}
+
+func (a *arrowStatsCollector) Primitive(dt iceberg.PrimitiveType) []statisticsCollector {
+	colName, ok := a.schema.FindColumnName(a.fieldID)
+	if !ok {
+		return []statisticsCollector{}
+	}
+
+	metMode, err := matchMetricsMode(a.defaultMode)
+	if err != nil {
+		panic(err)
+	}
+
+	colMode, ok := a.props[MetricsModeColumnConfPrefix+"."+colName]
+	if ok {
+		metMode, err = matchMetricsMode(colMode)
+		if err != nil {
+			panic(err)
+		}
+	}
+
+	switch dt.(type) {
+	case iceberg.StringType:
+	case iceberg.BinaryType:
+	default:
+		if metMode.typ == metricModeTruncate {
+			metMode = metricsMode{typ: metricModeFull, len: 0}
+		}
+	}
+
+	isNested := strings.Contains(colName, ".")
+	if isNested && (metMode.typ == metricModeTruncate || metMode.typ == metricModeFull) {
+		metMode = metricsMode{typ: metricModeCounts}
+	}
+
+	return []statisticsCollector{{
+		fieldID:    a.fieldID,
+		icebergTyp: dt,
+		colName:    colName,
+		mode:       metMode,
+	}}
+}
+
+func computeStatsPlan(sc *iceberg.Schema, props iceberg.Properties) (map[int]statisticsCollector, error) {
+	result := make(map[int]statisticsCollector)
+
+	visitor := &arrowStatsCollector{
+		schema: sc, props: props,
+		defaultMode: props.Get(DefaultWriteMetricsModeKey, DefaultWriteMetricsModeDefault),
+	}
+
+	collectors, err := iceberg.PreOrderVisit(sc, visitor)
+	if err != nil {
+		return nil, err
+	}
+
+	for _, entry := range collectors {
+		result[entry.fieldID] = entry
+	}
+
+	return result, nil
+}
+
+type id2ParquetPath struct {
+	fieldID int
+	path    string
+}
+
+type id2ParquetPathVisitor struct {
+	fieldID int
+	path    []string
+}
+
+func (v *id2ParquetPathVisitor) Schema(_ *iceberg.Schema, res func() []id2ParquetPath) []id2ParquetPath {
+	return res()
+}
+
+func (v *id2ParquetPathVisitor) Struct(_ iceberg.StructType, results []func() []id2ParquetPath) []id2ParquetPath {
+	result := make([]id2ParquetPath, 0, len(results))
+	for _, res := range results {
+		result = append(result, res()...)
+	}
+
+	return result
+}
+
+func (v *id2ParquetPathVisitor) Field(field iceberg.NestedField, res func() []id2ParquetPath) []id2ParquetPath {
+	v.fieldID = field.ID
+	v.path = append(v.path, field.Name)
+	result := res()
+	v.path = v.path[:len(v.path)-1]
+
+	return result
+}
+
+func (v *id2ParquetPathVisitor) List(listType iceberg.ListType, elemResult func() []id2ParquetPath) []id2ParquetPath {
+	v.fieldID = listType.ElementID
+	v.path = append(v.path, "list")
+	result := elemResult()
+	v.path = v.path[:len(v.path)-1]
+
+	return result
+}
+
+func (v *id2ParquetPathVisitor) Map(m iceberg.MapType, keyResult func() []id2ParquetPath, valResult func() []id2ParquetPath) []id2ParquetPath {
+	v.fieldID = m.KeyID
+	v.path = append(v.path, "key_value")
+	keyRes := keyResult()
+	v.path = v.path[:len(v.path)-1]
+
+	v.fieldID = m.ValueID
+	v.path = append(v.path, "key_value")
+	valRes := valResult()
+	v.path = v.path[:len(v.path)-1]
+
+	return append(keyRes, valRes...)
+}
+
+func (v *id2ParquetPathVisitor) Primitive(iceberg.PrimitiveType) []id2ParquetPath {
+	return []id2ParquetPath{{fieldID: v.fieldID, path: strings.Join(v.path, ".")}}
+}
+
+func parquetPathToIDMapping(sc *iceberg.Schema) (map[string]int, error) {
+	result := make(map[string]int)
+
+	paths, err := iceberg.PreOrderVisit(sc, &id2ParquetPathVisitor{})
+	if err != nil {
+		return nil, err
+	}
+
+	for _, entry := range paths {
+		result[entry.path] = entry.fieldID
+	}
+
+	return result, nil
+}
+
+type dataFileStatistics struct {
+	recordCount     int64
+	colSizes        map[int]int64
+	valueCounts     map[int]int64
+	nullValueCounts map[int]int64
+	nanValueCounts  map[int]int64
+	// column aggregates
+	splitOffsets []int64
+}
+
+func (d *dataFileStatistics) toDataFile(spec iceberg.PartitionSpec, path string, format iceberg.FileFormat, filesize int64) (iceberg.DataFile, error) {
+	bldr, err := iceberg.NewDataFileBuilder(spec, iceberg.EntryContentData,
+		path, iceberg.ParquetFile, nil, d.recordCount, filesize)
+	if err != nil {
+		return nil, err
+	}
+
+	bldr.ColumnSizes(d.colSizes)
+	bldr.ValueCounts(d.valueCounts)
+	bldr.NullValueCounts(d.nullValueCounts)
+	bldr.NaNValueCounts(d.nanValueCounts)
+	bldr.SplitOffsets(d.splitOffsets)
+
+	return bldr.Build(), nil
+}
+
+func dataFileStatsFromParquetMetadata(pqmeta *metadata.FileMetaData, statsCols map[int]statisticsCollector, colMapping map[string]int) *dataFileStatistics {
+	var (
+		colSizes        = make(map[int]int64)
+		valueCounts     = make(map[int]int64)
+		splitOffsets    = make([]int64, 0)
+		nullValueCounts = make(map[int]int64)
+		nanValueCounts  = make(map[int]int64)
+		invalidateCol   = make(map[int]struct{})
+	)
+
+	for rg := range pqmeta.NumRowGroups() {
+		// reference: https://github.com/apache/iceberg-python/blob/main/pyiceberg/io/pyarrow.py#L2285
+		rowGroup := pqmeta.RowGroup(rg)
+		colChunk, err := rowGroup.ColumnChunk(0)
+		if err != nil {
+			panic(err)
+		}
+
+		dataOffset, dictOffset := colChunk.DataPageOffset(), colChunk.DictionaryPageOffset()
+		if colChunk.HasDictionaryPage() && dictOffset < dataOffset {
+			splitOffsets = append(splitOffsets, dictOffset)
+		} else {
+			splitOffsets = append(splitOffsets, dataOffset)
+		}
+
+		for pos := range rowGroup.NumColumns() {
+			colChunk, err = rowGroup.ColumnChunk(pos)
+			if err != nil {
+				panic(err)
+			}
+
+			fieldID := colMapping[colChunk.PathInSchema().String()]
+			statsCol := statsCols[fieldID]
+			if statsCol.mode.typ == metricModeNone {
+				continue
+			}
+
+			colSizes[fieldID] += colChunk.TotalCompressedSize()
+			valueCounts[fieldID] += colChunk.NumValues()
+			set, err := colChunk.StatsSet()
+			if err != nil {
+				panic(err)
+			}
+
+			if !set {
+				invalidateCol[fieldID] = struct{}{}
+
+				continue
+			}
+
+			stats, err := colChunk.Statistics()
+			if err != nil {
+				invalidateCol[fieldID] = struct{}{}
+
+				continue
+			}
+
+			if stats.HasNullCount() {
+				nullValueCounts[fieldID] += stats.NullCount()
+			}
+
+			if statsCol.mode.typ == metricModeCounts {
+				continue
+			}
+
+			// TODO stats aggregator!
+		}
+
+	}
+
+	slices.Sort(splitOffsets)
+	maps.DeleteFunc(nullValueCounts, func(fieldID int, _ int64) bool {
+		_, ok := invalidateCol[fieldID]
+
+		return ok
+	})
+
+	return &dataFileStatistics{
+		recordCount:     pqmeta.GetNumRows(),
+		colSizes:        colSizes,
+		valueCounts:     valueCounts,
+		nullValueCounts: nullValueCounts,
+		nanValueCounts:  nanValueCounts,
+		splitOffsets:    splitOffsets,
+	}
+}
diff --git a/table/internal/interfaces.go b/table/internal/interfaces.go
index 9d837fe..d5697bf 100644
--- a/table/internal/interfaces.go
+++ b/table/internal/interfaces.go
@@ -58,7 +58,7 @@ type FileReader interface {
 	// that represents the underlying file schema with only the projected fields. It also
 	// returns the indexes of the projected columns to allow reading *only* the needed
 	// columns.
-	PrunedSchema(projectedIDs map[int]struct{}) (*arrow.Schema, []int, error)
+	PrunedSchema(projectedIDs map[int]struct{}, mapping iceberg.NameMapping) (*arrow.Schema, []int, error)
 	// GetRecords returns a record reader for only the provided columns (using nil will read
 	// all of the columns of the underlying file.) The `tester` is a function that can be used,
 	// if non-nil, to filter aspects of the file such as skipping row groups in a parquet file.
diff --git a/table/internal/parquet_files.go b/table/internal/parquet_files.go
index 0afa44f..d1b5059 100644
--- a/table/internal/parquet_files.go
+++ b/table/internal/parquet_files.go
@@ -49,8 +49,8 @@ func (w wrapPqArrowReader) Close() error {
 	return w.ParquetReader().Close()
 }
 
-func (w wrapPqArrowReader) PrunedSchema(projectedIDs map[int]struct{}) (*arrow.Schema, []int, error) {
-	return pruneParquetColumns(w.Manifest, projectedIDs, false)
+func (w wrapPqArrowReader) PrunedSchema(projectedIDs map[int]struct{}, mapping iceberg.NameMapping) (*arrow.Schema, []int, error) {
+	return pruneParquetColumns(w.Manifest, projectedIDs, false, mapping)
 }
 
 func (w wrapPqArrowReader) GetRecords(ctx context.Context, cols []int, tester any) (array.RecordReader, error) {
@@ -118,15 +118,15 @@ func (pfs *ParquetFileSource) GetReader(ctx context.Context) (FileReader, error)
 }
 
 type manifestVisitor[T any] interface {
-	Manifest(*pqarrow.SchemaManifest, []T) T
-	Field(pqarrow.SchemaField, T) T
-	Struct(pqarrow.SchemaField, []T) T
-	List(pqarrow.SchemaField, T) T
-	Map(pqarrow.SchemaField, T, T) T
-	Primitive(pqarrow.SchemaField) T
+	Manifest(*pqarrow.SchemaManifest, []T, *iceberg.MappedField) T
+	Field(pqarrow.SchemaField, T, *iceberg.MappedField) T
+	Struct(pqarrow.SchemaField, []T, *iceberg.MappedField) T
+	List(pqarrow.SchemaField, T, *iceberg.MappedField) T
+	Map(pqarrow.SchemaField, T, T, *iceberg.MappedField) T
+	Primitive(pqarrow.SchemaField, *iceberg.MappedField) T
 }
 
-func visitParquetManifest[T any](manifest *pqarrow.SchemaManifest, visitor manifestVisitor[T]) (res T, err error) {
+func visitParquetManifest[T any](manifest *pqarrow.SchemaManifest, visitor manifestVisitor[T], mapping *iceberg.MappedField) (res T, err error) {
 	if manifest == nil {
 		err = fmt.Errorf("%w: cannot visit nil manifest", iceberg.ErrInvalidArgument)
 
@@ -139,54 +139,71 @@ func visitParquetManifest[T any](manifest *pqarrow.SchemaManifest, visitor manif
 		}
 	}()
 
+	var fieldMap *iceberg.MappedField
+
 	results := make([]T, len(manifest.Fields))
 	for i, f := range manifest.Fields {
-		res := visitManifestField(f, visitor)
-		results[i] = visitor.Field(f, res)
+		if mapping != nil {
+			fieldMap = mapping.GetField(f.Field.Name)
+		}
+		res := visitManifestField(f, visitor, fieldMap)
+		results[i] = visitor.Field(f, res, fieldMap)
 	}
 
-	return visitor.Manifest(manifest, results), nil
+	return visitor.Manifest(manifest, results, mapping), nil
 }
 
-func visitParquetManifestStruct[T any](field pqarrow.SchemaField, visitor manifestVisitor[T]) T {
+func visitParquetManifestStruct[T any](field pqarrow.SchemaField, visitor manifestVisitor[T], mapping *iceberg.MappedField) T {
 	results := make([]T, len(field.Children))
-
+	var fieldMap *iceberg.MappedField
 	for i, f := range field.Children {
-		res := visitManifestField(f, visitor)
-		results[i] = visitor.Field(f, res)
+		if mapping != nil {
+			fieldMap = mapping.GetField(f.Field.Name)
+		}
+		res := visitManifestField(f, visitor, fieldMap)
+		results[i] = visitor.Field(f, res, fieldMap)
 	}
 
-	return visitor.Struct(field, results)
+	return visitor.Struct(field, results, mapping)
 }
 
-func visitManifestList[T any](field pqarrow.SchemaField, visitor manifestVisitor[T]) T {
+func visitManifestList[T any](field pqarrow.SchemaField, visitor manifestVisitor[T], mapping *iceberg.MappedField) T {
 	elemField := field.Children[0]
-	res := visitManifestField(elemField, visitor)
+	var elemMapping *iceberg.MappedField
+	if mapping != nil {
+		elemMapping = mapping.GetField("element")
+	}
+	res := visitManifestField(elemField, visitor, elemMapping)
 
-	return visitor.List(field, res)
+	return visitor.List(field, res, mapping)
 }
 
-func visitManifestMap[T any](field pqarrow.SchemaField, visitor manifestVisitor[T]) T {
+func visitManifestMap[T any](field pqarrow.SchemaField, visitor manifestVisitor[T], mapping *iceberg.MappedField) T {
 	kvfield := field.Children[0]
 	keyField, valField := kvfield.Children[0], kvfield.Children[1]
+	var keyMapping, valMapping *iceberg.MappedField
+	if mapping != nil {
+		keyMapping = mapping.GetField("key")
+		valMapping = mapping.GetField("value")
+	}
 
-	return visitor.Map(field, visitManifestField(keyField, visitor), visitManifestField(valField, visitor))
+	return visitor.Map(field, visitManifestField(keyField, visitor, keyMapping), visitManifestField(valField, visitor, valMapping), mapping)
 }
 
-func visitManifestField[T any](field pqarrow.SchemaField, visitor manifestVisitor[T]) T {
+func visitManifestField[T any](field pqarrow.SchemaField, visitor manifestVisitor[T], mapping *iceberg.MappedField) T {
 	switch field.Field.Type.(type) {
 	case *arrow.StructType:
-		return visitParquetManifestStruct(field, visitor)
+		return visitParquetManifestStruct(field, visitor, mapping)
 	case *arrow.MapType:
-		return visitManifestMap(field, visitor)
+		return visitManifestMap(field, visitor, mapping)
 	case arrow.ListLikeType:
-		return visitManifestList(field, visitor)
+		return visitManifestList(field, visitor, mapping)
 	default:
-		return visitor.Primitive(field)
+		return visitor.Primitive(field, mapping)
 	}
 }
 
-func pruneParquetColumns(manifest *pqarrow.SchemaManifest, selected map[int]struct{}, selectFullTypes bool) (*arrow.Schema, []int, error) {
+func pruneParquetColumns(manifest *pqarrow.SchemaManifest, selected map[int]struct{}, selectFullTypes bool, mapping iceberg.NameMapping) (*arrow.Schema, []int, error) {
 	visitor := &pruneParquetSchema{
 		selected:  selected,
 		manifest:  manifest,
@@ -194,7 +211,7 @@ func pruneParquetColumns(manifest *pqarrow.SchemaManifest, selected map[int]stru
 		indices:   []int{},
 	}
 
-	result, err := visitParquetManifest(manifest, visitor)
+	result, err := visitParquetManifest(manifest, visitor, &iceberg.MappedField{Fields: mapping})
 	if err != nil {
 		return nil, nil, err
 	}
@@ -229,7 +246,13 @@ type pruneParquetSchema struct {
 	indices []int
 }
 
-func (p *pruneParquetSchema) fieldID(field arrow.Field) int {
+func (p *pruneParquetSchema) fieldID(field arrow.Field, mapping *iceberg.MappedField) int {
+	if mapping != nil {
+		if mapping.FieldID != nil {
+			return *mapping.FieldID
+		}
+	}
+
 	if id := getFieldID(field); id != nil {
 		return *id
 	}
@@ -238,7 +261,7 @@ func (p *pruneParquetSchema) fieldID(field arrow.Field) int {
 		iceberg.ErrInvalidSchema, field))
 }
 
-func (p *pruneParquetSchema) Manifest(manifest *pqarrow.SchemaManifest, fields []arrow.Field) arrow.Field {
+func (p *pruneParquetSchema) Manifest(manifest *pqarrow.SchemaManifest, fields []arrow.Field, _ *iceberg.MappedField) arrow.Field {
 	finalFields := slices.DeleteFunc(fields, func(f arrow.Field) bool { return f.Type == nil })
 	result := arrow.Field{
 		Type: arrow.StructOf(finalFields...),
@@ -250,7 +273,7 @@ func (p *pruneParquetSchema) Manifest(manifest *pqarrow.SchemaManifest, fields [
 	return result
 }
 
-func (p *pruneParquetSchema) Struct(field pqarrow.SchemaField, children []arrow.Field) arrow.Field {
+func (p *pruneParquetSchema) Struct(field pqarrow.SchemaField, children []arrow.Field, _ *iceberg.MappedField) arrow.Field {
 	selected, fields := []arrow.Field{}, field.Children
 	sameType := true
 
@@ -285,8 +308,8 @@ func (p *pruneParquetSchema) Struct(field pqarrow.SchemaField, children []arrow.
 	return arrow.Field{}
 }
 
-func (p *pruneParquetSchema) Field(field pqarrow.SchemaField, result arrow.Field) arrow.Field {
-	_, ok := p.selected[p.fieldID(*field.Field)]
+func (p *pruneParquetSchema) Field(field pqarrow.SchemaField, result arrow.Field, mapping *iceberg.MappedField) arrow.Field {
+	_, ok := p.selected[p.fieldID(*field.Field, mapping)]
 	if !ok {
 		if result.Type != nil {
 			return result
@@ -315,8 +338,8 @@ func (p *pruneParquetSchema) Field(field pqarrow.SchemaField, result arrow.Field
 	return *field.Field
 }
 
-func (p *pruneParquetSchema) List(field pqarrow.SchemaField, elemResult arrow.Field) arrow.Field {
-	_, ok := p.selected[p.fieldID(*field.Children[0].Field)]
+func (p *pruneParquetSchema) List(field pqarrow.SchemaField, elemResult arrow.Field, mapping *iceberg.MappedField) arrow.Field {
+	_, ok := p.selected[p.fieldID(*field.Children[0].Field, mapping)]
 	if !ok {
 		if elemResult.Type != nil {
 			result := *field.Field
@@ -350,8 +373,13 @@ func (p *pruneParquetSchema) List(field pqarrow.SchemaField, elemResult arrow.Fi
 	return *field.Field
 }
 
-func (p *pruneParquetSchema) Map(field pqarrow.SchemaField, keyResult, valResult arrow.Field) arrow.Field {
-	_, ok := p.selected[p.fieldID(*field.Children[0].Children[1].Field)]
+func (p *pruneParquetSchema) Map(field pqarrow.SchemaField, keyResult, valResult arrow.Field, mapping *iceberg.MappedField) arrow.Field {
+	var valMapping *iceberg.MappedField
+	if mapping != nil {
+		valMapping = mapping.GetField("value")
+	}
+
+	_, ok := p.selected[p.fieldID(*field.Children[0].Children[1].Field, valMapping)]
 	if !ok {
 		if valResult.Type != nil {
 			result := *field.Field
@@ -360,10 +388,6 @@ func (p *pruneParquetSchema) Map(field pqarrow.SchemaField, keyResult, valResult
 			return result
 		}
 
-		if _, ok = p.selected[p.fieldID(*field.Children[0].Children[1].Field)]; ok {
-			return *field.Field
-		}
-
 		return arrow.Field{}
 	}
 
@@ -390,7 +414,7 @@ func (p *pruneParquetSchema) Map(field pqarrow.SchemaField, keyResult, valResult
 	return *field.Field
 }
 
-func (p *pruneParquetSchema) Primitive(_ pqarrow.SchemaField) arrow.Field {
+func (p *pruneParquetSchema) Primitive(_ pqarrow.SchemaField, _ *iceberg.MappedField) arrow.Field {
 	return arrow.Field{}
 }
 
diff --git a/table/metadata.go b/table/metadata.go
index a5cb877..6828979 100644
--- a/table/metadata.go
+++ b/table/metadata.go
@@ -114,8 +114,11 @@ type Metadata interface {
 	Properties() iceberg.Properties
 	// PreviousFiles returns the list of metadata log entries for the table.
 	PreviousFiles() iter.Seq[MetadataLogEntry]
-
 	Equals(Metadata) bool
+
+	NameMapping() iceberg.NameMapping
+
+	LastSequenceNumber() int64
 }
 
 type MetadataBuilder struct {
@@ -201,6 +204,16 @@ func MetadataBuilderFromBase(metadata Metadata) (*MetadataBuilder, error) {
 	return b, nil
 }
 
+func (b *MetadataBuilder) CurrentSpec() iceberg.PartitionSpec {
+	return b.specs[b.defaultSpecID]
+}
+
+func (b *MetadataBuilder) CurrentSchema() *iceberg.Schema {
+	s, _ := b.GetSchemaByID(b.currentSchemaID)
+
+	return s
+}
+
 func (b *MetadataBuilder) AddSchema(schema *iceberg.Schema, newLastColumnID int, initial bool) (*MetadataBuilder, error) {
 	if newLastColumnID < b.lastColumnId {
 		return nil, fmt.Errorf("%w: newLastColumnID %d, must be >= %d", iceberg.ErrInvalidArgument, newLastColumnID, b.lastColumnId)
@@ -602,8 +615,23 @@ func (b *MetadataBuilder) SnapshotByID(id int64) (*Snapshot, error) {
 	return nil, fmt.Errorf("snapshot with id %d not found", id)
 }
 
+func (b *MetadataBuilder) NameMapping() iceberg.NameMapping {
+	if nameMappingJson, ok := b.props[DefaultNameMappingKey]; ok {
+		nm := iceberg.NameMapping{}
+		if err := json.Unmarshal([]byte(nameMappingJson), &nm); err == nil {
+			return nm
+		}
+	}
+
+	return nil
+}
+
 func (b *MetadataBuilder) Build() (Metadata, error) {
 	common := b.buildCommonMetadata()
+	if err := common.validate(); err != nil {
+		return nil, err
+	}
+
 	switch b.formatVersion {
 	case 1:
 		schema, err := b.GetSchemaByID(b.currentSchemaID)
@@ -635,8 +663,8 @@ func (b *MetadataBuilder) Build() (Metadata, error) {
 		}
 
 		return &metadataV2{
-			LastSequenceNumber: lastSequenceNumber,
-			commonMetadata:     *common,
+			LastSeqNum:     lastSequenceNumber,
+			commonMetadata: *common,
 		}, nil
 
 	default:
@@ -930,6 +958,18 @@ func (c *commonMetadata) checkSortOrders() error {
 		ErrInvalidMetadata, c.DefaultSortOrderID, c.SortOrderList)
 }
 
+func (c *commonMetadata) constructRefs() {
+	if c.CurrentSnapshotID != nil {
+		_, ok := c.SnapshotRefs[MainBranch]
+		if !ok {
+			c.SnapshotRefs[MainBranch] = SnapshotRef{
+				SnapshotID:      *c.CurrentSnapshotID,
+				SnapshotRefType: BranchRef,
+			}
+		}
+	}
+}
+
 func (c *commonMetadata) validate() error {
 	if err := c.checkSchemas(); err != nil {
 		return err
@@ -943,6 +983,8 @@ func (c *commonMetadata) validate() error {
 		return err
 	}
 
+	c.constructRefs()
+
 	switch {
 	case c.LastUpdatedMS == 0:
 		// last-updated-ms is required
@@ -955,6 +997,17 @@ func (c *commonMetadata) validate() error {
 	return nil
 }
 
+func (c *commonMetadata) NameMapping() iceberg.NameMapping {
+	if nameMappingJson, ok := c.Props[DefaultNameMappingKey]; ok {
+		nm := iceberg.NameMapping{}
+		if err := json.Unmarshal([]byte(nameMappingJson), &nm); err == nil {
+			return nm
+		}
+	}
+
+	return nil
+}
+
 func (c *commonMetadata) Version() int { return c.FormatVersion }
 
 type metadataV1 struct {
@@ -964,6 +1017,8 @@ type metadataV1 struct {
 	commonMetadata
 }
 
+func (m *metadataV1) LastSequenceNumber() int64 { return 0 }
+
 func (m *metadataV1) Equals(other Metadata) bool {
 	rhs, ok := other.(*metadataV1)
 	if !ok {
@@ -1032,11 +1087,13 @@ func (m *metadataV1) ToV2() metadataV2 {
 }
 
 type metadataV2 struct {
-	LastSequenceNumber int64 `json:"last-sequence-number"`
+	LastSeqNum int64 `json:"last-sequence-number"`
 
 	commonMetadata
 }
 
+func (m *metadataV2) LastSequenceNumber() int64 { return m.LastSeqNum }
+
 func (m *metadataV2) Equals(other Metadata) bool {
 	rhs, ok := other.(*metadataV2)
 	if !ok {
@@ -1047,7 +1104,7 @@ func (m *metadataV2) Equals(other Metadata) bool {
 		return true
 	}
 
-	return m.LastSequenceNumber == rhs.LastSequenceNumber &&
+	return m.LastSeqNum == rhs.LastSeqNum &&
 		m.commonMetadata.Equals(&rhs.commonMetadata)
 }
 
diff --git a/table/properties.go b/table/properties.go
index 6c1d688..e2f9e8c 100644
--- a/table/properties.go
+++ b/table/properties.go
@@ -24,4 +24,11 @@ const (
 	WriteObjectStorePartitionedPathsDefault = true
 	ObjectStoreEnabledKey                   = "write.object-storage.enabled"
 	ObjectStoreEnabledDefault               = false
+
+	DefaultNameMappingKey = "schema.name-mapping.default"
+
+	MetricsModeColumnConfPrefix = "write.metadata.metrics.column"
+
+	DefaultWriteMetricsModeKey     = "write.metadata.metrics.default"
+	DefaultWriteMetricsModeDefault = "truncate(16)"
 )


Test patch:
diff --git a/table/name_mapping_test.go b/name_mapping_test.go
similarity index 77%
rename from table/name_mapping_test.go
rename to name_mapping_test.go
index c8f928b..599554f 100644
--- a/table/name_mapping_test.go
+++ b/name_mapping_test.go
@@ -15,39 +15,41 @@
 // specific language governing permissions and limitations
 // under the License.
 
-package table_test
+package iceberg_test
 
 import (
 	"encoding/json"
 	"testing"
 
-	"github.com/apache/iceberg-go/table"
+	"github.com/apache/iceberg-go"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
 
-var tableNameMappingNested = table.NameMapping{
+func makeID(v int) *int { return &v }
+
+var tableNameMappingNested = iceberg.NameMapping{
 	{FieldID: makeID(1), Names: []string{"foo"}},
 	{FieldID: makeID(2), Names: []string{"bar"}},
 	{FieldID: makeID(3), Names: []string{"baz"}},
 	{
 		FieldID: makeID(4), Names: []string{"qux"},
-		Fields: []table.MappedField{{FieldID: makeID(5), Names: []string{"element"}}},
+		Fields: []iceberg.MappedField{{FieldID: makeID(5), Names: []string{"element"}}},
 	},
-	{FieldID: makeID(6), Names: []string{"quux"}, Fields: []table.MappedField{
+	{FieldID: makeID(6), Names: []string{"quux"}, Fields: []iceberg.MappedField{
 		{FieldID: makeID(7), Names: []string{"key"}},
-		{FieldID: makeID(8), Names: []string{"value"}, Fields: []table.MappedField{
+		{FieldID: makeID(8), Names: []string{"value"}, Fields: []iceberg.MappedField{
 			{FieldID: makeID(9), Names: []string{"key"}},
 			{FieldID: makeID(10), Names: []string{"value"}},
 		}},
 	}},
-	{FieldID: makeID(11), Names: []string{"location"}, Fields: []table.MappedField{
-		{FieldID: makeID(12), Names: []string{"element"}, Fields: []table.MappedField{
+	{FieldID: makeID(11), Names: []string{"location"}, Fields: []iceberg.MappedField{
+		{FieldID: makeID(12), Names: []string{"element"}, Fields: []iceberg.MappedField{
 			{FieldID: makeID(13), Names: []string{"latitude"}},
 			{FieldID: makeID(14), Names: []string{"longitude"}},
 		}},
 	}},
-	{FieldID: makeID(15), Names: []string{"person"}, Fields: []table.MappedField{
+	{FieldID: makeID(15), Names: []string{"person"}, Fields: []iceberg.MappedField{
 		{FieldID: makeID(16), Names: []string{"name"}},
 		{FieldID: makeID(17), Names: []string{"age"}},
 	}},
@@ -57,22 +59,22 @@ func TestJsonMappedField(t *testing.T) {
 	tests := []struct {
 		name string
 		str  string
-		exp  table.MappedField
+		exp  iceberg.MappedField
 	}{
 		{
 			"simple", `{"field-id": 1, "names": ["id", "record_id"]}`,
-			table.MappedField{FieldID: makeID(1), Names: []string{"id", "record_id"}},
+			iceberg.MappedField{FieldID: makeID(1), Names: []string{"id", "record_id"}},
 		},
 		{
 			"with null fields", `{"field-id": 1, "names": ["id", "record_id"], "fields": null}`,
-			table.MappedField{FieldID: makeID(1), Names: []string{"id", "record_id"}},
+			iceberg.MappedField{FieldID: makeID(1), Names: []string{"id", "record_id"}},
 		},
-		{"no names", `{"field-id": 1, "names": []}`, table.MappedField{FieldID: makeID(1), Names: []string{}}},
+		{"no names", `{"field-id": 1, "names": []}`, iceberg.MappedField{FieldID: makeID(1), Names: []string{}}},
 	}
 
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			var n table.MappedField
+			var n iceberg.MappedField
 			require.NoError(t, json.Unmarshal([]byte(tt.str), &n))
 			assert.Equal(t, tt.exp, n)
 		})
@@ -90,14 +92,14 @@ func TestNameMappingFromJson(t *testing.T) {
 		]}
 	]`
 
-	var nm table.NameMapping
+	var nm iceberg.NameMapping
 	require.NoError(t, json.Unmarshal([]byte(mapping), &nm))
 
-	assert.Equal(t, nm, table.NameMapping{
+	assert.Equal(t, nm, iceberg.NameMapping{
 		{FieldID: nil, Names: []string{"foo", "bar"}},
 		{FieldID: makeID(1), Names: []string{"id", "record_id"}},
 		{FieldID: makeID(2), Names: []string{"data"}},
-		{FieldID: makeID(3), Names: []string{"location"}, Fields: []table.MappedField{
+		{FieldID: makeID(3), Names: []string{"location"}, Fields: []iceberg.MappedField{
 			{FieldID: makeID(4), Names: []string{"latitude", "lat"}},
 			{FieldID: makeID(5), Names: []string{"longitude", "long"}},
 		}},
@@ -138,11 +140,11 @@ func TestNameMappingToString(t *testing.T) {
 	([id, record_id] -> 1)
 	([data] -> 2)
 	([location] -> 3 ([lat, latitude] -> 4), ([long, longitude] -> 5))
-]`, table.NameMapping{
+]`, iceberg.NameMapping{
 		{Names: []string{"foo"}},
 		{FieldID: makeID(1), Names: []string{"id", "record_id"}},
 		{FieldID: makeID(2), Names: []string{"data"}},
-		{FieldID: makeID(3), Names: []string{"location"}, Fields: []table.MappedField{
+		{FieldID: makeID(3), Names: []string{"location"}, Fields: []iceberg.MappedField{
 			{FieldID: makeID(4), Names: []string{"lat", "latitude"}},
 			{FieldID: makeID(5), Names: []string{"long", "longitude"}},
 		}},
diff --git a/table/arrow_utils_internal_test.go b/table/arrow_utils_internal_test.go
new file mode 100644
index 0000000..85730d1
--- /dev/null
+++ b/table/arrow_utils_internal_test.go
@@ -0,0 +1,410 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package table
+
+import (
+	"bytes"
+	"cmp"
+	"math"
+	"slices"
+	"testing"
+
+	"github.com/apache/arrow-go/v18/arrow/array"
+	"github.com/apache/arrow-go/v18/arrow/memory"
+	"github.com/apache/arrow-go/v18/parquet"
+	"github.com/apache/arrow-go/v18/parquet/file"
+	"github.com/apache/arrow-go/v18/parquet/metadata"
+	"github.com/apache/arrow-go/v18/parquet/pqarrow"
+	"github.com/apache/iceberg-go"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"github.com/stretchr/testify/suite"
+)
+
+func constructTestTable(t *testing.T, writeStats bool) (*metadata.FileMetaData, Metadata) {
+	tableMeta, err := ParseMetadataString(`{
+		"format-version": 2,
+        "location": "s3://bucket/test/location",
+        "last-column-id": 7,
+        "current-schema-id": 0,
+		"last-updated-ms": -1,
+        "schemas": [
+            {
+                "type": "struct",
+                "schema-id": 0,
+                "fields": [
+                    {"id": 1, "name": "strings", "required": false, "type": "string"},
+                    {"id": 2, "name": "floats", "required": false, "type": "float"},
+                    {
+                        "id": 3,
+                        "name": "list",
+                        "required": false,
+                        "type": {"type": "list", "element-id": 6, "element": "long", "element-required": false}
+                    },
+                    {
+                        "id": 4,
+                        "name": "maps",
+                        "required": false,
+                        "type": {
+                            "type": "map",
+                            "key-id": 7,
+                            "key": "long",
+                            "value-id": 8,
+                            "value": "long",
+                            "value-required": false
+                        }
+                    },
+                    {
+                        "id": 5,
+                        "name": "structs",
+                        "required": false,
+                        "type": {
+                            "type": "struct",
+                            "fields": [
+                                {"id": 9, "name": "x", "required": false, "type": "long"},
+                                {"id": 10, "name": "y", "required": false, "type": "float", "doc": "comment"}
+                            ]
+                        }
+                    }
+                ]
+            }
+        ],
+        "default-spec-id": 0,
+        "partition-specs": [{"spec-id": 0, "fields": []}],
+        "properties": {}
+	}`)
+	require.NoError(t, err)
+
+	arrowSchema, err := SchemaToArrowSchema(tableMeta.Schemas()[0], nil, true, false)
+	require.NoError(t, err)
+
+	bldr := array.NewRecordBuilder(memory.DefaultAllocator, arrowSchema)
+	defer bldr.Release()
+
+	bldr.Field(0).(*array.StringBuilder).AppendValues([]string{
+		"zzzzzzzzzzzzzzzzzzzz", "rrrrrrrrrrrrrrrrrrrr", "", "aaaaaaaaaaaaaaaaaaaa",
+	},
+		[]bool{true, true, false, true})
+	bldr.Field(1).(*array.Float32Builder).AppendValues(
+		[]float32{3.14, float32(math.NaN()), 1.69, 100}, nil)
+	lb := bldr.Field(2).(*array.ListBuilder)
+	vb := lb.ValueBuilder().(*array.Int64Builder)
+	lb.Append(true)
+	vb.AppendValues([]int64{1, 2, 3}, nil)
+	lb.Append(true)
+	vb.AppendValues([]int64{4, 5, 6}, nil)
+	lb.AppendNull()
+	lb.Append(true)
+	vb.AppendValues([]int64{7, 8, 9}, nil)
+
+	mb := bldr.Field(3).(*array.MapBuilder)
+	kb := mb.KeyBuilder().(*array.Int64Builder)
+	vb = mb.ItemBuilder().(*array.Int64Builder)
+	mb.Append(true)
+	kb.AppendValues([]int64{1, 2}, nil)
+	vb.AppendValues([]int64{3, 4}, nil)
+	mb.AppendNull()
+	mb.Append(true)
+	kb.Append(5)
+	vb.Append(6)
+	mb.Append(true)
+
+	sb := bldr.Field(4).(*array.StructBuilder)
+	xb := sb.FieldBuilder(0).(*array.Int64Builder)
+	yb := sb.FieldBuilder(1).(*array.Float32Builder)
+	sb.Append(true)
+	xb.Append(1)
+	yb.Append(0.2)
+	sb.Append(true)
+	xb.AppendNull()
+	yb.Append(-1.34)
+	sb.AppendNull()
+	sb.Append(true)
+	xb.Append(54)
+	yb.AppendNull()
+
+	rec := bldr.NewRecord()
+	defer rec.Release()
+
+	var buf bytes.Buffer
+	wr, err := pqarrow.NewFileWriter(arrowSchema, &buf,
+		parquet.NewWriterProperties(parquet.WithStats(writeStats)),
+		pqarrow.DefaultWriterProps())
+	require.NoError(t, err)
+
+	require.NoError(t, wr.Write(rec))
+	require.NoError(t, wr.Close())
+
+	rdr, err := file.NewParquetReader(bytes.NewReader(buf.Bytes()))
+	require.NoError(t, err)
+	defer rdr.Close()
+
+	return rdr.MetaData(), tableMeta
+}
+
+type FileStatsMetricsSuite struct {
+	suite.Suite
+}
+
+func (suite *FileStatsMetricsSuite) getDataFile(meta iceberg.Properties) iceberg.DataFile {
+	fileMeta, tableMeta := constructTestTable(suite.T(), true)
+	require.NotNil(suite.T(), tableMeta)
+	require.NotNil(suite.T(), fileMeta)
+
+	schema := tableMeta.CurrentSchema()
+	if len(meta) > 0 {
+		bldr, err := MetadataBuilderFromBase(tableMeta)
+		suite.Require().NoError(err)
+		_, err = bldr.SetProperties(meta)
+		suite.Require().NoError(err)
+		tableMeta, err = bldr.Build()
+		suite.Require().NoError(err)
+	}
+
+	collector, err := computeStatsPlan(schema, tableMeta.Properties())
+	suite.Require().NoError(err)
+	mapping, err := parquetPathToIDMapping(schema)
+	suite.Require().NoError(err)
+
+	stats := dataFileStatsFromParquetMetadata(fileMeta, collector, mapping)
+	df, err := stats.toDataFile(tableMeta.PartitionSpec(), "fake-path.parquet",
+		iceberg.ParquetFile, fileMeta.GetSourceFileSize())
+	suite.Require().NoError(err)
+
+	return df
+}
+
+func (suite *FileStatsMetricsSuite) TestRecordCount() {
+	df := suite.getDataFile(nil)
+	suite.EqualValues(int64(4), df.Count())
+}
+
+func (suite *FileStatsMetricsSuite) TestValueCounts() {
+	df := suite.getDataFile(nil)
+	suite.Len(df.ValueCounts(), 7)
+	suite.EqualValues(4, df.ValueCounts()[1])
+	suite.EqualValues(4, df.ValueCounts()[2])
+	suite.EqualValues(10, df.ValueCounts()[6])
+	suite.EqualValues(5, df.ValueCounts()[7])
+	suite.EqualValues(5, df.ValueCounts()[8])
+	suite.EqualValues(4, df.ValueCounts()[9])
+	suite.EqualValues(4, df.ValueCounts()[10])
+}
+
+func (suite *FileStatsMetricsSuite) TestColumnSizes() {
+	df := suite.getDataFile(nil)
+	suite.Len(df.ColumnSizes(), 7)
+	suite.Greater(df.ColumnSizes()[1], int64(0))
+	suite.Greater(df.ColumnSizes()[2], int64(0))
+	suite.Greater(df.ColumnSizes()[6], int64(0))
+	suite.Greater(df.ColumnSizes()[7], int64(0))
+	suite.Greater(df.ColumnSizes()[8], int64(0))
+}
+
+func (suite *FileStatsMetricsSuite) TestOffsets() {
+	df := suite.getDataFile(nil)
+	suite.Len(df.SplitOffsets(), 1)
+	suite.EqualValues(4, df.SplitOffsets()[0])
+}
+
+func (suite *FileStatsMetricsSuite) TestNullValueCounts() {
+	df := suite.getDataFile(nil)
+	suite.Len(df.NullValueCounts(), 7)
+	suite.EqualValues(1, df.NullValueCounts()[1])
+	suite.EqualValues(0, df.NullValueCounts()[2])
+	suite.EqualValues(1, df.NullValueCounts()[6])
+	suite.EqualValues(2, df.NullValueCounts()[7])
+	suite.EqualValues(2, df.NullValueCounts()[8])
+	suite.EqualValues(2, df.NullValueCounts()[9])
+	suite.EqualValues(2, df.NullValueCounts()[10])
+
+	// pqarrow doesn't currently write the NaN counts
+}
+
+func (suite *FileStatsMetricsSuite) TestMetricsModeNone() {
+	df := suite.getDataFile(iceberg.Properties{"write.metadata.metrics.default": "none"})
+	suite.Len(df.ValueCounts(), 0)
+	suite.Len(df.ColumnSizes(), 0)
+	suite.Len(df.NullValueCounts(), 0)
+	suite.Len(df.NaNValueCounts(), 0)
+	suite.Len(df.LowerBoundValues(), 0)
+	suite.Len(df.UpperBoundValues(), 0)
+}
+
+func (suite *FileStatsMetricsSuite) TestMetricsModeCounts() {
+	df := suite.getDataFile(iceberg.Properties{"write.metadata.metrics.default": "counts"})
+	suite.Len(df.ValueCounts(), 7)
+	suite.Len(df.NullValueCounts(), 7)
+	suite.Len(df.NaNValueCounts(), 0)
+	suite.Len(df.LowerBoundValues(), 0)
+	suite.Len(df.UpperBoundValues(), 0)
+}
+
+func (suite *FileStatsMetricsSuite) TestMetricsModeFull() {
+	df := suite.getDataFile(iceberg.Properties{"write.metadata.metrics.default": "full"})
+	suite.Len(df.ValueCounts(), 7)
+	suite.Len(df.NullValueCounts(), 7)
+	suite.Len(df.NaNValueCounts(), 0)
+	// bound values not yet implemented
+}
+
+func (suite *FileStatsMetricsSuite) TestMetricsModeNonDefaultTrunc() {
+	df := suite.getDataFile(iceberg.Properties{"write.metadata.metrics.default": "truncate(2)"})
+	suite.Len(df.ValueCounts(), 7)
+	suite.Len(df.NullValueCounts(), 7)
+	suite.Len(df.NaNValueCounts(), 0)
+	// bound values not yet implemented
+}
+
+func (suite *FileStatsMetricsSuite) TestColumnMetricsMode() {
+	df := suite.getDataFile(iceberg.Properties{
+		"write.metadata.metrics.default":        "truncate(2)",
+		"write.metadata.metrics.column.strings": "none",
+	})
+
+	suite.Len(df.ValueCounts(), 6)
+	suite.Len(df.NullValueCounts(), 6)
+	suite.Len(df.NaNValueCounts(), 0)
+	// bound values not yet implemented
+}
+
+func TestFileMetrics(t *testing.T) {
+	suite.Run(t, new(FileStatsMetricsSuite))
+}
+
+func TestMetricsModePairs(t *testing.T) {
+	tests := []struct {
+		str      string
+		expected metricsMode
+	}{
+		{"none", metricsMode{typ: metricModeNone}},
+		{"nOnE", metricsMode{typ: metricModeNone}},
+		{"counts", metricsMode{typ: metricModeCounts}},
+		{"Counts", metricsMode{typ: metricModeCounts}},
+		{"full", metricsMode{typ: metricModeFull}},
+		{"FuLl", metricsMode{typ: metricModeFull}},
+		{" FuLl", metricsMode{typ: metricModeFull}},
+		{"truncate(16)", metricsMode{typ: metricModeTruncate, len: 16}},
+		{"trUncatE(16)", metricsMode{typ: metricModeTruncate, len: 16}},
+		{"trUncatE(7)", metricsMode{typ: metricModeTruncate, len: 7}},
+		{"trUncatE(07)", metricsMode{typ: metricModeTruncate, len: 7}},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.str, func(t *testing.T) {
+			mode, err := matchMetricsMode(tt.str)
+			require.NoError(t, err)
+			assert.Equal(t, tt.expected, mode)
+		})
+	}
+
+	_, err := matchMetricsMode("truncatE(-7)")
+	assert.ErrorContains(t, err, "malformed truncate metrics mode: truncatE(-7)")
+
+	_, err = matchMetricsMode("truncatE(0)")
+	assert.ErrorContains(t, err, "invalid truncate length: 0")
+}
+
+var tableSchemaNested = iceberg.NewSchemaWithIdentifiers(1,
+	[]int{1},
+	iceberg.NestedField{
+		ID: 1, Name: "foo", Type: iceberg.PrimitiveTypes.String, Required: false,
+	},
+	iceberg.NestedField{
+		ID: 2, Name: "bar", Type: iceberg.PrimitiveTypes.Int32, Required: true,
+	},
+	iceberg.NestedField{
+		ID: 3, Name: "baz", Type: iceberg.PrimitiveTypes.Bool, Required: false,
+	},
+	iceberg.NestedField{
+		ID: 4, Name: "qux", Required: true, Type: &iceberg.ListType{
+			ElementID: 5, Element: iceberg.PrimitiveTypes.String, ElementRequired: true,
+		},
+	},
+	iceberg.NestedField{
+		ID: 6, Name: "quux",
+		Type: &iceberg.MapType{
+			KeyID:   7,
+			KeyType: iceberg.PrimitiveTypes.String,
+			ValueID: 8,
+			ValueType: &iceberg.MapType{
+				KeyID:         9,
+				KeyType:       iceberg.PrimitiveTypes.String,
+				ValueID:       10,
+				ValueType:     iceberg.PrimitiveTypes.Int32,
+				ValueRequired: true,
+			},
+			ValueRequired: true,
+		},
+		Required: true,
+	},
+	iceberg.NestedField{
+		ID: 11, Name: "location", Type: &iceberg.ListType{
+			ElementID: 12, Element: &iceberg.StructType{
+				FieldList: []iceberg.NestedField{
+					{ID: 13, Name: "latitude", Type: iceberg.PrimitiveTypes.Float32, Required: false},
+					{ID: 14, Name: "longitude", Type: iceberg.PrimitiveTypes.Float32, Required: false},
+				},
+			},
+			ElementRequired: true,
+		},
+		Required: true,
+	},
+	iceberg.NestedField{
+		ID:   15,
+		Name: "person",
+		Type: &iceberg.StructType{
+			FieldList: []iceberg.NestedField{
+				{ID: 16, Name: "name", Type: iceberg.PrimitiveTypes.String, Required: false},
+				{ID: 17, Name: "age", Type: iceberg.PrimitiveTypes.Int32, Required: true},
+			},
+		},
+		Required: false,
+	},
+)
+
+func TestStatsTypes(t *testing.T) {
+	statsCols, err := iceberg.PreOrderVisit(tableSchemaNested,
+		&arrowStatsCollector{schema: tableSchemaNested, defaultMode: "full"})
+
+	require.NoError(t, err)
+
+	// field ids should be sorted
+	assert.True(t, slices.IsSortedFunc(statsCols, func(a, b statisticsCollector) int {
+		return cmp.Compare(a.fieldID, b.fieldID)
+	}))
+
+	actual := make([]iceberg.Type, len(statsCols))
+	for i, col := range statsCols {
+		actual[i] = col.icebergTyp
+	}
+
+	assert.Equal(t, []iceberg.Type{
+		iceberg.PrimitiveTypes.String,
+		iceberg.PrimitiveTypes.Int32,
+		iceberg.PrimitiveTypes.Bool,
+		iceberg.PrimitiveTypes.String,
+		iceberg.PrimitiveTypes.String,
+		iceberg.PrimitiveTypes.String,
+		iceberg.PrimitiveTypes.Int32,
+		iceberg.PrimitiveTypes.Float32,
+		iceberg.PrimitiveTypes.Float32,
+		iceberg.PrimitiveTypes.String,
+		iceberg.PrimitiveTypes.Int32,
+	}, actual)
+}
diff --git a/table/arrow_utils_test.go b/table/arrow_utils_test.go
index 844cec2..29d74d1 100644
--- a/table/arrow_utils_test.go
+++ b/table/arrow_utils_test.go
@@ -385,40 +385,40 @@ func TestArrowSchemaWithNameMapping(t *testing.T) {
 	tests := []struct {
 		name     string
 		schema   *arrow.Schema
-		mapping  table.NameMapping
+		mapping  iceberg.NameMapping
 		expected *iceberg.Schema
 		err      string
 	}{
-		{"simple", schemaWithoutIDs, table.NameMapping{
+		{"simple", schemaWithoutIDs, iceberg.NameMapping{
 			{FieldID: makeID(1), Names: []string{"foo"}},
 			{FieldID: makeID(2), Names: []string{"bar"}},
 			{FieldID: makeID(3), Names: []string{"baz"}},
 		}, icebergSchemaSimple, ""},
-		{"field missing", schemaWithoutIDs, table.NameMapping{
+		{"field missing", schemaWithoutIDs, iceberg.NameMapping{
 			{FieldID: makeID(1), Names: []string{"foo"}},
 		}, nil, "field missing from name mapping: bar"},
-		{"nested schema", schemaNestedWithoutIDs, table.NameMapping{
+		{"nested schema", schemaNestedWithoutIDs, iceberg.NameMapping{
 			{FieldID: makeID(1), Names: []string{"foo"}},
 			{FieldID: makeID(2), Names: []string{"bar"}},
 			{FieldID: makeID(3), Names: []string{"baz"}},
 			{
 				FieldID: makeID(4), Names: []string{"qux"},
-				Fields: []table.MappedField{{FieldID: makeID(5), Names: []string{"element"}}},
+				Fields: []iceberg.MappedField{{FieldID: makeID(5), Names: []string{"element"}}},
 			},
-			{FieldID: makeID(6), Names: []string{"quux"}, Fields: []table.MappedField{
+			{FieldID: makeID(6), Names: []string{"quux"}, Fields: []iceberg.MappedField{
 				{FieldID: makeID(7), Names: []string{"key"}},
-				{FieldID: makeID(8), Names: []string{"value"}, Fields: []table.MappedField{
+				{FieldID: makeID(8), Names: []string{"value"}, Fields: []iceberg.MappedField{
 					{FieldID: makeID(9), Names: []string{"key"}},
 					{FieldID: makeID(10), Names: []string{"value"}},
 				}},
 			}},
-			{FieldID: makeID(11), Names: []string{"location"}, Fields: []table.MappedField{
-				{FieldID: makeID(12), Names: []string{"element"}, Fields: []table.MappedField{
+			{FieldID: makeID(11), Names: []string{"location"}, Fields: []iceberg.MappedField{
+				{FieldID: makeID(12), Names: []string{"element"}, Fields: []iceberg.MappedField{
 					{FieldID: makeID(13), Names: []string{"latitude"}},
 					{FieldID: makeID(14), Names: []string{"longitude"}},
 				}},
 			}},
-			{FieldID: makeID(15), Names: []string{"person"}, Fields: []table.MappedField{
+			{FieldID: makeID(15), Names: []string{"person"}, Fields: []iceberg.MappedField{
 				{FieldID: makeID(16), Names: []string{"name"}},
 				{FieldID: makeID(17), Names: []string{"age"}},
 			}},
diff --git a/table/metadata_internal_test.go b/table/metadata_internal_test.go
index 00c6c03..e28dfd4 100644
--- a/table/metadata_internal_test.go
+++ b/table/metadata_internal_test.go
@@ -169,7 +169,7 @@ func TestMetadataV2Parsing(t *testing.T) {
 	data := meta.(*metadataV2)
 	assert.Equal(t, uuid.MustParse("9c12d441-03fe-4693-9a96-a0705ddf69c1"), data.UUID)
 	assert.Equal(t, "s3://bucket/test/location", data.Location())
-	assert.Equal(t, int64(34), data.LastSequenceNumber)
+	assert.Equal(t, int64(34), data.LastSeqNum)
 	assert.Equal(t, int64(1602638573590), data.LastUpdatedMS)
 	assert.Equal(t, 3, data.LastColumnId)
 	assert.Equal(t, 0, data.SchemaList[0].ID)
diff --git a/table/snapshots_internal_test.go b/table/snapshots_internal_test.go
index 4063ad8..c374178 100644
--- a/table/snapshots_internal_test.go
+++ b/table/snapshots_internal_test.go
@@ -32,14 +32,6 @@ var tableSchemaSimple = iceberg.NewSchemaWithIdentifiers(1,
 	iceberg.NestedField{ID: 3, Name: "baz", Type: iceberg.PrimitiveTypes.Bool},
 )
 
-func must[T any](v T, err error) T {
-	if err != nil {
-		panic(err)
-	}
-
-	return v
-}
-
 func TestSnapshotSummaryCollector(t *testing.T) {
 	var ssc SnapshotSummaryCollector