git.8kb.co.uk Git - postgresql/pg_update_chunks/blob - pg_chunk_update.plpgsql

   1 --
   2 -- Glyn Astill 25/02/2012
   3 -- Function to break up large / full table updates using dblink and
   4 -- intermediate vacuums.
   5 --
   6
   7 SET search_path=public;
   8
   9 DROP FUNCTION IF EXISTS pg_chunk_update(in_nspname text, in_relname text, in_fields text, in_values text, in_where_clause text, in_chunks integer, in_conninfo text);
  10 CREATE OR REPLACE FUNCTION pg_chunk_update(in_nspname text, in_relname text, in_fields text, in_values text, in_where_clause text, in_chunks integer, in_conninfo text)
  11 RETURNS void AS
  12 $BODY$
  13 DECLARE
  14         v_sql text;
  15         v_update_status text;
  16         v_vacuum_status text;
  17         v_chunk_size bigint;
  18         v_debug boolean := FALSE;
  19         v_start_timestamp timestamp;
  20         v_last_timestamp timestamp;
  21 BEGIN
  22         IF EXISTS (SELECT * FROM dblink_get_connections() WHERE ARRAY['chunk_updates'] <@ dblink_get_connections) THEN
  23                 IF (dblink_is_busy('chunk_updates') = 0)  THEN
  24                         RAISE NOTICE '%: Closing non-busy dblink connection', clock_timestamp();
  25                         PERFORM dblink_disconnect('chunk_updates');
  26                 ELSE
  27                         RAISE EXCEPTION '%: Found busy dblink connection', clock_timestamp();
  28                 END IF;
  29         END IF;
  30
  31         PERFORM dblink_connect('chunk_updates', in_conninfo);
  32
  33         -- A lot of this could be switched out for format() in 9.1+
  34         v_sql := 'SELECT count(*)/' || in_chunks || ' FROM ' || quote_ident(in_nspname) || '.' || quote_ident(in_relname) ||
  35                 ' WHERE ((' || in_fields || ') IS NULL OR (' || in_fields || ') <> (' || in_values || '))' ||
  36                 COALESCE(' AND ' || regexp_replace(in_where_clause, E'WHERE\\y','AND', 'i'),'') || ';';
  37         SELECT INTO v_chunk_size cs FROM dblink('chunk_updates', v_sql) AS result (cs bigint);
  38
  39         IF (v_chunk_size > 0) THEN
  40
  41                 v_start_timestamp := clock_timestamp();
  42                 RAISE NOTICE '%: Starting update; chunks = % chunk_size = %', v_start_timestamp, in_chunks, v_chunk_size;
  43                 RAISE NOTICE E'Equiv\' full SQL: \n\tUPDATE %.% SET (%)=(%) WHERE %;',
  44                         quote_ident(in_nspname), quote_ident(in_relname),in_fields,in_values,
  45                         COALESCE(regexp_replace(in_where_clause, E'WHERE\\y','AND', 'i'),'');
  46
  47                 FOR chunk IN 1..(in_chunks+1) LOOP
  48
  49                         v_sql := 'UPDATE ' || quote_ident(in_nspname) || '.' || quote_ident(in_relname) || ' a ' ||
  50                                 ' SET (' || in_fields || ')=(' || in_values || ')';
  51                         IF (chunk <= in_chunks) THEN
  52                                 RAISE NOTICE E'%: Updating chunk % of % (elapsed time: %s est\' remainig time: %s)', clock_timestamp(), chunk, in_chunks,
  53                                         extract('epoch' from (clock_timestamp()-v_start_timestamp)),
  54                                         COALESCE((extract('epoch' from (clock_timestamp()-v_last_timestamp))*(in_chunks+1-chunk))::text,'?');
  55
  56                                 v_sql :=  v_sql || ' FROM (SELECT ctid FROM ' || quote_ident(in_nspname) || '.' || quote_ident(in_relname) ||
  57                                         ' WHERE ((' || in_fields || ') IS NULL OR (' || in_fields || ') <> (' || in_values || '))' ||
  58                                         COALESCE(' AND ' || regexp_replace(in_where_clause, E'WHERE\\y','AND', 'i'),'') ||
  59                                         ' ORDER BY ctid LIMIT ' || v_chunk_size || ' ) b WHERE a.ctid = b.ctid AND';
  60                         ELSE
  61                                 RAISE NOTICE E'%: Final update pass (elapsed time: %s est\' remainig time: %s)', clock_timestamp(),
  62                                         extract('epoch' from (clock_timestamp()-v_start_timestamp)),
  63                                         COALESCE((extract('epoch' from (clock_timestamp()-v_last_timestamp))*(in_chunks+1-chunk))::text,'?');
  64                                 v_sql :=  v_sql || ' WHERE';
  65                         END IF;
  66
  67                         v_sql :=  v_sql || ' ((' || in_fields || ') IS NULL OR (' || in_fields || ') <> (' || in_values || '))' ||
  68                                 COALESCE(' AND ' || regexp_replace(in_where_clause, E'WHERE\\y','AND', 'i'),'') || ';';
  69
  70                         IF (v_debug) THEN
  71                                 RAISE NOTICE '%: SQL : %', clock_timestamp(), v_sql;
  72                         END IF;
  73
  74                         v_last_timestamp := clock_timestamp();
  75                         v_update_status := dblink_exec('chunk_updates', v_sql);
  76                         v_update_status := v_update_status || ' ' || dblink_error_message('chunk_updates');
  77
  78                         v_sql := 'VACUUM ' || quote_ident(in_nspname) || '.' || quote_ident(in_relname) || ';';
  79
  80                         IF (v_debug) THEN
  81                                 RAISE NOTICE '%: SQL : %', clock_timestamp(), v_sql;
  82                         END IF;
  83
  84                         v_vacuum_status := dblink_exec('chunk_updates', v_sql);
  85                         v_vacuum_status := v_vacuum_status || ' ' || dblink_error_message('chunk_updates');
  86
  87                         RAISE NOTICE '%: Chunk % status : % / %', clock_timestamp(), chunk, v_update_status, v_vacuum_status;
  88                 END LOOP;
  89         ELSE
  90                 RAISE NOTICE '%: Nothing to do!', clock_timestamp();
  91         END IF;
  92
  93         PERFORM dblink_disconnect('chunk_updates');
  94 END;
  95 $BODY$
  96 LANGUAGE plpgsql VOLATILE;