How to use common table expressions to do efficient data cleanup in partitioned PostgreSQL tables

There are some tricky challenges when cleaning up data within partitioned tables in PostgreSQL The biggest one being identifying specific data for deletion. This requires a sophisticated approach. 

Common table expressions (CTEs) are our go-to for handling this. They provide a structured mechanism that can span the partition hierarchy and pinpoint relevant data. This is essential, as that data might be distributed all throughout the partitioned tables.

Traditional DELETE operations can be resource-intensive, but you can get around this by using optimization techniques such as batch processing and leveraging indexes in WHERE conditions. To maintain consistency across partitions you need to emphasize the importance of atomic transactions. Well-designed CTEs will also help uphold database integrity.

We wanted to remove a substantial volume of data from numerous partitioned tables. But since each partitioned table was close to a terabyte in size, using an immediate data deletion strategy within the partitions would be careless. Potential restoration complications, especially during validation errors, were also daunting. Additionally, devising a precise logic to pinpoint the correct data for deletion introduced an extra layer of complexity.

We needed a thorough cleanup while avoiding any irreversible actions. Here’s what we came up with and how we implemented it. Use case

We structured our table with weekly partitions and housed Entity event overviews in an append-only format. These overviews functioned as consolidated snapshots of past events for a specific entity, with each version distinguished by the stateid column. This is what the parent table’s structure looked like:

        CREATE TABLE purple.entityevent (
  entityeventid BIGINT PRIMARY KEY,
  entityid BIGINT,
 stateid INTEGER,
 eventoverview VARCHAR
);
      

However, Postgres has some limitations that made our job more challenging: anytime we tried to update the same event overview, the updates triggered the vacuum and analysis processes. These presented complications for our sizable and high-throughput table. So instead of modifying existing overviews, we read from partitions to retrieve the latest event overview using eventId and stateId indexes. During write operations, we fetched the last overview, calculated the final version, and stored it in the latest weekly partition by inheritance. Here is what the child tables looked like:

        CREATE TABLE purple.entityevent_20210701_20210731 (
  entityeventid BIGINT PRIMARY KEY,
  entityid BIGINT,
 stateid INTEGER,
 eventoverview VARCHAR
);

ALTER TABLE purple.entityevent_20210701_20210731 inherit purple.entityevent;
      

This approach ensured that all prior iterations would be obsolete and never used again once we generated a new version. But when we saw that the sheer volume of data in those partitions had surpassed practical limits, we had to switch our focus to efficiency. This prompted us to streamline the database by eliminating redundant and outdated records.

Approach

The new challenge was cleaning up data on nearly a hundred partitions. We devised an automated approach for this, knowing we could trust its logic to identify deletable data. This step would also be reversible if the validation raised red flags. By splitting the entire task into multiple phases, it would follow these concepts/processes in no particular order:

• Identifying deletable data

• Performance considerations

• Maintaining database consistency

• Testing and validation

Let’s go through each of them:

Identifying deletable data Using common table expressions (CTEs) was our best bet here, because all the tables we wanted to clean up had dependencies on other tables to help us identify deletable data. 

CTEs provide a powerful tool for creating temporary result sets within a query, so we can construct a recursive query to traverse the partitions and identify the data requiring deletion. In addition to that, CTEs play a pivotal role in streamlining the identification and filtering of deletable data, especially when dealing with complex relationships across multiple partitioned tables.

When trying to delete data from such tables, using CTEs becomes an indispensable strategy. These temporary result sets provide a concise and organised way to articulate the logic required for identifying data for deletion.

Here is an example of two CTEs in an ecommerce platform with tables for customers, orders, and products. The goal is to retrieve a report showing each customer’s top-selling products based on the total amount spent:

        -- Sample tables for customers, orders, and products
CREATE TABLE customers (
  customer_id SERIAL PRIMARY KEY,
  customer_name VARCHAR(50)
);

CREATE TABLE products (
  product_id SERIAL PRIMARY KEY,
  product_name VARCHAR(50),
  unit_price DECIMAL
);

CREATE TABLE orders (
  order_id SERIAL PRIMARY KEY,
  customer_id INT REFERENCES customers(customer_id),
  product_id INT REFERENCES products(product_id),
  quantity INT,
  total_amount DECIMAL
);

-- Sample data
INSERT INTO customers (customer_name) VALUES
('Customer A'),
('Customer B');

INSERT INTO products (product_name, unit_price) VALUES
('Product X', 10),
('Product Y', 15),
('Product Z', 20);

INSERT INTO orders (customer_id, product_id, quantity, total_amount) VALUES
(1, 1, 3, 30),
(1, 2, 2, 30),
(2, 2, 5, 75),
(2, 3, 1, 20);

-- Query with two CTEs
WITH CustomerTotalSpent AS (
  SELECT
    o.customer_id,
    c.customer_name,
    SUM(o.total_amount) AS total_spent
  FROM
    orders o
    JOIN customers c ON o.customer_id = c.customer_id
  GROUP BY
    o.customer_id, c.customer_name
),

TopProductsPerCustomer AS (
  SELECT
    o.customer_id,
    c.customer_name,
    p.product_name,
    o.quantity,
    o.total_amount,
    RANK() OVER (PARTITION BY o.customer_id ORDER BY o.total_amount DESC) AS rank
  FROM
    orders o
    JOIN customers c ON o.customer_id = c.customer_id
    JOIN products p ON o.product_id = p.product_id
)

-- Main query combining results from CTEs
SELECT
  cts.customer_id,
  cts.customer_name,
  tpc.product_name,
  tpc.quantity,
  tpc.total_amount
FROM
  CustomerTotalSpent cts
  JOIN TopProductsPerCustomer tpc ON cts.customer_id = tpc.customer_id
WHERE
  tpc.rank = 1;  -- Retrieve only the top-selling product for each customer

      

Performance considerations When you’re cleaning up data within partitioned tables, it’s essential to consider performance. The sheer size of partitioned tables, often approaching terabytes, can pose a challenge regarding the performance impact of deletion operations. Optimization techniques are crucial in addressing this; they ensure the cleanup process is effective and resource-efficient. One great optimization strategy is implementing batch processing. Rather than executing a single, resource-intensive DELETE operation on the entire dataset, break the operation into smaller, manageable batches. This can significantly ease the impact on system resources. Batch processing prevents potential bottlenecks and allows for better monitoring and control over the deletion process. Leveraging the appropriate columns in WHERE clauses is another way to optimize performance. Making the most of Primary Keys and Indexes ensures that the database engine can quickly locate and process the relevant records for deletion. Maintaining database consistency Ensuring database consistency during the cleanup process is critical to avoiding potential pitfalls and maintaining data integrity. Using transactions plays a pivotal role in achieving this goal. By encapsulating the data cleanup operations within a transaction block, you can guarantee that the changes are treated as a single atomic unit of work. This means that either all cleanup operations are successfully executed, or none are. In the event of an error or interruption, the database can be rolled back to its original state, preventing partial or inconsistent modifications and safeguarding the overall consistency of the dataset. Testing and validation Before applying cleanup operations to a production database, we recommend doing  extensive testing in a controlled environment. This includes testing different scenarios, edge cases, and the impact on database performance to identify and address potential issues.

To find the most efficient way to detect deletable data using CTEs, we tested them thoroughly, then moved on to write a Stored Procedure to use for multiple partitions.

        'WITH entityIds as (
    select distinct entityid from purple.entityevent_%s where
        entityid >= %s::BIGINT order by entityid limit %s
),
entityEventData (entityid, maxstateid) as (
    select entityid, max(stateid)  from purple.entityevent_%s
        join entityIds using (entityid) group by entityid
),
entityEventIdToKeep as (
     select entityeventid from purple.entityevent_%s join entityEventData 
     	using (entityid) where stateid = maxstateid
) 
insert into purple.entityevent_%s_v2 
select ee.* from purple.entityevent_%s ee 
where ee.entityeventid in (
	select entityeventid from entityEventIdToKeep
)'
      

The idea was to make the stored procedure with the tasks (steps):

• open a transaction

• create an empty partition (new)

• move all the relevant data to this new partition in customizable batch size

• adjust the pointer to move to the next batch

• iterate till it moves all relevant data.

Once the stored procedure completes these steps, it:

• starts a loop to get a lock with the lock_timeout parameter on the session level (this lock was sufficient in our use case because the our application does not write to historical partitions)

• removes the old partition from inheritance, attaches the new partition

• renames the old partition with a suffix

• renames the new partition as the old one

• analyzes the newly created partition and ends the transaction.

During all these steps, you can also print required metrics wherever necessary. Here is what the complete stored procedure looks like:

        CREATE OR REPLACE PROCEDURE cleanup_entityevent(date_string text)
LANGUAGE plpgsql SECURITY DEFINER
AS
$$
DECLARE
  rows_inserted                BIGINT := 0;
  total_rows_inserted          BIGINT := 0;
  start_time                   TIMESTAMP;
  end_time                     TIMESTAMP;
  elapsed_time                 INTERVAL;
  start_point                  BIGINT := 0;
  end_point                  	 BIGINT := 0;
  batch_size                   BIGINT := 100001;
  batch_num                    BIGINT := 0;
  attach_lock_timeout CONSTANT INT    := 1000;
  attach_retries      CONSTANT INT    := 3;
  attach_retry_sleep  CONSTANT INT    := 5; -- seconds
BEGIN

  EXECUTE format('LOCK TABLE purple.entityevent_%s IN ACCESS SHARE MODE', date_string);
  EXECUTE format('select max(entityid) from purple.entityevent_%s', date_string) into end_point;
  EXECUTE format('CREATE TABLE purple.entityevent_%s_v2 (LIKE purple.entityevent_%s INCLUDING ALL)', date_string, date_string);

  LOOP
    start_time := clock_timestamp();
    EXECUTE format(
    'WITH entityIds as (
                select distinct entityid from purple.entityevent_%s where
                    entityid >= %s::BIGINT order by entityid limit %s
            ),
            entityEventData (entityid, maxstateid) as (
                select entityid, max(stateid)  from purple.entityevent_%s
                    join entityIds using (entityid) group by entityid
            ),
            entityEventIdToKeep as (
                select entityeventid from purple.entityevent_%s 
                join entityEventData 
                    using (entityid) where stateid = maxstateid
            ) 
            insert into purple.entityevent_%s_v2 
            select ee.* from purple.entityevent_%s ee 
            where ee.entityeventid in (
                select entityeventid from entityEventIdToKeep
            )', 
        date_string, start_point, batch_size, 
            date_string, date_string, date_string, date_string
   );
    end_time := clock_timestamp();
    elapsed_time := end_time - start_time;
    GET DIAGNOSTICS rows_inserted = ROW_COUNT;
    RAISE NOTICE 'Batch % processed: , processed count : %, current point: %, time taken: % ',
      batch_num, rows_inserted, start_point, elapsed_time;
    batch_num := batch_num + 1;
    total_rows_inserted := rows_inserted + total_rows_inserted;
    EXECUTE format(
      'WITH entityIds as (
                select distinct entityid from purple.entityevent_%s where
                    entityid >= %s::BIGINT order by entityid limit %s
            )
            select max(entityid)+1 from entityIds',
        date_string,start_point,batch_size
   ) INTO start_point;

   EXIT WHEN start_point > end_point;
   PERFORM pg_sleep(1);
  END LOOP;

 FOR loop_cnt in 1..attach_retries
   LOOP
     BEGIN
       -- Add the new table as partition to the parent table

       EXECUTE format('SET local lock_timeout TO %L', attach_lock_timeout);
       EXECUTE format('ALTER TSBLE purple.entityevent_%s no inherit purple.entityevent', date_string);
       EXECUTE format('ALTER TABLE purple.entityevent_%s_v2 inherit purple.entityevent', date_string);
       EXECUTE format('ALTER TABLE purple.entityevent_%s rename to entityevent_%s_old', date_string, date_string);
       EXECUTE format('ALTER TABEL purple.entityevent_%s_v2 rename to entityevent_%s', date_string, date_string);

       -- Attaching succeeded. Exit the loop.
       EXIT;
       EXECPTION;
       WHEN lock_not_avialable THEN
         RAISE NOTICE 'Lock not available ';

         IF loop_cnt = attach_retries THEN
             RAISE NOTICE 'Attaching table failed';
         END IF;
         perform pg_sleep(attach_retry_sleep);

     END; 
   END LOOP;

  RAISE NOTICE 'Current Date and Time: %', NOW();
  EXECUTE format('ANALYZE (VERBOSE) purple purple.entityevent_%s', date_string);
  RAISE NOTICE 'Current Date and Time: %', NOW();
  RAISE NOTICE 'All Inserted completed: % rows affected in total', total_rows_inserted;   
END;
$$;
      

When the stored procedure is ready, we could call it for all the required partitions, increasing or decreasing the batch size based on our database's available resources.

To call the procedure we execute this script:

        call cleanup_entityevent(‘20210701_20210731’);
      

With this approach, the rollback plan is ready in case the validation fails. After all the required validations, if we revert, we only need to remove the new partition from inheritance and add the old one back to inheritance. If the validation is all green, we drop the old partition.

Cleaning up data within partitioned PostgreSQL tables demands a strategic blend of careful planning and the right tools. Leveraging Common Table Expressions (CTEs) is an excellent way to  seamlessly traverse partition hierarchies and precisely identifyideletable data.