Re: POC PATCH: copy from ... exceptions to: (was Re: VLDB Features)

Hi!

Improved my patch by adding block subtransactions.
The block size is determined by the REPLAY_BUFFER_SIZE parameter.
I used the idea of a buffer for accumulating tuples in it.
If we read REPLAY_BUFFER_SIZE rows without errors, the subtransaction will
be committed.
If we find an error, the subtransaction will rollback and the buffer will
be replayed containing tuples.

In the patch REPLAY_BUFFER_SIZE equals 3, but it can be changed to any
other number (for example 1000).
There is an idea to create a GUC parameter for it.
Also maybe create a GUC parameter for the number of occurring WARNINGS by
rows with errors.

For CIM_MULTI and CIM_MULTI_CONDITIONAL cases the buffer is not needed.
It is needed for the CIM_SINGLE case.

Tests:

-- CIM_MULTI case
CREATE TABLE check_ign_err (n int, m int, k int);
COPY check_ign_err FROM STDIN WITH IGNORE_ERRORS;
WARNING:  COPY check_ign_err, line 2: "2 2 2 2"
WARNING:  COPY check_ign_err, line 3: "3 3"
WARNING:  COPY check_ign_err, line 4, column n: "a"
WARNING:  COPY check_ign_err, line 5, column m: "b"
WARNING:  COPY check_ign_err, line 6, column n: ""
1 1 1
2 2 2 2
3 3
a 4 4
5 b b

7 7 7
\.
SELECT * FROM check_ign_err;
WARNING:  COPY check_ign_err, line 2: "2 2 2 2"
WARNING:  COPY check_ign_err, line 3: "3 3"
WARNING:  COPY check_ign_err, line 4, column n: "a"
WARNING:  COPY check_ign_err, line 5, column m: "b"
WARNING:  COPY check_ign_err, line 6, column n: ""
 n | m | k
---+---+---
 1 | 1 | 1
 7 | 7 | 7
(2 rows)

##################################################


-- CIM_SINGLE case
-- BEFORE row trigger
CREATE TABLE trig_test(n int, m int);
CREATE FUNCTION fn_trig_before () RETURNS TRIGGER AS '
  BEGIN
    INSERT INTO trig_test VALUES(NEW.n, NEW.m);
    RETURN NEW;
  END;
' LANGUAGE plpgsql;
CREATE TRIGGER trig_before BEFORE INSERT ON check_ign_err
FOR EACH ROW EXECUTE PROCEDURE fn_trig_before();
COPY check_ign_err FROM STDIN WITH IGNORE_ERRORS;
WARNING:  COPY check_ign_err, line 2: "2 2 2 2"
WARNING:  COPY check_ign_err, line 3: "3 3"
WARNING:  COPY check_ign_err, line 4, column n: "a"
WARNING:  COPY check_ign_err, line 5, column m: "b"
WARNING:  COPY check_ign_err, line 6, column n: ""
1 1 1
2 2 2 2
3 3
a 4 4
5 b b

7 7 7
\.
SELECT * FROM check_ign_err;
 n | m | k
---+---+---
 1 | 1 | 1
 7 | 7 | 7
(2 rows)

##################################################


-- INSTEAD OF row trigger
CREATE VIEW check_ign_err_view AS SELECT * FROM check_ign_err;
CREATE FUNCTION fn_trig_instead_of () RETURNS TRIGGER AS '
  BEGIN
    INSERT INTO check_ign_err VALUES(NEW.n, NEW.m, NEW.k);
    RETURN NEW;
  END;
' LANGUAGE plpgsql;
CREATE TRIGGER trig_instead_of INSTEAD OF INSERT ON check_ign_err_view
FOR EACH ROW EXECUTE PROCEDURE fn_trig_instead_of();
COPY check_ign_err_view FROM STDIN WITH IGNORE_ERRORS;
WARNING:  COPY check_ign_err, line 2: "2 2 2 2"
WARNING:  COPY check_ign_err, line 3: "3 3"
WARNING:  COPY check_ign_err, line 4, column n: "a"
WARNING:  COPY check_ign_err, line 5, column m: "b"
WARNING:  COPY check_ign_err, line 6, column n: ""
SELECT * FROM check_ign_err;
1 1 1
2 2 2 2
3 3
a 4 4
5 b b

7 7 7
\.
SELECT * FROM check_ign_err_view;
 n | m | k
---+---+---
 1 | 1 | 1
 7 | 7 | 7
(2 rows)

##################################################

-- foreign table case in postgres_fdw extension

##################################################

-- volatile function in WHERE clause
COPY check_ign_err FROM STDIN WITH IGNORE_ERRORS
  WHERE n = floor(random()*(1-1+1))+1; /* find values equal 1 */
WARNING:  COPY check_ign_err, line 2: "2 2 2 2"
WARNING:  COPY check_ign_err, line 3: "3 3"
WARNING:  COPY check_ign_err, line 4, column n: "a"
WARNING:  COPY check_ign_err, line 5, column m: "b"
WARNING:  COPY check_ign_err, line 6, column n: ""
SELECT * FROM check_ign_err;
1 1 1
2 2 2 2
3 3
a 4 4
5 b b

7 7 7
\.
SELECT * FROM check_ign_err;
 n | m | k
---+---+---
 1 | 1 | 1
(1 row)

##################################################

-- CIM_MULTI_CONDITIONAL case
-- INSERT triggers for partition tables
CREATE TABLE check_ign_err (n int, m int, k int) PARTITION BY RANGE (n);
CREATE TABLE check_ign_err_part1 PARTITION OF check_ign_err
  FOR VALUES FROM (1) TO (4);
CREATE TABLE check_ign_err_part2 PARTITION OF check_ign_err
  FOR VALUES FROM (4) TO (8);
CREATE FUNCTION fn_trig_before_part () RETURNS TRIGGER AS '
  BEGIN
    INSERT INTO trig_test VALUES(NEW.n, NEW.m);
    RETURN NEW;
  END;
' LANGUAGE plpgsql;
CREATE TRIGGER trig_before_part BEFORE INSERT ON check_ign_err
FOR EACH ROW EXECUTE PROCEDURE fn_trig_before_part();
COPY check_ign_err FROM STDIN WITH IGNORE_ERRORS;
WARNING:  COPY check_ign_err, line 2: "2 2 2 2"
WARNING:  COPY check_ign_err, line 3: "3 3"
WARNING:  COPY check_ign_err, line 4, column n: "a"
WARNING:  COPY check_ign_err, line 5, column m: "b"
WARNING:  COPY check_ign_err, line 6, column n: ""
SELECT * FROM check_ign_err;
1 1 1
2 2 2 2
3 3
a 4 4
5 b b

7 7 7
\.
 n | m | k
---+---+---
 1 | 1 | 1
 7 | 7 | 7
(2 rows)

Thanks for feedback.
Regards, Damir