Read delete files in parallel.

[Reo-LEI]

In upsert/cdc case, we usually will get a lot of pos-delete and eq-delete files. When we read/rewrite data from the v2 table, DeleteFilter will open all referenced pos-delete files and eq-delete files for each data file to construct the posDeleteSet and eqDeleteSet.

Currently, that all work will handled by same thread for each CombinedScanTask and all delete files are read serially, that is mean iceberg read a delete file must wait for the last file to be read and DeleteFilter will take a lot of time to open and read delete files.
I think DeleteFilter should read delete files in parallel when construct the posDeleteSet and eqDeleteSet to speed up reading v2 table.

[rdblue]

Are you sure that this is the right approach? It seems to me that if you have so many delete files that you need to read them in parallel, that you should rewrite and merge the delete files into data files. Optimizing for a large number of delete files doesn't seem like a good idea to me.

I'm interested to hear what @stevenzwu, @openinx, @jackye1995, @aokolnychyi, and @RussellSpitzer think, too.

[Reo-LEI]

Are you sure that this is the right approach? It seems to me that if you have so many delete files that you need to read them in parallel, that you should rewrite and merge the delete files into data files.

@rdblue I think this optimization is needed. On the one hand, user usually need to query the latest data in streaming case, And these data may not have been rewritten yet. On the other hand, as you said, the best way is to rewrite and merge the delete files into data files, but the rewrite is depend on read. So we should optimizing the DeleteFilter and read delete files in parallel to speed up the read and rewrite.

[Reo-LEI]

Actually, in our production environment, we sync our mysql data to iceberg v2 by flink. I do some test of this optimization by spark rewrite action on these v2 table. And we found this optimization is more than 2 times faster to execute than before.

caseId	dataFile number	dataFile + deleteFile number	deleteFile read parallelism	target file size	executor number	executor-core number	spark job task number	elapsed time(min)	task min elapsed time	task median elapsed time	task max elapsed time
1	755	2585	1	100MB	4	1	4	456	33s	1.5m	7.6h
2	755	2585	10	100MB	4	1	4	168	26s	37s	2.8h

[jackye1995]

Just to confirm, @Reo-LEI are you mostly doing this through Flink?

I am asking because I think currently we have the following dilemma: delete files are mostly generated by CDC pipelines in Flink, but rewrite functionality is not yet ready for delete files, and even if it's ready it's in Spark.

#2867 is another PR that tries to tackle the same root issue in a different way.

We definitely need to speed up the delete compaction progress, making it the top of the top priority. On the other side, I think we should start considering developing some actions in Flink to run compaction natively. Maybe the second compactor in streaming pipeline approach is not avoidable although a bit complex.

[jackye1995]

And just curious, it looks like the code is running in the same application as #2680 , is your benchmark based on the in memory map or this rock DB backed map?

[Reo-LEI]

@jackye1995 Yes, my original intention was to resolve the rewrite of streaming cdc data. At first I am trying to speed up the delete compaction progress, but I found that all engines can benefit from this optimization.

Maybe the second compactor in streaming pipeline approach is not avoidable although a bit complex.

+1, I think flink really need a compactor to apply/compaction delete file in streaming.

And just curious, it looks like the code is running in the same application as #2680 , is your benchmark based on the in memory map or this rock DB backed map?

This benchmark is based on the in-memory map, I'm not apply this PR in test.

[stevenzwu]

On the one hand, user usually need to query the latest data in streaming case, And these data may not have been rewritten yet. On the other hand, as you said, the best way is to rewrite and merge the delete files into data files, but the rewrite is depend on read. So we should optimizing the DeleteFilter and read delete files in parallel to speed up the read and rewrite.

Regarding this approach, I like the two supporting arguments from @Reo-LEI

[rdblue]

I'm fine adding this as long as it is configurable. I don't think that parallelizing delete file reads is the right long term approach, but it certainly helps right now.

[Reo-LEI]

Are you sure that this is the right approach? It seems to me that if you have so many delete files that you need to read them in parallel, that you should rewrite and merge the delete files into data files.

@rdblue I think this optimization is needed. On the one hand, user usually need to query the latest data in streaming case, And these data may not have been rewritten yet. On the other hand, as you said, the best way is to rewrite and merge the delete files into data files, but the rewrite is depend on read. So we should optimizing the DeleteFilter and read delete files in parallel to speed up the read and rewrite.

[github-actions]

This issue has been automatically marked as stale because it has been open for 180 days with no activity. It will be closed in next 14 days if no further activity occurs. To permanently prevent this issue from being considered stale, add the label 'not-stale', but commenting on the issue is preferred when possible.

[github-actions]

This issue has been closed because it has not received any activity in the last 14 days since being marked as 'stale'