CIMV prototype with basic testing

src/backend/distributed/Makefile

@@ -18,7 +18,7 @@ generated_downgrade_sql_files += $(patsubst %,$(citus_abs_srcdir)/build/sql/%,$(
 DATA_built = $(generated_sql_files)
 
 # directories with source files
-SUBDIRS = . commands connection ddl deparser executor metadata operations planner progress relay safeclib test transaction utils worker
+SUBDIRS = . commands connection ddl deparser executor metadata operations planner progress relay safeclib test transaction utils worker cimv
 
 # Symlinks are not copied over to the build directory if a separete build
 # directory is used during configure (such as on CI)

src/backend/distributed/cimv/drop.c

@@ -0,0 +1,245 @@
+#include "postgres.h"
+
+#include "catalog/dependency.h"
+#include "catalog/objectaddress.h"
+#include "catalog/pg_class.h"
+#include "distributed/listutils.h"
+#include "distributed/metadata_cache.h"
+#include "distributed/pg_cimv.h"
+#include "executor/spi.h"
+#include "nodes/parsenodes.h"
+#include "utils/builtins.h"
+#include "utils/elog.h"
+#include "utils/lsyscache.h"
+#include "utils/syscache.h"
+
+#include "distributed/cimv.h"
+
+static void DropCimv(Form_pg_cimv formCimv, DropBehavior behavior);
+static void DropDmlTriggers(Form_pg_cimv cimv);
+static void DropCronJob(Form_pg_cimv cimv);
+
+extern void
+ProcessDropMaterializedViewStmt(DropStmt *stmt)
+{
+	if (stmt->removeType != OBJECT_MATVIEW)
+	{
+		return;
+	}
+
+	List *viewNameList = NULL;
+	List *cimvObjects = NIL;
+	List *formCimvs = NIL;
+	Form_pg_cimv formCimv;
+	bool hasCimv = false;
+	bool hasNonCimv = false;
+	foreach_ptr(viewNameList, stmt->objects)
+	{
+		RangeVar *viewRangeVar = makeRangeVarFromNameList(viewNameList);
+
+		Oid relationId = RangeVarGetRelid(viewRangeVar, NoLock, true);
+
+		if (relationId == InvalidOid)
+		{
+			hasNonCimv = true;
+			continue;
+		}
+
+		formCimv = LookupCimvFromCatalog(relationId, true);
+
+		if (formCimv == NULL)
+		{
+			hasNonCimv = true;
+			continue;
+		}
+
+		hasCimv = true;
+
+		cimvObjects = lappend(cimvObjects, list_make2(makeString(get_namespace_name(
+																	 get_rel_namespace(
+																		 formCimv->
+																		 mattable))),
+													  makeString(get_rel_name(
+																	 formCimv
+																	 ->
+																	 mattable))));
+		formCimvs = lappend(formCimvs, formCimv);
+	}
+
+	if (hasCimv && hasNonCimv)
+	{
+		ereport(ERROR, (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
+						errmsg(
+							"cannot drop regular and Citus materialized views with single command")));
+	}
+
+	if (hasCimv)
+	{
+		foreach_ptr(formCimv, formCimvs)
+		{
+			DropCimv(formCimv, stmt->behavior);
+		}
+
+		stmt->removeType = OBJECT_TABLE;
+		stmt->objects = cimvObjects;
+	}
+}
+
+
+extern void
+ProcessDropViewStmt(DropStmt *stmt)
+{
+	if (stmt->removeType != OBJECT_VIEW)
+	{
+		return;
+	}
+
+	List *viewNameList = NULL;
+	foreach_ptr(viewNameList, stmt->objects)
+	{
+		RangeVar *viewRangeVar = makeRangeVarFromNameList(viewNameList);
+
+		Oid relationId = RangeVarGetRelid(viewRangeVar, NoLock, true);
+
+		if (relationId == InvalidOid)
+		{
+			continue;
+		}
+
+		Form_pg_cimv formCimv = LookupCimvFromCatalog(relationId, true);
+
+		if (formCimv != NULL)
+		{
+			ereport(ERROR, (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
+							errmsg("DROP VIEW not supported for %s",
+								   viewRangeVar->relname),
+							errhint("use DROP MATERIALIZED VIEW")));
+		}
+	}
+}
+
+
+static void
+DropCimv(Form_pg_cimv formCimv, DropBehavior behavior)
+{
+	ObjectAddress matTableAddress;
+	matTableAddress.classId = RelationRelationId;
+	matTableAddress.objectId = formCimv->mattable;
+	matTableAddress.objectSubId = 0;
+
+	ObjectAddress userViewAddress;
+	userViewAddress.classId = RelationRelationId;
+	userViewAddress.objectId = formCimv->userview;
+	userViewAddress.objectSubId = 0;
+
+	ObjectAddress refreshViewAddress;
+	refreshViewAddress.classId = RelationRelationId;
+	refreshViewAddress.objectId = formCimv->refreshview;
+	refreshViewAddress.objectSubId = 0;
+
+	ObjectAddress landingTableAddress;
+	landingTableAddress.classId = RelationRelationId;
+	landingTableAddress.objectId = formCimv->landingtable;
+	landingTableAddress.objectSubId = 0;
+
+
+	if (SPI_connect() != SPI_OK_CONNECT)
+	{
+		elog(ERROR, "SPI_connect failed");
+	}
+
+	DropCronJob(formCimv);
+	DropDmlTriggers(formCimv);
+
+	/* Lock */
+	if (OidIsValid(matTableAddress.objectId))
+	{
+		AcquireDeletionLock(&matTableAddress, 0);
+	}
+
+	if (OidIsValid(userViewAddress.objectId))
+	{
+		AcquireDeletionLock(&userViewAddress, 0);
+	}
+
+	if (OidIsValid(refreshViewAddress.objectId))
+	{
+		AcquireDeletionLock(&refreshViewAddress, 0);
+	}
+
+	if (OidIsValid(landingTableAddress.objectId))
+	{
+		AcquireDeletionLock(&landingTableAddress, 0);
+	}
+
+	/* Drop views */
+	if (OidIsValid(userViewAddress.objectId))
+	{
+		performDeletion(&userViewAddress, behavior, 0);
+	}
+
+	if (OidIsValid(refreshViewAddress.objectId))
+	{
+		performDeletion(&refreshViewAddress, behavior, 0);
+	}
+
+	/* Drop landing table */
+	if (OidIsValid(landingTableAddress.objectId))
+	{
+		performDeletion(&landingTableAddress, behavior, 0);
+	}
+
+	DeletePgCimvRow(userViewAddress.objectId);
+
+	/* Close SPI context. */
+	if (SPI_finish() != SPI_OK_FINISH)
+	{
+		elog(ERROR, "SPI_finish failed");
+	}
+}
+
+
+static void
+DropDmlTriggers(Form_pg_cimv cimv)
+{
+	StringInfoData querybuf;
+	initStringInfo(&querybuf);
+
+	appendStringInfo(&querybuf,
+					 "DROP FUNCTION %s.%s CASCADE;",
+					 quote_identifier(NameStr(cimv->triggerfnnamespace)),
+					 quote_identifier(NameStr(cimv->triggerfnname)));
+
+	if (SPI_execute(querybuf.data, false, 0) != SPI_OK_UTILITY)
+	{
+		elog(ERROR, "SPI_exec failed: %s", querybuf.data);
+	}
+
+	pfree(querybuf.data);
+}
+
+
+static void
+DropCronJob(Form_pg_cimv cimv)
+{
+	if (cimv->jobid == 0)
+	{
+		return;
+	}
+
+	StringInfoData querybuf;
+	initStringInfo(&querybuf);
+
+	appendStringInfo(&querybuf,
+					 "SELECT * FROM cron.unschedule(" INT64_FORMAT ");",
+					 cimv->jobid);
+
+	int spiResult = SPI_execute(querybuf.data, false, 0);
+
+	if (spiResult != SPI_OK_SELECT)
+	{
+		elog(ERROR, "SPI_exec failed: %s", querybuf.data);
+	}
+
+	pfree(querybuf.data);
+}

src/backend/distributed/cimv/refresh.c

@@ -0,0 +1,193 @@
+#include "postgres.h"
+
+#include "catalog/dependency.h"
+#include "catalog/objectaddress.h"
+#include "catalog/pg_class.h"
+#include "distributed/listutils.h"
+#include "distributed/metadata_cache.h"
+#include "distributed/pg_cimv.h"
+#include "executor/spi.h"
+#include "nodes/parsenodes.h"
+#include "utils/builtins.h"
+#include "utils/lsyscache.h"
+#include "utils/snapmgr.h"
+
+#include "distributed/cimv.h"
+
+static void SpiExecuteSnapshot(StringInfo query, Snapshot snapshot, int expectedResult);
+static void SpiExecute(StringInfo query, int expectedResult);
+
+bool
+ProcessRefreshMaterializedViewStmt(RefreshMatViewStmt *stmt)
+{
+	Oid relationId = RangeVarGetRelid(stmt->relation, NoLock, true);
+
+	if (relationId == InvalidOid)
+	{
+		return false;
+	}
+
+	Form_pg_cimv formCimv = LookupCimvFromCatalog(relationId, true);
+
+	if (formCimv == NULL)
+	{
+		return false;
+	}
+
+	RefreshCimv(formCimv, stmt->skipData, false);
+	return true;
+}
+
+
+void
+RefreshCimv(Form_pg_cimv formCimv, bool skipData, bool isCreate)
+{
+	StringInfoData querybuf;
+	initStringInfo(&querybuf);
+
+	if (SPI_connect_ext(SPI_OPT_NONATOMIC) != SPI_OK_CONNECT)
+	{
+		elog(ERROR, "SPI_connect failed");
+	}
+
+	const char *matTableSchemaName = get_namespace_name(get_rel_namespace(
+															formCimv->mattable));
+	const char *matTableName = get_rel_name(formCimv->mattable);
+	matTableSchemaName = quote_identifier(matTableSchemaName);
+	matTableName = quote_identifier(matTableName);
+
+	const char *landingTableSchemaName = NULL;
+	const char *landingTableName = NULL;
+	if (formCimv->landingtable)
+	{
+		landingTableSchemaName = get_namespace_name(get_rel_namespace(
+														formCimv->landingtable));
+		landingTableName = get_rel_name(formCimv->landingtable);
+		landingTableSchemaName = quote_identifier(landingTableSchemaName);
+		landingTableName = quote_identifier(landingTableName);
+	}
+
+	if (skipData)
+	{
+		if (formCimv->landingtable)
+		{
+			appendStringInfo(&querybuf,
+							 "TRUNCATE TABLE %s.%s",
+							 landingTableSchemaName,
+							 landingTableName);
+
+			SpiExecute(&querybuf, SPI_OK_UTILITY);
+			resetStringInfo(&querybuf);
+		}
+		appendStringInfo(&querybuf,
+						 "TRUNCATE TABLE  %s.%s",
+						 matTableSchemaName,
+						 matTableName);
+
+		SpiExecute(&querybuf, SPI_OK_UTILITY);
+		resetStringInfo(&querybuf);
+	}
+	else
+	{
+		const char *refreshViewSchemaName = get_namespace_name(get_rel_namespace(
+																   formCimv->refreshview));
+		const char *refreshViewName = get_rel_name(formCimv->refreshview);
+		refreshViewSchemaName = quote_identifier(refreshViewSchemaName);
+		refreshViewName = quote_identifier(refreshViewName);
+
+		if (isCreate)
+		{
+			/* better: SPI_commit_and_chain(); */
+			SPI_commit();
+			SPI_start_transaction();
+
+			/* TODO: cleanup if this fails */
+			appendStringInfo(&querybuf,
+							 "INSERT INTO %s.%s "
+							 "SELECT * FROM %s.%s",
+							 matTableSchemaName,
+							 matTableName,
+							 refreshViewSchemaName,
+							 refreshViewName);
+			if (SPI_execute(querybuf.data, false, 0) != SPI_OK_INSERT)
+			{
+				elog(ERROR, "SPI_exec failed: %s", querybuf.data);
+			}
+		}
+		else
+		{
+			Snapshot snapshot = GetLatestSnapshot();
+
+			/* TODO: DELETE only if !isCreate */
+			appendStringInfo(&querybuf,
+							 "DELETE FROM  %s.%s",
+							 matTableSchemaName,
+							 matTableName);
+			SpiExecuteSnapshot(&querybuf, snapshot, SPI_OK_DELETE);
+			resetStringInfo(&querybuf);
+
+			appendStringInfo(&querybuf,
+							 "INSERT INTO %s.%s "
+							 "SELECT * FROM %s.%s",
+							 matTableSchemaName,
+							 matTableName,
+							 refreshViewSchemaName,
+							 refreshViewName);
+			SpiExecuteSnapshot(&querybuf, snapshot, SPI_OK_INSERT);
+			resetStringInfo(&querybuf);
+
+			if (formCimv->landingtable != InvalidOid)
+			{
+				/* TODO: DELETE only if !isCreate */
+				appendStringInfo(&querybuf,
+								 "DELETE FROM  %s.%s",
+								 landingTableSchemaName,
+								 landingTableName);
+				SpiExecuteSnapshot(&querybuf, snapshot, SPI_OK_DELETE);
+				resetStringInfo(&querybuf);
+			}
+		}
+	}
+
+	/* Close SPI context. */
+	if (SPI_finish() != SPI_OK_FINISH)
+	{
+		elog(ERROR, "SPI_finish failed");
+	}
+}
+
+
+static void
+SpiExecuteSnapshot(StringInfo query, Snapshot snapshot, int expectedResult)
+{
+	SPIPlanPtr qplan = SPI_prepare(query->data, 0, NULL);
+
+	if (qplan == NULL)
+	{
+		elog(ERROR, "SPI_prepare returned %s for %s",
+			 SPI_result_code_string(SPI_result), query->data);
+	}
+
+	int spi_result = SPI_execute_snapshot(qplan,
+										  NULL, NULL,
+										  snapshot,
+										  InvalidSnapshot,
+										  false, false, 0);
+
+	if (spi_result != expectedResult)
+	{
+		elog(ERROR, "SPI_exec failed: %s", query->data);
+	}
+}
+
+
+static void
+SpiExecute(StringInfo query, int expectedResult)
+{
+	int spi_result = SPI_execute(query->data, false, 0);
+
+	if (spi_result != expectedResult)
+	{
+		elog(ERROR, "SPI_exec failed: %s", query->data);
+	}
+}

src/backend/distributed/commands/utility_hook.c

@@ -59,6 +59,7 @@
 #include "distributed/transmit.h"
 #include "distributed/version_compat.h"
 #include "distributed/worker_transaction.h"
+#include "distributed/cimv.h"
 #include "lib/stringinfo.h"
 #include "nodes/parsenodes.h"
 #include "nodes/pg_list.h"
@@ -457,6 +458,35 @@ multi_ProcessUtility(PlannedStmt *pstmt,
 		}
 	}
 
+	bool continueProcessing = true;
+	if (IsA(parsetree, CreateTableAsStmt))
+	{
+		continueProcessing = !ProcessCreateMaterializedViewStmt((const
+																 CreateTableAsStmt *)
+																parsetree, queryString,
+																pstmt);
+	}
+
+	if (IsA(parsetree, RefreshMatViewStmt))
+	{
+		continueProcessing = !ProcessRefreshMaterializedViewStmt(
+			(RefreshMatViewStmt *) parsetree);
+	}
+
+	if (IsA(parsetree, DropStmt))
+	{
+		DropStmt *dropStatement = (DropStmt *) parsetree;
+
+		if (dropStatement->removeType == OBJECT_MATVIEW)
+		{
+			ProcessDropMaterializedViewStmt(dropStatement);
+		}
+		else if (dropStatement->removeType == OBJECT_VIEW)
+		{
+			ProcessDropViewStmt(dropStatement);
+		}
+	}
+
 	if (IsDropCitusExtensionStmt(parsetree))
 	{
 		StopMaintenanceDaemon(MyDatabaseId);
@@ -483,8 +513,11 @@ multi_ProcessUtility(PlannedStmt *pstmt,
 			citusCanBeUpdatedToAvailableVersion = !InstalledAndAvailableVersionsSame();
 		}
 
-		standard_ProcessUtility(pstmt, queryString, context,
+		if (continueProcessing)
-								params, queryEnv, dest, completionTag);
+		{
+			standard_ProcessUtility(pstmt, queryString, context,
+									params, queryEnv, dest, completionTag);
+		}
 
 		/*
 		 * if we are running ALTER EXTENSION citus UPDATE (to "<version>") command, we may need

src/backend/distributed/metadata/metadata_cache.c

@@ -33,6 +33,7 @@
 #include "commands/trigger.h"
 #include "distributed/colocation_utils.h"
 #include "distributed/connection_management.h"
+#include "distributed/cimv.h"
 #include "distributed/citus_ruleutils.h"
 #include "distributed/function_utils.h"
 #include "distributed/foreign_key_relationship.h"
@@ -140,6 +141,8 @@ typedef struct MetadataCacheData
 	Oid distColocationRelationId;
 	Oid distColocationConfigurationIndexId;
 	Oid distPartitionRelationId;
+	Oid pgCimvId;
+	Oid pgCimvIndexId;
 	Oid distPartitionLogicalRelidIndexId;
 	Oid distPartitionColocationidIndexId;
 	Oid distShardLogicalRelidIndexId;
@@ -2118,6 +2121,28 @@ DistPartitionRelationId(void)
 }
 
 
+/* return oid of pg_cimv relation */
+Oid
+PgCimvId(void)
+{
+	CachedRelationLookup("pg_cimv",
+						 &MetadataCache.pgCimvId);
+
+	return MetadataCache.pgCimvId;
+}
+
+
+/* return oid of pg_cimv_pkey index */
+Oid
+PgCimvIndexId(void)
+{
+	CachedRelationLookup("pg_cimv_pkey",
+						 &MetadataCache.pgCimvIndexId);
+
+	return MetadataCache.pgCimvIndexId;
+}
+
+
 /* return oid of pg_dist_partition_logical_relid_index index */
 Oid
 DistPartitionLogicalRelidIndexId(void)

src/backend/distributed/metadata/metadata_utility.c

@@ -46,6 +46,7 @@
 #include "distributed/pg_dist_partition.h"
 #include "distributed/pg_dist_shard.h"
 #include "distributed/pg_dist_placement.h"
+#include "distributed/pg_cimv.h"
 #include "distributed/reference_table_utils.h"
 #include "distributed/relay_utility.h"
 #include "distributed/resource_lock.h"
@@ -1031,6 +1032,110 @@ InsertIntoPgDistPartition(Oid relationId, char distributionMethod,
 }
 
 
+/*
+ * InsertIntoPgCimv inserts a new tuple into pg_cimv.
+ */
+void
+InsertIntoPgCimv(Form_pg_cimv cimv)
+{
+	Datum newValues[Natts_pg_cimv];
+	bool newNulls[Natts_pg_cimv];
+
+	/* open system catalog and insert new tuple */
+	Relation pgCimv = table_open(PgCimvId(), RowExclusiveLock);
+
+	/* form new tuple for pg_cimv */
+	memset(newValues, 0, sizeof(newValues));
+	memset(newNulls, false, sizeof(newNulls));
+
+	newValues[Anum_pg_cimv_userview - 1] = ObjectIdGetDatum(cimv->userview);
+	newValues[Anum_pg_cimv_basetable - 1] = ObjectIdGetDatum(cimv->basetable);
+	newValues[Anum_pg_cimv_mattable - 1] = ObjectIdGetDatum(cimv->mattable);
+	newValues[Anum_pg_cimv_refreshview - 1] = ObjectIdGetDatum(cimv->refreshview);
+	newValues[Anum_pg_cimv_triggernamespace - 1] = NameGetDatum(
+		&cimv->triggerfnnamespace);
+	newValues[Anum_pg_cimv_triggername - 1] = NameGetDatum(&cimv->triggerfnname);
+	newValues[Anum_pg_cimv_landingtable - 1] = ObjectIdGetDatum(cimv->landingtable);
+	newValues[Anum_pg_cimv_jobid - 1] = Int64GetDatum(cimv->jobid);
+
+	HeapTuple newTuple = heap_form_tuple(RelationGetDescr(pgCimv), newValues,
+										 newNulls);
+
+	/* finally insert tuple, build index entries & register cache invalidation */
+	CatalogTupleInsert(pgCimv, newTuple);
+
+
+	CommandCounterIncrement();
+	table_close(pgCimv, NoLock);
+}
+
+
+void
+DeletePgCimvRow(Oid userViewId)
+{
+	ScanKeyData scanKey[1];
+	int scanKeyCount = 1;
+
+	Relation pgCimv = table_open(PgCimvId(), RowExclusiveLock);
+
+	ScanKeyInit(&scanKey[0], Anum_pg_cimv_userview,
+				BTEqualStrategyNumber, F_OIDEQ, ObjectIdGetDatum(userViewId));
+
+	SysScanDesc scanDescriptor = systable_beginscan(pgCimv, InvalidOid, false,
+													NULL,
+													scanKeyCount, scanKey);
+
+	HeapTuple heapTuple = systable_getnext(scanDescriptor);
+	if (!HeapTupleIsValid(heapTuple))
+	{
+		ereport(ERROR, (errmsg("could not find valid entry for cimv %d",
+							   userViewId)));
+	}
+
+	simple_heap_delete(pgCimv, &heapTuple->t_self);
+
+	systable_endscan(scanDescriptor);
+
+	/* increment the counter so that next command can see the row */
+	CommandCounterIncrement();
+
+	table_close(pgCimv, NoLock);
+}
+
+
+Form_pg_cimv
+LookupCimvFromCatalog(Oid userViewId, bool missingOk)
+{
+	ScanKeyData scanKey[1];
+	int scanKeyCount = 1;
+	Form_pg_cimv cimvForm = NULL;
+	Relation pgCimv = table_open(PgCimvId(), AccessShareLock);
+
+	ScanKeyInit(&scanKey[0], Anum_pg_cimv_userview,
+				BTEqualStrategyNumber, F_OIDEQ, ObjectIdGetDatum(userViewId));
+
+	SysScanDesc scanDescriptor = systable_beginscan(pgCimv,
+													PgCimvIndexId(), true,
+													NULL, scanKeyCount, scanKey);
+
+	HeapTuple heapTuple = systable_getnext(scanDescriptor);
+	if (!HeapTupleIsValid(heapTuple) && !missingOk)
+	{
+		ereport(ERROR, (errmsg("could not find valid entry for view %u", userViewId)));
+	}
+
+	if (HeapTupleIsValid(heapTuple))
+	{
+		cimvForm = (Form_pg_cimv) GETSTRUCT(heapTuple);
+	}
+
+	systable_endscan(scanDescriptor);
+	table_close(pgCimv, NoLock);
+
+	return cimvForm;
+}
+
+
 /*
  * RecordDistributedRelationDependencies creates the dependency entries
  * necessary for a distributed relation in addition to the preexisting ones

src/backend/distributed/sql/citus--9.4-1--9.5-1.sql

@@ -20,4 +20,82 @@ DROP TRIGGER dist_poolinfo_task_tracker_cache_invalidate ON pg_catalog.pg_dist_p
 DROP FUNCTION task_tracker_conninfo_cache_invalidate();
 DROP FUNCTION master_drop_sequences(text[]);
 
+CREATE FUNCTION pg_catalog.partial_agg_sfunc(internal, oid, anyelement, int)
+RETURNS internal
+AS 'MODULE_PATHNAME'
+LANGUAGE C PARALLEL SAFE;
+
+CREATE FUNCTION pg_catalog.partial_agg_ffunc(internal)
+RETURNS bytea
+AS 'MODULE_PATHNAME'
+LANGUAGE C PARALLEL SAFE;
+
+CREATE AGGREGATE pg_catalog.partial_agg(oid, anyelement, int) (
+    STYPE = internal,
+    SFUNC = pg_catalog.partial_agg_sfunc,
+    FINALFUNC = pg_catalog.partial_agg_ffunc
+);
+
+CREATE FUNCTION pg_catalog.combine_agg_sfunc(internal, oid, bytea, anyelement)
+RETURNS internal
+AS 'MODULE_PATHNAME'
+LANGUAGE C PARALLEL SAFE;
+
+CREATE FUNCTION pg_catalog.combine_agg_sfunc(internal, oid, bytea)
+RETURNS internal
+AS 'MODULE_PATHNAME'
+LANGUAGE C PARALLEL SAFE;
+
+CREATE FUNCTION pg_catalog.combine_agg_ffunc(internal, oid, bytea, anyelement)
+RETURNS anyelement
+AS 'MODULE_PATHNAME'
+LANGUAGE C PARALLEL SAFE;
+
+CREATE AGGREGATE pg_catalog.combine_agg(oid, bytea) (
+    STYPE = internal,
+    SFUNC = pg_catalog.combine_agg_sfunc,
+    FINALFUNC = pg_catalog.partial_agg_ffunc
+
+);
+
+CREATE AGGREGATE pg_catalog.finalize_agg(oid, bytea, anyelement) (
+    STYPE = internal,
+    SFUNC = pg_catalog.combine_agg_sfunc,
+    FINALFUNC = pg_catalog.combine_agg_ffunc,
+    FINALFUNC_EXTRA
+);
+
+REVOKE ALL ON FUNCTION pg_catalog.partial_agg_sfunc FROM PUBLIC;
+REVOKE ALL ON FUNCTION pg_catalog.partial_agg_ffunc FROM PUBLIC;
+REVOKE ALL ON FUNCTION pg_catalog.partial_agg FROM PUBLIC;
+REVOKE ALL ON FUNCTION pg_catalog.combine_agg_sfunc(internal, oid, bytea, anyelement) FROM PUBLIC;
+REVOKE ALL ON FUNCTION pg_catalog.combine_agg_sfunc(internal, oid, bytea) FROM PUBLIC;
+REVOKE ALL ON FUNCTION pg_catalog.combine_agg_ffunc FROM PUBLIC;
+REVOKE ALL ON FUNCTION pg_catalog.combine_agg FROM PUBLIC;
+REVOKE ALL ON FUNCTION pg_catalog.finalize_agg FROM PUBLIC;
+
+GRANT EXECUTE ON FUNCTION pg_catalog.partial_agg_sfunc TO PUBLIC;
+GRANT EXECUTE ON FUNCTION pg_catalog.partial_agg_ffunc TO PUBLIC;
+GRANT EXECUTE ON FUNCTION pg_catalog.partial_agg TO PUBLIC;
+GRANT EXECUTE ON FUNCTION pg_catalog.combine_agg_sfunc(internal, oid, bytea, anyelement) TO PUBLIC;
+GRANT EXECUTE ON FUNCTION pg_catalog.combine_agg_sfunc(internal, oid, bytea) TO PUBLIC;
+GRANT EXECUTE ON FUNCTION pg_catalog.combine_agg_ffunc TO PUBLIC;
+GRANT EXECUTE ON FUNCTION pg_catalog.combine_agg TO PUBLIC;
+GRANT EXECUTE ON FUNCTION pg_catalog.finalize_agg TO PUBLIC;
+
+--  add pg_cimv TODO: finalize name of this table
+CREATE TABLE citus.pg_cimv(
+    userview regclass NOT NULL PRIMARY KEY,
+    basetable regclass NOT NULL,
+    mattable regclass NOT NULL,
+    refreshview regclass NOT NULL,
+    deltatriggerfnnamespace name NOT NULL,
+    deltatriggerfnname name NOT NULL,
+    landingtable regclass NOT NULL,
+    jobid bigint NOT NULL
+);
+
+ALTER TABLE citus.pg_cimv SET SCHEMA pg_catalog;
+GRANT SELECT ON pg_catalog.pg_cimv TO public;
+
 RESET search_path;

src/backend/distributed/sql/downgrades/citus--9.5-1--9.4-1.sql

@@ -92,3 +92,14 @@ COMMENT ON FUNCTION master_drop_sequences(text[])
 RESET search_path;
 
 DROP FUNCTION pg_catalog.undistribute_table(table_name regclass);
+
+DROP AGGREGATE pg_catalog.finalize_agg(oid, bytea, anyelement);
+DROP AGGREGATE pg_catalog.combine_agg(oid, bytea);
+DROP FUNCTION pg_catalog.combine_agg_ffunc(internal, oid, bytea, anyelement);
+DROP FUNCTION pg_catalog.combine_agg_sfunc(internal, oid, bytea);
+DROP FUNCTION pg_catalog.combine_agg_sfunc(internal, oid, bytea, anyelement);
+DROP AGGREGATE pg_catalog.partial_agg(oid, anyelement, int);
+DROP FUNCTION pg_catalog.partial_agg_ffunc(internal);
+DROP FUNCTION pg_catalog.partial_agg_sfunc(internal, oid, anyelement, int);
+
+DROP TABLE pg_catalog.pg_cimv;

src/backend/distributed/utils/aggregate_utils.c

@@ -22,9 +22,11 @@
 #include "catalog/pg_type.h"
 #include "distributed/version_compat.h"
 #include "nodes/nodeFuncs.h"
+#include "parser/parse_agg.h"
 #include "utils/acl.h"
 #include "utils/builtins.h"
 #include "utils/datum.h"
+#include "utils/expandeddatum.h"
 #include "utils/lsyscache.h"
 #include "utils/syscache.h"
 #include "utils/typcache.h"
@@ -37,6 +39,11 @@ PG_FUNCTION_INFO_V1(worker_partial_agg_ffunc);
 PG_FUNCTION_INFO_V1(coord_combine_agg_sfunc);
 PG_FUNCTION_INFO_V1(coord_combine_agg_ffunc);
 
+PG_FUNCTION_INFO_V1(partial_agg_sfunc);
+PG_FUNCTION_INFO_V1(partial_agg_ffunc);
+PG_FUNCTION_INFO_V1(combine_agg_sfunc);
+PG_FUNCTION_INFO_V1(combine_agg_ffunc);
+
 /*
  * Holds information describing the structure of aggregation arguments
  * and helps to efficiently handle both a single argument and multiple
@@ -73,6 +80,11 @@ typedef struct StypeBox
 	AggregationArgumentContext *aggregationArgumentContext;
 } StypeBox;
 
+/* TODO: replace PG_FUNCTION_ARGS with explicit parameters */
+static Datum partial_agg_sfunc_internal(PG_FUNCTION_ARGS);
+static Datum partial_agg_ffunc_internal(PG_FUNCTION_ARGS, bool serializeToCString);
+static Datum combine_agg_sfunc_internal(PG_FUNCTION_ARGS, bool deserializeFromCString);
+static Datum combine_agg_ffunc_internal(PG_FUNCTION_ARGS);
 static HeapTuple GetAggregateForm(Oid oid, Form_pg_aggregate *form);
 static HeapTuple GetProcForm(Oid oid, Form_pg_proc *form);
 static HeapTuple GetTypeForm(Oid oid, Form_pg_type *form);
@@ -482,6 +494,62 @@ HandleStrictUninit(StypeBox *box, FunctionCallInfo fcinfo, Datum value)
 }
 
 
+Datum
+worker_partial_agg_sfunc(PG_FUNCTION_ARGS)
+{
+	return partial_agg_sfunc_internal(fcinfo);
+}
+
+
+Datum
+worker_partial_agg_ffunc(PG_FUNCTION_ARGS)
+{
+	return partial_agg_ffunc_internal(fcinfo, true);
+}
+
+
+Datum
+coord_combine_agg_sfunc(PG_FUNCTION_ARGS)
+{
+	return combine_agg_sfunc_internal(fcinfo, true);
+}
+
+
+Datum
+coord_combine_agg_ffunc(PG_FUNCTION_ARGS)
+{
+	return combine_agg_ffunc_internal(fcinfo);
+}
+
+
+Datum
+partial_agg_sfunc(PG_FUNCTION_ARGS)
+{
+	return partial_agg_sfunc_internal(fcinfo);
+}
+
+
+Datum
+partial_agg_ffunc(PG_FUNCTION_ARGS)
+{
+	return partial_agg_ffunc_internal(fcinfo, false);
+}
+
+
+Datum
+combine_agg_sfunc(PG_FUNCTION_ARGS)
+{
+	return combine_agg_sfunc_internal(fcinfo, false);
+}
+
+
+Datum
+combine_agg_ffunc(PG_FUNCTION_ARGS)
+{
+	return combine_agg_ffunc_internal(fcinfo);
+}
+
+
 /*
  * worker_partial_agg_sfunc advances transition state,
  * essentially implementing the following pseudocode:
@@ -492,7 +560,7 @@ HandleStrictUninit(StypeBox *box, FunctionCallInfo fcinfo, Datum value)
  * return box
  */
 Datum
-worker_partial_agg_sfunc(PG_FUNCTION_ARGS)
+partial_agg_sfunc_internal(PG_FUNCTION_ARGS)
 {
 	StypeBox *box = NULL;
 	Form_pg_aggregate aggform;
@@ -500,6 +568,7 @@ worker_partial_agg_sfunc(PG_FUNCTION_ARGS)
 	FmgrInfo info;
 	int argumentIndex = 0;
 	bool initialCall = PG_ARGISNULL(0);
+	int mode = PG_NARGS() < 4 ? 0 : PG_GETARG_INT32(3);
 
 	if (initialCall)
 	{
@@ -525,7 +594,25 @@ worker_partial_agg_sfunc(PG_FUNCTION_ARGS)
 	}
 
 	HeapTuple aggtuple = GetAggregateForm(box->agg, &aggform);
-	Oid aggsfunc = aggform->aggtransfn;
+
+	if (mode == 1 &&
+		(aggform->aggminvtransfn == 0 ||
+		 aggform->aggtransfn != aggform->aggmtransfn))
+	{
+		ereport(ERROR, (errmsg("applying inverse state transition not supported "
+							   "for this aggregate")));
+	}
+
+	Oid aggsfunc = InvalidOid;
+	if (mode == 1)
+	{
+		aggsfunc = aggform->aggminvtransfn;
+	}
+	else
+	{
+		aggsfunc = aggform->aggtransfn;
+	}
+
 
 	if (initialCall)
 	{
@@ -604,11 +691,11 @@ worker_partial_agg_sfunc(PG_FUNCTION_ARGS)
  * worker_partial_agg_ffunc serializes transition state,
  * essentially implementing the following pseudocode:
  *
- * (box) -> text
+ * (box) -> text/byte
  * return box.agg.stype.output(box.value)
  */
 Datum
-worker_partial_agg_ffunc(PG_FUNCTION_ARGS)
+partial_agg_ffunc_internal(PG_FUNCTION_ARGS, bool serializeToCString)
 {
 	LOCAL_FCINFO(innerFcinfo, 1);
 	FmgrInfo info;
@@ -635,27 +722,66 @@ worker_partial_agg_ffunc(PG_FUNCTION_ARGS)
 							"worker_partial_agg_ffunc expects an aggregate with COMBINEFUNC")));
 	}
 
-	if (aggform->aggtranstype == INTERNALOID)
+	if (serializeToCString && aggform->aggtranstype == INTERNALOID)
 	{
 		ereport(ERROR,
 				(errmsg(
 					 "worker_partial_agg_ffunc does not support aggregates with INTERNAL transition state")));
 	}
-
 	Oid transtype = aggform->aggtranstype;
+
 	ReleaseSysCache(aggtuple);
 
-	getTypeOutputInfo(transtype, &typoutput, &typIsVarlena);
+	Datum result = (Datum) 0;
+	bool isNull = false;
 
-	fmgr_info(typoutput, &info);
+	/* TODO: move function call info into box to avoid initializing it repeatedly */
+	if (OidIsValid(aggform->aggserialfn))
+	{
+		Expr *serialfnexpr = NULL;
+		FmgrInfo serialfn;
+		build_aggregate_serialfn_expr(aggform->aggserialfn,
+									  &serialfnexpr);
+		fmgr_info(aggform->aggserialfn, &serialfn);
+		fmgr_info_set_expr((Node *) serialfnexpr, &serialfn);
 
-	InitFunctionCallInfoData(*innerFcinfo, &info, 1, fcinfo->fncollation,
+		LOCAL_FCINFO(serialfn_fcinfo, 2);
-							 fcinfo->context, fcinfo->resultinfo);
+		InitFunctionCallInfoData(*serialfn_fcinfo,
-	fcSetArgExt(innerFcinfo, 0, box->value, box->valueNull);
+								 &serialfn,
+								 1,
+								 InvalidOid,
+								 (void *) fcinfo->context, NULL);
 
-	Datum result = FunctionCallInvoke(innerFcinfo);
+		fcSetArgExt(serialfn_fcinfo, 0,
+					MakeExpandedObjectReadOnly(box->value, box->valueNull,
+											   box->transtypeLen),
+					false);
 
-	if (innerFcinfo->isnull)
+		result = FunctionCallInvoke(serialfn_fcinfo);
+		isNull = serialfn_fcinfo->isnull;
+	}
+	else
+	{
+		if (serializeToCString)
+		{
+			getTypeOutputInfo(transtype, &typoutput, &typIsVarlena);
+		}
+		else
+		{
+			getTypeBinaryOutputInfo(transtype, &typoutput, &typIsVarlena);
+		}
+
+		fmgr_info(typoutput, &info);
+
+		InitFunctionCallInfoData(*innerFcinfo, &info, 1, fcinfo->fncollation,
+								 fcinfo->context, fcinfo->resultinfo);
+		fcSetArgExt(innerFcinfo, 0, box->value, box->valueNull);
+
+		result = FunctionCallInvoke(innerFcinfo);
+		isNull = innerFcinfo->isnull;
+	}
+
+	if (isNull)
 	{
 		PG_RETURN_NULL();
 	}
@@ -674,7 +800,7 @@ worker_partial_agg_ffunc(PG_FUNCTION_ARGS)
  * return box
  */
 Datum
-coord_combine_agg_sfunc(PG_FUNCTION_ARGS)
+combine_agg_sfunc_internal(PG_FUNCTION_ARGS, bool deserializeFromCString)
 {
 	LOCAL_FCINFO(innerFcinfo, 3);
 	FmgrInfo info;
@@ -702,7 +828,7 @@ coord_combine_agg_sfunc(PG_FUNCTION_ARGS)
 							"coord_combine_agg_sfunc expects an aggregate with COMBINEFUNC")));
 	}
 
-	if (aggform->aggtranstype == INTERNALOID)
+	if (deserializeFromCString && aggform->aggtranstype == INTERNALOID)
 	{
 		ereport(ERROR,
 				(errmsg(
@@ -728,19 +854,67 @@ coord_combine_agg_sfunc(PG_FUNCTION_ARGS)
 	bool valueNull = PG_ARGISNULL(2);
 	HeapTuple transtypetuple = GetTypeForm(box->transtype, &transtypeform);
 	Oid ioparam = getTypeIOParam(transtypetuple);
-	Oid deserial = transtypeform->typinput;
+	Oid deserial = deserializeFromCString ? transtypeform->typinput :
+				   transtypeform->typreceive;
 	ReleaseSysCache(transtypetuple);
 
-	fmgr_info(deserial, &info);
+	bool strictDeserial = false;
-	if (valueNull && info.fn_strict)
+	if (OidIsValid(aggform->aggdeserialfn))
+	{
+		strictDeserial = true;
+	}
+	else
+	{
+		fmgr_info(deserial, &info);
+		strictDeserial = info.fn_strict;
+	}
+
+	if (valueNull && strictDeserial)
 	{
 		value = (Datum) 0;
 	}
+	/* TODO: store function call info in box to avoid recurring initialization overhead */
+	else if (OidIsValid(aggform->aggdeserialfn))
+	{
+		Expr *deserialfnexpr = NULL;
+		FmgrInfo deserialfn;
+		build_aggregate_deserialfn_expr(aggform->aggdeserialfn,
+										&deserialfnexpr);
+		fmgr_info(aggform->aggdeserialfn, &deserialfn);
+		fmgr_info_set_expr((Node *) deserialfnexpr, &deserialfn);
+
+		LOCAL_FCINFO(deserialfn_fcinfo, 2);
+		InitFunctionCallInfoData(*deserialfn_fcinfo,
+								 &deserialfn,
+								 2,
+								 InvalidOid,
+								 (void *) fcinfo->context, NULL);
+
+
+		fcSetArgExt(deserialfn_fcinfo, 0, PointerGetDatum(PG_GETARG_BYTEA_P(2)),
+					valueNull);
+		fcSetArgExt(deserialfn_fcinfo, 1, PointerGetDatum(NULL), false);
+
+		value = FunctionCallInvoke(deserialfn_fcinfo);
+		valueNull = deserialfn_fcinfo->isnull;
+	}
 	else
 	{
 		InitFunctionCallInfoData(*innerFcinfo, &info, 3, fcinfo->fncollation,
 								 fcinfo->context, fcinfo->resultinfo);
-		fcSetArgExt(innerFcinfo, 0, PG_GETARG_DATUM(2), valueNull);
+		if (deserializeFromCString)
+		{
+			fcSetArgExt(innerFcinfo, 0, PG_GETARG_DATUM(2), valueNull);
+		}
+		else
+		{
+			StringInfo string = makeStringInfo();
+			appendBinaryStringInfo(string,
+								   VARDATA_ANY(PG_GETARG_DATUM(2)),
+								   VARSIZE_ANY_EXHDR(PG_GETARG_DATUM(2)));
+			fcSetArgExt(innerFcinfo, 0, PointerGetDatum(string), valueNull);
+		}
+
 		fcSetArg(innerFcinfo, 1, ObjectIdGetDatum(ioparam));
 		fcSetArg(innerFcinfo, 2, Int32GetDatum(-1)); /* typmod */
 
@@ -788,7 +962,7 @@ coord_combine_agg_sfunc(PG_FUNCTION_ARGS)
  * return box.agg.ffunc(box.value)
  */
 Datum
-coord_combine_agg_ffunc(PG_FUNCTION_ARGS)
+combine_agg_ffunc_internal(PG_FUNCTION_ARGS)
 {
 	StypeBox *box = (StypeBox *) (PG_ARGISNULL(0) ? NULL : PG_GETARG_POINTER(0));
 	LOCAL_FCINFO(innerFcinfo, FUNC_MAX_ARGS);
@@ -872,7 +1046,7 @@ TypecheckWorkerPartialAggArgType(FunctionCallInfo fcinfo, StypeBox *box)
 		return false;
 	}
 
-	Assert(list_length(aggref->args) == 2);
+	Assert(list_length(aggref->args) == 2 || list_length(aggref->args) == 3);
 	TargetEntry *aggarg = list_nth(aggref->args, 1);
 
 	bool argtypesNull;

src/include/distributed/cimv.h

@@ -0,0 +1,20 @@
+#ifndef CIMV_H
+#define CIMV_H
+
+#include "postgres.h"
+#include "nodes/plannodes.h"
+
+#define CITUS_INTERNAL_SCHEMA "citus_internal"
+#define CITUS_NAMESPACE "citus"
+#define MATERIALIZATION_TABLE_SUFFIX "mt"
+#define LANDING_TABLE_SUFFIX "ld"
+#define REFRESH_VIEW_SUFFIX "rv"
+
+extern bool ProcessCreateMaterializedViewStmt(const CreateTableAsStmt *stmt, const
+											  char *query_string, PlannedStmt *pstmt);
+extern void ProcessDropMaterializedViewStmt(DropStmt *stmt);
+extern void ProcessDropViewStmt(DropStmt *stmt);
+extern bool ProcessRefreshMaterializedViewStmt(RefreshMatViewStmt *stmt);
+extern void RefreshCimv(Form_pg_cimv formCimv, bool skipData, bool isCreate);
+
+#endif

src/include/distributed/metadata_cache.h

@@ -203,6 +203,8 @@ extern Oid CitusCatalogNamespaceId(void);
 extern Oid DistColocationRelationId(void);
 extern Oid DistColocationConfigurationIndexId(void);
 extern Oid DistPartitionRelationId(void);
+extern Oid PgCimvId(void);
+extern Oid PgCimvIndexId(void);
 extern Oid DistShardRelationId(void);
 extern Oid DistPlacementRelationId(void);
 extern Oid DistNodeRelationId(void);

src/include/distributed/metadata_utility.h

@@ -23,6 +23,7 @@
 #include "catalog/objectaddress.h"
 #include "distributed/citus_nodes.h"
 #include "distributed/relay_utility.h"
+#include "distributed/pg_cimv.h"
 #include "utils/acl.h"
 #include "utils/relcache.h"
 
@@ -125,6 +126,9 @@ extern uint64 InsertShardPlacementRow(uint64 shardId, uint64 placementId,
 extern void InsertIntoPgDistPartition(Oid relationId, char distributionMethod,
 									  Var *distributionColumn, uint32 colocationId,
 									  char replicationModel);
+extern void InsertIntoPgCimv(Form_pg_cimv cimv);
+extern void DeletePgCimvRow(Oid userViewId);
+extern Form_pg_cimv LookupCimvFromCatalog(Oid userViewId, bool missingOk);
 extern void DeletePartitionRow(Oid distributedRelationId);
 extern void DeleteShardRow(uint64 shardId);
 extern void UpdatePartitionShardPlacementStates(ShardPlacement *parentShardPlacement,

src/include/distributed/pg_cimv.h

@@ -0,0 +1,51 @@
+/*-------------------------------------------------------------------------
+ *
+ * pg_cimv.h
+ *	  TODO
+ *
+ * Copyright (c) Citus Data, Inc.
+ *
+ *-------------------------------------------------------------------------
+ */
+
+#ifndef PG_CIMV_H
+#define PG_CIMV_H
+
+/* ----------------
+ *		pg_cimv definition.
+ * ----------------
+ */
+typedef struct FormData_pg_cimv
+{
+	Oid userview;
+	Oid basetable;
+	Oid mattable;
+	Oid refreshview;
+	NameData triggerfnnamespace;
+	NameData triggerfnname;
+	Oid landingtable;
+	int64 jobid;
+} FormData_pg_cimv;
+
+/* ----------------
+ *      FormData_pg_cimv corresponds to a pointer to a tuple with
+ *      the format of pg_cimv relation.
+ * ----------------
+ */
+typedef FormData_pg_cimv *Form_pg_cimv;
+
+/* ----------------
+ *      compiler constants for pg_cimv
+ * ----------------
+ */
+#define Natts_pg_cimv 8
+#define Anum_pg_cimv_userview 1
+#define Anum_pg_cimv_basetable 2
+#define Anum_pg_cimv_mattable 3
+#define Anum_pg_cimv_refreshview 4
+#define Anum_pg_cimv_triggernamespace 5
+#define Anum_pg_cimv_triggername 6
+#define Anum_pg_cimv_landingtable 7
+#define Anum_pg_cimv_jobid 8
+
+#endif   /* PG_CIMV_H */

src/test/regress/expected/multi_extension.out

@@ -448,10 +448,19 @@ SELECT * FROM print_extension_changes();
  function task_tracker_task_status(bigint,integer)                   |
  function worker_execute_sql_task(bigint,integer,text,boolean)       |
  function worker_merge_files_and_run_query(bigint,integer,text,text) |
+                                                                     | function combine_agg(oid,bytea)
+                                                                     | function combine_agg_ffunc(internal,oid,bytea,anyelement)
+                                                                     | function combine_agg_sfunc(internal,oid,bytea)
+                                                                     | function combine_agg_sfunc(internal,oid,bytea,anyelement)
                                                                      | function create_citus_local_table(regclass)
+                                                                     | function finalize_agg(oid,bytea,anyelement)
+                                                                     | function partial_agg(oid,anyelement,integer)
+                                                                     | function partial_agg_ffunc(internal)
+                                                                     | function partial_agg_sfunc(internal,oid,anyelement,integer)
                                                                      | function undistribute_table(regclass)
                                                                      | function worker_record_sequence_dependency(regclass,regclass,name)
-(10 rows)
+                                                                     | table pg_cimv
+(19 rows)
 
 DROP TABLE prev_objects, extension_diff;
 -- show running version

src/test/regress/multi_schedule

@@ -96,7 +96,7 @@ test: multi_subquery_in_where_reference_clause join geqo adaptive_executor propa
 test: multi_subquery_union multi_subquery_in_where_clause multi_subquery_misc statement_cancel_error_message
 test: multi_agg_distinct multi_agg_approximate_distinct multi_limit_clause_approximate multi_outer_join_reference multi_single_relation_subquery multi_prepare_plsql set_role_in_transaction
 test: multi_reference_table multi_select_for_update relation_access_tracking pg13_with_ties
-test: custom_aggregate_support aggregate_support tdigest_aggregate_support
+test: custom_aggregate_support aggregate_support tdigest_aggregate_support cimv
 test: multi_average_expression multi_working_columns multi_having_pushdown having_subquery
 test: multi_array_agg multi_limit_clause multi_orderby_limit_pushdown
 test: multi_jsonb_agg multi_jsonb_object_agg multi_json_agg multi_json_object_agg bool_agg ch_bench_having chbenchmark_all_queries expression_reference_join anonymous_columns

src/test/regress/sql/cimv.sql

@@ -0,0 +1,360 @@
+--
+-- CIMV
+--   Tests for Citus Incremental Materialized Views
+--
+
+\set VERBOSITY terse
+SET citus.next_shard_id TO 400000;
+CREATE SCHEMA cimv;
+SET search_path TO cimv, public;
+
+SET citus.shard_count TO 4;
+
+CREATE TABLE events (a int, b int, c double precision, d timestamp, e bigint);
+
+INSERT INTO events
+SELECT v % 10 AS a,
+       v % 100 AS b,
+       v / 3.0 AS c,
+       timestamp '2020-01-01 20:00:00' +
+       ((v / 10000.0) * (timestamp '2020-01-01 15:00:00' -
+                   timestamp '2020-01-01 10:00:00')) AS d,
+       v AS e
+FROM generate_series(1, 10000) v;
+
+CREATE MATERIALIZED VIEW mv WITH (citus.cimv) AS
+SELECT a,
+       date_trunc('hour', d) AS d_hour,
+       min(b) AS min_b,
+       max(b) AS max_b,
+       avg(b) AS avg_b,
+       min(c) AS min_c,
+       max(c) AS max_c,
+       avg(c) AS avg_c,
+       min(e) AS min_e,
+       max(e) AS max_e,
+       avg(e) AS avg_e
+FROM events
+WHERE b > 10
+GROUP BY a, d_hour;
+
+SELECT a,
+       d_hour,
+       min_b::numeric(12,2),
+       max_b::numeric(12,2),
+       avg_b::numeric(12,2),
+       min_c::numeric(12,2),
+       max_c::numeric(12,2),
+       avg_c::numeric(12,2),
+       min_e::numeric(12,2),
+       max_e::numeric(12,2),
+       avg_e::numeric(12,2)
+FROM mv
+ORDER BY a, d_hour;
+
+INSERT INTO events
+SELECT v % 10 AS a,
+       v % 100 AS b,
+       v / 3.0 AS c,
+       timestamp '2020-01-01 20:00:00' +
+       ((v / 10000.0) * (timestamp '2020-01-01 15:00:00' -
+                   timestamp '2020-01-01 10:00:00')) AS d,
+       v AS e
+FROM generate_series(10000, 11000) v;
+
+SELECT a,
+       d_hour,
+       min_b::numeric(12,2),
+       max_b::numeric(12,2),
+       avg_b::numeric(12,2),
+       min_c::numeric(12,2),
+       max_c::numeric(12,2),
+       avg_c::numeric(12,2),
+       min_e::numeric(12,2),
+       max_e::numeric(12,2),
+       avg_e::numeric(12,2)
+FROM mv
+ORDER BY a, d_hour;
+
+DELETE FROM events WHERE b < 100;
+
+DROP VIEW mv;
+
+DROP MATERIALIZED VIEW mv;
+
+CREATE MATERIALIZED VIEW mv WITH (citus.cimv) AS
+SELECT a,
+       date_trunc('hour', d) AS d_hour,
+       avg(b) AS avg_b
+FROM events
+WHERE b > 10
+GROUP BY a, d_hour;
+
+SELECT a,
+       d_hour,
+       avg_b::numeric(12,2)
+FROM mv
+ORDER BY a, d_hour;
+
+DELETE FROM events WHERE b < 20;
+
+SELECT a,
+       d_hour,
+       avg_b::numeric(12,2)
+FROM mv
+ORDER BY a, d_hour;
+
+UPDATE events SET b = b + b;
+
+SELECT a,
+       d_hour,
+       avg_b::numeric(12,2)
+FROM mv
+ORDER BY a, d_hour;
+
+DROP MATERIALIZED VIEW mv;
+
+CREATE MATERIALIZED VIEW mv WITH (citus.cimv, citus.insertonlycapture) AS
+SELECT a,
+       date_trunc('hour', d) AS d_hour,
+       avg(b) AS avg_b
+FROM events
+WHERE b > 10
+GROUP BY a, d_hour;
+
+SELECT a,
+       d_hour,
+       avg_b::numeric(12,2)
+FROM mv
+ORDER BY a, d_hour;
+
+
+INSERT INTO events
+SELECT v % 10 AS a,
+       v % 100 AS b,
+       v / 3.0 AS c,
+       timestamp '2020-01-01 20:00:00' +
+       ((v / 10000.0) * (timestamp '2020-01-01 15:00:00' -
+                   timestamp '2020-01-01 10:00:00')) AS d,
+       v AS e
+FROM generate_series(11000, 12000) v;
+
+SELECT a,
+       d_hour,
+       avg_b::numeric(12,2)
+FROM mv
+ORDER BY a, d_hour;
+
+DELETE FROM events WHERE b < 100;
+
+SELECT a,
+       d_hour,
+       avg_b::numeric(12,2)
+FROM mv
+ORDER BY a, d_hour;
+
+UPDATE events SET b = b + b;
+
+SELECT a,
+       d_hour,
+       avg_b::numeric(12,2)
+FROM mv
+ORDER BY a, d_hour;
+
+REFRESH MATERIALIZED VIEW mv WITH NO DATA;
+
+SELECT * FROM mv;
+
+REFRESH MATERIALIZED VIEW mv;
+
+SELECT a,
+       d_hour,
+       avg_b::numeric(12,2)
+FROM mv
+ORDER BY a, d_hour;
+
+
+DROP MATERIALIZED VIEW mv;
+
+CREATE MATERIALIZED VIEW mv WITH (citus.cimv, citus.insertonlycapture) AS
+SELECT a,
+       date_trunc('hour', d) AS d_hour,
+       avg(b) AS avg_b
+FROM events
+WHERE b > 10
+GROUP BY a, d_hour WITH NO DATA;
+
+SELECT * FROM mv;
+
+DROP MATERIALIZED VIEW mv;
+
+SELECT create_distributed_table('events', 'a');
+
+CREATE MATERIALIZED VIEW mv WITH (citus.cimv) AS
+SELECT a,
+       date_trunc('hour', d) AS d_hour,
+       min(b) AS min_b,
+       max(b) AS max_b,
+       avg(b) AS avg_b,
+       min(c) AS min_c,
+       max(c) AS max_c,
+       avg(c) AS avg_c,
+       min(e) AS min_e,
+       max(e) AS max_e,
+       avg(e) AS avg_e
+FROM events
+WHERE b > 10
+GROUP BY a, d_hour;
+
+SELECT a,
+       d_hour,
+       min_b::numeric(12,2),
+       max_b::numeric(12,2),
+       avg_b::numeric(12,2),
+       min_c::numeric(12,2),
+       max_c::numeric(12,2),
+       avg_c::numeric(12,2),
+       min_e::numeric(12,2),
+       max_e::numeric(12,2),
+       avg_e::numeric(12,2)
+FROM mv
+ORDER BY a, d_hour;
+
+INSERT INTO events
+SELECT v % 10 AS a,
+       v % 100 AS b,
+       v / 3.0 AS c,
+       timestamp '2020-01-01 20:00:00' +
+       ((v / 10000.0) * (timestamp '2020-01-01 15:00:00' -
+                   timestamp '2020-01-01 10:00:00')) AS d,
+       v AS e
+FROM generate_series(12000, 13000) v;
+
+SELECT a,
+       d_hour,
+       min_b::numeric(12,2),
+       max_b::numeric(12,2),
+       avg_b::numeric(12,2),
+       min_c::numeric(12,2),
+       max_c::numeric(12,2),
+       avg_c::numeric(12,2),
+       min_e::numeric(12,2),
+       max_e::numeric(12,2),
+       avg_e::numeric(12,2)
+FROM mv
+ORDER BY a, d_hour;
+
+DELETE FROM events WHERE b < 100;
+
+DROP VIEW mv;
+
+DROP MATERIALIZED VIEW mv;
+
+CREATE MATERIALIZED VIEW mv WITH (citus.cimv) AS
+SELECT a,
+       date_trunc('hour', d) AS d_hour,
+       avg(b) AS avg_b
+FROM events
+WHERE b > 10
+GROUP BY a, d_hour;
+
+SELECT a,
+       d_hour,
+       avg_b::numeric(12,2)
+FROM mv
+ORDER BY a, d_hour;
+
+DELETE FROM events WHERE b < 20;
+
+SELECT a,
+       d_hour,
+       avg_b::numeric(12,2)
+FROM mv
+ORDER BY a, d_hour;
+
+
+UPDATE events SET b = b + b;
+
+SELECT a,
+       d_hour,
+       avg_b::numeric(12,2)
+FROM mv
+ORDER BY a, d_hour;
+
+
+DROP MATERIALIZED VIEW mv;
+
+CREATE MATERIALIZED VIEW mv WITH (citus.cimv, citus.insertonlycapture) AS
+SELECT a,
+       date_trunc('hour', d) AS d_hour,
+       avg(b) AS avg_b
+FROM events
+WHERE b > 10
+GROUP BY a, d_hour;
+
+SELECT a,
+       d_hour,
+       avg_b::numeric(12,2)
+FROM mv
+ORDER BY a, d_hour;
+
+INSERT INTO events
+SELECT v % 10 AS a,
+       v % 100 AS b,
+       v / 3.0 AS c,
+       timestamp '2020-01-01 20:00:00' +
+       ((v / 10000.0) * (timestamp '2020-01-01 15:00:00' -
+                   timestamp '2020-01-01 10:00:00')) AS d,
+       v AS e
+FROM generate_series(13000, 14000) v;
+
+SELECT a,
+       d_hour,
+       avg_b::numeric(12,2)
+FROM mv
+ORDER BY a, d_hour;
+
+DELETE FROM events WHERE b < 100;
+
+SELECT a,
+       d_hour,
+       avg_b::numeric(12,2)
+FROM mv
+ORDER BY a, d_hour;
+
+UPDATE events SET b = b + b;
+
+SELECT a,
+       d_hour,
+       avg_b::numeric(12,2)
+FROM mv
+ORDER BY a, d_hour;
+
+REFRESH MATERIALIZED VIEW mv WITH NO DATA;
+
+SELECT * FROM mv;
+
+REFRESH MATERIALIZED VIEW mv;
+
+SELECT a,
+       d_hour,
+       avg_b::numeric(12,2)
+FROM mv
+ORDER BY a, d_hour;
+
+DROP MATERIALIZED VIEW mv;
+
+CREATE MATERIALIZED VIEW mv WITH (citus.cimv, citus.insertonlycapture) AS
+SELECT a,
+       date_trunc('hour', d) AS d_hour,
+       avg(b) AS avg_b
+FROM events
+WHERE b > 10
+GROUP BY a, d_hour WITH NO DATA;
+
+SELECT * FROM mv;
+
+DROP MATERIALIZED VIEW mv;
+
+SET client_min_messages TO WARNING; -- suppress cascade messages
+DROP SCHEMA cimv CASCADE;