badalprasadsingh commented on code in PR #2383: URL: https://github.com/apache/iceberg-go/pull/2383#discussion_r4458049767
########## table/rewrite_by_path.go: ########## @@ -0,0 +1,205 @@ +// 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" + "math" + "runtime" + + "github.com/apache/iceberg-go" +) + +// NewRewriteFromPaths builds a data-file rewrite from existing file paths. +// It resolves removed files against the transaction's target branch and reads +// the replacement Parquet footers to construct their data-file metadata. +// The current partition spec must be unpartitioned. +// +// Unlike ReplaceDataFilesByPath, which performs an overwrite, this operation +// produces an Iceberg replace snapshot. The caller must ensure that the new +// files contain exactly the same live rows as the removed files, including +// the effects of delete files. File metadata alone cannot establish that +// equivalence. This path-only helper requires the same physical record count +// in the replacement files, even when the table has delete files. A rewrite +// that materializes row-level deletes and reduces the physical row count must +// use NewRewrite with explicit DataFile metadata and delete-file handling. +// Matching record counts alone do not establish row-level equivalence. +// Format v3 is not supported because row lineage cannot be established +// from file paths and footer metadata alone. +// +// The returned builder can be configured with DataSequenceNumber or additional +// delete-file changes. Call RewriteFiles.Commit to stage the snapshot, then +// Transaction.Commit to persist it. The rewrite validator detects concurrent +// delete files that would make the replacement unsafe. +// +// Both path lists must be non-empty, unique, and disjoint. Removed paths must +// be live data files in the target branch; added paths must not already be +// referenced there. The method may stage the table's default name mapping +// before returning the builder, but does not stage a snapshot. +func (t *Transaction) NewRewriteFromPaths( + ctx context.Context, + filesToDelete, filesToAdd []string, + snapshotProps iceberg.Properties, +) (*RewriteFiles, error) { + meta, err := t.txnMeta() + if err != nil { + return nil, err + } + if err := ctx.Err(); err != nil { + return nil, err + } + if meta.formatVersion >= 3 { + return nil, fmt.Errorf("%w: path-based rewrites do not support v3 row lineage", iceberg.ErrNotImplemented) + } + spec, err := meta.CurrentSpec() + if err != nil { + return nil, err + } + if spec.NumFields() != 0 { + return nil, fmt.Errorf("%w: path-based rewrites do not support partitioned tables", iceberg.ErrNotImplemented) + } + if len(filesToDelete) == 0 || len(filesToAdd) == 0 { + return nil, fmt.Errorf("%w: a data-file rewrite requires both removed and added files", ErrInvalidOperation) + } + + toDelete := make(map[string]struct{}, len(filesToDelete)) + for _, path := range filesToDelete { + if path == "" { + return nil, fmt.Errorf("%w: removed file paths must be non-empty", ErrInvalidOperation) + } + if _, exists := toDelete[path]; exists { + return nil, fmt.Errorf("%w: duplicate removed file path %q", ErrInvalidOperation, path) + } + toDelete[path] = struct{}{} + } + + toAdd := make(map[string]struct{}, len(filesToAdd)) + for _, path := range filesToAdd { + if path == "" { + return nil, fmt.Errorf("%w: added file paths must be non-empty", ErrInvalidOperation) + } + if _, exists := toAdd[path]; exists { + return nil, fmt.Errorf("%w: duplicate added file path %q", ErrInvalidOperation, path) + } + if _, exists := toDelete[path]; exists { + return nil, fmt.Errorf("%w: cannot remove and add the same file %q", ErrInvalidOperation, path) + } + toAdd[path] = struct{}{} + } + + snapshot := t.planningSnapshot(meta) + if snapshot == nil { + return nil, fmt.Errorf("%w: cannot rewrite files in a table without an existing snapshot", ErrInvalidOperation) + } + + fs, err := t.tbl.fsF(ctx) + if err != nil { + return nil, err + } + if _, err := requireWriteFileIO(fs); err != nil { + return nil, err + } + + removed := make([]iceberg.DataFile, 0, len(filesToDelete)) + var originalRecordCount int64 + for df, err := range snapshot.dataFiles(fs, nil) { + if err != nil { + return nil, err + } + if err := ctx.Err(); err != nil { + return nil, err + } + if _, exists := toAdd[df.FilePath()]; exists { + return nil, fmt.Errorf("%w: added file %q is already referenced by the table", ErrInvalidOperation, df.FilePath()) + } + if _, exists := toDelete[df.FilePath()]; exists { + if df.ContentType() != iceberg.EntryContentData { + return nil, fmt.Errorf("%w: cannot rewrite non-data file %q", ErrInvalidOperation, df.FilePath()) + } + if df.Count() < 0 || df.Count() > math.MaxInt64-originalRecordCount { + return nil, fmt.Errorf("%w: invalid total record count for removed data files", ErrInvalidOperation) + } + originalRecordCount += df.Count() + removed = append(removed, df) + } + } + if len(removed) != len(filesToDelete) { + return nil, fmt.Errorf("%w: all removed files must be live in the target branch", ErrInvalidOperation) + } + + if err := t.ensureNameMapping(); err != nil { + return nil, err + } + // ensureNameMapping may have replaced the staged metadata builder. Review Comment: if a later check fails: this name mapping stays in the txn, and the next txn.Commit writes it. commit already adds the mapping, and reading footers doesn't need it so better: remove this call and the re-read below. ########## table/rewrite_by_path.go: ########## @@ -0,0 +1,205 @@ +// 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" + "math" + "runtime" + + "github.com/apache/iceberg-go" +) + +// NewRewriteFromPaths builds a data-file rewrite from existing file paths. +// It resolves removed files against the transaction's target branch and reads +// the replacement Parquet footers to construct their data-file metadata. +// The current partition spec must be unpartitioned. +// +// Unlike ReplaceDataFilesByPath, which performs an overwrite, this operation +// produces an Iceberg replace snapshot. The caller must ensure that the new +// files contain exactly the same live rows as the removed files, including +// the effects of delete files. File metadata alone cannot establish that +// equivalence. This path-only helper requires the same physical record count +// in the replacement files, even when the table has delete files. A rewrite +// that materializes row-level deletes and reduces the physical row count must +// use NewRewrite with explicit DataFile metadata and delete-file handling. +// Matching record counts alone do not establish row-level equivalence. +// Format v3 is not supported because row lineage cannot be established +// from file paths and footer metadata alone. +// +// The returned builder can be configured with DataSequenceNumber or additional +// delete-file changes. Call RewriteFiles.Commit to stage the snapshot, then +// Transaction.Commit to persist it. The rewrite validator detects concurrent +// delete files that would make the replacement unsafe. +// +// Both path lists must be non-empty, unique, and disjoint. Removed paths must +// be live data files in the target branch; added paths must not already be +// referenced there. The method may stage the table's default name mapping +// before returning the builder, but does not stage a snapshot. +func (t *Transaction) NewRewriteFromPaths( + ctx context.Context, + filesToDelete, filesToAdd []string, + snapshotProps iceberg.Properties, +) (*RewriteFiles, error) { + meta, err := t.txnMeta() + if err != nil { + return nil, err + } + if err := ctx.Err(); err != nil { + return nil, err + } + if meta.formatVersion >= 3 { + return nil, fmt.Errorf("%w: path-based rewrites do not support v3 row lineage", iceberg.ErrNotImplemented) + } + spec, err := meta.CurrentSpec() + if err != nil { + return nil, err + } + if spec.NumFields() != 0 { + return nil, fmt.Errorf("%w: path-based rewrites do not support partitioned tables", iceberg.ErrNotImplemented) + } + if len(filesToDelete) == 0 || len(filesToAdd) == 0 { + return nil, fmt.Errorf("%w: a data-file rewrite requires both removed and added files", ErrInvalidOperation) + } + + toDelete := make(map[string]struct{}, len(filesToDelete)) + for _, path := range filesToDelete { + if path == "" { + return nil, fmt.Errorf("%w: removed file paths must be non-empty", ErrInvalidOperation) + } + if _, exists := toDelete[path]; exists { + return nil, fmt.Errorf("%w: duplicate removed file path %q", ErrInvalidOperation, path) + } + toDelete[path] = struct{}{} + } + + toAdd := make(map[string]struct{}, len(filesToAdd)) + for _, path := range filesToAdd { + if path == "" { + return nil, fmt.Errorf("%w: added file paths must be non-empty", ErrInvalidOperation) + } + if _, exists := toAdd[path]; exists { + return nil, fmt.Errorf("%w: duplicate added file path %q", ErrInvalidOperation, path) + } + if _, exists := toDelete[path]; exists { + return nil, fmt.Errorf("%w: cannot remove and add the same file %q", ErrInvalidOperation, path) + } + toAdd[path] = struct{}{} + } + + snapshot := t.planningSnapshot(meta) + if snapshot == nil { + return nil, fmt.Errorf("%w: cannot rewrite files in a table without an existing snapshot", ErrInvalidOperation) + } + + fs, err := t.tbl.fsF(ctx) + if err != nil { + return nil, err + } + if _, err := requireWriteFileIO(fs); err != nil { + return nil, err + } + + removed := make([]iceberg.DataFile, 0, len(filesToDelete)) + var originalRecordCount int64 + for df, err := range snapshot.dataFiles(fs, nil) { + if err != nil { + return nil, err + } + if err := ctx.Err(); err != nil { + return nil, err + } + if _, exists := toAdd[df.FilePath()]; exists { + return nil, fmt.Errorf("%w: added file %q is already referenced by the table", ErrInvalidOperation, df.FilePath()) + } + if _, exists := toDelete[df.FilePath()]; exists { + if df.ContentType() != iceberg.EntryContentData { + return nil, fmt.Errorf("%w: cannot rewrite non-data file %q", ErrInvalidOperation, df.FilePath()) + } + if df.Count() < 0 || df.Count() > math.MaxInt64-originalRecordCount { + return nil, fmt.Errorf("%w: invalid total record count for removed data files", ErrInvalidOperation) + } + originalRecordCount += df.Count() + removed = append(removed, df) + } + } + if len(removed) != len(filesToDelete) { + return nil, fmt.Errorf("%w: all removed files must be live in the target branch", ErrInvalidOperation) + } + + if err := t.ensureNameMapping(); err != nil { + return nil, err + } + // ensureNameMapping may have replaced the staged metadata builder. + meta, err = t.txnMeta() + if err != nil { + return nil, err + } + + added, err := filesToDataFiles(ctx, fs, meta, filesToAdd, runtime.GOMAXPROCS(0)) + if err != nil { + return nil, err + } + + var replacementRecordCount int64 + for _, df := range added { + if df.Count() < 0 || df.Count() > math.MaxInt64-replacementRecordCount { + return nil, fmt.Errorf("%w: invalid total record count for replacement data files", ErrInvalidOperation) + } + replacementRecordCount += df.Count() Review Comment: the count check can't see a lost column, if a new file drops an optional column, this passes and its values read as null, IMO, compare `ValueCounts()` for each column, and fail if one changed. ########## website/src/api.md: ########## @@ -363,7 +363,37 @@ if err != nil { /* ... */ } newTbl, err := txn.Commit(ctx) ``` -`ReplaceDataFilesByPath(ctx, filesToDelete, filesToAdd, snapshotProps)` and `ReplaceDataFiles(ctx, filesToDelete, dataFilesToAdd, snapshotProps, opts...)` are also available on `*Transaction` for swapping files atomically. +`ReplaceDataFilesByPath(ctx, filesToDelete, filesToAdd, snapshotProps)` performs an +**overwrite** of existing files. It is suitable when the logical table data +changes. `ReplaceDataFiles(ctx, filesToDelete, dataFilesToAdd, snapshotProps, opts...)` +is the corresponding data-file metadata API. + +### Rewrite existing Parquet files + +For compaction or other **data-preserving** rewrites, use the path-based +`RewriteDataFilesByPath`. It reads the replacement Parquet file footers, +checks the removed files against the target branch's live data files, and +stages a `replace` snapshot rather than an `overwrite`: Review Comment: this sends compaction users here, but RewriteDataFiles below also handles deletes and partitions. Better: say this section is for files another engine wrote, link to compaction, and add the same-record-count rule, pointing live-rows-only rewrites to NewRewrite. ########## table/rewrite_by_path.go: ########## @@ -0,0 +1,205 @@ +// 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" + "math" + "runtime" + + "github.com/apache/iceberg-go" +) + +// NewRewriteFromPaths builds a data-file rewrite from existing file paths. +// It resolves removed files against the transaction's target branch and reads +// the replacement Parquet footers to construct their data-file metadata. +// The current partition spec must be unpartitioned. +// +// Unlike ReplaceDataFilesByPath, which performs an overwrite, this operation +// produces an Iceberg replace snapshot. The caller must ensure that the new +// files contain exactly the same live rows as the removed files, including +// the effects of delete files. File metadata alone cannot establish that +// equivalence. This path-only helper requires the same physical record count +// in the replacement files, even when the table has delete files. A rewrite +// that materializes row-level deletes and reduces the physical row count must +// use NewRewrite with explicit DataFile metadata and delete-file handling. +// Matching record counts alone do not establish row-level equivalence. +// Format v3 is not supported because row lineage cannot be established +// from file paths and footer metadata alone. +// +// The returned builder can be configured with DataSequenceNumber or additional +// delete-file changes. Call RewriteFiles.Commit to stage the snapshot, then +// Transaction.Commit to persist it. The rewrite validator detects concurrent +// delete files that would make the replacement unsafe. +// +// Both path lists must be non-empty, unique, and disjoint. Removed paths must +// be live data files in the target branch; added paths must not already be +// referenced there. The method may stage the table's default name mapping +// before returning the builder, but does not stage a snapshot. +func (t *Transaction) NewRewriteFromPaths( + ctx context.Context, + filesToDelete, filesToAdd []string, + snapshotProps iceberg.Properties, +) (*RewriteFiles, error) { + meta, err := t.txnMeta() + if err != nil { + return nil, err + } + if err := ctx.Err(); err != nil { + return nil, err + } + if meta.formatVersion >= 3 { + return nil, fmt.Errorf("%w: path-based rewrites do not support v3 row lineage", iceberg.ErrNotImplemented) + } + spec, err := meta.CurrentSpec() + if err != nil { + return nil, err + } + if spec.NumFields() != 0 { + return nil, fmt.Errorf("%w: path-based rewrites do not support partitioned tables", iceberg.ErrNotImplemented) + } + if len(filesToDelete) == 0 || len(filesToAdd) == 0 { + return nil, fmt.Errorf("%w: a data-file rewrite requires both removed and added files", ErrInvalidOperation) + } + + toDelete := make(map[string]struct{}, len(filesToDelete)) + for _, path := range filesToDelete { + if path == "" { + return nil, fmt.Errorf("%w: removed file paths must be non-empty", ErrInvalidOperation) + } + if _, exists := toDelete[path]; exists { + return nil, fmt.Errorf("%w: duplicate removed file path %q", ErrInvalidOperation, path) + } + toDelete[path] = struct{}{} + } + + toAdd := make(map[string]struct{}, len(filesToAdd)) + for _, path := range filesToAdd { + if path == "" { + return nil, fmt.Errorf("%w: added file paths must be non-empty", ErrInvalidOperation) + } + if _, exists := toAdd[path]; exists { + return nil, fmt.Errorf("%w: duplicate added file path %q", ErrInvalidOperation, path) + } + if _, exists := toDelete[path]; exists { + return nil, fmt.Errorf("%w: cannot remove and add the same file %q", ErrInvalidOperation, path) + } + toAdd[path] = struct{}{} + } + + snapshot := t.planningSnapshot(meta) + if snapshot == nil { + return nil, fmt.Errorf("%w: cannot rewrite files in a table without an existing snapshot", ErrInvalidOperation) + } + + fs, err := t.tbl.fsF(ctx) + if err != nil { + return nil, err + } + if _, err := requireWriteFileIO(fs); err != nil { + return nil, err + } + + removed := make([]iceberg.DataFile, 0, len(filesToDelete)) + var originalRecordCount int64 + for df, err := range snapshot.dataFiles(fs, nil) { + if err != nil { + return nil, err + } + if err := ctx.Err(); err != nil { + return nil, err + } + if _, exists := toAdd[df.FilePath()]; exists { + return nil, fmt.Errorf("%w: added file %q is already referenced by the table", ErrInvalidOperation, df.FilePath()) + } + if _, exists := toDelete[df.FilePath()]; exists { + if df.ContentType() != iceberg.EntryContentData { + return nil, fmt.Errorf("%w: cannot rewrite non-data file %q", ErrInvalidOperation, df.FilePath()) + } + if df.Count() < 0 || df.Count() > math.MaxInt64-originalRecordCount { + return nil, fmt.Errorf("%w: invalid total record count for removed data files", ErrInvalidOperation) + } + originalRecordCount += df.Count() + removed = append(removed, df) + } + } + if len(removed) != len(filesToDelete) { + return nil, fmt.Errorf("%w: all removed files must be live in the target branch", ErrInvalidOperation) + } + + if err := t.ensureNameMapping(); err != nil { + return nil, err + } + // ensureNameMapping may have replaced the staged metadata builder. + meta, err = t.txnMeta() + if err != nil { + return nil, err + } + + added, err := filesToDataFiles(ctx, fs, meta, filesToAdd, runtime.GOMAXPROCS(0)) + if err != nil { + return nil, err + } + + var replacementRecordCount int64 + for _, df := range added { + if df.Count() < 0 || df.Count() > math.MaxInt64-replacementRecordCount { + return nil, fmt.Errorf("%w: invalid total record count for replacement data files", ErrInvalidOperation) + } + replacementRecordCount += df.Count() + } + // A table-level delete file is not evidence that it applies to these + // removed files. Do not accept lower counts merely because any delete + // file exists somewhere in the table: it could hide lost rows. + if replacementRecordCount != originalRecordCount { + return nil, fmt.Errorf("%w: path-based rewrite record count changed from %d to %d", + ErrInvalidOperation, originalRecordCount, replacementRecordCount) + } Review Comment: P0: Deleted rows can come back. Say a delete file applies to a removed file. A new file with only the live rows fails this check. A full copy passes, but position deletes still point at the old path, and the new sequence number skips equality deletes, so the deleted rows return. The snapshot still says replace. Lets fail the call here instead: `t.Scan(WithSnapshotID(snapshot.SnapshotID))` with `PlanFiles` lists the delete files for each file, and it can replace the dataFiles loop. Please add a test with real delete files. ########## table/rewrite_by_path_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_test + +import ( + "context" + "testing" + + "github.com/apache/iceberg-go" + "github.com/apache/iceberg-go/table" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func newPathRewriteFixture(t *testing.T) (*table.Table, string, string) { + t.Helper() + + return newPathRewriteFixtureVersion(t, 2) +} + +func newPathRewriteFixtureVersion(t *testing.T, version int) (*table.Table, string, string) { + t.Helper() + + tbl, _ := newConcurrentRewriteTestTableVersion(t, version) + sc, err := table.SchemaToArrowSchema(tbl.Schema(), nil, false, false) + require.NoError(t, err) + + oldPath := tbl.Location() + "/data/original.parquet" + replacementPath := tbl.Location() + "/data/replacement.parquet" + rows := `[{"id":1,"data":"first"},{"id":2,"data":"second"}]` + writeParquetFile(t, oldPath, sc, rows) + writeParquetFile(t, replacementPath, sc, rows) + + txn := tbl.NewTransaction() + require.NoError(t, txn.AddDataFilesByPath(t.Context(), []string{oldPath}, nil, false)) + tbl, err = txn.Commit(t.Context()) + require.NoError(t, err) + + return tbl, oldPath, replacementPath +} + +func TestRewriteDataFilesByPathProducesReplaceSnapshot(t *testing.T) { + tbl, oldPath, replacementPath := newPathRewriteFixture(t) + before := tbl.CurrentSnapshot() + require.NotNil(t, before) + + txn := tbl.NewTransaction() + require.NoError(t, txn.RewriteDataFilesByPath(t.Context(), + []string{oldPath}, []string{replacementPath}, + iceberg.Properties{"reason": "compaction"})) + + tbl, err := txn.Commit(t.Context()) + require.NoError(t, err) + + snapshot := tbl.CurrentSnapshot() + require.NotNil(t, snapshot) + assert.Equal(t, table.OpReplace, snapshot.Summary.Operation) + assert.Equal(t, "compaction", snapshot.Summary.Properties["reason"]) + assert.Equal(t, []int64{1, 2}, idsOnRef(t, tbl, table.MainBranch)) + dataPaths, _ := liveFilesOnRef(t, tbl, table.MainBranch) + assert.Equal(t, []string{replacementPath}, dataPaths) + assert.NotEqual(t, before.SnapshotID, snapshot.SnapshotID) +} + +func TestRewriteDataFilesByPathRejectsInvalidRecordCounts(t *testing.T) { + for _, tc := range []struct { + name string + rows string + }{ + {"drops rows without deletes", `[{"id":1,"data":"first"}]`}, + {"adds rows", `[{"id":1,"data":"first"},{"id":2,"data":"second"},{"id":3,"data":"extra"}]`}, + } { + t.Run(tc.name, func(t *testing.T) { + tbl, oldPath, _ := newPathRewriteFixture(t) + sc, err := table.SchemaToArrowSchema(tbl.Schema(), nil, false, false) + require.NoError(t, err) + replacementPath := tbl.Location() + "/data/invalid-replacement.parquet" + writeParquetFile(t, replacementPath, sc, tc.rows) + + txn := tbl.NewTransaction() + err = txn.RewriteDataFilesByPath(t.Context(), + []string{oldPath}, []string{replacementPath}, nil) + require.ErrorIs(t, err, table.ErrInvalidOperation) + require.ErrorContains(t, err, "record count") + }) + } +} + +// A delete file elsewhere in the table does not prove that fewer physical +// rows in the replacement are valid. Only the explicit rewrite builder may +// handle delete-aware compaction with different physical row counts. +func TestRewriteDataFilesByPathRejectsReducedRowsWithExistingDeleteFile(t *testing.T) { + tbl, oldPath, _ := newPathRewriteFixture(t) + deletes := newEqDeleteFile(t, tbl.Location()+"/data/rewrite-count-delete.parquet") + peer := tbl.NewTransaction() + require.NoError(t, peer.NewRowDelta(nil).AddDeletes(deletes).Commit(t.Context())) + var err error + tbl, err = peer.Commit(t.Context()) + require.NoError(t, err) + + sc, err := table.SchemaToArrowSchema(tbl.Schema(), nil, false, false) + require.NoError(t, err) + replacementPath := tbl.Location() + "/data/short-replacement.parquet" + writeParquetFile(t, replacementPath, sc, `[{"id":1,"data":"first"}]`) + + txn := tbl.NewTransaction() + err = txn.RewriteDataFilesByPath(t.Context(), + []string{oldPath}, []string{replacementPath}, nil) + require.ErrorIs(t, err, table.ErrInvalidOperation) + require.ErrorContains(t, err, "record count") +} + +func TestNewRewriteFromPathsAllowsDataSequenceNumber(t *testing.T) { + tbl, oldPath, replacementPath := newPathRewriteFixture(t) + before := tbl.CurrentSnapshot() + require.NotNil(t, before) + + txn := tbl.NewTransaction() + rewrite, err := txn.NewRewriteFromPaths(t.Context(), + []string{oldPath}, []string{replacementPath}, nil) + require.NoError(t, err) + + require.NoError(t, rewrite.DataSequenceNumber(before.SequenceNumber).Commit(t.Context())) + tbl, err = txn.Commit(t.Context()) + require.NoError(t, err) + + assert.Equal(t, table.OpReplace, tbl.CurrentSnapshot().Summary.Operation) + assert.Equal(t, []int64{1, 2}, idsOnRef(t, tbl, table.MainBranch)) +} + +func TestNewRewriteFromPathsValidatesPaths(t *testing.T) { + tbl, oldPath, replacementPath := newPathRewriteFixture(t) + + cases := []struct { + name string + removed []string + added []string + }{ + {"no removed files", nil, []string{replacementPath}}, + {"no added files", []string{oldPath}, nil}, + {"blank removed path", []string{""}, []string{replacementPath}}, + {"blank added path", []string{oldPath}, []string{""}}, + {"duplicate removed path", []string{oldPath, oldPath}, []string{replacementPath}}, + {"duplicate added path", []string{oldPath}, []string{replacementPath, replacementPath}}, + {"overlapping path sets", []string{oldPath}, []string{oldPath}}, + {"unknown removed path", []string{tbl.Location() + "/unknown.parquet"}, []string{replacementPath}}, + {"existing added path", []string{oldPath}, []string{oldPath}}, Review Comment: "existing added path" has the same inputs as overlapping path sets, so no test reaches the already referenced branch. could you please commit a second file to the table here, and pass its path as the added file? ########## table/rewrite_by_path.go: ########## @@ -0,0 +1,205 @@ +// 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" + "math" + "runtime" + + "github.com/apache/iceberg-go" +) + +// NewRewriteFromPaths builds a data-file rewrite from existing file paths. +// It resolves removed files against the transaction's target branch and reads +// the replacement Parquet footers to construct their data-file metadata. +// The current partition spec must be unpartitioned. +// +// Unlike ReplaceDataFilesByPath, which performs an overwrite, this operation +// produces an Iceberg replace snapshot. The caller must ensure that the new +// files contain exactly the same live rows as the removed files, including +// the effects of delete files. File metadata alone cannot establish that +// equivalence. This path-only helper requires the same physical record count +// in the replacement files, even when the table has delete files. A rewrite +// that materializes row-level deletes and reduces the physical row count must +// use NewRewrite with explicit DataFile metadata and delete-file handling. +// Matching record counts alone do not establish row-level equivalence. +// Format v3 is not supported because row lineage cannot be established +// from file paths and footer metadata alone. +// +// The returned builder can be configured with DataSequenceNumber or additional +// delete-file changes. Call RewriteFiles.Commit to stage the snapshot, then +// Transaction.Commit to persist it. The rewrite validator detects concurrent +// delete files that would make the replacement unsafe. +// +// Both path lists must be non-empty, unique, and disjoint. Removed paths must +// be live data files in the target branch; added paths must not already be +// referenced there. The method may stage the table's default name mapping +// before returning the builder, but does not stage a snapshot. +func (t *Transaction) NewRewriteFromPaths( + ctx context.Context, + filesToDelete, filesToAdd []string, + snapshotProps iceberg.Properties, +) (*RewriteFiles, error) { + meta, err := t.txnMeta() + if err != nil { + return nil, err + } + if err := ctx.Err(); err != nil { + return nil, err + } + if meta.formatVersion >= 3 { + return nil, fmt.Errorf("%w: path-based rewrites do not support v3 row lineage", iceberg.ErrNotImplemented) + } + spec, err := meta.CurrentSpec() + if err != nil { + return nil, err + } + if spec.NumFields() != 0 { + return nil, fmt.Errorf("%w: path-based rewrites do not support partitioned tables", iceberg.ErrNotImplemented) + } + if len(filesToDelete) == 0 || len(filesToAdd) == 0 { + return nil, fmt.Errorf("%w: a data-file rewrite requires both removed and added files", ErrInvalidOperation) + } + + toDelete := make(map[string]struct{}, len(filesToDelete)) + for _, path := range filesToDelete { + if path == "" { + return nil, fmt.Errorf("%w: removed file paths must be non-empty", ErrInvalidOperation) + } + if _, exists := toDelete[path]; exists { + return nil, fmt.Errorf("%w: duplicate removed file path %q", ErrInvalidOperation, path) + } + toDelete[path] = struct{}{} + } + + toAdd := make(map[string]struct{}, len(filesToAdd)) + for _, path := range filesToAdd { + if path == "" { + return nil, fmt.Errorf("%w: added file paths must be non-empty", ErrInvalidOperation) + } + if _, exists := toAdd[path]; exists { + return nil, fmt.Errorf("%w: duplicate added file path %q", ErrInvalidOperation, path) + } + if _, exists := toDelete[path]; exists { + return nil, fmt.Errorf("%w: cannot remove and add the same file %q", ErrInvalidOperation, path) + } + toAdd[path] = struct{}{} + } + + snapshot := t.planningSnapshot(meta) + if snapshot == nil { + return nil, fmt.Errorf("%w: cannot rewrite files in a table without an existing snapshot", ErrInvalidOperation) + } + + fs, err := t.tbl.fsF(ctx) + if err != nil { + return nil, err + } + if _, err := requireWriteFileIO(fs); err != nil { + return nil, err + } + + removed := make([]iceberg.DataFile, 0, len(filesToDelete)) + var originalRecordCount int64 + for df, err := range snapshot.dataFiles(fs, nil) { + if err != nil { + return nil, err + } + if err := ctx.Err(); err != nil { + return nil, err + } + if _, exists := toAdd[df.FilePath()]; exists { + return nil, fmt.Errorf("%w: added file %q is already referenced by the table", ErrInvalidOperation, df.FilePath()) + } + if _, exists := toDelete[df.FilePath()]; exists { + if df.ContentType() != iceberg.EntryContentData { + return nil, fmt.Errorf("%w: cannot rewrite non-data file %q", ErrInvalidOperation, df.FilePath()) + } + if df.Count() < 0 || df.Count() > math.MaxInt64-originalRecordCount { + return nil, fmt.Errorf("%w: invalid total record count for removed data files", ErrInvalidOperation) + } + originalRecordCount += df.Count() + removed = append(removed, df) + } + } + if len(removed) != len(filesToDelete) { + return nil, fmt.Errorf("%w: all removed files must be live in the target branch", ErrInvalidOperation) + } Review Comment: let the error say which path is missing, please list the missing paths. ########## table/rewrite_by_path_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_test + +import ( + "context" + "testing" + + "github.com/apache/iceberg-go" + "github.com/apache/iceberg-go/table" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func newPathRewriteFixture(t *testing.T) (*table.Table, string, string) { + t.Helper() + + return newPathRewriteFixtureVersion(t, 2) +} + +func newPathRewriteFixtureVersion(t *testing.T, version int) (*table.Table, string, string) { + t.Helper() + + tbl, _ := newConcurrentRewriteTestTableVersion(t, version) + sc, err := table.SchemaToArrowSchema(tbl.Schema(), nil, false, false) + require.NoError(t, err) + + oldPath := tbl.Location() + "/data/original.parquet" + replacementPath := tbl.Location() + "/data/replacement.parquet" + rows := `[{"id":1,"data":"first"},{"id":2,"data":"second"}]` + writeParquetFile(t, oldPath, sc, rows) + writeParquetFile(t, replacementPath, sc, rows) + + txn := tbl.NewTransaction() + require.NoError(t, txn.AddDataFilesByPath(t.Context(), []string{oldPath}, nil, false)) + tbl, err = txn.Commit(t.Context()) + require.NoError(t, err) + + return tbl, oldPath, replacementPath +} + +func TestRewriteDataFilesByPathProducesReplaceSnapshot(t *testing.T) { + tbl, oldPath, replacementPath := newPathRewriteFixture(t) + before := tbl.CurrentSnapshot() + require.NotNil(t, before) + + txn := tbl.NewTransaction() + require.NoError(t, txn.RewriteDataFilesByPath(t.Context(), + []string{oldPath}, []string{replacementPath}, + iceberg.Properties{"reason": "compaction"})) + + tbl, err := txn.Commit(t.Context()) + require.NoError(t, err) + + snapshot := tbl.CurrentSnapshot() + require.NotNil(t, snapshot) + assert.Equal(t, table.OpReplace, snapshot.Summary.Operation) + assert.Equal(t, "compaction", snapshot.Summary.Properties["reason"]) + assert.Equal(t, []int64{1, 2}, idsOnRef(t, tbl, table.MainBranch)) + dataPaths, _ := liveFilesOnRef(t, tbl, table.MainBranch) + assert.Equal(t, []string{replacementPath}, dataPaths) + assert.NotEqual(t, before.SnapshotID, snapshot.SnapshotID) +} + +func TestRewriteDataFilesByPathRejectsInvalidRecordCounts(t *testing.T) { + for _, tc := range []struct { + name string + rows string + }{ + {"drops rows without deletes", `[{"id":1,"data":"first"}]`}, + {"adds rows", `[{"id":1,"data":"first"},{"id":2,"data":"second"},{"id":3,"data":"extra"}]`}, + } { + t.Run(tc.name, func(t *testing.T) { + tbl, oldPath, _ := newPathRewriteFixture(t) + sc, err := table.SchemaToArrowSchema(tbl.Schema(), nil, false, false) + require.NoError(t, err) + replacementPath := tbl.Location() + "/data/invalid-replacement.parquet" + writeParquetFile(t, replacementPath, sc, tc.rows) + + txn := tbl.NewTransaction() + err = txn.RewriteDataFilesByPath(t.Context(), + []string{oldPath}, []string{replacementPath}, nil) + require.ErrorIs(t, err, table.ErrInvalidOperation) + require.ErrorContains(t, err, "record count") + }) + } +} + +// A delete file elsewhere in the table does not prove that fewer physical +// rows in the replacement are valid. Only the explicit rewrite builder may +// handle delete-aware compaction with different physical row counts. +func TestRewriteDataFilesByPathRejectsReducedRowsWithExistingDeleteFile(t *testing.T) { + tbl, oldPath, _ := newPathRewriteFixture(t) + deletes := newEqDeleteFile(t, tbl.Location()+"/data/rewrite-count-delete.parquet") + peer := tbl.NewTransaction() + require.NoError(t, peer.NewRowDelta(nil).AddDeletes(deletes).Commit(t.Context())) + var err error + tbl, err = peer.Commit(t.Context()) + require.NoError(t, err) + + sc, err := table.SchemaToArrowSchema(tbl.Schema(), nil, false, false) + require.NoError(t, err) + replacementPath := tbl.Location() + "/data/short-replacement.parquet" + writeParquetFile(t, replacementPath, sc, `[{"id":1,"data":"first"}]`) + + txn := tbl.NewTransaction() + err = txn.RewriteDataFilesByPath(t.Context(), + []string{oldPath}, []string{replacementPath}, nil) + require.ErrorIs(t, err, table.ErrInvalidOperation) + require.ErrorContains(t, err, "record count") +} + +func TestNewRewriteFromPathsAllowsDataSequenceNumber(t *testing.T) { Review Comment: neither the rows nor `OpReplace` depend on `DataSequenceNumber`, so this test would pass if the value were ignored. could you read the new file's manifest entry and assert its `DataSequenceNumber()`? This goes away if comment we drop the builder. ########## table/rewrite_by_path.go: ########## @@ -0,0 +1,205 @@ +// 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" + "math" + "runtime" + + "github.com/apache/iceberg-go" +) + +// NewRewriteFromPaths builds a data-file rewrite from existing file paths. +// It resolves removed files against the transaction's target branch and reads +// the replacement Parquet footers to construct their data-file metadata. +// The current partition spec must be unpartitioned. +// +// Unlike ReplaceDataFilesByPath, which performs an overwrite, this operation +// produces an Iceberg replace snapshot. The caller must ensure that the new +// files contain exactly the same live rows as the removed files, including +// the effects of delete files. File metadata alone cannot establish that +// equivalence. This path-only helper requires the same physical record count +// in the replacement files, even when the table has delete files. A rewrite +// that materializes row-level deletes and reduces the physical row count must +// use NewRewrite with explicit DataFile metadata and delete-file handling. +// Matching record counts alone do not establish row-level equivalence. +// Format v3 is not supported because row lineage cannot be established +// from file paths and footer metadata alone. +// +// The returned builder can be configured with DataSequenceNumber or additional +// delete-file changes. Call RewriteFiles.Commit to stage the snapshot, then +// Transaction.Commit to persist it. The rewrite validator detects concurrent +// delete files that would make the replacement unsafe. +// +// Both path lists must be non-empty, unique, and disjoint. Removed paths must +// be live data files in the target branch; added paths must not already be +// referenced there. The method may stage the table's default name mapping +// before returning the builder, but does not stage a snapshot. +func (t *Transaction) NewRewriteFromPaths( Review Comment: any reason to expose this? once deletes are rejected, this builder has no job left, and `NewRewrite` with `FileToDataFile` already covers those callers. Removing a public method later breaks users, so lets keep only `RewriteDataFilesByPath`. ########## table/rewrite_by_path_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_test + +import ( + "context" + "testing" + + "github.com/apache/iceberg-go" + "github.com/apache/iceberg-go/table" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func newPathRewriteFixture(t *testing.T) (*table.Table, string, string) { + t.Helper() + + return newPathRewriteFixtureVersion(t, 2) +} + +func newPathRewriteFixtureVersion(t *testing.T, version int) (*table.Table, string, string) { + t.Helper() + + tbl, _ := newConcurrentRewriteTestTableVersion(t, version) + sc, err := table.SchemaToArrowSchema(tbl.Schema(), nil, false, false) + require.NoError(t, err) + + oldPath := tbl.Location() + "/data/original.parquet" + replacementPath := tbl.Location() + "/data/replacement.parquet" + rows := `[{"id":1,"data":"first"},{"id":2,"data":"second"}]` + writeParquetFile(t, oldPath, sc, rows) + writeParquetFile(t, replacementPath, sc, rows) + + txn := tbl.NewTransaction() + require.NoError(t, txn.AddDataFilesByPath(t.Context(), []string{oldPath}, nil, false)) + tbl, err = txn.Commit(t.Context()) + require.NoError(t, err) + + return tbl, oldPath, replacementPath +} + +func TestRewriteDataFilesByPathProducesReplaceSnapshot(t *testing.T) { Review Comment: compaction merges many files, but every test here rewrites one file into one, so a bug in adding up counts would go unnoticed. could you please add two-to-one and one-to-two cases, plus another live file that stays untouched? ########## table/rewrite_by_path.go: ########## @@ -0,0 +1,205 @@ +// 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" + "math" + "runtime" + + "github.com/apache/iceberg-go" +) + +// NewRewriteFromPaths builds a data-file rewrite from existing file paths. +// It resolves removed files against the transaction's target branch and reads +// the replacement Parquet footers to construct their data-file metadata. +// The current partition spec must be unpartitioned. +// +// Unlike ReplaceDataFilesByPath, which performs an overwrite, this operation +// produces an Iceberg replace snapshot. The caller must ensure that the new +// files contain exactly the same live rows as the removed files, including +// the effects of delete files. File metadata alone cannot establish that +// equivalence. This path-only helper requires the same physical record count +// in the replacement files, even when the table has delete files. A rewrite +// that materializes row-level deletes and reduces the physical row count must +// use NewRewrite with explicit DataFile metadata and delete-file handling. +// Matching record counts alone do not establish row-level equivalence. +// Format v3 is not supported because row lineage cannot be established +// from file paths and footer metadata alone. +// +// The returned builder can be configured with DataSequenceNumber or additional +// delete-file changes. Call RewriteFiles.Commit to stage the snapshot, then +// Transaction.Commit to persist it. The rewrite validator detects concurrent +// delete files that would make the replacement unsafe. +// +// Both path lists must be non-empty, unique, and disjoint. Removed paths must +// be live data files in the target branch; added paths must not already be +// referenced there. The method may stage the table's default name mapping +// before returning the builder, but does not stage a snapshot. +func (t *Transaction) NewRewriteFromPaths( + ctx context.Context, + filesToDelete, filesToAdd []string, + snapshotProps iceberg.Properties, +) (*RewriteFiles, error) { + meta, err := t.txnMeta() + if err != nil { + return nil, err + } + if err := ctx.Err(); err != nil { + return nil, err + } + if meta.formatVersion >= 3 { + return nil, fmt.Errorf("%w: path-based rewrites do not support v3 row lineage", iceberg.ErrNotImplemented) + } + spec, err := meta.CurrentSpec() + if err != nil { + return nil, err + } + if spec.NumFields() != 0 { + return nil, fmt.Errorf("%w: path-based rewrites do not support partitioned tables", iceberg.ErrNotImplemented) + } + if len(filesToDelete) == 0 || len(filesToAdd) == 0 { + return nil, fmt.Errorf("%w: a data-file rewrite requires both removed and added files", ErrInvalidOperation) + } + + toDelete := make(map[string]struct{}, len(filesToDelete)) + for _, path := range filesToDelete { + if path == "" { + return nil, fmt.Errorf("%w: removed file paths must be non-empty", ErrInvalidOperation) + } + if _, exists := toDelete[path]; exists { + return nil, fmt.Errorf("%w: duplicate removed file path %q", ErrInvalidOperation, path) + } + toDelete[path] = struct{}{} + } + + toAdd := make(map[string]struct{}, len(filesToAdd)) + for _, path := range filesToAdd { + if path == "" { + return nil, fmt.Errorf("%w: added file paths must be non-empty", ErrInvalidOperation) + } + if _, exists := toAdd[path]; exists { + return nil, fmt.Errorf("%w: duplicate added file path %q", ErrInvalidOperation, path) + } + if _, exists := toDelete[path]; exists { + return nil, fmt.Errorf("%w: cannot remove and add the same file %q", ErrInvalidOperation, path) + } + toAdd[path] = struct{}{} + } + + snapshot := t.planningSnapshot(meta) + if snapshot == nil { + return nil, fmt.Errorf("%w: cannot rewrite files in a table without an existing snapshot", ErrInvalidOperation) + } + + fs, err := t.tbl.fsF(ctx) + if err != nil { + return nil, err + } + if _, err := requireWriteFileIO(fs); err != nil { + return nil, err + } + + removed := make([]iceberg.DataFile, 0, len(filesToDelete)) + var originalRecordCount int64 + for df, err := range snapshot.dataFiles(fs, nil) { + if err != nil { + return nil, err + } + if err := ctx.Err(); err != nil { + return nil, err + } + if _, exists := toAdd[df.FilePath()]; exists { + return nil, fmt.Errorf("%w: added file %q is already referenced by the table", ErrInvalidOperation, df.FilePath()) + } + if _, exists := toDelete[df.FilePath()]; exists { + if df.ContentType() != iceberg.EntryContentData { + return nil, fmt.Errorf("%w: cannot rewrite non-data file %q", ErrInvalidOperation, df.FilePath()) + } + if df.Count() < 0 || df.Count() > math.MaxInt64-originalRecordCount { + return nil, fmt.Errorf("%w: invalid total record count for removed data files", ErrInvalidOperation) + } + originalRecordCount += df.Count() + removed = append(removed, df) + } + } + if len(removed) != len(filesToDelete) { + return nil, fmt.Errorf("%w: all removed files must be live in the target branch", ErrInvalidOperation) + } + + if err := t.ensureNameMapping(); err != nil { + return nil, err + } + // ensureNameMapping may have replaced the staged metadata builder. + meta, err = t.txnMeta() + if err != nil { + return nil, err + } + + added, err := filesToDataFiles(ctx, fs, meta, filesToAdd, runtime.GOMAXPROCS(0)) + if err != nil { + return nil, err + } + + var replacementRecordCount int64 + for _, df := range added { + if df.Count() < 0 || df.Count() > math.MaxInt64-replacementRecordCount { + return nil, fmt.Errorf("%w: invalid total record count for replacement data files", ErrInvalidOperation) + } + replacementRecordCount += df.Count() + } + // A table-level delete file is not evidence that it applies to these + // removed files. Do not accept lower counts merely because any delete + // file exists somewhere in the table: it could hide lost rows. + if replacementRecordCount != originalRecordCount { + return nil, fmt.Errorf("%w: path-based rewrite record count changed from %d to %d", + ErrInvalidOperation, originalRecordCount, replacementRecordCount) + } + + rewrite := t.NewRewrite(snapshotProps) + for _, df := range removed { + rewrite.DeleteFile(df) + } + for _, df := range added { + rewrite.AddDataFile(df) + } + + return rewrite, nil +} + +// RewriteDataFilesByPath stages an Iceberg replace snapshot for a +// data-preserving rewrite of files identified by their storage paths. +// Call Transaction.Commit to persist it. +// +// This is the single-step counterpart of NewRewriteFromPaths. Use the builder +// instead if the rewrite also needs DataSequenceNumber or delete-file changes. +// As with RewriteFiles, verifying that the replacement preserves live rows is +// the caller's responsibility. Format v3 is not supported. +func (t *Transaction) RewriteDataFilesByPath( Review Comment: don't you think that this name can mislead? `RewriteDataFiles` reads rows and applies deletes, but this method only swaps in files the caller already wrote. what about `CommitRewrittenFilesByPath`? open to discussion though. -- 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]
