badalprasadsingh commented on code in PR #2381: URL: https://github.com/apache/iceberg-go/pull/2381#discussion_r4492183879
########## table/dynamic_partition_overwrite.go: ########## @@ -0,0 +1,403 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package table + +import ( + "context" + "fmt" + "maps" + + "github.com/apache/arrow-go/v18/arrow" + "github.com/apache/arrow-go/v18/arrow/array" + "github.com/apache/iceberg-go" + "github.com/google/uuid" +) + +// dynamicPartitionKeys normalizes all partition values before comparing them. +// A manifest decoder may represent the same logical day as an int32 while the +// writer uses iceberg.Date. The shared canonical encoder handles both forms, +// decimals, bytes, UUIDs, and nulls without lossy string formatting. +func dynamicPartitionKeys(files []iceberg.DataFile) (map[string]struct{}, error) { + keys := make(map[string]struct{}, len(files)) + for _, file := range files { + key, err := canonicalPartitionKey(file.SpecID(), dataFilePartition(file)) + if err != nil { + return nil, fmt.Errorf("dynamic partition overwrite file %q: %w", file.FilePath(), err) + } + keys[key] = struct{}{} + } + + return keys, nil +} + +func (t *Transaction) dynamicOverwriteSpec() (*iceberg.PartitionSpec, error) { + meta, err := t.txnMeta() + if err != nil { + return nil, err + } + + if meta.formatVersion > 3 { + return nil, fmt.Errorf("%w: dynamic partition overwrite does not support format v%d", ErrInvalidOperation, meta.formatVersion) + } + + spec, err := meta.CurrentSpec() + if err != nil { + return nil, err + } + if spec == nil { + return nil, fmt.Errorf("%w: dynamic partition overwrite requires a current partition spec", ErrInvalidOperation) + } + if err := checkNoUnknownTransform(spec); err != nil { + return nil, err + } + + return spec, nil +} + +// validateDynamicOverwriteConflicts uses exact stored partition tuples so +// transformed and null partitions cannot become false-negative row predicates. +// It rejects concurrent additions and removals in touched partitions while +// allowing unrelated partition appends during refresh-and-replay. +func validateDynamicOverwriteConflicts(cc *conflictContext, specID int, touched map[string]struct{}) error { + if cc == nil || len(cc.concurrent) == 0 { + return nil + } + for _, snapshot := range cc.concurrent { + manifests, err := cc.manifestsFor(snapshot) + if err != nil { + return fmt.Errorf("dynamic partition overwrite: concurrent manifests for snapshot %d: %w", snapshot.SnapshotID, err) + } + for _, manifest := range manifests { + for entry, err := range manifest.Entries(cc.ctx, cc.manifestReadIO(), false) { + if err != nil { + return fmt.Errorf("dynamic partition overwrite: read concurrent manifest %s: %w", manifest.FilePath(), err) + } + if entry.SnapshotID() != snapshot.SnapshotID || entry.Status() == iceberg.EntryStatusEXISTING { + continue + } + file := entry.DataFile() + conflict := ErrConflictingDataFiles + if file.ContentType() != iceberg.EntryContentData { + conflict = ErrConflictingDeleteFiles + } + if int(file.SpecID()) != specID { + continue + } + key, err := canonicalPartitionKey(file.SpecID(), dataFilePartition(file)) + if err != nil { + return fmt.Errorf("dynamic partition overwrite: concurrent file %q: %w", file.FilePath(), err) + } + if _, ok := touched[key]; ok { + return fmt.Errorf("%w: concurrent snapshot %d changed file %q in an overwritten partition", + conflict, snapshot.SnapshotID, file.FilePath()) + } + } + } + } + + return nil +} + +// DynamicPartitionOverwriteDataFiles atomically replaces all live data in +// the partitions represented by filesToAdd. Empty input is a no-op; an +// unpartitioned table has one partition. New partitions still use OpOverwrite. +// +// Added files must use the table's current partition spec; existing live files +// under older specs are preserved. Supports v1-v3, including v3 row lineage +// and associated deletion vectors. On v3, a new data file may omit first_row_id +// and inherit it from its manifest. +// Concurrent writes to touched partitions fail while independent appends to +// other partitions can be retried unless delete-file removal prevents replay. +// A missing target branch is pinned together with main to guard its origin. +// Default partition and sort order changes also fail. +// +// All input files must already exist. Uncommitted data files remain the +// caller's responsibility after a staging or commit failure. By default this +// sets the table's name mapping; WithoutAutoNameMapping disables that step. +// WithoutDuplicateCheck is ignored to avoid adding an already live file twice. +func (t *Transaction) DynamicPartitionOverwriteDataFiles( + ctx context.Context, filesToAdd []iceberg.DataFile, snapshotProps iceberg.Properties, opts ...WriteOption, +) error { + spec, err := t.dynamicOverwriteSpec() + if err != nil { + return err + } + if len(filesToAdd) == 0 { + return nil + } + if err := ctx.Err(); err != nil { + return err + } + + var cfg dataFileCfg + for _, option := range opts { + option(&cfg) + } + if cfg.rewriteSemantics { + return fmt.Errorf("%w: dynamic partition overwrite cannot use data-preserving rewrite semantics", ErrInvalidOperation) + } + + // Dynamic overwrite adds new rows, so v3 first_row_id can be inherited + // from the manifest. Java ReplacePartitions uses the same inheritance. + addedPaths, err := t.validateDataFilesToAdd(filesToAdd, "DynamicPartitionOverwriteDataFiles", false) + if err != nil { + return err + } + for _, df := range filesToAdd { + if int(df.SpecID()) != spec.ID() { + return fmt.Errorf("%w: dynamic partition overwrite file %q uses spec %d, current spec is %d", + ErrInvalidOperation, df.FilePath(), df.SpecID(), spec.ID()) + } + } + + fs, err := t.tbl.fsF(ctx) + if err != nil { + return err + } + wfs, err := requireWriteFileIO(fs) + if err != nil { + return err + } + + keys, err := dynamicPartitionKeys(filesToAdd) + if err != nil { + return err + } + var toDelete, deleteFilesToRemove []iceberg.DataFile + validationPlan := newPartitionValidationPlan(spec) + if snapshot := t.planningSnapshot(t.meta); snapshot != nil { + for df, err := range snapshot.dataFiles(ctx, 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() { + continue + } + 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(ctx, fs, iceberg.ManifestContentDeletes, true) { + if err != nil { + return fmt.Errorf("dynamic partition overwrite: read delete files: %w", err) + } + if err := ctx.Err(); err != nil { + return err + } + df := entry.DataFile() + if _, exists := addedPaths[df.FilePath()]; exists { + return fmt.Errorf("%w: added file %q already exists as a delete file", ErrInvalidOperation, df.FilePath()) + } + if IsDeletionVector(df) { + continue + } + if int(df.SpecID()) != spec.ID() { + continue + } + key, err := canonicalPartitionKey(df.SpecID(), dataFilePartition(df)) + if err != nil { + return fmt.Errorf("dynamic partition overwrite live delete file %q: %w", df.FilePath(), err) + } + if _, touched := keys[key]; touched { + deleteFilesToRemove = append(deleteFilesToRemove, df) + } Review Comment: P0: on v2, a mor delete of an old-spec row is written under the current spec, with null for the new partition field. Overwriting that partition drops the delete but keeps the old file, so the row comes back. better: keep deletes whose ReferencedDataFile() stays live, with a test. ########## table/dynamic_partition_overwrite.go: ########## @@ -0,0 +1,403 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package table + +import ( + "context" + "fmt" + "maps" + + "github.com/apache/arrow-go/v18/arrow" + "github.com/apache/arrow-go/v18/arrow/array" + "github.com/apache/iceberg-go" + "github.com/google/uuid" +) + +// dynamicPartitionKeys normalizes all partition values before comparing them. +// A manifest decoder may represent the same logical day as an int32 while the +// writer uses iceberg.Date. The shared canonical encoder handles both forms, +// decimals, bytes, UUIDs, and nulls without lossy string formatting. +func dynamicPartitionKeys(files []iceberg.DataFile) (map[string]struct{}, error) { + keys := make(map[string]struct{}, len(files)) + for _, file := range files { + key, err := canonicalPartitionKey(file.SpecID(), dataFilePartition(file)) + if err != nil { + return nil, fmt.Errorf("dynamic partition overwrite file %q: %w", file.FilePath(), err) + } + keys[key] = struct{}{} + } + + return keys, nil +} + +func (t *Transaction) dynamicOverwriteSpec() (*iceberg.PartitionSpec, error) { + meta, err := t.txnMeta() + if err != nil { + return nil, err + } + + if meta.formatVersion > 3 { + return nil, fmt.Errorf("%w: dynamic partition overwrite does not support format v%d", ErrInvalidOperation, meta.formatVersion) + } + + spec, err := meta.CurrentSpec() + if err != nil { + return nil, err + } + if spec == nil { + return nil, fmt.Errorf("%w: dynamic partition overwrite requires a current partition spec", ErrInvalidOperation) + } + if err := checkNoUnknownTransform(spec); err != nil { + return nil, err + } + + return spec, nil +} + +// validateDynamicOverwriteConflicts uses exact stored partition tuples so +// transformed and null partitions cannot become false-negative row predicates. +// It rejects concurrent additions and removals in touched partitions while +// allowing unrelated partition appends during refresh-and-replay. +func validateDynamicOverwriteConflicts(cc *conflictContext, specID int, touched map[string]struct{}) error { + if cc == nil || len(cc.concurrent) == 0 { + return nil + } + for _, snapshot := range cc.concurrent { + manifests, err := cc.manifestsFor(snapshot) + if err != nil { + return fmt.Errorf("dynamic partition overwrite: concurrent manifests for snapshot %d: %w", snapshot.SnapshotID, err) + } + for _, manifest := range manifests { + for entry, err := range manifest.Entries(cc.ctx, cc.manifestReadIO(), false) { + if err != nil { + return fmt.Errorf("dynamic partition overwrite: read concurrent manifest %s: %w", manifest.FilePath(), err) + } + if entry.SnapshotID() != snapshot.SnapshotID || entry.Status() == iceberg.EntryStatusEXISTING { + continue + } + file := entry.DataFile() + conflict := ErrConflictingDataFiles + if file.ContentType() != iceberg.EntryContentData { + conflict = ErrConflictingDeleteFiles + } + if int(file.SpecID()) != specID { + continue + } + key, err := canonicalPartitionKey(file.SpecID(), dataFilePartition(file)) + if err != nil { + return fmt.Errorf("dynamic partition overwrite: concurrent file %q: %w", file.FilePath(), err) + } + if _, ok := touched[key]; ok { + return fmt.Errorf("%w: concurrent snapshot %d changed file %q in an overwritten partition", + conflict, snapshot.SnapshotID, file.FilePath()) + } + } + } + } + + return nil +} + +// DynamicPartitionOverwriteDataFiles atomically replaces all live data in +// the partitions represented by filesToAdd. Empty input is a no-op; an +// unpartitioned table has one partition. New partitions still use OpOverwrite. +// +// Added files must use the table's current partition spec; existing live files +// under older specs are preserved. Supports v1-v3, including v3 row lineage +// and associated deletion vectors. On v3, a new data file may omit first_row_id +// and inherit it from its manifest. +// Concurrent writes to touched partitions fail while independent appends to +// other partitions can be retried unless delete-file removal prevents replay. +// A missing target branch is pinned together with main to guard its origin. +// Default partition and sort order changes also fail. +// +// All input files must already exist. Uncommitted data files remain the +// caller's responsibility after a staging or commit failure. By default this +// sets the table's name mapping; WithoutAutoNameMapping disables that step. +// WithoutDuplicateCheck is ignored to avoid adding an already live file twice. +func (t *Transaction) DynamicPartitionOverwriteDataFiles( + ctx context.Context, filesToAdd []iceberg.DataFile, snapshotProps iceberg.Properties, opts ...WriteOption, +) error { + spec, err := t.dynamicOverwriteSpec() + if err != nil { + return err + } + if len(filesToAdd) == 0 { + return nil + } + if err := ctx.Err(); err != nil { + return err + } + + var cfg dataFileCfg + for _, option := range opts { + option(&cfg) + } + if cfg.rewriteSemantics { + return fmt.Errorf("%w: dynamic partition overwrite cannot use data-preserving rewrite semantics", ErrInvalidOperation) + } + + // Dynamic overwrite adds new rows, so v3 first_row_id can be inherited + // from the manifest. Java ReplacePartitions uses the same inheritance. + addedPaths, err := t.validateDataFilesToAdd(filesToAdd, "DynamicPartitionOverwriteDataFiles", false) + if err != nil { + return err + } + for _, df := range filesToAdd { + if int(df.SpecID()) != spec.ID() { + return fmt.Errorf("%w: dynamic partition overwrite file %q uses spec %d, current spec is %d", + ErrInvalidOperation, df.FilePath(), df.SpecID(), spec.ID()) + } + } + + fs, err := t.tbl.fsF(ctx) + if err != nil { + return err + } + wfs, err := requireWriteFileIO(fs) + if err != nil { + return err + } + + keys, err := dynamicPartitionKeys(filesToAdd) + if err != nil { + return err + } + var toDelete, deleteFilesToRemove []iceberg.DataFile + validationPlan := newPartitionValidationPlan(spec) + if snapshot := t.planningSnapshot(t.meta); snapshot != nil { + for df, err := range snapshot.dataFiles(ctx, 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() { + continue + } Review Comment: P0: here, after a partition field is removed, the spec is unpartitioned, this keeps old-spec files but drops current-spec equality deletes, which are global, so rows they removed come back. please replace every file here, also add a test pinning that -- 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]
