Change syntax for continuous aggregates

We change the syntax for defining continuous aggregates to use `CREATE
MATERIALIZED VIEW` rather than `CREATE VIEW`. The command still creates
a view, while `CREATE MATERIALIZED VIEW` creates a table.  Raise an
error if `CREATE VIEW` is used to create a continuous aggregate and
redirect to `CREATE MATERIALIZED VIEW`.

In a similar vein, `DROP MATERIALIZED VIEW` is used for continuous
aggregates and continuous aggregates cannot be dropped with `DROP
VIEW`.

Continuous aggregates are altered using `ALTER MATERIALIZED VIEW`
rather than `ALTER VIEW`, so we ensure that it works for `ALTER
MATERIALIZED VIEW` and gives an error if you try to use `ALTER VIEW` to
change a continuous aggregate.

Note that we allow `ALTER VIEW ... SET SCHEMA` to be used with the
partial view as well as with the direct view, so this is handled as a
special case.

Fixes #2233

Co-authored-by: =?UTF-8?q?Erik=20Nordstr=C3=B6m?= <erik@timescale.com>
Co-authored-by: Mats Kindahl <mats@timescale.com>

src/continuous_agg.c

@@ -514,6 +514,21 @@ ts_continuous_agg_find_by_relid(Oid relid)
 	return ts_continuous_agg_find_userview_name(schemaname, relname);
 }
 
+/*
+ * Find a continuous aggregate by range var.
+ */
+ContinuousAgg *
+ts_continuous_agg_find_by_rv(const RangeVar *rv)
+{
+	Oid relid;
+	if (rv == NULL)
+		return NULL;
+	relid = RangeVarGetRelid(rv, NoLock, true);
+	if (!OidIsValid(relid))
+		return NULL;
+	return ts_continuous_agg_find_by_relid(relid);
+}
+
 /*
  * Drops continuous aggs and all related objects.
  *
@@ -834,11 +849,14 @@ ts_continuous_agg_rename_schema_name(char *old_schema, char *new_schema)
 }
 
 extern void
-ts_continuous_agg_rename_view(char *old_schema, char *name, char *new_schema, char *new_name)
+ts_continuous_agg_rename_view(const char *old_schema, const char *name, const char *new_schema,
+							  const char *new_name, ObjectType *object_type)
 {
 	ScanIterator iterator =
 		ts_scan_iterator_create(CONTINUOUS_AGG, RowExclusiveLock, CurrentMemoryContext);
 
+	Assert(object_type);
+
 	ts_scanner_foreach(&iterator)
 	{
 		TupleInfo *tinfo = ts_scan_iterator_tuple_info(&iterator);
@@ -847,11 +865,24 @@ ts_continuous_agg_rename_view(char *old_schema, char *name, char *new_schema, ch
 		FormData_continuous_agg *data = (FormData_continuous_agg *) GETSTRUCT(tuple);
 		HeapTuple new_tuple = NULL;
 		ContinuousAggViewType vtyp = ts_continuous_agg_view_type(data, old_schema, name);
+
 		switch (vtyp)
 		{
 			case ContinuousAggUserView:
 			{
-				FormData_continuous_agg *new_data = ensure_new_tuple(tuple, &new_tuple);
+				FormData_continuous_agg *new_data;
+
+				if (*object_type == OBJECT_VIEW)
+					ereport(ERROR,
+							(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
+							 errmsg("cannot alter continuous aggregate using ALTER VIEW"),
+							 errhint(
+								 "Use ALTER MATERIALIZED VIEW to alter a continuous aggregate.")));
+
+				Assert(*object_type == OBJECT_MATVIEW);
+				*object_type = OBJECT_VIEW;
+
+				new_data = ensure_new_tuple(tuple, &new_tuple);
 				namestrcpy(&new_data->user_view_schema, new_schema);
 				namestrcpy(&new_data->user_view_name, new_name);
 				break;

src/continuous_agg.h

@@ -77,6 +77,7 @@ int64 ts_continuous_agg_get_completed_threshold(int32 materialization_id);
 extern TSDLLEXPORT ContinuousAgg *ts_continuous_agg_find_by_view_name(const char *schema,
 																	  const char *name);
 extern TSDLLEXPORT ContinuousAgg *ts_continuous_agg_find_by_relid(Oid relid);
+extern TSDLLEXPORT ContinuousAgg *ts_continuous_agg_find_by_rv(const RangeVar *rv);
 
 extern void ts_continuous_agg_drop_view_callback(ContinuousAgg *ca, const char *schema,
 												 const char *name);
@@ -87,8 +88,9 @@ extern TSDLLEXPORT ContinuousAggViewType ts_continuous_agg_view_type(FormData_co
 																	 const char *schema,
 																	 const char *name);
 extern void ts_continuous_agg_rename_schema_name(char *old_schema, char *new_schema);
-extern void ts_continuous_agg_rename_view(char *old_schema, char *name, char *new_schema,
-										  char *new_name);
+extern void ts_continuous_agg_rename_view(const char *old_schema, const char *name,
+										  const char *new_schema, const char *new_name,
+										  ObjectType *object_type);
 
 extern TSDLLEXPORT int32 ts_number_of_continuous_aggs(void);
 

src/cross_module_fn.c

@@ -245,7 +245,7 @@ error_no_default_fn_pg_enterprise(PG_FUNCTION_ARGS)
 }
 
 static DDLResult
-process_cagg_viewstmt_default(ViewStmt *stmt, const char *query_string, void *pstmt,
+process_cagg_viewstmt_default(Node *stmt, const char *query_string, void *pstmt,
 							  WithClauseResult *with_clause_options)
 {
 	return error_no_default_fn_bool_void_community();

src/cross_module_fn.h

@@ -88,7 +88,7 @@ typedef struct CrossModuleFunctions
 	PGFunction partialize_agg;
 	PGFunction finalize_agg_sfunc;
 	PGFunction finalize_agg_ffunc;
-	DDLResult (*process_cagg_viewstmt)(ViewStmt *stmt, const char *query_string, void *pstmt,
+	DDLResult (*process_cagg_viewstmt)(Node *stmt, const char *query_string, void *pstmt,
 									   WithClauseResult *with_clause_options);
 	PGFunction continuous_agg_invalidation_trigger;
 	PGFunction continuous_agg_refresh;

src/hypertable.c

@@ -2038,8 +2038,10 @@ ts_hypertable_create_from_info(Oid table_relid, int32 hypertable_id, uint32 flag
 					(errcode(ERRCODE_WRONG_OBJECT_TYPE),
 					 errmsg("table \"%s\" is already partitioned", get_rel_name(table_relid)),
 					 errdetail("It is not possible to turn partitioned tables into hypertables.")));
+		case RELKIND_MATVIEW:
 		case RELKIND_RELATION:
 			break;
+
 		default:
 			ereport(ERROR, (errcode(ERRCODE_WRONG_OBJECT_TYPE), errmsg("invalid relation type")));
 	}