mwa28 commented on issue #2166:
URL:
https://github.com/apache/iceberg-python/issues/2166#issuecomment-3028106959
I figured it out, it is not a simple schema update from outside Athena that
causes corruption but rather deleting the column that was set as Sort Order
Key.
This behavior is not allowed within Athena via
```sql
ALTER TABLE make_me_corrupt.corrupt_me DROP COLUMN col_2
```
as it gives :
```
Failed: Input does not satisfy Iceberg requirement
```
Meanwhile with pyiceberg it is allowed. Here is my test :
```python
import unittest
from pyathena.error import OperationalError
from pyathena import connect
from pyiceberg.catalog import load_catalog
from pyiceberg.schema import Schema
from pyiceberg.table.sorting import SortField, SortOrder
from pyiceberg.transforms import IdentityTransform
from pyiceberg.types import NestedField, StringType
class TestPyicebergAthenaCorrupt(unittest.TestCase):
def setUp(self):
self.catalog = load_catalog("glue", **{"type": "glue"})
self.TABLE_NAME = "make_me_corrupt.corrupt_me"
self.schema = Schema(
NestedField(1, "col_1", StringType(), required=True),
NestedField(2, "col_2", StringType(), required=True),
identifier_field_ids=[1], # 'col_1' is the primary key
)
self.iceberg_table = self.catalog.create_table(
identifier=self.TABLE_NAME,
schema=self.schema,
sort_order=SortOrder(
SortField(
source_id=2, transform=IdentityTransform()
) # col2 is the sort key
),
location="s3://<my-bucket>/icebergCorrupted/",
properties={
"write.format.default": "parquet",
"write.target-file-size-bytes": 134217728, # 128 MB
"write.metadata.delete-after-commit.enabled": True,
"write.metadata.previous-versions-max": 5,
},
)
assert self.iceberg_table.schema() == self.schema
self.cur = connect(
s3_staging_dir="s3://<my-bucket>/athena-query-result/",
region_name="eu-west-1",
).cursor()
def tearDown(self):
# Delete the table and drop it from the catalog
try:
self.catalog.drop_table(self.TABLE_NAME)
except Exception as e:
print(f"Error during teardown: {e}")
def test_corrupt_table(self):
# Assert table exists
self.assertTrue(self.catalog.table_exists(self.TABLE_NAME))
# Drop the sort order column
self.iceberg_table.update_schema().delete_column("col_2").commit()
new_schema = Schema(
NestedField(1, "col_1", StringType(), required=True),
identifier_field_ids=[1], # 'col_1' is the primary key
)
self.assertEqual(self.iceberg_table.schema(), new_schema)
with self.assertRaises(OperationalError) as context:
self.cur.execute(f"SELECT * FROM {self.TABLE_NAME}")
print(context.exception)
if __name__ == "__main__":
unittest.main()
```
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
