Repository: apache/iceberg-go
Pull request #353 (2025-03-20 17:23:41)

refactor(table): shift parquet-specific details to internal package
Refactoring the table writing code and abstracting the parquet-specific details from the general file writing logic so that it will be easier in the future to add support for other file types.


Patch:
diff --git a/README.md b/README.md
index 30ebcaa..2bfe7bb 100644
--- a/README.md
+++ b/README.md
@@ -65,6 +65,7 @@ $ cd iceberg-go/cmd/iceberg && go build .
 | Load Table               |  X   |      |          |  X   |  X  |
 | List Tables              |  X   |      |          |  X   |  X  |
 | Create Table             |  X   |      |          |  X   |  X  |
+| Register Table           |  X   |      |          |  X   |     |
 | Update Current Snapshot  |  X   |      |          |      |     |
 | Create New Snapshot      |  X   |      |          |      |     |
 | Rename Table             |  X   |      |          |  X   |  X  |
diff --git a/catalog/glue/glue.go b/catalog/glue/glue.go
index 3135d53..8e2bc13 100644
--- a/catalog/glue/glue.go
+++ b/catalog/glue/glue.go
@@ -217,7 +217,7 @@ func (c *Catalog) LoadTable(ctx context.Context, identifier table.Identifier, pr
 		return nil, fmt.Errorf("failed to load table %s.%s: %w", database, tableName, err)
 	}
 
-	icebergTable, err := table.NewFromLocation([]string{tableName}, location, iofs, c)
+	icebergTable, err := table.NewFromLocation(identifier, location, iofs, c)
 	if err != nil {
 		return nil, fmt.Errorf("failed to create table from location %s.%s: %w", database, tableName, err)
 	}
@@ -288,6 +288,49 @@ func (c *Catalog) CreateTable(ctx context.Context, identifier table.Identifier,
 	return createdTable, nil
 }
 
+// RegisterTable registers a new table using existing metadata.
+func (c *Catalog) RegisterTable(ctx context.Context, identifier table.Identifier, metadataLocation string) (*table.Table, error) {
+	database, tableName, err := identifierToGlueTable(identifier)
+	if err != nil {
+		return nil, err
+	}
+	// Load the metadata file to get table properties
+	ctx = utils.WithAwsConfig(ctx, c.awsCfg)
+	iofs, err := io.LoadFS(ctx, nil, metadataLocation)
+	if err != nil {
+		return nil, fmt.Errorf("failed to load metadata file at %s: %w", metadataLocation, err)
+	}
+	// Read the metadata file
+	metadata, err := table.NewFromLocation([]string{tableName}, metadataLocation, iofs, c)
+	if err != nil {
+		return nil, fmt.Errorf("failed to read table metadata from %s: %w", metadataLocation, err)
+	}
+	tableInput := &types.TableInput{
+		Name:       aws.String(tableName),
+		Parameters: map[string]string{},
+		TableType:  aws.String("EXTERNAL_TABLE"),
+		StorageDescriptor: &types.StorageDescriptor{
+			Location: aws.String(metadataLocation),
+			Columns:  schemaToGlueColumns(metadata.Schema()),
+		},
+	}
+	_, err = c.glueSvc.CreateTable(ctx, &glue.CreateTableInput{
+		CatalogId:    c.catalogId,
+		DatabaseName: aws.String(database),
+		TableInput:   tableInput,
+		OpenTableFormatInput: &types.OpenTableFormatInput{
+			IcebergInput: &types.IcebergInput{
+				MetadataOperation: types.MetadataOperationCreate,
+			},
+		},
+	})
+	if err != nil {
+		return nil, fmt.Errorf("failed to register table %s.%s: %w", database, tableName, err)
+	}
+
+	return c.LoadTable(ctx, identifier, nil)
+}
+
 func (c *Catalog) CommitTable(context.Context, *table.Table, []table.Requirement, []table.Update) (table.Metadata, string, error) {
 	panic("commit table not implemented for Glue Catalog")
 }
diff --git a/catalog/rest/options.go b/catalog/rest/options.go
index 5aae122..ae1782c 100644
--- a/catalog/rest/options.go
+++ b/catalog/rest/options.go
@@ -21,6 +21,7 @@ import (
 	"crypto/tls"
 	"net/url"
 
+	"github.com/apache/iceberg-go"
 	"github.com/aws/aws-sdk-go-v2/aws"
 )
 
@@ -100,6 +101,12 @@ func WithScope(scope string) Option {
 	}
 }
 
+func WithAdditionalProps(props iceberg.Properties) Option {
+	return func(o *options) {
+		o.additionalProps = props
+	}
+}
+
 type options struct {
 	awsConfig         aws.Config
 	tlsConfig         *tls.Config
@@ -113,4 +120,6 @@ type options struct {
 	prefix            string
 	authUri           *url.URL
 	scope             string
+
+	additionalProps iceberg.Properties
 }
diff --git a/catalog/rest/rest.go b/catalog/rest/rest.go
index cbeed0f..5eb870f 100644
--- a/catalog/rest/rest.go
+++ b/catalog/rest/rest.go
@@ -377,7 +377,9 @@ func handleNon200(rsp *http.Response, override map[int]error) error {
 }
 
 func fromProps(props iceberg.Properties) *options {
-	o := &options{}
+	o := &options{
+		additionalProps: iceberg.Properties{},
+	}
 	for k, v := range props {
 		switch k {
 		case keyOauthToken:
@@ -411,6 +413,11 @@ func fromProps(props iceberg.Properties) *options {
 			} else {
 				o.tlsConfig.InsecureSkipVerify = verify
 			}
+		case "uri", "type":
+		default:
+			if v != "" {
+				o.additionalProps[k] = v
+			}
 		}
 	}
 
@@ -418,7 +425,12 @@ func fromProps(props iceberg.Properties) *options {
 }
 
 func toProps(o *options) iceberg.Properties {
-	props := iceberg.Properties{}
+	var props iceberg.Properties
+	if o.additionalProps != nil {
+		props = o.additionalProps
+	} else {
+		props = iceberg.Properties{}
+	}
 
 	setIf := func(key, v string) {
 		if v != "" {
diff --git a/go.mod b/go.mod
index 95cd232..7838ffe 100644
--- a/go.mod
+++ b/go.mod
@@ -22,7 +22,7 @@ go 1.23.0
 toolchain go1.23.6
 
 require (
-	github.com/apache/arrow-go/v18 v18.2.0
+	github.com/apache/arrow-go/v18 v18.2.1-0.20250325140533-276892c275de
 	github.com/aws/aws-sdk-go-v2 v1.36.3
 	github.com/aws/aws-sdk-go-v2/config v1.29.9
 	github.com/aws/aws-sdk-go-v2/credentials v1.17.62
@@ -46,7 +46,7 @@ require (
 	github.com/uptrace/bun/extra/bundebug v1.2.11
 	gocloud.dev v0.40.0
 	golang.org/x/sync v0.12.0
-	google.golang.org/api v0.226.0
+	google.golang.org/api v0.227.0
 	gopkg.in/yaml.v3 v3.0.1
 )
 
diff --git a/go.sum b/go.sum
index 6531e58..fd35418 100644
--- a/go.sum
+++ b/go.sum
@@ -53,6 +53,8 @@ github.com/antlr4-go/antlr/v4 v4.13.1 h1:SqQKkuVZ+zWkMMNkjy5FZe5mr5WURWnlpmOuzYW
 github.com/antlr4-go/antlr/v4 v4.13.1/go.mod h1:GKmUxMtwp6ZgGwZSva4eWPC5mS6vUAmOABFgjdkM7Nw=
 github.com/apache/arrow-go/v18 v18.2.0 h1:QhWqpgZMKfWOniGPhbUxrHohWnooGURqL2R2Gg4SO1Q=
 github.com/apache/arrow-go/v18 v18.2.0/go.mod h1:Ic/01WSwGJWRrdAZcxjBZ5hbApNJ28K96jGYaxzzGUc=
+github.com/apache/arrow-go/v18 v18.2.1-0.20250325140533-276892c275de h1:wklDQWXjXSjWDoYXJvlPKii9nx6nXC8dEY/a4XRQOvI=
+github.com/apache/arrow-go/v18 v18.2.1-0.20250325140533-276892c275de/go.mod h1:EOkJNffq8UnNiioDTUIuynsY5mDoSojZHQZ5ELgGnWM=
 github.com/apache/thrift v0.21.0 h1:tdPmh/ptjE1IJnhbhrcl2++TauVjy242rkV/UzJChnE=
 github.com/apache/thrift v0.21.0/go.mod h1:W1H8aR/QRtYNvrPeFXBtobyRkd0/YVhTc6i07XIAgDw=
 github.com/atomicgo/cursor v0.0.1/go.mod h1:cBON2QmmrysudxNBFthvMtN32r3jxVRIvzkUiF/RuIk=
@@ -457,8 +459,8 @@ golang.org/x/xerrors v0.0.0-20240903120638-7835f813f4da h1:noIWHXmPHxILtqtCOPIhS
 golang.org/x/xerrors v0.0.0-20240903120638-7835f813f4da/go.mod h1:NDW/Ps6MPRej6fsCIbMTohpP40sJ/P/vI1MoTEGwX90=
 gonum.org/v1/gonum v0.15.1 h1:FNy7N6OUZVUaWG9pTiD+jlhdQ3lMP+/LcTpJ6+a8sQ0=
 gonum.org/v1/gonum v0.15.1/go.mod h1:eZTZuRFrzu5pcyjN5wJhcIhnUdNijYxX1T2IcrOGY0o=
-google.golang.org/api v0.226.0 h1:9A29y1XUD+YRXfnHkO66KggxHBZWg9LsTGqm7TkUvtQ=
-google.golang.org/api v0.226.0/go.mod h1:WP/0Xm4LVvMOCldfvOISnWquSRWbG2kArDZcg+W2DbY=
+google.golang.org/api v0.227.0 h1:QvIHF9IuyG6d6ReE+BNd11kIB8hZvjN8Z5xY5t21zYc=
+google.golang.org/api v0.227.0/go.mod h1:EIpaG6MbTgQarWF5xJvX0eOJPK9n/5D4Bynb9j2HXvQ=
 google.golang.org/appengine v1.1.0/go.mod h1:EbEs0AVv82hx2wNQdGPgUI5lhzA/G0D9YwlJXL52JkM=
 google.golang.org/appengine v1.4.0/go.mod h1:xpcJRLb0r/rnEns0DIKYYv+WjYCduHsrkT7/EB5XEv4=
 google.golang.org/genproto v0.0.0-20180817151627-c66870c02cf8/go.mod h1:JiN7NxoALGmiZfu7CAH4rXhgtRTLTxftemlI0sWmxmc=
diff --git a/internal/avro_schemas.go b/internal/avro_schemas.go
index 4fd4247..00fc01d 100644
--- a/internal/avro_schemas.go
+++ b/internal/avro_schemas.go
@@ -225,7 +225,7 @@ func init() {
 			WithFieldID(519))),
 	})))
 
-	AvroSchemaCache.Add("data_file_v1", Must(avro.NewRecordSchema("data_file", "", []*avro.Field{
+	AvroSchemaCache.Add("data_file_v1", Must(avro.NewRecordSchema("r2", "", []*avro.Field{
 		Must(avro.NewField("file_path",
 			StringSchema,
 			avro.WithDoc("Location URI with FS scheme"),
@@ -286,7 +286,7 @@ func init() {
 			WithFieldID(140))),
 	})))
 
-	AvroSchemaCache.Add("data_file_v2", Must(avro.NewRecordSchema("data_file", "", []*avro.Field{
+	AvroSchemaCache.Add("data_file_v2", Must(avro.NewRecordSchema("r2", "", []*avro.Field{
 		Must(avro.NewField("content", IntSchema,
 			avro.WithDoc("Content type"),
 			avro.WithDefault(0),
@@ -376,7 +376,7 @@ func newDataFileSchema(partitionType avro.Schema, version int) (avro.Schema, err
 		return nil, err
 	}
 
-	return avro.NewRecordSchema("data_file", "",
+	return avro.NewRecordSchema("r2", "",
 		append(schema.(*avro.RecordSchema).Fields(), partField))
 }
 
diff --git a/table/arrow_utils.go b/table/arrow_utils.go
index f1f5676..3271cc2 100644
--- a/table/arrow_utils.go
+++ b/table/arrow_utils.go
@@ -19,30 +19,21 @@ package table
 
 import (
 	"context"
-	"errors"
 	"fmt"
 	"iter"
-	"maps"
-	"regexp"
 	"slices"
 	"strconv"
 	"strings"
-	"unsafe"
 
 	"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/decimal128"
 	"github.com/apache/arrow-go/v18/arrow/extensions"
 	"github.com/apache/arrow-go/v18/arrow/memory"
-	"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/apache/iceberg-go/internal"
 	iceio "github.com/apache/iceberg-go/io"
 	tblutils "github.com/apache/iceberg-go/table/internal"
-	"github.com/google/uuid"
 	"github.com/pterm/pterm"
 )
 
@@ -1053,404 +1044,6 @@ func must[T any](v T, err error) T {
 	return v
 }
 
-func primitiveToPhysicalType(typ iceberg.Type) string {
-	switch t := typ.(type) {
-	case iceberg.BooleanType:
-		return "BOOLEAN"
-	case iceberg.Int32Type:
-		return "INT32"
-	case iceberg.Int64Type:
-		return "INT64"
-	case iceberg.Float32Type:
-		return "FLOAT"
-	case iceberg.Float64Type:
-		return "DOUBLE"
-	case iceberg.DateType:
-		return "INT32"
-	case iceberg.TimeType:
-		return "INT64"
-	case iceberg.TimestampType:
-		return "INT64"
-	case iceberg.TimestampTzType:
-		return "INT64"
-	case iceberg.StringType:
-		return "BYTE_ARRAY"
-	case iceberg.UUIDType:
-		return "FIXED_LEN_BYTE_ARRAY"
-	case iceberg.FixedType:
-		return "FIXED_LEN_BYTE_ARRAY"
-	case iceberg.BinaryType:
-		return "BYTE_ARRAY"
-	case iceberg.DecimalType:
-		switch {
-		case t.Precision() <= 9:
-			return "INT32"
-		case t.Precision() <= 18:
-			return "INT64"
-		default:
-			return "FIXED_LEN_BYTE_ARRAY"
-		}
-	default:
-		panic(fmt.Errorf("expected primitive type, got: %s", typ))
-	}
-}
-
-type typedStat[T iceberg.LiteralType] interface {
-	Min() T
-	Max() T
-}
-
-type wrappedBinaryStats struct {
-	*metadata.ByteArrayStatistics
-}
-
-func (w wrappedBinaryStats) Min() []byte {
-	return w.ByteArrayStatistics.Min()
-}
-
-func (w wrappedBinaryStats) Max() []byte {
-	return w.ByteArrayStatistics.Max()
-}
-
-type wrappedStringStats struct {
-	*metadata.ByteArrayStatistics
-}
-
-func (w wrappedStringStats) Min() string {
-	data := w.ByteArrayStatistics.Min()
-
-	return unsafe.String(unsafe.SliceData(data), len(data))
-}
-
-func (w wrappedStringStats) Max() string {
-	data := w.ByteArrayStatistics.Max()
-
-	return unsafe.String(unsafe.SliceData(data), len(data))
-}
-
-type wrappedUUIDStats struct {
-	*metadata.FixedLenByteArrayStatistics
-}
-
-func (w wrappedUUIDStats) Min() uuid.UUID {
-	uid, err := uuid.FromBytes(w.FixedLenByteArrayStatistics.Min())
-	if err != nil {
-		panic(err)
-	}
-
-	return uid
-}
-
-func (w wrappedUUIDStats) Max() uuid.UUID {
-	uid, err := uuid.FromBytes(w.FixedLenByteArrayStatistics.Max())
-	if err != nil {
-		panic(err)
-	}
-
-	return uid
-}
-
-type wrappedFLBAStats struct {
-	*metadata.FixedLenByteArrayStatistics
-}
-
-func (w wrappedFLBAStats) Min() []byte {
-	return w.FixedLenByteArrayStatistics.Min()
-}
-
-func (w wrappedFLBAStats) Max() []byte {
-	return w.FixedLenByteArrayStatistics.Max()
-}
-
-type wrappedDecStats struct {
-	*metadata.FixedLenByteArrayStatistics
-	scale int
-}
-
-func (w wrappedDecStats) Min() iceberg.Decimal {
-	dec, err := tblutils.BigEndianToDecimal(w.FixedLenByteArrayStatistics.Min())
-	if err != nil {
-		panic(err)
-	}
-
-	return iceberg.Decimal{Val: dec, Scale: w.scale}
-}
-
-func (w wrappedDecStats) Max() iceberg.Decimal {
-	dec, err := tblutils.BigEndianToDecimal(w.FixedLenByteArrayStatistics.Max())
-	if err != nil {
-		panic(err)
-	}
-
-	return iceberg.Decimal{Val: dec, Scale: w.scale}
-}
-
-type statsAgg interface {
-	min() iceberg.Literal
-	max() iceberg.Literal
-	update(stats metadata.TypedStatistics)
-	minAsBytes() ([]byte, error)
-	maxAsBytes() ([]byte, error)
-}
-
-type statsAggregator[T iceberg.LiteralType] struct {
-	curMin iceberg.TypedLiteral[T]
-	curMax iceberg.TypedLiteral[T]
-
-	cmp           iceberg.Comparator[T]
-	primitiveType iceberg.PrimitiveType
-	truncLen      int
-}
-
-func newStatAgg[T iceberg.LiteralType](typ iceberg.PrimitiveType, trunc int) statsAgg {
-	var z T
-
-	return &statsAggregator[T]{
-		primitiveType: typ,
-		truncLen:      trunc,
-		cmp:           iceberg.NewLiteral(z).(iceberg.TypedLiteral[T]).Comparator(),
-	}
-}
-
-func createStatsAgg(typ iceberg.PrimitiveType, physicalTypeStr string, truncLen int) (statsAgg, error) {
-	expectedPhysical := primitiveToPhysicalType(typ)
-	if physicalTypeStr != expectedPhysical {
-		switch {
-		case physicalTypeStr == "INT32" && expectedPhysical == "INT64":
-		case physicalTypeStr == "FLOAT" && expectedPhysical == "DOUBLE":
-		default:
-			return nil, fmt.Errorf("unexpected physical type %s for %s, expected %s",
-				physicalTypeStr, typ, expectedPhysical)
-		}
-	}
-
-	switch physicalTypeStr {
-	case "BOOLEAN":
-		return newStatAgg[bool](typ, truncLen), nil
-	case "INT32":
-		switch typ.(type) {
-		case iceberg.DecimalType:
-			return &decAsIntAgg[int32]{
-				newStatAgg[int32](typ, truncLen).(*statsAggregator[int32]),
-			}, nil
-		}
-
-		return newStatAgg[int32](typ, truncLen), nil
-	case "INT64":
-		switch typ.(type) {
-		case iceberg.DecimalType:
-			return &decAsIntAgg[int64]{
-				newStatAgg[int64](typ, truncLen).(*statsAggregator[int64]),
-			}, nil
-		}
-
-		return newStatAgg[int64](typ, truncLen), nil
-	case "FLOAT":
-		return newStatAgg[float32](typ, truncLen), nil
-	case "DOUBLE":
-		return newStatAgg[float64](typ, truncLen), nil
-	case "FIXED_LEN_BYTE_ARRAY":
-		switch typ.(type) {
-		case iceberg.UUIDType:
-			return newStatAgg[uuid.UUID](typ, truncLen), nil
-		case iceberg.DecimalType:
-			return newStatAgg[iceberg.Decimal](typ, truncLen), nil
-		default:
-			return newStatAgg[[]byte](typ, truncLen), nil
-		}
-	case "BYTE_ARRAY":
-		if typ.Equals(iceberg.PrimitiveTypes.String) {
-			return newStatAgg[string](typ, truncLen), nil
-		}
-
-		return newStatAgg[[]byte](typ, truncLen), nil
-	default:
-		return nil, fmt.Errorf("unsupported physical type: %s", physicalTypeStr)
-	}
-}
-
-func (s *statsAggregator[T]) min() iceberg.Literal { return s.curMin }
-func (s *statsAggregator[T]) max() iceberg.Literal { return s.curMax }
-
-func (s *statsAggregator[T]) update(stats metadata.TypedStatistics) {
-	st := stats.(typedStat[T])
-	s.updateMin(st.Min())
-	s.updateMax(st.Max())
-}
-
-func (s *statsAggregator[T]) toBytes(val iceberg.Literal) ([]byte, error) {
-	v, err := val.To(s.primitiveType)
-	if err != nil {
-		return nil, err
-	}
-
-	return v.MarshalBinary()
-}
-
-func (s *statsAggregator[T]) updateMin(val T) {
-	if s.curMin == nil {
-		s.curMin = iceberg.NewLiteral(val).(iceberg.TypedLiteral[T])
-	} else {
-		if s.cmp(val, s.curMin.Value()) < 0 {
-			s.curMin = iceberg.NewLiteral(val).(iceberg.TypedLiteral[T])
-		}
-	}
-}
-
-func (s *statsAggregator[T]) updateMax(val T) {
-	if s.curMax == nil {
-		s.curMax = iceberg.NewLiteral(val).(iceberg.TypedLiteral[T])
-	} else {
-		if s.cmp(val, s.curMax.Value()) > 0 {
-			s.curMax = iceberg.NewLiteral(val).(iceberg.TypedLiteral[T])
-		}
-	}
-}
-
-func (s *statsAggregator[T]) minAsBytes() ([]byte, error) {
-	if s.curMin == nil {
-		return nil, nil
-	}
-
-	if s.truncLen > 0 {
-		return s.toBytes((&iceberg.TruncateTransform{Width: s.truncLen}).
-			Apply(iceberg.Optional[iceberg.Literal]{Valid: true, Val: s.curMin}).Val)
-	}
-
-	return s.toBytes(s.curMin)
-}
-
-func (s *statsAggregator[T]) maxAsBytes() ([]byte, error) {
-	if s.curMax == nil {
-		return nil, nil
-	}
-
-	if s.truncLen <= 0 {
-		return s.toBytes(s.curMax)
-	}
-
-	switch s.primitiveType.(type) {
-	case iceberg.StringType:
-		if !s.curMax.Type().Equals(s.primitiveType) {
-			return nil, errors.New("expected current max to be a string")
-		}
-
-		curMax := any(s.curMax.Value()).(string)
-		result := truncateUpperBoundText(curMax, s.truncLen)
-		if result != "" {
-			return s.toBytes(iceberg.StringLiteral(result))
-		}
-
-		return nil, nil
-	case iceberg.BinaryType:
-		if !s.curMax.Type().Equals(s.primitiveType) {
-			return nil, errors.New("expected current max to be a binary")
-		}
-
-		curMax := any(s.curMax.Value()).([]byte)
-		result := truncateUpperBoundBinary(curMax, s.truncLen)
-		if len(result) > 0 {
-			return s.toBytes(iceberg.BinaryLiteral(result))
-		}
-
-		return nil, nil
-	default:
-		return nil, fmt.Errorf("%s cannot be truncated for upper bound", s.primitiveType)
-	}
-}
-
-type decAsIntAgg[T int32 | int64] struct {
-	*statsAggregator[T]
-}
-
-func (s *decAsIntAgg[T]) minAsBytes() ([]byte, error) {
-	if s.curMin == nil {
-		return nil, nil
-	}
-
-	lit := iceberg.DecimalLiteral(iceberg.Decimal{
-		Val:   decimal128.FromI64(int64(s.curMin.Value())),
-		Scale: s.primitiveType.(iceberg.DecimalType).Scale(),
-	})
-	if s.truncLen > 0 {
-		return s.toBytes((&iceberg.TruncateTransform{Width: s.truncLen}).
-			Apply(iceberg.Optional[iceberg.Literal]{Valid: true, Val: lit}).Val)
-	}
-
-	return s.toBytes(lit)
-}
-
-func (s *decAsIntAgg[T]) maxAsBytes() ([]byte, error) {
-	if s.curMax == nil {
-		return nil, nil
-	}
-
-	lit := iceberg.DecimalLiteral(iceberg.Decimal{
-		Val:   decimal128.FromI64(int64(s.curMax.Value())),
-		Scale: s.primitiveType.(iceberg.DecimalType).Scale(),
-	})
-	if s.truncLen <= 0 {
-		return s.toBytes(lit)
-	}
-
-	return nil, fmt.Errorf("%s cannot be truncated for upper bound", s.primitiveType)
-}
-
-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
@@ -1458,12 +1051,12 @@ type arrowStatsCollector struct {
 	defaultMode string
 }
 
-func (a *arrowStatsCollector) Schema(_ *iceberg.Schema, results func() []statisticsCollector) []statisticsCollector {
+func (a *arrowStatsCollector) Schema(_ *iceberg.Schema, results func() []tblutils.StatisticsCollector) []tblutils.StatisticsCollector {
 	return results()
 }
 
-func (a *arrowStatsCollector) Struct(_ iceberg.StructType, results []func() []statisticsCollector) []statisticsCollector {
-	result := make([]statisticsCollector, 0, len(results))
+func (a *arrowStatsCollector) Struct(_ iceberg.StructType, results []func() []tblutils.StatisticsCollector) []tblutils.StatisticsCollector {
+	result := make([]tblutils.StatisticsCollector, 0, len(results))
 	for _, res := range results {
 		result = append(result, res()...)
 	}
@@ -1471,19 +1064,19 @@ func (a *arrowStatsCollector) Struct(_ iceberg.StructType, results []func() []st
 	return result
 }
 
-func (a *arrowStatsCollector) Field(field iceberg.NestedField, fieldRes func() []statisticsCollector) []statisticsCollector {
+func (a *arrowStatsCollector) Field(field iceberg.NestedField, fieldRes func() []tblutils.StatisticsCollector) []tblutils.StatisticsCollector {
 	a.fieldID = field.ID
 
 	return fieldRes()
 }
 
-func (a *arrowStatsCollector) List(list iceberg.ListType, elemResult func() []statisticsCollector) []statisticsCollector {
+func (a *arrowStatsCollector) List(list iceberg.ListType, elemResult func() []tblutils.StatisticsCollector) []tblutils.StatisticsCollector {
 	a.fieldID = list.ElementID
 
 	return elemResult()
 }
 
-func (a *arrowStatsCollector) Map(m iceberg.MapType, keyResult func() []statisticsCollector, valResult func() []statisticsCollector) []statisticsCollector {
+func (a *arrowStatsCollector) Map(m iceberg.MapType, keyResult, valResult func() []tblutils.StatisticsCollector) []tblutils.StatisticsCollector {
 	a.fieldID = m.KeyID
 	keyRes := keyResult()
 
@@ -1493,20 +1086,20 @@ func (a *arrowStatsCollector) Map(m iceberg.MapType, keyResult func() []statisti
 	return append(keyRes, valRes...)
 }
 
-func (a *arrowStatsCollector) Primitive(dt iceberg.PrimitiveType) []statisticsCollector {
+func (a *arrowStatsCollector) Primitive(dt iceberg.PrimitiveType) []tblutils.StatisticsCollector {
 	colName, ok := a.schema.FindColumnName(a.fieldID)
 	if !ok {
-		return []statisticsCollector{}
+		return []tblutils.StatisticsCollector{}
 	}
 
-	metMode, err := matchMetricsMode(a.defaultMode)
+	metMode, err := tblutils.MatchMetricsMode(a.defaultMode)
 	if err != nil {
 		panic(err)
 	}
 
 	colMode, ok := a.props[MetricsModeColumnConfPrefix+"."+colName]
 	if ok {
-		metMode, err = matchMetricsMode(colMode)
+		metMode, err = tblutils.MatchMetricsMode(colMode)
 		if err != nil {
 			panic(err)
 		}
@@ -1516,26 +1109,26 @@ func (a *arrowStatsCollector) Primitive(dt iceberg.PrimitiveType) []statisticsCo
 	case iceberg.StringType:
 	case iceberg.BinaryType:
 	default:
-		if metMode.typ == metricModeTruncate {
-			metMode = metricsMode{typ: metricModeFull, len: 0}
+		if metMode.Typ == tblutils.MetricModeTruncate {
+			metMode = tblutils.MetricsMode{Typ: tblutils.MetricModeFull, Len: 0}
 		}
 	}
 
 	isNested := strings.Contains(colName, ".")
-	if isNested && (metMode.typ == metricModeTruncate || metMode.typ == metricModeFull) {
-		metMode = metricsMode{typ: metricModeCounts}
+	if isNested && (metMode.Typ == tblutils.MetricModeTruncate || metMode.Typ == tblutils.MetricModeFull) {
+		metMode = tblutils.MetricsMode{Typ: tblutils.MetricModeCounts}
 	}
 
-	return []statisticsCollector{{
-		fieldID:    a.fieldID,
-		icebergTyp: dt,
-		colName:    colName,
-		mode:       metMode,
+	return []tblutils.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)
+func computeStatsPlan(sc *iceberg.Schema, props iceberg.Properties) (map[int]tblutils.StatisticsCollector, error) {
+	result := make(map[int]tblutils.StatisticsCollector)
 
 	visitor := &arrowStatsCollector{
 		schema: sc, props: props,
@@ -1548,289 +1141,13 @@ func computeStatsPlan(sc *iceberg.Schema, props iceberg.Properties) (map[int]sta
 	}
 
 	for _, entry := range collectors {
-		result[entry.fieldID] = entry
+		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
-	colAggs         map[int]statsAgg
-	splitOffsets    []int64
-}
-
-func (d *dataFileStatistics) partitionValue(field iceberg.PartitionField, sc *iceberg.Schema) any {
-	agg, ok := d.colAggs[field.SourceID]
-	if !ok {
-		return nil
-	}
-
-	if !field.Transform.PreservesOrder() {
-		panic(fmt.Errorf("cannot infer partition value from parquet metadata for a non-linear partition field: %s with transform %s",
-			field.Name, field.Transform))
-	}
-
-	lowerVal, upperVal := agg.min(), agg.max()
-	if !lowerVal.Equals(upperVal) {
-		panic(fmt.Errorf("cannot infer partition value from parquet metadata as there is more than one value for partition field: %s. (low: %s, high: %s)",
-			field.Name, lowerVal, upperVal))
-	}
-
-	val := field.Transform.Apply(must(partitionRecordValue(field, lowerVal, sc)))
-	if !val.Valid {
-		return nil
-	}
-
-	return val.Val.Any()
-}
-
-func (d *dataFileStatistics) toDataFile(schema *iceberg.Schema, spec iceberg.PartitionSpec, path string, format iceberg.FileFormat, filesize int64) iceberg.DataFile {
-	var partitionData map[string]any
-	if !spec.Equals(*iceberg.UnpartitionedSpec) {
-		partitionData = make(map[string]any)
-		for field := range spec.Fields() {
-			val := d.partitionValue(field, schema)
-			if val != nil {
-				partitionData[field.Name] = val
-			}
-		}
-	}
-
-	bldr, err := iceberg.NewDataFileBuilder(spec, iceberg.EntryContentData,
-		path, format, partitionData, d.recordCount, filesize)
-	if err != nil {
-		panic(err)
-	}
-
-	lowerBounds := make(map[int][]byte)
-	upperBounds := make(map[int][]byte)
-
-	for fieldID, agg := range d.colAggs {
-		min := must(agg.minAsBytes())
-		max := must(agg.maxAsBytes())
-		if len(min) > 0 {
-			lowerBounds[fieldID] = min
-		}
-		if len(max) > 0 {
-			upperBounds[fieldID] = max
-		}
-	}
-
-	if len(lowerBounds) > 0 {
-		bldr.LowerBoundValues(lowerBounds)
-	}
-	if len(upperBounds) > 0 {
-		bldr.UpperBoundValues(upperBounds)
-	}
-
-	bldr.ColumnSizes(d.colSizes)
-	bldr.ValueCounts(d.valueCounts)
-	bldr.NullValueCounts(d.nullValueCounts)
-	bldr.NaNValueCounts(d.nanValueCounts)
-	bldr.SplitOffsets(d.splitOffsets)
-
-	return bldr.Build()
-}
-
-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{})
-		colAggs         = make(map[int]statsAgg)
-	)
-
-	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 || !stats.HasMinMax() {
-				continue
-			}
-
-			agg, ok := colAggs[fieldID]
-			if !ok {
-				agg, err = createStatsAgg(statsCol.icebergTyp, stats.Type().String(), statsCol.mode.len)
-				if err != nil {
-					panic(err)
-				}
-
-				colAggs[fieldID] = agg
-			}
-
-			switch typ := statsCol.icebergTyp.(type) {
-			case iceberg.BinaryType:
-				stats = wrappedBinaryStats{stats.(*metadata.ByteArrayStatistics)}
-			case iceberg.UUIDType:
-				stats = wrappedUUIDStats{stats.(*metadata.FixedLenByteArrayStatistics)}
-			case iceberg.StringType:
-				stats = wrappedStringStats{stats.(*metadata.ByteArrayStatistics)}
-			case iceberg.FixedType:
-				stats = wrappedFLBAStats{stats.(*metadata.FixedLenByteArrayStatistics)}
-			case iceberg.DecimalType:
-				if typ.Precision() > 18 {
-					stats = wrappedDecStats{stats.(*metadata.FixedLenByteArrayStatistics), typ.Scale()}
-				}
-			}
-
-			agg.update(stats)
-		}
-
-	}
-
-	slices.Sort(splitOffsets)
-	maps.DeleteFunc(nullValueCounts, func(fieldID int, _ int64) bool {
-		_, ok := invalidateCol[fieldID]
-
-		return ok
-	})
-	maps.DeleteFunc(colAggs, func(fieldID int, _ statsAgg) bool {
-		_, ok := invalidateCol[fieldID]
-
-		return ok
-	})
-
-	return &dataFileStatistics{
-		recordCount:     pqmeta.GetNumRows(),
-		colSizes:        colSizes,
-		valueCounts:     valueCounts,
-		nullValueCounts: nullValueCounts,
-		nanValueCounts:  nanValueCounts,
-		splitOffsets:    splitOffsets,
-		colAggs:         colAggs,
-	}
-}
-
-func parquetFilesToDataFiles(fileIO iceio.IO, meta *MetadataBuilder, paths iter.Seq[string]) iter.Seq2[iceberg.DataFile, error] {
+func filesToDataFiles(ctx context.Context, fileIO iceio.IO, meta *MetadataBuilder, paths iter.Seq[string]) iter.Seq2[iceberg.DataFile, error] {
 	return func(yield func(iceberg.DataFile, error) bool) {
 		defer func() {
 			if r := recover(); r != nil {
@@ -1846,14 +1163,11 @@ func parquetFilesToDataFiles(fileIO iceio.IO, meta *MetadataBuilder, paths iter.
 		currentSchema, currentSpec := meta.CurrentSchema(), meta.CurrentSpec()
 
 		for filePath := range paths {
-			inputFile := must(fileIO.Open(filePath))
-			defer inputFile.Close()
-
-			rdr := must(file.NewParquetReader(inputFile))
+			format := tblutils.FormatFromFileName(filePath)
+			rdr := must(format.Open(ctx, fileIO, filePath))
 			defer rdr.Close()
 
-			arrRdr := must(pqarrow.NewFileReader(rdr, pqarrow.ArrowReadProperties{}, memory.DefaultAllocator))
-			arrSchema := must(arrRdr.Schema())
+			arrSchema := must(rdr.Schema())
 
 			if hasIDs := must(VisitArrowSchema(arrSchema, hasIDs{})); hasIDs {
 				yield(nil, fmt.Errorf("%w: cannot add file %s because it has field-ids. add-files only supports the addition of files without field_ids",
@@ -1863,14 +1177,15 @@ func parquetFilesToDataFiles(fileIO iceio.IO, meta *MetadataBuilder, paths iter.
 			}
 
 			if err := checkArrowSchemaCompat(currentSchema, arrSchema, false); err != nil {
-				panic(err)
+				yield(nil, err)
+
+				return
 			}
 
-			statistics := dataFileStatsFromParquetMetadata(rdr.MetaData(),
-				must(computeStatsPlan(currentSchema, meta.props)),
-				must(parquetPathToIDMapping(currentSchema)))
+			statistics := format.DataFileStatsFromMeta(rdr.Metadata(), must(computeStatsPlan(currentSchema, meta.props)),
+				must(format.PathToIDMapping(currentSchema)))
 
-			df := statistics.toDataFile(currentSchema, currentSpec, filePath, iceberg.ParquetFile, rdr.MetaData().GetSourceFileSize())
+			df := statistics.ToDataFile(currentSchema, currentSpec, filePath, iceberg.ParquetFile, rdr.SourceFileSize())
 			if !yield(df, nil) {
 				return
 			}
diff --git a/table/internal/interfaces.go b/table/internal/interfaces.go
index d5697bf..720e80c 100644
--- a/table/internal/interfaces.go
+++ b/table/internal/interfaces.go
@@ -21,6 +21,7 @@ import (
 	"context"
 	"fmt"
 	"io"
+	"path"
 
 	"github.com/apache/arrow-go/v18/arrow"
 	"github.com/apache/arrow-go/v18/arrow/array"
@@ -51,9 +52,14 @@ type FileSource interface {
 	GetReader(context.Context) (FileReader, error)
 }
 
+type Metadata any
+
 type FileReader interface {
 	io.Closer
 
+	Metadata() Metadata
+	SourceFileSize() int64
+	Schema() (*arrow.Schema, error)
 	// PrunedSchema takes in the list of projected field IDs and returns the arrow schema
 	// 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
@@ -66,3 +72,27 @@ type FileReader interface {
 	// ReadTable reads the entire file and returns it as an arrow table.
 	ReadTable(context.Context) (arrow.Table, error)
 }
+
+type FileFormat interface {
+	Open(context.Context, iceio.IO, string) (FileReader, error)
+	PathToIDMapping(*iceberg.Schema) (map[string]int, error)
+	DataFileStatsFromMeta(rdr Metadata, statsCols map[int]StatisticsCollector, colMapping map[string]int) *DataFileStatistics
+}
+
+func GetFileFormat(format iceberg.FileFormat) FileFormat {
+	switch format {
+	case iceberg.ParquetFile:
+		return parquetFormat{}
+	default:
+		return nil
+	}
+}
+
+func FormatFromFileName(fileName string) FileFormat {
+	switch path.Ext(fileName) {
+	case ".parquet":
+		return parquetFormat{}
+	default:
+		return nil
+	}
+}
diff --git a/table/internal/parquet_files.go b/table/internal/parquet_files.go
index 770a9be..2223d5b 100644
--- a/table/internal/parquet_files.go
+++ b/table/internal/parquet_files.go
@@ -21,11 +21,16 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"maps"
 	"slices"
 	"strconv"
+	"strings"
+	"unsafe"
 
 	"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/decimal128"
 	"github.com/apache/arrow-go/v18/arrow/memory"
 	"github.com/apache/arrow-go/v18/parquet"
 	"github.com/apache/arrow-go/v18/parquet/file"
@@ -33,8 +38,378 @@ import (
 	"github.com/apache/arrow-go/v18/parquet/pqarrow"
 	"github.com/apache/iceberg-go"
 	iceio "github.com/apache/iceberg-go/io"
+	"github.com/google/uuid"
 )
 
+type parquetFormat struct{}
+
+func (parquetFormat) Open(ctx context.Context, fs iceio.IO, path string) (FileReader, error) {
+	inputfile, err := fs.Open(path)
+	if err != nil {
+		return nil, err
+	}
+
+	rdr, err := file.NewParquetReader(inputfile)
+	if err != nil {
+		return nil, err
+	}
+
+	alloc := compute.GetAllocator(ctx)
+	arrRdr, err := pqarrow.NewFileReader(rdr, pqarrow.ArrowReadProperties{}, alloc)
+	if err != nil {
+		return nil, err
+	}
+
+	return wrapPqArrowReader{arrRdr}, nil
+}
+
+func (parquetFormat) PathToIDMapping(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
+}
+
+func (p parquetFormat) createStatsAgg(typ iceberg.PrimitiveType, physicalTypeStr string, truncLen int) (StatsAgg, error) {
+	expectedPhysical := p.PrimitiveTypeToPhysicalType(typ)
+	if physicalTypeStr != expectedPhysical {
+		switch {
+		case physicalTypeStr == "INT32" && expectedPhysical == "INT64":
+		case physicalTypeStr == "FLOAT" && expectedPhysical == "DOUBLE":
+		default:
+			return nil, fmt.Errorf("unexpected physical type %s for %s, expected %s",
+				physicalTypeStr, typ, expectedPhysical)
+		}
+	}
+
+	switch physicalTypeStr {
+	case "BOOLEAN":
+		return newStatAgg[bool](typ, truncLen), nil
+	case "INT32":
+		switch typ.(type) {
+		case iceberg.DecimalType:
+			return &decAsIntAgg[int32]{
+				newStatAgg[int32](typ, truncLen).(*statsAggregator[int32]),
+			}, nil
+		}
+
+		return newStatAgg[int32](typ, truncLen), nil
+	case "INT64":
+		switch typ.(type) {
+		case iceberg.DecimalType:
+			return &decAsIntAgg[int64]{
+				newStatAgg[int64](typ, truncLen).(*statsAggregator[int64]),
+			}, nil
+		}
+
+		return newStatAgg[int64](typ, truncLen), nil
+	case "FLOAT":
+		return newStatAgg[float32](typ, truncLen), nil
+	case "DOUBLE":
+		return newStatAgg[float64](typ, truncLen), nil
+	case "FIXED_LEN_BYTE_ARRAY":
+		switch typ.(type) {
+		case iceberg.UUIDType:
+			return newStatAgg[uuid.UUID](typ, truncLen), nil
+		case iceberg.DecimalType:
+			return newStatAgg[iceberg.Decimal](typ, truncLen), nil
+		default:
+			return newStatAgg[[]byte](typ, truncLen), nil
+		}
+	case "BYTE_ARRAY":
+		if typ.Equals(iceberg.PrimitiveTypes.String) {
+			return newStatAgg[string](typ, truncLen), nil
+		}
+
+		return newStatAgg[[]byte](typ, truncLen), nil
+	default:
+		return nil, fmt.Errorf("unsupported physical type: %s", physicalTypeStr)
+	}
+}
+
+func (parquetFormat) PrimitiveTypeToPhysicalType(typ iceberg.PrimitiveType) string {
+	switch typ.(type) {
+	case iceberg.BooleanType:
+		return "BOOLEAN"
+	case iceberg.Int32Type:
+		return "INT32"
+	case iceberg.Int64Type:
+		return "INT64"
+	case iceberg.Float32Type:
+		return "FLOAT"
+	case iceberg.Float64Type:
+		return "DOUBLE"
+	case iceberg.DateType:
+		return "INT32"
+	case iceberg.TimeType:
+		return "INT64"
+	case iceberg.TimestampType:
+		return "INT64"
+	case iceberg.TimestampTzType:
+		return "INT64"
+	case iceberg.StringType:
+		return "BYTE_ARRAY"
+	case iceberg.UUIDType:
+		return "FIXED_LEN_BYTE_ARRAY"
+	case iceberg.FixedType:
+		return "FIXED_LEN_BYTE_ARRAY"
+	case iceberg.BinaryType:
+		return "BYTE_ARRAY"
+	case iceberg.DecimalType:
+		return "FIXED_LEN_BYTE_ARRAY"
+	default:
+		panic(fmt.Errorf("expected primitive type, got: %s", typ))
+	}
+}
+
+type decAsIntAgg[T int32 | int64] struct {
+	*statsAggregator[T]
+}
+
+func (s *decAsIntAgg[T]) MinAsBytes() ([]byte, error) {
+	if s.curMin == nil {
+		return nil, nil
+	}
+
+	lit := iceberg.DecimalLiteral(iceberg.Decimal{
+		Val:   decimal128.FromI64(int64(s.curMin.Value())),
+		Scale: s.primitiveType.(iceberg.DecimalType).Scale(),
+	})
+	if s.truncLen > 0 {
+		return s.toBytes((&iceberg.TruncateTransform{Width: s.truncLen}).
+			Apply(iceberg.Optional[iceberg.Literal]{Valid: true, Val: lit}).Val)
+	}
+
+	return s.toBytes(lit)
+}
+
+func (s *decAsIntAgg[T]) MaxAsBytes() ([]byte, error) {
+	if s.curMax == nil {
+		return nil, nil
+	}
+
+	lit := iceberg.DecimalLiteral(iceberg.Decimal{
+		Val:   decimal128.FromI64(int64(s.curMax.Value())),
+		Scale: s.primitiveType.(iceberg.DecimalType).Scale(),
+	})
+	if s.truncLen <= 0 {
+		return s.toBytes(lit)
+	}
+
+	return nil, fmt.Errorf("%s cannot be truncated for upper bound", s.primitiveType)
+}
+
+type wrappedBinaryStats struct {
+	*metadata.ByteArrayStatistics
+}
+
+func (w *wrappedBinaryStats) Min() []byte {
+	return w.ByteArrayStatistics.Min()
+}
+
+func (w *wrappedBinaryStats) Max() []byte {
+	return w.ByteArrayStatistics.Max()
+}
+
+type wrappedStringStats struct {
+	*metadata.ByteArrayStatistics
+}
+
+func (w *wrappedStringStats) Min() string {
+	data := w.ByteArrayStatistics.Min()
+
+	return unsafe.String(unsafe.SliceData(data), len(data))
+}
+
+func (w *wrappedStringStats) Max() string {
+	data := w.ByteArrayStatistics.Max()
+
+	return unsafe.String(unsafe.SliceData(data), len(data))
+}
+
+type wrappedUUIDStats struct {
+	*metadata.FixedLenByteArrayStatistics
+}
+
+func (w *wrappedUUIDStats) Min() uuid.UUID {
+	uid, err := uuid.FromBytes(w.FixedLenByteArrayStatistics.Min())
+	if err != nil {
+		panic(err)
+	}
+
+	return uid
+}
+
+func (w *wrappedUUIDStats) Max() uuid.UUID {
+	uid, err := uuid.FromBytes(w.FixedLenByteArrayStatistics.Max())
+	if err != nil {
+		panic(err)
+	}
+
+	return uid
+}
+
+type wrappedFLBAStats struct {
+	*metadata.FixedLenByteArrayStatistics
+}
+
+func (w *wrappedFLBAStats) Min() []byte {
+	return w.FixedLenByteArrayStatistics.Min()
+}
+
+func (w *wrappedFLBAStats) Max() []byte {
+	return w.FixedLenByteArrayStatistics.Max()
+}
+
+type wrappedDecStats struct {
+	*metadata.FixedLenByteArrayStatistics
+	scale int
+}
+
+func (w wrappedDecStats) Min() iceberg.Decimal {
+	dec, err := BigEndianToDecimal(w.FixedLenByteArrayStatistics.Min())
+	if err != nil {
+		panic(err)
+	}
+
+	return iceberg.Decimal{Val: dec, Scale: w.scale}
+}
+
+func (w wrappedDecStats) Max() iceberg.Decimal {
+	dec, err := BigEndianToDecimal(w.FixedLenByteArrayStatistics.Max())
+	if err != nil {
+		panic(err)
+	}
+
+	return iceberg.Decimal{Val: dec, Scale: w.scale}
+}
+
+func (p parquetFormat) DataFileStatsFromMeta(meta Metadata, statsCols map[int]StatisticsCollector, colMapping map[string]int) *DataFileStatistics {
+	pqmeta := meta.(*metadata.FileMetaData)
+	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{})
+		colAggs         = make(map[int]StatsAgg)
+	)
+
+	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 || !stats.HasMinMax() {
+				continue
+			}
+
+			agg, ok := colAggs[fieldID]
+			if !ok {
+				agg, err = p.createStatsAgg(statsCol.IcebergTyp, stats.Type().String(), statsCol.Mode.Len)
+				if err != nil {
+					panic(err)
+				}
+
+				colAggs[fieldID] = agg
+			}
+
+			switch t := statsCol.IcebergTyp.(type) {
+			case iceberg.BinaryType:
+				stats = &wrappedBinaryStats{stats.(*metadata.ByteArrayStatistics)}
+			case iceberg.UUIDType:
+				stats = &wrappedUUIDStats{stats.(*metadata.FixedLenByteArrayStatistics)}
+			case iceberg.StringType:
+				stats = &wrappedStringStats{stats.(*metadata.ByteArrayStatistics)}
+			case iceberg.FixedType:
+				stats = &wrappedFLBAStats{stats.(*metadata.FixedLenByteArrayStatistics)}
+			case iceberg.DecimalType:
+				stats = &wrappedDecStats{stats.(*metadata.FixedLenByteArrayStatistics), t.Scale()}
+			}
+
+			agg.Update(stats)
+		}
+
+	}
+
+	slices.Sort(splitOffsets)
+	maps.DeleteFunc(nullValueCounts, func(fieldID int, _ int64) bool {
+		_, ok := invalidateCol[fieldID]
+
+		return ok
+	})
+	maps.DeleteFunc(colAggs, func(fieldID int, _ StatsAgg) bool {
+		_, ok := invalidateCol[fieldID]
+
+		return ok
+	})
+
+	return &DataFileStatistics{
+		RecordCount:     pqmeta.GetNumRows(),
+		ColSizes:        colSizes,
+		ValueCounts:     valueCounts,
+		NullValueCounts: nullValueCounts,
+		NanValueCounts:  nanValueCounts,
+		SplitOffsets:    splitOffsets,
+		ColAggs:         colAggs,
+	}
+}
+
 type ParquetFileSource struct {
 	mem  memory.Allocator
 	fs   iceio.IO
@@ -45,6 +420,14 @@ type wrapPqArrowReader struct {
 	*pqarrow.FileReader
 }
 
+func (w wrapPqArrowReader) Metadata() Metadata {
+	return w.ParquetReader().MetaData()
+}
+
+func (w wrapPqArrowReader) SourceFileSize() int64 {
+	return w.ParquetReader().MetaData().GetSourceFileSize()
+}
+
 func (w wrapPqArrowReader) Close() error {
 	return w.ParquetReader().Close()
 }
@@ -464,3 +847,62 @@ func (p *pruneParquetSchema) projectMap(m *arrow.MapType, valResult arrow.DataTy
 
 	return arrow.MapOf(m.KeyType(), valResult)
 }
+
+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, ".")}}
+}
diff --git a/table/internal/utils.go b/table/internal/utils.go
index cab0d73..5a53b6a 100644
--- a/table/internal/utils.go
+++ b/table/internal/utils.go
@@ -18,12 +18,20 @@
 package internal
 
 import (
+	"bytes"
 	"container/heap"
 	"encoding/binary"
+	"errors"
 	"fmt"
+	"regexp"
+	"strconv"
+	"strings"
+	"unicode/utf8"
+	_ "unsafe"
 
 	"github.com/apache/arrow-go/v18/arrow/decimal"
 	"github.com/apache/arrow-go/v18/arrow/decimal128"
+	"github.com/apache/iceberg-go"
 )
 
 // Enumerated is a quick way to represent a sequenced value that can
@@ -139,3 +147,334 @@ func BigEndianToDecimal(buf []byte) (decimal.Decimal128, error) {
 
 	return decimal128.New(hi, uint64(lo)), nil
 }
+
+func PartitionRecordValue(field iceberg.PartitionField, val iceberg.Literal, schema *iceberg.Schema) (iceberg.Optional[iceberg.Literal], error) {
+	var ret iceberg.Optional[iceberg.Literal]
+	if val == nil {
+		return ret, nil
+	}
+
+	f, ok := schema.FindFieldByID(field.SourceID)
+	if !ok {
+		return ret, fmt.Errorf("%w: could not find source field in schema for %s",
+			iceberg.ErrInvalidSchema, field.String())
+	}
+
+	out, err := val.To(f.Type)
+	if err != nil {
+		return ret, err
+	}
+
+	ret.Val = out
+	ret.Valid = true
+
+	return ret, nil
+}
+
+func must[T any](v T, err error) T {
+	if err != nil {
+		panic(err)
+	}
+
+	return v
+}
+
+type TypedStats[T iceberg.LiteralType] interface {
+	Min() T
+	Max() T
+}
+
+type StatsAgg interface {
+	Min() iceberg.Literal
+	Max() iceberg.Literal
+	Update(stats interface{ HasMinMax() bool })
+	MinAsBytes() ([]byte, error)
+	MaxAsBytes() ([]byte, error)
+}
+
+type DataFileStatistics struct {
+	RecordCount     int64
+	ColSizes        map[int]int64
+	ValueCounts     map[int]int64
+	NullValueCounts map[int]int64
+	NanValueCounts  map[int]int64
+	ColAggs         map[int]StatsAgg
+	SplitOffsets    []int64
+}
+
+func (d *DataFileStatistics) PartitionValue(field iceberg.PartitionField, sc *iceberg.Schema) any {
+	agg, ok := d.ColAggs[field.SourceID]
+	if !ok {
+		return nil
+	}
+
+	if !field.Transform.PreservesOrder() {
+		panic(fmt.Errorf("cannot infer partition value from parquet metadata for a non-linear partition field: %s with transform %s",
+			field.Name, field.Transform))
+	}
+
+	lowerVal, upperVal := agg.Min(), agg.Max()
+	if !lowerVal.Equals(upperVal) {
+		panic(fmt.Errorf("cannot infer partition value from parquet metadata as there is more than one value for partition field: %s. (low: %s, high: %s)",
+			field.Name, lowerVal, upperVal))
+	}
+
+	val := field.Transform.Apply(must(PartitionRecordValue(field, lowerVal, sc)))
+	if !val.Valid {
+		return nil
+	}
+
+	return val.Val.Any()
+}
+
+func (d *DataFileStatistics) ToDataFile(schema *iceberg.Schema, spec iceberg.PartitionSpec, path string, format iceberg.FileFormat, filesize int64) iceberg.DataFile {
+	var partitionData map[string]any
+	if !spec.Equals(*iceberg.UnpartitionedSpec) {
+		partitionData = make(map[string]any)
+		for field := range spec.Fields() {
+			val := d.PartitionValue(field, schema)
+			if val != nil {
+				partitionData[field.Name] = val
+			}
+		}
+	}
+
+	bldr, err := iceberg.NewDataFileBuilder(spec, iceberg.EntryContentData,
+		path, format, partitionData, d.RecordCount, filesize)
+	if err != nil {
+		panic(err)
+	}
+
+	lowerBounds := make(map[int][]byte)
+	upperBounds := make(map[int][]byte)
+
+	for fieldID, agg := range d.ColAggs {
+		min := must(agg.MinAsBytes())
+		max := must(agg.MaxAsBytes())
+		if len(min) > 0 {
+			lowerBounds[fieldID] = min
+		}
+		if len(max) > 0 {
+			upperBounds[fieldID] = max
+		}
+	}
+
+	if len(lowerBounds) > 0 {
+		bldr.LowerBoundValues(lowerBounds)
+	}
+	if len(upperBounds) > 0 {
+		bldr.UpperBoundValues(upperBounds)
+	}
+
+	bldr.ColumnSizes(d.ColSizes)
+	bldr.ValueCounts(d.ValueCounts)
+	bldr.NullValueCounts(d.NullValueCounts)
+	bldr.NaNValueCounts(d.NanValueCounts)
+	bldr.SplitOffsets(d.SplitOffsets)
+
+	return bldr.Build()
+}
+
+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 typedStat[T iceberg.LiteralType] interface {
+	Min() T
+	Max() T
+}
+
+type statsAggregator[T iceberg.LiteralType] struct {
+	curMin iceberg.TypedLiteral[T]
+	curMax iceberg.TypedLiteral[T]
+
+	cmp           iceberg.Comparator[T]
+	primitiveType iceberg.PrimitiveType
+	truncLen      int
+}
+
+func newStatAgg[T iceberg.LiteralType](typ iceberg.PrimitiveType, trunc int) StatsAgg {
+	var z T
+
+	return &statsAggregator[T]{
+		primitiveType: typ,
+		truncLen:      trunc,
+		cmp:           iceberg.NewLiteral(z).(iceberg.TypedLiteral[T]).Comparator(),
+	}
+}
+
+func (s *statsAggregator[T]) Min() iceberg.Literal { return s.curMin }
+func (s *statsAggregator[T]) Max() iceberg.Literal { return s.curMax }
+
+func (s *statsAggregator[T]) Update(stats interface{ HasMinMax() bool }) {
+	st := stats.(typedStat[T])
+	s.updateMin(st.Min())
+	s.updateMax(st.Max())
+}
+
+func (s *statsAggregator[T]) toBytes(val iceberg.Literal) ([]byte, error) {
+	v, err := val.To(s.primitiveType)
+	if err != nil {
+		return nil, err
+	}
+
+	return v.MarshalBinary()
+}
+
+func (s *statsAggregator[T]) updateMin(val T) {
+	if s.curMin == nil {
+		s.curMin = iceberg.NewLiteral(val).(iceberg.TypedLiteral[T])
+	} else {
+		if s.cmp(val, s.curMin.Value()) < 0 {
+			s.curMin = iceberg.NewLiteral(val).(iceberg.TypedLiteral[T])
+		}
+	}
+}
+
+func (s *statsAggregator[T]) updateMax(val T) {
+	if s.curMax == nil {
+		s.curMax = iceberg.NewLiteral(val).(iceberg.TypedLiteral[T])
+	} else {
+		if s.cmp(val, s.curMax.Value()) > 0 {
+			s.curMax = iceberg.NewLiteral(val).(iceberg.TypedLiteral[T])
+		}
+	}
+}
+
+func (s *statsAggregator[T]) MinAsBytes() ([]byte, error) {
+	if s.curMin == nil {
+		return nil, nil
+	}
+
+	if s.truncLen > 0 {
+		return s.toBytes((&iceberg.TruncateTransform{Width: s.truncLen}).
+			Apply(iceberg.Optional[iceberg.Literal]{Valid: true, Val: s.curMin}).Val)
+	}
+
+	return s.toBytes(s.curMin)
+}
+
+func (s *statsAggregator[T]) MaxAsBytes() ([]byte, error) {
+	if s.curMax == nil {
+		return nil, nil
+	}
+
+	if s.truncLen <= 0 {
+		return s.toBytes(s.curMax)
+	}
+
+	switch s.primitiveType.(type) {
+	case iceberg.StringType:
+		if !s.curMax.Type().Equals(s.primitiveType) {
+			return nil, errors.New("expected current max to be a string")
+		}
+
+		curMax := any(s.curMax.Value()).(string)
+		result := TruncateUpperBoundText(curMax, s.truncLen)
+		if result != "" {
+			return s.toBytes(iceberg.StringLiteral(result))
+		}
+
+		return nil, nil
+	case iceberg.BinaryType:
+		if !s.curMax.Type().Equals(s.primitiveType) {
+			return nil, errors.New("expected current max to be a binary")
+		}
+
+		curMax := any(s.curMax.Value()).([]byte)
+		result := TruncateUpperBoundBinary(curMax, s.truncLen)
+		if len(result) > 0 {
+			return s.toBytes(iceberg.BinaryLiteral(result))
+		}
+
+		return nil, nil
+	default:
+		return nil, fmt.Errorf("%s cannot be truncated for upper bound", s.primitiveType)
+	}
+}
+
+func TruncateUpperBoundText(s string, trunc int) string {
+	if trunc == utf8.RuneCountInString(s) {
+		return s
+	}
+
+	result := []rune(s)[:trunc]
+	for i := len(result) - 1; i >= 0; i-- {
+		next := result[i] + 1
+		if utf8.ValidRune(next) {
+			result[i] = next
+
+			return string(result)
+		}
+	}
+
+	return ""
+}
+
+func TruncateUpperBoundBinary(val []byte, trunc int) []byte {
+	result := val[:trunc]
+	if bytes.Equal(result, val) {
+		return result
+	}
+
+	for i := len(result) - 1; i >= 0; i-- {
+		if result[i] < 255 {
+			result[i]++
+
+			return result
+		}
+	}
+
+	return nil
+}
diff --git a/table/scanner.go b/table/scanner.go
index a058d90..c0eaf93 100644
--- a/table/scanner.go
+++ b/table/scanner.go
@@ -488,26 +488,3 @@ func (scan *Scan) ToArrowTable(ctx context.Context) (arrow.Table, error) {
 
 	return array.NewTableFromRecords(schema, records), nil
 }
-
-func partitionRecordValue(field iceberg.PartitionField, val iceberg.Literal, schema *iceberg.Schema) (iceberg.Optional[iceberg.Literal], error) {
-	var ret iceberg.Optional[iceberg.Literal]
-	if val == nil {
-		return ret, nil
-	}
-
-	f, ok := schema.FindFieldByID(field.SourceID)
-	if !ok {
-		return ret, fmt.Errorf("%w: could not find source field in schema for %s",
-			iceberg.ErrInvalidSchema, field.String())
-	}
-
-	out, err := val.To(f.Type)
-	if err != nil {
-		return ret, err
-	}
-
-	ret.Val = out
-	ret.Valid = true
-
-	return ret, nil
-}
diff --git a/table/snapshot_producers.go b/table/snapshot_producers.go
index 01d39f7..0309063 100644
--- a/table/snapshot_producers.go
+++ b/table/snapshot_producers.go
@@ -18,14 +18,12 @@
 package table
 
 import (
-	"bytes"
 	"fmt"
 	"io"
 	"maps"
 	"slices"
 	"sync/atomic"
 	"time"
-	"unicode/utf8"
 
 	"github.com/apache/iceberg-go"
 	"github.com/apache/iceberg-go/internal"
@@ -349,38 +347,3 @@ func (sp *snapshotProducer) commit() ([]Update, []Requirement, error) {
 			AssertRefSnapshotID("main", sp.txn.meta.currentSnapshotID),
 		}, nil
 }
-
-func truncateUpperBoundText(s string, trunc int) string {
-	if trunc == utf8.RuneCountInString(s) {
-		return s
-	}
-
-	result := []rune(s)[:trunc]
-	for i := len(result) - 1; i >= 0; i-- {
-		next := result[i] + 1
-		if utf8.ValidRune(next) {
-			result[i] = next
-
-			return string(result)
-		}
-	}
-
-	return ""
-}
-
-func truncateUpperBoundBinary(val []byte, trunc int) []byte {
-	result := val[:trunc]
-	if bytes.Equal(result, val) {
-		return result
-	}
-
-	for i := len(result) - 1; i >= 0; i-- {
-		if result[i] < 255 {
-			result[i]++
-
-			return result
-		}
-	}
-
-	return nil
-}
diff --git a/table/transaction.go b/table/transaction.go
index b5cbdee..c9c3c06 100644
--- a/table/transaction.go
+++ b/table/transaction.go
@@ -121,7 +121,7 @@ func (t *Transaction) Append(rdr array.RecordReader, snapshotProps iceberg.Prope
 	return iceberg.ErrNotImplemented
 }
 
-func (t *Transaction) AddFiles(files []string, snapshotProps iceberg.Properties, ignoreDuplicates bool) error {
+func (t *Transaction) AddFiles(ctx context.Context, files []string, snapshotProps iceberg.Properties, ignoreDuplicates bool) error {
 	set := make(map[string]string)
 	for _, f := range files {
 		set[f] = f
@@ -163,7 +163,7 @@ func (t *Transaction) AddFiles(files []string, snapshotProps iceberg.Properties,
 
 	updater := t.updateSnapshot(snapshotProps).fastAppend()
 
-	dataFiles := parquetFilesToDataFiles(t.tbl.fs, t.meta, slices.Values(files))
+	dataFiles := filesToDataFiles(ctx, t.tbl.fs, t.meta, slices.Values(files))
 	for df, err := range dataFiles {
 		if err != nil {
 			return err


Test patch:
diff --git a/catalog/glue/glue_test.go b/catalog/glue/glue_test.go
index 5d2df2b..ddedf4a 100644
--- a/catalog/glue/glue_test.go
+++ b/catalog/glue/glue_test.go
@@ -938,3 +938,47 @@ func TestGlueCreateTableRollbackOnInvalidMetadata(t *testing.T) {
 	mockGlueSvc.AssertCalled(t, "DeleteTable", mock.Anything, mock.Anything, mock.Anything)
 	mockGlueSvc.AssertCalled(t, "GetTable", mock.Anything, mock.Anything, mock.Anything)
 }
+
+func TestRegisterTableMetadataNotFound(t *testing.T) {
+	assert := require.New(t)
+	mockGlueSvc := &mockGlueClient{}
+	awsCfg, err := config.LoadDefaultConfig(context.TODO())
+	assert.NoError(err)
+	cat := &Catalog{
+		glueSvc: mockGlueSvc,
+		awsCfg:  &awsCfg,
+	}
+	_, err = cat.RegisterTable(context.Background(), catalog.ToIdentifier("test_db", "test_table"), "s3://nonexistent-bucket/metadata/metadata.json")
+	assert.Error(err)
+	assert.Contains(err.Error(), "failed to read table metadata from s3://nonexistent-bucket/metadata/metadata.json")
+}
+
+func TestRegisterTableIntegration(t *testing.T) {
+	if os.Getenv("TEST_DATABASE_NAME") == "" {
+		t.Skip()
+	}
+	if os.Getenv("TEST_TABLE_LOCATION") == "" {
+		t.Skip()
+	}
+	assert := require.New(t)
+	dbName := os.Getenv("TEST_DATABASE_NAME")
+	tableName := "test_register_table_integration"
+	// Metadata location example s3://test-bucket/metadata/0000-0000-0000.metadata.json
+	metadataLocation := os.Getenv("TEST_TABLE_LOCATION")
+
+	awsCfg, err := config.LoadDefaultConfig(context.TODO(), config.WithClientLogMode(aws.LogRequest|aws.LogResponse))
+	assert.NoError(err)
+	ctlg := NewCatalog(WithAwsConfig(awsCfg))
+
+	// Drop table if it exists
+	_ = ctlg.DropTable(context.TODO(), TableIdentifier(dbName, tableName))
+
+	tbl, err := ctlg.RegisterTable(context.TODO(), TableIdentifier(dbName, tableName), metadataLocation)
+	defer func() {
+		err = ctlg.DropTable(context.TODO(), TableIdentifier(dbName, tableName))
+		assert.NoError(err)
+	}()
+	assert.NoError(err)
+	assert.Equal([]string{tableName}, tbl.Identifier())
+	assert.Equal(metadataLocation, tbl.MetadataLocation())
+}
diff --git a/catalog/rest/rest_integration_test.go b/catalog/rest/rest_integration_test.go
index b17725a..b68a32f 100644
--- a/catalog/rest/rest_integration_test.go
+++ b/catalog/rest/rest_integration_test.go
@@ -21,12 +21,17 @@ package rest_test
 
 import (
 	"context"
+	"net/url"
 	"testing"
 
+	"github.com/apache/arrow-go/v18/arrow/array"
+	"github.com/apache/arrow-go/v18/arrow/memory"
+	"github.com/apache/arrow-go/v18/parquet/pqarrow"
 	"github.com/apache/iceberg-go"
 	"github.com/apache/iceberg-go/catalog"
 	"github.com/apache/iceberg-go/catalog/rest"
 	"github.com/apache/iceberg-go/io"
+	"github.com/apache/iceberg-go/table"
 	"github.com/stretchr/testify/suite"
 )
 
@@ -43,6 +48,7 @@ func (s *RestIntegrationSuite) loadCatalog(ctx context.Context) *rest.Catalog {
 	cat, err := catalog.Load(ctx, "local", iceberg.Properties{
 		"type":               "rest",
 		"uri":                "http://localhost:8181",
+		io.S3Region:          "us-east-1",
 		io.S3AccessKeyID:     "admin",
 		io.S3SecretAccessKey: "password",
 	})
@@ -60,7 +66,7 @@ var (
 	tableSchemaNested = iceberg.NewSchemaWithIdentifiers(1,
 		[]int{1},
 		iceberg.NestedField{
-			ID: 1, Name: "foo", Type: iceberg.PrimitiveTypes.String, Required: false},
+			ID: 1, Name: "foo", Type: iceberg.PrimitiveTypes.String, Required: true},
 		iceberg.NestedField{
 			ID: 2, Name: "bar", Type: iceberg.PrimitiveTypes.Int32, Required: true},
 		iceberg.NestedField{
@@ -175,6 +181,72 @@ func (s *RestIntegrationSuite) TestCreateTable() {
 	s.Require().NoError(s.cat.DropTable(s.ctx, catalog.ToIdentifier(TestNamespaceIdent, "test-table")))
 }
 
+func (s *RestIntegrationSuite) TestWriteCommitTable() {
+	s.ensureNamespace()
+
+	const location = "s3://warehouse/iceberg"
+
+	tbl, err := s.cat.CreateTable(s.ctx,
+		catalog.ToIdentifier(TestNamespaceIdent, "test-table-2"),
+		tableSchemaNested, catalog.WithLocation(location))
+	s.Require().NoError(err)
+	s.Require().NotNil(tbl)
+
+	defer func() {
+		s.Require().NoError(s.cat.DropTable(s.ctx, catalog.ToIdentifier(TestNamespaceIdent, "test-table-2")))
+	}()
+
+	s.Equal(location, tbl.Location())
+
+	arrSchema, err := table.SchemaToArrowSchema(tableSchemaNested, nil, false, false)
+	s.Require().NoError(err)
+
+	table, err := array.TableFromJSON(memory.DefaultAllocator, arrSchema,
+		[]string{`[
+		{
+			"foo": "foo_string",
+			"bar": 123,
+			"baz": true,
+			"qux": ["a", "b", "c"],
+			"quux": [{"key": "gopher", "value": [
+				{"key": "golang", "value": "1337"}]}],
+			"location": [{"latitude": 37.7749, "longitude": -122.4194}],
+			"person": {"name": "gopher", "age": 10}
+		}
+	]`})
+	s.Require().NoError(err)
+	defer table.Release()
+
+	pqfile, err := url.JoinPath(location, "data", "test_commit_table_data", "test.parquet")
+	s.Require().NoError(err)
+
+	fw, err := tbl.FS().(io.WriteFileIO).Create(pqfile)
+	s.Require().NoError(err)
+	s.Require().NoError(pqarrow.WriteTable(table, fw, table.NumRows(),
+		nil, pqarrow.DefaultWriterProps()))
+	defer tbl.FS().Remove(pqfile)
+
+	txn := tbl.NewTransaction()
+	s.Require().NoError(txn.AddFiles([]string{pqfile}, nil, false))
+	updated, err := txn.Commit(s.ctx)
+	s.Require().NoError(err)
+
+	mf := []iceberg.ManifestFile{}
+	for m, err := range updated.AllManifests() {
+		s.Require().NoError(err)
+		s.Require().NotNil(m)
+		mf = append(mf, m)
+	}
+
+	s.Len(mf, 1)
+	s.EqualValues(1, mf[0].AddedDataFiles())
+	entries, err := mf[0].FetchEntries(updated.FS(), false)
+	s.Require().NoError(err)
+
+	s.Len(entries, 1)
+	s.Equal(pqfile, entries[0].DataFile().FilePath())
+}
+
 func TestRestIntegration(t *testing.T) {
 	suite.Run(t, new(RestIntegrationSuite))
 }
diff --git a/catalog/sql/sql_test.go b/catalog/sql/sql_test.go
index 0f3bb52..496f230 100644
--- a/catalog/sql/sql_test.go
+++ b/catalog/sql/sql_test.go
@@ -1028,7 +1028,7 @@ func (s *SqliteCatalogTestSuite) TestCommitTable() {
 		s.EqualValues(0, tbl.Metadata().CurrentSchema().ID)
 
 		tx := tbl.NewTransaction()
-		s.Require().NoError(tx.AddFiles([]string{pqfile}, nil, false))
+		s.Require().NoError(tx.AddFiles(ctx, []string{pqfile}, nil, false))
 		updated, err := tx.Commit(ctx)
 		s.Require().NoError(err)
 
diff --git a/table/arrow_utils_internal_test.go b/table/arrow_utils_internal_test.go
index 3b478e9..27fe6cb 100644
--- a/table/arrow_utils_internal_test.go
+++ b/table/arrow_utils_internal_test.go
@@ -21,21 +21,17 @@ import (
 	"bytes"
 	"cmp"
 	"math"
-	"math/big"
 	"slices"
-	"strings"
 	"testing"
-	"time"
 
 	"github.com/apache/arrow-go/v18/arrow/array"
-	"github.com/apache/arrow-go/v18/arrow/decimal128"
 	"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/google/uuid"
+	"github.com/apache/iceberg-go/table/internal"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"github.com/stretchr/testify/suite"
@@ -177,6 +173,8 @@ type FileStatsMetricsSuite struct {
 }
 
 func (suite *FileStatsMetricsSuite) getDataFile(meta iceberg.Properties, writeStats []string) iceberg.DataFile {
+	format := internal.GetFileFormat(iceberg.ParquetFile)
+
 	fileMeta, tableMeta := constructTestTable(suite.T(), writeStats)
 	require.NotNil(suite.T(), tableMeta)
 	require.NotNil(suite.T(), fileMeta)
@@ -193,12 +191,12 @@ func (suite *FileStatsMetricsSuite) getDataFile(meta iceberg.Properties, writeSt
 
 	collector, err := computeStatsPlan(schema, tableMeta.Properties())
 	suite.Require().NoError(err)
-	mapping, err := parquetPathToIDMapping(schema)
+	mapping, err := format.PathToIDMapping(schema)
 	suite.Require().NoError(err)
 
-	stats := dataFileStatsFromParquetMetadata(fileMeta, collector, mapping)
+	stats := format.DataFileStatsFromMeta(fileMeta, collector, mapping)
 
-	return stats.toDataFile(tableMeta.CurrentSchema(), tableMeta.PartitionSpec(), "fake-path.parquet",
+	return stats.ToDataFile(tableMeta.CurrentSchema(), tableMeta.PartitionSpec(), "fake-path.parquet",
 		iceberg.ParquetFile, fileMeta.GetSourceFileSize())
 }
 
@@ -357,39 +355,6 @@ 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{
@@ -455,13 +420,13 @@ func TestStatsTypes(t *testing.T) {
 	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)
+	assert.True(t, slices.IsSortedFunc(statsCols, func(a, b internal.StatisticsCollector) int {
+		return cmp.Compare(a.FieldID, b.FieldID)
 	}))
 
 	actual := make([]iceberg.Type, len(statsCols))
 	for i, col := range statsCols {
-		actual[i] = col.icebergTyp
+		actual[i] = col.IcebergTyp
 	}
 
 	assert.Equal(t, []iceberg.Type{
@@ -478,193 +443,3 @@ func TestStatsTypes(t *testing.T) {
 		iceberg.PrimitiveTypes.Int32,
 	}, actual)
 }
-
-func constructTestTablePrimitiveTypes(t *testing.T) (*metadata.FileMetaData, Metadata) {
-	tableMeta, err := ParseMetadataString(`{
-        "format-version": 2,
-        "location": "s3://bucket/test/location",
-        "last-column-id": 7,
-        "current-schema-id": 0,
-        "schemas": [
-            {
-                "type": "struct",
-                "schema-id": 0,
-                "fields": [
-                    {"id": 1, "name": "booleans", "required": false, "type": "boolean"},
-                    {"id": 2, "name": "ints", "required": false, "type": "int"},
-                    {"id": 3, "name": "longs", "required": false, "type": "long"},
-                    {"id": 4, "name": "floats", "required": false, "type": "float"},
-                    {"id": 5, "name": "doubles", "required": false, "type": "double"},
-                    {"id": 6, "name": "dates", "required": false, "type": "date"},
-                    {"id": 7, "name": "times", "required": false, "type": "time"},
-                    {"id": 8, "name": "timestamps", "required": false, "type": "timestamp"},
-                    {"id": 9, "name": "timestamptzs", "required": false, "type": "timestamptz"},
-                    {"id": 10, "name": "strings", "required": false, "type": "string"},
-                    {"id": 11, "name": "uuids", "required": false, "type": "uuid"},
-                    {"id": 12, "name": "binaries", "required": false, "type": "binary"},
-					{"id": 13, "name": "small_dec", "required": false, "type": "decimal(8, 2)"},
-					{"id": 14, "name": "med_dec", "required": false, "type": "decimal(16, 2)"},
-					{"id": 15, "name": "large_dec", "required": false, "type": "decimal(24, 2)"}
-                ]
-            }
-        ],
-		"last-updated-ms": -1,
-        "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)
-
-	rec, _, err := array.RecordFromJSON(memory.DefaultAllocator, arrowSchema, strings.NewReader(`[
-		{
-			"booleans": true, 
-			"ints": 23,
-			"longs": 54,
-			"floats": 454.1223,
-			"doubles": 8542.12,
-			"dates": "2022-01-02",
-			"times": "17:30:34",
-			"timestamps": "2022-01-02T17:30:34.399",
-			"timestamptzs": "2022-01-02T17:30:34.399",
-			"strings": "hello",
-			"uuids": "`+uuid.NewMD5(uuid.NameSpaceDNS, []byte("foo")).String()+`",
-			"binaries": "aGVsbG8=",
-			"small_dec": "123456.78",
-			"med_dec": "12345678901234.56",
-			"large_dec": "1234567890123456789012.34"
-		},
-		{
-			"booleans": false,
-			"ints": 89,
-			"longs": 2,
-			"floats": 24342.29,
-			"doubles": -43.9,
-			"dates": "2023-02-04",
-			"times": "13:21:04",
-			"timestamps": "2023-02-04T13:21:04.354",
-			"timestamptzs": "2023-02-04T13:21:04.354",
-			"strings": "world",
-			"uuids": "`+uuid.NewMD5(uuid.NameSpaceDNS, []byte("bar")).String()+`",
-			"binaries": "d29ybGQ=",
-			"small_dec": "876543.21",
-			"med_dec": "65432109876543.21",
-			"large_dec": "4321098765432109876543.21"
-		}
-	]`))
-	require.NoError(t, err)
-	defer rec.Release()
-
-	var buf bytes.Buffer
-	wr, err := pqarrow.NewFileWriter(arrowSchema, &buf,
-		parquet.NewWriterProperties(parquet.WithStats(true),
-			parquet.WithStoreDecimalAsInteger(true)),
-		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
-}
-
-func assertBounds[T iceberg.LiteralType](t *testing.T, bound []byte, typ iceberg.Type, expected T) {
-	lit, err := iceberg.LiteralFromBytes(typ, bound)
-	require.NoError(t, err)
-	assert.Equal(t, expected, lit.(iceberg.TypedLiteral[T]).Value())
-}
-
-func toDate(tm time.Time) iceberg.Date {
-	return iceberg.Date(tm.Truncate(24*time.Hour).Unix() / int64((time.Hour * 24).Seconds()))
-}
-
-func TestMetricsPrimitiveTypes(t *testing.T) {
-	meta, tblMeta := constructTestTablePrimitiveTypes(t)
-	require.NotNil(t, tblMeta)
-	require.NotNil(t, meta)
-
-	tblMeta.Properties()["write.metadata.metrics.default"] = "truncate(2)"
-
-	collector, err := computeStatsPlan(tblMeta.CurrentSchema(), tblMeta.Properties())
-	require.NoError(t, err)
-	mapping, err := parquetPathToIDMapping(tblMeta.CurrentSchema())
-	require.NoError(t, err)
-
-	stats := dataFileStatsFromParquetMetadata(meta, collector, mapping)
-	df := stats.toDataFile(tblMeta.CurrentSchema(), tblMeta.PartitionSpec(), "fake-path.parquet",
-		iceberg.ParquetFile, meta.GetSourceFileSize())
-
-	assert.Len(t, df.ValueCounts(), 15)
-	assert.Len(t, df.NullValueCounts(), 15)
-	assert.Len(t, df.NaNValueCounts(), 0)
-
-	assert.Len(t, df.LowerBoundValues(), 15)
-	assertBounds(t, df.LowerBoundValues()[1], iceberg.PrimitiveTypes.Bool, false)
-	assertBounds(t, df.LowerBoundValues()[2], iceberg.PrimitiveTypes.Int32, int32(23))
-	assertBounds(t, df.LowerBoundValues()[3], iceberg.PrimitiveTypes.Int64, int64(2))
-	assertBounds(t, df.LowerBoundValues()[4], iceberg.PrimitiveTypes.Float32, float32(454.1223))
-	assertBounds(t, df.LowerBoundValues()[5], iceberg.PrimitiveTypes.Float64, -43.9)
-	assertBounds(t, df.LowerBoundValues()[6], iceberg.PrimitiveTypes.Date,
-		toDate(time.Date(2022, time.January, 2, 0, 0, 0, 0, time.UTC)))
-	assertBounds(t, df.LowerBoundValues()[7], iceberg.PrimitiveTypes.Time,
-		iceberg.Time(time.Duration(13*time.Hour+21*time.Minute+4*time.Second).Microseconds()))
-	assertBounds(t, df.LowerBoundValues()[8], iceberg.PrimitiveTypes.Timestamp,
-		iceberg.Timestamp(time.Date(2022, time.January, 2, 17, 30, 34, 399000000, time.UTC).UnixMicro()))
-	assertBounds(t, df.LowerBoundValues()[9], iceberg.PrimitiveTypes.TimestampTz,
-		iceberg.Timestamp(time.Date(2022, time.January, 2, 17, 30, 34, 399000000, time.UTC).UnixMicro()))
-	assertBounds(t, df.LowerBoundValues()[10], iceberg.PrimitiveTypes.String, "he")
-	assertBounds(t, df.LowerBoundValues()[11], iceberg.PrimitiveTypes.UUID,
-		uuid.NewMD5(uuid.NameSpaceDNS, []byte("foo")))
-	assertBounds(t, df.LowerBoundValues()[12], iceberg.PrimitiveTypes.Binary, []byte("he"))
-	assertBounds(t, df.LowerBoundValues()[13], iceberg.DecimalTypeOf(8, 2), iceberg.Decimal{
-		Val:   decimal128.FromI64(12345678),
-		Scale: 2,
-	})
-	assertBounds(t, df.LowerBoundValues()[14], iceberg.DecimalTypeOf(16, 2), iceberg.Decimal{
-		Val:   decimal128.FromI64(1234567890123456),
-		Scale: 2,
-	})
-	expected, _ := (&big.Int{}).SetString("123456789012345678901234", 10)
-	assertBounds(t, df.LowerBoundValues()[15], iceberg.DecimalTypeOf(24, 2), iceberg.Decimal{
-		Val:   decimal128.FromBigInt(expected),
-		Scale: 2,
-	})
-
-	assert.Len(t, df.UpperBoundValues(), 15)
-	assertBounds(t, df.UpperBoundValues()[1], iceberg.PrimitiveTypes.Bool, true)
-	assertBounds(t, df.UpperBoundValues()[2], iceberg.PrimitiveTypes.Int32, int32(89))
-	assertBounds(t, df.UpperBoundValues()[3], iceberg.PrimitiveTypes.Int64, int64(54))
-	assertBounds(t, df.UpperBoundValues()[4], iceberg.PrimitiveTypes.Float32, float32(24342.29))
-	assertBounds(t, df.UpperBoundValues()[5], iceberg.PrimitiveTypes.Float64, 8542.12)
-	assertBounds(t, df.UpperBoundValues()[6], iceberg.PrimitiveTypes.Date,
-		toDate(time.Date(2023, time.February, 4, 0, 0, 0, 0, time.UTC)))
-	assertBounds(t, df.UpperBoundValues()[7], iceberg.PrimitiveTypes.Time,
-		iceberg.Time(time.Duration(17*time.Hour+30*time.Minute+34*time.Second).Microseconds()))
-	assertBounds(t, df.UpperBoundValues()[8], iceberg.PrimitiveTypes.Timestamp,
-		iceberg.Timestamp(time.Date(2023, time.February, 4, 13, 21, 4, 354000000, time.UTC).UnixMicro()))
-	assertBounds(t, df.UpperBoundValues()[9], iceberg.PrimitiveTypes.TimestampTz,
-		iceberg.Timestamp(time.Date(2023, time.February, 4, 13, 21, 4, 354000000, time.UTC).UnixMicro()))
-	assertBounds(t, df.UpperBoundValues()[10], iceberg.PrimitiveTypes.String, "wp")
-	assertBounds(t, df.UpperBoundValues()[11], iceberg.PrimitiveTypes.UUID,
-		uuid.NewMD5(uuid.NameSpaceDNS, []byte("bar")))
-	assertBounds(t, df.UpperBoundValues()[12], iceberg.PrimitiveTypes.Binary, []byte("wp"))
-	assertBounds(t, df.UpperBoundValues()[13], iceberg.DecimalTypeOf(8, 2), iceberg.Decimal{
-		Val:   decimal128.FromI64(87654321),
-		Scale: 2,
-	})
-	assertBounds(t, df.UpperBoundValues()[14], iceberg.DecimalTypeOf(16, 2), iceberg.Decimal{
-		Val:   decimal128.FromI64(6543210987654321),
-		Scale: 2,
-	})
-	expectedUpper, _ := (&big.Int{}).SetString("432109876543210987654321", 10)
-	assertBounds(t, df.UpperBoundValues()[15], iceberg.DecimalTypeOf(24, 2), iceberg.Decimal{
-		Val:   decimal128.FromBigInt(expectedUpper),
-		Scale: 2,
-	})
-}
diff --git a/table/internal/parquet_files_test.go b/table/internal/parquet_files_test.go
new file mode 100644
index 0000000..cde72d8
--- /dev/null
+++ b/table/internal/parquet_files_test.go
@@ -0,0 +1,325 @@
+// 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 internal_test
+
+import (
+	"bytes"
+	"math/big"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/apache/arrow-go/v18/arrow/array"
+	"github.com/apache/arrow-go/v18/arrow/decimal128"
+	"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/apache/iceberg-go/table"
+	"github.com/apache/iceberg-go/table/internal"
+	"github.com/google/uuid"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func constructTestTablePrimitiveTypes(t *testing.T) (*metadata.FileMetaData, table.Metadata) {
+	tableMeta, err := table.ParseMetadataString(`{
+        "format-version": 2,
+        "location": "s3://bucket/test/location",
+        "last-column-id": 7,
+        "current-schema-id": 0,
+        "schemas": [
+            {
+                "type": "struct",
+                "schema-id": 0,
+                "fields": [
+                    {"id": 1, "name": "booleans", "required": false, "type": "boolean"},
+                    {"id": 2, "name": "ints", "required": false, "type": "int"},
+                    {"id": 3, "name": "longs", "required": false, "type": "long"},
+                    {"id": 4, "name": "floats", "required": false, "type": "float"},
+                    {"id": 5, "name": "doubles", "required": false, "type": "double"},
+                    {"id": 6, "name": "dates", "required": false, "type": "date"},
+                    {"id": 7, "name": "times", "required": false, "type": "time"},
+                    {"id": 8, "name": "timestamps", "required": false, "type": "timestamp"},
+                    {"id": 9, "name": "timestamptzs", "required": false, "type": "timestamptz"},
+                    {"id": 10, "name": "strings", "required": false, "type": "string"},
+                    {"id": 11, "name": "uuids", "required": false, "type": "uuid"},
+                    {"id": 12, "name": "binaries", "required": false, "type": "binary"},
+					{"id": 13, "name": "small_dec", "required": false, "type": "decimal(8, 2)"},
+					{"id": 14, "name": "med_dec", "required": false, "type": "decimal(16, 2)"},
+					{"id": 15, "name": "large_dec", "required": false, "type": "decimal(24, 2)"}
+                ]
+            }
+        ],
+		"last-updated-ms": -1,
+        "default-spec-id": 0,
+        "partition-specs": [{"spec-id": 0, "fields": []}],
+        "properties": {}
+	}`)
+	require.NoError(t, err)
+
+	arrowSchema, err := table.SchemaToArrowSchema(tableMeta.Schemas()[0], nil, true, false)
+	require.NoError(t, err)
+
+	rec, _, err := array.RecordFromJSON(memory.DefaultAllocator, arrowSchema, strings.NewReader(`[
+		{
+			"booleans": true, 
+			"ints": 23,
+			"longs": 54,
+			"floats": 454.1223,
+			"doubles": 8542.12,
+			"dates": "2022-01-02",
+			"times": "17:30:34",
+			"timestamps": "2022-01-02T17:30:34.399",
+			"timestamptzs": "2022-01-02T17:30:34.399",
+			"strings": "hello",
+			"uuids": "`+uuid.NewMD5(uuid.NameSpaceDNS, []byte("foo")).String()+`",
+			"binaries": "aGVsbG8=",
+			"small_dec": "123456.78",
+			"med_dec": "12345678901234.56",
+			"large_dec": "1234567890123456789012.34"
+		},
+		{
+			"booleans": false,
+			"ints": 89,
+			"longs": 2,
+			"floats": 24342.29,
+			"doubles": -43.9,
+			"dates": "2023-02-04",
+			"times": "13:21:04",
+			"timestamps": "2023-02-04T13:21:04.354",
+			"timestamptzs": "2023-02-04T13:21:04.354",
+			"strings": "world",
+			"uuids": "`+uuid.NewMD5(uuid.NameSpaceDNS, []byte("bar")).String()+`",
+			"binaries": "d29ybGQ=",
+			"small_dec": "876543.21",
+			"med_dec": "65432109876543.21",
+			"large_dec": "4321098765432109876543.21"
+		}
+	]`))
+	require.NoError(t, err)
+	defer rec.Release()
+
+	var buf bytes.Buffer
+	wr, err := pqarrow.NewFileWriter(arrowSchema, &buf,
+		parquet.NewWriterProperties(parquet.WithStats(true)),
+		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
+}
+
+func assertBounds[T iceberg.LiteralType](t *testing.T, bound []byte, typ iceberg.Type, expected T) {
+	lit, err := iceberg.LiteralFromBytes(typ, bound)
+	require.NoError(t, err)
+	assert.Equal(t, expected, lit.(iceberg.TypedLiteral[T]).Value())
+}
+
+func toDate(tm time.Time) iceberg.Date {
+	return iceberg.Date(tm.Truncate(24*time.Hour).Unix() / int64((time.Hour * 24).Seconds()))
+}
+
+func getCollector() map[int]internal.StatisticsCollector {
+	modeTrunc := internal.MetricsMode{Typ: internal.MetricModeTruncate, Len: 2}
+	modeFull := internal.MetricsMode{Typ: internal.MetricModeFull}
+
+	return map[int]internal.StatisticsCollector{
+		1: {
+			FieldID:    1,
+			Mode:       modeFull,
+			ColName:    "booleans",
+			IcebergTyp: iceberg.PrimitiveTypes.Bool,
+		},
+		2: {
+			FieldID:    2,
+			Mode:       modeFull,
+			ColName:    "ints",
+			IcebergTyp: iceberg.PrimitiveTypes.Int32,
+		},
+		3: {
+			FieldID:    3,
+			Mode:       modeFull,
+			ColName:    "longs",
+			IcebergTyp: iceberg.PrimitiveTypes.Int64,
+		},
+		4: {
+			FieldID:    4,
+			Mode:       modeFull,
+			ColName:    "floats",
+			IcebergTyp: iceberg.PrimitiveTypes.Float32,
+		},
+		5: {
+			FieldID:    5,
+			Mode:       modeFull,
+			ColName:    "doubles",
+			IcebergTyp: iceberg.PrimitiveTypes.Float64,
+		},
+		6: {
+			FieldID:    6,
+			Mode:       modeFull,
+			ColName:    "dates",
+			IcebergTyp: iceberg.PrimitiveTypes.Date,
+		},
+		7: {
+			FieldID:    7,
+			Mode:       modeFull,
+			ColName:    "times",
+			IcebergTyp: iceberg.PrimitiveTypes.Time,
+		},
+		8: {
+			FieldID:    8,
+			Mode:       modeFull,
+			ColName:    "timestamps",
+			IcebergTyp: iceberg.PrimitiveTypes.Timestamp,
+		},
+		9: {
+			FieldID:    9,
+			Mode:       modeFull,
+			ColName:    "timestamptzs",
+			IcebergTyp: iceberg.PrimitiveTypes.TimestampTz,
+		},
+		10: {
+			FieldID:    10,
+			Mode:       modeTrunc,
+			ColName:    "strings",
+			IcebergTyp: iceberg.PrimitiveTypes.String,
+		},
+		11: {
+			FieldID:    11,
+			Mode:       modeFull,
+			ColName:    "uuids",
+			IcebergTyp: iceberg.PrimitiveTypes.UUID,
+		},
+		12: {
+			FieldID:    12,
+			Mode:       modeTrunc,
+			ColName:    "binaries",
+			IcebergTyp: iceberg.PrimitiveTypes.Binary,
+		},
+		13: {
+			FieldID:    13,
+			Mode:       modeFull,
+			ColName:    "small_dec",
+			IcebergTyp: iceberg.DecimalTypeOf(8, 2),
+		},
+		14: {
+			FieldID:    14,
+			Mode:       modeFull,
+			ColName:    "med_dec",
+			IcebergTyp: iceberg.DecimalTypeOf(16, 2),
+		},
+		15: {
+			FieldID:    15,
+			Mode:       modeFull,
+			ColName:    "large_dec",
+			IcebergTyp: iceberg.DecimalTypeOf(24, 2),
+		},
+	}
+}
+
+func TestMetricsPrimitiveTypes(t *testing.T) {
+	format := internal.GetFileFormat(iceberg.ParquetFile)
+
+	meta, tblMeta := constructTestTablePrimitiveTypes(t)
+	require.NotNil(t, tblMeta)
+	require.NotNil(t, meta)
+
+	mapping, err := format.PathToIDMapping(tblMeta.CurrentSchema())
+	require.NoError(t, err)
+
+	stats := format.DataFileStatsFromMeta(internal.Metadata(meta), getCollector(), mapping)
+	df := stats.ToDataFile(tblMeta.CurrentSchema(), tblMeta.PartitionSpec(), "fake-path.parquet",
+		iceberg.ParquetFile, meta.GetSourceFileSize())
+
+	assert.Len(t, df.ValueCounts(), 15)
+	assert.Len(t, df.NullValueCounts(), 15)
+	assert.Len(t, df.NaNValueCounts(), 0)
+
+	assert.Len(t, df.LowerBoundValues(), 15)
+	assertBounds(t, df.LowerBoundValues()[1], iceberg.PrimitiveTypes.Bool, false)
+	assertBounds(t, df.LowerBoundValues()[2], iceberg.PrimitiveTypes.Int32, int32(23))
+	assertBounds(t, df.LowerBoundValues()[3], iceberg.PrimitiveTypes.Int64, int64(2))
+	assertBounds(t, df.LowerBoundValues()[4], iceberg.PrimitiveTypes.Float32, float32(454.1223))
+	assertBounds(t, df.LowerBoundValues()[5], iceberg.PrimitiveTypes.Float64, -43.9)
+	assertBounds(t, df.LowerBoundValues()[6], iceberg.PrimitiveTypes.Date,
+		toDate(time.Date(2022, time.January, 2, 0, 0, 0, 0, time.UTC)))
+	assertBounds(t, df.LowerBoundValues()[7], iceberg.PrimitiveTypes.Time,
+		iceberg.Time(time.Duration(13*time.Hour+21*time.Minute+4*time.Second).Microseconds()))
+	assertBounds(t, df.LowerBoundValues()[8], iceberg.PrimitiveTypes.Timestamp,
+		iceberg.Timestamp(time.Date(2022, time.January, 2, 17, 30, 34, 399000000, time.UTC).UnixMicro()))
+	assertBounds(t, df.LowerBoundValues()[9], iceberg.PrimitiveTypes.TimestampTz,
+		iceberg.Timestamp(time.Date(2022, time.January, 2, 17, 30, 34, 399000000, time.UTC).UnixMicro()))
+	assertBounds(t, df.LowerBoundValues()[10], iceberg.PrimitiveTypes.String, "he")
+	assertBounds(t, df.LowerBoundValues()[11], iceberg.PrimitiveTypes.UUID,
+		uuid.NewMD5(uuid.NameSpaceDNS, []byte("foo")))
+	assertBounds(t, df.LowerBoundValues()[12], iceberg.PrimitiveTypes.Binary, []byte("he"))
+	assertBounds(t, df.LowerBoundValues()[13], iceberg.DecimalTypeOf(8, 2), iceberg.Decimal{
+		Val:   decimal128.FromI64(12345678),
+		Scale: 2,
+	})
+	assertBounds(t, df.LowerBoundValues()[14], iceberg.DecimalTypeOf(16, 2), iceberg.Decimal{
+		Val:   decimal128.FromI64(1234567890123456),
+		Scale: 2,
+	})
+	expected, _ := (&big.Int{}).SetString("123456789012345678901234", 10)
+	assertBounds(t, df.LowerBoundValues()[15], iceberg.DecimalTypeOf(24, 2), iceberg.Decimal{
+		Val:   decimal128.FromBigInt(expected),
+		Scale: 2,
+	})
+
+	assert.Len(t, df.UpperBoundValues(), 15)
+	assertBounds(t, df.UpperBoundValues()[1], iceberg.PrimitiveTypes.Bool, true)
+	assertBounds(t, df.UpperBoundValues()[2], iceberg.PrimitiveTypes.Int32, int32(89))
+	assertBounds(t, df.UpperBoundValues()[3], iceberg.PrimitiveTypes.Int64, int64(54))
+	assertBounds(t, df.UpperBoundValues()[4], iceberg.PrimitiveTypes.Float32, float32(24342.29))
+	assertBounds(t, df.UpperBoundValues()[5], iceberg.PrimitiveTypes.Float64, 8542.12)
+	assertBounds(t, df.UpperBoundValues()[6], iceberg.PrimitiveTypes.Date,
+		toDate(time.Date(2023, time.February, 4, 0, 0, 0, 0, time.UTC)))
+	assertBounds(t, df.UpperBoundValues()[7], iceberg.PrimitiveTypes.Time,
+		iceberg.Time(time.Duration(17*time.Hour+30*time.Minute+34*time.Second).Microseconds()))
+	assertBounds(t, df.UpperBoundValues()[8], iceberg.PrimitiveTypes.Timestamp,
+		iceberg.Timestamp(time.Date(2023, time.February, 4, 13, 21, 4, 354000000, time.UTC).UnixMicro()))
+	assertBounds(t, df.UpperBoundValues()[9], iceberg.PrimitiveTypes.TimestampTz,
+		iceberg.Timestamp(time.Date(2023, time.February, 4, 13, 21, 4, 354000000, time.UTC).UnixMicro()))
+	assertBounds(t, df.UpperBoundValues()[10], iceberg.PrimitiveTypes.String, "wp")
+	assertBounds(t, df.UpperBoundValues()[11], iceberg.PrimitiveTypes.UUID,
+		uuid.NewMD5(uuid.NameSpaceDNS, []byte("bar")))
+	assertBounds(t, df.UpperBoundValues()[12], iceberg.PrimitiveTypes.Binary, []byte("wp"))
+	assertBounds(t, df.UpperBoundValues()[13], iceberg.DecimalTypeOf(8, 2), iceberg.Decimal{
+		Val:   decimal128.FromI64(87654321),
+		Scale: 2,
+	})
+	assertBounds(t, df.UpperBoundValues()[14], iceberg.DecimalTypeOf(16, 2), iceberg.Decimal{
+		Val:   decimal128.FromI64(6543210987654321),
+		Scale: 2,
+	})
+	expectedUpper, _ := (&big.Int{}).SetString("432109876543210987654321", 10)
+	assertBounds(t, df.UpperBoundValues()[15], iceberg.DecimalTypeOf(24, 2), iceberg.Decimal{
+		Val:   decimal128.FromBigInt(expectedUpper),
+		Scale: 2,
+	})
+}
diff --git a/table/internal/utils_test.go b/table/internal/utils_test.go
new file mode 100644
index 0000000..95c83b1
--- /dev/null
+++ b/table/internal/utils_test.go
@@ -0,0 +1,70 @@
+// 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 internal_test
+
+import (
+	"testing"
+
+	"github.com/apache/iceberg-go/table/internal"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestMetricsModePairs(t *testing.T) {
+	tests := []struct {
+		str      string
+		expected internal.MetricsMode
+	}{
+		{"none", internal.MetricsMode{Typ: internal.MetricModeNone}},
+		{"nOnE", internal.MetricsMode{Typ: internal.MetricModeNone}},
+		{"counts", internal.MetricsMode{Typ: internal.MetricModeCounts}},
+		{"Counts", internal.MetricsMode{Typ: internal.MetricModeCounts}},
+		{"full", internal.MetricsMode{Typ: internal.MetricModeFull}},
+		{"FuLl", internal.MetricsMode{Typ: internal.MetricModeFull}},
+		{" FuLl", internal.MetricsMode{Typ: internal.MetricModeFull}},
+		{"truncate(16)", internal.MetricsMode{Typ: internal.MetricModeTruncate, Len: 16}},
+		{"trUncatE(16)", internal.MetricsMode{Typ: internal.MetricModeTruncate, Len: 16}},
+		{"trUncatE(7)", internal.MetricsMode{Typ: internal.MetricModeTruncate, Len: 7}},
+		{"trUncatE(07)", internal.MetricsMode{Typ: internal.MetricModeTruncate, Len: 7}},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.str, func(t *testing.T) {
+			mode, err := internal.MatchMetricsMode(tt.str)
+			require.NoError(t, err)
+			assert.Equal(t, tt.expected, mode)
+		})
+	}
+
+	_, err := internal.MatchMetricsMode("truncatE(-7)")
+	assert.ErrorContains(t, err, "malformed truncate metrics mode: truncatE(-7)")
+
+	_, err = internal.MatchMetricsMode("truncatE(0)")
+	assert.ErrorContains(t, err, "invalid truncate length: 0")
+}
+
+func TestTruncateUpperBoundString(t *testing.T) {
+	assert.Equal(t, "ab", internal.TruncateUpperBoundText("aaaa", 2))
+	// \u10FFFF\u10FFFF\x00
+	assert.Equal(t, "", internal.TruncateUpperBoundText("\xf4\x8f\xbf\xbf\xf4\x8f\xbf\xbf\x00", 2))
+}
+
+func TestTruncateUpperBoundBinary(t *testing.T) {
+	assert.Equal(t, []byte{0x01, 0x03}, internal.TruncateUpperBoundBinary([]byte{0x01, 0x02, 0x03}, 2))
+	assert.Nil(t, internal.TruncateUpperBoundBinary([]byte{0xff, 0xff, 0x00}, 2))
+}
diff --git a/table/scanner_test.go b/table/scanner_test.go
index e953817..aa0fad4 100644
--- a/table/scanner_test.go
+++ b/table/scanner_test.go
@@ -475,6 +475,30 @@ func (s *ScannerSuite) TestNestedColumns() {
 	s.EqualValues(5, results.NumRows())
 }
 
+func (s *ScannerSuite) TestIsInFilterTable() {
+	mem := memory.NewCheckedAllocator(memory.DefaultAllocator)
+	defer mem.AssertSize(s.T(), 0)
+
+	ident := catalog.ToIdentifier("default", "test_uuid_and_fixed_unpartitioned")
+
+	tbl, err := s.cat.LoadTable(s.ctx, ident, s.props)
+	s.Require().NoError(err)
+
+	ctx := compute.WithAllocator(s.ctx, mem)
+	results, err := tbl.Scan(table.WithRowFilter(
+		iceberg.NewNot(iceberg.IsIn(iceberg.Reference("uuid_col"),
+			"102cb62f-e6f8-4eb0-9973-d9b012ff0967",
+			"639cccce-c9d2-494a-a78c-278ab234f024"))),
+		table.WithSelectedFields("uuid_col")).ToArrowTable(ctx)
+	s.Require().NoError(err)
+	defer results.Release()
+
+	s.EqualValues(3, results.NumRows())
+	s.Equal([]string{"ec33e4b2-a834-4cc3-8c4a-a1d3bfc2f226",
+		"c1b0d8e0-0b0e-4b1e-9b0a-0e0b0d0c0a0b",
+		"923dae77-83d6-47cd-b4b0-d383e64ee57e"}, getStrValues(results.Column(0)))
+}
+
 func (s *ScannerSuite) TestUnpartitionedUUIDTable() {
 	mem := memory.NewCheckedAllocator(memory.DefaultAllocator)
 	defer mem.AssertSize(s.T(), 0)
diff --git a/table/snapshots_internal_test.go b/table/snapshots_internal_test.go
index 5c8afe0..c374178 100644
--- a/table/snapshots_internal_test.go
+++ b/table/snapshots_internal_test.go
@@ -194,14 +194,3 @@ func TestInvalidOperation(t *testing.T) {
 	_, err := updateSnapshotSummaries(Summary{Operation: OpReplace}, nil, false)
 	assert.ErrorIs(t, err, iceberg.ErrNotImplemented)
 }
-
-func TestTruncateUpperBoundString(t *testing.T) {
-	assert.Equal(t, "ab", truncateUpperBoundText("aaaa", 2))
-	// \u10FFFF\u10FFFF\x00
-	assert.Equal(t, "", truncateUpperBoundText("\xf4\x8f\xbf\xbf\xf4\x8f\xbf\xbf\x00", 2))
-}
-
-func TestTruncateUpperBoundBinary(t *testing.T) {
-	assert.Equal(t, []byte{0x01, 0x03}, truncateUpperBoundBinary([]byte{0x01, 0x02, 0x03}, 2))
-	assert.Nil(t, truncateUpperBoundBinary([]byte{0xff, 0xff, 0x00}, 2))
-}
diff --git a/table/table_test.go b/table/table_test.go
index 825fb7d..975f40d 100644
--- a/table/table_test.go
+++ b/table/table_test.go
@@ -142,6 +142,8 @@ func (t *TableTestSuite) TestSnapshotByName() {
 type TableWritingTestSuite struct {
 	suite.Suite
 
+	ctx context.Context
+
 	tableSchema      *iceberg.Schema
 	arrSchema        *arrow.Schema
 	arrTbl           arrow.Table
@@ -159,6 +161,7 @@ type TableWritingTestSuite struct {
 }
 
 func (t *TableWritingTestSuite) SetupSuite() {
+	t.ctx = context.Background()
 	mem := memory.DefaultAllocator
 
 	t.tableSchema = iceberg.NewSchema(0,
@@ -301,7 +304,7 @@ func (t *TableWritingTestSuite) TestAddFilesUnpartitioned() {
 	}
 
 	tx := tbl.NewTransaction()
-	t.Require().NoError(tx.AddFiles(files, nil, false))
+	t.Require().NoError(tx.AddFiles(t.ctx, files, nil, false))
 
 	stagedTbl, err := tx.StagedTable()
 	t.Require().NoError(err)
@@ -349,7 +352,7 @@ func (t *TableWritingTestSuite) TestAddFilesFileNotFound() {
 
 	files = append(files, t.location+"/unpartitioned_file_not_found/unknown.parquet")
 	tx := tbl.NewTransaction()
-	err := tx.AddFiles(files, nil, false)
+	err := tx.AddFiles(t.ctx, files, nil, false)
 	t.Error(err)
 	t.ErrorContains(err, "no such file or directory")
 }
@@ -369,7 +372,7 @@ func (t *TableWritingTestSuite) TestAddFilesUnpartitionedHasFieldIDs() {
 	}
 
 	tx := tbl.NewTransaction()
-	err := tx.AddFiles(files, nil, false)
+	err := tx.AddFiles(t.ctx, files, nil, false)
 	t.Error(err)
 	t.ErrorIs(err, iceberg.ErrNotImplemented)
 }
@@ -401,9 +404,9 @@ func (t *TableWritingTestSuite) TestAddFilesFailsSchemaMismatch() {
 	defer pterm.EnableOutput()
 
 	tx := tbl.NewTransaction()
-	err = tx.AddFiles(files, nil, false)
+	err = tx.AddFiles(t.ctx, files, nil, false)
 	t.Error(err)
-	t.EqualError(err, `error encountered during parquet file conversion: error encountered during schema visitor: mismatch in fields:
+	t.EqualError(err, `error encountered during schema visitor: mismatch in fields:
    | Table Field              | Requested Field         
 ✅ | 1: foo: optional boolean | 1: foo: optional boolean
 ✅ | 2: bar: optional string  | 2: bar: optional string 
@@ -441,7 +444,7 @@ func (t *TableWritingTestSuite) TestAddFilesPartitionedTable() {
 	}
 
 	tx := tbl.NewTransaction()
-	t.Require().NoError(tx.AddFiles(files, nil, false))
+	t.Require().NoError(tx.AddFiles(t.ctx, files, nil, false))
 
 	stagedTbl, err := tx.StagedTable()
 	t.Require().NoError(err)
@@ -499,7 +502,7 @@ func (t *TableWritingTestSuite) TestAddFilesToBucketPartitionedTableFails() {
 	}
 
 	tx := tbl.NewTransaction()
-	err := tx.AddFiles(files, nil, false)
+	err := tx.AddFiles(t.ctx, files, nil, false)
 	t.Error(err)
 	t.ErrorContains(err, "cannot infer partition value from parquet metadata for a non-linear partition field: baz_bucket_3 with transform bucket[3]")
 }
@@ -527,7 +530,7 @@ func (t *TableWritingTestSuite) TestAddFilesToPartitionedTableFailsLowerAndUpper
 	}
 
 	tx := tbl.NewTransaction()
-	err := tx.AddFiles(files, nil, false)
+	err := tx.AddFiles(t.ctx, files, nil, false)
 	t.Error(err)
 	t.ErrorContains(err, "cannot infer partition value from parquet metadata as there is more than one value for partition field: baz. (low: 123, high: 124)")
 }
@@ -565,7 +568,7 @@ func (t *TableWritingTestSuite) TestAddFilesWithLargeAndRegular() {
 	t.writeParquet(tbl.FS().(iceio.WriteFileIO), filePathLarge, arrTableLarge)
 
 	tx := tbl.NewTransaction()
-	t.Require().NoError(tx.AddFiles([]string{filePath, filePathLarge}, nil, false))
+	t.Require().NoError(tx.AddFiles(t.ctx, []string{filePath, filePathLarge}, nil, false))
 
 	scan, err := tx.Scan(table.WithOptions(iceberg.Properties{
 		table.ScanOptionArrowUseLargeTypes: "true",
@@ -588,7 +591,7 @@ func (t *TableWritingTestSuite) TestAddFilesValidUpcast() {
 	t.writeParquet(tbl.FS().(iceio.WriteFileIO), filePath, t.arrTablePromotedTypes)
 
 	tx := tbl.NewTransaction()
-	t.Require().NoError(tx.AddFiles([]string{filePath}, nil, false))
+	t.Require().NoError(tx.AddFiles(t.ctx, []string{filePath}, nil, false))
 
 	scan, err := tx.Scan()
 	t.Require().NoError(err)
@@ -636,7 +639,7 @@ func (t *TableWritingTestSuite) TestAddFilesSubsetOfSchema() {
 	t.writeParquet(tbl.FS().(iceio.WriteFileIO), filePath, withoutCol)
 
 	tx := tbl.NewTransaction()
-	t.Require().NoError(tx.AddFiles([]string{filePath}, nil, false))
+	t.Require().NoError(tx.AddFiles(t.ctx, []string{filePath}, nil, false))
 
 	scan, err := tx.Scan()
 	t.Require().NoError(err)
@@ -671,7 +674,7 @@ func (t *TableWritingTestSuite) TestAddFilesDuplicateFilesInFilePaths() {
 	t.writeParquet(tbl.FS().(iceio.WriteFileIO), filePath, t.arrTbl)
 
 	tx := tbl.NewTransaction()
-	err := tx.AddFiles([]string{filePath, filePath}, nil, false)
+	err := tx.AddFiles(t.ctx, []string{filePath, filePath}, nil, false)
 	t.Error(err)
 	t.ErrorContains(err, "file paths must be unique for AddFiles")
 }
@@ -691,9 +694,9 @@ func (t *TableWritingTestSuite) TestAddFilesReferencedByCurrentSnapshot() {
 	}
 
 	tx := tbl.NewTransaction()
-	t.Require().NoError(tx.AddFiles(files, nil, false))
+	t.Require().NoError(tx.AddFiles(t.ctx, files, nil, false))
 
-	err := tx.AddFiles(files[len(files)-1:], nil, false)
+	err := tx.AddFiles(t.ctx, files[len(files)-1:], nil, false)
 	t.Error(err)
 	t.ErrorContains(err, "cannot add files that are already referenced by table, files:")
 }
@@ -713,9 +716,9 @@ func (t *TableWritingTestSuite) TestAddFilesReferencedCurrentSnapshotIgnoreDupli
 	}
 
 	tx := tbl.NewTransaction()
-	t.Require().NoError(tx.AddFiles(files, nil, false))
+	t.Require().NoError(tx.AddFiles(t.ctx, files, nil, false))
 
-	t.Require().NoError(tx.AddFiles(files[len(files)-1:], nil, true))
+	t.Require().NoError(tx.AddFiles(t.ctx, files[len(files)-1:], nil, true))
 	staged, err := tx.StagedTable()
 	t.Require().NoError(err)