Create a partition to the iceberg tables created by S3-Data-Lake Connector

[sharon-clue]

https://docs.airbyte.com/integrations/destinations/s3-data-lake

I really like the new connector for Iceberg and really keen on using it in production.
One thing that prevents me to do so is that the tables are created with sorting but without partitions

When having tables with even millions of records, not using a partition is very problematic and make the performance very bad

I'd suggest to create the cursor field as a partition and sort by the primary key

another issue i noticed was that when i ran compactions on the table and rerun the airbyte sync i get this error
java.lang.IllegalArgumentException: Cannot fast-forward: main is not an ancestor of airbyte_staging

i'm assuming that the airbyte_staging branch suppose to be on the same snapshot as the main branch
to fix that i manually changed the snapshot on the airbyte_staging branch to align with the main branch
given the compactions on iceberg tables are a required procedure it would be great if this could be fixed

[edgao]

thanks for starting this discussion! some questions about how you want to partition:

• do you have any streams configured in a non-dedup mode? (airbyte doesn't actually provide PK/cursor information to the destination connector in append/overwrite mode currently)
• what data type are your cursor + PK?
• do you have any requirements on how we partition? E.g daily vs hourly partitioning. In other words - would it be fine for the connector to always set an hourly partition on (e.g.) timestamp cursor, or do you need to be able to configure the granularity
  • do you have any non-temporal cursors? e.g. integer, etc - my intuition is that there's no sane default here, i.e. we'd need to ask for user input
• iceberg supports some fancier partitioning options (e.g. case-insensitive string partitioning), do you need any of those?

(also, just to check - I think we're already sorting on PK, does that match what you're seeing?)

good catch on the compaction thing - lemme take a look at that, I have some theories about it

[sharon-clue]

thanks for the response! really appreciate the good work :)

well, i developed a destination (a clunky one that i would be happy to replace) in the past (using Athena, don't know what i thinking!!) which is currently used in our production
and had to deal with your questions exactly.

one of the limitations is, that you don't have optional parameters that you could set on the schema level, so you'd have to have a generic destination level decision, so technically you could create different destinations for each partition type you'd want which isn't ideal but that's not something that we'd solve now :)

if to have more specific answers from the experience i had so far:

1. non-dedup mode: i guess if you do incremental only you might not want to have a sort key and if you do full refresh then you don't want a sort-key or a partition.
2. in our case the cursor is always a timestamp and the PK is almost always a uuid (which is a varchar in Iceberg) but if the cursor is an integer then it shouldn't be a partition but maybe the sort-key.
3. the partition we use is always a daily, i think that would work for 95% of cases
4. we do use string partitions but they are more for downstream tables and not raw table, it's hard for me to think of a use case that would be relevant here.
5. at the moment i do see sorting on the PK

regarding compactions, yeah, it would be nice to have it fixed.
i was thinking of running a script to align the branches as a workaround but i can wait if you'd say that it could be solved from your end.

[edgao]

nice, thanks for the detailed info! I think that mostly lines up with how we've been thinking about it, lemme check with the rest of the team to get their thoughts on some of this

(internally, I think we're leaning pretty heavily towards "we need to ask the user for input on partitioning", i.e. we have known usecases that would require non-default behavior :/ )

also - I think I have a plan for the compaction thing. Roughly: (disclaimer: unreviewed, this is just my personal thoughts, subject to change)

• implement a rudimentary "rebase" function, to replay the airbyte_staging branch onto updated main (i.e. find the shared upstream snapshot, list all new data/delete files in staging against that upstream, create a new delta with those new files against main)
• at the start of each sync attempt, rebase airbyte_staging onto main
• in case of race (e.g. you run compaction while a sync is running, before the sync fast-forwards main -> staging): crash; the next sync attempt will rebase and recover