Repository: apache/iceberg-go
Pull request #194 (2024-11-10 08:46:14)

build(deps): bump github.com/aws/aws-sdk-go-v2/service/glue from 1.101.0 to 1.101.2
Bumps [github.com/aws/aws-sdk-go-v2/service/glue](https://github.com/aws/aws-sdk-go-v2) from 1.101.0 to 1.101.2.
<details>
<summary>Commits</summary>
<ul>
<li><a href="https://github.com/aws/aws-sdk-go-v2/commit/2f70834c690f3fecf04ef006f0ea7022884b367f"><code>2f70834</code></a> Release 2024-11-07</li>
<li><a href="https://github.com/aws/aws-sdk-go-v2/commit/ef9a3c319fb525f680f6d6bdffdfe4153eaa3135"><code>ef9a3c3</code></a> Regenerated Clients</li>
<li><a href="https://github.com/aws/aws-sdk-go-v2/commit/c054fe7021957875ecc9707b0cfd49512251d1e9"><code>c054fe7</code></a> Update endpoints model</li>
<li><a href="https://github.com/aws/aws-sdk-go-v2/commit/b68675cf5962947f8001f96ee8cd3b1d66aacc46"><code>b68675c</code></a> Update API model</li>
<li><a href="https://github.com/aws/aws-sdk-go-v2/commit/1d989f31b0248f4d800d6b07bdab20bde40b59c5"><code>1d989f3</code></a> send opt-in query-compatible header where applicable (<a href="https://redirect.github.com/aws/aws-sdk-go-v2/issues/2891">#2891</a>)</li>
<li><a href="https://github.com/aws/aws-sdk-go-v2/commit/5d0eb2386ab466c40e4760960f7cf160cdac557b"><code>5d0eb23</code></a> Merge pull request <a href="https://redirect.github.com/aws/aws-sdk-go-v2/issues/2889">#2889</a> from aws/customize-identity-store-exception-message</li>
<li><a href="https://github.com/aws/aws-sdk-go-v2/commit/70eb57ac775f782db6856c73f1ca22eae8e48ac2"><code>70eb57a</code></a> Release 2024-11-06</li>
<li><a href="https://github.com/aws/aws-sdk-go-v2/commit/cd2c6b1af3d2fd8b114804a604a6b621fbed92c3"><code>cd2c6b1</code></a> Regenerated Clients</li>
<li><a href="https://github.com/aws/aws-sdk-go-v2/commit/2b2a737b179ea3b7b699584ff6c391619b8f6de3"><code>2b2a737</code></a> Update API model</li>
<li><a href="https://github.com/aws/aws-sdk-go-v2/commit/8c9892f0427c4bf13dac6f944b250fe6512eb6cf"><code>8c9892f</code></a> bump smithy-go codegen to latest (<a href="https://redirect.github.com/aws/aws-sdk-go-v2/issues/2890">#2890</a>)</li>
<li>Additional commits viewable in <a href="https://github.com/aws/aws-sdk-go-v2/compare/service/ec2/v1.101.0...service/glue/v1.101.2">compare view</a></li>
</ul>
</details>
<br />


[![Dependabot compatibility score](https://dependabot-badges.githubapp.com/badges/compatibility_score?dependency-name=github.com/aws/aws-sdk-go-v2/service/glue&package-manager=go_modules&previous-version=1.101.0&new-version=1.101.2)](https://docs.github.com/en/github/managing-security-vulnerabilities/about-dependabot-security-updates#about-compatibility-scores)

Dependabot will resolve any conflicts with this PR as long as you don't alter it yourself. You can also trigger a rebase manually by commenting `@dependabot rebase`.

[//]: # (dependabot-automerge-start)
[//]: # (dependabot-automerge-end)

---

<details>
<summary>Dependabot commands and options</summary>
<br />

You can trigger Dependabot actions by commenting on this PR:
- `@dependabot rebase` will rebase this PR
- `@dependabot recreate` will recreate this PR, overwriting any edits that have been made to it
- `@dependabot merge` will merge this PR after your CI passes on it
- `@dependabot squash and merge` will squash and merge this PR after your CI passes on it
- `@dependabot cancel merge` will cancel a previously requested merge and block automerging
- `@dependabot reopen` will reopen this PR if it is closed
- `@dependabot close` will close this PR and stop Dependabot recreating it. You can achieve the same result by closing it manually
- `@dependabot show <dependency name> ignore conditions` will show all of the ignore conditions of the specified dependency
- `@dependabot ignore this major version` will close this PR and stop Dependabot creating any more for this major version (unless you reopen the PR or upgrade to it yourself)
- `@dependabot ignore this minor version` will close this PR and stop Dependabot creating any more for this minor version (unless you reopen the PR or upgrade to it yourself)
- `@dependabot ignore this dependency` will close this PR and stop Dependabot creating any more for this dependency (unless you reopen the PR or upgrade to it yourself)


</details>


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 12b40ae..973cbc1 100644
--- a/go.mod
+++ b/go.mod
@@ -26,7 +26,7 @@ require (
 	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/glue v1.101.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
diff --git a/go.sum b/go.sum
index 3ce7b97..efe1347 100644
--- a/go.sum
+++ b/go.sum
@@ -48,8 +48,8 @@ github.com/aws/aws-sdk-go-v2/internal/ini v1.8.1 h1:VaRN3TlFdd6KxX1x3ILT5ynH6HvK
 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.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/glue v1.101.2 h1:0gvXLmkF95b45HD/s0tenCZPRH7fv+C9CDwzyKVnUTg=
+github.com/aws/aws-sdk-go-v2/service/glue v1.101.2/go.mod h1:m9O1Fp+iJWbt7QmRTR/Zfh6HJpMEKNf0L8/chyp7bu8=
 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.4 h1:aaPpoG15S2qHkWm4KlEyF01zovK1nW4BBbyXuHNSE90=
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))
 }