Python: Alter table plumbing and REST support

python/mkdocs/docs/api.md

@@ -78,110 +78,15 @@ catalog.load_table(("nyc", "taxis"))
 # The tuple syntax can be used if the namespace or table contains a dot.
 ```
 
-This returns a `Table` that represents an Iceberg table:
-
-```python
-Table(
-  identifier=('nyc', 'taxis'),
-  metadata_location='s3a://warehouse/wh/nyc.db/taxis/metadata/00002-6ea51ce3-62aa-4197-9cf8-43d07c3440ca.metadata.json',
-  metadata=TableMetadataV2(
-    location='s3a://warehouse/wh/nyc.db/taxis',
-    table_uuid=UUID('ebd5d172-2162-453d-b586-1cdce52c1116'),
-    last_updated_ms=1662633437826,
-    last_column_id=19,
-    schemas=[Schema(
-        NestedField(field_id=1, name='VendorID', field_type=LongType(), required=False),
-        NestedField(field_id=2, name='tpep_pickup_datetime', field_type=TimestamptzType(), required=False),
-        NestedField(field_id=3, name='tpep_dropoff_datetime', field_type=TimestamptzType(), required=False),
-        NestedField(field_id=4, name='passenger_count', field_type=DoubleType(), required=False),
-        NestedField(field_id=5, name='trip_distance', field_type=DoubleType(), required=False),
-        NestedField(field_id=6, name='RatecodeID', field_type=DoubleType(), required=False),
-        NestedField(field_id=7, name='store_and_fwd_flag', field_type=StringType(), required=False),
-        NestedField(field_id=8, name='PULocationID', field_type=LongType(), required=False),
-        NestedField(field_id=9, name='DOLocationID', field_type=LongType(), required=False),
-        NestedField(field_id=10, name='payment_type', field_type=LongType(), required=False),
-        NestedField(field_id=11, name='fare_amount', field_type=DoubleType(), required=False),
-        NestedField(field_id=12, name='extra', field_type=DoubleType(), required=False),
-        NestedField(field_id=13, name='mta_tax', field_type=DoubleType(), required=False),
-        NestedField(field_id=14, name='tip_amount', field_type=DoubleType(), required=False),
-        NestedField(field_id=15, name='tolls_amount', field_type=DoubleType(), required=False),
-        NestedField(field_id=16, name='improvement_surcharge', field_type=DoubleType(), required=False),
-        NestedField(field_id=17, name='total_amount', field_type=DoubleType(), required=False),
-        NestedField(field_id=18, name='congestion_surcharge', field_type=DoubleType(), required=False),
-        NestedField(field_id=19, name='airport_fee', field_type=DoubleType(), required=False)
-      ),
-      schema_id=0,
-      identifier_field_ids=[]
-    )],
-    current_schema_id=0,
-    partition_specs=[PartitionSpec(spec_id=0)],
-    default_spec_id=0,
-    last_partition_id=999,
-    properties={
-      'owner': 'root',
-      'write.format.default': 'parquet'
-    },
-    current_snapshot_id=8334458494559715805,
-    snapshots=[
-      Snapshot(
-        snapshot_id=7910949481055846233,
-        parent_snapshot_id=None,
-        sequence_number=None,
-        timestamp_ms=1662489306555,
-        manifest_list='s3a://warehouse/wh/nyc.db/taxis/metadata/snap-7910949481055846233-1-3eb7a2e1-5b7a-4e76-a29a-3e29c176eea4.avro',
-        summary=Summary(
-          Operation.APPEND,
-          **{
-            'spark.app.id': 'local-1662489289173',
-            'added-data-files': '1',
-            'added-records': '2979431',
-            'added-files-size': '46600777',
-            'changed-partition-count': '1',
-            'total-records': '2979431',
-            'total-files-size': '46600777',
-            'total-data-files': '1',
-            'total-delete-files': '0',
-            'total-position-deletes': '0',
-            'total-equality-deletes': '0'
-          }
-        ),
-        schema_id=0
-      ),
-    ],
-    snapshot_log=[
-      SnapshotLogEntry(
-        snapshot_id='7910949481055846233',
-        timestamp_ms=1662489306555
-      )
-    ],
-    metadata_log=[
-      MetadataLogEntry(
-        metadata_file='s3a://warehouse/wh/nyc.db/taxis/metadata/00000-b58341ba-6a63-4eea-9b2f-e85e47c7d09f.metadata.json',
-        timestamp_ms=1662489306555
-      )
-    ],
-    sort_orders=[SortOrder(order_id=0)],
-    default_sort_order_id=0,
-    refs={
-      'main': SnapshotRef(
-        snapshot_id=8334458494559715805,
-        snapshot_ref_type=SnapshotRefType.BRANCH,
-        min_snapshots_to_keep=None,
-        max_snapshot_age_ms=None,
-        max_ref_age_ms=None
-      )
-    },
-    format_version=2,
-    last_sequence_number=1
-  )
-)
-```
+This returns a `Table` that represents an Iceberg table that can be queried and altered.
 
 ### Directly from a metadata file
 
 To load a table directly from a metadata file (i.e., **without** using a catalog), you can use a `StaticTable` as follows:
 
 ```python
+from pyiceberg.table import StaticTable
+
 table = StaticTable.from_metadata(
     "s3a://warehouse/wh/nyc.db/taxis/metadata/00002-6ea51ce3-62aa-4197-9cf8-43d07c3440ca.metadata.json"
 )
@@ -200,7 +105,7 @@ table = StaticTable.from_metadata(
 
 ## Create a table
 
-To create a table from a catalog:
+To create a table by passing the table to the catalog:
 
 ```python
 from pyiceberg.catalog import load_catalog
@@ -241,52 +146,88 @@ catalog.create_table(
 )
 ```
 
-Which returns a newly created table:
+## Altering the table metadata
+
+Using the Python API you can alter table metadata.
+
+### Update the schema
+
+Add a new field to the table:
+
+```python
+from pyiceberg.schema import Schema
+from pyiceberg.types import (
+    BooleanType,
+    DoubleType,
+    IntegerType,
+    NestedField,
+    StringType,
+    TimestampType,
+)
+
+schema = Schema(
+    NestedField(field_id=1, name="str", field_type=StringType(), required=False),
+    NestedField(field_id=2, name="int", field_type=IntegerType(), required=True),
+    NestedField(field_id=3, name="bool", field_type=BooleanType(), required=False),
+    NestedField(
+        field_id=4, name="datetime", field_type=TimestampType(), required=False
+    ),
+    # Add a new column to the table
+    NestedField(field_id=5, name="double", field_type=DoubleType(), required=False),
+)
+
+table = table.alter().set_schema(schema).commit()
+```
+
+### Update the partition spec
+
+Updates the partition spec that will be applied for all new data that's being added to the table.
 
 ```python
-Table(
-    identifier=('default', 'bids'),
-    metadata_location='/Users/fokkodriesprong/Desktop/docker-spark-iceberg/wh/bids//metadata/00000-c8cd93ab-f784-474d-a167-b1a86b05195f.metadata.json',
-    metadata=TableMetadataV2(
-        location='/Users/fokkodriesprong/Desktop/docker-spark-iceberg/wh/bids/',
-        table_uuid=UUID('38d4cb39-4945-4bf2-b374-984b5c4984d2'),
-        last_updated_ms=1661847562069,
-        last_column_id=4,
-        schemas=[
-            Schema(
-                NestedField(field_id=1, name='datetime', field_type=TimestampType(), required=False),
-                NestedField(field_id=2, name='bid', field_type=DoubleType(), required=False),
-                NestedField(field_id=3, name='ask', field_type=DoubleType(), required=False),
-                NestedField(field_id=4, name='symbol', field_type=StringType(), required=False)),
-                schema_id=1,
-                identifier_field_ids=[])
-        ],
-        current_schema_id=1,
-        partition_specs=[
-            PartitionSpec(
-                PartitionField(source_id=1, field_id=1000, transform=DayTransform(), name='datetime_day'),))
-        ],
-        default_spec_id=0,
-        last_partition_id=1000,
-        properties={},
-        current_snapshot_id=None,
-        snapshots=[],
-        snapshot_log=[],
-        metadata_log=[],
-        sort_orders=[
-            SortOrder(order_id=1, fields=[SortField(source_id=4, transform=IdentityTransform(), direction=SortDirection.ASC, null_order=NullOrder.NULLS_FIRST)])
-        ],
-        default_sort_order_id=1,
-        refs={},
-        format_version=2,
-        last_sequence_number=0
+from pyiceberg.partitioning import PartitionField, PartitionSpec
+from pyiceberg.transforms import DayTransform
+
+spec = PartitionSpec(
+    PartitionField(
+        source_id=4, field_id=1000, transform=DayTransform(), name="datetime_day"
     )
 )
+
+table = table.alter().set_partition_spec(spec).commit()
+```
+
+### Update the sort order
+
+Updates the sort order of the table.
+
+```python
+from pyiceberg.table.sorting import SortOrder, SortField
+from pyiceberg.transforms import IdentityTransform
+
+order = SortOrder(SortField(source_id=2, transform=IdentityTransform()))
+
+table = table.alter().set_sort_order(order).commit()
+```
+
+### Update the properties
+
+Add, update and remove properties:
+
+```python
+assert table.properties == {}
+
+table = table.alter().set_properties(abc="def").commit()
+
+assert table.properties == {"abc": "def"}
+
+table = table.alter().unset_properties("abc").commit()
+
+assert table.properties == {}
 ```
 
-## Query a table
+## Query the data
 
-To query a table, a table scan is needed. A table scan accepts a filter, columns and optionally a limit and a snapshot ID:
+To query a table, a table scan is needed. A table scan accepts a filter, columns, optionally a limit and a snapshot ID:
 
 ```python
 from pyiceberg.catalog import load_catalog

python/mkdocs/docs/feature-support.md

@@ -42,6 +42,22 @@ The goal is that the python library will provide a functional, performant subset
 | Drop Namespace           |  X   |   X    |
 | Set Namespace Properties |  X   |   X    |
 
+## Catalog capabilities
+
+The following table projects the capabilities of each of the available Catalog implementations:
+
+|                        | Rest | Hive | Glue | DynamoDB |
+| ---------------------- | ---- | ---- | ---- | -------- |
+| CRUD Namespaces        | ✓    | ✓    | ✓    | ✓        |
+| CRUD Tables            | ✓    | ✓    | ✓    | ✓        |
+| Update format version  | ✓    | ✕    | ✕    | ✕        |
+| Alter table properties | ✓    | ✕    | ✕    | ✕        |
+| Alter schema           | ✓    | ✕    | ✕    | ✕        |
+| Alter partition spec   | ✓    | ✕    | ✕    | ✕        |
+| Alter sort order       | ✓    | ✕    | ✕    | ✕        |
+| Alter snapshots        | ✕    | ✕    | ✕    | ✕        |
+| Refresh tables         | ✓    | ✕    | ✕    | ✕        |
+
 ## Types
 
 The types are kept in `pyiceberg.types`.

python/pyiceberg/catalog/__init__.py

@@ -40,7 +40,7 @@
 from pyiceberg.partitioning import UNPARTITIONED_PARTITION_SPEC, PartitionSpec
 from pyiceberg.schema import Schema
 from pyiceberg.serializers import ToOutputFile
-from pyiceberg.table import Table, TableMetadata
+from pyiceberg.table import BaseTableUpdate, Table, TableMetadata
 from pyiceberg.table.sorting import UNSORTED_SORT_ORDER, SortOrder
 from pyiceberg.typedef import (
     EMPTY_DICT,
@@ -323,6 +323,17 @@ def rename_table(self, from_identifier: Union[str, Identifier], to_identifier: U
             NoSuchTableError: If a table with the name does not exist
         """
 
+    @abstractmethod
+    def alter_table(self, identifier: Union[str, Identifier], updates: Tuple[BaseTableUpdate, ...]) -> Table:
+        """Updates the table
+
+        Args:
+            identifier (str | Identifier): Namespace identifier
+            updates (Tuple[BaseTableUpdate]): Updates to be applied to the table
+        Raises:
+            NoSuchTableError: If a table with the given identifier does not exist
+        """
+
     @abstractmethod
     def create_namespace(self, namespace: Union[str, Identifier], properties: Properties = EMPTY_DICT) -> None:
         """Create a namespace in the catalog.
@@ -393,7 +404,7 @@ def load_namespace_properties(self, namespace: Union[str, Identifier]) -> Proper
 
     @abstractmethod
     def update_namespace_properties(
-        self, namespace: Union[str, Identifier], removals: set[str] | None = None, updates: Properties = EMPTY_DICT
+        self, namespace: Union[str, Identifier], removals: Optional[Set[str]] = None, updates: Properties = EMPTY_DICT
     ) -> PropertiesUpdateSummary:
         """Removes provided property keys and updates properties for a namespace.
 

python/pyiceberg/catalog/dynamodb.py

@@ -22,6 +22,7 @@
     List,
     Optional,
     Set,
+    Tuple,
     Union,
 )
 
@@ -52,7 +53,7 @@
 from pyiceberg.partitioning import UNPARTITIONED_PARTITION_SPEC, PartitionSpec
 from pyiceberg.schema import Schema
 from pyiceberg.serializers import FromInputFile
-from pyiceberg.table import Table
+from pyiceberg.table import BaseTableUpdate, Table
 from pyiceberg.table.metadata import new_table_metadata
 from pyiceberg.table.sorting import UNSORTED_SORT_ORDER, SortOrder
 from pyiceberg.typedef import EMPTY_DICT
@@ -168,6 +169,17 @@ def create_table(
 
         return self.load_table(identifier=identifier)
 
+    def alter_table(self, identifier: Union[str, Identifier], updates: Tuple[BaseTableUpdate, ...]) -> Table:
+        """Updates the table
+
+        Args:
+            identifier (str | Identifier): Namespace identifier
+            updates (Tuple[BaseTableUpdate]): Updates to be applied to the table
+        Raises:
+            NoSuchTableError: If a table with the given identifier does not exist
+        """
+        raise NotImplementedError
+
     def load_table(self, identifier: Union[str, Identifier]) -> Table:
         """
         Loads the table's metadata and returns the table instance.
@@ -580,6 +592,7 @@ def _convert_dynamo_table_item_to_iceberg_table(self, dynamo_table_item: Dict[st
             metadata=metadata,
             metadata_location=metadata_location,
             io=self._load_file_io(metadata.properties, metadata_location),
+            catalog=self,
         )
 
 

python/pyiceberg/catalog/glue.py

@@ -22,6 +22,7 @@
     List,
     Optional,
     Set,
+    Tuple,
     Union,
 )
 
@@ -51,7 +52,7 @@
 from pyiceberg.partitioning import UNPARTITIONED_PARTITION_SPEC, PartitionSpec
 from pyiceberg.schema import Schema
 from pyiceberg.serializers import FromInputFile
-from pyiceberg.table import Table
+from pyiceberg.table import BaseTableUpdate, Table
 from pyiceberg.table.metadata import new_table_metadata
 from pyiceberg.table.sorting import UNSORTED_SORT_ORDER, SortOrder
 from pyiceberg.typedef import EMPTY_DICT
@@ -164,6 +165,7 @@ def _convert_glue_to_iceberg(self, glue_table: Dict[str, Any]) -> Table:
             metadata=metadata,
             metadata_location=metadata_location,
             io=self._load_file_io(metadata.properties, metadata_location),
+            catalog=self,
         )
 
     def _create_glue_table(self, database_name: str, table_name: str, table_input: Dict[str, Any]) -> None:
@@ -218,6 +220,17 @@ def create_table(
 
         return self.load_table(identifier=identifier)
 
+    def alter_table(self, identifier: Union[str, Identifier], updates: Tuple[BaseTableUpdate, ...]) -> Table:
+        """Updates the table
+
+        Args:
+            identifier (str | Identifier): Namespace identifier
+            updates (Tuple[BaseTableUpdate]): Updates that applied to the table
+        Raises:
+            NoSuchTableError: If a table with the given identifier does not exist
+        """
+        raise NotImplementedError
+
     def load_table(self, identifier: Union[str, Identifier]) -> Table:
         """Loads the table's metadata and returns the table instance.