[gelöst] Optimierung eines Mappings und entsprechender Inserts

Hallo,
ich hab eine collectd Instanz welche Daten in eine PG Datenbank loggt. Mein Problem ist, das der Prozess der Connection fast immer bei 100% hängt.

Die Probleme die ich habe:

  1. Die timestamp Werte kommen nicht 100% gleich rein bei zusammengehörigen Werten. Dadurch benötige ich ein Epsilon um zusammengehörige Werte zu gruppieren. Das macht das Anti Join in der Funktion zu einem Bottleneck.
  2. Die Werte müssen normalisiert werden und auf die richtigen Spalten gemappt werden.
  3. Die Statements müssen häufig zusammengebastelt werden und dann für jeden Wert neu geparst und ausgeführt werden.

Mit pg_stat_statements hab ich die durchschnittliche Ausführungsdauer von 110 ms auf 69 ms gedrückt. Ein reiner Insert in eine Tabelle ohne Irgendein Mapping ist bei 0.05 ms.


SELECT calls, total_exec_time, total_exec_time/calls as avg_time, query
FROM pg_stat_statements
ORDER BY total_exec_time DESC LIMIT 1;

18194 1277731.538421991 70.22818173144944 "CALL collectd_new_insert_v6($1, $2, $3, $4, $5, $6, $7, $8, $9)"

Vielleicht bringt es was nur sekundengenaue Timestamp Werte einzutragen? Wäre dann der Antijoin schneller?

Gibt es eine Möglichkeit diese Statements dynamisch vorzubereiten und dann immer wieder auszuführen? Ich habe leider keine Möglichkeit das in der Anwendung zu machen…

Hierzu nutze ich folgende Funktion:

CREATE OR REPLACE PROCEDURE public.collectd_new_insert_v6(
	IN timestamp with time zone,
	IN character varying,
	IN character varying,
	IN character varying,
	IN character varying,
	IN character varying,
	IN character varying[],
	IN character varying[],
	IN double precision[])
LANGUAGE 'plpgsql'
AS $BODY$
DECLARE
    p_time alias for $1;
    p_host alias for $2;
    p_plugin alias for $3;
    p_plugin_inst alias for $4;
    p_type alias for $5;
    p_type_inst alias for $6;
    p_value_names alias for $7;
    -- don't use the type info; for 'StoreRates true' it's 'gauge' anyway
    -- p_type_names alias for $8;
    p_values alias for $9;

	-- working variables
	i integer;
	stmt text;
	cnt integer;
	line RECORD;
	ds_id integer;
	epsilon interval;
	tstamp_l timestamp with time zone;
	tstamp_h timestamp with time zone;
BEGIN

	----------------------------------------------------------------------
	-- create the temporary table
	----------------------------------------------------------------------
	CREATE temporary TABLE IF NOT EXISTS temp_incoming
	(
		tstamp timestamp with time zone,
		host text,
		plugin text,
		plugin_inst text,
		tbl text,
		tbl_col text,
		val double precision
	);

	----------------------------------------------------------------------
	-- Insert data
	----------------------------------------------------------------------
	i := 1;
    LOOP
        EXIT WHEN i > array_upper(p_value_names, 1);	

		SELECT * INTO line FROM impl_location(p_plugin, p_plugin_inst, p_type, p_type_inst, p_value_names[i]);

		insert into temp_incoming	(tstamp, host, 		plugin, 	plugin_inst, 			tbl, 		tbl_col, 		val) 
		VALUES 						(p_time, p_host, 	p_plugin, 	line.corr_plugin_inst,	line.tbl,	line.tbl_col, 	p_values[i]);

		-- continue the loop
        i := i + 1;
    END LOOP;
	-- do a early commit as we might not finish the rest
	COMMIT;
	
	----------------------------------------------------------------------
	-- Determine if we need to move to the tables
	----------------------------------------------------------------------
	SELECT COUNT(*) INTO cnt FROM temp_incoming;
	IF cnt >= 20 THEN
		----------------------------------------------------------------------
		-- create the tables			
		----------------------------------------------------------------------
		--RAISE INFO 'Create the plugin table and tstamp idx';
		FOR line IN
			SELECT tbl
			FROM temp_incoming
			GROUP BY tbl
			ORDER BY tbl
		LOOP
			-- Create the plugin table
			stmt = format(
				'
					CREATE TABLE IF NOT EXISTS %I
					(
						id	smallint NOT NULL,
						tstamp timestamp with time zone NOT NULL,

						FOREIGN KEY (id) REFERENCES instance (id)
					);				
				', line.tbl);
			EXECUTE stmt;

			-- Create the tstamp_id_idx
			stmt = format(
				'
					CREATE INDEX IF NOT EXISTS %s_tstamp_id_idx
					ON %I USING brin
					(tstamp,id);
				',line.tbl, line.tbl);
			EXECUTE stmt;
		END LOOP;

		----------------------------------------------------------------------
		-- Create the columns
		----------------------------------------------------------------------
		--RAISE INFO 'Create the columns';
		FOR line IN
			SELECT tbl,tbl_col
			FROM temp_incoming
			GROUP BY tbl,tbl_col
			ORDER BY tbl,tbl_col
		LOOP
			stmt = format(
				'ALTER table %I ADD COLUMN IF NOT EXISTS %I real DEFAULT NULL;',
				line.tbl, line.tbl_col);	
			EXECUTE stmt;
		END LOOP;

		----------------------------------------------------------------------
		-- Create the instances
		----------------------------------------------------------------------
		--RAISE INFO 'Create the instances';
		INSERT INTO instance (host,plugin,plugin_inst)
			SELECT host,plugin, plugin_inst
			FROM temp_incoming
			WHERE impl_instance_id(host, plugin, plugin_inst) IS NULL
			GROUP BY host,plugin, plugin_inst
			ORDER BY host,plugin, plugin_inst;	

		----------------------------------------------------------------------
		-- Move the info -- Anti Join TAOP book
		----------------------------------------------------------------------
		--RAISE INFO 'Move incoming to tables';	
		epsilon = '0.5 seconds'::interval;
		FOR line IN
			SELECT *
			FROM temp_incoming
			ORDER BY tstamp, host, plugin
		LOOP			
			SELECT impl_instance_id(line.host, line.plugin, line.plugin_inst) INTO ds_id;
			SELECT line.tstamp - epsilon INTO tstamp_l;
			SELECT line.tstamp + epsilon INTO tstamp_h;				

			stmt = format(
				'
					WITH upd AS
					(
						UPDATE %I
						SET %I=%L
						WHERE 	%L <= tstamp AND	tstamp < %L
								AND id=%L
						returning 1
					),
					ins AS (
						INSERT INTO %I (tstamp,id,%I)
						SELECT %L,%L,%L
						WHERE NOT EXISTS
						(
							SELECT 1 FROM %I
							WHERE 	%L <= tstamp AND	tstamp < %L
									AND id=%L
						)
						returning 1
					)
					select (select count(*) from upd) as updates,
						   (select count(*) from ins) as inserts;

				',	line.tbl, 
					line.tbl_col, line.val, 
					tstamp_l, tstamp_h,
					ds_id, 

					line.tbl, line.tbl_col,
					line.tstamp, ds_id, line.val, 
					line.tbl, 
					tstamp_l, tstamp_h, ds_id
				);
			EXECUTE stmt;	
		END LOOP;

		-- We are done, clear the incoming table
		DELETE FROM temp_incoming;
		COMMIT;
	END IF;
	
END;
$BODY$;

dann füge sie in einer Transaktion ein, innerhalb einer Transaktion ist now() konstant:

postgres=# select now();
              now              
-------------------------------
 2023-10-27 08:12:37.152639+00
(1 row)

postgres=# select now();
              now              
-------------------------------
 2023-10-27 08:12:38.480761+00
(1 row)

postgres=# begin;
BEGIN
postgres=*# select now();
              now              
-------------------------------
 2023-10-27 08:12:42.383953+00
(1 row)

postgres=*# select now();
              now              
-------------------------------
 2023-10-27 08:12:42.383953+00
(1 row)

postgres=*# select now();
              now              
-------------------------------
 2023-10-27 08:12:42.383953+00
(1 row)

postgres=*#

Naja, das Problem ist das die einzeln übermittelt werden inklusive jeweils einmaliger Timestamps. Die Prozedur wird für jeden Wert einzeln aufgerufen. Ab und an sind 2 Werte im Array enthalten. Deshalb ist auch now() nicht genug.

Das Problem wurde gelöst indem ich die insert Prozedur so umgeschrieben habe das sie gleich das Zeitmapping macht und die Instanzen zuordnet bzw. erzeugt. Eine zweite Prozedur schiebt die Daten dann im Cronjob in die Zieltabellen aber erzeugt ein Sql Statement das INSERT () SELECT macht. Der Select ist ein crosstab der dynamisch erzeugt wird. So vermeide ich viel Parsarbeit und bin effizenter.

EDIT: Für interessierte Leser:

Die Insert Prozedur

CREATE OR REPLACE PROCEDURE public.collectd_new_insert_v6(
	IN timestamp with time zone,
	IN character varying,
	IN character varying,
	IN character varying,
	IN character varying,
	IN character varying,
	IN character varying[],
	IN character varying[],
	IN double precision[])
LANGUAGE 'plpgsql'
AS $BODY$
DECLARE
    p_time alias for $1;
    p_host alias for $2;
    p_plugin alias for $3;
    p_plugin_inst alias for $4;
    p_type alias for $5;
    p_type_inst alias for $6;
    p_value_names alias for $7;
    -- don't use the type info; for 'StoreRates true' it's 'gauge' anyway
    -- p_type_names alias for $8;
    p_values alias for $9;

	-- working variables
	i integer;
	line RECORD;
	ds_id integer;
	epsilon interval;
	tstamp_l timestamp with time zone;
	tstamp_h timestamp with time zone;
	
	tstamp_t timestamp with time zone;
BEGIN

	epsilon = '0.5 seconds'::interval;
	SELECT p_time - epsilon INTO tstamp_l;
	SELECT p_time + epsilon INTO tstamp_h;				

	----------------------------------------------------------------------
	-- Insert data
	----------------------------------------------------------------------
	i := 1;
    LOOP
        EXIT WHEN i > array_upper(p_value_names, 1);	

		SELECT * INTO line FROM impl_location(p_plugin, p_plugin_inst, p_type, p_type_inst, p_value_names[i]);

		SELECT impl_instance_id(p_host,p_plugin,line.corr_plugin_inst) INTO ds_id;

		SELECT tstamp INTO tstamp_t
		FROM incoming
		WHERE tstamp_l <= tstamp AND tstamp < tstamp_h
		AND id = ds_id;		
		
		INSERT INTO incoming	(tstamp, id, 	tbl, 		tbl_col, 		val) 
		VALUES 					(COALESCE(tstamp_t,p_time), 
								 		ds_id,	line.tbl,	line.tbl_col, 	p_values[i]);

		-- continue the loop
        i := i + 1;
    END LOOP;
	
END;
$BODY$;

Die Move Prozedur

CREATE OR REPLACE PROCEDURE public.move_data_to_table(	)
LANGUAGE 'plpgsql'
AS $BODY$
DECLARE
 	-- working variables
	line RECORD;
	test RECORD;
	stmt text;
	stmt_ins text;
	stmt_ct text;
	stmt_cp text;
	cols RECORD;
BEGIN

	----------------------------------------------------------------------
	-- create the tables and indices
	----------------------------------------------------------------------
	FOR line IN
		SELECT DISTINCT tbl FROM incoming
	LOOP
		EXECUTE '
			SELECT 1 FROM 
				pg_tables
			WHERE 
				schemaname = $1 AND 
				tablename  = $2
			' INTO test USING 'public', line.tbl; 
			
		IF test ISNULL THEN
			RAISE INFO 'Create %s', line.tbl;
			stmt = format(
				'
					CREATE TABLE IF NOT EXISTS %I
					(
						id	smallint NOT NULL,
						tstamp timestamp with time zone NOT NULL,

						FOREIGN KEY (id) REFERENCES instance (id)
					);				
				', line.tbl);
			EXECUTE stmt;

			-- Create the tstamp_id_idx
			stmt = format(
				'
					CREATE INDEX IF NOT EXISTS %s_tstamp_id_idx
					ON %I USING brin
					(tstamp,id);
				',line.tbl, line.tbl);
			EXECUTE stmt;									
		END IF;
	END LOOP;
	
	----------------------------------------------------------------------
	-- create the coluns in the table
	----------------------------------------------------------------------
	FOR line IN
		SELECT DISTINCT tbl, tbl_col FROM incoming
		ORDER BY tbl
	LOOP
		EXECUTE 
			'	SELECT 1
				FROM information_schema.columns 
				WHERE table_name=$1 and column_name=$2;
			' INTO test USING line.tbl, line.tbl_col;
		
		IF test ISNULL THEN
			stmt = format(
				'ALTER table %I ADD COLUMN IF NOT EXISTS %I real DEFAULT NULL;',
				line.tbl, line.tbl_col);	
			EXECUTE stmt;
		END IF;
	END LOOP;
	
	----------------------------------------------------------------------
	-- Now as all tables and all columns exists, move the data
	-- table by table and not line by line into the target tables
	----------------------------------------------------------------------
	FOR line IN
		SELECT DISTINCT tbl FROM incoming
		ORDER BY tbl
	LOOP
		----------------------------------------------------------------------
		-- build the crosstab and insert statement
		----------------------------------------------------------------------
		stmt_ins = 'INSERT INTO ' || line.tbl || ' (tstamp, id ';
		stmt_ct = format('					 
			SELECT *
			FROM crosstab(
				''select tstamp, id, tbl_col, val
				from incoming
				where tbl=''%L''
				order by 1,2'',
						
				''SELECT DISTINCT tbl_col FROM incoming WHERE tbl=''%L'' ''		 
						)
			AS ct(row_name timestamp with time zone, id smallint  ', line.tbl,line.tbl); 
		
		FOR cols IN
			SELECT DISTINCT tbl_col from incoming WHERE tbl = line.tbl
		LOOP
			stmt_ct = stmt_ct || ', "' || cols.tbl_col || '" real ';
			stmt_ins = stmt_ins || ', "' || cols.tbl_col || '" ';
		END LOOP;
			stmt_ct = stmt_ct || ');';
			stmt_ins = stmt_ins || ') ';
			stmt_cp = stmt_ins || ' ' || stmt_ct;
		--RAISE INFO 'ct %', stmt_ct;		
		--RAISE INFO 'ins %', stmt_ins;		
		--RAISE INFO 'total %', stmt_cp;		
		
		----------------------------------------------------------------------
		-- Insert into the target table
		----------------------------------------------------------------------
		EXECUTE stmt_cp;
	END LOOP;
	
	DELETE FROM incoming;
END;
$BODY$;

Dieses Thema wurde automatisch 60 Tage nach der letzten Antwort geschlossen. Es sind keine neuen Antworten mehr erlaubt.