Destination Snowflake: Sync Id, generation_id and Meta

[gisripa]

What

Adding migration for Snowflake Raw tables and final tables to include _airbyte_meta and _airbyte_generation_id columns.
Tests and refactors to adapt to latest CDK.

Review guide

User Impact

Can this PR be safely reverted and rolled back?

• YES 💚
• NO ❌

[vercel]

The latest updates on your projects. Learn more about Vercel for Git ↗︎

Name	Status	Preview	Comments	Updated (UTC)
airbyte-docs	✅ Ready (Inspect)	Visit Preview	💬 Add feedback	Jun 10, 2024 11:31pm

[gisripa]

• Destination Snowflake: Sync Id, generation_id and Meta #39107 👈
• Destinations CDK: CatalogParser sets default namespace #38121 : 1 other dependent PR (#38075 )
• Destination bigquery: switch cdk back to artifact #38770
• master

This stack of pull requests is managed by Graphite. Learn more about stacking.

Join @gisripa and the rest of your teammates on Graphite

[edgao]

looks reasonable, added some nitpicky comments but wouldn't block on any of them. Lmk if you want to ship this pr standalone and I'll approve 👍

[edgao]

a couple nitpicks, otherwise lgtm

[edgao]

we're already restricting to specifically airbyte_internal.whatever, why do we need to group by schema name / table name?

[gisripa]

just to avoid. list has 1 element and first() call i guess.

[edgao]

I think you get the same result with this?

TableDefinition(
            results.associateTo(LinkedHashMap()) {
                // return value of data_type in show columns is a json string.
                val dataType = Jsons.deserialize(it.get("data_type").asText())
                it.get("column_name").asText()!! to
                    ColumnDefinition(
                        it.get("column_name").asText(),
                        dataType.get("type").asText(),
                        0,
                        dataType.get("nullable").asBoolean(),
                    )
            }
        )

i.e. don't need to group on schema/table name at all, just directly build the map of column name to definition

(... but also, I would hope that kotlin has some easy .first accessor :P )

[oskarthansen]

It seems like this update makes "full refresh overwrite" fail..

Error message:

JavaScript execution error: Uncaught Execution of multiple statements failed on statement "INSERT INTO "?"."LINKEDINADS..." (at line 2, position 0).
SQL compilation error: error line 76 at position 12
invalid identifier '"_?_meta"' in SYSTEM$MULTISTMT at '    throw `Execution of multiple statements failed on statement {0} (at line {1}, position {2}).`.replace('{1}', LINES[i])' position 4
stackstrace: 
SYSTEM$MULTISTMT line: 10

[pKorsholm]

Hi @gisripa
This pr seems to just have broken a lot of snowflake destinations causing the following error:

Java.lang.RuntimeException: SQL compilation error: error line 63 at position 12 invalid identifier '"_airbyte_meta"' in SYSTEM$MULTISTMT at ' throw `Execution of multiple statements failed on statement {0} (at line {1}, position {2}).`.replace('{1}', LINES[i])' position 4 stackstrace: SYSTEM$MULTISTMT line: 10

(if you have a remedy for this we could implement that would be great)

[edgao]

@pKorsholm / @tuday2 we believe this only affects overwrite streams - can you confirm?

[tuday2]

For me its "Full Refresh / Overwrite" and "Incremental | Append + Deduped", i have at least 20+ examples

[edgao]

can you send the logs for one of the dedup streams running into problems? (feel free to slack me if you don't want to post them publicly)

so far we've only seen this happen on overwrite streams, so I'm not sure what's happening there

[edgao]

in the meantime, we have #39399, which should fix for overwrite streams. It's just pending green CI.

[edgao]

for vis: @tuday2 sent me the logs - thanks! confirmed that the streams with errors were all full refresh overwrite, which caused the UI to mark other streams as failed (b/c the entire sync terminated uncleanly)