You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@iceberg.apache.org by "alexandre-normand (via GitHub)" <gi...@apache.org> on 2026/02/12 05:22:05 UTC
[PR] feat(table): add support for merge-on-read delete [iceberg-go]
alexandre-normand opened a new pull request, #721:
URL: https://github.com/apache/iceberg-go/pull/721
(no comment)
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(table): add support for merge-on-read delete [iceberg-go]
Posted by "alexandre-normand (via GitHub)" <gi...@apache.org>.
alexandre-normand commented on code in PR #721:
URL: https://github.com/apache/iceberg-go/pull/721#discussion_r2850530628
##########
table/arrow_utils.go:
##########
@@ -1358,12 +1370,87 @@ func recordsToDataFiles(ctx context.Context, rootLocation string, meta *Metadata
}
}
- return writeFiles(ctx, rootLocation, args.fs, meta, nil, tasks)
- } else {
- rollingDataWriters := NewWriterFactory(rootLocation, args, meta, taskSchema, targetFileSize)
- partitionWriter := newPartitionedFanoutWriter(*currentSpec, meta.CurrentSchema(), args.itr, &rollingDataWriters)
- workers := config.EnvConfig.MaxWorkers
+ return cw.writeFiles(ctx, rootLocation, args.fs, meta, meta.props, nil, tasks)
+ }
+
+ factory := NewWriterFactory(rootLocation, args, meta, taskSchema, targetFileSize)
+ partitionWriter := newPartitionedFanoutWriter(*currentSpec, cw, meta.CurrentSchema(), args.itr, &factory)
+ workers := config.EnvConfig.MaxWorkers
- return partitionWriter.Write(ctx, workers)
+ return partitionWriter.Write(ctx, workers)
+}
+
+func positionDeleteRecordsToDataFiles(ctx context.Context, rootLocation string, meta *MetadataBuilder, partitionDataPerFile map[string]map[int]any, args recordWritingArgs) (ret iter.Seq2[iceberg.DataFile, error]) {
+ if args.counter == nil {
+ args.counter = internal.Counter(0)
}
+
+ defer func() {
+ if r := recover(); r != nil {
+ var err error
+ switch e := r.(type) {
+ case error:
+ err = fmt.Errorf("error encountered during position delete file writing: %w", e)
+ default:
+ err = fmt.Errorf("error encountered during position delete file writing: %v", e)
+ }
+ ret = func(yield func(iceberg.DataFile, error) bool) {
+ yield(nil, err)
+ }
+ }
+ }()
+
+ if args.writeUUID == nil {
+ u := uuid.Must(uuid.NewRandom())
+ args.writeUUID = &u
+ }
+
+ targetFileSize := int64(meta.props.GetInt(WriteTargetFileSizeBytesKey,
+ WriteTargetFileSizeBytesDefault))
+
+ currentSpec, err := meta.CurrentSpec()
Review Comment:
I tried to address this but let me know if that's correct. I extended the mapping per data file to include the spec id in addition to the partition data we were already tracking that way. That's passed down now and the position delete fanout writer uses the partition spec from the partition context.
That's done in cc0a966778dc56f5f8383f7f993d438c6750852f. The one case I was unsure about is the one where the table's latest metadata is unpartitioned. In that case, we were short-circuiting early and skipping the fanout writer. That case remains but I'm not sure what the ramifications are.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(table): add support for merge-on-read delete [iceberg-go]
Posted by "laskoviymishka (via GitHub)" <gi...@apache.org>.
laskoviymishka commented on code in PR #721:
URL: https://github.com/apache/iceberg-go/pull/721#discussion_r2849808292
##########
table/transaction.go:
##########
@@ -915,6 +920,45 @@ func (t *Transaction) performCopyOnWriteDeletion(ctx context.Context, operation
return updater, nil
}
+func (t *Transaction) performMergeOnReadDeletion(ctx context.Context, snapshotProps iceberg.Properties, filter iceberg.BooleanExpression, caseSensitive bool, concurrency int) (*snapshotProducer, error) {
+ fs, err := t.tbl.fsF(ctx)
+ if err != nil {
+ return nil, err
+ }
+
+ if t.meta.NameMapping() == nil {
+ nameMapping := t.meta.CurrentSchema().NameMapping()
+ mappingJson, err := json.Marshal(nameMapping)
+ if err != nil {
+ return nil, err
+ }
+ err = t.SetProperties(iceberg.Properties{DefaultNameMappingKey: string(mappingJson)})
+ if err != nil {
+ return nil, err
+ }
+ }
+
+ commitUUID := uuid.New()
+ updater := t.updateSnapshot(fs, snapshotProps, OpDelete).mergeOverwrite(&commitUUID)
Review Comment:
For merge-on-read deletes, this path still builds an overwrite-style snapshot producer (mergeOverwrite) instead of a RowDelta-style update. In Iceberg Java, MOR deletes are modeled through newRowDelta() (BaseTransaction#newRowDelta) and validated via BaseRowDelta conflict checks. Could we align this flow with RowDelta semantics to avoid divergence in commit validation behavior?
see:
https://github.com/apache/iceberg/blob/main/core/src/main/java/org/apache/iceberg/BaseTransaction.java
https://github.com/apache/iceberg/blob/main/core/src/main/java/org/apache/iceberg/BaseRowDelta.java
##########
table/transaction_test.go:
##########
@@ -536,6 +536,120 @@ func (s *SparkIntegrationTestSuite) TestDeleteInsensitive() {
+----------+---------+---+`)
}
+func (s *SparkIntegrationTestSuite) TestDeleteMergeOnReadUnpartitioned() {
+ icebergSchema := iceberg.NewSchema(0,
+ iceberg.NestedField{ID: 1, Name: "first_name", Type: iceberg.PrimitiveTypes.String},
+ iceberg.NestedField{ID: 2, Name: "last_name", Type: iceberg.PrimitiveTypes.String},
+ iceberg.NestedField{ID: 3, Name: "age", Type: iceberg.PrimitiveTypes.Int32},
+ )
+
+ tbl, err := s.cat.CreateTable(s.ctx, catalog.ToIdentifier("default", "go_test_merge_on_read_delete"), icebergSchema,
+ catalog.WithProperties(
+ map[string]string{
+ table.WriteDeleteModeKey: table.WriteModeMergeOnRead,
+ },
+ ),
+ )
+ s.Require().NoError(err)
+
+ arrowSchema, err := table.SchemaToArrowSchema(icebergSchema, nil, true, false)
+ s.Require().NoError(err)
+
+ initialTable, err := array.TableFromJSON(memory.DefaultAllocator, arrowSchema, []string{
+ `[
+ {"first_name": "alan", "last_name": "gopher", "age": 7},
+ {"first_name": "steve", "last_name": "gopher", "age": 5},
+ {"first_name": "dead", "last_name": "gopher", "age": 97}
+ ]`,
+ })
+ s.Require().NoError(err)
+ defer initialTable.Release()
+
+ tx := tbl.NewTransaction()
+ err = tx.AppendTable(s.ctx, initialTable, 3, nil)
+ s.Require().NoError(err)
+ tbl, err = tx.Commit(s.ctx)
+ s.Require().NoError(err)
+
+ // Delete the dead gopher and confirm that alan and steve are still present
+ filter := iceberg.EqualTo(iceberg.Reference("first_name"), "dead")
+ tx = tbl.NewTransaction()
+ err = tx.Delete(s.ctx, filter, nil)
+ s.Require().NoError(err)
+ _, err = tx.Commit(s.ctx)
+ s.Require().NoError(err)
+
+ output, err := recipe.ExecuteSpark(s.T(), "./validation.py", "--sql", "SELECT * FROM default.go_test_merge_on_read_delete ORDER BY age")
+ s.Require().NoError(err)
+ s.Require().Contains(output, `|first_name|last_name|age|
++----------+---------+---+
+|steve |gopher |5 |
+|alan |gopher |7 |
++----------+---------+---+`)
+}
+
+func (s *SparkIntegrationTestSuite) TestDeleteMergeOnReadPartitioned() {
+ icebergSchema := iceberg.NewSchema(0,
+ iceberg.NestedField{ID: 1, Name: "first_name", Type: iceberg.PrimitiveTypes.String},
+ iceberg.NestedField{ID: 2, Name: "last_name", Type: iceberg.PrimitiveTypes.String},
+ iceberg.NestedField{ID: 3, Name: "age", Type: iceberg.PrimitiveTypes.Int32},
+ )
+
+ spec := iceberg.NewPartitionSpec(iceberg.PartitionField{
+ SourceID: 3,
+ Name: "age_bucket",
+ Transform: iceberg.BucketTransform{
+ NumBuckets: 2,
+ },
+ })
+ tbl, err := s.cat.CreateTable(s.ctx, catalog.ToIdentifier("default", "go_test_merge_on_read_delete_partitioned"), icebergSchema,
+ catalog.WithProperties(
+ map[string]string{
+ table.WriteDeleteModeKey: table.WriteModeMergeOnRead,
+ },
+ ),
+ catalog.WithPartitionSpec(&spec),
+ )
+ s.Require().NoError(err)
+
+ arrowSchema, err := table.SchemaToArrowSchema(icebergSchema, nil, true, false)
+ s.Require().NoError(err)
+
+ initialTable, err := array.TableFromJSON(memory.DefaultAllocator, arrowSchema, []string{
+ `[
+ {"first_name": "alan", "last_name": "gopher", "age": 7},
+ {"first_name": "steve", "last_name": "gopher", "age": 5},
+ {"first_name": "dead", "last_name": "gopher", "age": 97}
Review Comment:
nit: missing coma, this should panic
##########
table/arrow_scanner.go:
##########
@@ -473,6 +523,78 @@ func (as *arrowScan) recordsFromTask(ctx context.Context, task internal.Enumerat
return err
}
+func (as *arrowScan) producePosDeletesFromTask(ctx context.Context, task internal.Enumerated[FileScanTask], positionalDeletes positionDeletes, out chan<- enumeratedRecord) (err error) {
+ defer func() {
+ if err != nil {
+ out <- enumeratedRecord{Task: task, Err: err}
+ }
+ }()
+
+ var (
+ rdr internal.FileReader
+ iceSchema *iceberg.Schema
+ colIndices []int
+ filterFunc recProcessFn
+ dropFile bool
+ )
+
+ iceSchema, colIndices, rdr, err = as.prepareToRead(ctx, task.Value.File)
+ if err != nil {
+ return err
+ }
+ defer iceinternal.CheckedClose(rdr, &err)
+
+ fields := append(iceSchema.Fields(), iceberg.PositionalDeleteSchema.Fields()...)
+ enrichedIcebergSchema := iceberg.NewSchema(iceSchema.ID+1, fields...)
+
+ pipeline := make([]recProcessFn, 0, 2)
+ pipeline = append(pipeline, enrichRecordsWithPosDeleteFields(ctx, task.Value.File))
+ if len(positionalDeletes) > 0 {
+ deletes := set[int64]{}
+ for _, chunk := range positionalDeletes {
+ for _, a := range chunk.Chunks() {
+ for _, v := range a.(*array.Int64).Int64Values() {
+ deletes[v] = struct{}{}
+ }
+ }
+ }
+
+ pipeline = append(pipeline, processPositionalDeletes(ctx, deletes))
+ }
+
+ filterFunc, dropFile, err = as.getRecordFilter(ctx, iceSchema)
+ if err != nil {
+ return err
+ }
+
+ // Nothing to delete in a dropped file
+ if dropFile {
+ var emptySchema *arrow.Schema
+ emptySchema, err = SchemaToArrowSchema(as.projectedSchema, nil, false, as.useLargeTypes)
Review Comment:
nit: when dropFile is true (filter resolves to AlwaysFalse for this file, so nothing to delete), the empty sentinel batch is built with the data schema (projectedSchema) rather than PositionalDeleteArrowSchema.
It's harmless today because createIterator skips zero-row batches, but it's a latent schema mismatch. Suggest using PositionalDeleteArrowSchema here to keep things consistent with the rest of the pipeline.
##########
table/transaction.go:
##########
@@ -1212,6 +1282,117 @@ func (t *Transaction) rewriteSingleFile(ctx context.Context, fs io.IO, originalF
return result, nil
}
+// writePositionDeletesForFiles rewrites data files by preserving only rows that do NOT match the filter
+func (t *Transaction) writePositionDeletesForFiles(ctx context.Context, fs io.IO, updater *snapshotProducer, files []iceberg.DataFile, filter iceberg.BooleanExpression, caseSensitive bool, concurrency int) error {
+ posDeleteRecIter, err := t.makePositionDeleteRecordsForFilter(ctx, fs, files, filter, caseSensitive, concurrency)
+ if err != nil {
+ return err
+ }
+
+ partitionDataPerFile := make(map[string]map[int]any, len(files))
+ for _, df := range files {
+ partitionPath, _ := path.Split(df.FilePath())
+ partitionDataPerFile[partitionPath] = df.Partition()
Review Comment:
Partition routing is currently keyed by the **parent directory string** (`path.Split(df.FilePath())`). That’s a path convention, not something guaranteed by the Iceberg spec.
If two data files from different partitions share the same directory (possible with a custom `LocationProvider` or flat `write.data.path`), the second entry silently overwrites the first. As a result, position delete files can be written with incorrect partition values.
In Java, routing is never derived from the file path. It is keyed by the `(PartitionSpec, partition)` pair taken directly from the data file metadata. See
`SparkPositionDeltaWrite.DeleteOnlyDeltaWriter.delete()`:
[https://github.com/apache/iceberg/blob/master/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/source/SparkPositionDeltaWrite.java](https://github.com/apache/iceberg/blob/master/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/source/SparkPositionDeltaWrite.java)
```java
int specId = metadata.getInt(specIdOrdinal);
PartitionSpec spec = specs.get(specId);
InternalRow partition = metadata.getStruct(partitionOrdinal, ...);
delegate.write(positionDelete, spec, partitionProjection); // (spec, partition)
```
And `ClusteredWriter.write()` opens a new writer per unique `(spec, partition)` pair — again, not path-based.
**Simplest compatible fix:**
Key the routing map by the full `df.FilePath()` (not its parent directory), and in `processBatch` look up by the full `file_path` column value instead of stripping the filename.
This is a potential data corruption, so CRIT.
##########
table/pos_delete_partitioned_fanout_writer.go:
##########
@@ -0,0 +1,132 @@
+// 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"
+ "iter"
+ "path"
+
+ "github.com/apache/arrow-go/v18/arrow"
+ "github.com/apache/arrow-go/v18/arrow/array"
+ "github.com/apache/iceberg-go"
+ "golang.org/x/sync/errgroup"
+)
+
+// positionDeletePartitionedFanoutWriter distributes Arrow position delete records across multiple partitions based on
+// a partition specification, writing data to separate delete files for each partition using
+// a fanout pattern with configurable parallelism.
+type positionDeletePartitionedFanoutWriter struct {
+ partitionSpec iceberg.PartitionSpec
+ partitionDataByFilePath map[string]map[int]any
+ schema *iceberg.Schema
+ itr iter.Seq2[arrow.RecordBatch, error]
+ writerFactory *writerFactory
+ concurrentDataFileWriter *concurrentDataFileWriter
+}
+
+// newPositionDeletePartitionedFanoutWriter creates a new PartitionedFanoutWriter with the specified
+// partition specification, schema, and record iterator.
+func newPositionDeletePartitionedFanoutWriter(partitionSpec iceberg.PartitionSpec, concurrentWriter *concurrentDataFileWriter, partitionDataByFilePath map[string]map[int]any, itr iter.Seq2[arrow.RecordBatch, error], writerFactory *writerFactory) *positionDeletePartitionedFanoutWriter {
+ return &positionDeletePartitionedFanoutWriter{
+ partitionSpec: partitionSpec,
+ partitionDataByFilePath: partitionDataByFilePath,
+ schema: iceberg.PositionalDeleteSchema,
+ itr: itr,
+ writerFactory: writerFactory,
+ concurrentDataFileWriter: concurrentWriter,
+ }
+}
+
+// Write writes the Arrow records to the specified location using a fanout pattern with
+// the specified number of workers. The returned iterator yields the data files written
+// by the fanout process.
+func (p *positionDeletePartitionedFanoutWriter) Write(ctx context.Context, workers int) iter.Seq2[iceberg.DataFile, error] {
+ inputRecordsCh := make(chan arrow.RecordBatch, workers)
+ outputDataFilesCh := make(chan iceberg.DataFile, workers)
+
+ fanoutWorkers, ctx := errgroup.WithContext(ctx)
+ startRecordFeeder(ctx, p.itr, fanoutWorkers, inputRecordsCh)
+
+ for range workers {
+ fanoutWorkers.Go(func() error {
+ return p.fanout(ctx, inputRecordsCh, outputDataFilesCh)
+ })
+ }
+
+ return p.yieldDataFiles(fanoutWorkers, outputDataFilesCh)
+}
+
+func (p *positionDeletePartitionedFanoutWriter) fanout(ctx context.Context, inputRecordsCh <-chan arrow.RecordBatch, dataFilesChannel chan<- iceberg.DataFile) error {
+ for {
+ select {
+ case <-ctx.Done():
+ return context.Cause(ctx)
+
+ case record, ok := <-inputRecordsCh:
+ if !ok {
+ return nil
+ }
+
+ err := p.processBatch(ctx, record, dataFilesChannel)
+ if err != nil {
+ return err
+ }
+ }
+ }
+}
+
+func (p *positionDeletePartitionedFanoutWriter) processBatch(ctx context.Context, batch arrow.RecordBatch, dataFilesChannel chan<- iceberg.DataFile) (err error) {
+ defer batch.Release()
+
+ select {
+ case <-ctx.Done():
+ return context.Cause(ctx)
+ default:
+ }
+
+ if batch.NumRows() == 0 {
+ return
+ }
+
+ columns := batch.Columns()
+ filePath := columns[0].(*array.String)
+ partitionPath, _ := path.Split(filePath.Value(0))
+
+ partitionValues, ok := p.partitionDataByFilePath[partitionPath]
Review Comment:
same as above
##########
table/snapshot_producers.go:
##########
@@ -628,11 +603,55 @@ func (sp *snapshotProducer) manifests() (_ []iceberg.ManifestFile, err error) {
return nil, err
}
- manifests := slices.Concat(results[0], results[1], results[2])
+ manifests := slices.Concat(addedManifests, positionDeleteManifests, deletedFilesManifests, existingManifests)
return sp.processManifests(manifests)
}
+func (sp *snapshotProducer) manifestProducer(content iceberg.ManifestContent, files []iceberg.DataFile, output *[]iceberg.ManifestFile) func() (err error) {
+ return func() (err error) {
+ out, path, err := sp.newManifestOutput()
+ if err != nil {
+ return err
+ }
+ defer internal.CheckedClose(out, &err)
+
+ counter := &internal.CountingWriter{W: out}
+ currentSpec, err := sp.txn.meta.CurrentSpec()
+ if err != nil || currentSpec == nil {
+ return fmt.Errorf("could not get current partition spec: %w", err)
+ }
+ wr, err := iceberg.NewManifestWriter(sp.txn.meta.formatVersion, counter,
+ *currentSpec, sp.txn.meta.CurrentSchema(),
+ sp.snapshotID)
+ if err != nil {
+ return err
+ }
+ defer internal.CheckedClose(wr, &err)
+
+ for _, df := range files {
+ err := wr.Add(iceberg.NewManifestEntry(iceberg.EntryStatusADDED, &sp.snapshotID,
+ nil, nil, df))
+ if err != nil {
+ return err
+ }
+ }
+
+ // close the writer to force a flush and ensure counter.Count is accurate
+ if err := wr.Close(); err != nil {
+ return err
+ }
+
+ mf, err := wr.ToManifestFile(path, counter.Count, iceberg.WithManifestFileContent(content))
+ if err != nil {
+ return err
+ }
+ *output = []iceberg.ManifestFile{mf}
+
+ return nil
+ }
+}
+
func (sp *snapshotProducer) summary(props iceberg.Properties) (Summary, error) {
Review Comment:
summary() is never updated to iterate positionDeleteFiles, so MoR delete snapshots will always report zero added-delete-files, added-position-delete-files, and added-position-deletes. The fix is a sibling loop to the existing addedFiles one in summary() — ssc.addFile already handles EntryContentPosDeletes correctly, the call just never happens for this slice.
(could be a follow up to this PR)
##########
table/transaction.go:
##########
@@ -1212,6 +1282,117 @@ func (t *Transaction) rewriteSingleFile(ctx context.Context, fs io.IO, originalF
return result, nil
}
+// writePositionDeletesForFiles rewrites data files by preserving only rows that do NOT match the filter
+func (t *Transaction) writePositionDeletesForFiles(ctx context.Context, fs io.IO, updater *snapshotProducer, files []iceberg.DataFile, filter iceberg.BooleanExpression, caseSensitive bool, concurrency int) error {
+ posDeleteRecIter, err := t.makePositionDeleteRecordsForFilter(ctx, fs, files, filter, caseSensitive, concurrency)
+ if err != nil {
+ return err
+ }
+
+ partitionDataPerFile := make(map[string]map[int]any, len(files))
+ for _, df := range files {
+ partitionPath, _ := path.Split(df.FilePath())
+ partitionDataPerFile[partitionPath] = df.Partition()
+ }
+
+ posDeleteFiles := positionDeleteRecordsToDataFiles(ctx, t.tbl.Location(), t.meta, partitionDataPerFile, recordWritingArgs{
Review Comment:
nit: writeUUID is not set here, so positionDeleteRecordsToDataFiles generates a fresh random UUID for the delete file names. Position delete files that belong to the same commit should share the commitUUID for traceability — it makes it easy to correlate which files were written together.
Maybe better passing writeUUID: &commitUUID.
##########
table/arrow_utils.go:
##########
@@ -1358,12 +1370,87 @@ func recordsToDataFiles(ctx context.Context, rootLocation string, meta *Metadata
}
}
- return writeFiles(ctx, rootLocation, args.fs, meta, nil, tasks)
- } else {
- rollingDataWriters := NewWriterFactory(rootLocation, args, meta, taskSchema, targetFileSize)
- partitionWriter := newPartitionedFanoutWriter(*currentSpec, meta.CurrentSchema(), args.itr, &rollingDataWriters)
- workers := config.EnvConfig.MaxWorkers
+ return cw.writeFiles(ctx, rootLocation, args.fs, meta, meta.props, nil, tasks)
+ }
+
+ factory := NewWriterFactory(rootLocation, args, meta, taskSchema, targetFileSize)
+ partitionWriter := newPartitionedFanoutWriter(*currentSpec, cw, meta.CurrentSchema(), args.itr, &factory)
+ workers := config.EnvConfig.MaxWorkers
- return partitionWriter.Write(ctx, workers)
+ return partitionWriter.Write(ctx, workers)
+}
+
+func positionDeleteRecordsToDataFiles(ctx context.Context, rootLocation string, meta *MetadataBuilder, partitionDataPerFile map[string]map[int]any, args recordWritingArgs) (ret iter.Seq2[iceberg.DataFile, error]) {
+ if args.counter == nil {
+ args.counter = internal.Counter(0)
}
+
+ defer func() {
+ if r := recover(); r != nil {
+ var err error
+ switch e := r.(type) {
+ case error:
+ err = fmt.Errorf("error encountered during position delete file writing: %w", e)
+ default:
+ err = fmt.Errorf("error encountered during position delete file writing: %v", e)
+ }
+ ret = func(yield func(iceberg.DataFile, error) bool) {
+ yield(nil, err)
+ }
+ }
+ }()
+
+ if args.writeUUID == nil {
+ u := uuid.Must(uuid.NewRandom())
+ args.writeUUID = &u
+ }
+
+ targetFileSize := int64(meta.props.GetInt(WriteTargetFileSizeBytesKey,
+ WriteTargetFileSizeBytesDefault))
+
+ currentSpec, err := meta.CurrentSpec()
Review Comment:
All position delete files are currently written using the table’s current partition spec, regardless of which spec the referenced data files were written under.
For tables that have undergone partition spec evolution, this can produce delete manifest entries with partition values encoded against the wrong spec. Readers that perform partition-based pruning on delete manifests may then incorrectly skip relevant deletes.
In Java, PositionDeleteWriter writes each delete file using the spec of the data file it references, not the table’s current spec.
At minimum, we should validate (and clearly document) that this implementation is only safe for tables that have not evolved their partition spec. Ideally, it should mirror the Java behavior and write delete files against the matching data file spec.
This one is also CRIT, i think this could lead to data corruption.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(table): add support for merge-on-read delete [iceberg-go]
Posted by "alexandre-normand (via GitHub)" <gi...@apache.org>.
alexandre-normand commented on code in PR #721:
URL: https://github.com/apache/iceberg-go/pull/721#discussion_r2834645942
##########
table/arrow_scanner.go:
##########
@@ -189,6 +192,54 @@ func processPositionalDeletes(ctx context.Context, deletes set[int64]) recProces
}
}
+// enrichRecordsWithPosDeleteFields enriches a RecordBatch with the columns declared in the PositionalDeleteArrowSchema
+// so that during the pipeline filtering stages that sheds filtered out records, we still have a way to
+// preserve the original position of those records.
+func enrichRecordsWithPosDeleteFields(ctx context.Context, filePath iceberg.DataFile) recProcessFn {
Review Comment:
Yes, those all make sense and I'll work on that today. I was holding off on the full testing until I got feedback I was on the right track because I wasn't super confident in my approach.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(table): add support for merge-on-read delete [iceberg-go]
Posted by "ferhatelmas (via GitHub)" <gi...@apache.org>.
ferhatelmas commented on code in PR #721:
URL: https://github.com/apache/iceberg-go/pull/721#discussion_r2875132752
##########
table/pos_delete_partitioned_fanout_writer.go:
##########
@@ -0,0 +1,146 @@
+// 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"
+ "iter"
+ "maps"
+ "slices"
+
+ "github.com/apache/arrow-go/v18/arrow"
+ "github.com/apache/arrow-go/v18/arrow/array"
+ "github.com/apache/iceberg-go"
+ "golang.org/x/sync/errgroup"
+)
+
+// positionDeletePartitionedFanoutWriter distributes Arrow position delete records across multiple partitions based on
+// a partition specification, writing data to separate delete files for each partition using
+// a fanout pattern with configurable parallelism.
+type positionDeletePartitionedFanoutWriter struct {
+ partitionContextByFilePath map[string]partitionContext
+ metadata Metadata
+ schema *iceberg.Schema
+ itr iter.Seq2[arrow.RecordBatch, error]
+ writerFactory *writerFactory
+ concurrentDataFileWriter *concurrentDataFileWriter
+}
+
+// newPositionDeletePartitionedFanoutWriter creates a new PartitionedFanoutWriter with the specified
+// partition specification, schema, and record iterator.
+func newPositionDeletePartitionedFanoutWriter(metadata Metadata, concurrentWriter *concurrentDataFileWriter, partitionContextByFilePath map[string]partitionContext, itr iter.Seq2[arrow.RecordBatch, error], writerFactory *writerFactory) *positionDeletePartitionedFanoutWriter {
+ return &positionDeletePartitionedFanoutWriter{
+ partitionContextByFilePath: partitionContextByFilePath,
+ metadata: metadata,
+ schema: iceberg.PositionalDeleteSchema,
+ itr: itr,
+ writerFactory: writerFactory,
+ concurrentDataFileWriter: concurrentWriter,
+ }
+}
+
+// Write writes the Arrow records to the specified location using a fanout pattern with
+// the specified number of workers. The returned iterator yields the data files written
+// by the fanout process.
+func (p *positionDeletePartitionedFanoutWriter) Write(ctx context.Context, workers int) iter.Seq2[iceberg.DataFile, error] {
+ inputRecordsCh := make(chan arrow.RecordBatch, workers)
+ outputDataFilesCh := make(chan iceberg.DataFile, workers)
+
+ fanoutWorkers, ctx := errgroup.WithContext(ctx)
+ startRecordFeeder(ctx, p.itr, fanoutWorkers, inputRecordsCh)
+
+ for range workers {
+ fanoutWorkers.Go(func() error {
+ return p.fanout(ctx, inputRecordsCh, outputDataFilesCh)
+ })
+ }
+
+ return p.yieldDataFiles(fanoutWorkers, outputDataFilesCh)
+}
+
+func (p *positionDeletePartitionedFanoutWriter) fanout(ctx context.Context, inputRecordsCh <-chan arrow.RecordBatch, dataFilesChannel chan<- iceberg.DataFile) error {
+ for {
+ select {
+ case <-ctx.Done():
+ return context.Cause(ctx)
+
+ case record, ok := <-inputRecordsCh:
+ if !ok {
+ return nil
+ }
+
+ err := p.processBatch(ctx, record, dataFilesChannel)
+ if err != nil {
+ return err
+ }
+ }
+ }
+}
+
+func (p *positionDeletePartitionedFanoutWriter) processBatch(ctx context.Context, batch arrow.RecordBatch, dataFilesChannel chan<- iceberg.DataFile) (err error) {
+ defer batch.Release()
+
+ select {
+ case <-ctx.Done():
+ return context.Cause(ctx)
+ default:
+ }
+
+ if batch.NumRows() == 0 {
+ return
+ }
+
+ columns := batch.Columns()
+ filePathArray := columns[0].(*array.String)
+ filePath := filePathArray.ValueStr(0)
+ partitionContext, ok := p.partitionContextByFilePath[filePath]
+ if !ok {
+ return fmt.Errorf("unexpected missing partition context for path %s", filePath)
+ }
+
+ partitionPath, err := p.partitionPath(partitionContext)
+ if err != nil {
+ return err
+ }
+ rollingDataWriter, err := p.writerFactory.getOrCreateRollingDataWriter(ctx, p.concurrentDataFileWriter, partitionPath, partitionContext.partitionData, dataFilesChannel)
+ if err != nil {
+ return err
+ }
+
+ err = rollingDataWriter.Add(batch)
+ if err != nil {
+ return err
+ }
+
+ return nil
+}
+
+func (p *positionDeletePartitionedFanoutWriter) partitionPath(partitionContext partitionContext) (string, error) {
+ data := partitionRecord(slices.Collect(maps.Values(partitionContext.partitionData)))
Review Comment:
I think order matters here: https://github.com/apache/iceberg-go/pull/767
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(table): add support for merge-on-read delete [iceberg-go]
Posted by "zeroshade (via GitHub)" <gi...@apache.org>.
zeroshade commented on code in PR #721:
URL: https://github.com/apache/iceberg-go/pull/721#discussion_r2843340522
##########
table/arrow_scanner.go:
##########
@@ -189,6 +191,54 @@ func processPositionalDeletes(ctx context.Context, deletes set[int64]) recProces
}
}
+// enrichRecordsWithPosDeleteFields enriches a RecordBatch with the columns declared in the PositionalDeleteArrowSchema
+// so that during the pipeline filtering stages that sheds filtered out records, we still have a way to
+// preserve the original position of those records.
+func enrichRecordsWithPosDeleteFields(ctx context.Context, filePath iceberg.DataFile) recProcessFn {
+ nextIdx, mem := int64(0), compute.GetAllocator(ctx)
+
+ return func(inData arrow.RecordBatch) (outData arrow.RecordBatch, err error) {
+ defer inData.Release()
+
+ schema := inData.Schema()
+ fieldIdx := schema.NumFields()
+ filePathField, ok := PositionalDeleteArrowSchema.FieldsByName("file_path")
+ if !ok {
+ panic("position delete schema should have required field 'file_path'")
+ }
+ posField, ok := PositionalDeleteArrowSchema.FieldsByName("pos")
+ if !ok {
+ panic("position delete schema should have required field 'pos'")
+ }
Review Comment:
You could pull these out of the function so we only have to do this once.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(table): add support for merge-on-read delete [iceberg-go]
Posted by "alexandre-normand (via GitHub)" <gi...@apache.org>.
alexandre-normand commented on code in PR #721:
URL: https://github.com/apache/iceberg-go/pull/721#discussion_r2850208052
##########
table/transaction.go:
##########
@@ -1212,6 +1282,117 @@ func (t *Transaction) rewriteSingleFile(ctx context.Context, fs io.IO, originalF
return result, nil
}
+// writePositionDeletesForFiles rewrites data files by preserving only rows that do NOT match the filter
+func (t *Transaction) writePositionDeletesForFiles(ctx context.Context, fs io.IO, updater *snapshotProducer, files []iceberg.DataFile, filter iceberg.BooleanExpression, caseSensitive bool, concurrency int) error {
+ posDeleteRecIter, err := t.makePositionDeleteRecordsForFilter(ctx, fs, files, filter, caseSensitive, concurrency)
+ if err != nil {
+ return err
+ }
+
+ partitionDataPerFile := make(map[string]map[int]any, len(files))
+ for _, df := range files {
+ partitionPath, _ := path.Split(df.FilePath())
+ partitionDataPerFile[partitionPath] = df.Partition()
+ }
+
+ posDeleteFiles := positionDeleteRecordsToDataFiles(ctx, t.tbl.Location(), t.meta, partitionDataPerFile, recordWritingArgs{
Review Comment:
Thanks. Done in 6f24160aeca21c04978ed04ddbaf8deb52733442.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(table): add support for merge-on-read delete [iceberg-go]
Posted by "alexandre-normand (via GitHub)" <gi...@apache.org>.
alexandre-normand commented on code in PR #721:
URL: https://github.com/apache/iceberg-go/pull/721#discussion_r2850367986
##########
table/transaction.go:
##########
@@ -1212,6 +1282,117 @@ func (t *Transaction) rewriteSingleFile(ctx context.Context, fs io.IO, originalF
return result, nil
}
+// writePositionDeletesForFiles rewrites data files by preserving only rows that do NOT match the filter
+func (t *Transaction) writePositionDeletesForFiles(ctx context.Context, fs io.IO, updater *snapshotProducer, files []iceberg.DataFile, filter iceberg.BooleanExpression, caseSensitive bool, concurrency int) error {
+ posDeleteRecIter, err := t.makePositionDeleteRecordsForFilter(ctx, fs, files, filter, caseSensitive, concurrency)
+ if err != nil {
+ return err
+ }
+
+ partitionDataPerFile := make(map[string]map[int]any, len(files))
+ for _, df := range files {
+ partitionPath, _ := path.Split(df.FilePath())
+ partitionDataPerFile[partitionPath] = df.Partition()
Review Comment:
I took the suggested fix but also used the partition spec to get the partition path for the new delete files. Done in ce0c40d983ad70456b50f0a7aa34a94c98f9f076.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(table): add support for merge-on-read delete [iceberg-go]
Posted by "laskoviymishka (via GitHub)" <gi...@apache.org>.
laskoviymishka commented on code in PR #721:
URL: https://github.com/apache/iceberg-go/pull/721#discussion_r2851247883
##########
table/transaction.go:
##########
@@ -915,6 +920,45 @@ func (t *Transaction) performCopyOnWriteDeletion(ctx context.Context, operation
return updater, nil
}
+func (t *Transaction) performMergeOnReadDeletion(ctx context.Context, snapshotProps iceberg.Properties, filter iceberg.BooleanExpression, caseSensitive bool, concurrency int) (*snapshotProducer, error) {
+ fs, err := t.tbl.fsF(ctx)
+ if err != nil {
+ return nil, err
+ }
+
+ if t.meta.NameMapping() == nil {
+ nameMapping := t.meta.CurrentSchema().NameMapping()
+ mappingJson, err := json.Marshal(nameMapping)
+ if err != nil {
+ return nil, err
+ }
+ err = t.SetProperties(iceberg.Properties{DefaultNameMappingKey: string(mappingJson)})
+ if err != nil {
+ return nil, err
+ }
+ }
+
+ commitUUID := uuid.New()
+ updater := t.updateSnapshot(fs, snapshotProps, OpDelete).mergeOverwrite(&commitUUID)
Review Comment:
I'm fine to keep it as is, this may be addressed separately.
I will dive a bit deeper in py/rust, to understand how critical is this.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(table): add support for merge-on-read delete [iceberg-go]
Posted by "alexandre-normand (via GitHub)" <gi...@apache.org>.
alexandre-normand commented on code in PR #721:
URL: https://github.com/apache/iceberg-go/pull/721#discussion_r2835329248
##########
table/arrow_utils.go:
##########
@@ -1358,13 +1361,81 @@ func recordsToDataFiles(ctx context.Context, rootLocation string, meta *Metadata
}
}
- return writeFiles(ctx, rootLocation, args.fs, meta, nil, tasks)
- } else {
- partitionWriter := newPartitionedFanoutWriter(*currentSpec, meta.CurrentSchema(), args.itr)
- rollingDataWriters := NewWriterFactory(rootLocation, args, meta, taskSchema, targetFileSize)
- partitionWriter.writers = &rollingDataWriters
- workers := config.EnvConfig.MaxWorkers
+ return cw.writeFiles(ctx, rootLocation, args.fs, meta, meta.props, nil, tasks)
+ }
+
+ partitionWriter := newPartitionedFanoutWriter(*currentSpec, cw, meta.CurrentSchema(), args.itr)
+ partitionWriter.writers = newWriterFactory(rootLocation, args, meta, taskSchema, targetFileSize)
+ workers := config.EnvConfig.MaxWorkers
+
+ return partitionWriter.Write(ctx, workers)
+}
+
+func positionDeleteRecordsToDataFiles(ctx context.Context, rootLocation string, meta *MetadataBuilder, partitionDataPerFile map[string]map[int]any, args recordWritingArgs) (ret iter.Seq2[iceberg.DataFile, error]) {
+ if args.counter == nil {
+ args.counter = internal.Counter(0)
+ }
+
+ defer func() {
+ if r := recover(); r != nil {
+ var err error
+ switch e := r.(type) {
+ case string:
+ err = fmt.Errorf("error encountered during position delete file writing %s", e)
+ case error:
+ err = fmt.Errorf("error encountered during position delete file writing: %w", e)
+ }
+ ret = func(yield func(iceberg.DataFile, error) bool) {
+ yield(nil, err)
+ }
+ }
+ }()
- return partitionWriter.Write(ctx, workers)
+ if args.writeUUID == nil {
+ u := uuid.Must(uuid.NewRandom())
+ args.writeUUID = &u
}
+
+ targetFileSize := int64(meta.props.GetInt(WriteTargetFileSizeBytesKey,
+ WriteTargetFileSizeBytesDefault))
+
+ currentSpec, err := meta.CurrentSpec()
+ if err != nil || currentSpec == nil {
+ panic(fmt.Errorf("%w: cannot write files without a current spec", err))
Review Comment:
Done in 42755749bc82c055102dd743b2d33d9790cbf5c7.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(table): add support for merge-on-read delete [iceberg-go]
Posted by "alexandre-normand (via GitHub)" <gi...@apache.org>.
alexandre-normand commented on code in PR #721:
URL: https://github.com/apache/iceberg-go/pull/721#discussion_r2843356460
##########
table/arrow_scanner.go:
##########
@@ -189,6 +191,54 @@ func processPositionalDeletes(ctx context.Context, deletes set[int64]) recProces
}
}
+// enrichRecordsWithPosDeleteFields enriches a RecordBatch with the columns declared in the PositionalDeleteArrowSchema
+// so that during the pipeline filtering stages that sheds filtered out records, we still have a way to
+// preserve the original position of those records.
+func enrichRecordsWithPosDeleteFields(ctx context.Context, filePath iceberg.DataFile) recProcessFn {
+ nextIdx, mem := int64(0), compute.GetAllocator(ctx)
+
+ return func(inData arrow.RecordBatch) (outData arrow.RecordBatch, err error) {
+ defer inData.Release()
+
+ schema := inData.Schema()
+ fieldIdx := schema.NumFields()
+ filePathField, ok := PositionalDeleteArrowSchema.FieldsByName("file_path")
+ if !ok {
+ panic("position delete schema should have required field 'file_path'")
+ }
+ posField, ok := PositionalDeleteArrowSchema.FieldsByName("pos")
+ if !ok {
+ panic("position delete schema should have required field 'pos'")
+ }
Review Comment:
ah yes, good catch. I'll do that now.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(table): add support for merge-on-read delete [iceberg-go]
Posted by "alexandre-normand (via GitHub)" <gi...@apache.org>.
alexandre-normand commented on code in PR #721:
URL: https://github.com/apache/iceberg-go/pull/721#discussion_r2801617558
##########
manifest.go:
##########
@@ -2314,5 +2314,5 @@ type ManifestEntry interface {
var PositionalDeleteSchema = NewSchema(0,
NestedField{ID: 2147483546, Type: PrimitiveTypes.String, Name: "file_path", Required: true},
- NestedField{ID: 2147483545, Type: PrimitiveTypes.Int32, Name: "pos", Required: true},
+ NestedField{ID: 2147483545, Type: PrimitiveTypes.Int64, Name: "pos", Required: true},
Review Comment:
The [spec](https://github.com/apache/iceberg-go/pull/721) says that `pos` is a long so I updated this to be `Int64` rather than `Int32`.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(table): add support for merge-on-read delete [iceberg-go]
Posted by "alexandre-normand (via GitHub)" <gi...@apache.org>.
alexandre-normand commented on code in PR #721:
URL: https://github.com/apache/iceberg-go/pull/721#discussion_r2843364413
##########
table/arrow_scanner.go:
##########
@@ -189,6 +191,54 @@ func processPositionalDeletes(ctx context.Context, deletes set[int64]) recProces
}
}
+// enrichRecordsWithPosDeleteFields enriches a RecordBatch with the columns declared in the PositionalDeleteArrowSchema
+// so that during the pipeline filtering stages that sheds filtered out records, we still have a way to
+// preserve the original position of those records.
+func enrichRecordsWithPosDeleteFields(ctx context.Context, filePath iceberg.DataFile) recProcessFn {
+ nextIdx, mem := int64(0), compute.GetAllocator(ctx)
+
+ return func(inData arrow.RecordBatch) (outData arrow.RecordBatch, err error) {
+ defer inData.Release()
+
+ schema := inData.Schema()
+ fieldIdx := schema.NumFields()
+ filePathField, ok := PositionalDeleteArrowSchema.FieldsByName("file_path")
+ if !ok {
+ panic("position delete schema should have required field 'file_path'")
+ }
+ posField, ok := PositionalDeleteArrowSchema.FieldsByName("pos")
+ if !ok {
+ panic("position delete schema should have required field 'pos'")
+ }
Review Comment:
Done in 3d3b8a90876b6fe023e84f0f0e28887ddb353108.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(table): add support for merge-on-read delete [iceberg-go]
Posted by "alexandre-normand (via GitHub)" <gi...@apache.org>.
alexandre-normand commented on PR #721:
URL: https://github.com/apache/iceberg-go/pull/721#issuecomment-3947615577
> this is looking good to me. There's a conflict to resolve in README.md, and I'll wait for @laskoviymishka to give feedback before merging.
Thanks for the nudge. I just rebased to fix the conflict on `README.md`.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(table): add support for merge-on-read delete [iceberg-go]
Posted by "alexandre-normand (via GitHub)" <gi...@apache.org>.
alexandre-normand commented on code in PR #721:
URL: https://github.com/apache/iceberg-go/pull/721#discussion_r2835291759
##########
table/arrow_scanner.go:
##########
@@ -189,6 +192,54 @@ func processPositionalDeletes(ctx context.Context, deletes set[int64]) recProces
}
}
+// enrichRecordsWithPosDeleteFields enriches a RecordBatch with the columns declared in the PositionalDeleteArrowSchema
+// so that during the pipeline filtering stages that sheds filtered out records, we still have a way to
+// preserve the original position of those records.
+func enrichRecordsWithPosDeleteFields(ctx context.Context, filePath iceberg.DataFile) recProcessFn {
Review Comment:
Added those tests in aae5b2f81471cea2fef873e0a2c20c52800512cb except for the missing required fields since that would mean injecting a different `PositionalDeleteArrowSchema` which is not really a good idea since that's a fixed thing.
While I was looking at that one, I decided to make the error on missing required field a panic since that would be a bug/regression introduced by someone changing the constant schema.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(table): add support for merge-on-read delete [iceberg-go]
Posted by "alexandre-normand (via GitHub)" <gi...@apache.org>.
alexandre-normand commented on code in PR #721:
URL: https://github.com/apache/iceberg-go/pull/721#discussion_r2850587186
##########
table/transaction.go:
##########
@@ -915,6 +920,45 @@ func (t *Transaction) performCopyOnWriteDeletion(ctx context.Context, operation
return updater, nil
}
+func (t *Transaction) performMergeOnReadDeletion(ctx context.Context, snapshotProps iceberg.Properties, filter iceberg.BooleanExpression, caseSensitive bool, concurrency int) (*snapshotProducer, error) {
+ fs, err := t.tbl.fsF(ctx)
+ if err != nil {
+ return nil, err
+ }
+
+ if t.meta.NameMapping() == nil {
+ nameMapping := t.meta.CurrentSchema().NameMapping()
+ mappingJson, err := json.Marshal(nameMapping)
+ if err != nil {
+ return nil, err
+ }
+ err = t.SetProperties(iceberg.Properties{DefaultNameMappingKey: string(mappingJson)})
+ if err != nil {
+ return nil, err
+ }
+ }
+
+ commitUUID := uuid.New()
+ updater := t.updateSnapshot(fs, snapshotProps, OpDelete).mergeOverwrite(&commitUUID)
Review Comment:
I'm not familiar with the Java implementation and this would definitely increase the scope/time investment on my end to review and align the implementations. I think it's very likely worthwhile but I'm just not feeling very confident that I can do this quickly enough.
Unless maybe you've got more specific comments/notes that can save me some time and get close to the RowDelta style?
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(table): add support for merge-on-read delete [iceberg-go]
Posted by "ferhatelmas (via GitHub)" <gi...@apache.org>.
ferhatelmas commented on code in PR #721:
URL: https://github.com/apache/iceberg-go/pull/721#discussion_r2874892805
##########
manifest.go:
##########
@@ -605,6 +605,9 @@ func NewManifestReader(file ManifestFile, in io.Reader) (*ManifestReader, error)
if err != nil {
return nil, err
}
+ defer func() {
+ _ = dec.Close()
+ }()
Review Comment:
I think there is a regression here to fix the leak https://github.com/apache/iceberg-go/pull/766
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(table): add support for merge-on-read delete [iceberg-go]
Posted by "alexandre-normand (via GitHub)" <gi...@apache.org>.
alexandre-normand commented on code in PR #721:
URL: https://github.com/apache/iceberg-go/pull/721#discussion_r2850220031
##########
table/arrow_scanner.go:
##########
@@ -473,6 +523,78 @@ func (as *arrowScan) recordsFromTask(ctx context.Context, task internal.Enumerat
return err
}
+func (as *arrowScan) producePosDeletesFromTask(ctx context.Context, task internal.Enumerated[FileScanTask], positionalDeletes positionDeletes, out chan<- enumeratedRecord) (err error) {
+ defer func() {
+ if err != nil {
+ out <- enumeratedRecord{Task: task, Err: err}
+ }
+ }()
+
+ var (
+ rdr internal.FileReader
+ iceSchema *iceberg.Schema
+ colIndices []int
+ filterFunc recProcessFn
+ dropFile bool
+ )
+
+ iceSchema, colIndices, rdr, err = as.prepareToRead(ctx, task.Value.File)
+ if err != nil {
+ return err
+ }
+ defer iceinternal.CheckedClose(rdr, &err)
+
+ fields := append(iceSchema.Fields(), iceberg.PositionalDeleteSchema.Fields()...)
+ enrichedIcebergSchema := iceberg.NewSchema(iceSchema.ID+1, fields...)
+
+ pipeline := make([]recProcessFn, 0, 2)
+ pipeline = append(pipeline, enrichRecordsWithPosDeleteFields(ctx, task.Value.File))
+ if len(positionalDeletes) > 0 {
+ deletes := set[int64]{}
+ for _, chunk := range positionalDeletes {
+ for _, a := range chunk.Chunks() {
+ for _, v := range a.(*array.Int64).Int64Values() {
+ deletes[v] = struct{}{}
+ }
+ }
+ }
+
+ pipeline = append(pipeline, processPositionalDeletes(ctx, deletes))
+ }
+
+ filterFunc, dropFile, err = as.getRecordFilter(ctx, iceSchema)
+ if err != nil {
+ return err
+ }
+
+ // Nothing to delete in a dropped file
+ if dropFile {
+ var emptySchema *arrow.Schema
+ emptySchema, err = SchemaToArrowSchema(as.projectedSchema, nil, false, as.useLargeTypes)
Review Comment:
Good catch. Our future selves will be thankful we didn't let this one slide. Done in 4cbcd7573e785510c46840f0db580f829a24d161.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(table): add support for merge-on-read delete [iceberg-go]
Posted by "laskoviymishka (via GitHub)" <gi...@apache.org>.
laskoviymishka commented on code in PR #721:
URL: https://github.com/apache/iceberg-go/pull/721#discussion_r2814108204
##########
manifest.go:
##########
@@ -1136,7 +1136,7 @@ func (w *ManifestWriter) Close() error {
return w.writer.Close()
}
-func (w *ManifestWriter) ToManifestFile(location string, length int64) (ManifestFile, error) {
+func (w *ManifestWriter) ToManifestFile(location string, length int64, content ManifestContent) (ManifestFile, error) {
Review Comment:
I think table/internal.DataFileStatistics.ToDataFile signature changes are acceptable (it's internal so who care?), but ManifestWriter.ToManifestFile is externally visible and now requires a new content arg.
This is backward incompatible change so 2 questions:
1. Is it really really needed?
2. If it's needed - we need to state it in changelog.
##########
table/arrow_scanner.go:
##########
@@ -189,6 +192,54 @@ func processPositionalDeletes(ctx context.Context, deletes set[int64]) recProces
}
}
+// enrichRecordsWithPosDeleteFields enriches a RecordBatch with the columns declared in the PositionalDeleteArrowSchema
+// so that during the pipeline filtering stages that sheds filtered out records, we still have a way to
+// preserve the original position of those records.
+func enrichRecordsWithPosDeleteFields(ctx context.Context, filePath iceberg.DataFile) recProcessFn {
Review Comment:
Please add focused unit tests for enrichRecordsWithPosDeleteFields:
- monotonically increasing pos across multiple input batches
- exact appended schema/column ordering
- empty batch behavior
- failure mode if required fields are missing from PositionalDeleteArrowSchema.
##########
table/pos_delete_partitioned_fanout_writer.go:
##########
@@ -0,0 +1,116 @@
+// 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"
+ "iter"
+ "path"
+
+ "github.com/apache/arrow-go/v18/arrow"
+ "github.com/apache/arrow-go/v18/arrow/array"
+ "github.com/apache/iceberg-go"
+ "golang.org/x/sync/errgroup"
+)
+
+// positionDeletePartitionedFanoutWriter distributes Arrow position delete records across multiple partitions based on
+// a partition specification, writing data to separate delete files for each partition using
+// a fanout pattern with configurable parallelism.
+type positionDeletePartitionedFanoutWriter struct {
Review Comment:
Could we add unit tests for positionDeletePartitionedFanoutWriter covering:
- mixed file_path values in one batch (assert/handle explicitly)
- missing partitionDataByFilePath entry (error instead of silent nil)
- empty batch behavior and cancellation path.
Also, consider avoiding defer record.Release() inside the loop to prevent delayed releases in long runs.
##########
table/arrow_utils.go:
##########
@@ -1358,13 +1361,81 @@ func recordsToDataFiles(ctx context.Context, rootLocation string, meta *Metadata
}
}
- return writeFiles(ctx, rootLocation, args.fs, meta, nil, tasks)
- } else {
- partitionWriter := newPartitionedFanoutWriter(*currentSpec, meta.CurrentSchema(), args.itr)
- rollingDataWriters := NewWriterFactory(rootLocation, args, meta, taskSchema, targetFileSize)
- partitionWriter.writers = &rollingDataWriters
- workers := config.EnvConfig.MaxWorkers
+ return cw.writeFiles(ctx, rootLocation, args.fs, meta, meta.props, nil, tasks)
+ }
+
+ partitionWriter := newPartitionedFanoutWriter(*currentSpec, cw, meta.CurrentSchema(), args.itr)
+ partitionWriter.writers = newWriterFactory(rootLocation, args, meta, taskSchema, targetFileSize)
+ workers := config.EnvConfig.MaxWorkers
+
+ return partitionWriter.Write(ctx, workers)
+}
+
+func positionDeleteRecordsToDataFiles(ctx context.Context, rootLocation string, meta *MetadataBuilder, partitionDataPerFile map[string]map[int]any, args recordWritingArgs) (ret iter.Seq2[iceberg.DataFile, error]) {
+ if args.counter == nil {
+ args.counter = internal.Counter(0)
+ }
+
+ defer func() {
+ if r := recover(); r != nil {
+ var err error
+ switch e := r.(type) {
+ case string:
+ err = fmt.Errorf("error encountered during position delete file writing %s", e)
+ case error:
+ err = fmt.Errorf("error encountered during position delete file writing: %w", e)
+ }
+ ret = func(yield func(iceberg.DataFile, error) bool) {
+ yield(nil, err)
+ }
+ }
+ }()
- return partitionWriter.Write(ctx, workers)
+ if args.writeUUID == nil {
+ u := uuid.Must(uuid.NewRandom())
+ args.writeUUID = &u
}
+
+ targetFileSize := int64(meta.props.GetInt(WriteTargetFileSizeBytesKey,
+ WriteTargetFileSizeBytesDefault))
+
+ currentSpec, err := meta.CurrentSpec()
+ if err != nil || currentSpec == nil {
+ panic(fmt.Errorf("%w: cannot write files without a current spec", err))
Review Comment:
Can we avoid panic/recover for expected failures here?
CurrentSpec() and schema conversion errors are regular error cases and returning an error iterator directly would be clearer/safer than panicking. Also, the recover block only handles string and error; any other panic type could produce yield(nil, nil), which masks failure.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(table): add support for merge-on-read delete [iceberg-go]
Posted by "alexandre-normand (via GitHub)" <gi...@apache.org>.
alexandre-normand commented on code in PR #721:
URL: https://github.com/apache/iceberg-go/pull/721#discussion_r2834651319
##########
manifest.go:
##########
@@ -1136,7 +1136,7 @@ func (w *ManifestWriter) Close() error {
return w.writer.Close()
}
-func (w *ManifestWriter) ToManifestFile(location string, length int64) (ManifestFile, error) {
+func (w *ManifestWriter) ToManifestFile(location string, length int64, content ManifestContent) (ManifestFile, error) {
Review Comment:
👍 I can make a new vararg parameter to take in options and have a `WithContent` option to override the `content` to keep backward compatibility.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(table): add support for merge-on-read delete [iceberg-go]
Posted by "alexandre-normand (via GitHub)" <gi...@apache.org>.
alexandre-normand commented on code in PR #721:
URL: https://github.com/apache/iceberg-go/pull/721#discussion_r2835330714
##########
manifest.go:
##########
@@ -1136,7 +1136,7 @@ func (w *ManifestWriter) Close() error {
return w.writer.Close()
}
-func (w *ManifestWriter) ToManifestFile(location string, length int64) (ManifestFile, error) {
+func (w *ManifestWriter) ToManifestFile(location string, length int64, content ManifestContent) (ManifestFile, error) {
Review Comment:
Addressed this in dc1ea8be7036c1fb63ef6c0974e8da71126a61c8 (with the vararg options).
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(table): add support for merge-on-read delete [iceberg-go]
Posted by "alexandre-normand (via GitHub)" <gi...@apache.org>.
alexandre-normand commented on code in PR #721:
URL: https://github.com/apache/iceberg-go/pull/721#discussion_r2850530628
##########
table/arrow_utils.go:
##########
@@ -1358,12 +1370,87 @@ func recordsToDataFiles(ctx context.Context, rootLocation string, meta *Metadata
}
}
- return writeFiles(ctx, rootLocation, args.fs, meta, nil, tasks)
- } else {
- rollingDataWriters := NewWriterFactory(rootLocation, args, meta, taskSchema, targetFileSize)
- partitionWriter := newPartitionedFanoutWriter(*currentSpec, meta.CurrentSchema(), args.itr, &rollingDataWriters)
- workers := config.EnvConfig.MaxWorkers
+ return cw.writeFiles(ctx, rootLocation, args.fs, meta, meta.props, nil, tasks)
+ }
+
+ factory := NewWriterFactory(rootLocation, args, meta, taskSchema, targetFileSize)
+ partitionWriter := newPartitionedFanoutWriter(*currentSpec, cw, meta.CurrentSchema(), args.itr, &factory)
+ workers := config.EnvConfig.MaxWorkers
- return partitionWriter.Write(ctx, workers)
+ return partitionWriter.Write(ctx, workers)
+}
+
+func positionDeleteRecordsToDataFiles(ctx context.Context, rootLocation string, meta *MetadataBuilder, partitionDataPerFile map[string]map[int]any, args recordWritingArgs) (ret iter.Seq2[iceberg.DataFile, error]) {
+ if args.counter == nil {
+ args.counter = internal.Counter(0)
}
+
+ defer func() {
+ if r := recover(); r != nil {
+ var err error
+ switch e := r.(type) {
+ case error:
+ err = fmt.Errorf("error encountered during position delete file writing: %w", e)
+ default:
+ err = fmt.Errorf("error encountered during position delete file writing: %v", e)
+ }
+ ret = func(yield func(iceberg.DataFile, error) bool) {
+ yield(nil, err)
+ }
+ }
+ }()
+
+ if args.writeUUID == nil {
+ u := uuid.Must(uuid.NewRandom())
+ args.writeUUID = &u
+ }
+
+ targetFileSize := int64(meta.props.GetInt(WriteTargetFileSizeBytesKey,
+ WriteTargetFileSizeBytesDefault))
+
+ currentSpec, err := meta.CurrentSpec()
Review Comment:
I tried to address this but let me know if that's correct. I extended the mapping per data file to include the spec id in addition to the partition data we were already tracking that way. That's passed down now and the position delete fanout writer uses the partition spec from the partition context.
That's done in 83c50c6f542bf085b0d54b271d9f2720153efc55. The one case I was unsure about is the one where the table's latest metadata is unpartitioned. In that case, we were short-circuiting early and skipping the fanout writer. That case remains but I'm not sure what the ramifications are.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(table): add support for merge-on-read delete [iceberg-go]
Posted by "alexandre-normand (via GitHub)" <gi...@apache.org>.
alexandre-normand commented on code in PR #721:
URL: https://github.com/apache/iceberg-go/pull/721#discussion_r2835811686
##########
table/pos_delete_partitioned_fanout_writer.go:
##########
@@ -0,0 +1,116 @@
+// 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"
+ "iter"
+ "path"
+
+ "github.com/apache/arrow-go/v18/arrow"
+ "github.com/apache/arrow-go/v18/arrow/array"
+ "github.com/apache/iceberg-go"
+ "golang.org/x/sync/errgroup"
+)
+
+// positionDeletePartitionedFanoutWriter distributes Arrow position delete records across multiple partitions based on
+// a partition specification, writing data to separate delete files for each partition using
+// a fanout pattern with configurable parallelism.
+type positionDeletePartitionedFanoutWriter struct {
Review Comment:
I did most of this in df4cef7e2cc850bf000b7bc8b14f9dcdc9ba3684:
* Extracted the processing of one record batch in processBatch so that defer record.Release() could be scoped to the processing of a single batch.
* Note that this is a common pattern in many places and I did not expand the scope beyond the new fanout writer. It might be nice to follow-up at some point to take the same approach on other defer released executed in loops.
* I added tests for processBatch that include the missing partitionDataByFilePath entry, empty batch, context cancellation, simple success as well as mixed file_path values.
* Note that for the mixed file_path case, I didn't want to pre-process the batch to validate that all paths were the same since the calling code already guarantees that. So I left the test case to illustrate the "limitation" as one that's intentional (at least for now).
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(table): add support for merge-on-read delete [iceberg-go]
Posted by "alexandre-normand (via GitHub)" <gi...@apache.org>.
alexandre-normand commented on code in PR #721:
URL: https://github.com/apache/iceberg-go/pull/721#discussion_r2850389817
##########
table/snapshot_producers.go:
##########
@@ -628,11 +603,55 @@ func (sp *snapshotProducer) manifests() (_ []iceberg.ManifestFile, err error) {
return nil, err
}
- manifests := slices.Concat(results[0], results[1], results[2])
+ manifests := slices.Concat(addedManifests, positionDeleteManifests, deletedFilesManifests, existingManifests)
return sp.processManifests(manifests)
}
+func (sp *snapshotProducer) manifestProducer(content iceberg.ManifestContent, files []iceberg.DataFile, output *[]iceberg.ManifestFile) func() (err error) {
+ return func() (err error) {
+ out, path, err := sp.newManifestOutput()
+ if err != nil {
+ return err
+ }
+ defer internal.CheckedClose(out, &err)
+
+ counter := &internal.CountingWriter{W: out}
+ currentSpec, err := sp.txn.meta.CurrentSpec()
+ if err != nil || currentSpec == nil {
+ return fmt.Errorf("could not get current partition spec: %w", err)
+ }
+ wr, err := iceberg.NewManifestWriter(sp.txn.meta.formatVersion, counter,
+ *currentSpec, sp.txn.meta.CurrentSchema(),
+ sp.snapshotID)
+ if err != nil {
+ return err
+ }
+ defer internal.CheckedClose(wr, &err)
+
+ for _, df := range files {
+ err := wr.Add(iceberg.NewManifestEntry(iceberg.EntryStatusADDED, &sp.snapshotID,
+ nil, nil, df))
+ if err != nil {
+ return err
+ }
+ }
+
+ // close the writer to force a flush and ensure counter.Count is accurate
+ if err := wr.Close(); err != nil {
+ return err
+ }
+
+ mf, err := wr.ToManifestFile(path, counter.Count, iceberg.WithManifestFileContent(content))
+ if err != nil {
+ return err
+ }
+ *output = []iceberg.ManifestFile{mf}
+
+ return nil
+ }
+}
+
func (sp *snapshotProducer) summary(props iceberg.Properties) (Summary, error) {
Review Comment:
That's a good catch and easy enough to fix on its own but, adding a test case for that to validate the snapshot statistics, I encountered a bug where scanning tries to parse of of the positional delete files as a data file which causes an error. I have the summary showing the positional delete count but that same test does a scan to check the number of rows post-delete and that fails so I have to address 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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(table): add support for merge-on-read delete [iceberg-go]
Posted by "alexandre-normand (via GitHub)" <gi...@apache.org>.
alexandre-normand commented on code in PR #721:
URL: https://github.com/apache/iceberg-go/pull/721#discussion_r2850030555
##########
table/transaction_test.go:
##########
@@ -536,6 +536,120 @@ func (s *SparkIntegrationTestSuite) TestDeleteInsensitive() {
+----------+---------+---+`)
}
+func (s *SparkIntegrationTestSuite) TestDeleteMergeOnReadUnpartitioned() {
+ icebergSchema := iceberg.NewSchema(0,
+ iceberg.NestedField{ID: 1, Name: "first_name", Type: iceberg.PrimitiveTypes.String},
+ iceberg.NestedField{ID: 2, Name: "last_name", Type: iceberg.PrimitiveTypes.String},
+ iceberg.NestedField{ID: 3, Name: "age", Type: iceberg.PrimitiveTypes.Int32},
+ )
+
+ tbl, err := s.cat.CreateTable(s.ctx, catalog.ToIdentifier("default", "go_test_merge_on_read_delete"), icebergSchema,
+ catalog.WithProperties(
+ map[string]string{
+ table.WriteDeleteModeKey: table.WriteModeMergeOnRead,
+ },
+ ),
+ )
+ s.Require().NoError(err)
+
+ arrowSchema, err := table.SchemaToArrowSchema(icebergSchema, nil, true, false)
+ s.Require().NoError(err)
+
+ initialTable, err := array.TableFromJSON(memory.DefaultAllocator, arrowSchema, []string{
+ `[
+ {"first_name": "alan", "last_name": "gopher", "age": 7},
+ {"first_name": "steve", "last_name": "gopher", "age": 5},
+ {"first_name": "dead", "last_name": "gopher", "age": 97}
+ ]`,
+ })
+ s.Require().NoError(err)
+ defer initialTable.Release()
+
+ tx := tbl.NewTransaction()
+ err = tx.AppendTable(s.ctx, initialTable, 3, nil)
+ s.Require().NoError(err)
+ tbl, err = tx.Commit(s.ctx)
+ s.Require().NoError(err)
+
+ // Delete the dead gopher and confirm that alan and steve are still present
+ filter := iceberg.EqualTo(iceberg.Reference("first_name"), "dead")
+ tx = tbl.NewTransaction()
+ err = tx.Delete(s.ctx, filter, nil)
+ s.Require().NoError(err)
+ _, err = tx.Commit(s.ctx)
+ s.Require().NoError(err)
+
+ output, err := recipe.ExecuteSpark(s.T(), "./validation.py", "--sql", "SELECT * FROM default.go_test_merge_on_read_delete ORDER BY age")
+ s.Require().NoError(err)
+ s.Require().Contains(output, `|first_name|last_name|age|
++----------+---------+---+
+|steve |gopher |5 |
+|alan |gopher |7 |
++----------+---------+---+`)
+}
+
+func (s *SparkIntegrationTestSuite) TestDeleteMergeOnReadPartitioned() {
+ icebergSchema := iceberg.NewSchema(0,
+ iceberg.NestedField{ID: 1, Name: "first_name", Type: iceberg.PrimitiveTypes.String},
+ iceberg.NestedField{ID: 2, Name: "last_name", Type: iceberg.PrimitiveTypes.String},
+ iceberg.NestedField{ID: 3, Name: "age", Type: iceberg.PrimitiveTypes.Int32},
+ )
+
+ spec := iceberg.NewPartitionSpec(iceberg.PartitionField{
+ SourceID: 3,
+ Name: "age_bucket",
+ Transform: iceberg.BucketTransform{
+ NumBuckets: 2,
+ },
+ })
+ tbl, err := s.cat.CreateTable(s.ctx, catalog.ToIdentifier("default", "go_test_merge_on_read_delete_partitioned"), icebergSchema,
+ catalog.WithProperties(
+ map[string]string{
+ table.WriteDeleteModeKey: table.WriteModeMergeOnRead,
+ },
+ ),
+ catalog.WithPartitionSpec(&spec),
+ )
+ s.Require().NoError(err)
+
+ arrowSchema, err := table.SchemaToArrowSchema(icebergSchema, nil, true, false)
+ s.Require().NoError(err)
+
+ initialTable, err := array.TableFromJSON(memory.DefaultAllocator, arrowSchema, []string{
+ `[
+ {"first_name": "alan", "last_name": "gopher", "age": 7},
+ {"first_name": "steve", "last_name": "gopher", "age": 5},
+ {"first_name": "dead", "last_name": "gopher", "age": 97}
Review Comment:
Good catch. It wasn't actually panicking for some reason but it was missing the coma which should have been there. Fixed in 3ca8a68abcd6e3c19c67b3eaaec19a82a335105b.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(table): add support for merge-on-read delete [iceberg-go]
Posted by "alexandre-normand (via GitHub)" <gi...@apache.org>.
alexandre-normand commented on code in PR #721:
URL: https://github.com/apache/iceberg-go/pull/721#discussion_r2834641853
##########
table/arrow_utils.go:
##########
@@ -1358,13 +1361,81 @@ func recordsToDataFiles(ctx context.Context, rootLocation string, meta *Metadata
}
}
- return writeFiles(ctx, rootLocation, args.fs, meta, nil, tasks)
- } else {
- partitionWriter := newPartitionedFanoutWriter(*currentSpec, meta.CurrentSchema(), args.itr)
- rollingDataWriters := NewWriterFactory(rootLocation, args, meta, taskSchema, targetFileSize)
- partitionWriter.writers = &rollingDataWriters
- workers := config.EnvConfig.MaxWorkers
+ return cw.writeFiles(ctx, rootLocation, args.fs, meta, meta.props, nil, tasks)
+ }
+
+ partitionWriter := newPartitionedFanoutWriter(*currentSpec, cw, meta.CurrentSchema(), args.itr)
+ partitionWriter.writers = newWriterFactory(rootLocation, args, meta, taskSchema, targetFileSize)
+ workers := config.EnvConfig.MaxWorkers
+
+ return partitionWriter.Write(ctx, workers)
+}
+
+func positionDeleteRecordsToDataFiles(ctx context.Context, rootLocation string, meta *MetadataBuilder, partitionDataPerFile map[string]map[int]any, args recordWritingArgs) (ret iter.Seq2[iceberg.DataFile, error]) {
+ if args.counter == nil {
+ args.counter = internal.Counter(0)
+ }
+
+ defer func() {
+ if r := recover(); r != nil {
+ var err error
+ switch e := r.(type) {
+ case string:
+ err = fmt.Errorf("error encountered during position delete file writing %s", e)
+ case error:
+ err = fmt.Errorf("error encountered during position delete file writing: %w", e)
+ }
+ ret = func(yield func(iceberg.DataFile, error) bool) {
+ yield(nil, err)
+ }
+ }
+ }()
- return partitionWriter.Write(ctx, workers)
+ if args.writeUUID == nil {
+ u := uuid.Must(uuid.NewRandom())
+ args.writeUUID = &u
}
+
+ targetFileSize := int64(meta.props.GetInt(WriteTargetFileSizeBytesKey,
+ WriteTargetFileSizeBytesDefault))
+
+ currentSpec, err := meta.CurrentSpec()
+ if err != nil || currentSpec == nil {
+ panic(fmt.Errorf("%w: cannot write files without a current spec", err))
Review Comment:
This makes sense to me but I want to point out that part of this was a copy of [recordsToDataFiles](https://github.com/apache/iceberg-go/blob/main/table/arrow_utils.go#L1300) which does that. @zeroshade : If you agree with this, should I go ahead and also update `recordsToDataFiles` to take the same approach?
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(table): add support for merge-on-read delete [iceberg-go]
Posted by "alexandre-normand (via GitHub)" <gi...@apache.org>.
alexandre-normand commented on code in PR #721:
URL: https://github.com/apache/iceberg-go/pull/721#discussion_r2850530628
##########
table/arrow_utils.go:
##########
@@ -1358,12 +1370,87 @@ func recordsToDataFiles(ctx context.Context, rootLocation string, meta *Metadata
}
}
- return writeFiles(ctx, rootLocation, args.fs, meta, nil, tasks)
- } else {
- rollingDataWriters := NewWriterFactory(rootLocation, args, meta, taskSchema, targetFileSize)
- partitionWriter := newPartitionedFanoutWriter(*currentSpec, meta.CurrentSchema(), args.itr, &rollingDataWriters)
- workers := config.EnvConfig.MaxWorkers
+ return cw.writeFiles(ctx, rootLocation, args.fs, meta, meta.props, nil, tasks)
+ }
+
+ factory := NewWriterFactory(rootLocation, args, meta, taskSchema, targetFileSize)
+ partitionWriter := newPartitionedFanoutWriter(*currentSpec, cw, meta.CurrentSchema(), args.itr, &factory)
+ workers := config.EnvConfig.MaxWorkers
- return partitionWriter.Write(ctx, workers)
+ return partitionWriter.Write(ctx, workers)
+}
+
+func positionDeleteRecordsToDataFiles(ctx context.Context, rootLocation string, meta *MetadataBuilder, partitionDataPerFile map[string]map[int]any, args recordWritingArgs) (ret iter.Seq2[iceberg.DataFile, error]) {
+ if args.counter == nil {
+ args.counter = internal.Counter(0)
}
+
+ defer func() {
+ if r := recover(); r != nil {
+ var err error
+ switch e := r.(type) {
+ case error:
+ err = fmt.Errorf("error encountered during position delete file writing: %w", e)
+ default:
+ err = fmt.Errorf("error encountered during position delete file writing: %v", e)
+ }
+ ret = func(yield func(iceberg.DataFile, error) bool) {
+ yield(nil, err)
+ }
+ }
+ }()
+
+ if args.writeUUID == nil {
+ u := uuid.Must(uuid.NewRandom())
+ args.writeUUID = &u
+ }
+
+ targetFileSize := int64(meta.props.GetInt(WriteTargetFileSizeBytesKey,
+ WriteTargetFileSizeBytesDefault))
+
+ currentSpec, err := meta.CurrentSpec()
Review Comment:
I tried to address this but let me know if that's correct. I extended the mapping per data file to include the spec id in addition to the partition data we were already tracking that way. That's passed down now and the position delete fanout writer uses the partition spec from the partition context.
That's done in cf4aaa879da73202c5740310f62f74810114fdbd. The one case I was unsure about is the one where the table's latest metadata is unpartitioned. In that case, we were short-circuiting early and skipping the fanout writer. That case remains but I'm not sure what the ramifications are.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org