Commit 6dd28e4

authored

fix: prevent athena.to_iceberg overwrite to delete table in order to preserve Iceberg transactions history (#2776)

1 parent 60beb95 commit 6dd28e4Copy full SHA for 6dd28e4

File tree

+14

-5

lines changed

+14

-5

lines changed

Lines changed: 14 additions & 5 deletions

Original file line number	Diff line number	Diff line change
`@@ -383,11 +383,6 @@ def to_iceberg(`
`383`	`383`	`glue_table_settings if glue_table_settings else {},`
`384`	`384`	`)`
`385`	`385`
`386`		`- if mode == "overwrite":`
`387`		`- catalog.delete_table_if_exists(`
`388`		`- database=database, table=table, catalog_id=catalog_id, boto3_session=boto3_session`
`389`		`- )`
`390`		`-`
`391`	`386`	`try:`
`392`	`387`	`# Create Iceberg table if it doesn't exist`
`393`	`388`	`if not catalog.does_table_exist(`
`@@ -469,6 +464,20 @@ def to_iceberg(`
`469`	`464`	`s3_additional_kwargs=s3_additional_kwargs,`
`470`	`465`	`catalog_id=catalog_id,`
`471`	`466`	`)`
	`467`	`+ # if mode == "overwrite", delete whole data from table (but not table itself)`
	`468`	`+ elif mode == "overwrite":`
	`469`	`+ delete_sql_statement = f"DELETE FROM {table}"`
	`470`	`+ delete_query_execution_id: str = _start_query_execution(`
	`471`	`+ sql=delete_sql_statement,`
	`472`	`+ workgroup=workgroup,`
	`473`	`+ wg_config=wg_config,`
	`474`	`+ database=database,`
	`475`	`+ data_source=data_source,`
	`476`	`+ encryption=encryption,`
	`477`	`+ kms_key=kms_key,`
	`478`	`+ boto3_session=boto3_session,`
	`479`	`+ )`
	`480`	`+ wait_query(query_execution_id=delete_query_execution_id, boto3_session=boto3_session)`
`472`	`481`
`473`	`482`	`# Create temporary external table, write the results`
`474`	`483`	`s3.to_parquet(`

Comments

(0)