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 b1206c60a feat(table): support dynamic partition overwrite (#2381)
b1206c60a is described below
commit b1206c60aeabb6a9a31f72daccf98305dcf5d759
Author: Minh Vu <[email protected]>
AuthorDate: Mon Nov 9 19:23:52 2026 +0100
feat(table): support dynamic partition overwrite (#2381)
* feat(table): support dynamic partition overwrite
Signed-off-by: Minh Vu <[email protected]>
* fix(table): honor cancellation before dynamic overwrite staging
Check cancellation before writing files and registering commit requirements.
Signed-off-by: Minh Vu <[email protected]>
* fix(table): preserve dynamic overwrite semantics for empty and
unpartitioned partitions
* test(table): cover unpartitioned dynamic overwrite and new partitions
* perf(table): reuse validated paths during partition overwrite
* test(table): cover v1 dynamic partition overwrite
Signed-off-by: Minh Vu <[email protected]>
* test(table): verify dynamic overwrite ref pin with commit retries
* fix(table): validate touched partitions on dynamic overwrite
* test(table): cover v3 and mixed-spec dynamic overwrite
* test(table): verify dynamic overwrite retry and branch fencing
* docs(table): clarify partition predicate scope
* test(table): fix partition iterator and assert shape rejection
* test(table): preserve only untouched partition deletes
* style(table): satisfy nlreturn in overwrite helpers
* style(table): satisfy nlreturn in partition test
* fix(table): inherit v3 row IDs during dynamic overwrite
* fix(table): resolve dynamic overwrite lint and test assertion
* test(table): cover null and date dynamic partition overwrites
* test(table): cover dynamic overwrite delete conflicts and DV cleanup
* style(table): satisfy nlreturn in dynamic overwrite tests
* style(table): gofmt dynamic overwrite integration tests
* fix(table): pass context to dynamic overwrite manifest APIs
* test(table): keep main intact on branch dynamic overwrite
* fix(table): preserve old specs in dynamic overwrite
* fix(table): remove unused overwrite locals
* fix(table): preserve deletes for surviving partition-spec data
* fix(table): fence old-spec writes in unpartitioned overwrite retries
* style(table): format dynamic overwrite regression tests
* test(table): filter delete files from dynamic overwrite data assertion
* docs(table): clarify dynamic overwrite with evolved partition specs
* test(table): cover staged dynamic overwrite and Arrow table API
---------
Signed-off-by: Minh Vu <[email protected]>
---
table/dynamic_partition_overwrite.go | 444 ++++++++++++++++++
...dynamic_partition_overwrite_integration_test.go | 436 ++++++++++++++++++
table/dynamic_partition_overwrite_test.go | 511 +++++++++++++++++++++
table/internal/partition_predicate.go | 10 +-
website/src/api.md | 35 ++
5 files changed, 1431 insertions(+), 5 deletions(-)
diff --git a/table/dynamic_partition_overwrite.go
b/table/dynamic_partition_overwrite.go
new file mode 100644
index 000000000..942bdd300
--- /dev/null
+++ b/table/dynamic_partition_overwrite.go
@@ -0,0 +1,444 @@
+// 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"
+ "maps"
+
+ "github.com/apache/arrow-go/v18/arrow"
+ "github.com/apache/arrow-go/v18/arrow/array"
+ "github.com/apache/iceberg-go"
+ "github.com/google/uuid"
+)
+
+// dynamicPartitionKeys normalizes all partition values before comparing them.
+// A manifest decoder may represent the same logical day as an int32 while the
+// writer uses iceberg.Date. The shared canonical encoder handles both forms,
+// decimals, bytes, UUIDs, and nulls without lossy string formatting.
+func dynamicPartitionKeys(files []iceberg.DataFile) (map[string]struct{},
error) {
+ keys := make(map[string]struct{}, len(files))
+ for _, file := range files {
+ key, err := canonicalPartitionKey(file.SpecID(),
dataFilePartition(file))
+ if err != nil {
+ return nil, fmt.Errorf("dynamic partition overwrite
file %q: %w", file.FilePath(), err)
+ }
+ keys[key] = struct{}{}
+ }
+
+ return keys, nil
+}
+
+func (t *Transaction) dynamicOverwriteSpec() (*iceberg.PartitionSpec, error) {
+ meta, err := t.txnMeta()
+ if err != nil {
+ return nil, err
+ }
+
+ if meta.formatVersion > 3 {
+ return nil, fmt.Errorf("%w: dynamic partition overwrite does
not support format v%d", ErrInvalidOperation, meta.formatVersion)
+ }
+
+ spec, err := meta.CurrentSpec()
+ if err != nil {
+ return nil, err
+ }
+ if spec == nil {
+ return nil, fmt.Errorf("%w: dynamic partition overwrite
requires a current partition spec", ErrInvalidOperation)
+ }
+ if err := checkNoUnknownTransform(spec); err != nil {
+ return nil, err
+ }
+
+ return spec, nil
+}
+
+// validateDynamicOverwriteConflicts uses exact stored partition tuples so
+// transformed and null partitions cannot become false-negative row predicates.
+// It rejects concurrent additions and removals in touched partitions while
+// allowing unrelated partition appends during refresh-and-replay.
+func validateDynamicOverwriteConflicts(cc *conflictContext, specID int,
unpartitioned bool, touched map[string]struct{}) error {
+ if cc == nil || len(cc.concurrent) == 0 {
+ return nil
+ }
+ for _, snapshot := range cc.concurrent {
+ manifests, err := cc.manifestsFor(snapshot)
+ if err != nil {
+ return fmt.Errorf("dynamic partition overwrite:
concurrent manifests for snapshot %d: %w", snapshot.SnapshotID, err)
+ }
+ for _, manifest := range manifests {
+ for entry, err := range manifest.Entries(cc.ctx,
cc.manifestReadIO(), false) {
+ if err != nil {
+ return fmt.Errorf("dynamic partition
overwrite: read concurrent manifest %s: %w", manifest.FilePath(), err)
+ }
+ if entry.SnapshotID() != snapshot.SnapshotID ||
entry.Status() == iceberg.EntryStatusEXISTING {
+ continue
+ }
+ file := entry.DataFile()
+ conflict := ErrConflictingDataFiles
+ if file.ContentType() !=
iceberg.EntryContentData {
+ conflict = ErrConflictingDeleteFiles
+ }
+ // No partition scope exists for an
unpartitioned overwrite.
+ // Every concurrent change matters, including
older spec IDs.
+ if unpartitioned {
+ return fmt.Errorf("%w: concurrent
snapshot %d changed file %q during an unpartitioned overwrite",
+ conflict, snapshot.SnapshotID,
file.FilePath())
+ }
+ if int(file.SpecID()) != specID {
+ continue
+ }
+ key, err :=
canonicalPartitionKey(file.SpecID(), dataFilePartition(file))
+ if err != nil {
+ return fmt.Errorf("dynamic partition
overwrite: concurrent file %q: %w", file.FilePath(), err)
+ }
+ if _, ok := touched[key]; ok {
+ return fmt.Errorf("%w: concurrent
snapshot %d changed file %q in an overwritten partition",
+ conflict, snapshot.SnapshotID,
file.FilePath())
+ }
+ }
+ }
+ }
+
+ return nil
+}
+
+// DynamicPartitionOverwriteDataFiles atomically replaces all live data in
+// the partitions represented by filesToAdd. Empty input is a no-op; an
+// unpartitioned table has one partition. New partitions still use OpOverwrite.
+//
+// Added files must use the table's current partition spec; existing live files
+// under older specs are preserved unless the current spec is unpartitioned,
+// in which case all existing files are replaced. Supports v1-v3, including v3
row lineage
+// and associated deletion vectors. On v3, a new data file may omit
first_row_id
+// and inherit it from its manifest.
+// Concurrent writes to touched partitions fail while independent appends to
+// other partitions can be retried unless delete-file removal prevents replay.
+// A missing target branch is pinned together with main to guard its origin.
+// Default partition and sort order changes also fail.
+//
+// All input files must already exist. Uncommitted data files remain the
+// caller's responsibility after a staging or commit failure. By default this
+// sets the table's name mapping; WithoutAutoNameMapping disables that step.
+// WithoutDuplicateCheck is ignored to avoid adding an already live file twice.
+func (t *Transaction) DynamicPartitionOverwriteDataFiles(
+ ctx context.Context, filesToAdd []iceberg.DataFile, snapshotProps
iceberg.Properties, opts ...WriteOption,
+) error {
+ spec, err := t.dynamicOverwriteSpec()
+ if err != nil {
+ return err
+ }
+ if len(filesToAdd) == 0 {
+ return nil
+ }
+ if err := ctx.Err(); err != nil {
+ return err
+ }
+
+ var cfg dataFileCfg
+ for _, option := range opts {
+ option(&cfg)
+ }
+ if cfg.rewriteSemantics {
+ return fmt.Errorf("%w: dynamic partition overwrite cannot use
data-preserving rewrite semantics", ErrInvalidOperation)
+ }
+
+ // Dynamic overwrite adds new rows, so v3 first_row_id can be inherited
+ // from the manifest. Java ReplacePartitions uses the same inheritance.
+ addedPaths, err := t.validateDataFilesToAdd(filesToAdd,
"DynamicPartitionOverwriteDataFiles", false)
+ if err != nil {
+ return err
+ }
+ for _, df := range filesToAdd {
+ if int(df.SpecID()) != spec.ID() {
+ return fmt.Errorf("%w: dynamic partition overwrite file
%q uses spec %d, current spec is %d",
+ ErrInvalidOperation, df.FilePath(),
df.SpecID(), spec.ID())
+ }
+ }
+
+ fs, err := t.tbl.fsF(ctx)
+ if err != nil {
+ return err
+ }
+ wfs, err := requireWriteFileIO(fs)
+ if err != nil {
+ return err
+ }
+
+ keys, err := dynamicPartitionKeys(filesToAdd)
+ if err != nil {
+ return err
+ }
+ var toDelete, deleteFilesToRemove []iceberg.DataFile
+ unpartitioned := spec.IsUnpartitioned()
+ survivingDataPaths := make(map[string]struct{})
+ var hasSurvivingOldSpecData bool
+ validationPlan := newPartitionValidationPlan(spec)
+ if snapshot := t.planningSnapshot(t.meta); snapshot != nil {
+ for df, err := range snapshot.dataFiles(ctx, fs, nil) {
+ if err != nil {
+ return fmt.Errorf("dynamic partition overwrite:
read data files: %w", err)
+ }
+ if err := ctx.Err(); err != nil {
+ return err
+ }
+ if _, exists := addedPaths[df.FilePath()]; exists {
+ return fmt.Errorf("%w: added file %q already
exists in the target snapshot", ErrInvalidOperation, df.FilePath())
+ }
+ if df.ContentType() != iceberg.EntryContentData {
+ continue
+ }
+ if int(df.SpecID()) != spec.ID() {
+ if !unpartitioned {
+ survivingDataPaths[df.FilePath()] =
struct{}{}
+ hasSurvivingOldSpecData = true
+
+ continue
+ }
+ } else if err := validationPlan.validate(df); err !=
nil {
+ return fmt.Errorf("%w: invalid live partition
in file %q: %w",
+ ErrInvalidOperation, df.FilePath(), err)
+ }
+
+ // A current unpartitioned spec represents the whole
table, even
+ // when existing files were written under older
partition specs.
+ if unpartitioned {
+ toDelete = append(toDelete, df)
+
+ continue
+ }
+
+ key, err := canonicalPartitionKey(df.SpecID(),
dataFilePartition(df))
+ if err != nil {
+ return fmt.Errorf("dynamic partition overwrite
live file %q: %w", df.FilePath(), err)
+ }
+ if _, touched := keys[key]; touched {
+ toDelete = append(toDelete, df)
+ } else {
+ survivingDataPaths[df.FilePath()] = struct{}{}
+ }
+ }
+
+ // A replaced partition has no surviving current-spec data that
+ // its old deletes should affect. Preserve deletes that can
still
+ // target surviving files under an older spec. Deletion vectors
are
+ // removed separately with their referenced data files.
+ for entry, err := range snapshot.entries(ctx, fs,
iceberg.ManifestContentDeletes, true) {
+ if err != nil {
+ return fmt.Errorf("dynamic partition overwrite:
read delete files: %w", err)
+ }
+ if err := ctx.Err(); err != nil {
+ return err
+ }
+ df := entry.DataFile()
+ if _, exists := addedPaths[df.FilePath()]; exists {
+ return fmt.Errorf("%w: added file %q already
exists as a delete file", ErrInvalidOperation, df.FilePath())
+ }
+ if IsDeletionVector(df) {
+ continue
+ }
+ if unpartitioned {
+ deleteFilesToRemove =
append(deleteFilesToRemove, df)
+
+ continue
+ }
+ if ref := df.ReferencedDataFile(); ref != nil && *ref
!= "" {
+ if _, survives := survivingDataPaths[*ref];
survives {
+ continue
+ }
+ } else if df.ContentType() ==
iceberg.EntryContentPosDeletes && hasSurvivingOldSpecData {
+ // V2 manifests do not carry
referenced_data_file. A position
+ // delete may still target an old-spec file, so
do not discard
+ // it while any such data file survives.
+ continue
+ }
+ if int(df.SpecID()) != spec.ID() {
+ continue
+ }
+ key, err := canonicalPartitionKey(df.SpecID(),
dataFilePartition(df))
+ if err != nil {
+ return fmt.Errorf("dynamic partition overwrite
live delete file %q: %w", df.FilePath(), err)
+ }
+ if _, touched := keys[key]; touched {
+ deleteFilesToRemove =
append(deleteFilesToRemove, df)
+ }
+ }
+ }
+
+ var dvs liveDeletionVectors
+ if t.meta.formatVersion >= 3 && len(toDelete) != 0 {
+ if dvs, err = t.readLiveDeletionVectors(ctx, fs); err != nil {
+ return err
+ }
+ }
+ if err := ctx.Err(); err != nil {
+ return err
+ }
+
+ if err := t.AssertDefaultShape(); err != nil {
+ return err
+ }
+ branch := t.branch
+ if branch == "" {
+ branch = MainBranch
+ }
+ if t.baseRefSnapshotID(branch) == nil {
+ if err := t.AssertRefSnapshotID(branch); err != nil {
+ return err
+ }
+ if branch != MainBranch {
+ if err := t.AssertRefSnapshotID(MainBranch); err != nil
{
+ return err
+ }
+ }
+ }
+
+ if !cfg.skipAutoNameMapping {
+ if err := t.ensureNameMapping(); err != nil {
+ return err
+ }
+ }
+
+ props := make(iceberg.Properties, len(snapshotProps)+1)
+ maps.Copy(props, snapshotProps)
+ props["replace-partitions"] = "true"
+
+ // Use one producer for additions and removals. In contrast,
+ // ReplaceDataFiles would scan manifests again and produce an append
+ // operation when there are no existing files in the touched partitions.
+ producer := newOverwriteFilesProducer(OpOverwrite, t, wfs, nil, props)
+ producer.setSkipDefaultValidator(true)
+ for _, file := range toDelete {
+ dvs.removeDataFile(producer, file)
+ }
+ for _, file := range deleteFilesToRemove {
+ producer.removeDeleteFile(file)
+ }
+ for _, file := range filesToAdd {
+ producer.appendDataFile(file)
+ }
+ updates, reqs, err := producer.commit(ctx)
+ if err != nil {
+ return err
+ }
+
+ // Publish the staged snapshot and validator together so concurrent
+ // Commit cannot see one without the other.
+ t.mx.Lock()
+ defer t.mx.Unlock()
+ if err := t.applyLocked(updates, reqs); err != nil {
+ return err
+ }
+ t.validators = append(t.validators, func(cc *conflictContext) error {
+ return validateDynamicOverwriteConflicts(cc, spec.ID(),
unpartitioned, keys)
+ })
+
+ return nil
+}
+
+// DynamicPartitionOverwrite writes Arrow records to data files and replaces
+// only the partitions represented by the written records. Empty input is
+// a no-op. Supports v1-v3 and preserves live files under older partition
+// specs. Concurrent touched-partition changes fail; removal of delete files or
+// DVs prevents replay. Written files remain the caller's responsibility after
+// failure. Pass WithoutAutoNameMapping to avoid changing that property.
+func (t *Transaction) DynamicPartitionOverwrite(
+ ctx context.Context, rdr array.RecordReader, snapshotProps
iceberg.Properties, opts ...WriteOption,
+) error {
+ if rdr == nil {
+ return fmt.Errorf("%w: nil record reader for dynamic partition
overwrite", ErrInvalidOperation)
+ }
+ if _, err := t.dynamicOverwriteSpec(); err != nil {
+ return err
+ }
+ if err := ctx.Err(); err != nil {
+ return err
+ }
+
+ meta, err := t.txnMeta()
+ if err != nil {
+ return err
+ }
+ fs, err := t.tbl.fsF(ctx)
+ if err != nil {
+ return err
+ }
+ wfs, err := requireWriteFileIO(fs)
+ if err != nil {
+ return err
+ }
+
+ writeUUID := uuid.New()
+ itr := recordsToDataFiles(ctx, t.tbl.Location(), meta,
recordWritingArgs{
+ sc: rdr.Schema(),
+ itr: array.IterFromReader(rdr),
+ fs: wfs,
+ writeUUID: &writeUUID,
+ })
+
+ var written []iceberg.DataFile
+ for df, err := range itr {
+ if err != nil {
+ return err
+ }
+ written = append(written, df)
+ }
+
+ return t.DynamicPartitionOverwriteDataFiles(ctx, written,
snapshotProps, opts...)
+}
+
+// DynamicPartitionOverwriteTable is the Arrow-table variant of
+// DynamicPartitionOverwrite. It shares the v3, older-spec file handling,
+// concurrency and uncommitted-file behavior of the record-reader method.
+func (t *Transaction) DynamicPartitionOverwriteTable(
+ ctx context.Context, tbl arrow.Table, batchSize int64, snapshotProps
iceberg.Properties, opts ...WriteOption,
+) error {
+ if tbl == nil {
+ return fmt.Errorf("%w: nil Arrow table for dynamic partition
overwrite", ErrInvalidOperation)
+ }
+ rdr := array.NewTableReader(tbl, batchSize)
+ defer rdr.Release()
+
+ return t.DynamicPartitionOverwrite(ctx, rdr, snapshotProps, opts...)
+}
+
+// DynamicPartitionOverwrite commits a dynamic overwrite in one transaction.
+// It shares the older-spec, v3, concurrency and uncommitted-file rules above.
+func (t Table) DynamicPartitionOverwrite(
+ ctx context.Context, rdr array.RecordReader, snapshotProps
iceberg.Properties, opts ...WriteOption,
+) (*Table, error) {
+ txn := t.NewTransaction()
+ if err := txn.DynamicPartitionOverwrite(ctx, rdr, snapshotProps,
opts...); err != nil {
+ return nil, err
+ }
+
+ return txn.Commit(ctx)
+}
+
+// DynamicPartitionOverwriteTable is the Arrow-table convenience wrapper.
+// It shares the older-spec, v3, concurrency and uncommitted-file rules above.
+func (t Table) DynamicPartitionOverwriteTable(
+ ctx context.Context, tbl arrow.Table, batchSize int64, snapshotProps
iceberg.Properties, opts ...WriteOption,
+) (*Table, error) {
+ txn := t.NewTransaction()
+ if err := txn.DynamicPartitionOverwriteTable(ctx, tbl, batchSize,
snapshotProps, opts...); err != nil {
+ return nil, err
+ }
+
+ return txn.Commit(ctx)
+}
diff --git a/table/dynamic_partition_overwrite_integration_test.go
b/table/dynamic_partition_overwrite_integration_test.go
new file mode 100644
index 000000000..9fbc4f7bd
--- /dev/null
+++ b/table/dynamic_partition_overwrite_integration_test.go
@@ -0,0 +1,436 @@
+// 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_test
+
+import (
+ "os"
+ "path/filepath"
+ "slices"
+ "testing"
+
+ "github.com/apache/arrow-go/v18/arrow/array"
+ "github.com/apache/arrow-go/v18/arrow/memory"
+ "github.com/apache/iceberg-go"
+ "github.com/apache/iceberg-go/table"
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+)
+
+func newDynamicOverwriteTable(t *testing.T) (*table.Table,
*concurrentTestCatalog) {
+ spec := iceberg.NewPartitionSpec(iceberg.PartitionField{SourceIDs:
[]int{1}, FieldID: 1000, Name: "id", Transform: iceberg.IdentityTransform{}})
+
+ return newDynamicOverwriteTableWithSpec(t, spec)
+}
+
+func newDynamicOverwriteTableWithSpec(t *testing.T, spec
iceberg.PartitionSpec) (*table.Table, *concurrentTestCatalog) {
+ t.Helper()
+
+ return newDynamicOverwriteTableWithFormat(t, spec, "2")
+}
+
+func newDynamicOverwriteTableWithFormat(t *testing.T, spec
iceberg.PartitionSpec, formatVersion string) (*table.Table,
*concurrentTestCatalog) {
+ t.Helper()
+ schema := iceberg.NewSchema(0, iceberg.NestedField{ID: 1, Name: "id",
Type: iceberg.PrimitiveTypes.Int64, Required: true})
+
+ return newDynamicOverwriteTableWithSchema(t, schema, spec,
formatVersion)
+}
+
+func newDynamicOverwriteTableWithSchema(t *testing.T, schema *iceberg.Schema,
spec iceberg.PartitionSpec, formatVersion string) (*table.Table,
*concurrentTestCatalog) {
+ t.Helper()
+ location := filepath.ToSlash(t.TempDir())
+ meta, err := table.NewMetadata(schema, &spec, table.UnsortedSortOrder,
location, iceberg.Properties{
+ table.PropertyFormatVersion: formatVersion,
+ table.CommitNumRetriesKey: "2",
+ table.CommitMinRetryWaitMsKey: "1",
+ table.CommitMaxRetryWaitMsKey: "2",
+ table.CommitTotalRetryTimeoutMsKey: "1000",
+ })
+ require.NoError(t, err)
+ metaLoc := location + "/metadata/v1.metadata.json"
+ cat := &concurrentTestCatalog{metadata: meta, location: metaLoc, fsF:
branchPlanningFS}
+
+ return table.New(branchPlanningIdent, meta, metaLoc, branchPlanningFS,
cat), cat
+}
+
+func TestDynamicPartitionOverwriteArrowAndConcurrentCommit(t *testing.T) {
+ base, cat := newDynamicOverwriteTable(t)
+ tbl := appendRowsOnRef(t, base, table.MainBranch, 1, 2)
+ txn := tbl.NewTransaction()
+ reader := idRecordReader(t, tbl, 1, 3)
+ require.NoError(t, txn.DynamicPartitionOverwrite(t.Context(), reader,
nil))
+ reader.Release()
+ updated, err := txn.Commit(t.Context())
+ require.NoError(t, err)
+ assert.ElementsMatch(t, []int64{1, 2, 3}, idsOnRef(t, updated,
table.MainBranch))
+
+ stale := updated.NewTransaction()
+ reader = idRecordReader(t, updated, 1)
+ require.NoError(t, stale.DynamicPartitionOverwrite(t.Context(), reader,
nil))
+ reader.Release()
+ updated = appendRowsOnRef(t, updated, table.MainBranch, 1)
+ attemptsBefore := cat.attempts.Load()
+ _, err = stale.Commit(t.Context())
+ require.ErrorIs(t, err, table.ErrConflictingDataFiles)
+ assert.Equal(t, attemptsBefore+1, cat.attempts.Load(),
+ "touched partition conflict should fail before another catalog
submission")
+ assert.ElementsMatch(t, []int64{1, 1, 2, 3}, idsOnRef(t, updated,
table.MainBranch))
+}
+
+// An append staged in the same transaction must be visible to dynamic
+// overwrite planning: touched rows are replaced, and other rows survive.
+func TestDynamicPartitionOverwriteAfterAppendInSameTransaction(t *testing.T) {
+ base, _ := newDynamicOverwriteTable(t)
+ tbl := appendRowsOnRef(t, base, table.MainBranch, 1, 2)
+ txn := tbl.NewTransaction()
+
+ appended := idRecordReader(t, tbl, 1, 3)
+ require.NoError(t, txn.Append(t.Context(), appended, nil))
+ appended.Release()
+
+ replacement := idRecordReader(t, tbl, 1)
+ require.NoError(t, txn.DynamicPartitionOverwrite(t.Context(),
replacement, nil))
+ replacement.Release()
+
+ updated, err := txn.Commit(t.Context())
+ require.NoError(t, err)
+ require.ElementsMatch(t, []int64{1, 2, 3}, idsOnRef(t, updated,
table.MainBranch))
+}
+
+func TestDynamicPartitionOverwriteTable(t *testing.T) {
+ base, _ := newDynamicOverwriteTable(t)
+ tbl := appendRowsOnRef(t, base, table.MainBranch, 1, 2)
+ arrowSchema, err := table.SchemaToArrowSchema(tbl.Schema(), nil, false,
false)
+ require.NoError(t, err)
+ data, err := array.TableFromJSON(memory.DefaultAllocator, arrowSchema,
+ []string{`[{"id":1},{"id":3}]`})
+ require.NoError(t, err)
+ defer data.Release()
+
+ updated, err := tbl.DynamicPartitionOverwriteTable(t.Context(), data,
1, nil)
+ require.NoError(t, err)
+ require.ElementsMatch(t, []int64{1, 2, 3}, idsOnRef(t, updated,
table.MainBranch))
+}
+
+func TestDynamicPartitionOverwriteAllowsConcurrentOtherPartition(t *testing.T)
{
+ base, cat := newDynamicOverwriteTable(t)
+ tbl := appendRowsOnRef(t, base, table.MainBranch, 1, 2)
+ stale := tbl.NewTransaction()
+ reader := idRecordReader(t, tbl, 1, 3)
+ require.NoError(t, stale.DynamicPartitionOverwrite(t.Context(), reader,
nil))
+ reader.Release()
+
+ peer := appendRowsOnRef(t, tbl, table.MainBranch, 4)
+ attemptsBefore := cat.attempts.Load()
+ result, err := stale.Commit(t.Context())
+ require.NoError(t, err)
+ assert.Equal(t, attemptsBefore+2, cat.attempts.Load(),
"different-partition append should allow refresh and retry")
+ assert.ElementsMatch(t, []int64{1, 2, 3, 4}, idsOnRef(t, result,
table.MainBranch))
+ assert.ElementsMatch(t, []int64{1, 2, 4}, idsOnRef(t, peer,
table.MainBranch))
+}
+
+// Overwriting an existing branch must plan against that branch's files,
+// rather than replacing files that only main references.
+func TestDynamicPartitionOverwriteExistingBranchKeepsMainUnchanged(t
*testing.T) {
+ base, _ := newDynamicOverwriteTable(t)
+ tbl := appendRowsOnRef(t, base, table.MainBranch, 1, 2)
+ tbl = appendRowsOnRef(t, tbl, branchPlanningRef, 1, 3)
+ tbl = appendRowsOnRef(t, tbl, table.MainBranch, 4)
+
+ require.ElementsMatch(t, []int64{1, 1, 2, 3}, idsOnRef(t, tbl,
branchPlanningRef))
+ require.ElementsMatch(t, []int64{1, 2, 4}, idsOnRef(t, tbl,
table.MainBranch))
+
+ txn, err := tbl.NewTransactionOnBranch(branchPlanningRef)
+ require.NoError(t, err)
+ reader := idRecordReader(t, tbl, 1, 5)
+ require.NoError(t, txn.DynamicPartitionOverwrite(t.Context(), reader,
nil))
+ reader.Release()
+ updated, err := txn.Commit(t.Context())
+ require.NoError(t, err)
+
+ require.ElementsMatch(t, []int64{1, 2, 3, 5}, idsOnRef(t, updated,
branchPlanningRef))
+ require.ElementsMatch(t, []int64{1, 2, 4}, idsOnRef(t, updated,
table.MainBranch))
+}
+
+func TestDynamicPartitionOverwriteMissingBranchPinsMain(t *testing.T) {
+ base, _ := newDynamicOverwriteTable(t)
+ tbl := appendRowsOnRef(t, base, table.MainBranch, 1, 2)
+ stale, err := tbl.NewTransactionOnBranch("audit")
+ require.NoError(t, err)
+ reader := idRecordReader(t, tbl, 1)
+ require.NoError(t, stale.DynamicPartitionOverwrite(t.Context(), reader,
nil))
+ reader.Release()
+
+ peer := appendRowsOnRef(t, tbl, table.MainBranch, 1)
+ _, err = stale.Commit(t.Context())
+ require.ErrorIs(t, err, table.ErrCommitFailed)
+ assert.ElementsMatch(t, []int64{1, 1, 2}, idsOnRef(t, peer,
table.MainBranch))
+}
+
+func TestDynamicPartitionOverwriteBucketTransformWithFiles(t *testing.T) {
+ spec := iceberg.NewPartitionSpec(iceberg.PartitionField{
+ SourceIDs: []int{1}, FieldID: 1000, Name: "id_bucket",
Transform: iceberg.BucketTransform{NumBuckets: 256},
+ })
+ base, _ := newDynamicOverwriteTableWithSpec(t, spec)
+ tbl := appendRowsOnRef(t, base, table.MainBranch, 1, 2)
+ reader := idRecordReader(t, tbl, 1, 3)
+ updated, err := tbl.DynamicPartitionOverwrite(t.Context(), reader, nil)
+ reader.Release()
+ require.NoError(t, err)
+ assert.ElementsMatch(t, []int64{1, 2, 3}, idsOnRef(t, updated,
table.MainBranch))
+}
+
+func TestDynamicPartitionOverwriteV3ArrowInheritsRowIDs(t *testing.T) {
+ spec := iceberg.NewPartitionSpec(iceberg.PartitionField{
+ SourceIDs: []int{1}, FieldID: 1000, Name: "id", Transform:
iceberg.IdentityTransform{},
+ })
+ base, _ := newDynamicOverwriteTableWithFormat(t, spec, "3")
+ tbl := appendRowsOnRef(t, base, table.MainBranch, 1, 2)
+
+ reader := idRecordReader(t, tbl, 1, 3)
+ updated, err := tbl.DynamicPartitionOverwrite(t.Context(), reader, nil)
+ reader.Release()
+ require.NoError(t, err)
+ assert.ElementsMatch(t, []int64{1, 2, 3}, idsOnRef(t, updated,
table.MainBranch))
+
+ head := updated.SnapshotByName(table.MainBranch)
+ require.NotNil(t, head)
+ require.NotNil(t, head.FirstRowID)
+ require.NotNil(t, head.AddedRows)
+ assert.Positive(t, *head.AddedRows)
+}
+
+func dynamicOverwriteJSONReader(t *testing.T, tbl *table.Table, rows string)
array.RecordReader {
+ t.Helper()
+ arrowSchema, err := table.SchemaToArrowSchema(tbl.Schema(), nil, false,
false)
+ require.NoError(t, err)
+ data, err := array.TableFromJSON(memory.DefaultAllocator, arrowSchema,
[]string{rows})
+ require.NoError(t, err)
+ t.Cleanup(data.Release)
+
+ return array.NewTableReader(data, -1)
+}
+
+// A null source value must be stored as a null partition tuple, and a
+// replacement of that partition must leave non-null partitions untouched.
+func TestDynamicPartitionOverwriteNullIdentityPartition(t *testing.T) {
+ schema := iceberg.NewSchema(0,
+ iceberg.NestedField{ID: 1, Name: "id", Type:
iceberg.PrimitiveTypes.Int64, Required: true},
+ iceberg.NestedField{ID: 2, Name: "group", Type:
iceberg.PrimitiveTypes.Int32},
+ )
+ spec := iceberg.NewPartitionSpec(iceberg.PartitionField{
+ SourceIDs: []int{2}, FieldID: 1000, Name: "group", Transform:
iceberg.IdentityTransform{},
+ })
+ tbl, _ := newDynamicOverwriteTableWithSchema(t, schema, spec, "2")
+
+ seed := dynamicOverwriteJSONReader(t, tbl,
`[{"id":1,"group":null},{"id":2,"group":7},{"id":3,"group":null}]`)
+ tx := tbl.NewTransaction()
+ require.NoError(t, tx.Append(t.Context(), seed, nil))
+ seed.Release()
+ tbl, err := tx.Commit(t.Context())
+ require.NoError(t, err)
+ require.ElementsMatch(t, []int64{1, 2, 3}, idsOnRef(t, tbl,
table.MainBranch))
+
+ replacement := dynamicOverwriteJSONReader(t, tbl,
`[{"id":4,"group":null},{"id":5,"group":null}]`)
+ updated, err := tbl.DynamicPartitionOverwrite(t.Context(), replacement,
nil)
+ replacement.Release()
+ require.NoError(t, err)
+ require.ElementsMatch(t, []int64{2, 4, 5}, idsOnRef(t, updated,
table.MainBranch))
+}
+
+// The manifest decoder and Arrow writer can represent day partitions with
+// different Go types. An overwrite must still match their logical values.
+func TestDynamicPartitionOverwriteDatePartition(t *testing.T) {
+ schema := iceberg.NewSchema(0,
+ iceberg.NestedField{ID: 1, Name: "id", Type:
iceberg.PrimitiveTypes.Int64, Required: true},
+ iceberg.NestedField{ID: 2, Name: "event_date", Type:
iceberg.PrimitiveTypes.Date, Required: true},
+ )
+ spec := iceberg.NewPartitionSpec(iceberg.PartitionField{
+ SourceIDs: []int{2}, FieldID: 1000, Name: "event_day",
Transform: iceberg.DayTransform{},
+ })
+ tbl, _ := newDynamicOverwriteTableWithSchema(t, schema, spec, "2")
+
+ seed := dynamicOverwriteJSONReader(t, tbl,
`[{"id":1,"event_date":"2026-02-10"},{"id":2,"event_date":"2026-02-11"},{"id":3,"event_date":"2026-02-10"}]`)
+ tx := tbl.NewTransaction()
+ require.NoError(t, tx.Append(t.Context(), seed, nil))
+ seed.Release()
+ tbl, err := tx.Commit(t.Context())
+ require.NoError(t, err)
+ require.ElementsMatch(t, []int64{1, 2, 3}, idsOnRef(t, tbl,
table.MainBranch))
+
+ replacement := dynamicOverwriteJSONReader(t, tbl,
`[{"id":4,"event_date":"2026-02-10"}]`)
+ updated, err := tbl.DynamicPartitionOverwrite(t.Context(), replacement,
nil)
+ replacement.Release()
+ require.NoError(t, err)
+ require.ElementsMatch(t, []int64{2, 4}, idsOnRef(t, updated,
table.MainBranch))
+}
+
+// newDynamicOverwriteV3MoRTable writes two rows in each partition so a
+// partial DELETE creates a deletion vector instead of dropping a whole file.
+func newDynamicOverwriteV3MoRTable(t *testing.T) (*table.Table,
*concurrentTestCatalog) {
+ t.Helper()
+ schema := iceberg.NewSchema(0,
+ iceberg.NestedField{ID: 1, Name: "id", Type:
iceberg.PrimitiveTypes.Int64, Required: true},
+ iceberg.NestedField{ID: 2, Name: "group", Type:
iceberg.PrimitiveTypes.Int64, Required: true},
+ )
+ spec := iceberg.NewPartitionSpec(iceberg.PartitionField{
+ SourceIDs: []int{2}, FieldID: 1000, Name: "group", Transform:
iceberg.IdentityTransform{},
+ })
+ tbl, cat := newDynamicOverwriteTableWithSchema(t, schema, spec, "3")
+ setMode := tbl.NewTransaction()
+ require.NoError(t,
setMode.SetProperties(iceberg.Properties{table.WriteDeleteModeKey:
table.WriteModeMergeOnRead}))
+ var err error
+ tbl, err = setMode.Commit(t.Context())
+ require.NoError(t, err)
+ seed := dynamicOverwriteJSONReader(t, tbl,
+
`[{"id":1,"group":10},{"id":2,"group":10},{"id":3,"group":20},{"id":4,"group":20}]`)
+ tbl, err = tbl.Append(t.Context(), seed, nil)
+ seed.Release()
+ require.NoError(t, err)
+ tasks, err := tbl.Scan().PlanFiles(t.Context())
+ require.NoError(t, err)
+ require.Len(t, tasks, 2, "each partition needs its own data file")
+
+ return tbl, cat
+}
+
+// The overwrite must remove the DV tied to the replaced data file, preserve
+// the DV for the untouched partition, and preserve row IDs for surviving rows.
+func TestDynamicPartitionOverwriteV3RemovesOnlyTouchedDeletionVectors(t
*testing.T) {
+ tbl, _ := newDynamicOverwriteV3MoRTable(t)
+ originalRowIDs := rowIDByID(t, tbl)
+ var err error
+ for _, id := range []int64{1, 3} {
+ tbl, err = tbl.Delete(t.Context(),
iceberg.EqualTo(iceberg.Reference("id"), id), nil)
+ require.NoError(t, err)
+ }
+ require.Equal(t, 2, liveDVCountOnRef(t, tbl, table.MainBranch))
+ require.ElementsMatch(t, []int64{2, 4}, idsOnRef(t, tbl,
table.MainBranch))
+ beforeData, _ := liveFilesOnRef(t, tbl, table.MainBranch)
+ require.Len(t, beforeData, 2)
+ replacement := dynamicOverwriteJSONReader(t, tbl,
`[{"id":5,"group":10}]`)
+ updated, err := tbl.DynamicPartitionOverwrite(t.Context(), replacement,
nil)
+ replacement.Release()
+ require.NoError(t, err)
+ require.ElementsMatch(t, []int64{4, 5}, idsOnRef(t, updated,
table.MainBranch))
+ require.Equal(t, 1, liveDVCountOnRef(t, updated, table.MainBranch))
+ afterData, _ := liveFilesOnRef(t, updated, table.MainBranch)
+ require.Len(t, afterData, 2)
+ var retained, replaced int
+ for _, path := range beforeData {
+ refs := map[string]struct{}{path: {}}
+ if slices.Contains(afterData, path) {
+ retained++
+ require.Len(t, deleteEntriesReferencing(t, updated,
refs), 1,
+ "the untouched data file must keep its deletion
vector")
+ } else {
+ replaced++
+ require.Empty(t, deleteEntriesReferencing(t, updated,
refs),
+ "the replaced data file must not keep its
deletion vector")
+ }
+ }
+ require.Equal(t, 1, retained)
+ require.Equal(t, 1, replaced)
+ rowIDs := rowIDByID(t, updated)
+ require.Equal(t, originalRowIDs[4], rowIDs[4], "untouched rows must
keep their _row_id")
+ require.Greater(t, rowIDs[5],
+ max(originalRowIDs[1], originalRowIDs[2], originalRowIDs[3],
originalRowIDs[4]),
+ "the replacement row must receive a fresh _row_id")
+}
+
+// Concurrent deletion vectors in replaced partitions must conflict, while
+// ones in other partitions must survive a refresh-and-replay of the overwrite.
+func TestDynamicPartitionOverwriteConcurrentDeletionVectors(t *testing.T) {
+ for _, tc := range []struct {
+ name string
+ deleteID int64
+ conflict bool
+ }{
+ {name: "touched partition", deleteID: 1, conflict: true},
+ {name: "other partition", deleteID: 3},
+ } {
+ t.Run(tc.name, func(t *testing.T) {
+ tbl, cat := newDynamicOverwriteV3MoRTable(t)
+ stale := tbl.NewTransaction()
+ replacement := dynamicOverwriteJSONReader(t, tbl,
`[{"id":5,"group":10}]`)
+ require.NoError(t,
stale.DynamicPartitionOverwrite(t.Context(), replacement, nil))
+ replacement.Release()
+ peer, err := tbl.Delete(t.Context(),
iceberg.EqualTo(iceberg.Reference("id"), tc.deleteID), nil)
+ require.NoError(t, err)
+ require.Equal(t, 1, liveDVCountOnRef(t, peer,
table.MainBranch),
+ "concurrent writer must add a deletion vector")
+ attemptsBefore := cat.attempts.Load()
+ updated, err := stale.Commit(t.Context())
+ if tc.conflict {
+ require.ErrorIs(t, err,
table.ErrConflictingDeleteFiles)
+ require.Equal(t, attemptsBefore+1,
cat.attempts.Load())
+ require.ElementsMatch(t, []int64{2, 3, 4},
idsOnRef(t, peer, table.MainBranch))
+
+ return
+ }
+ require.NoError(t, err)
+ require.Equal(t, attemptsBefore+2, cat.attempts.Load(),
+ "the independent delete must survive a refresh
and retry")
+ require.ElementsMatch(t, []int64{4, 5}, idsOnRef(t,
updated, table.MainBranch))
+ require.Equal(t, 1, liveDVCountOnRef(t, updated,
table.MainBranch))
+ })
+ }
+}
+
+// Count data files recursively to catch writes in existing partition
+// subdirectories as well as new ones.
+func dynamicOverwriteDataPaths(t *testing.T, tbl *table.Table) []string {
+ t.Helper()
+ var paths []string
+ err := filepath.WalkDir(filepath.Join(tbl.Location(), "data"),
func(path string, entry os.DirEntry, walkErr error) error {
+ if walkErr != nil {
+ return walkErr
+ }
+ if !entry.IsDir() {
+ paths = append(paths, filepath.ToSlash(path))
+ }
+
+ return nil
+ })
+ require.NoError(t, err)
+ slices.Sort(paths)
+
+ return paths
+}
+
+// Existing files use their own spec ID in the partition key, so an Arrow
+// overwrite with the current spec leaves old-spec rows untouched.
+func TestDynamicPartitionOverwriteArrowKeepsOldSpecFiles(t *testing.T) {
+ base, _ := newDynamicOverwriteTable(t)
+ tbl := appendRowsOnRef(t, base, table.MainBranch, 1, 2)
+ evolve := tbl.NewTransaction()
+ require.NoError(t, evolve.UpdateSpec(true).
+ RemoveField("id").
+ AddField("id", iceberg.BucketTransform{NumBuckets: 16},
"id_bucket").
+ Commit())
+ var err error
+ tbl, err = evolve.Commit(t.Context())
+ require.NoError(t, err)
+ before := dynamicOverwriteDataPaths(t, tbl)
+ require.NotEmpty(t, before, "test requires already-written data files")
+ replacement := idRecordReader(t, tbl, 3)
+ updated, err := tbl.DynamicPartitionOverwrite(t.Context(), replacement,
nil)
+ replacement.Release()
+ require.NoError(t, err)
+ require.Len(t, dynamicOverwriteDataPaths(t, updated), len(before)+1)
+ require.ElementsMatch(t, []int64{1, 2, 3}, idsOnRef(t, updated,
table.MainBranch))
+}
diff --git a/table/dynamic_partition_overwrite_test.go
b/table/dynamic_partition_overwrite_test.go
new file mode 100644
index 000000000..774c7ce49
--- /dev/null
+++ b/table/dynamic_partition_overwrite_test.go
@@ -0,0 +1,511 @@
+// 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"
+ "slices"
+ "testing"
+
+ "github.com/apache/iceberg-go"
+ iceio "github.com/apache/iceberg-go/io"
+ "github.com/stretchr/testify/require"
+)
+
+func TestDynamicPartitionMatcher(t *testing.T) {
+ spec := iceberg.NewPartitionSpecID(7,
+ iceberg.PartitionField{
+ SourceIDs: []int{1},
+ FieldID: 1000,
+ Name: "bucket_id",
+ Transform: iceberg.BucketTransform{NumBuckets: 16},
+ },
+ iceberg.PartitionField{
+ SourceIDs: []int{2},
+ FieldID: 1001,
+ Name: "region",
+ Transform: iceberg.IdentityTransform{},
+ },
+ )
+
+ files := []iceberg.DataFile{
+ newTestDataFile(t, spec, "file://new-1.parquet",
+ map[int]any{1000: int32(4), 1001: "north"}),
+ newTestDataFile(t, spec, "file://new-2.parquet",
+ map[int]any{1000: int32(4), 1001: "north"}),
+ newTestDataFile(t, spec, "file://new-3.parquet",
+ map[int]any{1000: int32(5), 1001: "south"}),
+ }
+ keys, err := dynamicPartitionKeys(files)
+ require.NoError(t, err)
+ require.Len(t, keys, 2)
+ for _, tc := range []struct {
+ partition map[int]any
+ want bool
+ }{
+ {map[int]any{1000: int32(4), 1001: "north"}, true},
+ {map[int]any{1000: int64(5), 1001: "south"}, true},
+ {map[int]any{1000: int32(4), 1001: "south"}, false},
+ {map[int]any{1000: int32(7), 1001: "north"}, false},
+ } {
+ key, err := canonicalPartitionKey(int32(spec.ID()),
tc.partition)
+ require.NoError(t, err)
+ _, found := keys[key]
+ require.Equal(t, tc.want, found)
+ }
+}
+
+func TestDynamicPartitionOverwriteDataFiles(t *testing.T) {
+ ctx := context.Background()
+ spec := partitionedSpec()
+ seed, fs := createTestTransactionWithMemIO(t, spec)
+
+ original := []iceberg.DataFile{
+ newTestDataFile(t, spec, "file://old-a.parquet",
map[int]any{1000: int32(1)}),
+ newTestDataFile(t, spec, "file://old-a2.parquet",
map[int]any{1000: int32(1)}),
+ newTestDataFile(t, spec, "file://old-b.parquet",
map[int]any{1000: int32(2)}),
+ }
+ require.NoError(t, seed.AddDataFiles(ctx, original, nil))
+ seedMeta, err := seed.meta.Build()
+ require.NoError(t, err)
+ tbl := New(Identifier{"db", "tbl"}, seedMeta, "metadata.json",
+ func(context.Context) (iceio.IO, error) { return fs, nil }, nil)
+
+ txn := tbl.NewTransaction()
+ replacement := []iceberg.DataFile{
+ newTestDataFile(t, spec, "file://new-a.parquet",
map[int]any{1000: int32(1)}),
+ newTestDataFile(t, spec, "file://new-a2.parquet",
map[int]any{1000: int32(1)}),
+ }
+
+ require.NoError(t, txn.DynamicPartitionOverwriteDataFiles(ctx,
replacement,
+ iceberg.Properties{"source": "dynamic-test"}))
+
+ updated, err := txn.meta.Build()
+ require.NoError(t, err)
+ snapshot := updated.CurrentSnapshot()
+ require.NotNil(t, snapshot)
+ require.NotNil(t, snapshot.Summary)
+ require.Equal(t, OpOverwrite, snapshot.Summary.Operation)
+ require.Equal(t, "true",
snapshot.Summary.Properties["replace-partitions"])
+ require.Equal(t, "dynamic-test", snapshot.Summary.Properties["source"])
+
+ var paths []string
+ for df, err := range snapshot.dataFiles(ctx, fs, nil) {
+ require.NoError(t, err)
+ if df.ContentType() == iceberg.EntryContentData {
+ paths = append(paths, df.FilePath())
+ }
+ }
+ slices.Sort(paths)
+ require.Equal(t, []string{"file://new-a.parquet",
"file://new-a2.parquet", "file://old-b.parquet"}, paths)
+ require.NotContains(t, txn.pinnedRefs, MainBranch, "existing branches
use partition-scoped conflict validation")
+}
+
+func TestDynamicPartitionOverwriteRejectsAlreadyLivePath(t *testing.T) {
+ spec := partitionedSpec()
+ seed, fs := createTestTransactionWithMemIO(t, spec)
+ existing := newTestDataFile(t, spec, "file://existing.parquet",
map[int]any{1000: int32(1)})
+ require.NoError(t, seed.AddDataFiles(t.Context(),
[]iceberg.DataFile{existing}, nil))
+ seedMeta, err := seed.meta.Build()
+ require.NoError(t, err)
+ tbl := New(Identifier{"db", "tbl"}, seedMeta, "metadata.json",
+ func(context.Context) (iceio.IO, error) { return fs, nil }, nil)
+
+ txn := tbl.NewTransaction()
+ err = txn.DynamicPartitionOverwriteDataFiles(t.Context(),
[]iceberg.DataFile{existing}, nil)
+ require.ErrorIs(t, err, ErrInvalidOperation)
+ require.ErrorContains(t, err, "already exists")
+ require.Empty(t, txn.meta.updates)
+ require.Empty(t, txn.pinnedRefs)
+}
+
+func TestDynamicPartitionOverwriteAppendsNewPartition(t *testing.T) {
+ ctx := context.Background()
+ spec := partitionedSpec()
+ txn, fs := createTestTransactionWithMemIO(t, spec)
+ newFile := newTestDataFile(t, spec, "file://first.parquet",
map[int]any{1000: int32(10)})
+
+ require.NoError(t, txn.DynamicPartitionOverwriteDataFiles(ctx,
[]iceberg.DataFile{newFile}, nil))
+ metadata, err := txn.meta.Build()
+ require.NoError(t, err)
+ snapshot := metadata.CurrentSnapshot()
+ require.NotNil(t, snapshot)
+ var files []string
+ for df, err := range snapshot.dataFiles(ctx, fs, nil) {
+ require.NoError(t, err)
+ files = append(files, df.FilePath())
+ }
+ require.Equal(t, []string{"file://first.parquet"}, files)
+ require.Equal(t, OpOverwrite, snapshot.Summary.Operation)
+ require.Equal(t, "true",
snapshot.Summary.Properties["replace-partitions"])
+ require.Contains(t, txn.pinnedRefs, MainBranch)
+}
+
+func TestDynamicPartitionOverwriteUnpartitionedReplacesAllFiles(t *testing.T) {
+ ctx := context.Background()
+ spec := iceberg.NewPartitionSpec()
+ seed, fs := createTestTransactionWithMemIO(t, spec)
+ original := []iceberg.DataFile{
+ newTestDataFile(t, spec, "file://old-1.parquet", nil),
+ newTestDataFile(t, spec, "file://old-2.parquet", nil),
+ }
+ require.NoError(t, seed.AddDataFiles(ctx, original, nil))
+ seedMeta, err := seed.meta.Build()
+ require.NoError(t, err)
+ tbl := New(Identifier{"db", "tbl"}, seedMeta, "metadata.json",
+ func(context.Context) (iceio.IO, error) { return fs, nil }, nil)
+
+ txn := tbl.NewTransaction()
+ replacement := newTestDataFile(t, spec, "file://replacement.parquet",
nil)
+ require.NoError(t, txn.DynamicPartitionOverwriteDataFiles(ctx,
[]iceberg.DataFile{replacement}, nil))
+ updated, err := txn.meta.Build()
+ require.NoError(t, err)
+ snapshot := updated.CurrentSnapshot()
+ require.Equal(t, OpOverwrite, snapshot.Summary.Operation)
+ require.Equal(t, "true",
snapshot.Summary.Properties["replace-partitions"])
+ var paths []string
+ for df, err := range snapshot.dataFiles(ctx, fs, nil) {
+ require.NoError(t, err)
+ if df.ContentType() == iceberg.EntryContentData {
+ paths = append(paths, df.FilePath())
+ }
+ }
+ require.Equal(t, []string{"file://replacement.parquet"}, paths)
+}
+
+func TestDynamicPartitionOverwriteEmptyInputIsNoOp(t *testing.T) {
+ txn, _ := createTestTransactionWithMemIO(t, partitionedSpec())
+ require.NoError(t, txn.DynamicPartitionOverwriteDataFiles(t.Context(),
nil, nil))
+ require.Empty(t, txn.meta.updates)
+ require.Empty(t, txn.pinnedRefs)
+}
+
+func TestDynamicPartitionOverwriteRespectsCanceledContext(t *testing.T) {
+ spec := partitionedSpec()
+ txn, _ := createTestTransactionWithMemIO(t, spec)
+ ctx, cancel := context.WithCancel(t.Context())
+ cancel()
+
+ file := newTestDataFile(t, spec, "file://canceled.parquet",
map[int]any{1000: int32(5)})
+ err := txn.DynamicPartitionOverwriteDataFiles(ctx,
[]iceberg.DataFile{file}, nil)
+ require.ErrorIs(t, err, context.Canceled)
+ require.Empty(t, txn.meta.updates)
+ require.Empty(t, txn.pinnedRefs)
+}
+
+func TestDynamicPartitionOverwriteV3RowLineage(t *testing.T) {
+ ctx := context.Background()
+ spec := partitionedSpec()
+ txn, _ := createTestTransactionWithMemIO(t, spec)
+ require.NoError(t, txn.meta.SetFormatVersion(3))
+ file := newTestDataFile(t, spec, "file://v3.parquet", map[int]any{1000:
int32(1)})
+ require.NoError(t, txn.DynamicPartitionOverwriteDataFiles(ctx,
[]iceberg.DataFile{file}, nil))
+ meta, err := txn.meta.Build()
+ require.NoError(t, err)
+ snapshot := meta.CurrentSnapshot()
+ require.NotNil(t, snapshot)
+ require.Equal(t, OpOverwrite, snapshot.Summary.Operation)
+ require.NotNil(t, snapshot.FirstRowID)
+ require.NotNil(t, snapshot.AddedRows)
+}
+
+func TestDynamicPartitionOverwriteRejectsUnsupportedLayouts(t *testing.T) {
+ ctx := context.Background()
+ t.Run("wrong incoming spec", func(t *testing.T) {
+ spec := partitionedSpec()
+ txn, _ := createTestTransactionWithMemIO(t, spec)
+ otherSpec := iceberg.NewPartitionSpec(iceberg.PartitionField{
+ SourceIDs: []int{1}, FieldID: 1001, Name: "bucket_id",
Transform: iceberg.BucketTransform{NumBuckets: 16},
+ })
+ require.NoError(t, txn.meta.AddPartitionSpec(&otherSpec, false))
+ require.NoError(t, txn.meta.SetDefaultSpecID(-1))
+ updatesBefore := slices.Clone(txn.meta.updates)
+ file := newTestDataFile(t, spec, "file://wrong-spec.parquet",
map[int]any{1000: int32(1)})
+ err := txn.DynamicPartitionOverwriteDataFiles(ctx,
[]iceberg.DataFile{file}, nil)
+ require.ErrorIs(t, err, ErrInvalidOperation)
+ require.ErrorContains(t, err, "current spec")
+ require.Equal(t, updatesBefore, txn.meta.updates)
+ })
+
+ t.Run("nil transaction", func(t *testing.T) {
+ var txn *Transaction
+ require.Error(t, txn.DynamicPartitionOverwriteDataFiles(ctx,
nil, nil))
+ })
+}
+
+func TestDynamicPartitionOverwriteV1(t *testing.T) {
+ ctx := t.Context()
+ spec := partitionedSpec()
+ seed, fs := createTestTransactionWithMemIO(t, spec)
+ seed.meta.formatVersion = 1
+ old := []iceberg.DataFile{
+ newTestDataFile(t, spec, "file://v1-old-a.parquet",
map[int]any{1000: int32(1)}),
+ newTestDataFile(t, spec, "file://v1-old-b.parquet",
map[int]any{1000: int32(2)}),
+ }
+ require.NoError(t, seed.AddDataFiles(ctx, old, nil))
+ base, err := seed.meta.Build()
+ require.NoError(t, err)
+ require.Equal(t, 1, base.FormatVersion())
+ tbl := New(Identifier{"db", "tbl"}, base, "metadata.json",
func(context.Context) (iceio.IO, error) { return fs, nil }, nil)
+ tx := tbl.NewTransaction()
+ replacement := newTestDataFile(t, spec, "file://v1-new-a.parquet",
map[int]any{1000: int32(1)})
+ require.NoError(t, tx.DynamicPartitionOverwriteDataFiles(ctx,
[]iceberg.DataFile{replacement}, nil))
+ updated, err := tx.meta.Build()
+ require.NoError(t, err)
+ snapshot := updated.CurrentSnapshot()
+ require.NotNil(t, snapshot)
+ require.Equal(t, OpOverwrite, snapshot.Summary.Operation)
+ var paths []string
+ for file, err := range snapshot.dataFiles(ctx, fs, nil) {
+ require.NoError(t, err)
+ if file.ContentType() == iceberg.EntryContentData {
+ paths = append(paths, file.FilePath())
+ }
+ }
+ require.ElementsMatch(t, []string{"file://v1-new-a.parquet",
"file://v1-old-b.parquet"}, paths)
+}
+
+func TestDynamicPartitionOverwriteKeepsLiveOldSpec(t *testing.T) {
+ ctx := t.Context()
+ oldSpec := partitionedSpec()
+ seed, fs := createTestTransactionWithMemIO(t, oldSpec)
+ old := newTestDataFile(t, oldSpec, "file://old-spec.parquet",
map[int]any{1000: int32(1)})
+ require.NoError(t, seed.AddDataFiles(ctx, []iceberg.DataFile{old}, nil))
+ oldDelete := newTestPosDeleteFileForSpec(t, oldSpec,
"file://old-spec-delete.parquet",
+ map[int]any{1000: int32(1)}, old.FilePath())
+ require.NoError(t,
seed.NewRowDelta(nil).AddDeletes(oldDelete).Commit(ctx))
+ base, err := seed.meta.Build()
+ require.NoError(t, err)
+
+ tbl := New(Identifier{"db", "tbl"}, base, "metadata.json",
+ func(context.Context) (iceio.IO, error) { return fs, nil }, nil)
+ evolve := tbl.NewTransaction()
+ nextSpec := iceberg.NewPartitionSpec(iceberg.PartitionField{
+ SourceIDs: []int{1}, FieldID: 1001, Name: "bucket_id",
Transform: iceberg.BucketTransform{NumBuckets: 16},
+ })
+ require.NoError(t, evolve.meta.AddPartitionSpec(&nextSpec, false))
+ require.NoError(t, evolve.meta.SetDefaultSpecID(-1))
+ evolved, err := evolve.meta.Build()
+ require.NoError(t, err)
+
+ txn := New(Identifier{"db", "tbl"}, evolved, "metadata.json",
+ func(context.Context) (iceio.IO, error) { return fs, nil },
nil).NewTransaction()
+ currentSpec := evolved.PartitionSpec()
+ var partitionFieldID int
+ for _, field := range currentSpec.Fields() {
+ partitionFieldID = field.FieldID
+
+ break
+ }
+ require.NotZero(t, partitionFieldID)
+ replacement := newTestDataFile(t, currentSpec,
"file://new-spec.parquet",
+ map[int]any{partitionFieldID: int32(1)})
+ require.NoError(t, txn.DynamicPartitionOverwriteDataFiles(ctx,
[]iceberg.DataFile{replacement}, nil))
+ updated, err := txn.meta.Build()
+ require.NoError(t, err)
+ snapshot := updated.CurrentSnapshot()
+ require.NotNil(t, snapshot)
+
+ var liveData []string
+ for file, err := range snapshot.dataFiles(ctx, fs, nil) {
+ require.NoError(t, err)
+ if file.ContentType() == iceberg.EntryContentData {
+ liveData = append(liveData, file.FilePath())
+ }
+ }
+ require.ElementsMatch(t, []string{"file://old-spec.parquet",
"file://new-spec.parquet"}, liveData)
+
+ var liveDeletes []string
+ for entry, err := range snapshot.entries(ctx, fs,
iceberg.ManifestContentDeletes, true) {
+ require.NoError(t, err)
+ liveDeletes = append(liveDeletes, entry.DataFile().FilePath())
+ }
+ require.Equal(t, []string{"file://old-spec-delete.parquet"},
liveDeletes)
+}
+
+func TestDynamicPartitionOverwriteUnpartitionedAfterSpecEvolution(t
*testing.T) {
+ ctx := t.Context()
+ oldSpec := partitionedSpec()
+ seed, fs := createTestTransactionWithMemIO(t, oldSpec)
+ oldData := newTestDataFile(t, oldSpec, "file://old-spec.parquet",
map[int]any{1000: int32(1)})
+ require.NoError(t, seed.AddDataFiles(ctx, []iceberg.DataFile{oldData},
nil))
+ base, err := seed.meta.Build()
+ require.NoError(t, err)
+
+ fsF := func(context.Context) (iceio.IO, error) { return fs, nil }
+ evolve := New(Identifier{"db", "tbl"}, base, "metadata.json", fsF,
nil).NewTransaction()
+ emptySpec := iceberg.NewPartitionSpec()
+ require.NoError(t, evolve.meta.AddPartitionSpec(&emptySpec, false))
+ require.NoError(t, evolve.meta.SetDefaultSpecID(-1))
+ evolved, err := evolve.meta.Build()
+ require.NoError(t, err)
+ currentSpec := evolved.PartitionSpec()
+ require.True(t, currentSpec.IsUnpartitioned())
+
+ // Equality deletes written with an unpartitioned spec are global.
+ builder, err := iceberg.NewDataFileBuilder(currentSpec,
iceberg.EntryContentEqDeletes,
+ "file://global-delete.parquet", iceberg.ParquetFile, nil, nil,
nil, 1, 1)
+ require.NoError(t, err)
+ eqDelete := builder.EqualityFieldIDs([]int{1}).Build()
+ addDelete := New(Identifier{"db", "tbl"}, evolved, "metadata.json",
fsF, nil).NewTransaction()
+ require.NoError(t,
addDelete.NewRowDelta(nil).AddDeletes(eqDelete).Commit(ctx))
+ withDelete, err := addDelete.meta.Build()
+ require.NoError(t, err)
+
+ overwrite := New(Identifier{"db", "tbl"}, withDelete, "metadata.json",
fsF, nil).NewTransaction()
+ replacement := newTestDataFile(t, currentSpec,
"file://replacement.parquet", nil)
+ require.NoError(t, overwrite.DynamicPartitionOverwriteDataFiles(ctx,
[]iceberg.DataFile{replacement}, nil))
+ updated, err := overwrite.meta.Build()
+ require.NoError(t, err)
+
+ var dataPaths []string
+ for df, err := range updated.CurrentSnapshot().dataFiles(ctx, fs, nil) {
+ require.NoError(t, err)
+ dataPaths = append(dataPaths, df.FilePath())
+ }
+ require.Equal(t, []string{"file://replacement.parquet"}, dataPaths,
+ "unpartitioned overwrite must replace data under old specs too")
+
+ var deletePaths []string
+ for entry, err := range updated.CurrentSnapshot().entries(ctx, fs,
iceberg.ManifestContentDeletes, true) {
+ require.NoError(t, err)
+ deletePaths = append(deletePaths, entry.DataFile().FilePath())
+ }
+ require.Empty(t, deletePaths, "no global delete may survive a
full-table replacement")
+}
+
+func TestDynamicPartitionOverwriteKeepsDeleteForSurvivingOldSpecData(t
*testing.T) {
+ ctx := t.Context()
+ oldSpec := partitionedSpec()
+ seed, fs := createTestTransactionWithMemIO(t, oldSpec)
+ oldData := newTestDataFile(t, oldSpec, "file://old-spec.parquet",
map[int]any{1000: int32(1)})
+ require.NoError(t, seed.AddDataFiles(ctx, []iceberg.DataFile{oldData},
nil))
+ base, err := seed.meta.Build()
+ require.NoError(t, err)
+
+ fsF := func(context.Context) (iceio.IO, error) { return fs, nil }
+ evolve := New(Identifier{"db", "tbl"}, base, "metadata.json", fsF,
nil).NewTransaction()
+ newSpec := iceberg.NewPartitionSpec(iceberg.PartitionField{
+ SourceIDs: []int{1}, FieldID: 1001, Name: "id_bucket",
Transform: iceberg.BucketTransform{NumBuckets: 16},
+ })
+ require.NoError(t, evolve.meta.AddPartitionSpec(&newSpec, false))
+ require.NoError(t, evolve.meta.SetDefaultSpecID(-1))
+ evolved, err := evolve.meta.Build()
+ require.NoError(t, err)
+ currentSpec := evolved.PartitionSpec()
+ var fieldID int
+ for _, field := range currentSpec.Fields() {
+ fieldID = field.FieldID
+
+ break
+ }
+ require.NotZero(t, fieldID)
+ nullPartition := map[int]any{fieldID: nil}
+
+ // A v2 position delete can target an old-spec data file even if its
+ // manifest stores it under a new partition tuple with a null value.
+ posDelete := newTestPosDeleteFileForSpec(t, currentSpec,
"file://old-spec-delete.parquet",
+ nullPartition, oldData.FilePath())
+ addDelete := New(Identifier{"db", "tbl"}, evolved, "metadata.json",
fsF, nil).NewTransaction()
+ require.NoError(t,
addDelete.NewRowDelta(nil).AddDeletes(posDelete).Commit(ctx))
+ withDelete, err := addDelete.meta.Build()
+ require.NoError(t, err)
+
+ overwrite := New(Identifier{"db", "tbl"}, withDelete, "metadata.json",
fsF, nil).NewTransaction()
+ replacement := newTestDataFile(t, currentSpec,
"file://new-spec.parquet", nullPartition)
+ require.NoError(t, overwrite.DynamicPartitionOverwriteDataFiles(ctx,
[]iceberg.DataFile{replacement}, nil))
+ updated, err := overwrite.meta.Build()
+ require.NoError(t, err)
+
+ var dataPaths []string
+ for df, err := range updated.CurrentSnapshot().dataFiles(ctx, fs, nil) {
+ require.NoError(t, err)
+ if df.ContentType() == iceberg.EntryContentData {
+ dataPaths = append(dataPaths, df.FilePath())
+ }
+ }
+ require.ElementsMatch(t, []string{oldData.FilePath(),
replacement.FilePath()}, dataPaths)
+
+ var deletePaths []string
+ for entry, err := range updated.CurrentSnapshot().entries(ctx, fs,
iceberg.ManifestContentDeletes, true) {
+ require.NoError(t, err)
+ deletePaths = append(deletePaths, entry.DataFile().FilePath())
+ }
+ require.Equal(t, []string{posDelete.FilePath()}, deletePaths,
+ "deletes for surviving old-spec data must not be dropped")
+}
+
+func TestDynamicPartitionOverwriteWithoutAutoNameMapping(t *testing.T) {
+ ctx := t.Context()
+ spec := partitionedSpec()
+ txn, _ := createTestTransactionWithMemIO(t, spec)
+ file := newTestDataFile(t, spec, "file://no-name-mapping.parquet",
map[int]any{1000: int32(1)})
+ require.NoError(t, txn.DynamicPartitionOverwriteDataFiles(ctx,
[]iceberg.DataFile{file}, nil, WithoutAutoNameMapping()))
+ updated, err := txn.meta.Build()
+ require.NoError(t, err)
+ require.Nil(t, updated.NameMapping())
+}
+
+func TestDynamicPartitionOverwriteRejectsStagedDefaultSpecChange(t *testing.T)
{
+ baseSpec := iceberg.NewPartitionSpec()
+ txn, _ := createTestTransactionWithMemIO(t, baseSpec)
+ changedSpec := iceberg.NewPartitionSpec(iceberg.PartitionField{
+ SourceIDs: []int{1}, FieldID: 1000, Name: "id", Transform:
iceberg.IdentityTransform{},
+ })
+ require.NoError(t, txn.meta.AddPartitionSpec(&changedSpec, false))
+ require.NoError(t, txn.meta.SetDefaultSpecID(-1))
+ spec, err := txn.meta.CurrentSpec()
+ require.NoError(t, err)
+ file := newTestDataFile(t, *spec, "file://staged-spec.parquet",
+ map[int]any{1000: int32(1)})
+ err = txn.DynamicPartitionOverwriteDataFiles(t.Context(),
[]iceberg.DataFile{file}, nil)
+ require.ErrorIs(t, err, ErrCommitFailed)
+ require.ErrorContains(t, err, "default spec id has changed")
+}
+
+func TestDynamicPartitionOverwriteRemovesOnlyTouchedPartitionDeletes(t
*testing.T) {
+ ctx := t.Context()
+ spec := partitionedSpec()
+ seed, fs := createTestTransactionWithMemIO(t, spec)
+ dataA := newTestDataFile(t, spec, "file://old-a.parquet",
map[int]any{1000: int32(1)})
+ dataB := newTestDataFile(t, spec, "file://old-b.parquet",
map[int]any{1000: int32(2)})
+ require.NoError(t, seed.AddDataFiles(ctx, []iceberg.DataFile{dataA,
dataB}, nil))
+ base, err := seed.meta.Build()
+ require.NoError(t, err)
+ fsF := func(context.Context) (iceio.IO, error) { return fs, nil }
+ tbl := New(Identifier{"db", "tbl"}, base, "metadata.json", fsF, nil)
+
+ deleteA := newTestPosDeleteFileForSpec(t, spec,
"file://delete-a.parquet", map[int]any{1000: int32(1)}, dataA.FilePath())
+ deleteB := newTestPosDeleteFileForSpec(t, spec,
"file://delete-b.parquet", map[int]any{1000: int32(2)}, dataB.FilePath())
+ addDeletes := tbl.NewTransaction()
+ require.NoError(t, addDeletes.NewRowDelta(nil).AddDeletes(deleteA,
deleteB).Commit(ctx))
+ withDeletes, err := addDeletes.meta.Build()
+ require.NoError(t, err)
+
+ overwrite := New(Identifier{"db", "tbl"}, withDeletes, "metadata.json",
fsF, nil).NewTransaction()
+ replacement := newTestDataFile(t, spec, "file://new-a.parquet",
map[int]any{1000: int32(1)})
+ require.NoError(t, overwrite.DynamicPartitionOverwriteDataFiles(ctx,
[]iceberg.DataFile{replacement}, nil))
+ updated, err := overwrite.meta.Build()
+ require.NoError(t, err)
+ snapshot := updated.CurrentSnapshot()
+ var liveDeletes []string
+ for entry, err := range snapshot.entries(ctx, fs,
iceberg.ManifestContentDeletes, true) {
+ require.NoError(t, err)
+ liveDeletes = append(liveDeletes, entry.DataFile().FilePath())
+ }
+ require.Equal(t, []string{"file://delete-b.parquet"}, liveDeletes)
+}
diff --git a/table/internal/partition_predicate.go
b/table/internal/partition_predicate.go
index 609d5e627..c47ceecdc 100644
--- a/table/internal/partition_predicate.go
+++ b/table/internal/partition_predicate.go
@@ -29,10 +29,10 @@ import (
)
// BuildPartitionMatchPredicate constructs a BooleanExpression matching all
rows
-// that belong to any of the given partitions. It is the core of dynamic
-// partition overwrite (see https://github.com/apache/iceberg-go/issues/1216):
-// the returned expression selects the existing data to delete before the new
-// data is appended.
+// that belong to any of the given partitions. It can be used by row-filter
+// operations involving partition transforms (issue #1216), but dynamic
+// partition overwrite now matches stored partition tuples directly.
+// Partial-file rewrites still need strict partition matching (issue #1215).
//
// partitions holds partition tuples as returned by DataFile.Partition(), each
// keyed by partition-field ID and carrying the field's (transformed) value.
The
@@ -46,7 +46,7 @@ import (
//
// Duplicate tuples collapse to a single clause, and an empty input yields
// AlwaysFalse (matching nothing). Callers are expected to pass a partitioned
-// spec; dynamic partition overwrite rejects unpartitioned tables upstream.
+// spec; unpartitioned dynamic overwrites do not need a row predicate.
// Void fields are also accepted when their source column has been dropped (the
// spec represents those tombstones with source ID 0), because void always
// produces a null partition value and does not need a source column.
diff --git a/website/src/api.md b/website/src/api.md
index 954a38358..9e351c05c 100644
--- a/website/src/api.md
+++ b/website/src/api.md
@@ -359,6 +359,41 @@ newTbl, err = tbl.Overwrite(ctx, recordReader, nil,
`OverwriteTable` is the `arrow.Table` variant.
+### Dynamic partition overwrite
+
+Replace only the partitions present in the incoming data. Unlike a filtered
+row-level overwrite, this removes **all existing data files** in those
partitions
+while retaining files in other partitions.
+
+```go
+// Write Arrow record batches, replacing only partitions receiving new rows.
+newTbl, err := tbl.DynamicPartitionOverwrite(ctx, recordReader, nil)
+
+// In-memory Arrow tables work too.
+newTbl, err = tbl.DynamicPartitionOverwriteTable(ctx, arrowTbl, 1024, nil)
+
+// Or commit already-written Iceberg DataFile metadata through a transaction.
+txn := tbl.NewTransaction()
+err = txn.DynamicPartitionOverwriteDataFiles(ctx, newDataFiles, nil)
+if err == nil {
+ newTbl, err = txn.Commit(ctx)
+}
+```
+
+This supports Iceberg format v1-v3. Incoming files must use the current
+partition spec. On partitioned tables, files under older specs are preserved;
+only matching current-spec partitions are replaced. If the current spec is
+unpartitioned, non-empty input replaces all live files, including files under
+older specs. For v3, replacement also removes deletion vectors tied to
+replaced data files. Empty input leaves the table untouched.
+
+A concurrent add or removal in a replaced partition fails validation, while
+independent changes to other partitions can be retried after a catalog refresh.
+A new target branch also pins the source branch to prevent missing concurrent
+writes. Default partition-spec and sort-order changes fail validation. Written
+data files are not automatically cleaned up if staging or commit fails, so
+callers must handle those files before retrying with a new transaction.
+
### Delete
```go
