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         v_sql := 'SELECT count(*)/' || in_chunks || ' FROM ' || quote_ident(in_nspname) || '.' || quote_ident(in_relname) ||
  34                 ' WHERE ((' || in_fields || ') IS NULL OR (' || in_fields || ') <> (' || in_values || '))' ||
  35                 COALESCE(' AND ' || regexp_replace(in_where_clause, E'WHERE\\y','AND', 'i'),'') || ';';
  36         SELECT INTO v_chunk_size cs FROM dblink('chunk_updates', v_sql) AS result (cs bigint);
  37
  38         IF (v_chunk_size > 0) THEN
  39
  40                 v_start_timestamp := clock_timestamp();
  41                 RAISE NOTICE '%: Starting update; chunks = % chunk_size = %', v_start_timestamp, in_chunks, v_chunk_size;
  42                 RAISE NOTICE E'Equiv\' full SQL: \n\tUPDATE %.% SET (%)=(%) WHERE %;',
  43                         quote_ident(in_nspname), quote_ident(in_relname),in_fields,in_values,
  44                         COALESCE(regexp_replace(in_where_clause, E'WHERE\\y','AND', 'i'),'');
  45
  46                 FOR chunk IN 1..(in_chunks+1) LOOP
  47
  48                         v_sql := 'UPDATE ' || quote_ident(in_nspname) || '.' || quote_ident(in_relname) || ' a ' ||
  49                                 ' SET (' || in_fields || ')=(' || in_values || ')';
  50                         IF (chunk <= in_chunks) THEN
  51                                 RAISE NOTICE E'%: Updating chunk % of % (elapsed time: %s est\' remainig time: %s)', clock_timestamp(), chunk, in_chunks,
  52                                         extract('epoch' from (clock_timestamp()-v_start_timestamp)),
  53                                         COALESCE((extract('epoch' from (clock_timestamp()-v_last_timestamp))*(in_chunks+1-chunk))::text,'?');
  54
  55                                 v_sql :=  v_sql || ' FROM (SELECT ctid FROM ' || quote_ident(in_nspname) || '.' || quote_ident(in_relname) ||
  56                                         ' WHERE ((' || in_fields || ') IS NULL OR (' || in_fields || ') <> (' || in_values || '))' ||
  57                                         COALESCE(' AND ' || regexp_replace(in_where_clause, E'WHERE\\y','AND', 'i'),'') ||
  58                                         ' ORDER BY ctid LIMIT ' || v_chunk_size || ' ) b WHERE a.ctid = b.ctid AND';
  59                         ELSE
  60                                 RAISE NOTICE E'%: Final update pass (elapsed time: %s est\' remainig time: %s)', clock_timestamp(),
  61                                         extract('epoch' from (clock_timestamp()-v_start_timestamp)),
  62                                         COALESCE((extract('epoch' from (clock_timestamp()-v_last_timestamp))*(in_chunks+1-chunk))::text,'?');
  63                                 v_sql :=  v_sql || ' WHERE';
  64                         END IF;
  65
  66                         v_sql :=  v_sql || ' ((' || in_fields || ') IS NULL OR (' || in_fields || ') <> (' || in_values || '))' ||
  67                                 COALESCE(' AND ' || regexp_replace(in_where_clause, E'WHERE\\y','AND', 'i'),'') || ';';
  68
  69                         IF (v_debug) THEN
  70                                 RAISE NOTICE '%: SQL : %', clock_timestamp(), v_sql;
  71                         END IF;
  72
  73                         v_last_timestamp := clock_timestamp();
  74                         v_update_status := dblink_exec('chunk_updates', v_sql);
  75                         v_update_status := v_update_status || ' ' || dblink_error_message('chunk_updates');
  76
  77                         v_sql := 'VACUUM ' || quote_ident(in_nspname) || '.' || quote_ident(in_relname) || ';';
  78
  79                         IF (v_debug) THEN
  80                                 RAISE NOTICE '%: SQL : %', clock_timestamp(), v_sql;
  81                         END IF;
  82
  83                         v_vacuum_status := dblink_exec('chunk_updates', v_sql);
  84                         v_vacuum_status := v_vacuum_status || ' ' || dblink_error_message('chunk_updates');
  85
  86                         RAISE NOTICE '%: Chunk % status : % / %', clock_timestamp(), chunk, v_update_status, v_vacuum_status;
  87                 END LOOP;
  88         ELSE
  89                 RAISE NOTICE '%: Nothing to do!', clock_timestamp();
  90         END IF;
  91
  92         PERFORM dblink_disconnect('chunk_updates');
  93 END;
  94 $BODY$
  95 LANGUAGE plpgsql VOLATILE;