show_table_history

Instructions

Implementation Reference

• src/server.py:266-288 (handler)

  MCP tool handler function 'show_table_history' registered with @mcp.tool decorator. Accepts catalog, schema_name, and table parameters, delegates to client.show_table_history().

  @mcp.tool(description="Show Iceberg table history/changelog")
  def show_table_history(
      catalog: str = Field(description="catalog name "),
      schema_name: str = Field(description="schema name "),
      table: str = Field(description="The name of the table"),
  ) -> str:
      """Show Iceberg table history/changelog.
  
      The history contains:
      - made_current_at: When snapshot became active
      - snapshot_id: Identifier of the snapshot
      - parent_id: Identifier of the parent snapshot
      - is_current_ancestor: Whether snapshot is an ancestor of current
  
      Args:
          catalog: catalog name
          schema_name: schema name
          table: The name of the table
  
      Returns:
          str: JSON-formatted table history
      """
      return client.show_table_history(catalog, schema_name, table)

• src/server.py:266-288 (registration)

  Tool registered via @mcp.tool(description='Show Iceberg table history/changelog') decorator on line 266.

  @mcp.tool(description="Show Iceberg table history/changelog")
  def show_table_history(
      catalog: str = Field(description="catalog name "),
      schema_name: str = Field(description="schema name "),
      table: str = Field(description="The name of the table"),
  ) -> str:
      """Show Iceberg table history/changelog.
  
      The history contains:
      - made_current_at: When snapshot became active
      - snapshot_id: Identifier of the snapshot
      - parent_id: Identifier of the parent snapshot
      - is_current_ancestor: Whether snapshot is an ancestor of current
  
      Args:
          catalog: catalog name
          schema_name: schema name
          table: The name of the table
  
      Returns:
          str: JSON-formatted table history
      """
      return client.show_table_history(catalog, schema_name, table)

• src/trino_client.py:370-393 (helper)

  Client method that executes the Iceberg table history query: 'SELECT * FROM "{catalog}.{schema}.{table}$history"' against Trino, returning JSON-formatted results.

  def show_table_history(self, table: str, catalog: str, schema: str) -> str:
      """Show Iceberg table history/changelog.
  
      The history contains:
      - made_current_at: TIMESTAMP(3) WITH TIME ZONE - Time when snapshot became active
      - snapshot_id: BIGINT - Identifier of the snapshot
      - parent_id: BIGINT - Identifier of the parent snapshot
      - is_current_ancestor: BOOLEAN - Whether this snapshot is an ancestor of current
  
      Args:
          table: The name of the table
          catalog: Optional catalog name (defaults to configured catalog)
          schema: Optional schema name (defaults to configured schema)
  
      Returns:
          str: JSON-formatted string containing table history
      """
      catalog = catalog or self.config.catalog
      schema = schema or self.config.schema
      if not catalog or not schema:
          raise CatalogSchemaError
      table_identifier = f"{catalog}.{schema}.{table}"
      query = 'SELECT * FROM "{}$history"'
      return self.execute_query(query.format(table_identifier))