You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iceberg.apache.org by at...@apache.org on 2026/09/23 21:17:08 UTC

(iceberg-go) branch main updated: fix(table): validate deletion vector fields in RowDelta (#2004 follow-up) (#2016)

This is an automated email from the ASF dual-hosted git repository.

laskoviymishka pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg-go.git


The following commit(s) were added to refs/heads/main by this push:
     new 7dd736493 fix(table): validate deletion vector fields in RowDelta (#2004 follow-up) (#2016)
7dd736493 is described below

commit 7dd7364930b6c888aa939a4fe19038ea51cc1086
Author: Badal Prasad Singh <ba...@datazip.io>
AuthorDate: Thu Sep 24 02:47:03 2026 +0530

    fix(table): validate deletion vector fields in RowDelta (#2004 follow-up) (#2016)
    
    * fix: reject dvs below v3
    
    Signed-off-by: badalprasadsingh <ba...@datazip.io>
    
    * minor follow ups
    
    Signed-off-by: badalprasadsingh <ba...@datazip.io>
    
    * minor
    
    Signed-off-by: badalprasadsingh <ba...@datazip.io>
    
    * chore: minor
    
    Signed-off-by: badalprasadsingh <ba...@datazip.io>
    
    * minor
    
    Signed-off-by: badalprasadsingh <ba...@datazip.io>
    
    * minor
    
    Signed-off-by: badalprasadsingh <ba...@datazip.io>
    
    ---------
    
    Signed-off-by: badalprasadsingh <ba...@datazip.io>
---
 table/data_file_meta_test.go           |   2 +-
 table/row_delta.go                     |   9 ++-
 table/row_delta_test.go                |  88 +++++++++++++++++----------
 table/row_lineage_prune_delete_test.go |  49 ++++++++++-----
 table/row_lineage_rewrite_test.go      |   2 +-
 table/row_lineage_scan_test.go         |  16 ++---
 table/transaction.go                   | 107 +++++++++++++++++----------------
 7 files changed, 158 insertions(+), 115 deletions(-)

diff --git a/table/data_file_meta_test.go b/table/data_file_meta_test.go
index b1101ea46..9ca145ed8 100644
--- a/table/data_file_meta_test.go
+++ b/table/data_file_meta_test.go
@@ -1228,7 +1228,7 @@ func TestDataFileFromMetadata_V3RewriteAcceptsNilFirstRowID(t *testing.T) {
 }
 
 func TestDataFileFromMetadata_EndToEndRowDelta(t *testing.T) {
-	tbl := newRowDeltaCommitTestTableVersion(t, 3)
+	tbl := newRowDeltaCommitTestTableVersion(t, 2)
 	sch := tbl.Schema()
 
 	const dataPath = "s3://bucket/test/data/insert.parquet"
diff --git a/table/row_delta.go b/table/row_delta.go
index bc37809f9..0e9508528 100644
--- a/table/row_delta.go
+++ b/table/row_delta.go
@@ -198,8 +198,13 @@ func (rd *RowDelta) Commit(ctx context.Context) error {
 				ct, f.FilePath())
 		}
 
-		if err := validateDeletionVectorFormatVersion(f, meta.formatVersion, "row delta"); err != nil {
-			return err
+		if IsDeletionVector(f) {
+			if err := validateDeletionVectorToAdd(f, meta.formatVersion, "row delta"); err != nil {
+				return err
+			}
+		} else if ct == iceberg.EntryContentPosDeletes && meta.formatVersion >= 3 {
+			return fmt.Errorf("position delete file %s must be a deletion vector for v%d table for row delta",
+				f.FilePath(), meta.formatVersion)
 		}
 
 		// Equality delete files must declare which columns form the delete key,
diff --git a/table/row_delta_test.go b/table/row_delta_test.go
index 38f791217..237f0457c 100644
--- a/table/row_delta_test.go
+++ b/table/row_delta_test.go
@@ -1108,66 +1108,86 @@ func TestRowDeltaRemoveDeletesFailsInsteadOfReplaying(t *testing.T) {
 		"the data file must carry exactly one live DV: the peer's")
 }
 
-func buildPuffinPosDeleteWithoutRef(t *testing.T, path string) iceberg.DataFile {
-	t.Helper()
-
-	b, err := iceberg.NewDataFileBuilder(*iceberg.UnpartitionedSpec, iceberg.EntryContentPosDeletes,
-		path, iceberg.PuffinFile, nil, nil, nil, 2, 128)
-	require.NoError(t, err)
-
-	return b.Build()
-}
-
-func TestRowDeltaRejectsDeletionVectorBelowV3(t *testing.T) {
+func TestRowDeltaRejectsInvalidDeletionVector(t *testing.T) {
 	const (
 		dataPath = "s3://bucket/data/insert.parquet"
 		dvPath   = "s3://bucket/data/dv-001.puffin"
 	)
+	offset, length := int64(4), int64(64)
+	negativeOffset, zeroLength := int64(-1), int64(0)
 
 	tests := []struct {
 		name          string
 		formatVersion int
 		rows          []iceberg.DataFile
-		deletes       func(*testing.T) []iceberg.DataFile
+		deletes       []iceberg.DataFile
 		errContains   string
 	}{
 		{
 			name:          "deletion vector alone on v2",
 			formatVersion: 2,
-			deletes: func(t *testing.T) []iceberg.DataFile {
-				return []iceberg.DataFile{buildDVFile(t, dvPath, dataPath)}
-			},
-			errContains: "requires table format version >= 3",
+			deletes:       []iceberg.DataFile{buildDVFile(t, dvPath, dataPath)},
+			errContains:   "requires table format version >= 3",
 		},
 		{
 			name:          "deletion vector beside valid files on v2",
 			formatVersion: 2,
 			rows:          []iceberg.DataFile{buildDataFile(t, dataPath)},
-			deletes: func(t *testing.T) []iceberg.DataFile {
-				return []iceberg.DataFile{
-					buildPosDeleteFile(t, "s3://bucket/data/pos-del.parquet"),
-					buildEqDeleteFile(t, "s3://bucket/data/eq-del.parquet", []int{1}),
-					buildDVFile(t, dvPath, dataPath),
-				}
+			deletes: []iceberg.DataFile{
+				buildPosDeleteFile(t, "s3://bucket/data/pos-del.parquet"),
+				buildEqDeleteFile(t, "s3://bucket/data/eq-del.parquet", []int{1}),
+				buildDVFile(t, dvPath, dataPath),
 			},
 			errContains: "requires table format version >= 3",
 		},
 		{
-			name:          "deletion vector without referenced data file on v2",
+			name:          "format version check precedes field checks on v2",
 			formatVersion: 2,
-			deletes: func(t *testing.T) []iceberg.DataFile {
-				return []iceberg.DataFile{buildPuffinPosDeleteWithoutRef(t, dvPath)}
-			},
-			errContains: "requires table format version >= 3",
+			deletes:       []iceberg.DataFile{newRewriteDeletionVector(t, dvPath, "", nil, nil)},
+			errContains:   "requires table format version >= 3",
 		},
 		{
 			// v1 rejects every delete file before the DV check is reached.
 			name:          "deletion vector on v1",
 			formatVersion: 1,
-			deletes: func(t *testing.T) []iceberg.DataFile {
-				return []iceberg.DataFile{buildDVFile(t, dvPath, dataPath)}
-			},
-			errContains: "format version >= 2",
+			deletes:       []iceberg.DataFile{buildDVFile(t, dvPath, dataPath)},
+			errContains:   "format version >= 2",
+		},
+		{
+			name:          "missing referenced data file on v3",
+			formatVersion: 3,
+			deletes:       []iceberg.DataFile{newRewriteDeletionVector(t, dvPath, "", &offset, &length)},
+			errContains:   "missing referenced_data_file",
+		},
+		{
+			name:          "missing content offset on v3",
+			formatVersion: 3,
+			deletes:       []iceberg.DataFile{newRewriteDeletionVector(t, dvPath, dataPath, nil, &length)},
+			errContains:   "missing content_offset",
+		},
+		{
+			name:          "negative content offset on v3",
+			formatVersion: 3,
+			deletes:       []iceberg.DataFile{newRewriteDeletionVector(t, dvPath, dataPath, &negativeOffset, &length)},
+			errContains:   "invalid content_offset -1",
+		},
+		{
+			name:          "missing content size on v3",
+			formatVersion: 3,
+			deletes:       []iceberg.DataFile{newRewriteDeletionVector(t, dvPath, dataPath, &offset, nil)},
+			errContains:   "missing content_size_in_bytes",
+		},
+		{
+			name:          "nonpositive content size on v3",
+			formatVersion: 3,
+			deletes:       []iceberg.DataFile{newRewriteDeletionVector(t, dvPath, dataPath, &offset, &zeroLength)},
+			errContains:   "invalid content_size_in_bytes 0",
+		},
+		{
+			name:          "parquet position delete on v3",
+			formatVersion: 3,
+			deletes:       []iceberg.DataFile{buildPosDeleteFile(t, "s3://bucket/data/pos-del.parquet")},
+			errContains:   "must be a deletion vector for v3 table",
 		},
 	}
 
@@ -1175,7 +1195,7 @@ func TestRowDeltaRejectsDeletionVectorBelowV3(t *testing.T) {
 		t.Run(tt.name, func(t *testing.T) {
 			tbl := newRowDeltaCommitTestTableVersion(t, tt.formatVersion)
 
-			rd := tbl.NewTransaction().NewRowDelta(nil).AddRows(tt.rows...).AddDeletes(tt.deletes(t)...)
+			rd := tbl.NewTransaction().NewRowDelta(nil).AddRows(tt.rows...).AddDeletes(tt.deletes...)
 
 			err := rd.Commit(t.Context())
 			require.Error(t, err)
@@ -1235,13 +1255,15 @@ func TestRowDeltaRemoveDeletesRequiresV3(t *testing.T) {
 // DV) while adding a valid DV.
 // Assertion: Commit fails identifying the live entry as not a DV.
 func TestRowDeltaRemoveDeletesRejectsNonDV(t *testing.T) {
-	tbl := newRowDeltaCommitTestTableVersion(t, 3)
+	tbl := newRowDeltaCommitTestTableVersion(t, 2)
 	location := tbl.Location()
 
+	// v3 rejects new Parquet pos-deletes, so the live one predates an upgrade.
 	posDelPath := location + "/data/pos-del-001.parquet"
 	tx := tbl.NewTransaction()
 	require.NoError(t, tx.NewRowDelta(nil).
 		AddDeletes(buildPosDeleteFile(t, posDelPath)).Commit(t.Context()))
+	require.NoError(t, tx.UpgradeFormatVersion(3))
 	tbl, err := tx.Commit(t.Context())
 	require.NoError(t, err)
 
diff --git a/table/row_lineage_prune_delete_test.go b/table/row_lineage_prune_delete_test.go
index 9710a6b9f..1716b2fc1 100644
--- a/table/row_lineage_prune_delete_test.go
+++ b/table/row_lineage_prune_delete_test.go
@@ -43,12 +43,13 @@ import (
 // test prunes a leading row group and deletes a row that lives only in the
 // surviving group.
 
-// TestScanPruningWithPositionalDeletes covers a Parquet positional delete whose
-// target (pos 6 = id=7) is in the second row group while the filter prunes the
-// first. The deleted row must vanish and survivors must keep their _row_id.
+// TestScanPruningWithPositionalDeletes covers a legacy Parquet positional delete,
+// committed before the v3 upgrade, whose target (pos 6 = id=7) is in the second row group
+// while the filter prunes the first.
+// The deleted row must vanish and survivors must keep their _row_id.
 func TestScanPruningWithPositionalDeletes(t *testing.T) {
 	ctx := context.Background()
-	tbl := buildTwoRowGroupV3Table(t)
+	tbl := buildTwoRowGroupTable(t, newV2RowLineageTestTable(t))
 
 	tasks, err := tbl.Scan().PlanFiles(ctx)
 	require.NoError(t, err)
@@ -64,10 +65,7 @@ func TestScanPruningWithPositionalDeletes(t *testing.T) {
 		*iceberg.UnpartitionedSpec, iceberg.EntryContentPosDeletes,
 		posDelPath, iceberg.ParquetFile, nil, nil, nil, 1, 256)
 	require.NoError(t, err)
-	tx := tbl.NewTransaction()
-	require.NoError(t, tx.NewRowDelta(nil).AddDeletes(b.Build()).Commit(ctx))
-	tbl, err = tx.Commit(ctx)
-	require.NoError(t, err)
+	tbl = commitLegacyPosDelete(t, tbl, b.Build())
 
 	filter := table.WithRowFilter(iceberg.GreaterThan(iceberg.Reference("id"), int64(5)))
 
@@ -85,7 +83,7 @@ func TestScanPruningWithPositionalDeletes(t *testing.T) {
 // drops pos 6 (id=7) in the surviving second row group while the first is pruned.
 func TestScanPruningWithDeletionVector(t *testing.T) {
 	ctx := context.Background()
-	tbl := buildTwoRowGroupV3Table(t)
+	tbl := buildTwoRowGroupTable(t, newV3RowLineageTestTable(t))
 
 	tasks, err := tbl.Scan().PlanFiles(ctx)
 	require.NoError(t, err)
@@ -119,7 +117,7 @@ func TestScanPruningWithDeletionVector(t *testing.T) {
 // must keep advancing across batch boundaries, not reset per batch.
 func TestScanPruningSurvivingGroupSpansBatches(t *testing.T) {
 	ctx := context.Background()
-	tbl := buildTwoRowGroupV3Table(t)
+	tbl := buildTwoRowGroupTable(t, newV3RowLineageTestTable(t))
 
 	tx := tbl.NewTransaction()
 	require.NoError(t, tx.SetProperties(iceberg.Properties{table.ParquetBatchSizeKey: "2"}))
@@ -133,7 +131,7 @@ func TestScanPruningSurvivingGroupSpansBatches(t *testing.T) {
 
 func TestScanPruningAllRowGroupsCompletes(t *testing.T) {
 	ctx := context.Background()
-	tbl := buildTwoRowGroupV3Table(t)
+	tbl := buildTwoRowGroupTable(t, newV3RowLineageTestTable(t))
 
 	_, itr, err := tbl.Scan(
 		table.WithRowLineage(),
@@ -222,12 +220,11 @@ func TestReadTaskDeletionVectorSupersedesPositionalDeletes(t *testing.T) {
 // scan cannot prove bloom skipped a group, since a correct scan emits the same
 // rows whether or not the empty group was physically read.
 
-// buildTwoRowGroupV3Table writes a v3 row-lineage table with ids 1..10 across two
-// 5-row row groups, so an id>5 filter prunes the first group via stats.
-func buildTwoRowGroupV3Table(t *testing.T) *table.Table {
+// buildTwoRowGroupTable appends ids 1..10 to tbl across two 5-row row groups,
+// so an id>5 filter prunes the first group via stats.
+func buildTwoRowGroupTable(t *testing.T, tbl *table.Table) *table.Table {
 	t.Helper()
 	ctx := context.Background()
-	tbl := newV3RowLineageTestTable(t)
 
 	tx := tbl.NewTransaction()
 	require.NoError(t, tx.SetProperties(iceberg.Properties{table.ParquetRowGroupLimitKey: "5"}))
@@ -255,6 +252,28 @@ func buildTwoRowGroupV3Table(t *testing.T) *table.Table {
 	return tbl
 }
 
+// commitLegacyPosDelete adds posDel while tbl is still v2, then upgrades it to v3.
+// The no-op Delete must keep committing a v3 snapshot: that snapshot gives the existing rows a _row_id.
+func commitLegacyPosDelete(t *testing.T, tbl *table.Table, posDel iceberg.DataFile) *table.Table {
+	t.Helper()
+	ctx := context.Background()
+
+	tx := tbl.NewTransaction()
+	require.NoError(t, tx.NewRowDelta(nil).AddDeletes(posDel).Commit(ctx))
+	require.NoError(t, tx.UpgradeFormatVersion(3))
+	require.NoError(t, tx.Delete(ctx, iceberg.AlwaysFalse{}, nil))
+	tbl, err := tx.Commit(ctx)
+	require.NoError(t, err)
+
+	tasks, err := tbl.Scan().PlanFiles(ctx)
+	require.NoError(t, err)
+	for _, task := range tasks {
+		require.NotNil(t, task.FirstRowID, "upgraded data file %s must inherit a first_row_id", task.File.FilePath())
+	}
+
+	return tbl
+}
+
 // idsMatchingFilter scans without row lineage and returns the surviving ids, sorted.
 func idsMatchingFilter(t *testing.T, tbl *table.Table, opts ...table.ScanOption) []int64 {
 	t.Helper()
diff --git a/table/row_lineage_rewrite_test.go b/table/row_lineage_rewrite_test.go
index 2c9b87c7b..d6f7e5ae2 100644
--- a/table/row_lineage_rewrite_test.go
+++ b/table/row_lineage_rewrite_test.go
@@ -302,7 +302,7 @@ func TestCoWRewriteDoesNotPruneMissingInitialDefault(t *testing.T) {
 
 func TestCoWRewriteNormalizesTheSurvivorComplement(t *testing.T) {
 	ctx := context.Background()
-	tbl := buildTwoRowGroupV3Table(t)
+	tbl := buildTwoRowGroupTable(t, newV3RowLineageTestTable(t))
 
 	var err error
 	tbl, err = tbl.Delete(ctx,
diff --git a/table/row_lineage_scan_test.go b/table/row_lineage_scan_test.go
index 909dfdd79..619c872e1 100644
--- a/table/row_lineage_scan_test.go
+++ b/table/row_lineage_scan_test.go
@@ -120,11 +120,12 @@ func TestScanRowLineagePreservedThroughEqualityDeletes(t *testing.T) {
 		"equality-deleted rows must not renumber surviving rows' _row_id")
 }
 
-// TestScanRowLineagePreservedThroughPositionalDeletes: a Parquet position-delete
-// file drops rows during the scan; survivors must keep their original _row_id.
+// TestScanRowLineagePreservedThroughPositionalDeletes: a legacy Parquet position-delete
+// file, committed before the v3 upgrade, drops rows during the scan; survivors must
+// keep their original _row_id.
 func TestScanRowLineagePreservedThroughPositionalDeletes(t *testing.T) {
 	ctx := context.Background()
-	tbl := newV3RowLineageTestTable(t)
+	tbl := newV2RowLineageTestTable(t)
 
 	arrowSchema := arrow.NewSchema([]arrow.Field{
 		{Name: "id", Type: arrow.PrimitiveTypes.Int64, Nullable: false},
@@ -138,7 +139,6 @@ func TestScanRowLineagePreservedThroughPositionalDeletes(t *testing.T) {
 
 	tbl, err = tbl.Append(ctx, array.NewTableReader(data, -1), nil)
 	require.NoError(t, err)
-	require.Equal(t, map[int64]int64{1: 0, 2: 1, 3: 2, 4: 3, 5: 4}, rowIDByID(t, tbl))
 
 	tasks, err := tbl.Scan().PlanFiles(ctx)
 	require.NoError(t, err)
@@ -156,13 +156,7 @@ func TestScanRowLineagePreservedThroughPositionalDeletes(t *testing.T) {
 		*iceberg.UnpartitionedSpec, iceberg.EntryContentPosDeletes,
 		posDelPath, iceberg.ParquetFile, nil, nil, nil, 2, 256)
 	require.NoError(t, err)
-
-	tx := tbl.NewTransaction()
-	rd := tx.NewRowDelta(nil)
-	rd.AddDeletes(b.Build())
-	require.NoError(t, rd.Commit(ctx))
-	tbl, err = tx.Commit(ctx)
-	require.NoError(t, err)
+	tbl = commitLegacyPosDelete(t, tbl, b.Build())
 
 	assert.Equal(t, map[int64]int64{1: 0, 3: 2, 5: 4}, rowIDByID(t, tbl),
 		"position-deleted rows must not renumber surviving rows' _row_id")
diff --git a/table/transaction.go b/table/transaction.go
index 1b5508f24..6a96b7d5f 100644
--- a/table/transaction.go
+++ b/table/transaction.go
@@ -1189,9 +1189,31 @@ type deleteFilesToAddSet struct {
 	dvsByRef map[string]rewriteDeleteFileAddition
 }
 
-func validateDeletionVectorFormatVersion(df iceberg.DataFile, formatVersion int, operation string) error {
-	if IsDeletionVector(df) && formatVersion < 3 {
-		return fmt.Errorf("deletion vector %s requires table format version >= 3 for %s", df.FilePath(), operation)
+// validateDeletionVectorToAdd is safe on any DataFile and no-ops on non-DVs.
+func validateDeletionVectorToAdd(df iceberg.DataFile, formatVersion int, operation string) error {
+	if !IsDeletionVector(df) {
+		return nil
+	}
+	path := df.FilePath()
+	if formatVersion < 3 {
+		return fmt.Errorf("deletion vector %s requires table format version >= 3 for %s", path, operation)
+	}
+	if ref := df.ReferencedDataFile(); ref == nil || *ref == "" {
+		return fmt.Errorf("deletion vector %s is missing referenced_data_file for %s", path, operation)
+	}
+	offset := df.ContentOffset()
+	if offset == nil {
+		return fmt.Errorf("deletion vector %s is missing content_offset for %s", path, operation)
+	}
+	if *offset < 0 {
+		return fmt.Errorf("deletion vector %s has invalid content_offset %d for %s", path, *offset, operation)
+	}
+	length := df.ContentSizeInBytes()
+	if length == nil {
+		return fmt.Errorf("deletion vector %s is missing content_size_in_bytes for %s", path, operation)
+	}
+	if *length <= 0 {
+		return fmt.Errorf("deletion vector %s has invalid content_size_in_bytes %d for %s", path, *length, operation)
 	}
 
 	return nil
@@ -1265,10 +1287,6 @@ func (t *Transaction) validateDeleteFilesToAdd(deleteFiles []rewriteDeleteFileAd
 			}
 		}
 
-		if err := validateDeletionVectorFormatVersion(df, meta.formatVersion, operation); err != nil {
-			return nil, err
-		}
-
 		if !IsDeletionVector(df) {
 			if meta.formatVersion >= 3 && df.ContentType() == iceberg.EntryContentPosDeletes {
 				return nil, fmt.Errorf("position delete file %s must be a deletion vector for v%d table for %s",
@@ -1286,54 +1304,39 @@ func (t *Transaction) validateDeleteFilesToAdd(deleteFiles []rewriteDeleteFileAd
 			continue
 		}
 
-		if IsDeletionVector(df) {
-			ref := df.ReferencedDataFile()
-			if ref == nil || *ref == "" {
-				return nil, fmt.Errorf("deletion vector to add is missing referenced_data_file for %s", operation)
-			}
-			offset := df.ContentOffset()
-			if offset == nil {
-				return nil, fmt.Errorf("deletion vector %s is missing content_offset for %s", path, operation)
-			}
-			if *offset < 0 {
-				return nil, fmt.Errorf("deletion vector %s has invalid content_offset %d for %s", path, *offset, operation)
-			}
-			length := df.ContentSizeInBytes()
-			if length == nil {
-				return nil, fmt.Errorf("deletion vector %s is missing content_size_in_bytes for %s", path, operation)
-			}
-			if *length <= 0 {
-				return nil, fmt.Errorf("deletion vector %s has invalid content_size_in_bytes %d for %s", path, *length, operation)
-			}
+		if err := validateDeletionVectorToAdd(df, meta.formatVersion, operation); err != nil {
+			return nil, err
+		}
+		// non-nil: validateDeletionVectorToAdd checked all three.
+		ref, offset, length := df.ReferencedDataFile(), df.ContentOffset(), df.ContentSizeInBytes()
 
-			blob := deletionVectorBlobKey{path: path, offset: *offset, length: *length}
-			if setToAdd.dvBlobs == nil {
-				setToAdd.dvBlobs = make(map[deletionVectorBlobKey]struct{}, len(deleteFiles))
-			}
-			if _, ok := setToAdd.dvBlobs[blob]; ok {
-				return nil, fmt.Errorf("deletion vector blob identity must be unique for %s: %s at offset %d with length %d",
-					operation, path, *offset, *length)
-			}
-			if _, ok := setToAdd.dvsByRef[*ref]; ok {
-				return nil, fmt.Errorf("deletion vectors to add must reference distinct data files for %s: %s",
-					operation, *ref)
-			}
-			if pathAlreadyAdded {
-				if _, ok := setToAdd.dvPaths[path]; !ok {
-					return nil, fmt.Errorf("delete file path %s cannot identify both a deletion vector container and a regular delete file for %s",
-						path, operation)
-				}
-			}
-			if setToAdd.dvPaths == nil {
-				setToAdd.dvPaths = make(map[string]struct{}, len(deleteFiles))
-			}
-			if setToAdd.dvsByRef == nil {
-				setToAdd.dvsByRef = make(map[string]rewriteDeleteFileAddition, len(deleteFiles))
+		blob := deletionVectorBlobKey{path: path, offset: *offset, length: *length}
+		if setToAdd.dvBlobs == nil {
+			setToAdd.dvBlobs = make(map[deletionVectorBlobKey]struct{}, len(deleteFiles))
+		}
+		if _, ok := setToAdd.dvBlobs[blob]; ok {
+			return nil, fmt.Errorf("deletion vector blob identity must be unique for %s: %s at offset %d with length %d",
+				operation, path, *offset, *length)
+		}
+		if _, ok := setToAdd.dvsByRef[*ref]; ok {
+			return nil, fmt.Errorf("deletion vectors to add must reference distinct data files for %s: %s",
+				operation, *ref)
+		}
+		if pathAlreadyAdded {
+			if _, ok := setToAdd.dvPaths[path]; !ok {
+				return nil, fmt.Errorf("delete file path %s cannot identify both a deletion vector container and a regular delete file for %s",
+					path, operation)
 			}
-			setToAdd.dvPaths[path] = struct{}{}
-			setToAdd.dvBlobs[blob] = struct{}{}
-			setToAdd.dvsByRef[*ref] = addition
 		}
+		if setToAdd.dvPaths == nil {
+			setToAdd.dvPaths = make(map[string]struct{}, len(deleteFiles))
+		}
+		if setToAdd.dvsByRef == nil {
+			setToAdd.dvsByRef = make(map[string]rewriteDeleteFileAddition, len(deleteFiles))
+		}
+		setToAdd.dvPaths[path] = struct{}{}
+		setToAdd.dvBlobs[blob] = struct{}{}
+		setToAdd.dvsByRef[*ref] = addition
 	}
 
 	return setToAdd, nil