get_table_partitions
Retrieve partition information for Iceberg tables to analyze data organization and optimize queries.
Instructions
Provides the partitions for a given Iceberg table
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| namespace | Yes | ||
| table_name | Yes |
Implementation Reference
- iceberg_mcp/iceberg_server.py:87-106 (handler)Handler function for get_table_partitions tool. Loads the Iceberg table using the catalog and processes its partitions into a list of dictionaries containing partition details, record count, and size.
@mcp.tool() def get_table_partitions( namespace: str, table_name: str ) -> list[dict[str, int]]: """Provides the partitions for a given Iceberg table""" catalog: Catalog = get_catalog() table_obj = catalog.load_table((namespace, table_name)) partitions = table_obj.inspect.partitions().to_pylist() result = [] for p in partitions: result.append( { "partition": p['partition'], "record_count": p['record_count'], "size_in_bytes": p['total_data_file_size_in_bytes'], } ) return result - Helper function to create and return the GlueCatalog instance used by the get_table_partitions handler and other tools.
def get_catalog() -> GlueCatalog: try: session = boto3.Session(profile_name=iceberg_config.profile_name) credentials = session.get_credentials().get_frozen_credentials() catalog = GlueCatalog( "glue", **{ "client.access-key-id": credentials.access_key, "client.secret-access-key": credentials.secret_key, "client.session-token": credentials.token, "client.region": iceberg_config.region, }, ) except Exception as e: logger.error(f"Error creating AWS connection: {str(e)}") raise return catalog