Add additional tests for Continuous Aggregates

Tests include:
1. UPDATEs get logged to the invalidation log.
2. the invalidation log is not written to on ABORT.
3. custom time-partitioning functions are not supported
   (also updates error message for this case).

tsl/src/continuous_aggs/create.c

@@ -707,6 +707,11 @@ cagg_validate_query(Query *query)
 		if (ht != NULL)
 		{
 			part_dimension = hyperspace_get_open_dimension(ht->space, 0);
+			if (part_dimension->partitioning != NULL)
+				ereport(ERROR,
+						(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
+						 errmsg(
+							 "continuous aggregate do not support custom partitioning functions")));
 			caggtimebucketinfo_init(&ret,
 									ht->fd.id,
 									ht->main_table_relid,

tsl/test/expected/continuous_aggs_errors.out

@@ -453,3 +453,24 @@ select time_bucket(BIGINT '100', timec), min(location), sum(temperature),sum(hum
 from conditions
 group by 1;
 NOTICE:  adding not-null constraint to column "time_partition_col"
+-- custom time partition functions are not supported with invalidations
+CREATE FUNCTION text_part_func(TEXT) RETURNS BIGINT
+    AS $$ SELECT length($1)::BIGINT $$
+    LANGUAGE SQL IMMUTABLE;
+CREATE TABLE text_time(time TEXT);
+    SELECT create_hypertable('text_time', 'time', chunk_time_interval => 10, time_partitioning_func => 'text_part_func');
+NOTICE:  adding not-null constraint to column "time"
+   create_hypertable    
+------------------------
+ (8,public,text_time,t)
+(1 row)
+
+\set ON_ERROR_STOP 0
+CREATE VIEW text_view
+    WITH ( timescaledb.continuous, timescaledb.refresh_interval='72 hours')
+    AS SELECT time_bucket('5', text_part_func(time)), COUNT(time)
+        FROM text_time
+        GROUP BY 1;
+ERROR:  continuous aggregate do not support custom partitioning functions
+\set ON_ERROR_STOP 1
+DROP TABLE text_time CASCADE;

tsl/test/expected/continuous_aggs_watermark.out

@@ -224,34 +224,60 @@ SELECT * from _timescaledb_catalog.continuous_aggs_hypertable_invalidation_log;
 (0 rows)
 
 \c :TEST_DBNAME :ROLE_SUPERUSER
-INSERT INTO _timescaledb_catalog.continuous_aggs_invalidation_threshold VALUES (2, 10);
+INSERT INTO _timescaledb_catalog.continuous_aggs_invalidation_threshold VALUES (2, 15);
 \c :TEST_DBNAME :ROLE_DEFAULT_PERM_USER
-INSERT INTO ca_inval_test SELECT generate_series(5, 10);
+INSERT INTO ca_inval_test SELECT generate_series(5, 15);
 SELECT * FROM _timescaledb_catalog.continuous_aggs_invalidation_threshold;
  hypertable_id | watermark 
 ---------------+-----------
-             2 |        10
+             2 |        15
 (1 row)
 
 SELECT * from _timescaledb_catalog.continuous_aggs_hypertable_invalidation_log;
  hypertable_id | lowest_modified_value | greatest_modified_value 
 ---------------+-----------------------+-------------------------
-             2 |                     5 |                      10
+             2 |                     5 |                      15
 (1 row)
 
-INSERT INTO ca_inval_test SELECT generate_series(11, 20);
+INSERT INTO ca_inval_test SELECT generate_series(16, 20);
 SELECT * FROM _timescaledb_catalog.continuous_aggs_invalidation_threshold;
  hypertable_id | watermark 
 ---------------+-----------
-             2 |        10
+             2 |        15
 (1 row)
 
 SELECT * from _timescaledb_catalog.continuous_aggs_hypertable_invalidation_log;
  hypertable_id | lowest_modified_value | greatest_modified_value 
 ---------------+-----------------------+-------------------------
-             2 |                     5 |                      10
+             2 |                     5 |                      15
 (1 row)
 
+\c :TEST_DBNAME :ROLE_SUPERUSER
+TRUNCATE _timescaledb_catalog.continuous_aggs_hypertable_invalidation_log;
+\c :TEST_DBNAME :ROLE_DEFAULT_PERM_USER
+-- updates below the threshold update both the old and new values
+UPDATE ca_inval_test SET time = 5 WHERE time = 6;
+UPDATE ca_inval_test SET time = 7 WHERE time = 5;
+UPDATE ca_inval_test SET time = 17 WHERE time = 14;
+UPDATE ca_inval_test SET time = 12 WHERE time = 16;
+-- updates purely above the threshold are not logged
+UPDATE ca_inval_test SET time = 19 WHERE time = 18;
+UPDATE ca_inval_test SET time = 17 WHERE time = 19;
+SELECT * FROM _timescaledb_catalog.continuous_aggs_invalidation_threshold;
+ hypertable_id | watermark 
+---------------+-----------
+             2 |        15
+(1 row)
+
+SELECT * from _timescaledb_catalog.continuous_aggs_hypertable_invalidation_log;
+ hypertable_id | lowest_modified_value | greatest_modified_value 
+---------------+-----------------------+-------------------------
+             2 |                     5 |                       6
+             2 |                     5 |                       7
+             2 |                    14 |                      17
+             2 |                    12 |                      16
+(4 rows)
+
 DROP TABLE ca_inval_test CASCADE;
 NOTICE:  drop cascades to 2 other objects
 \c :TEST_DBNAME :ROLE_SUPERUSER
@@ -302,3 +328,25 @@ SELECT * from _timescaledb_catalog.continuous_aggs_hypertable_invalidation_log;
              4 |                     1 |                       1
 (1 row)
 
+-- aborts don't get written
+BEGIN;
+    INSERT INTO ts_continuous_test VALUES (-20, -20);
+ABORT;
+SELECT * FROM _timescaledb_catalog.continuous_aggs_invalidation_threshold;
+ hypertable_id | watermark 
+---------------+-----------
+             4 |         2
+(1 row)
+
+SELECT * from _timescaledb_catalog.continuous_aggs_hypertable_invalidation_log;
+ hypertable_id | lowest_modified_value | greatest_modified_value 
+---------------+-----------------------+-------------------------
+             4 |                     1 |                       1
+(1 row)
+
+DROP TABLE ts_continuous_test CASCADE;
+NOTICE:  drop cascades to 2 other objects
+\c :TEST_DBNAME :ROLE_SUPERUSER
+TRUNCATE _timescaledb_catalog.continuous_aggs_hypertable_invalidation_log;
+TRUNCATE _timescaledb_catalog.continuous_aggs_invalidation_threshold;
+\c :TEST_DBNAME :ROLE_DEFAULT_PERM_USER

tsl/test/sql/.gitignore

@@ -4,3 +4,4 @@
 /continuous_aggs_ddl-9.6.sql
 /continuous_aggs_ddl-10.sql
 /continuous_aggs_ddl-11.sql
+/contaggviews_ddl-10.sql

tsl/test/sql/continuous_aggs_errors.sql

@@ -443,3 +443,20 @@ as
 select time_bucket(BIGINT '100', timec), min(location), sum(temperature),sum(humidity)
 from conditions
 group by 1;
+
+-- custom time partition functions are not supported with invalidations
+CREATE FUNCTION text_part_func(TEXT) RETURNS BIGINT
+    AS $$ SELECT length($1)::BIGINT $$
+    LANGUAGE SQL IMMUTABLE;
+
+CREATE TABLE text_time(time TEXT);
+    SELECT create_hypertable('text_time', 'time', chunk_time_interval => 10, time_partitioning_func => 'text_part_func');
+
+\set ON_ERROR_STOP 0
+CREATE VIEW text_view
+    WITH ( timescaledb.continuous, timescaledb.refresh_interval='72 hours')
+    AS SELECT time_bucket('5', text_part_func(time)), COUNT(time)
+        FROM text_time
+        GROUP BY 1;
+\set ON_ERROR_STOP 1
+DROP TABLE text_time CASCADE;

tsl/test/sql/continuous_aggs_watermark.sql

@@ -110,15 +110,32 @@ SELECT * FROM _timescaledb_catalog.continuous_aggs_invalidation_threshold;
 SELECT * from _timescaledb_catalog.continuous_aggs_hypertable_invalidation_log;
 
 \c :TEST_DBNAME :ROLE_SUPERUSER
-INSERT INTO _timescaledb_catalog.continuous_aggs_invalidation_threshold VALUES (2, 10);
+INSERT INTO _timescaledb_catalog.continuous_aggs_invalidation_threshold VALUES (2, 15);
 \c :TEST_DBNAME :ROLE_DEFAULT_PERM_USER
 
-INSERT INTO ca_inval_test SELECT generate_series(5, 10);
+INSERT INTO ca_inval_test SELECT generate_series(5, 15);
 
 SELECT * FROM _timescaledb_catalog.continuous_aggs_invalidation_threshold;
 SELECT * from _timescaledb_catalog.continuous_aggs_hypertable_invalidation_log;
 
-INSERT INTO ca_inval_test SELECT generate_series(11, 20);
+INSERT INTO ca_inval_test SELECT generate_series(16, 20);
+
+SELECT * FROM _timescaledb_catalog.continuous_aggs_invalidation_threshold;
+SELECT * from _timescaledb_catalog.continuous_aggs_hypertable_invalidation_log;
+
+\c :TEST_DBNAME :ROLE_SUPERUSER
+TRUNCATE _timescaledb_catalog.continuous_aggs_hypertable_invalidation_log;
+\c :TEST_DBNAME :ROLE_DEFAULT_PERM_USER
+
+-- updates below the threshold update both the old and new values
+UPDATE ca_inval_test SET time = 5 WHERE time = 6;
+UPDATE ca_inval_test SET time = 7 WHERE time = 5;
+UPDATE ca_inval_test SET time = 17 WHERE time = 14;
+UPDATE ca_inval_test SET time = 12 WHERE time = 16;
+
+-- updates purely above the threshold are not logged
+UPDATE ca_inval_test SET time = 19 WHERE time = 18;
+UPDATE ca_inval_test SET time = 17 WHERE time = 19;
 
 SELECT * FROM _timescaledb_catalog.continuous_aggs_invalidation_threshold;
 SELECT * from _timescaledb_catalog.continuous_aggs_hypertable_invalidation_log;
@@ -152,3 +169,17 @@ INSERT INTO ts_continuous_test VALUES (1, 1);
 
 SELECT * FROM _timescaledb_catalog.continuous_aggs_invalidation_threshold;
 SELECT * from _timescaledb_catalog.continuous_aggs_hypertable_invalidation_log;
+
+-- aborts don't get written
+BEGIN;
+    INSERT INTO ts_continuous_test VALUES (-20, -20);
+ABORT;
+
+SELECT * FROM _timescaledb_catalog.continuous_aggs_invalidation_threshold;
+SELECT * from _timescaledb_catalog.continuous_aggs_hypertable_invalidation_log;
+
+DROP TABLE ts_continuous_test CASCADE;
+\c :TEST_DBNAME :ROLE_SUPERUSER
+TRUNCATE _timescaledb_catalog.continuous_aggs_hypertable_invalidation_log;
+TRUNCATE _timescaledb_catalog.continuous_aggs_invalidation_threshold;
+\c :TEST_DBNAME :ROLE_DEFAULT_PERM_USER