You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iceberg.apache.org by ze...@apache.org on 2026/03/31 20:43:43 UTC
(iceberg-go) branch main updated: feat(table): equality delete writing for partitioned tables (#823)
This is an automated email from the ASF dual-hosted git repository.
zeroshade 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 e48015ba feat(table): equality delete writing for partitioned tables (#823)
e48015ba is described below
commit e48015ba25577dd9f8964c1a425ae3ba15711a3b
Author: Andrei Tserakhau <la...@gmail.com>
AuthorDate: Tue Mar 31 22:43:32 2026 +0200
feat(table): equality delete writing for partitioned tables (#823)
Closes #808.
For partitioned tables, the equality delete writer now expands the
record schema to include partition source columns alongside the equality
key columns. This allows the existing partitionedFanoutWriter to extract
partition values and route records to the correct partition directories.
The actual delete files still contain only the equality key columns —
partition columns are used for routing and stripped during the
ToRequestedSchema projection.
When partition source columns overlap with the equality key (e.g.,
deleting by partition column), no extra columns are needed.
---
table/equality_delete_writer.go | 70 +++++++++++++++++++++++++++----
table/equality_delete_writer_test.go | 80 +++++++++++++++++++++++++++++++-----
table/rolling_data_writer.go | 38 ++++++++++-------
3 files changed, 154 insertions(+), 34 deletions(-)
diff --git a/table/equality_delete_writer.go b/table/equality_delete_writer.go
index fa39522d..317ca1ad 100644
--- a/table/equality_delete_writer.go
+++ b/table/equality_delete_writer.go
@@ -25,6 +25,7 @@ import (
"github.com/apache/arrow-go/v18/arrow"
"github.com/apache/iceberg-go"
+ "github.com/apache/iceberg-go/config"
"github.com/apache/iceberg-go/internal"
iceio "github.com/apache/iceberg-go/io"
"github.com/google/uuid"
@@ -62,8 +63,11 @@ func equalityDeleteSchema(tableSchema *iceberg.Schema, fieldIDs []int) (*iceberg
//
// The table must use format version 2 or higher.
//
-// Note: partitioned tables are not yet supported for equality delete
-// writing. See https://github.com/apache/iceberg-go/issues/808
+// For partitioned tables, the provided records must include the partition
+// source columns in addition to the equality key columns so that records
+// can be routed to the correct partition directories. If the partition
+// source columns overlap with the equality key columns, no extra columns
+// are needed.
//
// Usage:
//
@@ -162,14 +166,28 @@ func equalityDeleteRecordsToDataFiles(ctx context.Context, rootLocation string,
return nil, err
}
- // TODO(#808): support partitioned tables for equality delete writing.
- // The partitioned fanout writer assumes partition source columns are
- // present in the record, but equality delete records only contain the
- // delete key columns. Needs either explicit partition values from the
- // caller (Java pattern) or automatic partition column inclusion.
- // See https://github.com/apache/iceberg-go/issues/808
if !latestMetadata.PartitionSpec().IsUnpartitioned() {
- return nil, errors.New("equality delete writing for partitioned tables is not yet supported")
+ tableSchema := meta.CurrentSchema()
+ partitionSpec := latestMetadata.PartitionSpec()
+
+ writeSchema, err := equalityDeleteWriteSchema(tableSchema, equalityFieldIDs, partitionSpec)
+ if err != nil {
+ return nil, err
+ }
+
+ factory, err := newWriterFactory(rootLocation, args, meta, writeSchema, targetFileSize,
+ withContentType(iceberg.EntryContentEqDeletes),
+ withFactoryFileSchema(deleteSchema),
+ withFactoryEqualityFieldIDs(equalityFieldIDs))
+ if err != nil {
+ return nil, err
+ }
+
+ partitionWriter := newPartitionedFanoutWriter(
+ partitionSpec, cw, writeSchema, args.itr, factory)
+ workers := config.EnvConfig.MaxWorkers
+
+ return partitionWriter.Write(ctx, workers), nil
}
nextCount, stopCount := iter.Pull(args.counter)
@@ -196,3 +214,37 @@ func equalityDeleteRecordsToDataFiles(ctx context.Context, rootLocation string,
return cw.writeFiles(ctx, rootLocation, args.fs, meta, meta.props, nil, tasks), nil
}
+
+// equalityDeleteWriteSchema returns a schema containing the union of equality
+// key columns and partition source columns. This allows the partitioned fanout
+// writer to extract partition values from the records while keeping the delete
+// key as the equality identifier.
+func equalityDeleteWriteSchema(tableSchema *iceberg.Schema, equalityFieldIDs []int, spec iceberg.PartitionSpec) (*iceberg.Schema, error) {
+ seen := make(map[int]struct{}, len(equalityFieldIDs))
+ names := make([]string, 0, len(equalityFieldIDs))
+
+ // Equality key columns first (deterministic order).
+ for _, id := range equalityFieldIDs {
+ name, ok := tableSchema.FindColumnName(id)
+ if !ok {
+ return nil, fmt.Errorf("%w: field ID %d not found in table schema", iceberg.ErrInvalidSchema, id)
+ }
+ seen[id] = struct{}{}
+ names = append(names, name)
+ }
+
+ // Partition source columns not already in the equality key.
+ for _, f := range spec.Fields() {
+ if _, ok := seen[f.SourceID]; ok {
+ continue
+ }
+ name, ok := tableSchema.FindColumnName(f.SourceID)
+ if !ok {
+ return nil, fmt.Errorf("%w: partition source field ID %d not found in table schema", iceberg.ErrInvalidSchema, f.SourceID)
+ }
+ seen[f.SourceID] = struct{}{}
+ names = append(names, name)
+ }
+
+ return tableSchema.Select(true, names...)
+}
diff --git a/table/equality_delete_writer_test.go b/table/equality_delete_writer_test.go
index 72c66a04..1820c51d 100644
--- a/table/equality_delete_writer_test.go
+++ b/table/equality_delete_writer_test.go
@@ -19,6 +19,7 @@ package table_test
import (
"context"
+ "os"
"path/filepath"
"strings"
"testing"
@@ -225,7 +226,9 @@ func TestWriteEqualityDeleteFilesRejectsInvalidFieldID(t *testing.T) {
require.ErrorIs(t, err, iceberg.ErrInvalidSchema)
}
-func TestWriteEqualityDeleteFilesRejectsPartitionedTable(t *testing.T) {
+func newPartitionedEqDeleteTestTable(t *testing.T) *table.Table {
+ t.Helper()
+
location := filepath.ToSlash(t.TempDir())
iceSchema := iceberg.NewSchema(0,
@@ -242,27 +245,84 @@ func TestWriteEqualityDeleteFilesRejectsPartitionedTable(t *testing.T) {
iceberg.Properties{table.PropertyFormatVersion: "2"})
require.NoError(t, err)
- tbl := table.New(
- table.Identifier{"db", "partitioned_test"},
+ return table.New(
+ table.Identifier{"db", "partitioned_eq_del_test"},
meta, location+"/metadata/v1.metadata.json",
func(ctx context.Context) (iceio.IO, error) {
return iceio.LocalFS{}, nil
},
&rowDeltaCatalog{metadata: meta},
)
+}
- delArrowSc, err := table.SchemaToArrowSchema(
- iceberg.NewSchema(0, iceberg.NestedField{ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int64, Required: true}),
- nil, true, false)
+func TestWriteEqualityDeleteFilesPartitionedTable(t *testing.T) {
+ tbl := newPartitionedEqDeleteTestTable(t)
+
+ // Delete records must include both the equality key (id) and partition source (category)
+ delSchema := iceberg.NewSchema(0,
+ iceberg.NestedField{ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int64, Required: true},
+ iceberg.NestedField{ID: 2, Name: "category", Type: iceberg.PrimitiveTypes.String, Required: true},
+ )
+ delArrowSc, err := table.SchemaToArrowSchema(delSchema, nil, true, false)
require.NoError(t, err)
- records, release := makeEqDeleteRecords(t, delArrowSc, `[{"id": 1}]`)
+ // 2 books + 1 music = should produce 2 partitioned files
+ records, release := makeEqDeleteRecords(t, delArrowSc,
+ `[{"id": 10, "category": "books"}, {"id": 20, "category": "music"}, {"id": 30, "category": "books"}]`)
defer release()
tx := tbl.NewTransaction()
- _, err = tx.WriteEqualityDeletes(t.Context(), []int{1}, records)
- require.Error(t, err)
- assert.Contains(t, err.Error(), "partitioned tables is not yet supported")
+ files, err := tx.WriteEqualityDeletes(t.Context(), []int{1}, records)
+ require.NoError(t, err)
+ require.Len(t, files, 2, "should produce one file per partition")
+
+ for _, df := range files {
+ assert.Equal(t, iceberg.EntryContentEqDeletes, df.ContentType())
+ assert.Equal(t, []int{1}, df.EqualityFieldIDs())
+ assert.True(t, df.Count() > 0)
+ }
+
+ // Verify Parquet file content: files should contain only the equality key column (id), not category
+ for _, df := range files {
+ func() {
+ f, err := os.Open(df.FilePath())
+ require.NoError(t, err)
+ defer f.Close()
+
+ rdr, err := file.NewParquetReader(f)
+ require.NoError(t, err)
+ defer rdr.Close()
+
+ pqSchema := rdr.MetaData().Schema
+ assert.Equal(t, 1, pqSchema.NumColumns(), "parquet file should contain only the equality key column")
+ assert.Equal(t, "id", pqSchema.Column(0).Name())
+ }()
+ }
+}
+
+func TestWriteEqualityDeleteFilesPartitionedKeyOverlapsPartition(t *testing.T) {
+ tbl := newPartitionedEqDeleteTestTable(t)
+
+ // When the equality key IS the partition column, no extra columns needed
+ delSchema := iceberg.NewSchema(0,
+ iceberg.NestedField{ID: 2, Name: "category", Type: iceberg.PrimitiveTypes.String, Required: true},
+ )
+ delArrowSc, err := table.SchemaToArrowSchema(delSchema, nil, true, false)
+ require.NoError(t, err)
+
+ records, release := makeEqDeleteRecords(t, delArrowSc,
+ `[{"category": "books"}, {"category": "music"}]`)
+ defer release()
+
+ tx := tbl.NewTransaction()
+ files, err := tx.WriteEqualityDeletes(t.Context(), []int{2}, records)
+ require.NoError(t, err)
+ require.NotEmpty(t, files)
+
+ for _, df := range files {
+ assert.Equal(t, iceberg.EntryContentEqDeletes, df.ContentType())
+ assert.Equal(t, []int{2}, df.EqualityFieldIDs())
+ }
}
func TestWriteEqualityDeleteFilesCommitViaRowDelta(t *testing.T) {
diff --git a/table/rolling_data_writer.go b/table/rolling_data_writer.go
index 5462e0e3..8e06755d 100644
--- a/table/rolling_data_writer.go
+++ b/table/rolling_data_writer.go
@@ -44,15 +44,16 @@ type writerFactory struct {
taskSchema *iceberg.Schema
targetFileSize int64
- locProvider LocationProvider
- fileSchema *iceberg.Schema
- arrowSchema *arrow.Schema
- writeProps any
- statsCols map[int]tblutils.StatisticsCollector
- currentSpec iceberg.PartitionSpec
- fileFormat iceberg.FileFormat
- format tblutils.FileFormat
- content iceberg.ManifestEntryContent
+ locProvider LocationProvider
+ fileSchema *iceberg.Schema
+ arrowSchema *arrow.Schema
+ writeProps any
+ statsCols map[int]tblutils.StatisticsCollector
+ currentSpec iceberg.PartitionSpec
+ fileFormat iceberg.FileFormat
+ format tblutils.FileFormat
+ content iceberg.ManifestEntryContent
+ equalityFieldIDs []int
writers sync.Map
partitionLocProviders sync.Map
@@ -71,6 +72,12 @@ func withContentType(content iceberg.ManifestEntryContent) writerFactoryOption {
}
}
+func withFactoryEqualityFieldIDs(ids []int) writerFactoryOption {
+ return func(w *writerFactory) {
+ w.equalityFieldIDs = ids
+ }
+}
+
func withFactoryFileSchema(schema *iceberg.Schema) writerFactoryOption {
return func(w *writerFactory) {
w.fileSchema = schema
@@ -193,12 +200,13 @@ func (w *writerFactory) openFileWriter(ctx context.Context, partitionPath string
}
return w.format.NewFileWriter(ctx, w.fs, partitionValues, tblutils.WriteFileInfo{
- FileSchema: w.fileSchema,
- FileName: filePath,
- StatsCols: w.statsCols,
- WriteProps: w.writeProps,
- Spec: w.currentSpec,
- Content: w.content,
+ FileSchema: w.fileSchema,
+ FileName: filePath,
+ StatsCols: w.statsCols,
+ WriteProps: w.writeProps,
+ Spec: w.currentSpec,
+ Content: w.content,
+ EqualityFieldIDs: w.equalityFieldIDs,
}, w.arrowSchema)
}