CATALOG_NOT_FOUND errors on insert statements with Iceberg connector with Glue & S3

[peterklingelhofer]

We're seeing CATALOG_NOT_FOUND errors from Trino when we attempt INSERT statements. SELECT and CREATE TABLE statements work successfully (the select statement works when we grab data that was already inserted via AWS Athena, as Trino data insertion does not work here yet). When we run SHOW CATALOGS, we rightly see Available catalogs: [['glue'], ['system'], ['tpcds'], ['tpch']], and when we run SHOW SCHEMAS IN glue we rightly see Available schemas in glue: [['default'], ['information_schema'], ['test_database']]. Select statements work successfully and grab data from the source. Table creation works fine, as does creating new databases. This service worker's Role ARN has access to all S3 resources with s3:* and glue:*.

Weirdly, when I change the commit type to IsolationLevel.READ_COMMITTED, we see Catalog only supports writes using autocommit: glue (which tells me the catalog is indeed present...).

Any ideas? Example code:

Trino helm chart values overrides (using Trino helm chart 0.28.0):

# Full chart: https://github.com/trinodb/charts/blob/main/charts/trino/values.yaml
serviceAccount:
  create: true
  name: trino-sa
  annotations:
    eks.amazonaws.com/role-arn: arn:aws:iam::############::role/trino-role

server:
  workers: 2

# https://trino.io/docs/current/connector/iceberg.html
additionalCatalogs:
  glue: |-
    connector.name=iceberg
    iceberg.catalog.type=glue
    iceberg.file-format=PARQUET
    iceberg.compression-codec=ZSTD
    iceberg.security=ALLOW_ALL
    hive.metastore.glue.region=us-east-1
    hive.metastore.glue.endpoint-url=glue.us-east-1.amazonaws.com
    hive.metastore.glue.catalogid=############

(including the hive.metastore.glue properties does not seem to have any effect on this catalog not found error)

import pandas as pd
import trino
from loguru import logger


def execute_trino_query(conn: trino.dbapi.Connection, query: str) -> pd.DataFrame | None:
    """Execute a query using Trino and return a Pandas DataFrame or None for non-select queries."""
    logger.info(f"Executing query: {query}")
    cursor = conn.cursor()
    try:
        cursor.execute(query)

        if cursor.description is None:
            logger.info("Query executed successfully, no result set to fetch.")
            return None

        result = cursor.fetchall()
        columns = [desc[0] for desc in cursor.description]
        df = pd.DataFrame(result, columns=columns)
        return df

    except Exception as e:
        logger.error(f"Error executing query: {str(e)}")
        raise

    finally:
        cursor.close()


def trino_init() -> trino.dbapi.Connection:
    """Initialize Trino connection"""
    return trino.dbapi.connect(
        host="localhost",
        port=8080,
        user="admin",
        catalog="glue",
        schema="test_database",
    )


test_row = {
    "timestamp": "2024-08-01T00:00:00.000000",
    "id": "Test1",
    "name": "Test1",
    "type": "TEST",
    # ...
}

df = pd.DataFrame([test_row])

df["timestamp"] = pd.to_datetime(df["timestamp"])
df["id"] = df["id"].astype(str)
df["name"] = df["name"].astype(str)
df["type"] = df["type"].astype(str)
df["alarm_status__has_warning"] = df["alarm_status__has_warning"].astype(bool)
df["alarm_status__has_fault"] = df["alarm_status__has_fault"].astype(bool)
df["communication_status__communicating"] = df["communication_status__communicating"].astype(bool)
df["operating_status__operating"] = df["operating_status__operating"].astype(bool)
df["standby_status__standby"] = df["standby_status__standby"].astype(bool)
df["lock_status__locked"] = df["lock_status__locked"].astype(bool)
df.fillna(value={"real_power_limits__max_charge_real_power_limit__w": 0}, inplace=True)


def create_glue_table(conn: trino.dbapi.Connection, table_name: str):
    query = f"""
    CREATE TABLE IF NOT EXISTS test_database.{table_name} (
        timestamp TIMESTAMP,
        id VARCHAR,
        name VARCHAR,
        type VARCHAR,
        ac_power_status__real_power__w DOUBLE,
        ac_power_status__reactive_power__var DOUBLE,
        ac_power_status__apparent_power__va DOUBLE
    )
    WITH (
        format = 'PARQUET',
        partitioning = ARRAY['timestamp']
    )
    """
    logger.info(f"Creating table with query: {query}")
    execute_trino_query(conn, query)
    logger.info(f"Table {table_name} created successfully in Glue")


def insert_data_into_glue_table(conn: trino.dbapi.Connection, table_name: str, df: pd.DataFrame):
    df["timestamp"] = pd.to_datetime(df["timestamp"])

    values = []
    for row in df.itertuples(index=False):
        formatted_timestamp = f"TIMESTAMP '{row.timestamp.strftime('%Y-%m-%d %H:%M:%S.%f')}'"
        values.append(
            f"({formatted_timestamp}, '{row.id}', '{row.name}', '{row.type}', {float(row.ac_power_status__real_power__w)}, {float(row.ac_power_status__reactive_power__var)}, {float(row.ac_power_status__apparent_power__va)})"
        )

    values_str = ",".join(values)

    query = f"""
    INSERT INTO test_database.{table_name} (timestamp, id, name, type, ac_power_status__real_power__w, ac_power_status__reactive_power__var, ac_power_status__apparent_power__va)
    VALUES {values_str}
    """
    logger.info(f"Inserting data into table {table_name}")
    execute_trino_query(conn, query)
    logger.info(f"Data inserted into {table_name} successfully")


def run_glue_table_creation_and_insertion(conn: trino.dbapi.Connection):
    table_name = "trino_test_table"

    data = {
        "timestamp": ["2024-08-01T00:00:00.442000"],
        "id": ["Test1"],
        "name": ["Test1"],
        "type": ["TEST"],
        "ac_power_status__real_power__w": [1],
        "ac_power_status__reactive_power__var": [1],
        "ac_power_status__apparent_power__va": [1],
    }
    df = pd.DataFrame(data)

    insert_data_into_glue_table(conn, table_name, df)


def test_trino_connection(conn: trino.dbapi.Connection):
    cursor = conn.cursor()
    query = "SELECT * FROM trino_test_table LIMIT 1"
    try:
        cursor.execute("SHOW CATALOGS")
        catalogs = cursor.fetchall()
        logger.info(f"Available catalogs: {catalogs}")
        cursor.execute("SHOW SCHEMAS IN glue")
        schemas = cursor.fetchall()
        logger.info(f"Available schemas in glue: {schemas}")
        logger.info(f"Testing connection with query: {query}")

        result = execute_trino_query(conn, query)

        logger.info(f"Query Result: {result}")

        logger.info(f"Successfully executed the test query: {query}")
    except Exception as e:
        logger.error(f"Failed to execute query: {str(e)}")
    finally:
        cursor.close()


def run_test():
    try:
        conn = trino_init()

        test_trino_connection(conn)

        run_glue_table_creation_and_insertion(conn)

        logger.info("Test completed successfully")

    except Exception as e:
        logger.error(f"Test failed: {str(e)}")
        raise


if __name__ == "__main__":
    run_test()

[peterklingelhofer]

The issue was that the Trino workers were not getting the latest catalog config via helmfile apply, only the Trino coordinator was. I checked the deployment and noticed the coordinator had been updated recently via helm but the workers were old. I deleted my deployment (so the workers would be removed) and did a completely fresh deployment, and I no longer see the missing Glue table error.