Predicate locking in rum index

Makefile

@@ -35,6 +35,14 @@ endif
 wal-check: temp-install
 	$(prove_check)
 
+check: isolationcheck
+
+submake-isolation:
+	$(MAKE) -C $(top_builddir)/src/test/isolation all
+submake-rum:
+	$(MAKE) -C $(top_builddir)/contrib/rum
+
+
 all: rum--1.1.sql
 
 #9.6 requires 1.1 file but 10.0 could live with 1.0 + 1.0-1.1 files
@@ -48,3 +56,9 @@ install: installincludes
 
 installincludes:
 	$(INSTALL_DATA) $(addprefix $(srcdir)/, $(INCLUDES)) '$(DESTDIR)$(includedir_server)/'
+
+ISOLATIONCHECKS= predicate-rum predicate-rum-2
+isolationcheck: | submake-isolation submake-rum temp-install
+	$(pg_isolation_regress_check) \
+	    --temp-config $(top_srcdir)/contrib/rum/logical.conf \
+	    $(ISOLATIONCHECKS)

logical.conf

@@ -0,0 +1,2 @@
+wal_level = logical
+max_replication_slots = 4

specs/predicate-rum-2.spec

@@ -0,0 +1,59 @@
+# Test for page level predicate locking in rum
+#
+# Test to check reduced false positives
+#
+# Queries are written in such a way that an index scan(from one transaction) and an index insert(from another transaction) will try to access different parts(sub-tree) of the index.
+
+setup
+{
+ CREATE EXTENSION rum;
+
+ CREATE TABLE rum_tbl (id serial, tsv tsvector);
+
+ CREATE TABLE text_table (id1 serial, t text[]);
+
+ SELECT SETSEED(0.5);
+
+ INSERT INTO text_table(t) SELECT array[chr(i) || chr(j)] FROM generate_series(65,90) i,
+ generate_series(65,90) j ; 
+
+ INSERT INTO rum_tbl(tsv) SELECT to_tsvector('simple', t[1] ) FROM  text_table; 
+
+ DO $$
+ BEGIN
+ FOR j in 1..10 LOOP
+ UPDATE rum_tbl SET tsv = tsv || q.t1 FROM (SELECT id1,to_tsvector('simple', t[1] )
+ as t1 FROM text_table) as q WHERE id = (random()*q.id1)::integer;
+ END LOOP;
+ END;
+ $$;
+
+ CREATE INDEX rum_tbl_idx ON rum_tbl USING rum (tsv rum_tsvector_ops);
+}
+
+teardown
+{
+ DROP TABLE text_table;
+ DROP TABLE rum_tbl;
+ DROP EXTENSION rum;
+}
+
+session "s1"
+setup		{ 
+		  BEGIN ISOLATION LEVEL SERIALIZABLE;
+		  set enable_seqscan=off;
+		}
+step "rxy1"	{ SELECT id, tsv FROM rum_tbl WHERE tsv @@ 'hx'; }
+step "wx1"	{ INSERT INTO rum_tbl(tsv) values('ab'); }
+step "c1"	{ COMMIT; }
+
+session "s2"
+setup		{ 
+		  BEGIN ISOLATION LEVEL SERIALIZABLE;
+		  set enable_seqscan=off;
+		}
+
+step "rxy2"	{ SELECT id, tsv FROM rum_tbl WHERE tsv @@ 'qh'; }
+step "wy2"	{ INSERT INTO rum_tbl(tsv) values('xz'); }
+step "c2"	{ COMMIT; }
+

specs/predicate-rum.spec

@@ -0,0 +1,58 @@
+# Test for page level predicate locking in rum
+#
+# Test to verify serialization failures
+#
+# Queries are written in such a way that an index scan(from one transaction) and an index insert(from another transaction) will try to access the same part(sub-tree) of the index.
+
+setup
+{
+ CREATE EXTENSION rum;
+
+ CREATE TABLE rum_tbl (id serial, tsv tsvector);
+
+ CREATE TABLE text_table (id1 serial, t text[]);
+
+ SELECT SETSEED(0.5);
+
+ INSERT INTO text_table(t) SELECT array[chr(i) || chr(j)] FROM generate_series(65,90) i,
+ generate_series(65,90) j ; 
+
+ INSERT INTO rum_tbl(tsv) SELECT to_tsvector('simple', t[1] ) FROM  text_table; 
+
+ DO $$
+ BEGIN
+ FOR j in 1..10 LOOP
+ UPDATE rum_tbl SET tsv = tsv || q.t1 FROM (SELECT id1,to_tsvector('simple', t[1] )
+ as t1 FROM text_table) as q WHERE id = (random()*q.id1)::integer;
+ END LOOP;
+ END;
+ $$;
+
+ CREATE INDEX rum_tbl_idx ON rum_tbl USING rum (tsv rum_tsvector_ops);
+}
+
+teardown
+{
+ DROP TABLE text_table;
+ DROP TABLE rum_tbl;
+ DROP EXTENSION rum;
+}
+
+session "s1"
+setup		{ 
+		  BEGIN ISOLATION LEVEL SERIALIZABLE;
+		  set enable_seqscan=off;
+		}
+step "rxy1"	{ SELECT id, tsv FROM rum_tbl WHERE tsv @@ 'hx'; }
+step "wx1"	{ INSERT INTO rum_tbl(tsv) values('qh'); }
+step "c1"	{ COMMIT; }
+
+session "s2"
+setup		{ 
+		  BEGIN ISOLATION LEVEL SERIALIZABLE;
+		  set enable_seqscan=off;
+		}
+
+step "rxy2"	{ SELECT id, tsv FROM rum_tbl WHERE tsv @@ 'qh'; }
+step "wy2"	{ INSERT INTO rum_tbl(tsv) values('hx'); }
+step "c2"	{ COMMIT; }

src/rumbtree.c

@@ -15,6 +15,7 @@
 
 #include "access/generic_xlog.h"
 #include "miscadmin.h"
+#include "storage/predicate.h"
 
 #include "rum.h"
 
@@ -485,6 +486,14 @@ rumInsertValue(Relation index, RumBtree btree, RumBtreeStack * stack,
 				btree->fillRoot(btree, stack->buffer, lbuffer, rbuffer,
 								page, lpage, rpage);
 
+				PredicateLockPageSplit(btree->index,
+							BufferGetBlockNumber(stack->buffer),
+							BufferGetBlockNumber(lbuffer));
+
+				PredicateLockPageSplit(btree->index,
+							BufferGetBlockNumber(stack->buffer),
+							BufferGetBlockNumber(rbuffer));
+
 				if (btree->rumstate->isBuild)
 				{
 					START_CRIT_SECTION();
@@ -548,6 +557,10 @@ rumInsertValue(Relation index, RumBtree btree, RumBtreeStack * stack,
 				RumPageGetOpaque(rpage)->leftlink = BufferGetBlockNumber(stack->buffer);
 				RumPageGetOpaque(newlpage)->rightlink = BufferGetBlockNumber(rbuffer);
 
+				PredicateLockPageSplit(btree->index,
+						BufferGetBlockNumber(stack->buffer),
+						BufferGetBlockNumber(rbuffer));
+
 				/*
 				 * it's safe because we don't have right-to-left walking
 				 * with locking bth pages except vacuum. But vacuum will