Add support for materializing ModelSQL based on a table query

trytond/CHANGELOG

@@ -1,3 +1,4 @@
+* Add support for materializing ModelSQL based on a table query
 * Add an option to trytond-console to run a script file
 * Store only immutable structure in MemoryCache
 * Replace ``clean_days`` with ``log_size`` in the cron configuration

trytond/doc/topics/configuration.rst

@@ -516,6 +516,27 @@ For example::
     account.invoice.line = acc_inv_line
     account.invoice.tax = acc_inv_tax
 
+.. _config-table_query_materialized:
+
+table_query_materialized
+------------------------
+
+This section allow to define which :class:`~trytond.model.ModelSQL` with
+:meth:`~trytond.model.ModelSQL.table_query` must be materialized and at which
+interval they must be refreshed.
+
+The interval value is expressed in seconds.
+
+For example::
+
+    [table_query_materialized]
+    account.account.party = 1440
+
+.. note::
+
+   Only the :meth:`~trytond.model.ModelSQL.table_query` that does not depend on
+   the :attr:`~trytond.transaction.Transaction.context`.
+
 .. _config-ssl:
 
 ssl

trytond/trytond/backend/database.py

@@ -185,6 +185,19 @@ class DatabaseInterface(object):
     def has_channel(self):
         return False
 
+    @classmethod
+    def has_materialized_views(cls):
+        return False
+
+    def create_materialized_view(self, connection, view_name, query):
+        raise NotImplementedError
+
+    def drop_materialized_view(self, connection):
+        raise NotImplementedError
+
+    def refresh_materialized_view(self, connection, view_name):
+        raise NotImplementedError
+
     def sql_type(self, type_):
         pass
 

trytond/trytond/backend/postgresql/database.py

@@ -535,6 +535,16 @@ class Database(DatabaseInterface):
             if cursor.rowcount:
                 return schema
 
+    def get_view_schema(self, connection, view_name):
+        cursor = connection.cursor()
+        for schema in self.search_path:
+            cursor.execute('SELECT 1 '
+                'FROM pg_matviews '
+                'WHERE matviewname = %s AND schemaname = %s',
+                (view_name, schema))
+            if cursor.rowcount:
+                return schema
+
     @property
     def current_user(self):
         if self._current_user is None:
@@ -860,6 +870,36 @@ class Database(DatabaseInterface):
     def json_contains(self, column, json):
         return JSONContains(Cast(column, 'jsonb'), Cast(json, 'jsonb'))
 
+    @classmethod
+    def has_materialized_views(cls):
+        return True
+
+    def create_materialized_view(self, connection, view_name, query):
+        cursor = connection.cursor()
+        cursor.execute(
+            SQL(
+                "CREATE MATERIALIZED VIEW IF NOT EXISTS {} AS {} "
+                "WITH NO DATA"
+                ).format(Identifier(view_name), SQL(str(query))),
+            query.params)
+        cursor.execute(
+            SQL("CREATE UNIQUE INDEX ON {}(id)").format(Identifier(view_name)))
+
+    def drop_materialized_view(self, connection, view_name):
+        cursor = connection.cursor()
+        cursor.execute(
+            SQL("DROP MATERIALIZED VIEW {}").format(Identifier(view_name)))
+
+    def refresh_materialized_view(
+            self, connection, view_name, concurrently=True):
+        cursor = connection.cursor()
+        cursor.execute(
+            SQL("REFRESH MATERIALIZED VIEW {} {}")
+            .format(
+                SQL('CONCURRENTLY' if concurrently else ''),
+                Identifier(view_name)
+                ))
+
 
 register_type(UNICODE)
 if PYDATE:

trytond/trytond/backend/postgresql/table.py

@@ -32,17 +32,28 @@ class TableHandler(TableHandlerInterface):
         cursor = transaction.connection.cursor()
 
         # Create new table if necessary
-        if not self.table_exist(self.table_name):
+        if (not (view_exists := self.view_exist(self.table_name))
+                and not self.table_exist(self.table_name)):
             cursor.execute(SQL('CREATE TABLE {} ()').format(
                     Identifier(self.table_name)))
-        self.table_schema = transaction.database.get_table_schema(
-            transaction.connection, self.table_name)
-
-        cursor.execute('SELECT tableowner = current_user FROM pg_tables '
-            'WHERE tablename = %s AND schemaname = %s',
-            (self.table_name, self.table_schema))
+        if view_exists:
+            self.table_schema = transaction.database.get_view_schema(
+                transaction.connection, self.table_name)
+            is_owner_query = (
+                'SELECT matviewowner = current_user FROM pg_matviews '
+                'WHERE matviewname = %s and schemaname = %s')
+        else:
+            self.table_schema = transaction.database.get_table_schema(
+                transaction.connection, self.table_name)
+            is_owner_query = (
+                'SELECT tableowner = current_user FROM pg_tables '
+                'WHERE tablename = %s AND schemaname = %s')
+        cursor.execute(is_owner_query, (self.table_name, self.table_schema))
         self.is_owner, = cursor.fetchone()
 
+        if view_exists:
+            return
+
         if model.__doc__ and self.is_owner:
             cursor.execute(SQL('COMMENT ON TABLE {} IS %s').format(
                         Identifier(self.table_name)),
@@ -125,6 +136,12 @@ class TableHandler(TableHandlerInterface):
         return bool(transaction.database.get_table_schema(
                 transaction.connection, table_name))
 
+    @classmethod
+    def view_exist(cls, view_name):
+        transaction = Transaction()
+        return bool(transaction.database.get_view_schema(
+                transaction.connection, view_name))
+
     @classmethod
     def table_rename(cls, old_name, new_name):
         transaction = Transaction()

trytond/trytond/backend/sqlite/table.py

@@ -56,6 +56,10 @@ class TableHandler(TableHandlerInterface):
             return False
         return True
 
+    @classmethod
+    def view_exist(cls, view_name):
+        return False
+
     @classmethod
     def table_rename(cls, old_name, new_name):
         cursor = Transaction().connection.cursor()

trytond/trytond/backend/table.py

@@ -45,6 +45,10 @@ class TableHandlerInterface(object):
     def table_exist(cls, table_name):
         raise NotImplementedError
 
+    @classmethod
+    def view_exist(cls, view_name):
+        raise NotImplementedError
+
     @classmethod
     def table_rename(cls, old_name, new_name):
         raise NotImplementedError

trytond/trytond/ir/action.py

@@ -271,6 +271,7 @@ class ActionKeyword(ModelSQL, ModelView):
         Action = pool.get('ir.action')
         Menu = pool.get('ir.ui.menu')
         ModelAccess = pool.get('ir.model.access')
+        ModelData = pool.get('ir.model.data')
         User = pool.get('res.user')
         groups = User.get_groups()
         key = (Transaction().language, groups, keyword, tuple(value))
@@ -279,6 +280,7 @@ class ActionKeyword(ModelSQL, ModelView):
             return keywords
         keywords = []
         model, record_id = value
+        Model = pool.get(model)
 
         clause = [
             ('keyword', '=', keyword),
@@ -301,6 +303,12 @@ class ActionKeyword(ModelSQL, ModelView):
         for action_keyword in action_keywords:
             type_ = action_keyword.action.type
             types[type_].append(action_keyword.action.id)
+        if (keyword == 'form_action'
+                and issubclass(Model, ModelSQL)
+                and Model._table_query_materialized()):
+            refresh_action = ModelData.get_id(
+                'ir', 'wizard_model_refresh_materialized')
+            types['ir.action.wizard'].append(refresh_action)
         for type_, action_ids in types.items():
             for value in Action.get_action_values(type_, action_ids):
                 if (type_ == 'ir.action.act_window'

trytond/trytond/ir/cron.py

@@ -94,6 +94,7 @@ class Cron(DeactivableMixin, ModelSQL, ModelView):
             ('ir.queue|clean', "Clean Task Queue"),
             ('ir.error|clean', "Clean Errors"),
             ('ir.cron.log|clean', "Clean Cron Logs"),
+            ('ir.model|refresh_materialized', "Refresh Materialized Models"),
             ], "Method", required=True, states=_states)
 
     logs = fields.One2Many('ir.cron.log', 'cron', "Logs", readonly=True)

trytond/trytond/ir/model.py

@@ -27,7 +27,8 @@ from trytond.rpc import RPC
 from trytond.tools import cursor_dict, grouped_slice, is_instance_method
 from trytond.tools.string_ import StringMatcher
 from trytond.transaction import Transaction, without_check_access
-from trytond.wizard import Button, StateAction, StateView, Wizard
+from trytond.wizard import (
+    Button, StateAction, StateTransition, StateView, Wizard)
 
 from .resource import ResourceAccessMixin
 
@@ -242,6 +243,13 @@ class Model(
     def get_name(cls, model):
         return cls.get_names().get(model, model)
 
+    @classmethod
+    def refresh_materialized(cls):
+        pool = Pool()
+        for _, Model in pool.iterobject():
+            if hasattr(Model, '_table_query_refresh'):
+                Model._table_query_refresh()
+
 
 class ModelField(
         fields.fmany2one(
@@ -1724,3 +1732,15 @@ class ModelWorkflowGraph(Report):
                         f'"{record.name}--{to}"',
                         arrowhead='normal')
                 subgraph.add_edge(edge)
+
+
+class RefreshMaterialized(Wizard):
+    __name__ = "ir.model.refresh_materialized"
+    start = StateTransition()
+
+    def transition_start(self):
+        pool = Pool()
+        Model = pool.get(Transaction().context['active_model'])
+        if hasattr(Model, '_table_query_refresh'):
+            Model._table_query_refresh(force=True)
+        return 'end'

trytond/trytond/ir/model.xml

@@ -329,5 +329,18 @@ this repository contains the full copyright notices and license terms. -->
             <field name="perm_create" eval="False"/>
             <field name="perm_delete" eval="False"/>
         </record>
+
+        <record model="ir.action.wizard" id="wizard_model_refresh_materialized">
+            <field name="name">Refresh Data</field>
+            <field name="wiz_name">ir.model.refresh_materialized</field>
+        </record>
+    </data>
+
+    <data noupdate="1">
+        <record model="ir.cron" id="cron_model_refresh_materialized">
+            <field name="method">ir.model|refresh_materialized</field>
+            <field name="interval_number" eval="1"/>
+            <field name="interval_type">hours</field>
+        </record>
     </data>
 </tryton>

trytond/trytond/ir/tryton.cfg

@@ -121,6 +121,7 @@ wizard:
     translation.TranslationReport
     ui.view.ShowView
     model.PrintModelGraph
+    model.RefreshMaterialized
     module.ModuleConfigWizard
     module.ModuleActivateUpgrade
     module.ModuleConfig

trytond/trytond/model/modelsql.py

@@ -269,7 +269,7 @@ class Index:
 def no_table_query(func):
     @wraps(func)
     def wrapper(cls, *args, **kwargs):
-        if callable(cls.table_query):
+        if cls._is_table_query():
             raise NotImplementedError("On table_query")
         return func(cls, *args, **kwargs)
     return wrapper
@@ -335,11 +335,13 @@ class ModelSQL(ModelStorage):
         cls._sql_constraints = []
         cls._sql_indexes = set()
         cls._history_sql_indexes = set()
-        if not callable(cls.table_query):
+        if (not cls._is_table_query()
+                or cls._table_query_materialized()):
             table = cls.__table__()
-            cls._sql_constraints.append(
-                ('id_positive', Check(table, table.id >= 0),
-                    'ir.msg_id_positive'))
+            if not cls._is_table_query():
+                cls._sql_constraints.append(
+                    ('id_positive', Check(table, table.id >= 0),
+                        'ir.msg_id_positive'))
             rec_name_field = getattr(cls, cls._rec_name, None)
             if (isinstance(rec_name_field, fields.Field)
                     and not hasattr(rec_name_field, 'set')):
@@ -428,7 +430,7 @@ class ModelSQL(ModelStorage):
 
     @classmethod
     def __table__(cls):
-        if callable(cls.table_query):
+        if cls._is_table_query() and not cls._table_query_materialized():
             return cls.table_query()
         else:
             return Table(cls._table)
@@ -445,10 +447,20 @@ class ModelSQL(ModelStorage):
 
     @classmethod
     def __register__(cls, module_name):
-        cursor = Transaction().connection.cursor()
+        transaction = Transaction()
+        cursor = transaction.connection.cursor()
         super().__register__(module_name)
 
-        if callable(cls.table_query):
+        if cls._is_table_query():
+            if transaction.database.has_materialized_views():
+                if backend.TableHandler.view_exist(cls._table):
+                    transaction.database.drop_materialized_view(
+                        transaction.connection, cls._table)
+                if cls._table_query_materialized():
+                    query = cls.table_query()
+                    query.columns += (CurrentTimestamp().as_('_refreshed'),)
+                    transaction.database.create_materialized_view(
+                        transaction.connection, cls._table, query)
             return
 
         pool = Pool()
@@ -567,7 +579,8 @@ class ModelSQL(ModelStorage):
                 if j != index and index < j:
                     return False
             return True
-        if not callable(cls.table_query):
+        if (not cls._is_table_query()
+                or cls._table_query_materialized()):
             table_h = cls.__table_handler__()
             indexes = filter(no_subset, cls._sql_indexes)
             table_h.set_indexes(indexes, concurrently=concurrently)
@@ -585,6 +598,36 @@ class ModelSQL(ModelStorage):
                     continue
                 history_table.add_column(field_name, field._sql_type)
 
+    @classmethod
+    def _is_table_query(cls):
+        return callable(cls.table_query)
+
+    @classmethod
+    def _table_query_materialized(cls):
+        if cls._is_table_query():
+            return config.getint(
+                'table_query_materialized', cls.__name__, default=0)
+
+    @classmethod
+    def _table_query_refresh(cls, concurrently=True, force=False):
+        table = cls.__table__()
+        transaction = Transaction()
+        if not transaction.database.has_materialized_views():
+            return
+        cursor = transaction.connection.cursor()
+        if not (interval := cls._table_query_materialized()):
+            return
+        if not force:
+            cursor.execute(*table.select(
+                    CurrentTimestamp() - table._refreshed, limit=1))
+            if row := cursor.fetchone():
+                delta, = row
+                interval = datetime.timedelta(seconds=interval)
+                if delta < interval:
+                    return
+        transaction.database.refresh_materialized_view(
+            transaction.connection, cls._table, concurrently=concurrently)
+
     @classmethod
     @without_check_access
     def __raise_integrity_error(
@@ -1126,7 +1169,7 @@ class ModelSQL(ModelStorage):
         history_limit = None
         if (cls._history
                 and transaction.context.get('_datetime')
-                and not callable(cls.table_query)):
+                and not cls._is_table_query()):
             in_max = 1
             table = cls.__table_history__()
             column = Coalesce(table.write_date, table.create_date)
@@ -1138,7 +1181,7 @@ class ModelSQL(ModelStorage):
         for f in all_fields:
             field = cls._fields.get(f)
             if field and field.sql_type():
-                if f in _TABLE_QUERY_COLUMNS and callable(cls.table_query):
+                if f in _TABLE_QUERY_COLUMNS and cls._is_table_query():
                     column = _TABLE_QUERY_COLUMNS[f]
                 else:
                     column = field.sql_column(table)
@@ -1146,7 +1189,7 @@ class ModelSQL(ModelStorage):
                 if backend.name == 'sqlite':
                     columns[f].output_name += ' [%s]' % field.sql_type().base
             elif f in {'_write', '_delete'}:
-                if not callable(cls.table_query):
+                if not cls._is_table_query():
                     rule_domain = Rule.domain_get(
                         cls.__name__, mode=f.lstrip('_'))
                     # No need to compute rule domain if it is the same as the
@@ -1168,7 +1211,7 @@ class ModelSQL(ModelStorage):
                         columns[f] = rule_expression.as_(f)
                     else:
                         columns[f] = Literal(True).as_(f)
-            elif f == '_timestamp' and not callable(cls.table_query):
+            elif f == '_timestamp' and not cls._is_table_query():
                 sql_type = fields.Char('timestamp').sql_type().base
                 columns[f] = Extract(
                     'EPOCH', Coalesce(table.write_date, table.create_date)
@@ -1641,7 +1684,7 @@ class ModelSQL(ModelStorage):
         if (mode == 'read'
                 and cls._history
                 and transaction.context.get('_datetime')
-                and not callable(cls.table_query)):
+                and not cls._is_table_query()):
             in_max = 1
             table = cls.__table_history__()
             column = Coalesce(table.write_date, table.create_date)
@@ -1824,7 +1867,7 @@ class ModelSQL(ModelStorage):
             columns.append(Column(table, '__id').as_('__id'))
 
         if eager:
-            table_query = callable(cls.table_query)
+            table_query = cls._is_table_query()
             columns += [f.sql_column(table).as_(n)
                 for n, f in sorted(cls._fields.items())
                 if not hasattr(f, 'get')

trytond/trytond/modules/__init__.py

@@ -181,6 +181,7 @@ def load_module_graph(graph, pool, update=None, lang=None, indexes=None):
     modules_todo = []
     models_to_update_history = set()
     models_with_indexes = set()
+    models_to_refresh = set()
 
     # Load also parent languages
     lang = set(lang)
@@ -257,6 +258,9 @@ def load_module_graph(graph, pool, update=None, lang=None, indexes=None):
                 for model in classes['model']:
                     if hasattr(model, '_history'):
                         models_to_update_history.add(model.__name__)
+                    if (hasattr(model, '_table_query_materialized')
+                            and model._table_query_materialized()):
+                        models_to_refresh.add(model.__name__)
                     if hasattr(model, '_update_sql_indexes'):
                         models_with_indexes.add(model.__name__)
 
@@ -327,6 +331,10 @@ def load_module_graph(graph, pool, update=None, lang=None, indexes=None):
         if update:
             cursor.execute(*ir_configuration.update(
                     [ir_configuration.series], [__series__]))
+            for model_name in models_to_refresh:
+                model = pool.get(model_name)
+                logger.info('refresh %s', model.__name__)
+                model._table_query_refresh(concurrently=False, force=True)
             if indexes or indexes is None:
                 create_indexes(concurrently=False)
             else:

trytond/trytond/tests/modelsql.py

@@ -1,6 +1,7 @@
 # This file is part of Tryton.  The COPYRIGHT file at the top level of
 # this repository contains the full copyright notices and license terms.
 from sql import Literal
+from sql.aggregate import Count, Min
 from sql.operators import Equal
 
 from trytond.model import (
@@ -267,6 +268,36 @@ class ModelLock(ModelSQL):
     __name__ = 'test.modelsql.lock'
 
 
+class ModelMaterializedViewData(ModelSQL):
+    "Model to provide test data to the materialized views"
+    __name__ = 'test.modelsql.materialized.data'
+    name = fields.Char("Name")
+
+
+class ModelMaterializedView(ModelSQL):
+    "Model to test the materialized views"
+    __name__ = 'test.modelsql.materialized'
+
+    name = fields.Char("Name")
+    count = fields.Integer("Count")
+
+    @classmethod
+    def table_query(cls):
+        pool = Pool()
+        Model = pool.get('test.modelsql.materialized.data')
+        model = Model.__table__()
+
+        return model.select(
+            Min(model.id).as_('id'),
+            model.name.as_('name'),
+            Count(model.id).as_('count'),
+            group_by=[model.name])
+
+    @classmethod
+    def _table_query_materialized(cls):
+        return 1
+
+
 def register(module):
     Pool.register(
         ModelSQLRead,
@@ -297,4 +328,6 @@ def register(module):
         ModelUnique,
         ModelExclude,
         ModelLock,
+        ModelMaterializedViewData,
+        ModelMaterializedView,
         module=module, type_='model')

trytond/trytond/tests/test_modelsql.py

@@ -1406,6 +1406,31 @@ class ModelSQLTestCase(TestCase):
                     ['OR', ('c', '=', 3), ('d.e', '=', 4)]]
                 ])
 
+    @unittest.skipUnless(
+        backend.Database.has_materialized_views(),
+        "Database backend does not support materialized views")
+    @with_transaction()
+    def test_table_query_refresh(self):
+        "Test refresh materialized table query"
+        pool = Pool()
+        Data = pool.get('test.modelsql.materialized.data')
+        MaterializedView = pool.get('test.modelsql.materialized')
+
+        Data.create([{
+                    'name': "Foo",
+                    }, {
+                    'name': "Foo",
+                    }, {
+                    'name': "Bar",
+                    }])
+        self.assertEqual(
+            MaterializedView.search([]),
+            [])
+
+        MaterializedView._table_query_refresh()
+        stats = [(s.name, s.count) for s in MaterializedView.search([])]
+        self.assertEqual(stats, [("Foo", 2), ("Bar", 1)])
+
 
 class TranslationTestCase(TestCase):
     default_language = 'fr'