|
| 1 | +# Copyright (c) 2021-2025, Crate.io Inc. |
| 2 | +# Distributed under the terms of the AGPLv3 license, see LICENSE. |
| 3 | +import logging |
| 4 | +import typing as t |
| 5 | +import uuid |
| 6 | + |
| 7 | +import sqlalchemy as sa |
| 8 | +from sqlalchemy import CursorResult, MetaData, Table |
| 9 | +from sqlalchemy.orm import Session |
| 10 | + |
| 11 | +from cratedb_toolkit.materialized.model import MaterializedView, MaterializedViewSettings |
| 12 | +from cratedb_toolkit.model import TableAddress |
| 13 | +from cratedb_toolkit.util.database import DatabaseAdapter |
| 14 | + |
| 15 | +logger = logging.getLogger(__name__) |
| 16 | + |
| 17 | + |
| 18 | +class MaterializedViewStore: |
| 19 | + """ |
| 20 | + A wrapper around the materialized view management table. |
| 21 | + """ |
| 22 | + |
| 23 | + def __init__(self, settings: MaterializedViewSettings): |
| 24 | + self.settings = settings |
| 25 | + |
| 26 | + if self.settings.materialized_table.table is None: |
| 27 | + raise ValueError("Unable to create MaterializedViewStore without table name") |
| 28 | + |
| 29 | + logger.info( |
| 30 | + f"Connecting to database {self.settings.database.safe}, table {self.settings.materialized_table.fullname}" |
| 31 | + ) |
| 32 | + |
| 33 | + # Set up generic database adapter. |
| 34 | + self.database: DatabaseAdapter = DatabaseAdapter(dburi=self.settings.database.dburi) |
| 35 | + |
| 36 | + # Set up SQLAlchemy Core adapter for materialized view management table. |
| 37 | + metadata = MetaData(schema=self.settings.materialized_table.schema) |
| 38 | + self.table = Table(self.settings.materialized_table.table, metadata, autoload_with=self.database.engine) |
| 39 | + |
| 40 | + def create(self, mview: MaterializedView, ignore: t.Optional[str] = None): |
| 41 | + """ |
| 42 | + Create a new materialized view, and return its identifier. |
| 43 | +
|
| 44 | + TODO: Generalize, see `RetentionPolicyStore`. |
| 45 | + """ |
| 46 | + |
| 47 | + # TODO: Sanity check, whether target table already exists? |
| 48 | + |
| 49 | + ignore = ignore or "" |
| 50 | + |
| 51 | + # Sanity checks. |
| 52 | + if mview.table_schema is None: |
| 53 | + raise ValueError("Table schema needs to be defined") |
| 54 | + if mview.table_name is None: |
| 55 | + raise ValueError("Table name needs to be defined") |
| 56 | + if self.exists(mview): |
| 57 | + if not ignore.startswith("DuplicateKey"): |
| 58 | + raise ValueError(f"Materialized view '{mview.table_schema}.{mview.table_name}' already exists") |
| 59 | + |
| 60 | + table = self.table |
| 61 | + # TODO: Add UUID as converter to CrateDB driver? |
| 62 | + identifier = str(uuid.uuid4()) |
| 63 | + data = mview.to_storage_dict(identifier=identifier) |
| 64 | + insertable = sa.insert(table).values(**data).returning(table.c.id) |
| 65 | + cursor = self.execute(insertable) |
| 66 | + identifier = cursor.one()[0] |
| 67 | + self.synchronize() |
| 68 | + return identifier |
| 69 | + |
| 70 | + def retrieve(self): |
| 71 | + """ |
| 72 | + Retrieve all records from database table. |
| 73 | +
|
| 74 | + TODO: Add filtering capabilities. |
| 75 | + TODO: Generalize, see `RetentionPolicyStore`. |
| 76 | + """ |
| 77 | + |
| 78 | + # Run SELECT statement, and return result. |
| 79 | + selectable = sa.select(self.table) |
| 80 | + records = self.query(selectable) |
| 81 | + return records |
| 82 | + |
| 83 | + def get_by_table(self, table_address: TableAddress) -> MaterializedView: |
| 84 | + """ |
| 85 | + Retrieve effective policies to process, by strategy and tags. |
| 86 | + """ |
| 87 | + table = self.table |
| 88 | + selectable = sa.select(table).where( |
| 89 | + table.c.table_schema == table_address.schema, |
| 90 | + table.c.table_name == table_address.table, |
| 91 | + ) |
| 92 | + logger.info(f"View definition DQL: {selectable}") |
| 93 | + try: |
| 94 | + record = self.query(selectable)[0] |
| 95 | + except IndexError as ex: |
| 96 | + raise KeyError( |
| 97 | + f"Synthetic materialized table definition does not exist: {table_address.schema}.{table_address.table}" |
| 98 | + ) from ex |
| 99 | + mview = MaterializedView.from_record(record) |
| 100 | + return mview |
| 101 | + |
| 102 | + def delete(self, identifier: str) -> int: |
| 103 | + """ |
| 104 | + Delete materialized view by identifier. |
| 105 | +
|
| 106 | + TODO: Generalize, see `RetentionPolicyStore`. |
| 107 | + """ |
| 108 | + table = self.table |
| 109 | + constraint = table.c.id == identifier |
| 110 | + deletable = sa.delete(table).where(constraint) |
| 111 | + result = self.execute(deletable) |
| 112 | + self.synchronize() |
| 113 | + if result.rowcount == 0: |
| 114 | + logger.warning(f"Materialized view not found with id: {identifier}") |
| 115 | + return result.rowcount |
| 116 | + |
| 117 | + def execute(self, statement) -> CursorResult: |
| 118 | + """ |
| 119 | + Execute SQL statement, and return result object. |
| 120 | +
|
| 121 | + TODO: Generalize, see `RetentionPolicyStore`. |
| 122 | + """ |
| 123 | + with Session(self.database.engine) as session: |
| 124 | + result = session.execute(statement) |
| 125 | + session.commit() |
| 126 | + return result |
| 127 | + |
| 128 | + def query(self, statement) -> t.List[t.Dict]: |
| 129 | + """ |
| 130 | + Execute SQL statement, fetch result rows, and return them converted to dictionaries. |
| 131 | +
|
| 132 | + TODO: Generalize, see `RetentionPolicyStore`. |
| 133 | + """ |
| 134 | + cursor = self.execute(statement) |
| 135 | + rows = cursor.mappings().fetchall() |
| 136 | + records = [dict(row.items()) for row in rows] |
| 137 | + return records |
| 138 | + |
| 139 | + def exists(self, mview: MaterializedView): |
| 140 | + """ |
| 141 | + Check if retention policy for specific table already exists. |
| 142 | +
|
| 143 | + TODO: Generalize, see `RetentionPolicyStore`. |
| 144 | + """ |
| 145 | + table = self.table |
| 146 | + selectable = sa.select(table).where( |
| 147 | + table.c.table_schema == mview.table_schema, |
| 148 | + table.c.table_name == mview.table_name, |
| 149 | + ) |
| 150 | + result = self.query(selectable) |
| 151 | + return bool(result) |
| 152 | + |
| 153 | + def synchronize(self): |
| 154 | + """ |
| 155 | + Synchronize data by issuing `REFRESH TABLE` statement. |
| 156 | + """ |
| 157 | + sql = f"REFRESH TABLE {self.settings.materialized_table.fullname};" |
| 158 | + self.database.run_sql(sql) |
0 commit comments