Repository: apache/iceberg-go
Pull request #417 (2025-05-03 14:58:11)

Add ManifestReader, which provides access to manifest metadata
This is an alternative to #415.

This closes #386.

This also exports the `ReadManifest` helper (previously named `readManifestEntries`), so the read flow has symmetry with the existing `NewManifestWriter` and `WriteManifest` for the write flow.



Patch:
diff --git a/catalog/catalog.go b/catalog/catalog.go
index 01b0e28..2866276 100644
--- a/catalog/catalog.go
+++ b/catalog/catalog.go
@@ -36,9 +36,7 @@ import (
 	"strings"
 
 	"github.com/apache/iceberg-go"
-	"github.com/apache/iceberg-go/catalog/internal"
 	iceinternal "github.com/apache/iceberg-go/internal"
-	"github.com/apache/iceberg-go/io"
 	"github.com/apache/iceberg-go/table"
 )
 
@@ -216,51 +214,3 @@ func getUpdatedPropsAndUpdateSummary(currentProps iceberg.Properties, removals [
 
 	return updatedProps, summary, nil
 }
-
-//lint:ignore U1000 this is linked to by catalogs via go:linkname but we don't want to export it
-func updateAndStageTable(ctx context.Context, current *table.Table, ident table.Identifier, reqs []table.Requirement, updates []table.Update, cat table.CatalogIO) (*table.StagedTable, error) {
-	var (
-		baseMeta    table.Metadata
-		metadataLoc string
-	)
-
-	if current != nil {
-		for _, r := range reqs {
-			if err := r.Validate(current.Metadata()); err != nil {
-				return nil, err
-			}
-		}
-
-		baseMeta = current.Metadata()
-		metadataLoc = current.MetadataLocation()
-	} else {
-		var err error
-		baseMeta, err = table.NewMetadata(iceberg.NewSchema(0), nil, table.UnsortedSortOrder, "", nil)
-		if err != nil {
-			return nil, err
-		}
-	}
-
-	updated, err := internal.UpdateTableMetadata(baseMeta, updates, metadataLoc)
-	if err != nil {
-		return nil, err
-	}
-
-	provider, err := table.LoadLocationProvider(updated.Location(), updated.Properties())
-	if err != nil {
-		return nil, err
-	}
-
-	newVersion := internal.ParseMetadataVersion(metadataLoc) + 1
-	newLocation, err := provider.NewTableMetadataFileLocation(newVersion)
-	if err != nil {
-		return nil, err
-	}
-
-	fs, err := io.LoadFS(ctx, updated.Properties(), newLocation)
-	if err != nil {
-		return nil, err
-	}
-
-	return &table.StagedTable{Table: table.New(ident, updated, newLocation, fs, cat)}, nil
-}
diff --git a/catalog/glue/glue.go b/catalog/glue/glue.go
index 75b19eb..ab15c2d 100644
--- a/catalog/glue/glue.go
+++ b/catalog/glue/glue.go
@@ -65,6 +65,10 @@ const (
 	Endpoint        = "glue.endpoint"
 	MaxRetries      = "glue.max-retries"
 	RetryMode       = "glue.retry-mode"
+
+	icebergFieldIDKey       = "iceberg.field.id"
+	icebergFieldOptionalKey = "iceberg.field.optional"
+	icebergFieldCurrentKey  = "iceberg.field.current"
 )
 
 var _ catalog.Catalog = (*Catalog)(nil)
@@ -130,6 +134,7 @@ type Catalog struct {
 	glueSvc   glueAPI
 	catalogId *string
 	awsCfg    *aws.Config
+	props     iceberg.Properties
 }
 
 // NewCatalog creates a new instance of glue.Catalog with the given options.
@@ -151,6 +156,7 @@ func NewCatalog(opts ...Option) *Catalog {
 		glueSvc:   glue.NewFromConfig(glueOps.awsConfig),
 		catalogId: catalogId,
 		awsCfg:    &glueOps.awsConfig,
+		props:     iceberg.Properties(glueOps.awsProperties),
 	}
 }
 
@@ -232,46 +238,55 @@ func (c *Catalog) CatalogType() catalog.Type {
 }
 
 // CreateTable creates a new Iceberg table in the Glue catalog.
-// AWS Glue will create a new table and a new metadata file in S3 with the format: metadataLocation/metadata/00000-00000-00000-00000-00000.metadata.json.
+// This function will create the metadata file in S3 using the catalog and table properties,
+// to determine the bucket and key for the metadata location.
 func (c *Catalog) CreateTable(ctx context.Context, identifier table.Identifier, schema *iceberg.Schema, opts ...catalog.CreateTableOpt) (*table.Table, error) {
+	staged, err := internal.CreateStagedTable(ctx, c.props, c.LoadNamespaceProperties, identifier, schema, opts...)
+	if err != nil {
+		return nil, err
+	}
+
 	database, tableName, err := identifierToGlueTable(identifier)
 	if err != nil {
 		return nil, err
 	}
-	var cfg catalog.CreateTableCfg
-	for _, opt := range opts {
-		opt(&cfg)
+
+	wfs, ok := staged.FS().(io.WriteFileIO)
+	if !ok {
+		return nil, errors.New("loaded filesystem IO does not support writing")
 	}
-	if cfg.Location == "" {
-		return nil, errors.New("metadata location is required for table creation")
+
+	if err := internal.WriteTableMetadata(staged.Metadata(), wfs, staged.MetadataLocation()); err != nil {
+		return nil, err
 	}
-	parameters := map[string]string{}
-	for k, v := range cfg.Properties {
-		parameters[k] = v
+
+	var tableDescription *string
+	if desc := staged.Properties().Get("Description", ""); desc != "" {
+		tableDescription = aws.String(desc)
 	}
+
 	tableInput := &types.TableInput{
-		Name:       aws.String(tableName),
-		Parameters: parameters,
-		TableType:  aws.String("EXTERNAL_TABLE"),
+		Name: aws.String(tableName),
+		Parameters: map[string]string{
+			tableTypePropsKey:        glueTypeIceberg,
+			metadataLocationPropsKey: staged.MetadataLocation(),
+		},
+		TableType: aws.String("EXTERNAL_TABLE"),
 		StorageDescriptor: &types.StorageDescriptor{
-			Location: aws.String(cfg.Location),
-			Columns:  schemaToGlueColumns(schema),
+			Location: aws.String(staged.Metadata().Location()),
+			Columns:  schemaToGlueColumns(schema, true),
 		},
+		Description: tableDescription,
 	}
 	_, 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 create table %s.%s: %w", database, tableName, err)
 	}
-	createdTable, err := c.LoadTable(ctx, identifier, cfg.Properties)
+	createdTable, err := c.LoadTable(ctx, identifier, nil)
 	if err != nil {
 		// Attempt to clean up the table if loading fails
 		_, cleanupErr := c.glueSvc.DeleteTable(ctx, &glue.DeleteTableInput{
@@ -313,7 +328,7 @@ func (c *Catalog) RegisterTable(ctx context.Context, identifier table.Identifier
 		TableType:  aws.String("EXTERNAL_TABLE"),
 		StorageDescriptor: &types.StorageDescriptor{
 			Location: aws.String(metadataLocation),
-			Columns:  schemaToGlueColumns(metadata.Schema()),
+			Columns:  schemaToGlueColumns(metadata.Schema(), true),
 		},
 	}
 	_, err = c.glueSvc.CreateTable(ctx, &glue.CreateTableInput{
@@ -333,9 +348,6 @@ func (c *Catalog) RegisterTable(ctx context.Context, identifier table.Identifier
 	return c.LoadTable(ctx, identifier, nil)
 }
 
-//go:linkname updateAndStageTable github.com/apache/iceberg-go/catalog.updateAndStageTable
-func updateAndStageTable(ctx context.Context, current *table.Table, ident table.Identifier, reqs []table.Requirement, updates []table.Update, cat table.CatalogIO) (*table.StagedTable, error)
-
 func (c *Catalog) CommitTable(ctx context.Context, tbl *table.Table, requirements []table.Requirement, updates []table.Update) (table.Metadata, string, error) {
 	// Load current table
 	database, tableName, err := identifierToGlueTable(tbl.Identifier())
@@ -348,7 +360,7 @@ func (c *Catalog) CommitTable(ctx context.Context, tbl *table.Table, requirement
 	}
 
 	// Create a staging table with the updates applied
-	staged, err := updateAndStageTable(ctx, tbl, tbl.Identifier(), requirements, updates, c)
+	staged, err := internal.UpdateAndStageTable(ctx, tbl, tbl.Identifier(), requirements, updates, c)
 	if err != nil {
 		return nil, "", err
 	}
@@ -764,7 +776,7 @@ func buildGlueTableInput(ctx context.Context, database string, tableName string,
 		existingColumnMap[*column.Name] = *column.Comment
 	}
 	var glueColumns []types.Column
-	for _, column := range schemaToGlueColumns(staged.Metadata().CurrentSchema()) {
+	for _, column := range schemaToGlueColumns(staged.Metadata().CurrentSchema(), true) {
 		col := types.Column{
 			Name:    column.Name,
 			Comment: column.Comment,
diff --git a/catalog/glue/schema.go b/catalog/glue/schema.go
index 27b2f6d..167b41d 100644
--- a/catalog/glue/schema.go
+++ b/catalog/glue/schema.go
@@ -19,6 +19,7 @@ package glue
 
 import (
 	"fmt"
+	"strconv"
 	"strings"
 
 	"github.com/apache/iceberg-go"
@@ -27,22 +28,27 @@ import (
 )
 
 // schemaToGlueColumns converts an Iceberg schema to a list of Glue columns.
-func schemaToGlueColumns(schema *iceberg.Schema) []types.Column {
+func schemaToGlueColumns(schema *iceberg.Schema, isCurrent bool) []types.Column {
 	var columns []types.Column
 	for _, field := range schema.Fields() {
-		columns = append(columns, fieldToGlueColumn(field))
+		columns = append(columns, fieldToGlueColumn(field, isCurrent))
 	}
 
 	return columns
 }
 
 // fieldToGlueColumn converts an Iceberg nested field to a Glue column.
-func fieldToGlueColumn(field iceberg.NestedField) types.Column {
+func fieldToGlueColumn(field iceberg.NestedField, isCurrent bool) types.Column {
 	column := types.Column{
 		Name:    aws.String(field.Name),
 		Comment: aws.String(field.Doc),
+		Type:    aws.String(icebergTypeToGlueType(field.Type)),
+		Parameters: map[string]string{
+			icebergFieldIDKey:       strconv.Itoa(field.ID),
+			icebergFieldOptionalKey: strconv.FormatBool(!field.Required),
+			icebergFieldCurrentKey:  strconv.FormatBool(isCurrent),
+		},
 	}
-	column.Type = aws.String(icebergTypeToGlueType(field.Type))
 
 	return column
 }
diff --git a/catalog/internal/utils.go b/catalog/internal/utils.go
index 204f7ea..7e4c8ca 100644
--- a/catalog/internal/utils.go
+++ b/catalog/internal/utils.go
@@ -22,11 +22,15 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"maps"
+	"net/url"
 	"path"
 	"regexp"
 	"strconv"
+	"strings"
 
 	"github.com/apache/iceberg-go"
+	"github.com/apache/iceberg-go/catalog"
 	"github.com/apache/iceberg-go/io"
 	"github.com/apache/iceberg-go/table"
 	"github.com/google/uuid"
@@ -37,6 +41,16 @@ func GetMetadataLoc(location string, newVersion uint) string {
 		location, newVersion, uuid.New().String())
 }
 
+func WriteTableMetadata(metadata table.Metadata, fs io.WriteFileIO, loc string) error {
+	out, err := fs.Create(loc)
+	if err != nil {
+		return nil
+	}
+	defer out.Close()
+
+	return json.NewEncoder(out).Encode(metadata)
+}
+
 func WriteMetadata(ctx context.Context, metadata table.Metadata, loc string, props iceberg.Properties) error {
 	fs, err := io.LoadFS(ctx, props, loc)
 	if err != nil {
@@ -90,6 +104,75 @@ func UpdateTableMetadata(base table.Metadata, updates []table.Update, metadataLo
 	return bldr.Build()
 }
 
+func CreateStagedTable(ctx context.Context, catprops iceberg.Properties, nspropsFn GetNamespacePropsFn, ident table.Identifier, sc *iceberg.Schema, opts ...catalog.CreateTableOpt) (table.StagedTable, error) {
+	var cfg catalog.CreateTableCfg
+	for _, opt := range opts {
+		opt(&cfg)
+	}
+
+	dbIdent := catalog.NamespaceFromIdent(ident)
+	tblname := catalog.TableNameFromIdent(ident)
+	dbname := strings.Join(dbIdent, ".")
+
+	loc, err := ResolveTableLocation(ctx, cfg.Location, dbname, tblname, catprops, nspropsFn)
+	if err != nil {
+		return table.StagedTable{}, err
+	}
+
+	provider, err := table.LoadLocationProvider(loc, cfg.Properties)
+	if err != nil {
+		return table.StagedTable{}, err
+	}
+
+	metadataLoc, err := provider.NewTableMetadataFileLocation(0)
+	if err != nil {
+		return table.StagedTable{}, err
+	}
+
+	metadata, err := table.NewMetadata(sc, cfg.PartitionSpec, cfg.SortOrder, loc, cfg.Properties)
+	if err != nil {
+		return table.StagedTable{}, err
+	}
+
+	ioProps := maps.Clone(catprops)
+	maps.Copy(ioProps, cfg.Properties)
+	fs, err := io.LoadFS(ctx, ioProps, metadataLoc)
+	if err != nil {
+		return table.StagedTable{}, err
+	}
+
+	return table.StagedTable{
+		Table: table.New(ident, metadata, metadataLoc, fs, nil),
+	}, nil
+}
+
+type GetNamespacePropsFn func(context.Context, table.Identifier) (iceberg.Properties, error)
+
+func ResolveTableLocation(ctx context.Context, loc, dbname, tablename string, catprops iceberg.Properties, nsprops GetNamespacePropsFn) (string, error) {
+	if len(loc) == 0 {
+		dbprops, err := nsprops(ctx, strings.Split(dbname, "."))
+		if err != nil {
+			return "", err
+		}
+
+		return getDefaultWarehouseLocation(dbname, tablename, dbprops, catprops)
+	}
+
+	return strings.TrimSuffix(loc, "/"), nil
+}
+
+func getDefaultWarehouseLocation(dbname, tablename string, nsprops, catprops iceberg.Properties) (string, error) {
+	if dblocation := nsprops.Get("location", ""); dblocation != "" {
+		return url.JoinPath(dblocation, tablename)
+	}
+
+	if warehousepath := catprops.Get("warehouse", ""); warehousepath != "" {
+		return url.JoinPath(warehousepath, dbname+".db", tablename)
+	}
+
+	return "", errors.New("no default path set, please specify a location when creating a table")
+}
+
 // (\d+)            -> version number
 // -                -> separator
 // ([\w-]{36})      -> UUID (36 characters, including hyphens)
@@ -116,3 +199,50 @@ func ParseMetadataVersion(location string) int {
 
 	return v
 }
+
+func UpdateAndStageTable(ctx context.Context, current *table.Table, ident table.Identifier, reqs []table.Requirement, updates []table.Update, cat table.CatalogIO) (*table.StagedTable, error) {
+	var (
+		baseMeta    table.Metadata
+		metadataLoc string
+	)
+
+	if current != nil {
+		for _, r := range reqs {
+			if err := r.Validate(current.Metadata()); err != nil {
+				return nil, err
+			}
+		}
+
+		baseMeta = current.Metadata()
+		metadataLoc = current.MetadataLocation()
+	} else {
+		var err error
+		baseMeta, err = table.NewMetadata(iceberg.NewSchema(0), nil, table.UnsortedSortOrder, "", nil)
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	updated, err := UpdateTableMetadata(baseMeta, updates, metadataLoc)
+	if err != nil {
+		return nil, err
+	}
+
+	provider, err := table.LoadLocationProvider(updated.Location(), updated.Properties())
+	if err != nil {
+		return nil, err
+	}
+
+	newVersion := ParseMetadataVersion(metadataLoc) + 1
+	newLocation, err := provider.NewTableMetadataFileLocation(newVersion)
+	if err != nil {
+		return nil, err
+	}
+
+	fs, err := io.LoadFS(ctx, updated.Properties(), newLocation)
+	if err != nil {
+		return nil, err
+	}
+
+	return &table.StagedTable{Table: table.New(ident, updated, newLocation, fs, cat)}, nil
+}
diff --git a/catalog/sql/sql.go b/catalog/sql/sql.go
index 9373f27..bb8a083 100644
--- a/catalog/sql/sql.go
+++ b/catalog/sql/sql.go
@@ -24,7 +24,6 @@ import (
 	"fmt"
 	"iter"
 	"maps"
-	"net/url"
 	"slices"
 	"strings"
 	"sync"
@@ -258,31 +257,6 @@ func (c *Catalog) namespaceExists(ctx context.Context, ns string) (bool, error)
 	})
 }
 
-func (c *Catalog) getDefaultWarehouseLocation(ctx context.Context, nsname, tableName string) (string, error) {
-	dbprops, err := c.LoadNamespaceProperties(ctx, strings.Split(nsname, "."))
-	if err != nil {
-		return "", err
-	}
-
-	if dblocation := dbprops.Get("location", ""); dblocation != "" {
-		return url.JoinPath(dblocation, tableName)
-	}
-
-	if warehousepath := c.props.Get("warehouse", ""); warehousepath != "" {
-		return url.JoinPath(warehousepath, nsname+".db", tableName)
-	}
-
-	return "", errors.New("no default path set, please specify a location when creating a table")
-}
-
-func (c *Catalog) resolveTableLocation(ctx context.Context, loc, nsname, tablename string) (string, error) {
-	if len(loc) == 0 {
-		return c.getDefaultWarehouseLocation(ctx, nsname, tablename)
-	}
-
-	return strings.TrimSuffix(loc, "/"), nil
-}
-
 func checkValidNamespace(ident table.Identifier) error {
 	if len(ident) < 1 {
 		return fmt.Errorf("%w: empty namespace identifier", catalog.ErrNoSuchNamespace)
@@ -292,9 +266,9 @@ func checkValidNamespace(ident table.Identifier) error {
 }
 
 func (c *Catalog) CreateTable(ctx context.Context, ident table.Identifier, sc *iceberg.Schema, opts ...catalog.CreateTableOpt) (*table.Table, error) {
-	var cfg catalog.CreateTableCfg
-	for _, opt := range opts {
-		opt(&cfg)
+	staged, err := internal.CreateStagedTable(ctx, c.props, c.LoadNamespaceProperties, ident, sc, opts...)
+	if err != nil {
+		return nil, err
 	}
 
 	nsIdent := catalog.NamespaceFromIdent(ident)
@@ -309,18 +283,12 @@ func (c *Catalog) CreateTable(ctx context.Context, ident table.Identifier, sc *i
 		return nil, fmt.Errorf("%w: %s", catalog.ErrNoSuchNamespace, ns)
 	}
 
-	loc, err := c.resolveTableLocation(ctx, cfg.Location, ns, tblIdent)
-	if err != nil {
-		return nil, err
-	}
-
-	metadataLocation := internal.GetMetadataLoc(loc, 0)
-	metadata, err := table.NewMetadata(sc, cfg.PartitionSpec, cfg.SortOrder, loc, cfg.Properties)
-	if err != nil {
-		return nil, err
+	wfs, ok := staged.FS().(io.WriteFileIO)
+	if !ok {
+		return nil, errors.New("loaded filesystem IO does not support writing")
 	}
 
-	if err := internal.WriteMetadata(ctx, metadata, metadataLocation, c.props); err != nil {
+	if err := internal.WriteTableMetadata(staged.Metadata(), wfs, staged.MetadataLocation()); err != nil {
 		return nil, err
 	}
 
@@ -329,7 +297,7 @@ func (c *Catalog) CreateTable(ctx context.Context, ident table.Identifier, sc *i
 			CatalogName:      c.name,
 			TableNamespace:   ns,
 			TableName:        tblIdent,
-			MetadataLocation: sql.NullString{String: metadataLocation, Valid: true},
+			MetadataLocation: sql.NullString{String: staged.MetadataLocation(), Valid: true},
 		}).Exec(ctx)
 		if err != nil {
 			return fmt.Errorf("failed to create table: %w", err)
@@ -341,12 +309,9 @@ func (c *Catalog) CreateTable(ctx context.Context, ident table.Identifier, sc *i
 		return nil, err
 	}
 
-	return c.LoadTable(ctx, ident, cfg.Properties)
+	return c.LoadTable(ctx, ident, staged.Properties())
 }
 
-//go:linkname updateAndStageTable github.com/apache/iceberg-go/catalog.updateAndStageTable
-func updateAndStageTable(ctx context.Context, current *table.Table, ident table.Identifier, reqs []table.Requirement, updates []table.Update, cat table.CatalogIO) (*table.StagedTable, error)
-
 func (c *Catalog) CommitTable(ctx context.Context, tbl *table.Table, reqs []table.Requirement, updates []table.Update) (table.Metadata, string, error) {
 	ns := catalog.NamespaceFromIdent(tbl.Identifier())
 	tblName := catalog.TableNameFromIdent(tbl.Identifier())
@@ -356,7 +321,7 @@ func (c *Catalog) CommitTable(ctx context.Context, tbl *table.Table, reqs []tabl
 		return nil, "", err
 	}
 
-	staged, err := updateAndStageTable(ctx, current, tbl.Identifier(), reqs, updates, c)
+	staged, err := internal.UpdateAndStageTable(ctx, current, tbl.Identifier(), reqs, updates, c)
 	if err != nil {
 		return nil, "", err
 	}
diff --git a/manifest.go b/manifest.go
index e09e892..e741734 100644
--- a/manifest.go
+++ b/manifest.go
@@ -450,62 +450,7 @@ func fetchManifestEntries(m ManifestFile, fs iceio.IO, discardDeleted bool) ([]M
 	}
 	defer f.Close()
 
-	return readManifestEntries(m, f, discardDeleted)
-}
-
-func readManifestEntries(m ManifestFile, f io.Reader, discardDeleted bool) ([]ManifestEntry, error) {
-	dec, err := ocf.NewDecoder(f, ocf.WithDecoderSchemaCache(&avro.SchemaCache{}))
-	if err != nil {
-		return nil, err
-	}
-
-	metadata := dec.Metadata()
-	sc := dec.Schema()
-
-	fieldNameToID, fieldIDToLogicalType := getFieldIDMap(sc)
-	isFallback := false
-	if string(metadata["format-version"]) == "1" {
-		for _, f := range sc.(*avro.RecordSchema).Fields() {
-			if f.Name() == "snapshot_id" {
-				if f.Type().Type() != avro.Union {
-					isFallback = true
-				}
-
-				break
-			}
-		}
-	}
-
-	results := make([]ManifestEntry, 0)
-	for dec.HasNext() {
-		var tmp ManifestEntry
-		if isFallback {
-			tmp = &fallbackManifestEntry{
-				manifestEntry: manifestEntry{Data: &dataFile{}},
-			}
-		} else {
-			tmp = &manifestEntry{Data: &dataFile{}}
-		}
-
-		if err := dec.Decode(tmp); err != nil {
-			return nil, err
-		}
-
-		if isFallback {
-			tmp = tmp.(*fallbackManifestEntry).toEntry()
-		}
-
-		if !discardDeleted || tmp.Status() != EntryStatusDELETED {
-			tmp.inherit(m)
-			if fieldToIDMap, ok := tmp.DataFile().(hasFieldToIDMap); ok {
-				fieldToIDMap.setFieldNameToIDMap(fieldNameToID)
-				fieldToIDMap.setFieldIDToLogicalTypeMap(fieldIDToLogicalType)
-			}
-			results = append(results, tmp)
-		}
-	}
-
-	return results, dec.Error()
+	return ReadManifest(m, f, discardDeleted)
 }
 
 // ManifestFile is the interface which covers both V1 and V2 manifest files.
@@ -614,6 +559,213 @@ func decodeManifests[I interface {
 	return results, dec.Error()
 }
 
+// ManifestReader reads the metadata and data from an avro manifest file.
+// This type is not thread-safe; its methods should not be called from
+// multiple goroutines.
+type ManifestReader struct {
+	dec           *ocf.Decoder
+	file          ManifestFile
+	formatVersion int
+	isFallback    bool
+	content       ManifestContent
+	fieldNameToID map[string]int
+	fieldIDToType map[int]avro.LogicalType
+
+	// The rest are lazily populated, on demand. Most readers
+	// will likely only try to load the entries.
+	schema              Schema
+	schemaLoaded        bool
+	partitionSpec       PartitionSpec
+	partitionSpecLoaded bool
+}
+
+// NewManifestReader returns a value that can read the contents of an avro manifest
+// file. If the caller is interested in the manifest entries in the file, it must call
+// [ManifestReader.Entries] before closing the provided reader.
+func NewManifestReader(file ManifestFile, in io.Reader) (*ManifestReader, error) {
+	dec, err := ocf.NewDecoder(in, ocf.WithDecoderSchemaCache(&avro.SchemaCache{}))
+	if err != nil {
+		return nil, err
+	}
+
+	metadata := dec.Metadata()
+	sc := dec.Schema()
+
+	formatVersion, err := strconv.Atoi(string(metadata["format-version"]))
+	if err != nil {
+		return nil, fmt.Errorf("manifest file's 'format-version' metadata is invalid: %w", err)
+	}
+	if formatVersion != file.Version() {
+		return nil, fmt.Errorf("manifest file's 'format-version' metadata indicates version %d, but entry from manifest list indicates version %d",
+			formatVersion, file.Version())
+	}
+
+	var content ManifestContent
+	switch contentStr := string(metadata["content"]); contentStr {
+	case "data":
+		content = ManifestContentData
+	case "deletes":
+		content = ManifestContentDeletes
+	default:
+		return nil, fmt.Errorf("manifest file's 'content' metadata is invalid, should be \"data\" or \"deletes\" but instead is %q",
+			contentStr)
+	}
+	if content != file.ManifestContent() {
+		return nil, fmt.Errorf("manifest file's 'content' metadata indicates %q, but entry from manifest list indicates %q",
+			content.String(), file.ManifestContent().String())
+	}
+
+	isFallback := false
+	if formatVersion == 1 {
+		for _, f := range sc.(*avro.RecordSchema).Fields() {
+			if f.Name() == "snapshot_id" {
+				if f.Type().Type() != avro.Union {
+					isFallback = true
+				}
+
+				break
+			}
+		}
+	}
+	fieldNameToID, fieldIDToType := getFieldIDMap(sc)
+
+	return &ManifestReader{
+		dec:           dec,
+		file:          file,
+		formatVersion: formatVersion,
+		isFallback:    isFallback,
+		content:       content,
+		fieldNameToID: fieldNameToID,
+		fieldIDToType: fieldIDToType,
+	}, nil
+}
+
+// Version returns the file's format version.
+func (c *ManifestReader) Version() int {
+	return c.formatVersion
+}
+
+// ManifestContent returns the type of content in the manifest file.
+func (c *ManifestReader) ManifestContent() ManifestContent {
+	return c.content
+}
+
+// SchemaID returns the schema ID encoded in the avro file's metadata.
+func (c *ManifestReader) SchemaID() (int, error) {
+	id, err := strconv.Atoi(string(c.dec.Metadata()["schema-id"]))
+	if err != nil {
+		return 0, fmt.Errorf("manifest file's 'schema-id' metadata is invalid: %w", err)
+	}
+
+	return id, nil
+}
+
+// Schema returns the schema encoded in the avro file's metadata.
+func (c *ManifestReader) Schema() (*Schema, error) {
+	if !c.schemaLoaded {
+		schemaID, err := c.SchemaID()
+		if err != nil {
+			return nil, err
+		}
+		if err := json.Unmarshal(c.dec.Metadata()["schema"], &c.schema); err != nil {
+			return nil, fmt.Errorf("manifest file's 'schema' metadata is invalid: %w", err)
+		}
+		c.schema.ID = schemaID
+		c.schemaLoaded = true
+	}
+
+	return &c.schema, nil
+}
+
+// PartitionSpecID returns the partition spec ID encoded in the avro file's metadata.
+func (c *ManifestReader) PartitionSpecID() (int, error) {
+	id, err := strconv.Atoi(string(c.dec.Metadata()["partition-spec-id"]))
+	if err != nil {
+		return 0, fmt.Errorf("manifest file's 'partition-spec-id' metadata is invalid: %w", err)
+	}
+	if id != int(c.file.PartitionSpecID()) {
+		return 0, fmt.Errorf("manifest file's 'partition-spec-id' metadata indicates %d, but entry from manifest list indicates %d",
+			id, c.file.PartitionSpecID())
+	}
+
+	return id, nil
+}
+
+// PartitionSpec returns the partition spec encoded in the avro file's metadata.
+func (c *ManifestReader) PartitionSpec() (*PartitionSpec, error) {
+	if !c.partitionSpecLoaded {
+		partitionSpecID, err := c.PartitionSpecID()
+		if err != nil {
+			return nil, err
+		}
+		if err := json.Unmarshal(c.dec.Metadata()["partition-spec"], &c.partitionSpec.fields); err != nil {
+			return nil, fmt.Errorf("manifest file's 'partition-spec' metadata is invalid: %w", err)
+		}
+		c.partitionSpec.id = partitionSpecID
+		c.partitionSpec.initialize()
+		c.partitionSpecLoaded = true
+	}
+
+	return &c.partitionSpec, nil
+}
+
+// ReadEntry reads the next manifest entry in the avro file's data.
+func (c *ManifestReader) ReadEntry() (ManifestEntry, error) {
+	if err := c.dec.Error(); err != nil {
+		return nil, err
+	}
+	if !c.dec.HasNext() {
+		return nil, io.EOF
+	}
+	var tmp ManifestEntry
+	if c.isFallback {
+		tmp = &fallbackManifestEntry{
+			manifestEntry: manifestEntry{Data: &dataFile{}},
+		}
+	} else {
+		tmp = &manifestEntry{Data: &dataFile{}}
+	}
+
+	if err := c.dec.Decode(tmp); err != nil {
+		return nil, err
+	}
+	if c.isFallback {
+		tmp = tmp.(*fallbackManifestEntry).toEntry()
+	}
+	tmp.inherit(c.file)
+	if fieldToIDMap, ok := tmp.DataFile().(hasFieldToIDMap); ok {
+		fieldToIDMap.setFieldNameToIDMap(c.fieldNameToID)
+		fieldToIDMap.setFieldIDToLogicalTypeMap(c.fieldIDToType)
+	}
+
+	return tmp, nil
+}
+
+// ReadManifest reads in an avro list file and returns a slice
+// of manifest entries or an error if one is encountered. If discardDeleted
+// is true, the returned slice omits entries whose status is "deleted".
+func ReadManifest(m ManifestFile, f io.Reader, discardDeleted bool) ([]ManifestEntry, error) {
+	manifestReader, err := NewManifestReader(m, f)
+	if err != nil {
+		return nil, err
+	}
+	var results []ManifestEntry
+	for {
+		entry, err := manifestReader.ReadEntry()
+		if err != nil {
+			if errors.Is(err, io.EOF) {
+				return results, nil
+			}
+
+			return results, err
+		}
+		if discardDeleted && entry.Status() == EntryStatusDELETED {
+			continue
+		}
+		results = append(results, entry)
+	}
+}
+
 // ReadManifestList reads in an avro manifest list file and returns a slice
 // of manifest files or an error if one is encountered.
 func ReadManifestList(in io.Reader) ([]ManifestFile, error) {


Test patch:
diff --git a/catalog/glue/schema_test.go b/catalog/glue/schema_test.go
index ff85c86..13825d5 100644
--- a/catalog/glue/schema_test.go
+++ b/catalog/glue/schema_test.go
@@ -130,6 +130,11 @@ func TestFieldToGlueColumn(t *testing.T) {
 				Name:    aws.String("simple_field"),
 				Type:    aws.String("string"),
 				Comment: aws.String("A simple string field"),
+				Parameters: map[string]string{
+					icebergFieldIDKey:       "1",
+					icebergFieldOptionalKey: "false",
+					icebergFieldCurrentKey:  "true",
+				},
 			},
 		},
 		{
@@ -138,22 +143,28 @@ func TestFieldToGlueColumn(t *testing.T) {
 				ID:       2,
 				Name:     "price",
 				Type:     iceberg.DecimalTypeOf(10, 2),
-				Required: true,
+				Required: false,
 				Doc:      "Price with 2 decimal places",
 			},
 			expected: types.Column{
 				Name:    aws.String("price"),
 				Type:    aws.String("decimal(10,2)"),
 				Comment: aws.String("Price with 2 decimal places"),
+				Parameters: map[string]string{
+					icebergFieldIDKey:       "2",
+					icebergFieldOptionalKey: "true",
+					icebergFieldCurrentKey:  "true",
+				},
 			},
 		},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			result := fieldToGlueColumn(tt.field)
-			assert.Equal(t, aws.ToString(tt.expected.Name), aws.ToString(result.Name))
-			assert.Equal(t, aws.ToString(tt.expected.Type), aws.ToString(result.Type))
-			assert.Equal(t, aws.ToString(tt.expected.Comment), aws.ToString(result.Comment))
+			result := fieldToGlueColumn(tt.field, true)
+			assert.Equal(t, tt.expected.Name, result.Name)
+			assert.Equal(t, tt.expected.Type, result.Type)
+			assert.Equal(t, tt.expected.Comment, result.Comment)
+			assert.Equal(t, tt.expected.Parameters, result.Parameters)
 		})
 	}
 }
@@ -191,7 +202,7 @@ func TestNestedStructType(t *testing.T) {
 		Required: true,
 		Doc:      "User address",
 	}
-	column := fieldToGlueColumn(field)
+	column := fieldToGlueColumn(field, true)
 	assert.Equal(t, "address", aws.ToString(column.Name))
 	assert.Equal(t, expected, aws.ToString(column.Type))
 	assert.Equal(t, "User address", aws.ToString(column.Comment))
@@ -212,7 +223,7 @@ func TestNestedListType(t *testing.T) {
 		Required: false,
 		Doc:      "User tags",
 	}
-	column := fieldToGlueColumn(field)
+	column := fieldToGlueColumn(field, false)
 	assert.Equal(t, "tags", aws.ToString(column.Name))
 	assert.Equal(t, "array<string>", aws.ToString(column.Type))
 	assert.Equal(t, "User tags", aws.ToString(column.Comment))
@@ -235,7 +246,7 @@ func TestNestedMapType(t *testing.T) {
 		Required: true,
 		Doc:      "User properties",
 	}
-	column := fieldToGlueColumn(field)
+	column := fieldToGlueColumn(field, true)
 	assert.Equal(t, "properties", aws.ToString(column.Name))
 	assert.Equal(t, "map<string,int>", aws.ToString(column.Type))
 	assert.Equal(t, "User properties", aws.ToString(column.Comment))
@@ -281,7 +292,7 @@ func TestComplexNestedTypes(t *testing.T) {
 		Required: true,
 		Doc:      "A complex nested field",
 	}
-	column := fieldToGlueColumn(field)
+	column := fieldToGlueColumn(field, true)
 	assert.Equal(t, "complex_field", aws.ToString(column.Name))
 	assert.Equal(t, expected, aws.ToString(column.Type))
 	assert.Equal(t, "A complex nested field", aws.ToString(column.Comment))
@@ -338,7 +349,7 @@ func TestSchemaToGlueColumns(t *testing.T) {
 		},
 	}
 	schema := iceberg.NewSchema(1, fields...)
-	columns := schemaToGlueColumns(schema)
+	columns := schemaToGlueColumns(schema, true)
 	assert.Equal(t, 4, len(columns))
 	assert.Equal(t, "id", aws.ToString(columns[0].Name))
 	assert.Equal(t, "bigint", aws.ToString(columns[0].Type))
diff --git a/manifest_test.go b/manifest_test.go
index aeda897..66bc07f 100644
--- a/manifest_test.go
+++ b/manifest_test.go
@@ -832,7 +832,7 @@ func (m *ManifestTestSuite) TestReadManifestIncompleteSchema() {
 	)
 	m.NoError(err)
 
-	entries, err := readManifestEntries(file, &buf, false)
+	entries, err := ReadManifest(file, &buf, false)
 	m.NoError(err)
 	m.Len(entries, 1)
 
@@ -907,26 +907,43 @@ func (m *ManifestTestSuite) TestReadManifestIncompleteSchema() {
 	}
 
 	// This should fail because the file's schema is incomplete.
-	_, err = readManifestEntries(file, &buf, false)
+	_, err = ReadManifest(file, &buf, false)
 	m.ErrorContains(err, "unknown type: r2")
 }
 
 func (m *ManifestTestSuite) TestManifestEntriesV2() {
-	var mockfs internal.MockFS
 	manifest := manifestFile{
 		version: 2,
+		SpecID:  1,
 		Path:    manifestFileRecordsV2[0].FilePath(),
 	}
 
-	mockfs.Test(m.T())
-	mockfs.On("Open", manifest.FilePath()).Return(&internal.MockFile{
+	partitionSpec := NewPartitionSpecID(1,
+		PartitionField{FieldID: 1000, SourceID: 1, Name: "VendorID", Transform: IdentityTransform{}},
+		PartitionField{FieldID: 1001, SourceID: 2, Name: "tpep_pickup_datetime", Transform: IdentityTransform{}})
+
+	mockedFile := &internal.MockFile{
 		Contents: bytes.NewReader(m.v2ManifestEntries.Bytes()),
-	}, nil)
-	defer mockfs.AssertExpectations(m.T())
-	entries, err := manifest.FetchEntries(&mockfs, false)
+	}
+	manifestReader, err := NewManifestReader(&manifest, mockedFile)
 	m.Require().NoError(err)
-	m.Len(entries, 2)
-	m.Zero(manifest.PartitionSpecID())
+	m.Equal(2, manifestReader.Version())
+	m.Equal(ManifestContentData, manifestReader.ManifestContent())
+	loadedSchema, err := manifestReader.Schema()
+	m.Require().NoError(err)
+	m.True(loadedSchema.Equals(testSchema))
+	loadedPartitionSpec, err := manifestReader.PartitionSpec()
+	m.Require().NoError(err)
+	m.True(loadedPartitionSpec.Equals(partitionSpec))
+
+	entry1, err := manifestReader.ReadEntry()
+	m.Require().NoError(err)
+	_, err = manifestReader.ReadEntry()
+	m.Require().NoError(err)
+	_, err = manifestReader.ReadEntry()
+	m.Require().ErrorIs(err, io.EOF)
+
+	m.Equal(int32(1), manifest.PartitionSpecID())
 	m.Zero(manifest.SnapshotID())
 	m.Zero(manifest.AddedDataFiles())
 	m.Zero(manifest.ExistingDataFiles())
@@ -935,8 +952,6 @@ func (m *ManifestTestSuite) TestManifestEntriesV2() {
 	m.Zero(manifest.DeletedRows())
 	m.Zero(manifest.AddedRows())
 
-	entry1 := entries[0]
-
 	m.Equal(EntryStatusADDED, entry1.Status())
 	m.Equal(entrySnapshotID, entry1.SnapshotID())
 	m.Zero(entry1.SequenceNum())