Repository: apache/iceberg-go
Pull request #177 (2024-10-17 04:24:54)

feat: more builders and writing manifests
#172 

This PR adds Builders for v1/2 ManifestEntry and for DataFile as well as helper funcs to write ManifestLists and Manifest files.

Bonus: fixes a max recursion depth panic in `dataFile.EqualityFieldIDs()`

One thing I'm not super happy with is `manifestEntryV1/2.Data`'s type has been changed from `dataFile` to `DataFile` (the interface type). This was the best way I could come up with to make the builders for both work, but I also feel like it diverges from the pattern a little bit. This may not be a huge deal, but something to consider. You can see how the effects rippled out a little bit in the tests. 


nastra: @dwilson1988 can you please rebase/fix the merge conflicts?
dwilson1988: > @dwilson1988 can you please rebase/fix the merge conflicts?

@nastra done!

Patch:
diff --git a/LICENSE b/LICENSE
index 515fd54..d645695 100644
--- a/LICENSE
+++ b/LICENSE
@@ -200,116 +200,3 @@
    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.
-
---------------------------------------------------------------------------------
-
-This product includes a gradle wrapper.
-
-* gradlew and gradle/wrapper/gradle-wrapper.properties
-
-Copyright: 2010-2019 Gradle Authors.
-Home page: https://github.com/gradle/gradle
-License: https://www.apache.org/licenses/LICENSE-2.0
-
---------------------------------------------------------------------------------
-
-This product includes code from Apache Avro.
-
-* Conversion in DecimalWriter is based on Avro's Conversions.DecimalConversion.
-
-Copyright: 2014-2017 The Apache Software Foundation.
-Home page: https://avro.apache.org/
-License: https://www.apache.org/licenses/LICENSE-2.0
-
---------------------------------------------------------------------------------
-
-This product includes code from Apache Parquet.
-
-* DynMethods.java
-* DynConstructors.java
-* AssertHelpers.java
-* IOUtil.java readFully and tests
-* ByteBufferInputStream implementations and tests
-
-Copyright: 2014-2017 The Apache Software Foundation.
-Home page: https://parquet.apache.org/
-License: https://www.apache.org/licenses/LICENSE-2.0
-
---------------------------------------------------------------------------------
-
-This product includes code from Cloudera Kite.
-
-* SchemaVisitor and visit methods
-
-Copyright: 2013-2017 Cloudera Inc.
-Home page: https://kitesdk.org/
-License: https://www.apache.org/licenses/LICENSE-2.0
-
---------------------------------------------------------------------------------
-
-This product includes code from Presto.
-
-* Retry wait and jitter logic in Tasks.java
-* S3FileIO logic derived from PrestoS3FileSystem.java in S3InputStream.java
-  and S3OutputStream.java
-* SQL grammar rules for parsing CALL statements in IcebergSqlExtensions.g4
-* some aspects of handling stored procedures
-
-Copyright: 2016 Facebook and contributors
-Home page: https://prestodb.io/
-License: https://www.apache.org/licenses/LICENSE-2.0
-
---------------------------------------------------------------------------------
-
-This product includes code from Apache iBATIS.
-
-* Hive ScriptRunner.java
-
-Copyright: 2004 Clinton Begin
-Home page: https://ibatis.apache.org/
-License: https://www.apache.org/licenses/LICENSE-2.0
-
---------------------------------------------------------------------------------
-
-This product includes code from Apache Hive.
-
-* Hive metastore derby schema in hive-schema-3.1.0.derby.sql
-
-Copyright: 2011-2018 The Apache Software Foundation
-Home page: https://hive.apache.org/
-License: https://www.apache.org/licenses/LICENSE-2.0
-
---------------------------------------------------------------------------------
-
-This product includes code from Apache Spark.
-
-* dev/check-license script
-* vectorized reading of definition levels in BaseVectorizedParquetValuesReader.java
-* portions of the extensions parser
-* casting logic in AssignmentAlignmentSupport
-* implementation of SetAccumulator.
-* Connector expressions.
-
-Copyright: 2011-2018 The Apache Software Foundation
-Home page: https://spark.apache.org/
-License: https://www.apache.org/licenses/LICENSE-2.0
-
---------------------------------------------------------------------------------
-
-This product includes code from Delta Lake.
-
-* AssignmentAlignmentSupport is an independent development but UpdateExpressionsSupport in Delta was used as a reference.
-
-Copyright: 2020 The Delta Lake Project Authors.
-Home page: https://delta.io/
-License: https://www.apache.org/licenses/LICENSE-2.0
-
---------------------------------------------------------------------------------
-
-This product includes code from Apache Commons.
-
-* Core ArrayUtil.
-
-Copyright: 2020 The Apache Software Foundation
-Home page: https://commons.apache.org/
-License: https://www.apache.org/licenses/LICENSE-2.0
diff --git a/dev/release/README.md b/dev/release/README.md
index 1f9285f..8323d8e 100644
--- a/dev/release/README.md
+++ b/dev/release/README.md
@@ -53,8 +53,7 @@ may be helpful.
 
 Your PGP key must be registered to the following:
 
-    * https://dist.apache.org/repos/dist/dev/iceberg/KEYS
-    * https://dist.apache.org/repos/dist/release/iceberg/KEYS
+    * https://downloads.apache.org/iceberg/KEYS
 
 See the header comment of them for how to add a PGP key.
 
@@ -62,7 +61,7 @@ Apache Iceberg committers can update them by Subversion client with their ASF ac
 e.g.:
 
 ```console
-$ svn co https://dist.apache.org/repos/dist/dev/iceberg
+$ svn co https://dist.apache.org/repos/dist/release/iceberg
 $ cd iceberg
 $ editor KEYS
 $ svn ci KEYS
diff --git a/dev/release/release_rc.sh b/dev/release/release_rc.sh
index 3c93f84..0259521 100755
--- a/dev/release/release_rc.sh
+++ b/dev/release/release_rc.sh
@@ -22,7 +22,7 @@ set -eu
 SOURCE_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
 SOURCE_TOP_DIR="$(cd "${SOURCE_DIR}/../../" && pwd)"
 
-if [ "$#" -ne 1 ]; then
+if [ "$#" -ne 2 ]; then
   echo "Usage: $0 <version> <rc>"
   echo " e.g.: $0 1.0.0 1"
   exit 1
diff --git a/dev/release/verify_rc.sh b/dev/release/verify_rc.sh
index a84af66..6177fe8 100755
--- a/dev/release/verify_rc.sh
+++ b/dev/release/verify_rc.sh
@@ -34,7 +34,7 @@ set -x
 VERSION="$1"
 RC="$2"
 
-ICEBERG_DIST_BASE_URL="https://dist.apache.org/repos/dist/dev/iceberg"
+ICEBERG_DIST_BASE_URL="https://downloads.apache.org/iceberg"
 DOWNLOAD_RC_BASE_URL="https://github.com/apache/iceberg-go/releases/download/v${VERSION}-rc${RC}"
 ARCHIVE_BASE_NAME="apache-iceberg-go-${VERSION}"
 
diff --git a/go.mod b/go.mod
index 66c7d5a..12b40ae 100644
--- a/go.mod
+++ b/go.mod
@@ -27,14 +27,14 @@ require (
 	github.com/aws/aws-sdk-go-v2/config v1.28.1
 	github.com/aws/aws-sdk-go-v2/credentials v1.17.42
 	github.com/aws/aws-sdk-go-v2/service/glue v1.101.0
-	github.com/aws/aws-sdk-go-v2/service/s3 v1.66.2
+	github.com/aws/aws-sdk-go-v2/service/s3 v1.66.3
 	github.com/aws/smithy-go v1.22.0
 	github.com/docopt/docopt-go v0.0.0-20180111231733-ee0de3bc6815
 	github.com/google/uuid v1.6.0
 	github.com/hamba/avro/v2 v2.27.0
 	github.com/pterm/pterm v0.12.79
 	github.com/stretchr/testify v1.9.0
-	github.com/substrait-io/substrait-go v1.1.0
+	github.com/substrait-io/substrait-go v1.2.0
 	github.com/twmb/murmur3 v1.1.8
 	github.com/wolfeidau/s3iofs v1.5.2
 	golang.org/x/sync v0.9.0
@@ -51,14 +51,14 @@ require (
 	github.com/apache/thrift v0.21.0 // indirect
 	github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.6.6 // indirect
 	github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.16.18 // indirect
-	github.com/aws/aws-sdk-go-v2/internal/configsources v1.3.22 // indirect
-	github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.6.22 // indirect
+	github.com/aws/aws-sdk-go-v2/internal/configsources v1.3.23 // indirect
+	github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.6.23 // indirect
 	github.com/aws/aws-sdk-go-v2/internal/ini v1.8.1 // indirect
-	github.com/aws/aws-sdk-go-v2/internal/v4a v1.3.22 // indirect
+	github.com/aws/aws-sdk-go-v2/internal/v4a v1.3.23 // indirect
 	github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.12.0 // indirect
-	github.com/aws/aws-sdk-go-v2/service/internal/checksum v1.4.3 // indirect
-	github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.12.3 // indirect
-	github.com/aws/aws-sdk-go-v2/service/internal/s3shared v1.18.3 // indirect
+	github.com/aws/aws-sdk-go-v2/service/internal/checksum v1.4.4 // indirect
+	github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.12.4 // indirect
+	github.com/aws/aws-sdk-go-v2/service/internal/s3shared v1.18.4 // indirect
 	github.com/aws/aws-sdk-go-v2/service/sso v1.24.3 // indirect
 	github.com/aws/aws-sdk-go-v2/service/ssooidc v1.28.3 // indirect
 	github.com/aws/aws-sdk-go-v2/service/sts v1.32.3 // indirect
@@ -88,6 +88,7 @@ require (
 	github.com/pmezard/go-difflib v1.0.0 // indirect
 	github.com/rivo/uniseg v0.4.4 // indirect
 	github.com/stretchr/objx v0.5.2 // indirect
+	github.com/substrait-io/substrait v0.57.1 // indirect
 	github.com/xo/terminfo v0.0.0-20220910002029-abceb7e1c41e // indirect
 	github.com/zeebo/xxh3 v1.0.2 // indirect
 	golang.org/x/exp v0.0.0-20240909161429-701f63a606c0 // indirect
diff --git a/go.sum b/go.sum
index 6de9569..3ce7b97 100644
--- a/go.sum
+++ b/go.sum
@@ -40,26 +40,26 @@ github.com/aws/aws-sdk-go-v2/credentials v1.17.42 h1:sBP0RPjBU4neGpIYyx8mkU2QqLP
 github.com/aws/aws-sdk-go-v2/credentials v1.17.42/go.mod h1:FwZBfU530dJ26rv9saAbxa9Ej3eF/AK0OAY86k13n4M=
 github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.16.18 h1:68jFVtt3NulEzojFesM/WVarlFpCaXLKaBxDpzkQ9OQ=
 github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.16.18/go.mod h1:Fjnn5jQVIo6VyedMc0/EhPpfNlPl7dHV916O6B+49aE=
-github.com/aws/aws-sdk-go-v2/internal/configsources v1.3.22 h1:Jw50LwEkVjuVzE1NzkhNKkBf9cRN7MtE1F/b2cOKTUM=
-github.com/aws/aws-sdk-go-v2/internal/configsources v1.3.22/go.mod h1:Y/SmAyPcOTmpeVaWSzSKiILfXTVJwrGmYZhcRbhWuEY=
-github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.6.22 h1:981MHwBaRZM7+9QSR6XamDzF/o7ouUGxFzr+nVSIhrs=
-github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.6.22/go.mod h1:1RA1+aBEfn+CAB/Mh0MB6LsdCYCnjZm7tKXtnk499ZQ=
+github.com/aws/aws-sdk-go-v2/internal/configsources v1.3.23 h1:A2w6m6Tmr+BNXjDsr7M90zkWjsu4JXHwrzPg235STs4=
+github.com/aws/aws-sdk-go-v2/internal/configsources v1.3.23/go.mod h1:35EVp9wyeANdujZruvHiQUAo9E3vbhnIO1mTCAxMlY0=
+github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.6.23 h1:pgYW9FCabt2M25MoHYCfMrVY2ghiiBKYWUVXfwZs+sU=
+github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.6.23/go.mod h1:c48kLgzO19wAu3CPkDWC28JbaJ+hfQlsdl7I2+oqIbk=
 github.com/aws/aws-sdk-go-v2/internal/ini v1.8.1 h1:VaRN3TlFdd6KxX1x3ILT5ynH6HvKgqdiXoTxAF4HQcQ=
 github.com/aws/aws-sdk-go-v2/internal/ini v1.8.1/go.mod h1:FbtygfRFze9usAadmnGJNc8KsP346kEe+y2/oyhGAGc=
-github.com/aws/aws-sdk-go-v2/internal/v4a v1.3.22 h1:yV+hCAHZZYJQcwAaszoBNwLbPItHvApxT0kVIw6jRgs=
-github.com/aws/aws-sdk-go-v2/internal/v4a v1.3.22/go.mod h1:kbR1TL8llqB1eGnVbybcA4/wgScxdylOdyAd51yxPdw=
+github.com/aws/aws-sdk-go-v2/internal/v4a v1.3.23 h1:1SZBDiRzzs3sNhOMVApyWPduWYGAX0imGy06XiBnCAM=
+github.com/aws/aws-sdk-go-v2/internal/v4a v1.3.23/go.mod h1:i9TkxgbZmHVh2S0La6CAXtnyFhlCX/pJ0JsOvBAS6Mk=
 github.com/aws/aws-sdk-go-v2/service/glue v1.101.0 h1:UiKyNrUwlM2FfHk1D8TefZIPVf4ubM3Qr3vmdNKfxtE=
 github.com/aws/aws-sdk-go-v2/service/glue v1.101.0/go.mod h1:TjtkCUyO8rZfxl0K6c3BF2L0K+ZbhiM7gClYk4wXyJ0=
 github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.12.0 h1:TToQNkvGguu209puTojY/ozlqy2d/SFNcoLIqTFi42g=
 github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.12.0/go.mod h1:0jp+ltwkf+SwG2fm/PKo8t4y8pJSgOCO4D8Lz3k0aHQ=
-github.com/aws/aws-sdk-go-v2/service/internal/checksum v1.4.3 h1:kT6BcZsmMtNkP/iYMcRG+mIEA/IbeiUimXtGmqF39y0=
-github.com/aws/aws-sdk-go-v2/service/internal/checksum v1.4.3/go.mod h1:Z8uGua2k4PPaGOYn66pK02rhMrot3Xk3tpBuUFPomZU=
-github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.12.3 h1:qcxX0JYlgWH3hpPUnd6U0ikcl6LLA9sLkXE2w1fpMvY=
-github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.12.3/go.mod h1:cLSNEmI45soc+Ef8K/L+8sEA3A3pYFEYf5B5UI+6bH4=
-github.com/aws/aws-sdk-go-v2/service/internal/s3shared v1.18.3 h1:ZC7Y/XgKUxwqcdhO5LE8P6oGP1eh6xlQReWNKfhvJno=
-github.com/aws/aws-sdk-go-v2/service/internal/s3shared v1.18.3/go.mod h1:WqfO7M9l9yUAw0HcHaikwRd/H6gzYdz7vjejCA5e2oY=
-github.com/aws/aws-sdk-go-v2/service/s3 v1.66.2 h1:p9TNFL8bFUMd+38YIpTAXpoxyz0MxC7FlbFEH4P4E1U=
-github.com/aws/aws-sdk-go-v2/service/s3 v1.66.2/go.mod h1:fNjyo0Coen9QTwQLWeV6WO2Nytwiu+cCcWaTdKCAqqE=
+github.com/aws/aws-sdk-go-v2/service/internal/checksum v1.4.4 h1:aaPpoG15S2qHkWm4KlEyF01zovK1nW4BBbyXuHNSE90=
+github.com/aws/aws-sdk-go-v2/service/internal/checksum v1.4.4/go.mod h1:eD9gS2EARTKgGr/W5xwgY/ik9z/zqpW+m/xOQbVxrMk=
+github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.12.4 h1:tHxQi/XHPK0ctd/wdOw0t7Xrc2OxcRCnVzv8lwWPu0c=
+github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.12.4/go.mod h1:4GQbF1vJzG60poZqWatZlhP31y8PGCCVTvIGPdaaYJ0=
+github.com/aws/aws-sdk-go-v2/service/internal/s3shared v1.18.4 h1:E5ZAVOmI2apR8ADb72Q63KqwwwdW1XcMeXIlrZ1Psjg=
+github.com/aws/aws-sdk-go-v2/service/internal/s3shared v1.18.4/go.mod h1:wezzqVUOVVdk+2Z/JzQT4NxAU0NbhRe5W8pIE72jsWI=
+github.com/aws/aws-sdk-go-v2/service/s3 v1.66.3 h1:neNOYJl72bHrz9ikAEED4VqWyND/Po0DnEx64RW6YM4=
+github.com/aws/aws-sdk-go-v2/service/s3 v1.66.3/go.mod h1:TMhLIyRIyoGVlaEMAt+ITMbwskSTpcGsCPDq91/ihY0=
 github.com/aws/aws-sdk-go-v2/service/sso v1.24.3 h1:UTpsIf0loCIWEbrqdLb+0RxnTXfWh2vhw4nQmFi4nPc=
 github.com/aws/aws-sdk-go-v2/service/sso v1.24.3/go.mod h1:FZ9j3PFHHAR+w0BSEjK955w5YD2UwB/l/H0yAK3MJvI=
 github.com/aws/aws-sdk-go-v2/service/ssooidc v1.28.3 h1:2YCmIXv3tmiItw0LlYf6v7gEHebLY45kBEnPezbUKyU=
@@ -176,8 +176,10 @@ github.com/stretchr/testify v1.6.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/
 github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
 github.com/stretchr/testify v1.9.0 h1:HtqpIVDClZ4nwg75+f6Lvsy/wHu+3BoSGCbBAcpTsTg=
 github.com/stretchr/testify v1.9.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY=
-github.com/substrait-io/substrait-go v1.1.0 h1:wUXoXV/ESMXgaOWu/05kvMI/UBUyhtaTRfkT5p1b5Ck=
-github.com/substrait-io/substrait-go v1.1.0/go.mod h1:LHzL5E0VL620yw4kBQCP+sQPmxhepPTQMDJQRbOe/T4=
+github.com/substrait-io/substrait v0.57.1 h1:GW8nnYfSowMseHR8Os82/X6lNtQGIK7p4p+lr6r+auw=
+github.com/substrait-io/substrait v0.57.1/go.mod h1:q9s+tjo+gK0lsA+SqYB0lhojNuxvdPdfYlGUP0hjbrA=
+github.com/substrait-io/substrait-go v1.2.0 h1:3ZNRkc8FYD7ifCagKEOZQtUcgMceMQfwo2N1NGaK4Q4=
+github.com/substrait-io/substrait-go v1.2.0/go.mod h1:IPsy24rdjp/buXR+T8ENl6QCnSCS6h+uM8P+GaZez7c=
 github.com/twmb/murmur3 v1.1.8 h1:8Yt9taO/WN3l08xErzjeschgZU2QSrwm1kclYq+0aRg=
 github.com/twmb/murmur3 v1.1.8/go.mod h1:Qq/R7NUyOfr65zD+6Q5IHKsJLwP7exErjN6lyyq3OSQ=
 github.com/wolfeidau/s3iofs v1.5.2 h1:2dmzSxdrSY29GsILVheJqBbURVQX3KglggSBtVWCYj4=
diff --git a/manifest.go b/manifest.go
index 700702b..208e09a 100644
--- a/manifest.go
+++ b/manifest.go
@@ -18,10 +18,12 @@
 package iceberg
 
 import (
+	"fmt"
 	"io"
 	"sync"
 	"time"
 
+	"github.com/apache/iceberg-go/internal"
 	iceio "github.com/apache/iceberg-go/io"
 
 	"github.com/hamba/avro/v2"
@@ -213,6 +215,11 @@ func (m *manifestFileV1) FetchEntries(fs iceio.IO, discardDeleted bool) ([]Manif
 	return fetchManifestEntries(m, fs, discardDeleted)
 }
 
+// WriteEntries writes a list of manifest entries to an avro file.
+func (m *manifestFileV1) WriteEntries(out io.Writer, entries []ManifestEntry) error {
+	return writeManifestEntries(out, entries, m.Version())
+}
+
 // ManifestV2Builder is a helper for building a V2 manifest file
 // struct which will conform to the ManifestFile interface.
 type ManifestV2Builder struct {
@@ -363,6 +370,11 @@ func (m *manifestFileV2) FetchEntries(fs iceio.IO, discardDeleted bool) ([]Manif
 	return fetchManifestEntries(m, fs, discardDeleted)
 }
 
+// WriteEntries writes a list of manifest entries to an avro file.
+func (m *manifestFileV2) WriteEntries(out io.Writer, entries []ManifestEntry) error {
+	return writeManifestEntries(out, entries, m.Version())
+}
+
 func getFieldIDMap(sc avro.Schema) (map[string]int, map[int]avro.LogicalType) {
 	getField := func(rs *avro.RecordSchema, name string) *avro.Field {
 		for _, f := range rs.Fields() {
@@ -439,12 +451,12 @@ func fetchManifestEntries(m ManifestFile, fs iceio.IO, discardDeleted bool) ([]M
 		var tmp ManifestEntry
 		if isVer1 {
 			if isFallback {
-				tmp = &fallbackManifestEntryV1{}
+				tmp = &fallbackManifestEntryV1{manifestEntryV1: manifestEntryV1{Data: &dataFile{}}}
 			} else {
-				tmp = &manifestEntryV1{}
+				tmp = &manifestEntryV1{Data: &dataFile{}}
 			}
 		} else {
-			tmp = &manifestEntryV2{}
+			tmp = &manifestEntryV2{Data: &dataFile{}}
 		}
 
 		if err := dec.Decode(tmp); err != nil {
@@ -528,6 +540,10 @@ type ManifestFile interface {
 	// If discardDeleted is true, entries for files containing deleted rows
 	// will be skipped.
 	FetchEntries(fs iceio.IO, discardDeleted bool) ([]ManifestEntry, error)
+	// WriteEntries writes a list of manifest entries to a provided
+	// io.Writer. The version of the manifest file is used to determine the
+	// schema to use for writing the entries.
+	WriteEntries(out io.Writer, entries []ManifestEntry) error
 }
 
 // ReadManifestList reads in an avro manifest list file and returns a slice
@@ -580,6 +596,50 @@ func ReadManifestList(in io.Reader) ([]ManifestFile, error) {
 	return out, dec.Error()
 }
 
+// WriteManifestList writes a list of v2 manifest files to an avro file.
+func WriteManifestList(out io.Writer, files []ManifestFile) error {
+	if len(files) == 0 {
+		return nil
+	}
+
+	version := files[0].Version()
+
+	for _, file := range files[1:] {
+		if file.Version() != version {
+			return fmt.Errorf(
+				"%w: ManifestFile '%s' has non-matching version %d instead of %d",
+				ErrInvalidArgument, file.FilePath(), file.Version(), version,
+			)
+		}
+	}
+
+	var key string
+	switch version {
+	case 1:
+		key = internal.ManifestListV1Key
+	case 2:
+		key = internal.ManifestListV2Key
+	default:
+		return fmt.Errorf("%w: non-recognized version %d", ErrInvalidArgument, version)
+	}
+
+	return avroEncode(key, version, files, out)
+}
+
+func writeManifestEntries(out io.Writer, entries []ManifestEntry, version int) error {
+	var key string
+	switch version {
+	case 1:
+		key = internal.ManifestEntryV1Key
+	case 2:
+		key = internal.ManifestEntryV2Key
+	default:
+		return fmt.Errorf("%w: non-recognized version %d", ErrInvalidArgument, version)
+	}
+
+	return avroEncode(key, version, entries, out)
+}
+
 // ManifestEntryStatus defines constants for the entry status of
 // existing, added or deleted.
 type ManifestEntryStatus int8
@@ -640,6 +700,18 @@ func avroColMapToMap[K comparable, V any](c *[]colMap[K, V]) map[K]V {
 	return out
 }
 
+func mapToAvroColMap[K comparable, V any](m map[K]V) *[]colMap[K, V] {
+	if m == nil {
+		return nil
+	}
+
+	out := make([]colMap[K, V], 0, len(m))
+	for k, v := range m {
+		out = append(out, colMap[K, V]{Key: k, Value: v})
+	}
+	return &out
+}
+
 func avroPartitionData(input map[string]any, nameToID map[string]int, logicalTypes map[int]avro.LogicalType) map[string]any {
 	out := make(map[string]any)
 	for k, v := range input {
@@ -791,12 +863,35 @@ func (d *dataFile) EqualityFieldIDs() []int {
 
 func (d *dataFile) SortOrderID() *int { return d.SortOrder }
 
+// ManifestEntryV1Builder is a helper for building a V1 manifest entry
+// struct which will conform to the ManifestEntry interface.
+type ManifestEntryV1Builder struct {
+	m *manifestEntryV1
+}
+
+// NewManifestEntryV1Builder is passed all of the required fields and then allows
+// all of the optional fields to be set by calling the corresponding methods
+// before calling [ManifestEntryV1Builder.Build] to construct the object.
+func NewManifestEntryV1Builder(status ManifestEntryStatus, snapshotID int64, data DataFile) (*ManifestEntryV1Builder, error) {
+	return &ManifestEntryV1Builder{
+		m: &manifestEntryV1{
+			EntryStatus: status,
+			Snapshot:    snapshotID,
+			Data:        data,
+		},
+	}, nil
+}
+
+func (b *ManifestEntryV1Builder) Build() ManifestEntry {
+	return b.m
+}
+
 type manifestEntryV1 struct {
 	EntryStatus ManifestEntryStatus `avro:"status"`
 	Snapshot    int64               `avro:"snapshot_id"`
 	SeqNum      *int64
 	FileSeqNum  *int64
-	Data        dataFile `avro:"data_file"`
+	Data        DataFile `avro:"data_file"`
 }
 
 type fallbackManifestEntryV1 struct {
@@ -825,14 +920,53 @@ func (m *manifestEntryV1) FileSequenceNum() *int64 {
 	return m.FileSeqNum
 }
 
-func (m *manifestEntryV1) DataFile() DataFile { return &m.Data }
+func (m *manifestEntryV1) DataFile() DataFile { return m.Data }
+
+// ManifestEntryV2Builder is a helper for building a V2 manifest entry
+// struct which will conform to the ManifestEntry interface.
+type ManifestEntryV2Builder struct {
+	m *manifestEntryV2
+}
+
+// NewManifestEntryV2Builder is passed all of the required fields and then allows
+// all of the optional fields to be set by calling the corresponding methods
+// before calling [ManifestEntryV2Builder.Build] to construct the object.
+func NewManifestEntryV2Builder(status ManifestEntryStatus, snapshotID int64, data DataFile) *ManifestEntryV2Builder {
+	return &ManifestEntryV2Builder{
+		m: &manifestEntryV2{
+			EntryStatus: status,
+			Snapshot:    &snapshotID,
+			Data:        data,
+		},
+	}
+}
+
+// SequenceNum sets the sequence number for the manifest entry.
+func (b *ManifestEntryV2Builder) SequenceNum(num int64) *ManifestEntryV2Builder {
+	b.m.SeqNum = &num
+	return b
+}
+
+// FileSequenceNum sets the file sequence number for the manifest entry.
+func (b *ManifestEntryV2Builder) FileSequenceNum(num int64) *ManifestEntryV2Builder {
+	b.m.FileSeqNum = &num
+	return b
+}
+
+// Build returns the constructed manifest entry, after calling Build this
+// builder should not be used further as we avoid copying by just returning
+// a pointer to the constructed manifest entry. Further calls to the modifier
+// methods after calling build would modify the constructed ManifestEntry.
+func (b *ManifestEntryV2Builder) Build() ManifestEntry {
+	return b.m
+}
 
 type manifestEntryV2 struct {
 	EntryStatus ManifestEntryStatus `avro:"status"`
 	Snapshot    *int64              `avro:"snapshot_id"`
 	SeqNum      *int64              `avro:"sequence_number"`
 	FileSeqNum  *int64              `avro:"file_sequence_number"`
-	Data        dataFile            `avro:"data_file"`
+	Data        DataFile            `avro:"data_file"`
 }
 
 func (m *manifestEntryV2) inheritSeqNum(manifest ManifestFile) {
@@ -870,7 +1004,138 @@ func (m *manifestEntryV2) FileSequenceNum() *int64 {
 	return m.FileSeqNum
 }
 
-func (m *manifestEntryV2) DataFile() DataFile { return &m.Data }
+func (m *manifestEntryV2) DataFile() DataFile { return m.Data }
+
+// DataFileBuilder is a helper for building a data file struct which will
+// conform to the DataFile interface.
+type DataFileBuilder struct {
+	d *dataFile
+}
+
+// NewDataFileBuilder is passed all of the required fields and then allows
+// all of the optional fields to be set by calling the corresponding methods
+// before calling [DataFileBuilder.Build] to construct the object.
+func NewDataFileBuilder(
+	content ManifestEntryContent,
+	path string,
+	format FileFormat,
+	partitionData map[string]any,
+	recordCount int64,
+	fileSize int64,
+) (*DataFileBuilder, error) {
+	if content != EntryContentData && content != EntryContentPosDeletes && content != EntryContentEqDeletes {
+		return nil, fmt.Errorf(
+			"%w: content must be one of %s, %s, or %s",
+			ErrInvalidArgument, EntryContentData, EntryContentPosDeletes, EntryContentEqDeletes,
+		)
+	}
+
+	if path == "" {
+		return nil, fmt.Errorf("%w: path cannot be empty", ErrInvalidArgument)
+	}
+
+	if format != AvroFile && format != OrcFile && format != ParquetFile {
+		return nil, fmt.Errorf(
+			"%w: format must be one of %s, %s, or %s",
+			ErrInvalidArgument, AvroFile, OrcFile, ParquetFile,
+		)
+	}
+
+	if recordCount <= 0 {
+		return nil, fmt.Errorf("%w: record count must be greater than 0", ErrInvalidArgument)
+	}
+
+	if fileSize <= 0 {
+		return nil, fmt.Errorf("%w: file size must be greater than 0", ErrInvalidArgument)
+	}
+
+	return &DataFileBuilder{
+		d: &dataFile{
+			Content:       content,
+			Path:          path,
+			Format:        format,
+			PartitionData: partitionData,
+			RecordCount:   recordCount,
+			FileSize:      fileSize,
+		},
+	}, nil
+}
+
+// BlockSizeInBytes sets the block size in bytes for the data file. Deprecated in v2.
+func (b *DataFileBuilder) BlockSizeInBytes(size int64) *DataFileBuilder {
+	b.d.BlockSizeInBytes = size
+	return b
+}
+
+// ColumnSizes sets the column sizes for the data file.
+func (b *DataFileBuilder) ColumnSizes(sizes map[int]int64) *DataFileBuilder {
+	b.d.ColSizes = mapToAvroColMap(sizes)
+	return b
+}
+
+// ValueCounts sets the value counts for the data file.
+func (b *DataFileBuilder) ValueCounts(counts map[int]int64) *DataFileBuilder {
+	b.d.ValCounts = mapToAvroColMap(counts)
+	return b
+}
+
+// NullValueCounts sets the null value counts for the data file.
+func (b *DataFileBuilder) NullValueCounts(counts map[int]int64) *DataFileBuilder {
+	b.d.NullCounts = mapToAvroColMap(counts)
+	return b
+}
+
+// NaNValueCounts sets the NaN value counts for the data file.
+func (b *DataFileBuilder) NaNValueCounts(counts map[int]int64) *DataFileBuilder {
+	b.d.NaNCounts = mapToAvroColMap(counts)
+	return b
+}
+
+// DistinctValueCounts sets the distinct value counts for the data file.
+func (b *DataFileBuilder) DistinctValueCounts(counts map[int]int64) *DataFileBuilder {
+	b.d.DistinctCounts = mapToAvroColMap(counts)
+	return b
+}
+
+// LowerBoundValues sets the lower bound values for the data file.
+func (b *DataFileBuilder) LowerBoundValues(bounds map[int][]byte) *DataFileBuilder {
+	b.d.LowerBounds = mapToAvroColMap(bounds)
+	return b
+}
+
+// UpperBoundValues sets the upper bound values for the data file.
+func (b *DataFileBuilder) UpperBoundValues(bounds map[int][]byte) *DataFileBuilder {
+	b.d.UpperBounds = mapToAvroColMap(bounds)
+	return b
+}
+
+// KeyMetadata sets the key metadata for the data file.
+func (b *DataFileBuilder) KeyMetadata(key []byte) *DataFileBuilder {
+	b.d.Key = &key
+	return b
+}
+
+// SplitOffsets sets the split offsets for the data file.
+func (b *DataFileBuilder) SplitOffsets(offsets []int64) *DataFileBuilder {
+	b.d.Splits = &offsets
+	return b
+}
+
+// EqualityFieldIDs sets the equality field ids for the data file.
+func (b *DataFileBuilder) EqualityFieldIDs(ids []int) *DataFileBuilder {
+	b.d.EqualityIDs = &ids
+	return b
+}
+
+// SortOrderID sets the sort order id for the data file.
+func (b *DataFileBuilder) SortOrderID(id int) *DataFileBuilder {
+	b.d.SortOrder = &id
+	return b
+}
+
+func (b *DataFileBuilder) Build() DataFile {
+	return b.d
+}
 
 // DataFile is the interface for reading the information about a
 // given data file indicated by an entry in a manifest list.
diff --git a/utils.go b/utils.go
index 7807acb..087fb25 100644
--- a/utils.go
+++ b/utils.go
@@ -21,9 +21,14 @@ import (
 	"cmp"
 	"fmt"
 	"hash/maphash"
+	"io"
 	"maps"
 	"runtime/debug"
+	"strconv"
 	"strings"
+
+	"github.com/apache/iceberg-go/internal"
+	"github.com/hamba/avro/v2/ocf"
 )
 
 var version string
@@ -212,3 +217,24 @@ func Difference(a, b []string) []string {
 	}
 	return diff
 }
+
+func avroEncode[T any](key string, version int, vals []T, out io.Writer) error {
+	enc, err := ocf.NewEncoderWithSchema(
+		internal.AvroSchemaCache.Get(key),
+		out, ocf.WithMetadata(map[string][]byte{
+			"format-version": []byte(strconv.Itoa(version)),
+		}),
+		ocf.WithCodec(ocf.Deflate),
+	)
+	if err != nil {
+		return err
+	}
+
+	for _, file := range vals {
+		if err := enc.Encode(file); err != nil {
+			return err
+		}
+	}
+
+	return enc.Close()
+}


Test patch:
diff --git a/manifest_test.go b/manifest_test.go
index 486c6a8..0e6e04c 100644
--- a/manifest_test.go
+++ b/manifest_test.go
@@ -23,7 +23,6 @@ import (
 	"time"
 
 	"github.com/apache/iceberg-go/internal"
-	"github.com/hamba/avro/v2/ocf"
 	"github.com/stretchr/testify/suite"
 )
 
@@ -69,7 +68,7 @@ var (
 		{
 			EntryStatus: EntryStatusADDED,
 			Snapshot:    entrySnapshotID,
-			Data: dataFile{
+			Data: &dataFile{
 				// bad value for Content but this field doesn't exist in V1
 				// so it shouldn't get written and shouldn't be read back out
 				// so the roundtrip test asserts that we get the default value
@@ -193,7 +192,7 @@ var (
 		{
 			EntryStatus: EntryStatusADDED,
 			Snapshot:    8744736658442914487,
-			Data: dataFile{
+			Data: &dataFile{
 				Path:             "/home/iceberg/warehouse/nyc/taxis_partitioned/data/VendorID=1/00000-633-d8a4223e-dc97-45a1-86e1-adaba6e8abd7-00002.parquet",
 				Format:           ParquetFile,
 				PartitionData:    map[string]any{"VendorID": int(1), "tpep_pickup_datetime": time.Unix(1925, 0)},
@@ -321,45 +320,48 @@ var (
 		},
 	}
 
+	dataRecord0 = manifestEntryV1Records[0].Data.(*dataFile)
+	dataRecord1 = manifestEntryV1Records[1].Data.(*dataFile)
+
 	manifestEntryV2Records = []*manifestEntryV2{
 		{
 			EntryStatus: EntryStatusADDED,
 			Snapshot:    &entrySnapshotID,
-			Data: dataFile{
-				Path:             manifestEntryV1Records[0].Data.Path,
-				Format:           manifestEntryV1Records[0].Data.Format,
-				PartitionData:    manifestEntryV1Records[0].Data.PartitionData,
-				RecordCount:      manifestEntryV1Records[0].Data.RecordCount,
-				FileSize:         manifestEntryV1Records[0].Data.FileSize,
-				BlockSizeInBytes: manifestEntryV1Records[0].Data.BlockSizeInBytes,
-				ColSizes:         manifestEntryV1Records[0].Data.ColSizes,
-				ValCounts:        manifestEntryV1Records[0].Data.ValCounts,
-				NullCounts:       manifestEntryV1Records[0].Data.NullCounts,
-				NaNCounts:        manifestEntryV1Records[0].Data.NaNCounts,
-				LowerBounds:      manifestEntryV1Records[0].Data.LowerBounds,
-				UpperBounds:      manifestEntryV1Records[0].Data.UpperBounds,
-				Splits:           manifestEntryV1Records[0].Data.Splits,
-				SortOrder:        manifestEntryV1Records[0].Data.SortOrder,
+			Data: &dataFile{
+				Path:             dataRecord0.Path,
+				Format:           dataRecord0.Format,
+				PartitionData:    dataRecord0.PartitionData,
+				RecordCount:      dataRecord0.RecordCount,
+				FileSize:         dataRecord0.FileSize,
+				BlockSizeInBytes: dataRecord0.BlockSizeInBytes,
+				ColSizes:         dataRecord0.ColSizes,
+				ValCounts:        dataRecord0.ValCounts,
+				NullCounts:       dataRecord0.NullCounts,
+				NaNCounts:        dataRecord0.NaNCounts,
+				LowerBounds:      dataRecord0.LowerBounds,
+				UpperBounds:      dataRecord0.UpperBounds,
+				Splits:           dataRecord0.Splits,
+				SortOrder:        dataRecord0.SortOrder,
 			},
 		},
 		{
 			EntryStatus: EntryStatusADDED,
 			Snapshot:    &entrySnapshotID,
-			Data: dataFile{
-				Path:             manifestEntryV1Records[1].Data.Path,
-				Format:           manifestEntryV1Records[1].Data.Format,
-				PartitionData:    manifestEntryV1Records[1].Data.PartitionData,
-				RecordCount:      manifestEntryV1Records[1].Data.RecordCount,
-				FileSize:         manifestEntryV1Records[1].Data.FileSize,
-				BlockSizeInBytes: manifestEntryV1Records[1].Data.BlockSizeInBytes,
-				ColSizes:         manifestEntryV1Records[1].Data.ColSizes,
-				ValCounts:        manifestEntryV1Records[1].Data.ValCounts,
-				NullCounts:       manifestEntryV1Records[1].Data.NullCounts,
-				NaNCounts:        manifestEntryV1Records[1].Data.NaNCounts,
-				LowerBounds:      manifestEntryV1Records[1].Data.LowerBounds,
-				UpperBounds:      manifestEntryV1Records[1].Data.UpperBounds,
-				Splits:           manifestEntryV1Records[1].Data.Splits,
-				SortOrder:        manifestEntryV1Records[1].Data.SortOrder,
+			Data: &dataFile{
+				Path:             dataRecord1.Path,
+				Format:           dataRecord1.Format,
+				PartitionData:    dataRecord1.PartitionData,
+				RecordCount:      dataRecord1.RecordCount,
+				FileSize:         dataRecord1.FileSize,
+				BlockSizeInBytes: dataRecord1.BlockSizeInBytes,
+				ColSizes:         dataRecord1.ColSizes,
+				ValCounts:        dataRecord1.ValCounts,
+				NullCounts:       dataRecord1.NullCounts,
+				NaNCounts:        dataRecord1.NaNCounts,
+				LowerBounds:      dataRecord1.LowerBounds,
+				UpperBounds:      dataRecord1.UpperBounds,
+				Splits:           dataRecord1.Splits,
+				SortOrder:        dataRecord1.SortOrder,
 			},
 		},
 	}
@@ -376,50 +378,23 @@ type ManifestTestSuite struct {
 }
 
 func (m *ManifestTestSuite) writeManifestList() {
-	enc, err := ocf.NewEncoder(internal.AvroSchemaCache.Get(internal.ManifestListV1Key).String(),
-		&m.v1ManifestList, ocf.WithMetadata(map[string][]byte{
-			"avro.codec": []byte("deflate"),
-		}),
-		ocf.WithCodec(ocf.Deflate))
-	m.Require().NoError(err)
-
-	m.Require().NoError(enc.Encode(manifestFileRecordsV1[0]))
-	enc.Close()
-
-	enc, err = ocf.NewEncoder(internal.AvroSchemaCache.Get(internal.ManifestListV2Key).String(),
-		&m.v2ManifestList, ocf.WithMetadata(map[string][]byte{
-			"format-version": []byte("2"),
-			"avro.codec":     []byte("deflate"),
-		}), ocf.WithCodec(ocf.Deflate))
-	m.Require().NoError(err)
-
-	m.Require().NoError(enc.Encode(manifestFileRecordsV2[0]))
-	enc.Close()
+	m.Require().NoError(WriteManifestList(&m.v1ManifestList, manifestFileRecordsV1))
+	m.Require().NoError(WriteManifestList(&m.v2ManifestList, manifestFileRecordsV2))
 }
 
 func (m *ManifestTestSuite) writeManifestEntries() {
-	enc, err := ocf.NewEncoder(internal.AvroSchemaCache.Get(internal.ManifestEntryV1Key).String(), &m.v1ManifestEntries,
-		ocf.WithMetadata(map[string][]byte{
-			"format-version": []byte("1"),
-		}), ocf.WithCodec(ocf.Deflate))
-	m.Require().NoError(err)
-
-	for _, ent := range manifestEntryV1Records {
-		m.Require().NoError(enc.Encode(ent))
+	manifestEntryV1Recs := make([]ManifestEntry, len(manifestEntryV1Records))
+	for i, rec := range manifestEntryV1Records {
+		manifestEntryV1Recs[i] = rec
 	}
-	m.Require().NoError(enc.Close())
 
-	enc, err = ocf.NewEncoder(internal.AvroSchemaCache.Get(internal.ManifestEntryV2Key).String(),
-		&m.v2ManifestEntries, ocf.WithMetadata(map[string][]byte{
-			"format-version": []byte("2"),
-			"avro.codec":     []byte("deflate"),
-		}), ocf.WithCodec(ocf.Deflate))
-	m.Require().NoError(err)
-
-	for _, ent := range manifestEntryV2Records {
-		m.Require().NoError(enc.Encode(ent))
+	manifestEntryV2Recs := make([]ManifestEntry, len(manifestEntryV2Records))
+	for i, rec := range manifestEntryV2Records {
+		manifestEntryV2Recs[i] = rec
 	}
-	m.Require().NoError(enc.Close())
+
+	m.Require().NoError(writeManifestEntries(&m.v1ManifestEntries, manifestEntryV1Recs, 1))
+	m.Require().NoError(writeManifestEntries(&m.v2ManifestEntries, manifestEntryV2Recs, 2))
 }
 
 func (m *ManifestTestSuite) SetupSuite() {
@@ -767,6 +742,91 @@ func (m *ManifestTestSuite) TestManifestEntriesV2() {
 	m.Zero(*datafile.SortOrderID())
 }
 
+func (m *ManifestTestSuite) TestManifestEntryBuilder() {
+	dataFileBuilder, err := NewDataFileBuilder(
+		EntryContentData,
+		"sample.parquet",
+		ParquetFile,
+		map[string]any{"int": int(1), "datetime": time.Unix(1925, 0)},
+		1,
+		2,
+	)
+	m.Require().NoError(err)
+
+	dataFileBuilder.ColumnSizes(map[int]int64{
+		1: 1,
+		2: 2,
+	}).ValueCounts(map[int]int64{
+		1: 1,
+		2: 2,
+	}).DistinctValueCounts(map[int]int64{
+		1: 1,
+		2: 2,
+	}).NullValueCounts(map[int]int64{
+		1: 0,
+		2: 0,
+	}).NaNValueCounts(map[int]int64{
+		1: 0,
+		2: 0,
+	}).LowerBoundValues(map[int][]byte{
+		1: {0x01, 0x00, 0x00, 0x00},
+		2: []byte("2020-04-01 00:00"),
+	}).UpperBoundValues(map[int][]byte{
+		1: {0x01, 0x00, 0x00, 0x00},
+		2: []byte("2020-04-30 23:5:"),
+	}).SplitOffsets([]int64{4}).EqualityFieldIDs([]int{1, 1}).SortOrderID(0)
+
+	builder, err := NewManifestEntryV1Builder(
+		EntryStatusEXISTING,
+		1,
+		dataFileBuilder.Build(),
+	)
+	m.Require().NoError(err)
+
+	entry := builder.Build()
+	m.Assert().Equal(EntryStatusEXISTING, entry.Status())
+	m.Assert().EqualValues(1, entry.SnapshotID())
+	m.Assert().Equal(int64(0), entry.SequenceNum())
+	m.Assert().Nil(entry.FileSequenceNum())
+	data := entry.DataFile()
+	m.Assert().Equal(EntryContentData, data.ContentType())
+	m.Assert().Equal("sample.parquet", data.FilePath())
+	m.Assert().Equal(ParquetFile, data.FileFormat())
+	m.Assert().EqualValues(1, data.Count())
+	m.Assert().EqualValues(2, data.FileSizeBytes())
+	m.Assert().Equal(map[int]int64{
+		1: 1,
+		2: 2,
+	}, data.ColumnSizes())
+	m.Assert().Equal(map[int]int64{
+		1: 1,
+		2: 2,
+	}, data.ValueCounts())
+	m.Assert().Equal(map[int]int64{
+		1: 0,
+		2: 0,
+	}, data.NullValueCounts())
+	m.Assert().Equal(map[int]int64{
+		1: 1,
+		2: 2,
+	}, data.DistinctValueCounts())
+	m.Assert().Equal(map[int]int64{
+		1: 0,
+		2: 0,
+	}, data.NaNValueCounts())
+	m.Assert().Equal(map[int][]byte{
+		1: {0x01, 0x00, 0x00, 0x00},
+		2: []byte("2020-04-01 00:00"),
+	}, data.LowerBoundValues())
+	m.Assert().Equal(map[int][]byte{
+		1: {0x01, 0x00, 0x00, 0x00},
+		2: []byte("2020-04-30 23:5:"),
+	}, data.UpperBoundValues())
+	m.Assert().Equal([]int64{4}, data.SplitOffsets())
+	m.Assert().Equal([]int{1, 1}, data.EqualityFieldIDs())
+	m.Assert().Equal(0, *data.SortOrderID())
+}
+
 func TestManifests(t *testing.T) {
 	suite.Run(t, new(ManifestTestSuite))
 }