Rewrite iceberg small files with flink succeeds but no snapshot is generated (V2 - upsert model)

[SHuixo]

Apache Iceberg version

0.14.1

Query engine

Flink

Please describe the bug 🐞

flink: 1.13.5
iceberg: 1.13.2 / 1.14.1

When using Rewrite files action API rewriteDataFiles(), the new compressed file is generated without a corresponding manifest file, I tried iceberg versions 1.13.2 and 1.14.1 which has a similar problem under the iceberg-catalog of Hive and Hadoop.

The Iceberg Maven dependent, table structure and code to compress the file using the Java API is as follows：

      <dependency>
            <groupId>org.apache.iceberg</groupId>
            <artifactId>iceberg-flink-runtime-1.13</artifactId>
<!--            <version>0.13.2</version>-->
            <version>0.14.1</version>
        </dependency>

name: iceberg_hive_catalog
type: iceberg
catalog-type: hive
uri: thrift://xxxxx:9083
clients: 5
property-version: 1
warehouse: hdfs://nameservice1/user/hive/warehouse/

create table iceberg_hive_catalog.dhome_db.ods_d_base_inf_229_iceberg (
`did` string,
`name` string,
`address` string,
`did_seq`  string,
PRIMARY KEY (did_seq) NOT ENFORCED
) with (
 'format-version'='2',
 'write.upsert.enabled'='true',
 'write.metadata.delete-after-commit.enabled'='true',
 'write.metadata.previous-versions-max'='5',
 'flink.rewrite.enable' = 'true',
 'flink.rewrite.parallelism' = '5',
 'flink.rewrite.target-file-size-bytes' = '536870912',
 'flink.rewrite.max-files-count' = '5'
);

final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
TableIdentifier identifier = TableIdentifier.of("dhome_db", "ods_d_base_inf_229_iceberg");
TableLoader tableLoader = TableLoader.fromCatalog(hive_iceberg, identifier);
tableLoader.open();
Table table_iceberg = tableLoader.loadTable();

Actions.forTable(env, table_iceberg)
		.rewriteDataFiles()
		.maxParallelism(5)
		.targetSizeInBytes(128*1024*1024)
		.execute();

The results:

If there is anything wrong with the question, please correct it, thank you.

[luoyuxia]

Seems the files has been compressed, but it fails to to generate snapshot. Maybe some exception happen when commit.
Is there any error or warning in the log?

[SHuixo]

Think you @luoyuxia for your replay.

Later, I tried again a few times and found that when the cumulative storage of iceberg data small files was relatively small, the flink 1.13.5 compressed file ran normally and could generate snapshot files.

However, when the file volume accumulates a lot, it takes a long time to rewrite the data each time, and it is easy to cause OOM exceptions, here are my attempts at Flink 1.13.5 / 1.15.2, iceberg 1.14.1 and the log logs generated by the task.

The following figure shows that the compression task has been in the Map stage：

OOM exception information that occurs when the compression task occurs：

flink 1.13.5:
error-flink-1.13.5.log

flink 1.15.2:
error-flink-1.15.2.log

Here I want to ask, if the data is continuously written to iceberg, the problem of data compression OOM is inevitable, and the compression time will become longer and longer.

I see that there are API methods appendsBetween() / appendsAfter() related to incremental compression in the source code, does this mean that incremental compression can be used to replace the repeated compression process of full data in the future?

thanks.

[luoyuxia]

the repeated compression process of full data

Do you mean each compaction will compact all data? The compaction won't be repeated. I mean during one compaction, some files are compacted to one file. In next compaction, it won't compact these files again. I don't think it's a repeated compression process. From my side, it's still a incremental process.

Back to your problem, actuall, there's a pr #2680 to use disk to avoid OOM. And I think it'll also fix your problem.
I think you can try to
1: increase the memory of TM.
2: do compaction more frequently , so that every compaction won't compact much data, which may well relieve the OOM.

[SHuixo]

the repeated compression process of full data

Do you mean each compaction will compact all data? The compaction won't be repeated. I mean during one compaction, some files are compacted to one file. In next compaction, it won't compact these files again. I don't think it's a repeated compression process. From my side, it's still a incremental process.

Back to your problem, actuall, there's a pr #2680 to use disk to avoid OOM. And I think it'll also fix your problem. I think you can try to 1: increase the memory of TM. 2: do compaction more frequently , so that every compaction won't compact much data, which may well relieve the OOM.

The new features of rocksdb mentioned in the PR #2680 have not found the relevant configuration items in the latest 1.0.0 version? Do you have to wait for a subsequent version to be added?

In compressing data it is indeed possible to solve the OOM problem by increasing the memory of TaskManger and increasing the compression frequency of data.

However, when processing the Flink cdc real-time stream written to iceberg's table, the checkpoint interval is set to 5 minutes and the compressor interval is 30 minutes, and the commit exception problem will be encountered when the data compression task is performed.
The subsequent restarted compression task will remain in the Map phase until the task processing timeout exception and exited.

commit exception:

2022-11-11 17:06:57,583 WARN  org.apache.iceberg.actions.BaseRewriteDataFilesAction        [] - Failed to commit rewrite, cleaning up rewritten files
org.apache.iceberg.exceptions.ValidationException: Cannot commit, found new delete for replaced data file: GenericDataFile{content=data, file_path=hdfs://nameservice1/user/hive/warehouse/dhome_db.db/ods_d_base_inf_229_iceberg/data/00001-0-70f034b7-9725-4d90-b1ad-95907d30ed19-00001.parquet, file_format=PARQUET, spec_id=0, partition=PartitionData{}, record_count=951380, file_size_in_bytes=38224976, column_sizes=null, value_counts=org.apache.iceberg.util.SerializableMap@187f476, null_value_counts=org.apache.iceberg.util.SerializableMap@17a, nan_value_counts=org.apache.iceberg.util.SerializableMap@0, lower_bounds=org.apache.iceberg.SerializableByteBufferMap@f05dccea, upper_bounds=org.apache.iceberg.SerializableByteBufferMap@ffcfbef8, key_metadata=null, split_offsets=[4], equality_ids=null, sort_order_id=null}
	at org.apache.iceberg.exceptions.ValidationException.check(ValidationException.java:50) ~[DevXXXX-0.14.1-1.13.5.jar:?]
	at org.apache.iceberg.MergingSnapshotProducer.validateNoNewDeletesForDataFiles(MergingSnapshotProducer.java:418) ~[DevXXXX-0.14.1-1.13.5.jar:?]
	at org.apache.iceberg.MergingSnapshotProducer.validateNoNewDeletesForDataFiles(MergingSnapshotProducer.java:367) ~[DevXXXX-0.14.1-1.13.5.jar:?]
	at org.apache.iceberg.BaseRewriteFiles.validate(BaseRewriteFiles.java:108) ~[DevXXXX-0.14.1-1.13.5.jar:?]
	at org.apache.iceberg.SnapshotProducer.apply(SnapshotProducer.java:175) ~[DevXXXX-0.14.1-1.13.5.jar:?]
	at org.apache.iceberg.SnapshotProducer.lambda$commit$2(SnapshotProducer.java:296) ~[DevXXXX-0.14.1-1.13.5.jar:?]
	at org.apache.iceberg.util.Tasks$Builder.runTaskWithRetry(Tasks.java:404) ~[DevXXXX-0.14.1-1.13.5.jar:?]
	at org.apache.iceberg.util.Tasks$Builder.runSingleThreaded(Tasks.java:214) ~[DevXXXX-0.14.1-1.13.5.jar:?]
	at org.apache.iceberg.util.Tasks$Builder.run(Tasks.java:198) ~[DevXXXX-0.14.1-1.13.5.jar:?]
	at org.apache.iceberg.util.Tasks$Builder.run(Tasks.java:190) ~[DevXXXX-0.14.1-1.13.5.jar:?]
	at org.apache.iceberg.SnapshotProducer.commit(SnapshotProducer.java:295) ~[DevXXXX-0.14.1-1.13.5.jar:?]
	at org.apache.iceberg.actions.BaseSnapshotUpdateAction.commit(BaseSnapshotUpdateAction.java:41) ~[DevXXXX-0.14.1-1.13.5.jar:?]
	at org.apache.iceberg.actions.BaseRewriteDataFilesAction.doReplace(BaseRewriteDataFilesAction.java:298) ~[DevXXXX-0.14.1-1.13.5.jar:?]
	at org.apache.iceberg.actions.BaseRewriteDataFilesAction.replaceDataFiles(BaseRewriteDataFilesAction.java:277) ~[DevXXXX-0.14.1-1.13.5.jar:?]
	at org.apache.iceberg.actions.BaseRewriteDataFilesAction.execute(BaseRewriteDataFilesAction.java:252) ~[DevXXXX-0.14.1-1.13.5.jar:?]
	at org.XXXX.FlinkCompactIcebergData.main(FlinkCompactIcebergData.java:73) ~[DevXXXX-0.14.1-1.13.5.jar:?]
	at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) ~[?:1.8.0_74]
	at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62) ~[?:1.8.0_74]
	at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) ~[?:1.8.0_74]
	at java.lang.reflect.Method.invoke(Method.java:498) ~[?:1.8.0_74]
	at org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:355) ~[flink-dist_2.11-1.13.5.jar:1.13.5]
	at org.apache.flink.client.program.PackagedProgram.invokeInteractiveModeForExecution(PackagedProgram.java:222) ~[flink-dist_2.11-1.13.5.jar:1.13.5]
	at org.apache.flink.client.ClientUtils.executeProgram(ClientUtils.java:114) ~[flink-dist_2.11-1.13.5.jar:1.13.5]
	at org.apache.flink.client.deployment.application.ApplicationDispatcherBootstrap.runApplicationEntryPoint(ApplicationDispatcherBootstrap.java:242) ~[flink-dist_2.11-1.13.5.jar:1.13.5]
	at org.apache.flink.client.deployment.application.ApplicationDispatcherBootstrap.lambda$runApplicationAsync$1(ApplicationDispatcherBootstrap.java:212) ~[flink-dist_2.11-1.13.5.jar:1.13.5]
	at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) [?:1.8.0_74]
	at java.util.concurrent.FutureTask.run(FutureTask.java:266) [?:1.8.0_74]
	at org.apache.flink.runtime.concurrent.akka.ActorSystemScheduledExecutorAdapter$ScheduledFutureTask.run(ActorSystemScheduledExecutorAdapter.java:159) [flink-dist_2.11-1.13.5.jar:1.13.5]
	at akka.dispatch.TaskInvocation.run(AbstractDispatcher.scala:40) [flink-dist_2.11-1.13.5.jar:1.13.5]
	at akka.dispatch.ForkJoinExecutorConfigurator$AkkaForkJoinTask.exec(ForkJoinExecutorConfigurator.scala:44) [flink-dist_2.11-1.13.5.jar:1.13.5]
	at akka.dispatch.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260) [flink-dist_2.11-1.13.5.jar:1.13.5]
	at akka.dispatch.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1339) [flink-dist_2.11-1.13.5.jar:1.13.5]
	at akka.dispatch.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979) [flink-dist_2.11-1.13.5.jar:1.13.5]
	at akka.dispatch.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107) [flink-dist_2.11-1.13.5.jar:1.13.5]

compact-iceberg-file-commit-exception.log

[luoyuxia]

The new features of rocksdb mentioned in the PR #2680 have not found the relevant configuration items in the latest 1.0.0 version? Do you have to wait for a subsequent version to be added?

Yes, we have to wait it to be merged.

In compressing data it is indeed possible to solve the OOM problem by increasing the memory of TaskManger and increasing the compression frequency of data.

I agree with you, so I think use rocksdb as backend is a better solution

and the commit exception problem will be encountered when the data compression task is performed.

Had a look about your exception log. The reason is the cdc contains a delete row, but the compaction for such case that contains delete files hasn't been supported.

[SHuixo]

Yes, we have to wait it to be merged.

Good, looking forward to the merger of this rockdb new feature.

Had a look about your exception log. The reason is the cdc contains a delete row, but the compaction for such case that contains delete files hasn't been supported.

This means that in the CDC data that is streaming to Iceberg, don't have a viable data compression scheme for data streams that contain delete operations at this stage?

In order to verify whether the commit exception in the flow scenario has a similar problem in the batch scenario, we made the following attempts:

1. Enable flink CDC streaming writing iceberg, and the checkpoint is 5 minutes; 
2. When the writer is running, start the compression program until a **commit exception** occurs; 
3. When the above **commit exception** occurs, stop the CDC data writing program and compression program; 
4. Turn on the data compression program again until the program is up and running. 

The following figure shows the start and end times when flink CDC writes data to iceberg：

Some of the logs are as follows:
compact-data-when-stream-write.log
compact-data-when-write-finished.log

Here's a question,is it possible to pause the writer for data compression once, and when the data compression is completed, resume the data writing from the checkpoint again, and handle the above commit exception by cyclically suspending, compressing, and writing again?

[luoyuxia]

This means that in the CDC data that is streaming to Iceberg, don't have a viable data compression scheme for data streams that contain delete operations at this stage?

Yes, I'm afraid so.

Here's a question,is it possible to pause the writer for data compression once, and when the data compression is completed, resume the data writing from the checkpoint again, and handle the above commit exception by cyclically suspending, compressing, and writing again?

I think it's possible. From the code, IIUC, the exception happens when you start a compression, but it find delete files before finish compression. And once it won't produce any deletes files between start and finish a compression, there shouldn't throw exception.
But I'm not sure, you can have a try and to see whether it works.

[SHuixo]

Yes, we tried it and found it worked.
The following two log messages are the results of the compression test.

compact-data-when-stream-write.log
compact-data-when-write-finished.log