You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iceberg.apache.org by ze...@apache.org on 2025/03/28 20:20:18 UTC
(iceberg-go) branch main updated: feat(table): add ReplaceFiles method to transaction (#356)
This is an automated email from the ASF dual-hosted git repository.
zeroshade pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg-go.git
The following commit(s) were added to refs/heads/main by this push:
new 3cc39f5 feat(table): add ReplaceFiles method to transaction (#356)
3cc39f5 is described below
commit 3cc39f57c7381c630cf3f93c233d38281093196f
Author: Matt Topol <zo...@gmail.com>
AuthorDate: Fri Mar 28 16:20:13 2025 -0400
feat(table): add ReplaceFiles method to transaction (#356)
Implementing a simple `ReplaceDataFiles` method for transactions to
create an `Overwrite` operation for deleting/adding files in a single
operation for use with compaction operations or otherwise replacing one
or more data files with new data files.
---------
Co-authored-by: Arnaud Briche <br...@gmail.com>
Co-authored-by: Kevin Liu <ke...@users.noreply.github.com>
---
config/config.go | 7 +-
table/internal/utils.go | 59 +++++++++++++
table/snapshot_producers.go | 177 ++++++++++++++++++++++++++++++++++++++-
table/snapshots.go | 31 +------
table/snapshots_internal_test.go | 43 +---------
table/table_test.go | 100 ++++++++++++++++++++++
table/transaction.go | 105 +++++++++++++++++++++++
7 files changed, 449 insertions(+), 73 deletions(-)
diff --git a/config/config.go b/config/config.go
index d4cf88a..3d645dd 100644
--- a/config/config.go
+++ b/config/config.go
@@ -25,12 +25,14 @@ import (
)
const (
- cfgFile = ".iceberg-go.yaml"
+ cfgFile = ".iceberg-go.yaml"
+ defaultMaxWorkers = 5
)
type Config struct {
DefaultCatalog string `yaml:"default-catalog"`
Catalogs map[string]CatalogConfig `yaml:"catalog"`
+ MaxWorkers int `yaml:"max-workers"`
}
type CatalogConfig struct {
@@ -88,6 +90,9 @@ func fromConfigFiles() Config {
if cfg.DefaultCatalog == "" {
cfg.DefaultCatalog = "default"
}
+ if cfg.MaxWorkers <= 0 {
+ cfg.MaxWorkers = defaultMaxWorkers
+ }
return cfg
}
diff --git a/table/internal/utils.go b/table/internal/utils.go
index 5a53b6a..3e67d8b 100644
--- a/table/internal/utils.go
+++ b/table/internal/utils.go
@@ -23,7 +23,9 @@ import (
"encoding/binary"
"errors"
"fmt"
+ "iter"
"regexp"
+ "runtime"
"strconv"
"strings"
"unicode/utf8"
@@ -32,6 +34,7 @@ import (
"github.com/apache/arrow-go/v18/arrow/decimal"
"github.com/apache/arrow-go/v18/arrow/decimal128"
"github.com/apache/iceberg-go"
+ "golang.org/x/sync/errgroup"
)
// Enumerated is a quick way to represent a sequenced value that can
@@ -478,3 +481,59 @@ 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] {
+ if nWorkers <= 0 {
+ nWorkers = runtime.GOMAXPROCS(0)
+ }
+
+ nWorkers = min(nWorkers, len(slice))
+
+ var g errgroup.Group
+ ch := make(chan T, len(slice))
+ out := make(chan S, nWorkers)
+
+ for range nWorkers {
+ g.Go(func() error {
+ for v := range ch {
+ result, err := fn(v)
+ if err != nil {
+ return err
+ }
+ out <- result
+ }
+
+ return nil
+ })
+ }
+
+ for _, v := range slice {
+ ch <- v
+ }
+ close(ch)
+
+ var err error
+ go func() {
+ defer close(out)
+ err = g.Wait()
+ }()
+
+ return func(yield func(S, error) bool) {
+ defer func() {
+ // drain out if we exit early
+ for range out {
+ }
+ }()
+
+ for v := range out {
+ if !yield(v, nil) {
+ return
+ }
+ }
+
+ if err != nil {
+ var z S
+ yield(z, err)
+ }
+ }
+}
diff --git a/table/snapshot_producers.go b/table/snapshot_producers.go
index 0309063..7519c9a 100644
--- a/table/snapshot_producers.go
+++ b/table/snapshot_producers.go
@@ -26,15 +26,24 @@ import (
"time"
"github.com/apache/iceberg-go"
+ "github.com/apache/iceberg-go/config"
"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"
"golang.org/x/sync/errgroup"
)
type producerImpl interface {
+ // to perform any post-processing on the manifests before writing them
+ // to the new snapshot. This will be called as the last step
+ // before writing a manifest list file, using the result of this function
+ // as the final list of manifests to write.
processManifests(manifests []iceberg.ManifestFile) ([]iceberg.ManifestFile, error)
+ // perform any processing necessary and return the list of existing
+ // manifests that should be included in the snapshot
existingManifests() ([]iceberg.ManifestFile, error)
+ // return the deleted entries for writing delete file manifests
deletedEntries() ([]iceberg.ManifestEntry, error)
}
@@ -87,9 +96,147 @@ func (fa *fastAppendFiles) existingManifests() ([]iceberg.ManifestFile, error) {
}
func (fa *fastAppendFiles) deletedEntries() ([]iceberg.ManifestEntry, error) {
+ // for fast appends, there are no deleted entries
return nil, nil
}
+type overwriteFiles struct {
+ base *snapshotProducer
+}
+
+func newOverwriteFilesProducer(op Operation, txn *Transaction, fs iceio.WriteFileIO, commitUUID *uuid.UUID, snapshotProps iceberg.Properties) *snapshotProducer {
+ prod := createSnapshotProducer(op, txn, fs, commitUUID, snapshotProps)
+ prod.producerImpl = &overwriteFiles{base: prod}
+
+ return prod
+}
+
+func (of *overwriteFiles) processManifests(manifests []iceberg.ManifestFile) ([]iceberg.ManifestFile, error) {
+ // no post processing
+ return manifests, nil
+}
+
+func (of *overwriteFiles) existingManifests() ([]iceberg.ManifestFile, error) {
+ // determine if there are any existing manifest files
+ existingFiles := make([]iceberg.ManifestFile, 0)
+
+ snap := of.base.txn.meta.currentSnapshot()
+ if snap == nil {
+ return existingFiles, nil
+ }
+
+ manifestList, err := snap.Manifests(of.base.io)
+ if err != nil {
+ return existingFiles, err
+ }
+
+ for _, m := range manifestList {
+ entries, err := of.base.fetchManifestEntry(m, true)
+ if err != nil {
+ return existingFiles, err
+ }
+
+ foundDeleted := make([]iceberg.ManifestEntry, 0)
+ notDeleted := make([]iceberg.ManifestEntry, 0, len(entries))
+ for _, entry := range entries {
+ if _, ok := of.base.deletedFiles[entry.DataFile().FilePath()]; ok {
+ foundDeleted = append(foundDeleted, entry)
+ } else {
+ notDeleted = append(notDeleted, entry)
+ }
+ }
+
+ if len(foundDeleted) == 0 {
+ existingFiles = append(existingFiles, m)
+
+ continue
+ }
+
+ if len(notDeleted) == 0 {
+ continue
+ }
+
+ spec, err := of.base.txn.meta.GetSpecByID(int(m.PartitionSpecID()))
+ if err != nil {
+ return existingFiles, err
+ }
+
+ wr, path, counter, err := of.base.newManifestWriter(*spec)
+ if err != nil {
+ return existingFiles, err
+ }
+ defer counter.W.(io.Closer).Close()
+
+ for _, entry := range notDeleted {
+ if err := wr.Existing(entry); err != nil {
+ return existingFiles, err
+ }
+ }
+
+ mf, err := wr.ToManifestFile(path, counter.Count)
+ if err != nil {
+ return existingFiles, err
+ }
+
+ existingFiles = append(existingFiles, mf)
+ }
+
+ return existingFiles, nil
+}
+
+func (of *overwriteFiles) deletedEntries() ([]iceberg.ManifestEntry, error) {
+ // determine if we need to record any deleted entries
+ //
+ // with a full overwrite all the entries are considered deleted
+ // with partial overwrites we have to use the predicate to evaluate
+ // which entries are affected
+ if of.base.parentSnapshotID <= 0 {
+ return nil, nil
+ }
+
+ parent, err := of.base.txn.meta.SnapshotByID(of.base.parentSnapshotID)
+ if err != nil {
+ return nil, fmt.Errorf("%w: cannot overwrite empty table", err)
+ }
+
+ previousManifests, err := parent.Manifests(of.base.io)
+ if err != nil {
+ return nil, err
+ }
+
+ getEntries := func(m iceberg.ManifestFile) ([]iceberg.ManifestEntry, error) {
+ entries, err := of.base.fetchManifestEntry(m, true)
+ if err != nil {
+ return nil, err
+ }
+
+ result := make([]iceberg.ManifestEntry, 0, len(entries))
+ for _, entry := range entries {
+ _, ok := of.base.deletedFiles[entry.DataFile().FilePath()]
+ if ok && entry.DataFile().ContentType() == iceberg.EntryContentData {
+ seqNum := entry.SequenceNum()
+ result = append(result,
+ iceberg.NewManifestEntry(iceberg.EntryStatusDELETED,
+ &of.base.snapshotID, &seqNum, entry.FileSequenceNum(),
+ entry.DataFile()))
+ }
+ }
+
+ return result, nil
+ }
+
+ nWorkers := config.EnvConfig.MaxWorkers
+ finalResult := make([]iceberg.ManifestEntry, 0, len(previousManifests))
+ for entries, err := range tblutils.MapExec(nWorkers, previousManifests, getEntries) {
+ if err != nil {
+ return nil, err
+ }
+ finalResult = append(finalResult, entries...)
+ }
+
+ return finalResult, nil
+}
+
type snapshotProducer struct {
producerImpl
@@ -148,6 +295,30 @@ func (sp *snapshotProducer) appendDataFile(df iceberg.DataFile) *snapshotProduce
return sp
}
+func (sp *snapshotProducer) deleteDataFile(df iceberg.DataFile) *snapshotProducer {
+ sp.deletedFiles[df.FilePath()] = df
+
+ return sp
+}
+
+func (sp *snapshotProducer) newManifestWriter(spec iceberg.PartitionSpec) (*iceberg.ManifestWriter, string, *internal.CountingWriter, error) {
+ out, path, err := sp.newManifestOutput()
+ if err != nil {
+ return nil, "", nil, err
+ }
+
+ counter := &internal.CountingWriter{W: out}
+ wr, err := iceberg.NewManifestWriter(sp.txn.meta.formatVersion, counter, spec,
+ sp.txn.meta.CurrentSchema(), sp.snapshotID)
+ if err != nil {
+ defer out.Close()
+
+ return nil, "", nil, err
+ }
+
+ return wr, path, counter, nil
+}
+
func (sp *snapshotProducer) newManifestOutput() (io.WriteCloser, string, error) {
provider, err := sp.txn.tbl.LocationProvider()
if err != nil {
@@ -163,6 +334,10 @@ func (sp *snapshotProducer) newManifestOutput() (io.WriteCloser, string, error)
return f, filepath, nil
}
+func (sp *snapshotProducer) fetchManifestEntry(m iceberg.ManifestFile, discardDeleted bool) ([]iceberg.ManifestEntry, error) {
+ return m.FetchEntries(sp.io, discardDeleted)
+}
+
func (sp *snapshotProducer) manifests() ([]iceberg.ManifestFile, error) {
var g errgroup.Group
@@ -290,7 +465,7 @@ func (sp *snapshotProducer) summary(props iceberg.Properties) (Summary, error) {
return updateSnapshotSummaries(Summary{
Operation: sp.op,
Properties: summaryProps,
- }, previousSummary, sp.op == OpOverwrite)
+ }, previousSummary)
}
func (sp *snapshotProducer) commit() ([]Update, []Requirement, error) {
diff --git a/table/snapshots.go b/table/snapshots.go
index 45b21ad..b034529 100644
--- a/table/snapshots.go
+++ b/table/snapshots.go
@@ -435,32 +435,7 @@ func (s *SnapshotSummaryCollector) build() iceberg.Properties {
return props
}
-func truncateTableSummary(sum Summary, previous iceberg.Properties) Summary {
- keys := []string{
- totalDataFilesKey, totalDeleteFilesKey, totalRecordsKey,
- totalFileSizeKey, totalPosDeletesKey, totalEqDeletesKey,
- }
- for _, prop := range keys {
- sum.Properties[prop] = "0"
- }
-
- updateProp := func(prop, prevProp string) {
- if val := previous.GetInt(prevProp, 0); val > 0 {
- sum.Properties[prop] = strconv.Itoa(val)
- }
- }
-
- updateProp(deletedDataFilesKey, totalDataFilesKey)
- updateProp(removedDeleteFilesKey, totalDeleteFilesKey)
- updateProp(deletedRecordsKey, totalRecordsKey)
- updateProp(removedFileSizeKey, totalFileSizeKey)
- updateProp(removedPosDeletesKey, totalPosDeletesKey)
- updateProp(removedEqDeletesKey, totalEqDeletesKey)
-
- return sum
-}
-
-func updateSnapshotSummaries(sum Summary, previous iceberg.Properties, truncateFullTable bool) (Summary, error) {
+func updateSnapshotSummaries(sum Summary, previous iceberg.Properties) (Summary, error) {
switch sum.Operation {
case OpAppend, OpOverwrite, OpDelete:
default:
@@ -471,10 +446,6 @@ func updateSnapshotSummaries(sum Summary, previous iceberg.Properties, truncateF
sum.Properties = make(iceberg.Properties)
}
- if truncateFullTable && sum.Operation == OpOverwrite && previous != nil {
- sum = truncateTableSummary(sum, previous)
- }
-
if previous == nil {
previous = iceberg.Properties{
totalDataFilesKey: "0",
diff --git a/table/snapshots_internal_test.go b/table/snapshots_internal_test.go
index c374178..0838abd 100644
--- a/table/snapshots_internal_test.go
+++ b/table/snapshots_internal_test.go
@@ -142,55 +142,16 @@ func TestMergeSnapshotSummaries(t *testing.T) {
"total-equality-deletes": "3",
}},
},
- {
- Summary{Operation: OpOverwrite, Properties: iceberg.Properties{
- "added-data-files": "1",
- "added-delete-files": "2",
- "added-equality-deletes": "3",
- "added-files-size": "4",
- "added-position-deletes": "5",
- "added-records": "6",
- }},
- iceberg.Properties{
- "total-data-files": "1",
- "total-delete-files": "1",
- "total-equality-deletes": "1",
- "total-files-size": "1",
- "total-position-deletes": "1",
- "total-records": "1",
- },
- true,
- Summary{Operation: OpOverwrite, Properties: iceberg.Properties{
- "added-data-files": "1",
- "added-delete-files": "2",
- "added-equality-deletes": "3",
- "added-files-size": "4",
- "added-position-deletes": "5",
- "added-records": "6",
- "total-data-files": "1",
- "total-records": "6",
- "total-delete-files": "2",
- "total-equality-deletes": "3",
- "total-files-size": "4",
- "total-position-deletes": "5",
- "deleted-data-files": "1",
- "removed-delete-files": "1",
- "deleted-records": "1",
- "removed-files-size": "1",
- "removed-position-deletes": "1",
- "removed-equality-deletes": "1",
- }},
- },
}
for _, tt := range tests {
- result, err := updateSnapshotSummaries(tt.sum, tt.previous, tt.truncateTable)
+ result, err := updateSnapshotSummaries(tt.sum, tt.previous)
require.NoError(t, err)
assert.Equal(t, tt.expected, result)
}
}
func TestInvalidOperation(t *testing.T) {
- _, err := updateSnapshotSummaries(Summary{Operation: OpReplace}, nil, false)
+ _, err := updateSnapshotSummaries(Summary{Operation: OpReplace}, nil)
assert.ErrorIs(t, err, iceberg.ErrNotImplemented)
}
diff --git a/table/table_test.go b/table/table_test.go
index 975f40d..4d7eb88 100644
--- a/table/table_test.go
+++ b/table/table_test.go
@@ -735,6 +735,106 @@ func (t *TableWritingTestSuite) TestAddFilesReferencedCurrentSnapshotIgnoreDupli
t.Equal([]int32{0, 0, 0}, deleted)
}
+type mockedCatalog struct{}
+
+func (m *mockedCatalog) LoadTable(ctx context.Context, ident table.Identifier, props iceberg.Properties) (*table.Table, error) {
+ return nil, nil
+}
+
+func (m *mockedCatalog) CommitTable(ctx context.Context, tbl *table.Table, reqs []table.Requirement, updates []table.Update) (table.Metadata, string, error) {
+ bldr, err := table.MetadataBuilderFromBase(tbl.Metadata())
+ if err != nil {
+ return nil, "", err
+ }
+
+ for _, u := range updates {
+ if err := u.Apply(bldr); err != nil {
+ return nil, "", err
+ }
+ }
+
+ meta, err := bldr.Build()
+ if err != nil {
+ return nil, "", err
+ }
+
+ return meta, "", nil
+}
+
+func (t *TableWritingTestSuite) TestReplaceDataFiles() {
+ fs := iceio.LocalFS{}
+
+ files := make([]string, 0)
+ for i := range 5 {
+ filePath := fmt.Sprintf("%s/replace_data_files_v%d/data-%d.parquet", t.location, t.formatVersion, i)
+ t.writeParquet(fs, filePath, t.arrTablePromotedTypes)
+ files = append(files, filePath)
+ }
+
+ ident := table.Identifier{"default", "replace_data_files_v" + strconv.Itoa(t.formatVersion)}
+ meta, err := table.NewMetadata(t.tableSchemaPromotedTypes, iceberg.UnpartitionedSpec,
+ table.UnsortedSortOrder, t.location, iceberg.Properties{"format-version": strconv.Itoa(t.formatVersion)})
+ t.Require().NoError(err)
+
+ ctx := context.Background()
+
+ tbl := table.New(ident, meta, t.getMetadataLoc(), fs, &mockedCatalog{})
+ for i := range 5 {
+ tx := tbl.NewTransaction()
+ t.Require().NoError(tx.AddFiles(ctx, files[i:i+1], nil, false))
+ tbl, err = tx.Commit(ctx)
+ t.Require().NoError(err)
+ }
+
+ mflist, err := tbl.CurrentSnapshot().Manifests(tbl.FS())
+ t.Require().NoError(err)
+ t.Len(mflist, 5)
+
+ // create a parquet file that is essentially as if we merged two of
+ // the data files together
+ cols := make([]arrow.Column, 0, t.arrTablePromotedTypes.NumCols())
+ for i := range int(t.arrTablePromotedTypes.NumCols()) {
+ chkd := t.arrTablePromotedTypes.Column(i).Data()
+ duplicated := arrow.NewChunked(chkd.DataType(), append(chkd.Chunks(), chkd.Chunks()...))
+ defer duplicated.Release()
+
+ col := arrow.NewColumn(t.arrSchemaPromotedTypes.Fields()[i], duplicated)
+ defer col.Release()
+
+ cols = append(cols, *col)
+ }
+
+ combined := array.NewTable(t.arrSchemaPromotedTypes, cols, -1)
+ defer combined.Release()
+
+ combinedFilePath := fmt.Sprintf("%s/replace_data_files_v%d/combined.parquet", t.location, t.formatVersion)
+ t.writeParquet(fs, combinedFilePath, combined)
+
+ tx := tbl.NewTransaction()
+ t.Require().NoError(tx.ReplaceDataFiles(ctx, files[:2], []string{combinedFilePath}, nil))
+
+ staged, err := tx.StagedTable()
+ t.Require().NoError(err)
+
+ t.Equal(&table.Summary{
+ Operation: table.OpOverwrite,
+ Properties: iceberg.Properties{
+ "added-data-files": "1",
+ "added-files-size": "1082",
+ "added-records": "4",
+ "deleted-data-files": "2",
+ "deleted-records": "4",
+ "removed-files-size": "2164",
+ "total-data-files": "4",
+ "total-delete-files": "0",
+ "total-equality-deletes": "0",
+ "total-files-size": "4328",
+ "total-position-deletes": "0",
+ "total-records": "10",
+ },
+ }, staged.CurrentSnapshot().Summary)
+}
+
func TestTableWriting(t *testing.T) {
suite.Run(t, &TableWritingTestSuite{formatVersion: 1})
suite.Run(t, &TableWritingTestSuite{formatVersion: 2})
diff --git a/table/transaction.go b/table/transaction.go
index c9c3c06..8b6d8cb 100644
--- a/table/transaction.go
+++ b/table/transaction.go
@@ -30,6 +30,7 @@ import (
"github.com/apache/arrow-go/v18/arrow/array"
"github.com/apache/iceberg-go"
"github.com/apache/iceberg-go/io"
+ "github.com/google/uuid"
)
type snapshotUpdate struct {
@@ -42,6 +43,15 @@ func (s snapshotUpdate) fastAppend() *snapshotProducer {
return newFastAppendFilesProducer(OpAppend, s.txn, s.io, nil, s.snapshotProps)
}
+func (s snapshotUpdate) mergeOverwrite(commitUUID *uuid.UUID) *snapshotProducer {
+ op := OpOverwrite
+ if s.txn.meta.currentSnapshot() == nil {
+ op = OpAppend
+ }
+
+ return newOverwriteFilesProducer(op, s.txn, s.io, commitUUID, s.snapshotProps)
+}
+
type Transaction struct {
tbl *Table
meta *MetadataBuilder
@@ -117,6 +127,101 @@ func (t *Transaction) SetProperties(props iceberg.Properties) error {
return nil
}
+// ReplaceFiles is actually just an overwrite operation with multiple
+// files deleted and added.
+//
+// TODO: technically, this could be a REPLACE operation but we aren't performing
+// any validation here that there are no changes to the underlying data. A REPLACE
+// operation is only valid if the data is exactly the same as the previous snapshot.
+//
+// For now, we'll keep using an overwrite operation.
+func (t *Transaction) ReplaceDataFiles(ctx context.Context, filesToDelete, filesToAdd []string, snapshotProps iceberg.Properties) error {
+ if len(filesToDelete) == 0 {
+ if len(filesToAdd) > 0 {
+ return t.AddFiles(ctx, filesToAdd, snapshotProps, false)
+ }
+ }
+
+ var (
+ setToDelete = make(map[string]struct{})
+ setToAdd = make(map[string]struct{})
+ )
+
+ for _, f := range filesToDelete {
+ setToDelete[f] = struct{}{}
+ }
+
+ for _, f := range filesToAdd {
+ setToAdd[f] = struct{}{}
+ }
+
+ if len(setToDelete) != len(filesToDelete) {
+ return errors.New("delete file paths must be unique for ReplaceDataFiles")
+ }
+
+ if len(setToAdd) != len(filesToAdd) {
+ return errors.New("add file paths must be unique for ReplaceDataFiles")
+ }
+
+ s := t.meta.currentSnapshot()
+ if s == nil {
+ return fmt.Errorf("%w: cannot replace files in a table without an existing snapshot", ErrInvalidOperation)
+ }
+
+ markedForDeletion := make([]iceberg.DataFile, 0, len(setToDelete))
+ for df, err := range s.dataFiles(t.tbl.fs, nil) {
+ if err != nil {
+ return err
+ }
+
+ if _, ok := setToDelete[df.FilePath()]; ok {
+ markedForDeletion = append(markedForDeletion, df)
+ }
+
+ if _, ok := setToAdd[df.FilePath()]; ok {
+ return fmt.Errorf("cannot add files that are already referenced by table, files: %s", df.FilePath())
+ }
+ }
+
+ if len(markedForDeletion) != len(setToDelete) {
+ return errors.New("cannot delete files that do not belong to the table")
+ }
+
+ if t.meta.NameMapping() == nil {
+ nameMapping := t.meta.CurrentSchema().NameMapping()
+ mappingJson, err := json.Marshal(nameMapping)
+ if err != nil {
+ return err
+ }
+ err = t.SetProperties(iceberg.Properties{DefaultNameMappingKey: string(mappingJson)})
+ if err != nil {
+ return err
+ }
+ }
+
+ commitUUID := uuid.New()
+ updater := t.updateSnapshot(snapshotProps).mergeOverwrite(&commitUUID)
+
+ for _, df := range markedForDeletion {
+ updater.deleteDataFile(df)
+ }
+
+ dataFiles := filesToDataFiles(ctx, t.tbl.fs, t.meta, slices.Values(filesToAdd))
+ for df, err := range dataFiles {
+ if err != nil {
+ return err
+ }
+ updater.appendDataFile(df)
+ }
+
+ updates, reqs, err := updater.commit()
+ if err != nil {
+ return err
+ }
+
+ return t.apply(updates, reqs)
+}
+
func (t *Transaction) Append(rdr array.RecordReader, snapshotProps iceberg.Properties) error {
return iceberg.ErrNotImplemented
}