badalprasadsingh commented on code in PR #2381: URL: https://github.com/apache/iceberg-go/pull/2381#discussion_r4478832808
########## table/dynamic_partition_overwrite_integration_test.go: ########## @@ -0,0 +1,221 @@ +// 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 ( + "path/filepath" + "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.WriteUpdateIsolationLevelKey: string(table.IsolationSnapshot), + 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)) +} + +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)) +} + +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) { Review Comment: the doc promises removal of deletion vectors, but no test checks it. If `dvs.removeDataFile` broke, the old dv would stay live unnoticed. please add a v3 test with dvs in two partitions, overwrite one, and assert only its dv is gone and `_row_ids` stay right? ########## table/dynamic_partition_overwrite.go: ########## @@ -0,0 +1,445 @@ +// 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" + iceio "github.com/apache/iceberg-go/io" + "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 +} + +// checkDynamicOverwriteExistingSpecs rejects live files under older specs +// before an Arrow writer creates data files. Different partition transforms +// cannot be compared safely by comparing their stored partition tuples. +func (t *Transaction) checkDynamicOverwriteExistingSpecs(ctx context.Context, fs iceio.IO, specID int) error { + snapshot := t.planningSnapshot(t.meta) + if snapshot == nil { + return nil + } + manifests, err := snapshot.Manifests(fs) + if err != nil { + return fmt.Errorf("dynamic partition overwrite: read manifests: %w", err) + } + for _, manifest := range manifests { + if err := ctx.Err(); err != nil { + return err + } + if int(manifest.PartitionSpecID()) == specID { + continue + } + for entry, err := range manifest.Entries(fs, true) { + if err != nil { + return fmt.Errorf("dynamic partition overwrite: read manifest %s: %w", manifest.FilePath(), err) + } + + return fmt.Errorf("%w: dynamic partition overwrite cannot replace live file %q under partition spec %d (current %d)", + ErrInvalidOperation, entry.DataFile().FilePath(), manifest.PartitionSpecID(), specID) + } + } + + return 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, 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.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 + } + if int(file.SpecID()) != specID { + return fmt.Errorf("%w: concurrent snapshot %d changed file %q under partition spec %d (current %d)", + conflict, snapshot.SnapshotID, file.FilePath(), file.SpecID(), specID) + } Review Comment: no test has another writer add a delete file, so nothing reaches `ErrConflictingDeleteFiles` or this spec check. please add two cases: a concurrent delete in a touched partition should fail, and one in another partition should replay ########## table/dynamic_partition_overwrite_integration_test.go: ########## @@ -0,0 +1,221 @@ +// 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 ( + "path/filepath" + "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.WriteUpdateIsolationLevelKey: string(table.IsolationSnapshot), Review Comment: I suggested this setting when the pin wsa the guard, but it hides a gap now: no test needs `setSkipDefaultValidator(true)`, and without that call, an append to another partition fails at default isolation please remove the line ########## table/dynamic_partition_overwrite.go: ########## @@ -0,0 +1,445 @@ +// 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" + iceio "github.com/apache/iceberg-go/io" + "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 +} + +// checkDynamicOverwriteExistingSpecs rejects live files under older specs +// before an Arrow writer creates data files. Different partition transforms +// cannot be compared safely by comparing their stored partition tuples. +func (t *Transaction) checkDynamicOverwriteExistingSpecs(ctx context.Context, fs iceio.IO, specID int) error { + snapshot := t.planningSnapshot(t.meta) + if snapshot == nil { + return nil + } + manifests, err := snapshot.Manifests(fs) + if err != nil { + return fmt.Errorf("dynamic partition overwrite: read manifests: %w", err) + } + for _, manifest := range manifests { + if err := ctx.Err(); err != nil { + return err + } + if int(manifest.PartitionSpecID()) == specID { + continue + } + for entry, err := range manifest.Entries(fs, true) { + if err != nil { + return fmt.Errorf("dynamic partition overwrite: read manifest %s: %w", manifest.FilePath(), err) + } + + return fmt.Errorf("%w: dynamic partition overwrite cannot replace live file %q under partition spec %d (current %d)", + ErrInvalidOperation, entry.DataFile().FilePath(), manifest.PartitionSpecID(), specID) + } + } + + return 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, 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.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 + } + if int(file.SpecID()) != specID { + return fmt.Errorf("%w: concurrent snapshot %d changed file %q under partition spec %d (current %d)", + conflict, snapshot.SnapshotID, file.FilePath(), file.SpecID(), specID) + } + 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 and existing live files must use the table's current partition spec. +// 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 + } + if err := t.checkDynamicOverwriteExistingSpecs(ctx, fs, spec.ID()); 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 + validationPlan := newPartitionValidationPlan(spec) + if snapshot := t.planningSnapshot(t.meta); snapshot != nil { + for df, err := range snapshot.dataFiles(fs, nil) { + if err != nil { + return 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() { + return fmt.Errorf("%w: dynamic partition overwrite cannot replace data spanning partition spec %d (current %d)", + ErrInvalidOperation, df.SpecID(), spec.ID()) + } + if err := validationPlan.validate(df); err != nil { + return fmt.Errorf("%w: invalid live partition in file %q: %w", + ErrInvalidOperation, df.FilePath(), err) + } + 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) + } + } + + // A replaced partition has no surviving data that its old equality + // or position deletes should affect. Deletion vectors are tied to a + // particular data file and are removed separately with that file. + for entry, err := range snapshot.entries(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 int(df.SpecID()) != spec.ID() { + return fmt.Errorf("%w: dynamic partition overwrite cannot replace deletes under spec %d (current %d)", + ErrInvalidOperation, df.SpecID(), spec.ID()) + } + 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(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(), 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 rejects live files under other 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) + } + spec, err := t.dynamicOverwriteSpec() + if 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 + } + if err := t.checkDynamicOverwriteExistingSpecs(ctx, fs, spec.ID()); err != nil { + return err + } Review Comment: this fixes the orphan files, but no test protects it, since `TestDynamicPartitionOverwriteRejectsLiveOldSpec` calls the `DataFiles` method. could you add an Arrow case on a table with an older-spec file, and assert no new files appear in the data directory? that would be really helpful -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
