Repository: apache/iceberg-go
Pull request #369 (2025-03-31 20:44:11)

feat(table): write an Arrow Table or stream of Arrow records
This creates the initial implementation to write a stream of data or a materialized Table of arrow data  to unpartitioned tables. Implementing writing with partitions will be done in a follow-up PR.

For now this will only write Parquet files, but everything is abstracted well enough to easily allow for other file types to be implemented.

laskoviymishka: Maybe it worth to add an e2e test suite that verify that external readers are able to read this data afterwards?
Like test that do write and then read with pandas/duckdb/clickhouse whatever?
Fokko: Ideally something that uses the Java SDK since that's the most feature complete one :)
Fokko: @zeroshade found an issue with the `snapshot_id` being updated:

```json
"snapshots": [
  {
    "snapshot-id": 3510441037372461000,
    "sequence-number": 0,
    "timestamp-ms": 1743534505231,
    "manifest-list": "/tmp/merge-manifests/metadata/snap-3510441037372461004-0-23f676b6-17d5-43a5-aa58-11a7579a0ad6.avro",
    "summary": {
      "added-data-files": "1",
      "added-files-size": "4848",
      "added-records": "3",
      "operation": "append",
      "total-data-files": "1",
      "total-delete-files": "0",
      "total-equality-deletes": "0",
      "total-files-size": "4848",
      "total-position-deletes": "0",
      "total-records": "3"
    },
    "schema-id": 0
  },
  {
    "snapshot-id": 7555903136473091000,
    "parent-snapshot-id": 3510441037372461000,
    "sequence-number": 0,
    "timestamp-ms": 1743534505235,
    "manifest-list": "/tmp/merge-manifests/metadata/snap-7555903136473090822-0-0208f3f6-8436-4c8c-802b-9565fbb68f15.avro",
    "summary": {
      "added-data-files": "1",
      "added-files-size": "4848",
      "added-records": "3",
      "operation": "append",
      "total-data-files": "2",
      "total-delete-files": "0",
      "total-equality-deletes": "0",
      "total-files-size": "9696",
      "total-position-deletes": "0",
      "total-records": "6"
    },
    "schema-id": 0
  },
  {
    "snapshot-id": 7590986702594961000,
    "parent-snapshot-id": 7555903136473091000,
    "sequence-number": 0,
    "timestamp-ms": 1743534505239,
    "manifest-list": "/tmp/merge-manifests/metadata/snap-7590986702594961103-0-94fb271e-d83d-45e6-88d8-ba6fc60e7669.avro",
    "summary": {
      "added-data-files": "1",
      "added-files-size": "4848",
      "added-records": "3",
      "operation": "append",
      "total-data-files": "3",
      "total-delete-files": "0",
      "total-equality-deletes": "0",
      "total-files-size": "14544",
      "total-position-deletes": "0",
      "total-records": "9"
    },
    "schema-id": 0
  }
],
```

```json
avro-tools tojson /tmp/merge-manifests/metadata/snap-7590986702594961103-0-94fb271e-d83d-45e6-88d8-ba6fc60e7669.avro | jq .
{
  "manifest_path": "/tmp/merge-manifests/metadata/94fb271e-d83d-45e6-88d8-ba6fc60e7669-m2.avro",
  "manifest_length": 3827,
  "partition_spec_id": 0,
  "added_snapshot_id": 7590986702594961103,
  "added_files_count": {
    "int": 1
  },
  "existing_files_count": {
    "int": 2
  },
  "deleted_files_count": {
    "int": 0
  },
  "partitions": {
    "array": []
  },
  "added_rows_count": {
    "long": 3
  },
  "existing_rows_count": {
    "long": 6
  },
  "deleted_rows_count": {
    "long": 0
  },
  "key_metadata": null
}
```

```json
{
  "status": 1,
  "snapshot_id": 7590986702594961103,
  "data_file": {
    "file_path": "/tmp/merge-manifests/data/00000-0-94fb271e-d83d-45e6-88d8-ba6fc60e7669.parquet",
    "file_format": "PARQUET",
    "record_count": 3,
    "file_size_in_bytes": 4848,
    "block_size_in_bytes": 0,
    "column_sizes": { ... },
    "value_counts": { ... },
    "null_value_counts": { ... },
    "nan_value_counts": { ... },
    "lower_bounds": { ... },
    "upper_bounds": { ... },
    "key_metadata": null,
    "split_offsets": {
      "array": [
        4
      ]
    },
    "sort_order_id": null,
    "partition": {}
  }
}
{
  "status": 0,
  "snapshot_id": 7590986702594961103,
  "data_file": {
    "file_path": "/tmp/merge-manifests/data/00000-0-0208f3f6-8436-4c8c-802b-9565fbb68f15.parquet",
    "file_format": "PARQUET",
    "record_count": 3,
    "file_size_in_bytes": 4848,
    "block_size_in_bytes": 0,
    "column_sizes": { ... },
    "value_counts": { ... },
    "null_value_counts": { ... },
    "nan_value_counts": { ... },
    "lower_bounds": { ... },
    "upper_bounds": { ... },
    "key_metadata": null,
    "split_offsets": {
      "array": [
        4
      ]
    },
    "sort_order_id": null,
    "partition": {}
  }
}
{
  "status": 0,
  "snapshot_id": 7590986702594961103,
  "data_file": {
    "file_path": "/tmp/merge-manifests/data/00000-0-23f676b6-17d5-43a5-aa58-11a7579a0ad6.parquet",
    "file_format": "PARQUET",
    "record_count": 3,
    "file_size_in_bytes": 4848,
    "block_size_in_bytes": 0,
    "column_sizes": { ... },
    "value_counts": { ... },
    "null_value_counts": { ... },
    "nan_value_counts": { ... },
    "lower_bounds": { ... },
    "upper_bounds": { ... },
    "key_metadata": null,
    "split_offsets": {
      "array": [
        4
      ]
    },
    "sort_order_id": null,
    "partition": {}
  }
}
```

Tne snapshot-ID should be kept from when the manifest was added, so we should see three different ones here:

![image](https://github.com/user-attachments/assets/6fbe7196-9662-47a8-b744-aebeb35d52e7)



Fokko: Also, V2 tables need a bit of TLC. I've added the comments using a `#`:

```json
{
  "status": 1,
  "snapshot_id": {
    "long": 3987426542168867673  # The snapshot should be inherited
  },
  "sequence_number": {
    "long": -1  # This should be null, since status is added, it should be inherited like file_sequence_number
  },
  "file_sequence_number": null,
  "data_file": {
    "content": 0,
    "file_path": "/tmp/merge-manifests/data/00000-0-8d0f1dd8-0398-4143-bb08-ff49c710311e.parquet",
    "file_format": "PARQUET",
    "record_count": 3,
    "file_size_in_bytes": 4848,
    "column_sizes": { ... },
    "value_counts": { ... },
    "null_value_counts": { ... },
    "nan_value_counts": { ... },
    "lower_bounds": { ... },
    "upper_bounds": { ... },
    "key_metadata": null,
    "split_offsets": {
      "array": [
        4
      ]
    },
    "equality_ids": null,
    "sort_order_id": null,
    "partition": {}
  }
}
{
  "status": 0,
  "snapshot_id": {
    "long": 3987426542168867673. # Similar to the issue above, this should be the snapshot-id when the file was added
  },
  "sequence_number": {
    "long": -1  # This should be 2, with status existing, snapshot inheritance is not possible
  },
  "file_sequence_number": {
    "long": 2
  },
  "data_file": {
    "content": 0,
    "file_path": "/tmp/merge-manifests/data/00000-0-a1142919-65c2-4eeb-8345-0ecdb5210a26.parquet",
    "file_format": "PARQUET",
    "record_count": 3,
    "file_size_in_bytes": 4848,
    "column_sizes": { ... },
    "value_counts": { ... },
    "null_value_counts": { ... },
    "nan_value_counts": { ... },
    "lower_bounds": { ... },
    "upper_bounds": { ... },
    "key_metadata": null,
    "split_offsets": {
      "array": [
        4
      ]
    },
    "equality_ids": null,
    "sort_order_id": null,
    "partition": {}
  }
}
{
  "status": 0,
  "snapshot_id": {
    "long": 3987426542168867673
  },
  "sequence_number": {
    "long": 1
  },
  "file_sequence_number": {
    "long": 1
  },
  "data_file": {
    "content": 0,
    "file_path": "/tmp/merge-manifests/data/00000-0-4bbe328f-e532-4931-b008-adca58fbb51d.parquet",
    "file_format": "PARQUET",
    "record_count": 3,
    "file_size_in_bytes": 4848,
    "column_sizes": { ... },
    "value_counts": { ... },
    "null_value_counts": { ... },
    "nan_value_counts": { ... },
    "lower_bounds": { ... },
    "upper_bounds": { ... },
    "key_metadata": null,
    "split_offsets": {
      "array": [
        4
      ]
    },
    "equality_ids": null,
    "sort_order_id": null,
    "partition": {}
  }
}
```

It would be good to have some tests here as well
zeroshade: @Fokko Updated from your feedback and added to the `TestMergeManifests` test so that it verifies the snapshot IDs and sequence numbers after the merging occurs

Patch:
diff --git a/README.md b/README.md
index 2bfe7bb..2837614 100644
--- a/README.md
+++ b/README.md
@@ -66,8 +66,8 @@ $ cd iceberg-go/cmd/iceberg && go build .
 | List Tables              |  X   |      |          |  X   |  X  |
 | Create Table             |  X   |      |          |  X   |  X  |
 | Register Table           |  X   |      |          |  X   |     |
-| Update Current Snapshot  |  X   |      |          |      |     |
-| Create New Snapshot      |  X   |      |          |      |     |
+| Update Current Snapshot  |  X   |      |          |      |  X  |
+| Create New Snapshot      |  X   |      |          |      |  X  |
 | Rename Table             |  X   |      |          |  X   |  X  |
 | Drop Table               |  X   |      |          |  X   |  X  |
 | Alter Table              |  X   |      |          |      |  X  |
@@ -91,7 +91,7 @@ the table, the following tracks the current write support:
 
 | Operation         |Supported|
 |:-----------------:|:-------:|
-| Append Stream     |         |
+| Append Stream     |   X     |
 | Append Data Files |   X     |
 | Rewrite Files     |         |
 | Rewrite manifests |         |
diff --git a/internal/utils.go b/internal/utils.go
index d29ab8a..09d3b6a 100644
--- a/internal/utils.go
+++ b/internal/utils.go
@@ -21,6 +21,7 @@ import (
 	"cmp"
 	"fmt"
 	"io"
+	"iter"
 	"slices"
 )
 
@@ -54,13 +55,7 @@ func (b *Bin[T]) Add(item T, weight int64) {
 	b.items = append(b.items, item)
 }
 
-type SlicePacker[T any] struct {
-	TargetWeight    int64
-	Lookback        int
-	LargestBinFirst bool
-}
-
-func (s *SlicePacker[T]) Pack(items []T, weightFunc func(T) int64) [][]T {
+func PackingIterator[T any](itr iter.Seq[T], targetWeight int64, lookback int, weightFunc func(T) int64, largestBinFirst bool) iter.Seq[[]T] {
 	bins := make([]Bin[T], 0)
 	findBin := func(weight int64) *Bin[T] {
 		for i := range bins {
@@ -73,7 +68,7 @@ func (s *SlicePacker[T]) Pack(items []T, weightFunc func(T) int64) [][]T {
 	}
 
 	removeBin := func() Bin[T] {
-		if s.LargestBinFirst {
+		if largestBinFirst {
 			maxBin := slices.MaxFunc(bins, func(a, b Bin[T]) int {
 				return cmp.Compare(a.Weight(), b.Weight())
 			})
@@ -92,18 +87,18 @@ func (s *SlicePacker[T]) Pack(items []T, weightFunc func(T) int64) [][]T {
 		return out
 	}
 
-	return slices.Collect(func(yield func([]T) bool) {
-		for _, item := range items {
+	return func(yield func([]T) bool) {
+		for item := range itr {
 			w := weightFunc(item)
 			bin := findBin(w)
 			if bin != nil {
 				bin.Add(item, w)
 			} else {
-				bin := Bin[T]{targetWeight: s.TargetWeight}
+				bin := Bin[T]{targetWeight: targetWeight}
 				bin.Add(item, w)
 				bins = append(bins, bin)
 
-				if len(bins) > s.Lookback {
+				if len(bins) > lookback {
 					if !yield(removeBin().items) {
 						return
 					}
@@ -116,7 +111,18 @@ func (s *SlicePacker[T]) Pack(items []T, weightFunc func(T) int64) [][]T {
 				return
 			}
 		}
-	})
+	}
+}
+
+type SlicePacker[T any] struct {
+	TargetWeight    int64
+	Lookback        int
+	LargestBinFirst bool
+}
+
+func (s *SlicePacker[T]) Pack(items []T, weightFunc func(T) int64) [][]T {
+	return slices.Collect(PackingIterator(slices.Values(items), s.TargetWeight,
+		s.Lookback, weightFunc, s.LargestBinFirst))
 }
 
 func (s *SlicePacker[T]) PackEnd(items []T, weightFunc func(T) int64) [][]T {
@@ -155,3 +161,14 @@ func RecoverError(err *error) {
 		}
 	}
 }
+
+func Counter(start int) iter.Seq[int] {
+	return func(yield func(int) bool) {
+		for {
+			if !yield(start) {
+				return
+			}
+			start++
+		}
+	}
+}
diff --git a/manifest.go b/manifest.go
index ed921c0..a63c6c0 100644
--- a/manifest.go
+++ b/manifest.go
@@ -660,6 +660,13 @@ type v1writerImpl struct{}
 
 func (v1writerImpl) content() ManifestContent { return ManifestContentData }
 func (v1writerImpl) prepareEntry(entry *manifestEntry, sn int64) (ManifestEntry, error) {
+	if entry.Snapshot != nil && *entry.Snapshot != sn {
+		if entry.EntryStatus != EntryStatusEXISTING {
+			return nil, fmt.Errorf("mismatched snapshot id for entry: %d vs %d", *entry.Snapshot, sn)
+		}
+		sn = *entry.Snapshot
+	}
+
 	return &fallbackManifestEntry{
 		manifestEntry: *entry,
 		Snapshot:      sn,
@@ -1018,7 +1025,8 @@ func (w *ManifestWriter) Delete(entry ManifestEntry) error {
 }
 
 func (w *ManifestWriter) Existing(entry ManifestEntry) error {
-	w.reusedEntry.wrap(EntryStatusEXISTING, &w.snapshotID, entry.(*manifestEntry).SeqNum, entry.FileSequenceNum(), entry.DataFile())
+	snapshotID := entry.SnapshotID()
+	w.reusedEntry.wrap(EntryStatusEXISTING, &snapshotID, entry.(*manifestEntry).SeqNum, entry.FileSequenceNum(), entry.DataFile())
 
 	return w.addEntry(&w.reusedEntry)
 }
@@ -1522,18 +1530,24 @@ func (m *manifestEntry) inherit(manifest ManifestFile) {
 	}
 
 	manifestSequenceNum := manifest.SequenceNum()
-	if m.SeqNum == nil && (manifestSequenceNum == initialSequenceNumber || m.EntryStatus == EntryStatusADDED) {
-		m.SeqNum = &manifestSequenceNum
-	}
+	if manifestSequenceNum != -1 {
+		if m.SeqNum == nil && (manifestSequenceNum == initialSequenceNumber || m.EntryStatus == EntryStatusADDED) {
+			m.SeqNum = &manifestSequenceNum
+		}
 
-	if m.FileSeqNum == nil && (manifestSequenceNum == initialSequenceNumber || m.EntryStatus == EntryStatusADDED) {
-		m.FileSeqNum = &manifestSequenceNum
+		if m.FileSeqNum == nil && (manifestSequenceNum == initialSequenceNumber || m.EntryStatus == EntryStatusADDED) {
+			m.FileSeqNum = &manifestSequenceNum
+		}
 	}
 
 	m.Data.(*dataFile).specID = manifest.PartitionSpecID()
 }
 
 func (m *manifestEntry) wrap(status ManifestEntryStatus, newSnapID, newSeq, newFileSeq *int64, data DataFile) ManifestEntry {
+	if newSeq != nil && *newSeq == -1 {
+		newSeq = nil
+	}
+
 	m.EntryStatus = status
 	m.Snapshot = newSnapID
 	m.SeqNum = newSeq
diff --git a/partitions.go b/partitions.go
index 84f2727..8bfc957 100644
--- a/partitions.go
+++ b/partitions.go
@@ -163,7 +163,7 @@ func (ps *PartitionSpec) ID() int                    { return ps.id }
 func (ps *PartitionSpec) NumFields() int             { return len(ps.fields) }
 func (ps *PartitionSpec) Field(i int) PartitionField { return ps.fields[i] }
 
-func (ps *PartitionSpec) IsUnpartitioned() bool {
+func (ps PartitionSpec) IsUnpartitioned() bool {
 	if len(ps.fields) == 0 {
 		return true
 	}
diff --git a/schema.go b/schema.go
index e32918e..0dcab76 100644
--- a/schema.go
+++ b/schema.go
@@ -25,6 +25,7 @@ import (
 	"strings"
 	"sync"
 	"sync/atomic"
+	"unicode"
 )
 
 // Schema is an Iceberg table schema, represented as a struct with
@@ -1430,3 +1431,103 @@ func visitTypeWithPartner[T, P any](t Type, fieldPartner P, visitor SchemaWithPa
 		return visitor.Primitive(t.(PrimitiveType), fieldPartner)
 	}
 }
+
+func makeCompatibleName(n string) string {
+	if !validAvroName(n) {
+		return sanitizeName(n)
+	}
+
+	return n
+}
+
+func validAvroName(n string) bool {
+	if len(n) == 0 {
+		panic("cannot validate empty name")
+	}
+
+	if !unicode.IsLetter(rune(n[0])) && n[0] != '_' {
+		return false
+	}
+
+	for _, r := range n[1:] {
+		if !unicode.In(r, unicode.Number, unicode.Letter) && r != '_' {
+			return false
+		}
+	}
+
+	return true
+}
+
+func sanitize(r rune) string {
+	if unicode.IsDigit(r) {
+		return "_" + string(r)
+	}
+
+	return fmt.Sprintf("_x%X", r)
+}
+
+func sanitizeName(n string) string {
+	var b strings.Builder
+	b.Grow(len(n))
+
+	first := n[0]
+	if !(unicode.IsLetter(rune(first)) || first == '_') {
+		b.WriteString(sanitize(rune(first)))
+	} else {
+		b.WriteByte(first)
+	}
+
+	for _, r := range n[1:] {
+		if !unicode.In(r, unicode.Number, unicode.Letter) && r != '_' {
+			b.WriteString(sanitize(r))
+		} else {
+			b.WriteRune(r)
+		}
+	}
+
+	return b.String()
+}
+
+func SanitizeColumnNames(sc *Schema) (*Schema, error) {
+	result, err := Visit(sc, sanitizeColumnNameVisitor{})
+	if err != nil {
+		return nil, err
+	}
+
+	return NewSchemaWithIdentifiers(sc.ID, sc.IdentifierFieldIDs,
+		result.Type.(*StructType).FieldList...), nil
+}
+
+type sanitizeColumnNameVisitor struct{}
+
+func (sanitizeColumnNameVisitor) Schema(_ *Schema, structResult NestedField) NestedField {
+	return structResult
+}
+
+func (sanitizeColumnNameVisitor) Field(field NestedField, fieldResult NestedField) NestedField {
+	field.Type = fieldResult.Type
+	field.Name = makeCompatibleName(field.Name)
+
+	return field
+}
+
+func (sanitizeColumnNameVisitor) Struct(_ StructType, fieldResults []NestedField) NestedField {
+	return NestedField{Type: &StructType{FieldList: fieldResults}}
+}
+
+func (sanitizeColumnNameVisitor) List(list ListType, elemResult NestedField) NestedField {
+	list.Element = elemResult.Type
+
+	return NestedField{Type: &list}
+}
+
+func (sanitizeColumnNameVisitor) Map(mapType MapType, keyResult, valueResult NestedField) NestedField {
+	mapType.KeyType = keyResult.Type
+	mapType.ValueType = valueResult.Type
+
+	return NestedField{Type: &mapType}
+}
+
+func (sanitizeColumnNameVisitor) Primitive(p PrimitiveType) NestedField {
+	return NestedField{Type: p}
+}
diff --git a/table/arrow_utils.go b/table/arrow_utils.go
index 3271cc2..4743175 100644
--- a/table/arrow_utils.go
+++ b/table/arrow_utils.go
@@ -34,6 +34,7 @@ import (
 	"github.com/apache/iceberg-go/internal"
 	iceio "github.com/apache/iceberg-go/io"
 	tblutils "github.com/apache/iceberg-go/table/internal"
+	"github.com/google/uuid"
 	"github.com/pterm/pterm"
 )
 
@@ -1153,9 +1154,9 @@ func filesToDataFiles(ctx context.Context, fileIO iceio.IO, meta *MetadataBuilde
 			if r := recover(); r != nil {
 				switch e := r.(type) {
 				case string:
-					yield(nil, fmt.Errorf("error encountered during parquet file conversion: %s", e))
+					yield(nil, fmt.Errorf("error encountered during file conversion: %s", e))
 				case error:
-					yield(nil, fmt.Errorf("error encountered during parquet file conversion: %w", e))
+					yield(nil, fmt.Errorf("error encountered during file conversion: %w", e))
 				}
 			}
 		}()
@@ -1192,3 +1193,93 @@ func filesToDataFiles(ctx context.Context, fileIO iceio.IO, meta *MetadataBuilde
 		}
 	}
 }
+
+func recordNBytes(rec arrow.Record) (total int64) {
+	for _, c := range rec.Columns() {
+		total += int64(c.Data().SizeInBytes())
+	}
+
+	return total
+}
+
+func binPackRecords(itr iter.Seq2[arrow.Record, error], recordLookback int, targetFileSize int64) iter.Seq[[]arrow.Record] {
+	return internal.PackingIterator(func(yield func(arrow.Record) bool) {
+		for rec, err := range itr {
+			if err != nil {
+				panic(err)
+			}
+
+			rec.Retain()
+			if !yield(rec) {
+				return
+			}
+		}
+	}, targetFileSize, recordLookback, recordNBytes, false)
+}
+
+type recordWritingArgs struct {
+	sc        *arrow.Schema
+	itr       iter.Seq2[arrow.Record, error]
+	fs        iceio.WriteFileIO
+	writeUUID *uuid.UUID
+	counter   iter.Seq[int]
+}
+
+func recordsToDataFiles(ctx context.Context, rootLocation string, meta *MetadataBuilder, args recordWritingArgs) (ret iter.Seq2[iceberg.DataFile, error]) {
+	if args.counter == nil {
+		args.counter = internal.Counter(0)
+	}
+
+	defer func() {
+		if r := recover(); r != nil {
+			var err error
+			switch e := r.(type) {
+			case string:
+				err = fmt.Errorf("error encountered during file writing %s", e)
+			case error:
+				err = fmt.Errorf("error encountered during file writing: %w", e)
+			}
+			ret = func(yield func(iceberg.DataFile, error) bool) {
+				yield(nil, err)
+			}
+		}
+	}()
+
+	if args.writeUUID == nil {
+		u := uuid.Must(uuid.NewRandom())
+		args.writeUUID = &u
+	}
+
+	targetFileSize := int64(meta.props.GetInt(WriteTargetFileSizeBytesKey,
+		WriteTargetFileSizeBytesDefault))
+
+	nameMapping := meta.CurrentSchema().NameMapping()
+	taskSchema, err := ArrowSchemaToIceberg(args.sc, false, nameMapping)
+	if err != nil {
+		panic(err)
+	}
+
+	nextCount, stopCount := iter.Pull(args.counter)
+	if meta.CurrentSpec().IsUnpartitioned() {
+		tasks := func(yield func(WriteTask) bool) {
+			defer stopCount()
+
+			for batch := range binPackRecords(args.itr, 20, targetFileSize) {
+				cnt, _ := nextCount()
+				t := WriteTask{
+					Uuid:    *args.writeUUID,
+					ID:      cnt,
+					Schema:  taskSchema,
+					Batches: batch,
+				}
+				if !yield(t) {
+					return
+				}
+			}
+		}
+
+		return writeFiles(ctx, rootLocation, args.fs, meta, tasks)
+	}
+
+	panic(fmt.Errorf("%w: write stream with partitions", iceberg.ErrNotImplemented))
+}
diff --git a/table/internal/interfaces.go b/table/internal/interfaces.go
index 720e80c..d7c1717 100644
--- a/table/internal/interfaces.go
+++ b/table/internal/interfaces.go
@@ -77,6 +77,8 @@ type FileFormat interface {
 	Open(context.Context, iceio.IO, string) (FileReader, error)
 	PathToIDMapping(*iceberg.Schema) (map[string]int, error)
 	DataFileStatsFromMeta(rdr Metadata, statsCols map[int]StatisticsCollector, colMapping map[string]int) *DataFileStatistics
+	GetWriteProperties(iceberg.Properties) any
+	WriteDataFile(ctx context.Context, fs iceio.WriteFileIO, info WriteFileInfo, batches []arrow.Record) (iceberg.DataFile, error)
 }
 
 func GetFileFormat(format iceberg.FileFormat) FileFormat {
@@ -96,3 +98,11 @@ func FormatFromFileName(fileName string) FileFormat {
 		return nil
 	}
 }
+
+type WriteFileInfo struct {
+	FileSchema *iceberg.Schema
+	Spec       iceberg.PartitionSpec
+	FileName   string
+	StatsCols  map[int]StatisticsCollector
+	WriteProps any
+}
diff --git a/table/internal/parquet_files.go b/table/internal/parquet_files.go
index 2223d5b..c33e820 100644
--- a/table/internal/parquet_files.go
+++ b/table/internal/parquet_files.go
@@ -33,14 +33,36 @@ import (
 	"github.com/apache/arrow-go/v18/arrow/decimal128"
 	"github.com/apache/arrow-go/v18/arrow/memory"
 	"github.com/apache/arrow-go/v18/parquet"
+	"github.com/apache/arrow-go/v18/parquet/compress"
 	"github.com/apache/arrow-go/v18/parquet/file"
 	"github.com/apache/arrow-go/v18/parquet/metadata"
 	"github.com/apache/arrow-go/v18/parquet/pqarrow"
 	"github.com/apache/iceberg-go"
+	"github.com/apache/iceberg-go/internal"
 	iceio "github.com/apache/iceberg-go/io"
 	"github.com/google/uuid"
 )
 
+const (
+	ParquetRowGroupSizeBytesKey              = "write.parquet.row-group-size-bytes"
+	ParquetRowGroupSizeBytesDefault          = 128 * 1024 * 1024 // 128 MB
+	ParquetRowGroupLimitKey                  = "write.parquet.row-group-limit"
+	ParquetRowGroupLimitDefault              = 1048576
+	ParquetPageSizeBytesKey                  = "write.parquet.page-size-bytes"
+	ParquetPageSizeBytesDefault              = 1024 * 1024 // 1 MB
+	ParquetPageRowLimitKey                   = "write.parquet.page-row-limit"
+	ParquetPageRowLimitDefault               = 20000
+	ParquetDictSizeBytesKey                  = "write.parquet.dict-size-bytes"
+	ParquetDictSizeBytesDefault              = 2 * 1024 * 1024 // 2 MB
+	ParquetCompressionKey                    = "write.parquet.compression-codec"
+	ParquetCompressionDefault                = "zstd"
+	ParquetCompressionLevelKey               = "write.parquet.compression-level"
+	ParquetCompressionLevelDefault           = -1
+	ParquetBloomFilterMaxBytesKey            = "write.parquet.bloom-filter-max-bytes"
+	ParquetBloomFilterMaxBytesDefault        = 1024 * 1024
+	ParquetBloomFilterColumnEnabledKeyPrefix = "write.parquet.bloom-filter-enabled.column"
+)
+
 type parquetFormat struct{}
 
 func (parquetFormat) Open(ctx context.Context, fs iceio.IO, path string) (FileReader, error) {
@@ -170,6 +192,91 @@ func (parquetFormat) PrimitiveTypeToPhysicalType(typ iceberg.PrimitiveType) stri
 	}
 }
 
+func (parquetFormat) GetWriteProperties(props iceberg.Properties) any {
+	writerProps := []parquet.WriterProperty{
+		parquet.WithDictionaryDefault(false),
+		parquet.WithMaxRowGroupLength(int64(props.GetInt(ParquetRowGroupLimitKey,
+			ParquetRowGroupLimitDefault))),
+		parquet.WithDataPageSize(int64(props.GetInt(ParquetPageSizeBytesKey,
+			ParquetPageSizeBytesDefault))),
+		parquet.WithDataPageVersion(parquet.DataPageV2),
+		parquet.WithBatchSize(int64(props.GetInt(ParquetPageRowLimitKey,
+			ParquetPageRowLimitDefault))),
+		parquet.WithDictionaryPageSizeLimit(int64(props.GetInt(ParquetDictSizeBytesKey,
+			ParquetDictSizeBytesDefault))),
+	}
+
+	compression := props.Get(ParquetCompressionKey, ParquetCompressionDefault)
+	compressionLevel := props.GetInt(ParquetCompressionLevelKey,
+		ParquetCompressionLevelDefault)
+
+	var codec compress.Compression
+	switch compression {
+	case "snappy":
+		codec = compress.Codecs.Snappy
+	case "zstd":
+		codec = compress.Codecs.Zstd
+	case "uncompressed":
+		codec = compress.Codecs.Uncompressed
+	case "gzip":
+		codec = compress.Codecs.Gzip
+	case "brotli":
+		codec = compress.Codecs.Brotli
+	case "lz4":
+		codec = compress.Codecs.Lz4
+	case "lz4raw":
+		codec = compress.Codecs.Lz4Raw
+	case "lzo":
+		codec = compress.Codecs.Lzo
+	default:
+		// warn
+	}
+
+	return append(writerProps, parquet.WithCompression(codec),
+		parquet.WithCompressionLevel(compressionLevel))
+}
+
+func (p parquetFormat) WriteDataFile(ctx context.Context, fs iceio.WriteFileIO, info WriteFileInfo, batches []arrow.Record) (iceberg.DataFile, error) {
+	fw, err := fs.Create(info.FileName)
+	if err != nil {
+		return nil, err
+	}
+	defer fw.Close()
+
+	cntWriter := internal.CountingWriter{W: fw}
+	mem := compute.GetAllocator(ctx)
+	writerProps := parquet.NewWriterProperties(info.WriteProps.([]parquet.WriterProperty)...)
+	arrProps := pqarrow.NewArrowWriterProperties(pqarrow.WithAllocator(mem), pqarrow.WithStoreSchema())
+
+	writer, err := pqarrow.NewFileWriter(batches[0].Schema(), &cntWriter, writerProps, arrProps)
+	if err != nil {
+		return nil, err
+	}
+
+	for _, batch := range batches {
+		if err := writer.WriteBuffered(batch); err != nil {
+			return nil, err
+		}
+	}
+
+	if err := writer.Close(); err != nil {
+		return nil, err
+	}
+
+	filemeta, err := writer.FileMetadata()
+	if err != nil {
+		return nil, err
+	}
+
+	colMapping, err := p.PathToIDMapping(info.FileSchema)
+	if err != nil {
+		return nil, err
+	}
+
+	return p.DataFileStatsFromMeta(filemeta, info.StatsCols, colMapping).
+		ToDataFile(info.FileSchema, info.Spec, info.FileName, iceberg.ParquetFile, cntWriter.Count), nil
+}
+
 type decAsIntAgg[T int32 | int64] struct {
 	*statsAggregator[T]
 }
diff --git a/table/internal/utils.go b/table/internal/utils.go
index 3e67d8b..00c1785 100644
--- a/table/internal/utils.go
+++ b/table/internal/utils.go
@@ -448,7 +448,7 @@ func (s *statsAggregator[T]) MaxAsBytes() ([]byte, error) {
 }
 
 func TruncateUpperBoundText(s string, trunc int) string {
-	if trunc == utf8.RuneCountInString(s) {
+	if trunc >= utf8.RuneCountInString(s) {
 		return s
 	}
 
@@ -466,6 +466,10 @@ func TruncateUpperBoundText(s string, trunc int) string {
 }
 
 func TruncateUpperBoundBinary(val []byte, trunc int) []byte {
+	if trunc >= len(val) {
+		return val
+	}
+
 	result := val[:trunc]
 	if bytes.Equal(result, val) {
 		return result
@@ -482,15 +486,13 @@ func TruncateUpperBoundBinary(val []byte, trunc int) []byte {
 	return nil
 }
 
-func MapExec[T, S any](nWorkers int, slice []T, fn func(T) (S, error)) iter.Seq2[S, error] {
+func MapExec[T, S any](nWorkers int, slice iter.Seq[T], fn func(T) (S, error)) iter.Seq2[S, error] {
 	if nWorkers <= 0 {
 		nWorkers = runtime.GOMAXPROCS(0)
 	}
 
-	nWorkers = min(nWorkers, len(slice))
-
 	var g errgroup.Group
-	ch := make(chan T, len(slice))
+	ch := make(chan T, nWorkers)
 	out := make(chan S, nWorkers)
 
 	for range nWorkers {
@@ -507,7 +509,7 @@ func MapExec[T, S any](nWorkers int, slice []T, fn func(T) (S, error)) iter.Seq2
 		})
 	}
 
-	for _, v := range slice {
+	for v := range slice {
 		ch <- v
 	}
 	close(ch)
diff --git a/table/locations.go b/table/locations.go
index a95ef78..46478cb 100644
--- a/table/locations.go
+++ b/table/locations.go
@@ -26,6 +26,7 @@ import (
 )
 
 type LocationProvider interface {
+	NewDataLocation(dataFileName string) string
 	NewTableMetadataFileLocation(newVersion int) (string, error)
 	NewMetadataLocation(metadataFileName string) string
 }
@@ -37,6 +38,10 @@ type simpleLocationProvider struct {
 	metadataPath *url.URL
 }
 
+func (slp *simpleLocationProvider) NewDataLocation(dataFileName string) string {
+	return slp.dataPath.JoinPath(dataFileName).String()
+}
+
 func (slp *simpleLocationProvider) NewTableMetadataFileLocation(newVersion int) (string, error) {
 	if newVersion < 0 {
 		return "", fmt.Errorf("%w: table metadata version %d must be a non-negative integer",
diff --git a/table/properties.go b/table/properties.go
index 5639b98..5f44224 100644
--- a/table/properties.go
+++ b/table/properties.go
@@ -17,6 +17,8 @@
 
 package table
 
+import "github.com/apache/iceberg-go/table/internal"
+
 const (
 	WriteDataPathKey                        = "write.data.path"
 	WriteMetadataPathKey                    = "write.metadata.path"
@@ -31,6 +33,24 @@ const (
 	DefaultWriteMetricsModeKey     = "write.metadata.metrics.default"
 	DefaultWriteMetricsModeDefault = "truncate(16)"
 
+	ParquetRowGroupSizeBytesKey              = internal.ParquetRowGroupSizeBytesKey
+	ParquetRowGroupSizeBytesDefault          = internal.ParquetRowGroupSizeBytesDefault
+	ParquetRowGroupLimitKey                  = internal.ParquetRowGroupLimitKey
+	ParquetRowGroupLimitDefault              = internal.ParquetRowGroupLimitDefault
+	ParquetPageSizeBytesKey                  = internal.ParquetPageSizeBytesKey
+	ParquetPageSizeBytesDefault              = internal.ParquetPageSizeBytesDefault
+	ParquetPageRowLimitKey                   = internal.ParquetPageRowLimitKey
+	ParquetPageRowLimitDefault               = internal.ParquetPageRowLimitDefault
+	ParquetDictSizeBytesKey                  = internal.ParquetDictSizeBytesKey
+	ParquetDictSizeBytesDefault              = internal.ParquetDictSizeBytesDefault
+	ParquetCompressionKey                    = internal.ParquetCompressionKey
+	ParquetCompressionDefault                = internal.ParquetCompressionDefault
+	ParquetCompressionLevelKey               = internal.ParquetCompressionLevelKey
+	ParquetCompressionLevelDefault           = internal.ParquetCompressionLevelDefault
+	ParquetBloomFilterMaxBytesKey            = internal.ParquetBloomFilterMaxBytesKey
+	ParquetBloomFilterMaxBytesDefault        = internal.ParquetBloomFilterMaxBytesDefault
+	ParquetBloomFilterColumnEnabledKeyPrefix = internal.ParquetBloomFilterColumnEnabledKeyPrefix
+
 	ManifestMergeEnabledKey     = "commit.manifest-merge.enabled"
 	ManifestMergeEnabledDefault = false
 
@@ -48,4 +68,7 @@ const (
 
 	MetadataPreviousVersionsMaxKey     = "write.metadata.previous-versions-max"
 	MetadataPreviousVersionsMaxDefault = 100
+
+	WriteTargetFileSizeBytesKey     = "write.target-file-size-bytes"
+	WriteTargetFileSizeBytesDefault = 512 * 1024 * 1024 // 512 MB
 )
diff --git a/table/snapshot_producers.go b/table/snapshot_producers.go
index 7519c9a..1a5d6fd 100644
--- a/table/snapshot_producers.go
+++ b/table/snapshot_producers.go
@@ -227,7 +227,7 @@ func (of *overwriteFiles) deletedEntries() ([]iceberg.ManifestEntry, error) {
 
 	nWorkers := config.EnvConfig.MaxWorkers
 	finalResult := make([]iceberg.ManifestEntry, 0, len(previousManifests))
-	for entries, err := range tblutils.MapExec(nWorkers, previousManifests, getEntries) {
+	for entries, err := range tblutils.MapExec(nWorkers, slices.Values(previousManifests), getEntries) {
 		if err != nil {
 			return nil, err
 		}
@@ -237,6 +237,171 @@ func (of *overwriteFiles) deletedEntries() ([]iceberg.ManifestEntry, error) {
 	return finalResult, nil
 }
 
+type manifestMergeManager struct {
+	targetSizeBytes int
+	minCountToMerge int
+	mergeEnabled    bool
+	snap            *snapshotProducer
+}
+
+func (m *manifestMergeManager) groupBySpec(manifests []iceberg.ManifestFile) map[int][]iceberg.ManifestFile {
+	groups := make(map[int][]iceberg.ManifestFile)
+	for _, m := range manifests {
+		specid := int(m.PartitionSpecID())
+		group := groups[specid]
+		groups[specid] = append(group, m)
+	}
+
+	return groups
+}
+
+func (m *manifestMergeManager) createManifest(specID int, bin []iceberg.ManifestFile) (iceberg.ManifestFile, error) {
+	wr, path, counter, err := m.snap.newManifestWriter(m.snap.spec(specID))
+	if err != nil {
+		return nil, err
+	}
+	defer counter.W.(io.Closer).Close()
+
+	for _, manifest := range bin {
+		entries, err := m.snap.fetchManifestEntry(manifest, false)
+		if err != nil {
+			return nil, err
+		}
+
+		for _, entry := range entries {
+			switch {
+			case entry.Status() == iceberg.EntryStatusDELETED && entry.SnapshotID() == m.snap.snapshotID:
+				// only files deleted by this snapshot should be added to the new manifest
+				wr.Delete(entry)
+			case entry.Status() == iceberg.EntryStatusADDED && entry.SnapshotID() == m.snap.snapshotID:
+				// added entries from this snapshot are still added, otherwise they should be existing
+				wr.Add(entry)
+			case entry.Status() != iceberg.EntryStatusDELETED:
+				// add all non-deleted files from the old manifest as existing files
+				wr.Existing(entry)
+			}
+		}
+	}
+
+	return wr.ToManifestFile(path, counter.Count)
+}
+
+func (m *manifestMergeManager) mergeGroup(firstManifest iceberg.ManifestFile, specID int, manifests []iceberg.ManifestFile) ([]iceberg.ManifestFile, error) {
+	packer := internal.SlicePacker[iceberg.ManifestFile]{
+		TargetWeight:    int64(m.targetSizeBytes),
+		Lookback:        1,
+		LargestBinFirst: false,
+	}
+	bins := packer.PackEnd(manifests, func(m iceberg.ManifestFile) int64 {
+		return m.Length()
+	})
+
+	mergeBin := func(bin []iceberg.ManifestFile) ([]iceberg.ManifestFile, error) {
+		output := make([]iceberg.ManifestFile, 0, 1)
+		if len(bin) == 1 {
+			output = append(output, bin[0])
+		} else if len(bin) < m.minCountToMerge && slices.ContainsFunc(bin, func(m iceberg.ManifestFile) bool { return m == firstManifest }) {
+			// if the bin has the first manifest (the new data files or an appended
+			// manifest file) then only merge it if the number of manifests is above
+			// the minimum count. this is applied only to bins with an in-memory manifest
+			// so that large manifests don't prevent merging older groups
+			output = append(output, bin...)
+		} else {
+			created, err := m.createManifest(specID, bin)
+			if err != nil {
+				return nil, err
+			}
+			output = append(output, created)
+		}
+
+		return output, nil
+	}
+
+	binResults := make([][]iceberg.ManifestFile, len(bins))
+	g := errgroup.Group{}
+	for i, bin := range bins {
+		i, bin := i, bin
+		g.Go(func() error {
+			var err error
+			binResults[i], err = mergeBin(bin)
+
+			return err
+		})
+	}
+
+	if err := g.Wait(); err != nil {
+		return nil, err
+	}
+
+	return slices.Concat(binResults...), nil
+}
+
+func (m *manifestMergeManager) mergeManifests(manifests []iceberg.ManifestFile) ([]iceberg.ManifestFile, error) {
+	if !m.mergeEnabled || len(manifests) == 0 {
+		return manifests, nil
+	}
+
+	first := manifests[0]
+	groups := m.groupBySpec(manifests)
+
+	merged := make([]iceberg.ManifestFile, 0, len(groups))
+	for _, specID := range slices.Backward(slices.Sorted(maps.Keys(groups))) {
+		manifests, err := m.mergeGroup(first, specID, groups[specID])
+		if err != nil {
+			return nil, err
+		}
+
+		merged = append(merged, manifests...)
+	}
+
+	return merged, nil
+}
+
+type mergeAppendFiles struct {
+	fastAppendFiles
+
+	targetSizeBytes int
+	minCountToMerge int
+	mergeEnabled    bool
+}
+
+func newMergeAppendFilesProducer(op Operation, txn *Transaction, fs iceio.WriteFileIO, commitUUID *uuid.UUID, snapshotProps iceberg.Properties) *snapshotProducer {
+	prod := createSnapshotProducer(op, txn, fs, commitUUID, snapshotProps)
+	prod.producerImpl = &mergeAppendFiles{
+		fastAppendFiles: fastAppendFiles{base: prod},
+		targetSizeBytes: txn.meta.props.GetInt(ManifestTargetSizeBytesKey, ManifestTargetSizeBytesDefault),
+		minCountToMerge: txn.meta.props.GetInt(ManifestMinMergeCountKey, ManifestMinMergeCountDefault),
+		mergeEnabled:    txn.meta.props.GetBool(ManifestMergeEnabledKey, ManifestMergeEnabledDefault),
+	}
+
+	return prod
+}
+
+func (m *mergeAppendFiles) processManifests(manifests []iceberg.ManifestFile) ([]iceberg.ManifestFile, error) {
+	unmergedDataManifests, unmergedDeleteManifests := []iceberg.ManifestFile{}, []iceberg.ManifestFile{}
+	for _, m := range manifests {
+		if m.ManifestContent() == iceberg.ManifestContentData {
+			unmergedDataManifests = append(unmergedDataManifests, m)
+		} else if m.ManifestContent() == iceberg.ManifestContentDeletes {
+			unmergedDeleteManifests = append(unmergedDeleteManifests, m)
+		}
+	}
+
+	dataManifestMergeMgr := manifestMergeManager{
+		targetSizeBytes: m.targetSizeBytes,
+		minCountToMerge: m.minCountToMerge,
+		mergeEnabled:    m.mergeEnabled,
+		snap:            m.base,
+	}
+
+	result, err := dataManifestMergeMgr.mergeManifests(unmergedDataManifests)
+	if err != nil {
+		return nil, err
+	}
+
+	return append(result, unmergedDeleteManifests...), nil
+}
+
 type snapshotProducer struct {
 	producerImpl
 
diff --git a/table/table.go b/table/table.go
index 07b8a38..05f40b0 100644
--- a/table/table.go
+++ b/table/table.go
@@ -24,6 +24,8 @@ import (
 	"runtime"
 	"slices"
 
+	"github.com/apache/arrow-go/v18/arrow"
+	"github.com/apache/arrow-go/v18/arrow/array"
 	"github.com/apache/iceberg-go"
 	"github.com/apache/iceberg-go/internal"
 	"github.com/apache/iceberg-go/io"
@@ -88,6 +90,26 @@ func (t Table) NewTransaction() *Transaction {
 	}
 }
 
+// AppendTable is a shortcut for NewTransaction().AppendTable() and then committing the transaction
+func (t Table) AppendTable(ctx context.Context, tbl arrow.Table, batchSize int64, snapshotProps iceberg.Properties) (*Table, error) {
+	txn := t.NewTransaction()
+	if err := txn.AppendTable(ctx, tbl, batchSize, snapshotProps); err != nil {
+		return nil, err
+	}
+
+	return txn.Commit(ctx)
+}
+
+// Append is a shortcut for NewTransaction().Append() and then committing the transaction
+func (t Table) Append(ctx context.Context, rdr array.RecordReader, snapshotProps iceberg.Properties) (*Table, error) {
+	txn := t.NewTransaction()
+	if err := txn.Append(ctx, rdr, snapshotProps); err != nil {
+		return nil, err
+	}
+
+	return txn.Commit(ctx)
+}
+
 func (t Table) AllManifests() iter.Seq2[iceberg.ManifestFile, error] {
 	type list = tblutils.Enumerated[[]iceberg.ManifestFile]
 	g := errgroup.Group{}
diff --git a/table/transaction.go b/table/transaction.go
index 8b6d8cb..26fa844 100644
--- a/table/transaction.go
+++ b/table/transaction.go
@@ -27,6 +27,7 @@ import (
 	"sync"
 	"time"
 
+	"github.com/apache/arrow-go/v18/arrow"
 	"github.com/apache/arrow-go/v18/arrow/array"
 	"github.com/apache/iceberg-go"
 	"github.com/apache/iceberg-go/io"
@@ -52,6 +53,10 @@ func (s snapshotUpdate) mergeOverwrite(commitUUID *uuid.UUID) *snapshotProducer
 	return newOverwriteFilesProducer(op, s.txn, s.io, commitUUID, s.snapshotProps)
 }
 
+func (s snapshotUpdate) mergeAppend() *snapshotProducer {
+	return newMergeAppendFilesProducer(OpAppend, s.txn, s.io, nil, s.snapshotProps)
+}
+
 type Transaction struct {
 	tbl  *Table
 	meta *MetadataBuilder
@@ -111,6 +116,16 @@ func (t *Transaction) apply(updates []Update, reqs []Requirement) error {
 	return nil
 }
 
+func (t *Transaction) appendSnapshotProducer(props iceberg.Properties) *snapshotProducer {
+	manifestMerge := t.meta.props.GetBool(ManifestMergeEnabledKey, ManifestMergeEnabledDefault)
+	updateSnapshot := t.updateSnapshot(props)
+	if manifestMerge {
+		return updateSnapshot.mergeAppend()
+	}
+
+	return updateSnapshot.fastAppend()
+}
+
 func (t *Transaction) updateSnapshot(props iceberg.Properties) snapshotUpdate {
 	return snapshotUpdate{
 		txn:           t,
@@ -127,6 +142,38 @@ func (t *Transaction) SetProperties(props iceberg.Properties) error {
 	return nil
 }
 
+func (t *Transaction) AppendTable(ctx context.Context, tbl arrow.Table, batchSize int64, snapshotProps iceberg.Properties) error {
+	rdr := array.NewTableReader(tbl, batchSize)
+	defer rdr.Release()
+
+	return t.Append(ctx, rdr, snapshotProps)
+}
+
+func (t *Transaction) Append(ctx context.Context, rdr array.RecordReader, snapshotProps iceberg.Properties) error {
+	appendFiles := t.appendSnapshotProducer(snapshotProps)
+
+	itr := recordsToDataFiles(ctx, t.tbl.Location(), t.meta, recordWritingArgs{
+		sc:        rdr.Schema(),
+		itr:       array.IterFromReader(rdr),
+		fs:        t.tbl.fs.(io.WriteFileIO),
+		writeUUID: &appendFiles.commitUuid,
+	})
+
+	for df, err := range itr {
+		if err != nil {
+			return err
+		}
+		appendFiles.appendDataFile(df)
+	}
+
+	updates, reqs, err := appendFiles.commit()
+	if err != nil {
+		return err
+	}
+
+	return t.apply(updates, reqs)
+}
+
 // ReplaceFiles is actually just an overwrite operation with multiple
 // files deleted and added.
 //
@@ -222,10 +269,6 @@ func (t *Transaction) ReplaceDataFiles(ctx context.Context, filesToDelete, files
 	return t.apply(updates, reqs)
 }
 
-func (t *Transaction) Append(rdr array.RecordReader, snapshotProps iceberg.Properties) error {
-	return iceberg.ErrNotImplemented
-}
-
 func (t *Transaction) AddFiles(ctx context.Context, files []string, snapshotProps iceberg.Properties, ignoreDuplicates bool) error {
 	set := make(map[string]string)
 	for _, f := range files {
diff --git a/table/writer.go b/table/writer.go
new file mode 100644
index 0000000..d4e9ae7
--- /dev/null
+++ b/table/writer.go
@@ -0,0 +1,127 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package table
+
+import (
+	"context"
+	"fmt"
+	"iter"
+
+	"github.com/apache/arrow-go/v18/arrow"
+	"github.com/apache/iceberg-go"
+	"github.com/apache/iceberg-go/config"
+	"github.com/apache/iceberg-go/io"
+	"github.com/apache/iceberg-go/table/internal"
+	"github.com/google/uuid"
+)
+
+type WriteTask struct {
+	Uuid        uuid.UUID
+	ID          int
+	Schema      *iceberg.Schema
+	Batches     []arrow.Record
+	SortOrderID int
+}
+
+func (w WriteTask) GenerateDataFileName(extension string) string {
+	// Mimics the behavior in the Java API:
+	// https://github.com/apache/iceberg/blob/a582968975dd30ff4917fbbe999f1be903efac02/core/src/main/java/org/apache/iceberg/io/OutputFileFactory.java#L92-L101
+	return fmt.Sprintf("00000-%d-%s.%s", w.ID, w.Uuid, extension)
+}
+
+type writer struct {
+	loc        LocationProvider
+	fs         io.WriteFileIO
+	fileSchema *iceberg.Schema
+	format     internal.FileFormat
+	props      any
+	meta       *MetadataBuilder
+}
+
+func (w *writer) writeFile(ctx context.Context, task WriteTask) (iceberg.DataFile, error) {
+	defer func() {
+		for _, b := range task.Batches {
+			b.Release()
+		}
+	}()
+
+	batches := make([]arrow.Record, len(task.Batches))
+	for i, b := range task.Batches {
+		rec, err := ToRequestedSchema(ctx, w.fileSchema,
+			task.Schema, b, false, true, false)
+		if err != nil {
+			return nil, err
+		}
+		batches[i] = rec
+	}
+
+	statsCols, err := computeStatsPlan(w.fileSchema, w.meta.props)
+	if err != nil {
+		return nil, err
+	}
+
+	filePath := w.loc.NewDataLocation(
+		task.GenerateDataFileName("parquet"))
+
+	return w.format.WriteDataFile(ctx, w.fs, internal.WriteFileInfo{
+		FileSchema: w.fileSchema,
+		FileName:   filePath,
+		StatsCols:  statsCols,
+		WriteProps: w.props,
+	}, batches)
+}
+
+func writeFiles(ctx context.Context, rootLocation string, fs io.WriteFileIO, meta *MetadataBuilder, tasks iter.Seq[WriteTask]) iter.Seq2[iceberg.DataFile, error] {
+	locProvider, err := LoadLocationProvider(rootLocation, meta.props)
+	if err != nil {
+		return func(yield func(iceberg.DataFile, error) bool) {
+			yield(nil, err)
+		}
+	}
+
+	format := internal.GetFileFormat(iceberg.ParquetFile)
+	fileSchema := meta.CurrentSchema()
+	sanitized, err := iceberg.SanitizeColumnNames(fileSchema)
+	if err != nil {
+		return func(yield func(iceberg.DataFile, error) bool) {
+			yield(nil, err)
+		}
+	}
+
+	// if the schema needs to be transformed, use the transformed schema
+	// and adjust the arrow schema appropriately. otherwise we just
+	// use the original schema.
+	if !sanitized.Equals(fileSchema) {
+		fileSchema = sanitized
+	}
+
+	w := &writer{
+		loc:        locProvider,
+		fs:         fs,
+		fileSchema: fileSchema,
+		format:     format,
+		props:      format.GetWriteProperties(meta.props),
+		meta:       meta,
+	}
+
+	nworkers := config.EnvConfig.MaxWorkers
+
+	return internal.MapExec(nworkers, tasks, func(t WriteTask) (iceberg.DataFile, error) {
+		return w.writeFile(ctx, t)
+	})
+}


Test patch:
diff --git a/table/table_test.go b/table/table_test.go
index 4d7eb88..7f5e66a 100644
--- a/table/table_test.go
+++ b/table/table_test.go
@@ -21,7 +21,9 @@ import (
 	"bytes"
 	"context"
 	"fmt"
+	"slices"
 	"strconv"
+	"strings"
 	"testing"
 
 	"github.com/apache/arrow-go/v18/arrow"
@@ -29,12 +31,15 @@ import (
 	"github.com/apache/arrow-go/v18/arrow/memory"
 	"github.com/apache/arrow-go/v18/parquet/pqarrow"
 	"github.com/apache/iceberg-go"
+	"github.com/apache/iceberg-go/catalog"
+	"github.com/apache/iceberg-go/catalog/sql"
 	"github.com/apache/iceberg-go/internal"
 	iceio "github.com/apache/iceberg-go/io"
 	"github.com/apache/iceberg-go/table"
 	"github.com/google/uuid"
 	"github.com/pterm/pterm"
 	"github.com/stretchr/testify/suite"
+	"github.com/uptrace/bun/driver/sqliteshim"
 )
 
 type TableTestSuite struct {
@@ -259,7 +264,7 @@ func (t *TableWritingTestSuite) SetupSuite() {
 }
 
 func (t *TableWritingTestSuite) SetupTest() {
-	t.location = t.T().TempDir()
+	t.location = strings.Replace(t.T().TempDir(), "#", "", -1)
 }
 
 func (t *TableWritingTestSuite) TearDownSuite() {
@@ -835,6 +840,297 @@ func (t *TableWritingTestSuite) TestReplaceDataFiles() {
 	}, staged.CurrentSnapshot().Summary)
 }
 
+func (t *TableWritingTestSuite) TestWriteSpecialCharacterColumn() {
+	ident := table.Identifier{"default", "write_special_character_column"}
+	colNameWithSpecialChar := "letter/abc"
+
+	s := iceberg.NewSchema(0,
+		iceberg.NestedField{ID: 1, Name: colNameWithSpecialChar, Type: iceberg.PrimitiveTypes.String},
+		iceberg.NestedField{ID: 2, Name: "id", Type: iceberg.PrimitiveTypes.Int32},
+		iceberg.NestedField{ID: 3, Name: "name", Type: iceberg.PrimitiveTypes.String, Required: true},
+		iceberg.NestedField{ID: 4, Name: "address", Required: true, Type: &iceberg.StructType{
+			FieldList: []iceberg.NestedField{
+				{ID: 5, Name: "street", Type: iceberg.PrimitiveTypes.String, Required: true},
+				{ID: 6, Name: "city", Type: iceberg.PrimitiveTypes.String, Required: true},
+				{ID: 7, Name: "zip", Type: iceberg.PrimitiveTypes.Int32, Required: true},
+				{ID: 8, Name: colNameWithSpecialChar, Type: iceberg.PrimitiveTypes.String, Required: true},
+			},
+		}})
+
+	arrowSchema := arrow.NewSchema([]arrow.Field{
+		{Name: colNameWithSpecialChar, Type: arrow.BinaryTypes.String, Nullable: true},
+		{Name: "id", Type: arrow.PrimitiveTypes.Int32, Nullable: true},
+		{Name: "name", Type: arrow.BinaryTypes.String},
+		{Name: "address", Type: arrow.StructOf(
+			arrow.Field{Name: "street", Type: arrow.BinaryTypes.String},
+			arrow.Field{Name: "city", Type: arrow.BinaryTypes.String},
+			arrow.Field{Name: "zip", Type: arrow.PrimitiveTypes.Int32},
+			arrow.Field{Name: colNameWithSpecialChar, Type: arrow.BinaryTypes.String},
+		)},
+	}, nil)
+
+	arrowTable, err := array.TableFromJSON(memory.DefaultAllocator, arrowSchema, []string{
+		`[
+			{
+				"letter/abc": "a",
+				"id": 1,
+				"name": "AB",
+				"address": {"street": "123", "city": "SFO", "zip": 12345, "letter/abc": "a"}
+			},
+			{
+				"letter/abc": null,
+				"id": 2,
+				"name": "CD",
+				"address": {"street": "456", "city": "SW", "zip": 67890, "letter/abc": "b"}
+			},
+			{
+				"letter/abc": "z",
+				"id": 3,
+				"name": "EF",
+				"address": {"street": "789", "city": "Random", "zip": 10112, "letter/abc": "c"}
+			}
+		 ]`,
+	})
+	t.Require().NoError(err)
+	defer arrowTable.Release()
+
+	tbl := t.createTable(ident, t.formatVersion, *iceberg.UnpartitionedSpec, s)
+	rdr := array.NewTableReader(arrowTable, 1)
+	defer rdr.Release()
+
+	tx := tbl.NewTransaction()
+	t.Require().NoError(tx.Append(t.ctx, rdr, nil))
+
+	scan, err := tx.Scan()
+	t.Require().NoError(err)
+
+	result, err := scan.ToArrowTable(t.ctx)
+	t.Require().NoError(err)
+	defer result.Release()
+
+	t.True(array.TableEqual(arrowTable, result), "expected:\n %s\ngot:\n %s", arrowTable, result)
+}
+
+func (t *TableWritingTestSuite) getInMemCatalog() catalog.Catalog {
+	cat, err := catalog.Load(context.Background(), "default", iceberg.Properties{
+		"uri":          ":memory:",
+		"type":         "sql",
+		sql.DriverKey:  sqliteshim.ShimName,
+		sql.DialectKey: string(sql.SQLite),
+		"warehouse":    "file://" + t.location,
+	})
+	t.Require().NoError(err)
+
+	return cat
+}
+
+func (t *TableWritingTestSuite) createTableWithProps(identifier table.Identifier, props iceberg.Properties, sc *iceberg.Schema) *table.Table {
+	cat := t.getInMemCatalog()
+	cat.DropTable(t.ctx, identifier)
+	cat.DropNamespace(t.ctx, catalog.NamespaceFromIdent(identifier))
+
+	t.Require().NoError(cat.CreateNamespace(t.ctx, catalog.NamespaceFromIdent(identifier), nil))
+	tbl, err := cat.CreateTable(t.ctx, identifier, sc, catalog.WithProperties(props),
+		catalog.WithLocation(t.location))
+
+	t.Require().NoError(err)
+
+	return tbl
+}
+
+func tableSchema() *iceberg.Schema {
+	return iceberg.NewSchema(0,
+		iceberg.NestedField{ID: 1, Name: "bool", Type: iceberg.PrimitiveTypes.Bool},
+		iceberg.NestedField{ID: 2, Name: "string", Type: iceberg.PrimitiveTypes.String},
+		iceberg.NestedField{ID: 3, Name: "string_long", Type: iceberg.PrimitiveTypes.String},
+		iceberg.NestedField{ID: 4, Name: "int", Type: iceberg.PrimitiveTypes.Int32},
+		iceberg.NestedField{ID: 5, Name: "long", Type: iceberg.PrimitiveTypes.Int64},
+		iceberg.NestedField{ID: 6, Name: "float", Type: iceberg.PrimitiveTypes.Float32},
+		iceberg.NestedField{ID: 7, Name: "double", Type: iceberg.PrimitiveTypes.Float64},
+		iceberg.NestedField{ID: 8, Name: "time", Type: iceberg.PrimitiveTypes.Time},
+		iceberg.NestedField{ID: 9, Name: "timestamp", Type: iceberg.PrimitiveTypes.Timestamp},
+		iceberg.NestedField{ID: 10, Name: "timestamptz", Type: iceberg.PrimitiveTypes.TimestampTz},
+		iceberg.NestedField{ID: 11, Name: "date", Type: iceberg.PrimitiveTypes.Date},
+		iceberg.NestedField{ID: 12, Name: "uuid", Type: iceberg.PrimitiveTypes.UUID},
+		iceberg.NestedField{ID: 13, Name: "binary", Type: iceberg.PrimitiveTypes.Binary},
+		iceberg.NestedField{ID: 14, Name: "fixed", Type: iceberg.FixedTypeOf(16)},
+		iceberg.NestedField{ID: 15, Name: "small_dec", Type: iceberg.DecimalTypeOf(8, 2)},
+		iceberg.NestedField{ID: 16, Name: "med_dec", Type: iceberg.DecimalTypeOf(16, 2)},
+		iceberg.NestedField{ID: 17, Name: "large_dec", Type: iceberg.DecimalTypeOf(24, 2)},
+	)
+}
+
+func arrowTableWithNull() arrow.Table {
+	sc, err := table.SchemaToArrowSchema(tableSchema(), nil, true, false)
+	if err != nil {
+		panic(err)
+	}
+
+	arrTable, err := array.TableFromJSON(memory.DefaultAllocator, sc, []string{
+		`[
+			{
+				"bool": false,
+				"string": "a",
+				"string_long": "` + strings.Repeat("a", 22) + `",
+				"int": 1,
+				"long": 1,
+				"float": 0.0,
+				"double": 0.0,
+				"time": "00:00:01.000000",
+				"timestamp": "2023-01-01T19:25:00.000000+08:00",
+				"timestamptz": "2023-01-01T19:25:00.000000Z",
+				"date": "2023-01-01",
+				"uuid": "00000000-0000-0000-0000-000000000000",
+				"binary": "AQ==",
+				"fixed": "AAAAAAAAAAAAAAAAAAAAAA==",
+				"small_dec": "123456.78",
+				"med_dec": "12345678901234.56",
+				"large_dec": "1234567890123456789012.34"
+			},
+			{
+				"bool": null,
+				"string": null,
+				"string_long": null,
+				"int": null,
+				"long": null,
+				"float": null,
+				"double": null,
+				"time": null,
+				"timestamp": null,
+				"timestamptz": null,
+				"date": null,
+				"uuid": null,
+				"binary": null,
+				"fixed": null,
+				"small_dec": null,
+				"med_dec": null,
+				"large_dec": null
+			},
+			{
+				"bool": true,
+				"string": "z",
+				"string_long": "` + strings.Repeat("z", 22) + `",
+				"int": 9,
+				"long": 9,
+				"float": 0.9,
+				"double": 0.9,
+				"time": "00:00:03.000000",
+				"timestamp": "2023-03-01T19:25:00.000000+08:00",
+				"timestamptz": "2023-03-01T19:25:00.000000Z",
+				"date": "2023-03-01",
+				"uuid": "11111111-1111-1111-1111-111111111111",
+				"binary": "Eg==",
+				"fixed": "EREREREREREREREREREREQ==",
+				"small_dec": "876543.21",
+				"med_dec": "65432109876543.21",
+				"large_dec": "4321098765432109876543.21"
+			}
+		 ]`,
+	})
+	if err != nil {
+		panic(err)
+	}
+
+	return arrTable
+}
+
+func (t *TableWritingTestSuite) TestMergeManifests() {
+	tblA := t.createTableWithProps(table.Identifier{"default", "merge_manifest_a"},
+		iceberg.Properties{
+			table.ParquetCompressionKey:    "snappy",
+			table.ManifestMergeEnabledKey:  "true",
+			table.ManifestMinMergeCountKey: "1",
+			"format-version":               strconv.Itoa(t.formatVersion),
+		}, tableSchema())
+
+	tblB := t.createTableWithProps(table.Identifier{"default", "merge_manifest_b"},
+		iceberg.Properties{
+			table.ParquetCompressionKey:      "snappy",
+			table.ManifestMergeEnabledKey:    "true",
+			table.ManifestMinMergeCountKey:   "1",
+			table.ManifestTargetSizeBytesKey: "1",
+			"format-version":                 strconv.Itoa(t.formatVersion),
+		}, tableSchema())
+
+	tblC := t.createTableWithProps(table.Identifier{"default", "merge_manifest_c"},
+		iceberg.Properties{
+			table.ParquetCompressionKey:    "snappy",
+			table.ManifestMinMergeCountKey: "1",
+			"format-version":               strconv.Itoa(t.formatVersion),
+		}, tableSchema())
+
+	arrTable := arrowTableWithNull()
+	defer arrTable.Release()
+
+	var err error
+	// tblA should merge all manifests into 1
+	tblA, err = tblA.AppendTable(t.ctx, arrTable, 1, nil)
+	t.Require().NoError(err)
+	tblA, err = tblA.AppendTable(t.ctx, arrTable, 1, nil)
+	t.Require().NoError(err)
+	tblA, err = tblA.AppendTable(t.ctx, arrTable, 1, nil)
+	t.Require().NoError(err)
+
+	// tblB should not merge any manifests because the target size is too small
+	tblB, err = tblB.AppendTable(t.ctx, arrTable, 1, nil)
+	t.Require().NoError(err)
+	tblB, err = tblB.AppendTable(t.ctx, arrTable, 1, nil)
+	t.Require().NoError(err)
+	tblB, err = tblB.AppendTable(t.ctx, arrTable, 1, nil)
+	t.Require().NoError(err)
+
+	// tblC should not merge any manifests because merging is disabled
+	tblC, err = tblC.AppendTable(t.ctx, arrTable, 1, nil)
+	t.Require().NoError(err)
+	tblC, err = tblC.AppendTable(t.ctx, arrTable, 1, nil)
+	t.Require().NoError(err)
+	tblC, err = tblC.AppendTable(t.ctx, arrTable, 1, nil)
+	t.Require().NoError(err)
+
+	manifestList, err := tblA.CurrentSnapshot().Manifests(tblA.FS())
+	t.Require().NoError(err)
+	t.Len(manifestList, 1)
+
+	entries, err := manifestList[0].FetchEntries(tblA.FS(), false)
+	t.Require().NoError(err)
+	t.Len(entries, 3)
+
+	// entries should match the snapshot ID they were added in
+	snapshotList := tblA.Metadata().Snapshots()
+	slices.Reverse(snapshotList)
+	for i, entry := range entries {
+		t.Equal(snapshotList[i].SnapshotID, entry.SnapshotID())
+		if t.formatVersion > 1 {
+			t.EqualValues(3-i, entry.SequenceNum())
+		}
+	}
+
+	manifestList, err = tblB.CurrentSnapshot().Manifests(tblB.FS())
+	t.Require().NoError(err)
+	t.Len(manifestList, 3)
+
+	manifestList, err = tblC.CurrentSnapshot().Manifests(tblC.FS())
+	t.Require().NoError(err)
+	t.Len(manifestList, 3)
+
+	resultA, err := tblA.Scan().ToArrowTable(t.ctx)
+	t.Require().NoError(err)
+	defer resultA.Release()
+
+	resultB, err := tblB.Scan().ToArrowTable(t.ctx)
+	t.Require().NoError(err)
+	defer resultB.Release()
+
+	resultC, err := tblC.Scan().ToArrowTable(t.ctx)
+	t.Require().NoError(err)
+	defer resultC.Release()
+
+	// tblA and tblC should contain the same data
+	t.True(array.TableEqual(resultA, resultC), "expected:\n %s\ngot:\n %s", resultA, resultC)
+	// tblB and tblC should contain the same data
+	t.True(array.TableEqual(resultB, resultC), "expected:\n %s\ngot:\n %s", resultB, resultC)
+}
+
 func TestTableWriting(t *testing.T) {
 	suite.Run(t, &TableWritingTestSuite{formatVersion: 1})
 	suite.Run(t, &TableWritingTestSuite{formatVersion: 2})