Hi Manu, Just to clarify: - Are you proposing to create a user facing locking feature in Iceberg, or just something something for internal use?
I think we shouldn't add locking to Iceberg's user facing scope in this stage. A fully featured locking system has many more features that we need (priorities, fairness, timeouts etc). I could be tempted when we are talking about the REST catalog, but I think that should be further down the road, if ever... About using the tags: - I whole-heartedly agree that using tags is not intuitive, and I see your points in most of your arguments. OTOH, introducing new requirement (locking mechanism) seems like a wrong direction to me. - We already defined a requirement (atomic changes on the table) for the Catalog implementations which could be used to archive our goal here. - We also already store technical data in snapshot properties in Flink jobs (JobId, OperatorId, CheckpointId). Maybe technical tags/table properties is not a big stretch. Or we can look at these tags or metadata as 'technical data' which is internal to Iceberg, and shouldn't expressed on the external API. My concern is: - Would it be used often enough to worth the additional complexity? Knowing that Spark compaction is struggling with the same issue is a good indicator, but probably we would need more use cases for introducing a new feature with this complexity, or simpler solution. Thanks, Peter On Mon, Apr 1, 2024, 10:18 Manu Zhang <owenzhang1...@gmail.com> wrote: > What would the community think of exploiting tags for preventing >> concurrent maintenance loop executions. > > > This issue is not specific to Flink maintenance jobs. We have a service > scheduling Spark maintenance jobs by watching table commits. When we don't > check in-progress maintenance jobs for the same table, multiple jobs will > run concurrently and have conflicts. > > Basically, I think we need a lock mechanism like the metastore lock > <https://iceberg.apache.org/docs/nightly/configuration/#hadoop-configuration> > if we want to handle it for users. However, using TAG doesn't look > intuitive to me. We are also mixing user data with system metadata. > Maybe we can define some general interfaces and leave the implementation > to users in the first version. > > Regards, > Manu > > > > On Fri, Mar 29, 2024 at 1:59 PM Péter Váry <peter.vary.apa...@gmail.com> > wrote: > >> What would the community think of exploiting tags for preventing >> concurrent maintenance loop executions. >> >> The issue: >> Some maintenance tasks couldn't run parallel, like DeleteOrphanFiles vs. >> ExpireSnapshots, or RewriteDataFiles vs. RewriteManifestFiles. We make >> sure, not to run tasks started by a single trigger concurrently by >> serializing them, but there are no loops in Flink, so we can't synchronize >> tasks started by the next trigger. >> >> In the document, I describe that we need to rely on the user to ensure >> that the rate limit is high enough to prevent concurrent triggers. >> >> Proposal: >> When firing a trigger, RateLimiter could check and create an Iceberg >> table tag [1] for the current table snapshot, with the name: >> '__flink_maitenance'. When the execution finishes we remove this tag. The >> RateLimiter keep accumulating changes, and doesn't fire new triggers until >> it finds this tag on the table. >> The solution relies on Flink 'forceNonParallel' to prevent concurrent >> execution of placing the tag, and on Iceberg to store it. >> >> This not uses the tags as intended, but seems like a better solution than >> adding/removing table properties which would clutter the table history with >> technical data. >> >> Your thoughts? Any other suggestions/solutions would be welcome. >> >> Thanks, >> Peter >> >> [1] >> https://iceberg.apache.org/docs/latest/java-api-quickstart/#branching-and-tagging >> >> On Thu, Mar 28, 2024, 14:44 Péter Váry <peter.vary.apa...@gmail.com> >> wrote: >> >>> Hi Team, >>> >>> As discussed on yesterday's community sync, I am working on adding a >>> possibility to the Flink Iceberg connector to run maintenance tasks on the >>> Iceberg tables. This will fix the small files issues and in the long run >>> help compacting the high number of positional and equality deletes created >>> by Flink tasks writing CDC data to Iceberg tables without the need of Spark >>> in the infrastructure. >>> >>> I did some planning, prototyping and currently trying out the solution >>> on a larger scale. >>> >>> I put together a document how my current solution looks like: >>> >>> https://docs.google.com/document/d/16g3vR18mVBy8jbFaLjf2JwAANuYOmIwr15yDDxovdnA/edit?usp=sharing >>> >>> I would love to hear your thoughts and feedback on this to find a good >>> final solution. >>> >>> Thanks, >>> Peter >>> >>
