apache / apache/iceberg

Spark: rewrite_table_path fails on second run with FileAlreadyExistsException for position delete files

Open
#16,418 4 comments 0 reactions 0 assignees View on GitHub
bug
Dominant language
Java
Stars
9.2k
Forks
3.5k
Avg merge
2d 16h
Merged PRs (30d)
129

Description

### Apache Iceberg version

1.10.1

### Query engine

Spark

### Please describe the bug 🐞

I think RewriteTablePathSparkAction should be safely executed repeatedly to support incremental data migration.

When I execute rewrite_table_stath for the second time on a MOR table with a posdelete file, I get the following error
```
2026-05-19 15:56:23,417 [ERROR] [Executor task launch worker for task 0.0 in stage 57.0 (TID 69)] Executor: Exception in task 0.0 in stage 57.0 (TID 69)
java.io.UncheckedIOException: Failed to create Parquet file
at org.apache.iceberg.parquet.ParquetWriter.ensureWriterInitialized(ParquetWriter.java:124)
at org.apache.iceberg.parquet.ParquetWriter.flushRowGroup(ParquetWriter.java:214)
at org.apache.iceberg.parquet.ParquetWriter.close(ParquetWriter.java:258)
at org.apache.iceberg.deletes.PositionDeleteWriter.close(PositionDeleteWriter.java:92)
at org.apache.iceberg.RewriteTablePathUtil.rewritePositionDeleteFile(RewriteTablePathUtil.java:633)
at org.apache.iceberg.spark.actions.RewriteTablePathSparkAction.lambda$rewritePositionDelete$a4760a1f$1(RewriteTablePathSparkAction.java:673)
at org.apache.spark.sql.Dataset.$anonfun$foreach$2(Dataset.scala:3507)
at org.apache.spark.sql.Dataset.$anonfun$foreach$2$adapted(Dataset.scala:3507)
at scala.collection.Iterator.foreach(Iterator.scala:943)
at scala.collection.Iterator.foreach$(Iterator.scala:943)
at scala.collection.AbstractIterator.foreach(Iterator.scala:1431)
at org.apache.spark.rdd.RDD.$anonfun$foreach$2(RDD.scala:1031)
at org.apache.spark.rdd.RDD.$anonfun$foreach$2$adapted(RDD.scala:1031)
at org.apache.spark.SparkContext.$anonfun$runJob$5(SparkContext.scala:2433)
at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:93)
at org.apache.spark.TaskContext.runTaskWithListeners(TaskContext.scala:166)
at org.apache.spark.scheduler.Task.run(Task.scala:141)
at org.apache.spark.executor.Executor$TaskRunner.$anonfun$run$4(Executor.scala:621)
at org.apache.spark.util.SparkErrorUtils.tryWithSafeFinally(SparkErrorUtils.scala:64)
at org.apache.spark.util.SparkErrorUtils.tryWithSafeFinally$(SparkErrorUtils.scala:61)
at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:94)
at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:624)
at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
at java.base/java.lang.Thread.run(Thread.java:829)
Caused by: org.apache.hadoop.fs.FileAlreadyExistsException: /usr/hive/warehouse/staging/orders/data/region=cn/dt=2024-01-01/00059-1-8c49b5a8-f44f-415c-9836-fb8d832aaf5c-00001-deletes.parquet for client 10.xx.x.xx already exists
at org.apache.hadoop.hdfs.server.namenode.FSDirWriteFileOp.startFile(FSDirWriteFileOp.java:389)
at org.apache.hadoop.hdfs.server.namenode.FSNamesystem.startFileInt(FSNamesystem.java:2583)
at org.apache.hadoop.hdfs.server.namenode.FSNamesystem.startFile(FSNamesystem.java:2479)
at org.apache.hadoop.hdfs.server.namenode.NameNodeRpcServer.create(NameNodeRpcServer.java:841)
at org.apache.hadoop.hdfs.protocolPB.ClientNamenodeProtocolServerSideTranslatorPB.create(ClientNamenodeProtocolServerSideTranslatorPB.java:326)
at org.apache.hadoop.hdfs.protocol.proto.ClientNamenodeProtocolProtos$ClientNamenodeProtocol$2.callBlockingMethod(ClientNamenodeProtocolProtos.java)
at org.apache.hadoop.ipc.ProtobufRpcEngine$Server$ProtoBufRpcInvoker.call(ProtobufRpcEngine.java:529)
at org.apache.hadoop.ipc.RPC$Server.call(RPC.java:1073)
at org.apache.hadoop.ipc.Server$RpcCall.run(Server.java:1039)
at org.apache.hadoop.ipc.Server$RpcCall.run(Server.java:963)
at java.security.AccessController.doPrivileged(Native Method)
at javax.security.auth.Subject.doAs(Subject.java:422)
at org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:2065)
at org.apache.hadoop.ipc.Server$Handler.run(Server.java:3047)
```

Every other artifact written to the staging directory uses overwrite semantics, except for PositionDeleteFile and DeletionVectorFile.

|Artifact|Write call|Mode|
|-|-|-|
|vN-*.metadata.json|TableMetadataParser.overwrite(...)|overwrite|
|snap-*.avro |ManifestLists.write (Avro default)|overwrite|
|*.avro|ManifestFiles.write (Avro default)|overwrite|
|file-list|outputFile.createOrOverwrite()|overwrite|
|position delete *.parquet/avro/orc|Parquet.writeDeletes(outputFile)…buildPositionWriter()|CREATE|
|deletion vector *.puffin|PuffinWriter.write(...).build()|CREATE|

How to reproduce
```
CREATE TABLE spark_catalog.sales.orders (
id BIGINT,
user_id BIGINT,
amount DOUBLE,
status STRING,
region STRING,
dt STRING,
updated_at TIMESTAMP
) USING iceberg
PARTITIONED BY (region, dt)
TBLPROPERTIES (
'format-version' = '2',
'write.delete.mode' = 'merge-on-read',
'write.update.mode' = 'merge-on-read',
'write.merge.mode' = 'merge-on-read'
);

INSERT INTO spark_catalog.sales.orders VALUES
(1, 1001, 100.5, 'paid', 'cn', '2024-01-01', timestamp'2024-01-01 10:00:00'),
(2, 1002, 200.0, 'paid', 'cn', '2024-01-01', timestamp'2024-01-01 11:00:00'),
(3, 1003, 300.0, 'pending', 'cn', '2024-01-02', timestamp'2024-01-02 09:00:00'),
(4, 1004, 400.0, 'paid', 'us', '2024-01-01', timestamp'2024-01-01 12:00:00'),
(5, 1005, 500.0, 'cancel', 'us', '2024-01-02', timestamp'2024-01-02 13:00:00'),
(6, 1006, 600.0, 'paid', 'eu', '2024-01-01', timestamp'2024-01-01 14:00:00');
INSERT INTO spark_catalog.sales.orders VALUES
(7, 1007, 700.0, 'paid', 'cn', '2024-01-03', timestamp'2024-01-03 10:00:00'),
(8, 1008, 800.0, 'paid', 'us', '2024-01-03', timestamp'2024-01-03 11:00:00'),
(9, 1009, 900.0, 'pending','jp', '2024-01-03', timestamp'2024-01-03 12:00:00'),
(10, 1010, 150.0, 'paid', 'cn', '2024-01-02', timestamp'2024-01-02 15:00:00');

DELETE FROM spark_catalog.sales.orders WHERE id = 5;
DELETE FROM spark_catalog.sales.orders WHERE status = 'pending';

UPDATE spark_catalog.sales.orders SET amount = amount * 1.1 WHERE region = 'cn';
```

```sql
-- Execute twice
CALL spark_catalog.system.rewrite_table_path(
table => 'sales.orders',
source_prefix => 'hdfs://xxx.xx.xx.xx:xx/usr/hive/warehouse/sales.db/orders',
target_prefix => 'hdfs://xxxx/iceberg-warehouse/sales/orders',
staging_location => 'hdfs://xxx.xx.xx.xx:xx/usr/hive/warehouse/staging/orders'
);
```

### Willingness to contribute

- [x] I can contribute a fix for this bug independently
- [ ] I would be willing to contribute a fix for this bug with guidance from the Iceberg community
- [ ] I cannot contribute a fix for this bug at this time

Contributor guide

Open the contributing guide

Research direction

Begin at RewriteTablePathSparkAction.rewritePositionDelete and RewriteTablePathUtil.rewritePositionDeleteFile, then reproduce the issue by running the documented rewrite_table_path call twice on a merge-on-read table with position deletes. Compare the position-delete and deletion-vector write paths with the overwrite-based artifacts; done means the second run completes without FileAlreadyExistsException.

Written by the indexing model from the issue text.

Assessment

Tech stack
java, spark
Domain
data-engineering, databases
Issue type
Bug
Difficulty
3/5
Estimated time
1-2 days
Activity status
Quiet
Clarity
Mostly clear
Newbie friendliness
62/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.