alexandre-normand commented on code in PR #718:
URL: https://github.com/apache/iceberg-go/pull/718#discussion_r2790901198
##########
table/transaction.go:
##########
@@ -595,55 +595,136 @@ func (t *Transaction) Overwrite(ctx context.Context, rdr
array.RecordReader, sna
apply(&overwrite)
}
+ updater, err := t.performCopyOnWriteDeletion(ctx, snapshotProps,
overwrite.filter, overwrite.caseSensitive, overwrite.concurrency)
+ if err != nil {
+ return err
+ }
+
fs, err := t.tbl.fsF(ctx)
if err != nil {
return err
}
+ itr := recordsToDataFiles(ctx, t.tbl.Location(), t.meta,
recordWritingArgs{
+ sc: rdr.Schema(),
+ itr: array.IterFromReader(rdr),
+ fs: fs.(io.WriteFileIO),
+ writeUUID: &updater.commitUuid,
+ })
+
+ for df, err := range itr {
+ if err != nil {
+ return err
+ }
+ updater.appendDataFile(df)
+ }
+
+ updates, reqs, err := updater.commit()
+ if err != nil {
+ return err
+ }
+
+ return t.apply(updates, reqs)
+}
+
+func (t *Transaction) performCopyOnWriteDeletion(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 err
+ return nil, err
}
err = t.SetProperties(iceberg.Properties{DefaultNameMappingKey:
string(mappingJson)})
if err != nil {
- return err
+ return nil, err
}
}
commitUUID := uuid.New()
updater := t.updateSnapshot(fs,
snapshotProps).mergeOverwrite(&commitUUID)
- filesToDelete, filesToRewrite, err := t.classifyFilesForOverwrite(ctx,
fs, overwrite.filter, overwrite.caseSensitive, overwrite.concurrency)
+ filesToDelete, filesToRewrite, err := t.classifyFilesForOverwrite(ctx,
fs, filter, caseSensitive, concurrency)
if err != nil {
- return err
+ return nil, err
}
for _, df := range filesToDelete {
updater.deleteDataFile(df)
}
if len(filesToRewrite) > 0 {
- if err := t.rewriteFilesWithFilter(ctx, fs, updater,
filesToRewrite, overwrite.filter, overwrite.caseSensitive,
overwrite.concurrency); err != nil {
- return err
+ if err := t.rewriteFilesWithFilter(ctx, fs, updater,
filesToRewrite, filter, caseSensitive, concurrency); err != nil {
+ return nil, err
}
}
- itr := recordsToDataFiles(ctx, t.tbl.Location(), t.meta,
recordWritingArgs{
- sc: rdr.Schema(),
- itr: array.IterFromReader(rdr),
- fs: fs.(io.WriteFileIO),
- writeUUID: &updater.commitUuid,
- })
+ return updater, nil
+}
- for df, err := range itr {
- if err != nil {
- return err
+type DeleteOption func(deleteOp *deleteOperation)
+
+type deleteOperation struct {
+ caseSensitive bool
+ concurrency int
+}
+
+// WithDeleteConcurrency overwrites the default concurrency for delete
operations.
+// Default: runtime.GOMAXPROCS(0)
+func WithDeleteConcurrency(concurrency int) OverwriteOption {
Review Comment:
🤦 . Fixed in 6ff00fb003dce8a9ac68b48b1a444f580f2ca668.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
