Merge pull request #352 from django-daiquiri/fix-max-records-trim

Improve performance of the table trim to max_records

Author: kimakan

daiquiri/core/adapter/database/base.py

@@ -14,8 +14,8 @@ def __init__(self, database_key, database_config):
     def connection(self):
         return connections[self.database_key]
 
-    def execute(self, sql):
-        return self.connection().cursor().execute(sql)
+    def execute(self, sql, args=None):
+        return self.connection().cursor().execute(sql, args)
 
     def fetchone(self, sql, args=None, as_dict=False):
         cursor = self.connection().cursor()

daiquiri/core/adapter/database/postgres.py

@@ -369,14 +369,17 @@ def trim_table_rows(self, schema_name, table_name, max_records):
         if not self.table_exists(schema_name, table_name):
             return
 
-        query = (
-            'DELETE FROM '
-            + f'{self.escape_identifier(schema_name)}.{self.escape_identifier(table_name)} '
-            + 'WHERE ctid NOT IN (SELECT ctid FROM '
-            + f'{self.escape_identifier(schema_name)}.{self.escape_identifier(table_name)} '
-            + f'LIMIT {max_records} );'
-        )
-        self.execute(query)
+        user_table = f'{self.escape_identifier(schema_name)}.{self.escape_identifier(table_name)}'
+        query = f"""DELETE FROM {user_table} as t
+        USING (
+            SELECT ctid
+            FROM {user_table}
+            ORDER BY ctid
+            OFFSET %s
+        ) as d
+        WHERE t.ctid = d.ctid;
+        """
+        self.execute(query, args=[max_records,])
 
     def table_exists(self, schema_name, table_name):
         check_query = (

daiquiri/query/tasks.py

@@ -28,6 +28,7 @@ def on_failure(self, exc, task_id, args, kwargs, einfo):
 
         # log raised exception
         logger.error('run_query %s raised an exception (%s)', job_id, exc)
+        logger.debug('run_query %s failed with an error: %s', job_id, einfo)
 
         # set phase and error_summary of the crashed job
         job = QueryJob.objects.get(pk=job_id)