Repository: apache/iceberg-go
Pull request #198 (2024-11-11 22:44:37)

Add option to set max concurrency for table scan operations
This PR introduces an option to set `MaxConcurrency` for the `Scanner`, allowing users to control the level of concurrent downloads in Scanner. This configuration can be beneficial for workloads running multiple simultaneous queries.

Additionally, this PR changes the default value of max concurrency from `runtime.NumCPU()` to `runtime.GOMAXPROCS`. While both are initially set to the number of available CPUs, `GOMAXPROCS` is adjustable by users, providing greater flexibility. This change addresses a limitation of using `runtime.NumCPU()`, which does not account for cgroup limits. For instance, in Kubernetes environments, using `runtime.NumCPU()` may incorrectly assume access to all CPU cores on a node, rather than only the cores allocated to the specific pod. This mismatch can lead to performance degradation in Kubernetes-deployed applications. See [uber-go/automaxprocs](https://pkg.go.dev/go.uber.org/automaxprocs).


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 c41a0a7..12b40ae 100644
--- a/go.mod
+++ b/go.mod
@@ -23,21 +23,21 @@ toolchain go1.23.2
 
 require (
 	github.com/apache/arrow-go/v18 v18.0.1-0.20241029153821-f0c5d9939d3f
-	github.com/aws/aws-sdk-go-v2 v1.32.3
+	github.com/aws/aws-sdk-go-v2 v1.32.4
 	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.8.0
+	golang.org/x/sync v0.9.0
 	gopkg.in/yaml.v3 v3.0.1
 )
 
@@ -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 0ed534d..3ce7b97 100644
--- a/go.sum
+++ b/go.sum
@@ -25,15 +25,13 @@ github.com/alecthomas/repr v0.2.0 h1:HAzS41CIzNW5syS8Mf9UwXhNH1J9aix/BvDRf1Ml2Yk
 github.com/alecthomas/repr v0.2.0/go.mod h1:Fr0507jx4eOXV7AlPV6AVZLYrLIuIeSOWtW57eE/O/4=
 github.com/andybalholm/brotli v1.1.1 h1:PR2pgnyFznKEugtsUo0xLdDop5SKXd5Qf5ysW+7XdTA=
 github.com/andybalholm/brotli v1.1.1/go.mod h1:05ib4cKhjx3OQYUY22hTVd34Bc8upXjOLL2rKwwZBoA=
-github.com/apache/arrow-go/v18 v18.0.1-0.20241022184425-56b794f52a9b h1:E79+ggEd/cv9g4iLe1B8XyBhpqlvqKbCGDmf+RPSwbA=
-github.com/apache/arrow-go/v18 v18.0.1-0.20241022184425-56b794f52a9b/go.mod h1:kVPeNv6eFSRhkfWZx1BIRXU6EZnp5g2NqKsuJmKXsO8=
 github.com/apache/arrow-go/v18 v18.0.1-0.20241029153821-f0c5d9939d3f h1:k14GhTGJuvq27vRgLxf4iuufzLt7GeN3UOytJmU7W/A=
 github.com/apache/arrow-go/v18 v18.0.1-0.20241029153821-f0c5d9939d3f/go.mod h1:kVPeNv6eFSRhkfWZx1BIRXU6EZnp5g2NqKsuJmKXsO8=
 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=
-github.com/aws/aws-sdk-go-v2 v1.32.3 h1:T0dRlFBKcdaUPGNtkBSwHZxrtis8CQU17UpNBZYd0wk=
-github.com/aws/aws-sdk-go-v2 v1.32.3/go.mod h1:2SK5n0a2karNTv5tbP1SjsX0uhttou00v/HpXKM1ZUo=
+github.com/aws/aws-sdk-go-v2 v1.32.4 h1:S13INUiTxgrPueTmrm5DZ+MiAo99zYzHEFh1UNkOxNE=
+github.com/aws/aws-sdk-go-v2 v1.32.4/go.mod h1:2SK5n0a2karNTv5tbP1SjsX0uhttou00v/HpXKM1ZUo=
 github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.6.6 h1:pT3hpW0cOHRJx8Y0DfJUEQuqPild8jRGmSFmBgvydr0=
 github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.6.6/go.mod h1:j/I2++U0xX+cr44QjHay4Cvxj6FUbnxrgmqN3H1jTZA=
 github.com/aws/aws-sdk-go-v2/config v1.28.1 h1:oxIvOUXy8x0U3fR//0eq+RdCKimWI900+SV+10xsCBw=
@@ -42,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=
@@ -178,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=
@@ -213,8 +213,8 @@ golang.org/x/net v0.30.0/go.mod h1:2wGyMJ5iFasEhkwi13ChkO/t1ECNC4X4eBKkVFyYFlU=
 golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20220722155255-886fb9371eb4/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.1.0/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
-golang.org/x/sync v0.8.0 h1:3NFvSEYkUoMifnESzZl15y791HH1qU2xm6eCJU5ZPXQ=
-golang.org/x/sync v0.8.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk=
+golang.org/x/sync v0.9.0 h1:fEo0HyrW1GIgZdpbhCRO0PkJajUS5H9IFUztCgEo2jQ=
+golang.org/x/sync v0.9.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk=
 golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20210124154548-22da62e12c0c/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
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/table/arrow_scanner.go b/table/arrow_scanner.go
index 814ae77..97147ce 100644
--- a/table/arrow_scanner.go
+++ b/table/arrow_scanner.go
@@ -21,7 +21,6 @@ import (
 	"context"
 	"io"
 	"iter"
-	"runtime"
 	"strconv"
 	"sync"
 
@@ -45,7 +44,7 @@ const (
 type positionDeletes = []*arrow.Chunked
 type perFilePosDeletes = map[string]positionDeletes
 
-func readAllDeleteFiles(ctx context.Context, fs iceio.IO, tasks []FileScanTask) (perFilePosDeletes, error) {
+func readAllDeleteFiles(ctx context.Context, fs iceio.IO, tasks []FileScanTask, concurrency int) (perFilePosDeletes, error) {
 	var (
 		deletesPerFile = make(perFilePosDeletes)
 		uniqueDeletes  = make(map[string]iceberg.DataFile)
@@ -69,9 +68,9 @@ func readAllDeleteFiles(ctx context.Context, fs iceio.IO, tasks []FileScanTask)
 	}
 
 	g, ctx := errgroup.WithContext(ctx)
-	g.SetLimit(runtime.NumCPU())
+	g.SetLimit(concurrency)
 
-	perFileChan := make(chan map[string]*arrow.Chunked, runtime.NumCPU())
+	perFileChan := make(chan map[string]*arrow.Chunked, concurrency)
 	go func() {
 		defer close(perFileChan)
 		for _, v := range uniqueDeletes {
@@ -213,6 +212,7 @@ type arrowScan struct {
 	options         iceberg.Properties
 
 	useLargeTypes bool
+	concurrency   int
 }
 
 func (as *arrowScan) projectedFieldIDs() (set[int], error) {
@@ -534,7 +534,7 @@ func (as *arrowScan) recordBatchesFromTasksAndDeletes(ctx context.Context, tasks
 	taskChan := make(chan internal.Enumerated[FileScanTask], len(tasks))
 
 	// numWorkers := 1
-	numWorkers := min(runtime.NumCPU(), len(tasks))
+	numWorkers := min(as.concurrency, len(tasks))
 	records := make(chan enumeratedRecord, numWorkers)
 
 	var wg sync.WaitGroup
@@ -592,7 +592,7 @@ func (as *arrowScan) GetRecords(ctx context.Context, tasks []FileScanTask) (*arr
 		return resultSchema, func(yield func(arrow.Record, error) bool) {}, nil
 	}
 
-	deletesPerFile, err := readAllDeleteFiles(ctx, as.fs, tasks)
+	deletesPerFile, err := readAllDeleteFiles(ctx, as.fs, tasks, as.concurrency)
 	if err != nil {
 		return nil, nil, err
 	}
diff --git a/table/scanner.go b/table/scanner.go
index f7aab11..bfb183e 100644
--- a/table/scanner.go
+++ b/table/scanner.go
@@ -22,7 +22,6 @@ import (
 	"context"
 	"fmt"
 	"iter"
-	"runtime"
 	"slices"
 	"sync"
 
@@ -132,6 +131,7 @@ type Scan struct {
 	limit          int64
 
 	partitionFilters *keyDefaultMap[int, iceberg.BooleanExpression]
+	concurrency      int
 }
 
 func (scan *Scan) UseRowLimit(n int64) *Scan {
@@ -294,7 +294,7 @@ func (scan *Scan) PlanFiles(ctx context.Context) ([]FileScanTask, error) {
 	dataEntries := make([]iceberg.ManifestEntry, 0)
 	positionalDeleteEntries := make([]iceberg.ManifestEntry, 0)
 
-	nworkers := min(runtime.NumCPU(), len(manifestList))
+	nworkers := min(scan.concurrency, len(manifestList))
 	var wg sync.WaitGroup
 
 	manifestChan := make(chan iceberg.ManifestFile, len(manifestList))
@@ -434,6 +434,7 @@ func (scan *Scan) ToArrowRecords(ctx context.Context) (*arrow.Schema, iter.Seq2[
 		caseSensitive:   scan.caseSensitive,
 		rowLimit:        scan.limit,
 		options:         scan.options,
+		concurrency:     scan.concurrency,
 	}).GetRecords(ctx, tasks)
 }
 
diff --git a/table/table.go b/table/table.go
index 350782a..ac9cd2b 100644
--- a/table/table.go
+++ b/table/table.go
@@ -18,6 +18,7 @@
 package table
 
 import (
+	"runtime"
 	"slices"
 
 	"github.com/apache/iceberg-go"
@@ -110,6 +111,18 @@ func WithLimit(n int64) ScanOption {
 	}
 }
 
+// WitMaxConcurrency sets the maximum concurrency for table scan and plan
+// operations. When unset it defaults to runtime.GOMAXPROCS.
+func WitMaxConcurrency(n int) ScanOption {
+	if n <= 0 {
+		return noopOption
+	}
+
+	return func(scan *Scan) {
+		scan.concurrency = n
+	}
+}
+
 func WithOptions(opts iceberg.Properties) ScanOption {
 	if opts == nil {
 		return noopOption
@@ -128,6 +141,7 @@ func (t Table) Scan(opts ...ScanOption) *Scan {
 		selectedFields: []string{"*"},
 		caseSensitive:  true,
 		limit:          ScanNoLimit,
+		concurrency:    runtime.GOMAXPROCS(0),
 	}
 
 	for _, opt := range opts {
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))
 }