[GitHub] [iceberg] rdblue commented on pull request #3323: Flink: Incrementally rewrite data files in streaming.

[GitBox <gi...@apache.org>]

rdblue commented on pull request #3323:
URL: https://github.com/apache/iceberg/pull/3323#issuecomment-962068331


   @jackye1995, thanks for taking the time to write up your thoughts. That doc is really helpful.
   
   I had previously been thinking about this mostly in terms of an append stream and v1 tables. Like @Reo-LEI notes, inline compaction for append streams is safe (for v1 tables) and is a good idea if you want frequent checkpoints but don't want a ton of small files. For v1 tables, we know that there won't be equality or position delete files, so we can easily compact. For v2 tables, append streams are pretty much like CDC streams because we may have concurrent writers adding equality or position delete files.
   
   That brings us back to Jack's points about adapting this idea to CDC streams. First, equality deletes create sequence number boundaries that make it difficult to compact. But I think there are still some use cases where this is valuable even if we only plan on compacting within a sequence number. Write tasks necessarily align with partitions, so compacting across tasks may still be valuable. For example, CDC writes to an unpartitioned table from multiple tasks will create a data file per task (per checkpoint) that is probably not full size.
   
   While there may be multiple files per partition, I think that the larger use case is compacting across checkpoints and that will require addressing compaction across sequence numbers. Let's assume that we have a fairly regular CDC stream for a bucketed table so that the commit for each bucket has one equality delete and one data file per checkpoint, and optionally a position delete file for the data file. That's the worst case that Jack describes, where there is nothing to compact within each sequence number except position deletes. Here are the files committed by sequence number for just one bucket (bucket_id = 0):
   
   | Type | Seq 107 | Seq 106 | Seq 105 | Seq 104 | Seq 103 |
   | -- | -- | -- | -- | -- | -- |
   | data | data-107.parquet | data-106.parquet | data-105.parquet | data-104.parquet | data-103.parquet |
   | eq-deletes | eq-107.parquet | eq-106.parquet | eq-105.parquet | eq-104.parquet | eq-103.parquet |
   | pos-deletes |  | pos-106.parquet |  |  |  pos-103.parquet |
   
   I think that we actually can do some compaction in this situation. Say we want to compact `data-103.parquet` through `data-107.parquet`. Then we need to apply all newer equality deletes to each data file:
   
   | File | Delete files to apply |
   | -- | -- |
   | data-103.parquet | eq-{104,105,106,107}.parquet, pos-103.parquet |
   | data-104.parquet | eq-{105,106,107}.parquet |
   | data-105.parquet | eq-{106,107}.parquet |
   | data-106.parquet | eq-107.parquet, pos-106.parquet |
   | data-107.parquet |  |
   
   The position delete files can be removed because they only reference data-103.parquet and data-106.parquet. The equality deletes must remain in the table in case they deleted data in other files. The new compacted data file should be data-107-compacted.parquet and should be committed at sequence number 107 so that future equality deletes are still applied correctly.
   
   Another thing to keep in mind is that we may have equality and position deletes coming from concurrent writers. But in this case all we would need to do is update the task emitting committed data files to emit all delete files, even those from concurrent writers. For example, maybe sequence number 5 was written by a concurrent commit. Then the delete files for that commit, eq-105.parquet, should be used. But the data file should not be part of the compaction (it would be compacted by a parallel Flink job).
   
   I think that this compaction strategy would address Jack's concern and is a safe way to compact across sequence numbers with equality deletes. The key idea is that we leave the equality delete files in the table so we don't have to worry about existing data files that aren't part of the stream we are compacting.
   
   I think that what I'm proposing is similar to @Reo-LEI's comment, except that it limits the scope of the compaction differently. Using time-based partitions would make us reasonably confident that the compaction is targeted, but the drawback is that we would still need to plan a scan, and it may still be impacted by concurrent deletes. But my strategy of keeping the compaction scoped to just files written by one Flink job and incrementally including delete files is a clean way to handle the issues and ensure that compaction won't conflict with one another.


-- 
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