badalprasadsingh commented on code in PR #2381: URL: https://github.com/apache/iceberg-go/pull/2381#discussion_r4459599592
########## table/dynamic_partition_overwrite.go: ########## @@ -0,0 +1,290 @@ +// 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 { + // Row lineage in v3 needs dedicated first-row-ID reassignment when + // replacing partitions. Do not use rewrite semantics to bypass it. + return nil, fmt.Errorf("%w: dynamic partition overwrite does not support format v3 row lineage", ErrInvalidOperation) + } + + 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 +} + +// DynamicPartitionOverwriteDataFiles atomically replaces all live data files +// in the partitions represented by filesToAdd. Other partitions are unchanged. +// An unpartitioned table has one logical partition, so a non-empty input +// replaces all of its data, matching Java's ReplacePartitions operation. +// +// The added files must use the table's current partition spec. Existing live +// data files must use that spec as well: overlapping partitions from an older +// spec cannot be identified by comparing tuples from different transforms. +// Rather than silently leave stale rows, this operation rejects mixed-spec +// tables. It supports identity and transformed partition fields alike. +// +// The operation pins the target branch head and default table shape. A +// concurrent snapshot commit or partition evolution fails instead of silently +// leaving newly appended files in a replaced partition. This also means the +// method should run on a fresh transaction, before other staged snapshot or +// schema/spec changes. +// +// An empty input is a no-op. Format-v3 tables are rejected. All files must be pre-written and have valid DataFile metadata, +// just as with AddDataFiles. Uncommitted files remain the caller's responsibility +// if staging or the eventual transaction commit fails. +func (t *Transaction) DynamicPartitionOverwriteDataFiles( + ctx context.Context, filesToAdd []iceberg.DataFile, snapshotProps iceberg.Properties, +) error { + spec, err := t.dynamicOverwriteSpec() + if err != nil { + return err + } + if len(filesToAdd) == 0 { + return nil + } + if err := ctx.Err(); err != nil { + return err + } + + addedPaths, err := t.validateDataFilesToAdd(filesToAdd, "DynamicPartitionOverwriteDataFiles", true) + 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 + } + + keys, err := dynamicPartitionKeys(filesToAdd) + if err != nil { + return err + } + var toDelete []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) + } + } + } + + if err := ctx.Err(); err != nil { + return err + } + + // A ref fence is important even if no files matched: an append to + // touched partition between planning and commit must not survive. + if err := t.AssertDefaultShape(); err != nil { + return err + } + if err := t.AssertRefSnapshotID(""); err != nil { + return err + } Review Comment: This pin misses a branch that doesn't exist yet. On the SQL catalog, if someone appends to the same partition on main first, the retry keeps their file, and Commit returns nil. Could you also pin main with `t.AssertRefSnapshotID(MainBranch)` when the branch is absent? ########## table/dynamic_partition_overwrite.go: ########## @@ -0,0 +1,290 @@ +// 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 { + // Row lineage in v3 needs dedicated first-row-ID reassignment when + // replacing partitions. Do not use rewrite semantics to bypass it. + return nil, fmt.Errorf("%w: dynamic partition overwrite does not support format v3 row lineage", ErrInvalidOperation) + } + + 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 +} + +// DynamicPartitionOverwriteDataFiles atomically replaces all live data files +// in the partitions represented by filesToAdd. Other partitions are unchanged. +// An unpartitioned table has one logical partition, so a non-empty input +// replaces all of its data, matching Java's ReplacePartitions operation. +// +// The added files must use the table's current partition spec. Existing live +// data files must use that spec as well: overlapping partitions from an older +// spec cannot be identified by comparing tuples from different transforms. +// Rather than silently leave stale rows, this operation rejects mixed-spec +// tables. It supports identity and transformed partition fields alike. +// +// The operation pins the target branch head and default table shape. A +// concurrent snapshot commit or partition evolution fails instead of silently +// leaving newly appended files in a replaced partition. This also means the Review Comment: any commit on the branch fails this overwrite, even an append to another partition, so a long backfill on a busy table may never commit. can we pin only a missing branch and add a validator that rejects only files added to touched partitions? ########## table/dynamic_partition_overwrite_integration_test.go: ########## @@ -0,0 +1,71 @@ +// 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/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) { + t.Helper() + location := filepath.ToSlash(t.TempDir()) + schema := iceberg.NewSchema(0, iceberg.NestedField{ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int64, Required: true}) + spec := iceberg.NewPartitionSpec(iceberg.PartitionField{SourceIDs: []int{1}, FieldID: 1000, Name: "id", Transform: iceberg.IdentityTransform{}}) + meta, err := table.NewMetadata(schema, &spec, table.UnsortedSortOrder, location, iceberg.Properties{ + table.PropertyFormatVersion: "2", + 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() Review Comment: this test can't prove the pin works, at default serializable isolation, the overwrite's own validator also rejects this append, with or without the pin. could you set `table.WriteUpdateIsolationLevelKey`: "snapshot" in `newDynamicOverwriteTable`, so the pin is the only guard? ########## table/dynamic_partition_overwrite.go: ########## @@ -0,0 +1,290 @@ +// 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 { + // Row lineage in v3 needs dedicated first-row-ID reassignment when + // replacing partitions. Do not use rewrite semantics to bypass it. + return nil, fmt.Errorf("%w: dynamic partition overwrite does not support format v3 row lineage", ErrInvalidOperation) + } Review Comment: IMO, I don't think v3 needs this. Overwrite already handles row IDs and DVs on v3. Could you support v3 for the Arrow path, or file an issue, cite it here and correct this comment? ########## table/dynamic_partition_overwrite.go: ########## @@ -0,0 +1,290 @@ +// 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 { + // Row lineage in v3 needs dedicated first-row-ID reassignment when + // replacing partitions. Do not use rewrite semantics to bypass it. + return nil, fmt.Errorf("%w: dynamic partition overwrite does not support format v3 row lineage", ErrInvalidOperation) + } + + 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 +} + +// DynamicPartitionOverwriteDataFiles atomically replaces all live data files +// in the partitions represented by filesToAdd. Other partitions are unchanged. +// An unpartitioned table has one logical partition, so a non-empty input +// replaces all of its data, matching Java's ReplacePartitions operation. +// +// The added files must use the table's current partition spec. Existing live +// data files must use that spec as well: overlapping partitions from an older +// spec cannot be identified by comparing tuples from different transforms. +// Rather than silently leave stale rows, this operation rejects mixed-spec +// tables. It supports identity and transformed partition fields alike. +// +// The operation pins the target branch head and default table shape. A +// concurrent snapshot commit or partition evolution fails instead of silently +// leaving newly appended files in a replaced partition. This also means the +// method should run on a fresh transaction, before other staged snapshot or +// schema/spec changes. +// +// An empty input is a no-op. Format-v3 tables are rejected. All files must be pre-written and have valid DataFile metadata, +// just as with AddDataFiles. Uncommitted files remain the caller's responsibility +// if staging or the eventual transaction commit fails. +func (t *Transaction) DynamicPartitionOverwriteDataFiles( + ctx context.Context, filesToAdd []iceberg.DataFile, snapshotProps iceberg.Properties, +) error { Review Comment: Unlike `AddDataFiles`, this has no `opts ...WriteOption`, so callers can't pass `WithoutAutoNameMapping()`, and catalogs that reject the name mapping, can't use it. Adding it later is a breaking change. Can you add it now? ########## table/dynamic_partition_overwrite.go: ########## @@ -0,0 +1,290 @@ +// 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 { + // Row lineage in v3 needs dedicated first-row-ID reassignment when + // replacing partitions. Do not use rewrite semantics to bypass it. + return nil, fmt.Errorf("%w: dynamic partition overwrite does not support format v3 row lineage", ErrInvalidOperation) + } + + 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 +} + +// DynamicPartitionOverwriteDataFiles atomically replaces all live data files +// in the partitions represented by filesToAdd. Other partitions are unchanged. +// An unpartitioned table has one logical partition, so a non-empty input +// replaces all of its data, matching Java's ReplacePartitions operation. +// +// The added files must use the table's current partition spec. Existing live +// data files must use that spec as well: overlapping partitions from an older +// spec cannot be identified by comparing tuples from different transforms. +// Rather than silently leave stale rows, this operation rejects mixed-spec +// tables. It supports identity and transformed partition fields alike. +// +// The operation pins the target branch head and default table shape. A +// concurrent snapshot commit or partition evolution fails instead of silently +// leaving newly appended files in a replaced partition. This also means the +// method should run on a fresh transaction, before other staged snapshot or +// schema/spec changes. +// +// An empty input is a no-op. Format-v3 tables are rejected. All files must be pre-written and have valid DataFile metadata, +// just as with AddDataFiles. Uncommitted files remain the caller's responsibility +// if staging or the eventual transaction commit fails. +func (t *Transaction) DynamicPartitionOverwriteDataFiles( + ctx context.Context, filesToAdd []iceberg.DataFile, snapshotProps iceberg.Properties, +) error { + spec, err := t.dynamicOverwriteSpec() + if err != nil { + return err + } + if len(filesToAdd) == 0 { + return nil + } + if err := ctx.Err(); err != nil { + return err + } + + addedPaths, err := t.validateDataFilesToAdd(filesToAdd, "DynamicPartitionOverwriteDataFiles", true) + 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 + } + + keys, err := dynamicPartitionKeys(filesToAdd) + if err != nil { + return err + } + var toDelete []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 + } Review Comment: Delete files in a replaced partition stay live, though they can never apply again. Could you collect the ones with a matching key here and remove them with removeDeleteFile? ########## table/dynamic_partition_overwrite.go: ########## @@ -0,0 +1,290 @@ +// 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 { + // Row lineage in v3 needs dedicated first-row-ID reassignment when + // replacing partitions. Do not use rewrite semantics to bypass it. + return nil, fmt.Errorf("%w: dynamic partition overwrite does not support format v3 row lineage", ErrInvalidOperation) + } + + 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 +} + +// DynamicPartitionOverwriteDataFiles atomically replaces all live data files +// in the partitions represented by filesToAdd. Other partitions are unchanged. +// An unpartitioned table has one logical partition, so a non-empty input +// replaces all of its data, matching Java's ReplacePartitions operation. +// +// The added files must use the table's current partition spec. Existing live +// data files must use that spec as well: overlapping partitions from an older +// spec cannot be identified by comparing tuples from different transforms. +// Rather than silently leave stale rows, this operation rejects mixed-spec +// tables. It supports identity and transformed partition fields alike. +// +// The operation pins the target branch head and default table shape. A +// concurrent snapshot commit or partition evolution fails instead of silently +// leaving newly appended files in a replaced partition. This also means the +// method should run on a fresh transaction, before other staged snapshot or +// schema/spec changes. +// +// An empty input is a no-op. Format-v3 tables are rejected. All files must be pre-written and have valid DataFile metadata, +// just as with AddDataFiles. Uncommitted files remain the caller's responsibility +// if staging or the eventual transaction commit fails. +func (t *Transaction) DynamicPartitionOverwriteDataFiles( + ctx context.Context, filesToAdd []iceberg.DataFile, snapshotProps iceberg.Properties, +) error { + spec, err := t.dynamicOverwriteSpec() + if err != nil { + return err + } + if len(filesToAdd) == 0 { + return nil + } + if err := ctx.Err(); err != nil { + return err + } + + addedPaths, err := t.validateDataFilesToAdd(filesToAdd, "DynamicPartitionOverwriteDataFiles", true) + 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 + } + + keys, err := dynamicPartitionKeys(filesToAdd) + if err != nil { + return err + } + var toDelete []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) + } + } + } + + if err := ctx.Err(); err != nil { + return err + } + + // A ref fence is important even if no files matched: an append to + // touched partition between planning and commit must not survive. + if err := t.AssertDefaultShape(); err != nil { + return err + } + if err := t.AssertRefSnapshotID(""); err != nil { + return err + } + + props := make(iceberg.Properties, len(snapshotProps)+1) + maps.Copy(props, snapshotProps) + props["replace-partitions"] = "true" + + if len(toDelete) == 0 { + // ReplaceDataFiles delegates additions without removals to AddDataFiles, + // recording an append snapshot. ReplacePartitions must produce an + // overwrite snapshot even when it creates a previously empty partition: + // the operation type is part of Iceberg's cherry-pick semantics. + if err := t.ensureNameMapping(); err != nil { + return err + } + wfs, err := requireWriteFileIO(fs) + if err != nil { + return err + } + commitUUID := uuid.New() + producer := newOverwriteFilesProducer(OpOverwrite, t, wfs, &commitUUID, props) + for _, file := range filesToAdd { + producer.appendDataFile(file) + } + updates, reqs, err := producer.commit(ctx) + if err != nil { + return err + } + + return t.apply(updates, reqs) + } + + return t.ReplaceDataFiles(ctx, toDelete, filesToAdd, props) +} + +// DynamicPartitionOverwrite writes Arrow records to data files and replaces +// only the partitions represented by the written records. In particular, +// an empty reader never clears existing partitions. +func (t *Transaction) DynamicPartitionOverwrite( Review Comment: Only `DynamicPartitionOverwriteDataFiles` documents the rules. Could this and the other Arrow methods also say that v3 and mixed-spec tables are rejected, a concurrent commit fails, and written files stay behind on failure? Overwrite repeats its note on all four methods. ########## table/dynamic_partition_overwrite.go: ########## @@ -0,0 +1,290 @@ +// 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) { Review Comment: Matching tuples looks right to me. But `BuildPartitionMatchPredicate` is left with no caller, while its doc calls it "the core of dynamic partition overwrite". Could you delete it, or say why it stays? ########## table/dynamic_partition_overwrite_test.go: ########## @@ -0,0 +1,287 @@ +// 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" + "math" + "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 TestDynamicPartitionValueEquality(t *testing.T) { + key := func(value any) string { + result, err := canonicalPartitionKey(0, map[int]any{1000: value}) + require.NoError(t, err) + + return result + } + require.Equal(t, key(nil), key(nil)) + require.Equal(t, key([]byte{0, 1, 255}), key([]byte{0, 1, 255})) + require.NotEqual(t, key([]byte{0, 1}), key([]byte{1, 0})) + require.Equal(t, key(int32(1)), key(int64(1))) + require.Equal(t, key(iceberg.Date(1)), key(int32(1))) + require.Equal(t, key(float32(math.NaN())), key(float32(math.NaN()))) + require.Equal(t, key(math.NaN()), key(math.NaN())) + require.NotEqual(t, key(math.NaN()), key(0.0)) +} + +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(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.Contains(t, txn.pinnedRefs, MainBranch, "concurrent appends must not bypass the overwrite") +} + +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(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(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 TestDynamicPartitionOverwriteRejectsUnsupportedLayouts(t *testing.T) { + ctx := context.Background() + t.Run("format v3", func(t *testing.T) { + spec := partitionedSpec() + txn, _ := createTestTransactionWithMemIO(t, spec) + txn.meta.formatVersion = 3 + err := txn.DynamicPartitionOverwriteDataFiles(ctx, nil, nil) + require.ErrorIs(t, err, ErrInvalidOperation) + require.ErrorContains(t, err, "v3") + }) + + t.Run("wrong incoming spec", func(t *testing.T) { + spec := partitionedSpec() + txn, _ := createTestTransactionWithMemIO(t, spec) + otherSpec := iceberg.NewPartitionSpecID(9, iceberg.PartitionField{ + SourceIDs: []int{1}, FieldID: 1000, Name: "id", Transform: iceberg.IdentityTransform{}, + }) + file := newTestDataFile(t, otherSpec, "file://wrong-spec.parquet", map[int]any{1000: int32(1)}) Review Comment: spec 9 isn't registered, so this fails before the check at L112. No test reaches that check, the mixed-spec rejection at L143, or AssertDefaultShape. please add a case for each, asserting ErrorIs and the message? ########## table/dynamic_partition_overwrite.go: ########## @@ -0,0 +1,290 @@ +// 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 { + // Row lineage in v3 needs dedicated first-row-ID reassignment when + // replacing partitions. Do not use rewrite semantics to bypass it. + return nil, fmt.Errorf("%w: dynamic partition overwrite does not support format v3 row lineage", ErrInvalidOperation) + } + + 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 +} + +// DynamicPartitionOverwriteDataFiles atomically replaces all live data files +// in the partitions represented by filesToAdd. Other partitions are unchanged. +// An unpartitioned table has one logical partition, so a non-empty input +// replaces all of its data, matching Java's ReplacePartitions operation. +// +// The added files must use the table's current partition spec. Existing live +// data files must use that spec as well: overlapping partitions from an older +// spec cannot be identified by comparing tuples from different transforms. +// Rather than silently leave stale rows, this operation rejects mixed-spec +// tables. It supports identity and transformed partition fields alike. +// +// The operation pins the target branch head and default table shape. A +// concurrent snapshot commit or partition evolution fails instead of silently +// leaving newly appended files in a replaced partition. This also means the +// method should run on a fresh transaction, before other staged snapshot or +// schema/spec changes. +// +// An empty input is a no-op. Format-v3 tables are rejected. All files must be pre-written and have valid DataFile metadata, +// just as with AddDataFiles. Uncommitted files remain the caller's responsibility +// if staging or the eventual transaction commit fails. +func (t *Transaction) DynamicPartitionOverwriteDataFiles( + ctx context.Context, filesToAdd []iceberg.DataFile, snapshotProps iceberg.Properties, +) error { + spec, err := t.dynamicOverwriteSpec() + if err != nil { + return err + } + if len(filesToAdd) == 0 { + return nil + } + if err := ctx.Err(); err != nil { + return err + } + + addedPaths, err := t.validateDataFilesToAdd(filesToAdd, "DynamicPartitionOverwriteDataFiles", true) + 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 + } + + keys, err := dynamicPartitionKeys(filesToAdd) + if err != nil { + return err + } + var toDelete []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) + } + } + } + + if err := ctx.Err(); err != nil { + return err + } + + // A ref fence is important even if no files matched: an append to + // touched partition between planning and commit must not survive. + if err := t.AssertDefaultShape(); err != nil { + return err + } + if err := t.AssertRefSnapshotID(""); err != nil { + return err + } + + props := make(iceberg.Properties, len(snapshotProps)+1) + maps.Copy(props, snapshotProps) + props["replace-partitions"] = "true" + + if len(toDelete) == 0 { + // ReplaceDataFiles delegates additions without removals to AddDataFiles, + // recording an append snapshot. ReplacePartitions must produce an + // overwrite snapshot even when it creates a previously empty partition: + // the operation type is part of Iceberg's cherry-pick semantics. + if err := t.ensureNameMapping(); err != nil { + return err + } + wfs, err := requireWriteFileIO(fs) + if err != nil { + return err + } + commitUUID := uuid.New() + producer := newOverwriteFilesProducer(OpOverwrite, t, wfs, &commitUUID, props) + for _, file := range filesToAdd { + producer.appendDataFile(file) + } + updates, reqs, err := producer.commit(ctx) + if err != nil { + return err + } + + return t.apply(updates, reqs) + } + + return t.ReplaceDataFiles(ctx, toDelete, filesToAdd, props) Review Comment: These two paths nearly match, and `ReplaceDataFiles` reads every manifest again to find files we already hold. Could one `newOverwriteFilesProducer(OpOverwrite, ...)` serve both, with `deleteDataFile` and `appendDataFile`? That removes a full manifest pass. ########## table/dynamic_partition_overwrite_test.go: ########## @@ -0,0 +1,287 @@ +// 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" + "math" + "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 TestDynamicPartitionValueEquality(t *testing.T) { Review Comment: This only tests `canonicalPartitionKey`, which already has tests. No test replaces a null, transformed or date partition using real files. Can you swap it for a table test that appends, overwrites one partition and scans? ########## table/dynamic_partition_overwrite.go: ########## @@ -0,0 +1,290 @@ +// 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 { + // Row lineage in v3 needs dedicated first-row-ID reassignment when + // replacing partitions. Do not use rewrite semantics to bypass it. + return nil, fmt.Errorf("%w: dynamic partition overwrite does not support format v3 row lineage", ErrInvalidOperation) + } + + 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 +} + +// DynamicPartitionOverwriteDataFiles atomically replaces all live data files +// in the partitions represented by filesToAdd. Other partitions are unchanged. +// An unpartitioned table has one logical partition, so a non-empty input +// replaces all of its data, matching Java's ReplacePartitions operation. +// +// The added files must use the table's current partition spec. Existing live +// data files must use that spec as well: overlapping partitions from an older +// spec cannot be identified by comparing tuples from different transforms. +// Rather than silently leave stale rows, this operation rejects mixed-spec +// tables. It supports identity and transformed partition fields alike. +// +// The operation pins the target branch head and default table shape. A +// concurrent snapshot commit or partition evolution fails instead of silently +// leaving newly appended files in a replaced partition. This also means the +// method should run on a fresh transaction, before other staged snapshot or +// schema/spec changes. +// +// An empty input is a no-op. Format-v3 tables are rejected. All files must be pre-written and have valid DataFile metadata, +// just as with AddDataFiles. Uncommitted files remain the caller's responsibility +// if staging or the eventual transaction commit fails. +func (t *Transaction) DynamicPartitionOverwriteDataFiles( + ctx context.Context, filesToAdd []iceberg.DataFile, snapshotProps iceberg.Properties, +) error { + spec, err := t.dynamicOverwriteSpec() + if err != nil { + return err + } + if len(filesToAdd) == 0 { + return nil + } + if err := ctx.Err(); err != nil { + return err + } + + addedPaths, err := t.validateDataFilesToAdd(filesToAdd, "DynamicPartitionOverwriteDataFiles", true) + 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 + } + + keys, err := dynamicPartitionKeys(filesToAdd) + if err != nil { + return err + } + var toDelete []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) + } + } + } + + if err := ctx.Err(); err != nil { + return err + } + + // A ref fence is important even if no files matched: an append to + // touched partition between planning and commit must not survive. + if err := t.AssertDefaultShape(); err != nil { + return err + } + if err := t.AssertRefSnapshotID(""); err != nil { + return err + } + + props := make(iceberg.Properties, len(snapshotProps)+1) + maps.Copy(props, snapshotProps) + props["replace-partitions"] = "true" + + if len(toDelete) == 0 { + // ReplaceDataFiles delegates additions without removals to AddDataFiles, + // recording an append snapshot. ReplacePartitions must produce an + // overwrite snapshot even when it creates a previously empty partition: + // the operation type is part of Iceberg's cherry-pick semantics. + if err := t.ensureNameMapping(); err != nil { + return err + } + wfs, err := requireWriteFileIO(fs) + if err != nil { + return err + } + commitUUID := uuid.New() + producer := newOverwriteFilesProducer(OpOverwrite, t, wfs, &commitUUID, props) + for _, file := range filesToAdd { + producer.appendDataFile(file) + } + updates, reqs, err := producer.commit(ctx) + if err != nil { + return err + } + + return t.apply(updates, reqs) + } + + return t.ReplaceDataFiles(ctx, toDelete, filesToAdd, props) +} + +// DynamicPartitionOverwrite writes Arrow records to data files and replaces +// only the partitions represented by the written records. In particular, +// an empty reader never clears existing partitions. +func (t *Transaction) DynamicPartitionOverwrite( + ctx context.Context, rdr array.RecordReader, snapshotProps iceberg.Properties, +) 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 { Review Comment: the mixed-spec check runs after all files are written. A table with older-spec files gets the whole input written, then an error, and the caller has no paths to clean up. please check PartitionSpecID() on the manifests before recordsToDataFiles? -- 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]
