Drop chunks from materialized hypertables

CHANGELOG.md

@@ -9,9 +9,10 @@ accidentally triggering the load of a previous DB version.**
 **Major Features**
 
 **Minor Features**
+* #1666 Support drop_chunks API for continuous aggregates
 
 **Bugfixes**
-* #1648 Drop chunks for materialized hypertable
+* #1648 Drop chunks from materialized hypertable
 * #1665 Add ignore_invalidation_older_than to timescaledb_information.continuous_aggregates view
 * #1668 Cannot add dimension if hypertable has empty chunks
 * #1674 Fix time_bucket_gapfill's interaction with GROUP BY

src/chunk.c

@@ -2392,11 +2392,14 @@ ts_chunk_drop_process_materialization(Oid hypertable_relid,
 	ts_cache_release(hcache);
 }
 
+/* Continuous agg materialization hypertables can be dropped
+ * only if a user explicitly specifies the table name
+ */
 List *
 ts_chunk_do_drop_chunks(Oid table_relid, Datum older_than_datum, Datum newer_than_datum,
 						Oid older_than_type, Oid newer_than_type, bool cascade,
 						CascadeToMaterializationOption cascades_to_materializations,
-						int32 log_level)
+						int32 log_level, bool user_supplied_table_name)
 {
 	uint64 i = 0;
 	uint64 num_chunks = 0;
@@ -2413,9 +2416,18 @@ ts_chunk_do_drop_chunks(Oid table_relid, Datum older_than_datum, Datum newer_tha
 	switch (ts_continuous_agg_hypertable_status(hypertable_id))
 	{
 		case HypertableIsMaterialization:
+			if (user_supplied_table_name == false)
+			{
+				elog(ERROR, "cannot drop chunks on a continuous aggregate materialization table");
+			}
+			has_continuous_aggs = false;
+			break;
 		case HypertableIsMaterializationAndRaw:
-			elog(ERROR, "cannot drop_chunks on a continuous aggregate materialization table");
-			pg_unreachable();
+			if (user_supplied_table_name == false)
+			{
+				elog(ERROR, "cannot drop chunks on a continuous aggregate materialization table");
+			}
+			has_continuous_aggs = true;
 			break;
 		case HypertableIsRawTable:
 			if (cascades_to_materializations == CASCADE_TO_MATERIALIZATION_UNKNOWN)
@@ -2485,7 +2497,8 @@ ts_chunk_do_drop_chunks(Oid table_relid, Datum older_than_datum, Datum newer_tha
 																older_than_type,
 																newer_than_type,
 																cascade,
-																log_level);
+																log_level,
+																user_supplied_table_name);
 	}
 	return dropped_chunk_names;
 }
@@ -2550,6 +2563,7 @@ ts_chunk_drop_chunks(PG_FUNCTION_ARGS)
 	bool cascade, verbose;
 	CascadeToMaterializationOption cascades_to_materializations;
 	int elevel;
+	bool user_supplied_table_name = true;
 
 	/*
 	 * When past the first call of the SRF, dropping has already been completed,
@@ -2585,10 +2599,27 @@ ts_chunk_drop_chunks(PG_FUNCTION_ARGS)
 	if (table_name != NULL)
 	{
 		if (ht_oids == NIL)
-			ereport(ERROR,
-					(errcode(ERRCODE_TS_HYPERTABLE_NOT_EXIST),
-					 errmsg("hypertable \"%s\" does not exist", NameStr(*table_name))));
+		{
+			ContinuousAgg *ca = NULL;
+			ca = ts_continuous_agg_find_userview_name(schema_name ? NameStr(*schema_name) : NULL,
+													  NameStr(*table_name));
+			if (ca == NULL)
+				ereport(ERROR,
+						(errcode(ERRCODE_TS_HYPERTABLE_NOT_EXIST),
+						 errmsg("\"%s\" is not a hypertable or a continuous aggregate view",
+								NameStr(*table_name)),
+						 errhint("It is only possible to drop chunks from a hypertable or "
+								 "continuous aggregate view")));
+			else
+			{
+				int32 matid = ca->data.mat_hypertable_id;
+				Hypertable *mat_ht = ts_hypertable_get_by_id(matid);
+				ht_oids = lappend_oid(ht_oids, mat_ht->main_table_relid);
+			}
+		}
 	}
+	else
+		user_supplied_table_name = false;
 
 	/* Initial multi function call setup */
 	funcctx = SRF_FIRSTCALL_INIT();
@@ -2657,7 +2688,8 @@ ts_chunk_drop_chunks(PG_FUNCTION_ARGS)
 										  newer_than_type,
 										  cascade,
 										  cascades_to_materializations,
-										  elevel);
+										  elevel,
+										  user_supplied_table_name);
 		dc_names = list_concat(dc_names, dc_temp);
 
 		MemoryContextSwitchTo(oldcontext);

src/chunk.h

@@ -139,7 +139,7 @@ extern TSDLLEXPORT List *
 ts_chunk_do_drop_chunks(Oid table_relid, Datum older_than_datum, Datum newer_than_datum,
 						Oid older_than_type, Oid newer_than_type, bool cascade,
 						CascadeToMaterializationOption cascades_to_materializations,
-						int32 log_level);
+						int32 log_level, bool user_supplied_table_name);
 extern TSDLLEXPORT Chunk *
 ts_chunk_get_chunks_in_time_range(Oid table_relid, Datum older_than_datum, Datum newer_than_datum,
 								  Oid older_than_type, Oid newer_than_type, char *caller_name,

src/continuous_agg.c

@@ -401,6 +401,41 @@ ts_continuous_agg_find_by_view_name(const char *schema, const char *name)
 	return ca;
 }
 
+ContinuousAgg *
+ts_continuous_agg_find_userview_name(const char *schema, const char *name)
+{
+	ScanIterator iterator =
+		ts_scan_iterator_create(CONTINUOUS_AGG, AccessShareLock, CurrentMemoryContext);
+	ContinuousAgg *ca = NULL;
+	int count = 0;
+	const char *chkschema = schema;
+
+	ts_scanner_foreach(&iterator)
+	{
+		ContinuousAggViewType vtyp;
+		FormData_continuous_agg *data =
+			(FormData_continuous_agg *) GETSTRUCT(ts_scan_iterator_tuple(&iterator));
+		if (schema == NULL)
+		{
+			/* only user visible views will be returned */
+			Oid relid = RelnameGetRelid(NameStr(data->user_view_name));
+			if (relid == InvalidOid)
+				continue;
+			chkschema = NameStr(data->user_view_schema);
+		}
+
+		vtyp = ts_continuous_agg_view_type(data, chkschema, name);
+		if (vtyp == ContinuousAggUserView)
+		{
+			ca = palloc0(sizeof(*ca));
+			continuous_agg_init(ca, data);
+			count++;
+		}
+	}
+	Assert(count <= 1);
+	return ca;
+}
+
 ContinuousAgg *
 ts_continuous_agg_find_by_job_id(int32 job_id)
 {
@@ -783,3 +818,51 @@ ts_continuous_agg_get_user_view_oid(ContinuousAgg *agg)
 		elog(ERROR, "could not find user view for continuous agg");
 	return view_relid;
 }
+
+static int32
+find_raw_hypertable_for_materialization(int32 mat_hypertable_id)
+{
+	short count = 0;
+	int32 htid = INVALID_HYPERTABLE_ID;
+	ScanIterator iterator =
+		ts_scan_iterator_create(CONTINUOUS_AGG, RowExclusiveLock, CurrentMemoryContext);
+
+	init_scan_by_mat_hypertable_id(&iterator, mat_hypertable_id);
+	ts_scanner_foreach(&iterator)
+	{
+		TupleInfo *ti = ts_scan_iterator_tuple_info(&iterator);
+		HeapTuple tuple = ti->tuple;
+		Form_continuous_agg form = (Form_continuous_agg) GETSTRUCT(tuple);
+		htid = form->raw_hypertable_id;
+		count++;
+	}
+	Assert(count <= 1);
+	ts_scan_iterator_close(&iterator);
+	return htid;
+}
+
+/* Continuous aggregate materialization hypertables inherit integer_now func
+ * from the raw hypertable (unless it was explictly reset for cont. aggregate.
+ * Walk the materialzation hyperatable ->raw hypertable tree till
+ * we find a hypertable that has integer_now_func set.
+ */
+TSDLLEXPORT Dimension *
+ts_continous_agg_find_integer_now_func_by_materialization_id(int32 mat_htid)
+{
+	int32 raw_htid = mat_htid;
+	Dimension *par_dim = NULL;
+	while (raw_htid != INVALID_HYPERTABLE_ID)
+	{
+		Hypertable *raw_ht = ts_hypertable_get_by_id(raw_htid);
+		Dimension *open_dim = hyperspace_get_open_dimension(raw_ht->space, 0);
+		if (strlen(NameStr(open_dim->fd.integer_now_func)) != 0 &&
+			strlen(NameStr(open_dim->fd.integer_now_func_schema)) != 0)
+		{
+			par_dim = open_dim;
+			break;
+		}
+		mat_htid = raw_htid;
+		raw_htid = find_raw_hypertable_for_materialization(mat_htid);
+	}
+	return par_dim;
+}

src/continuous_agg.h

@@ -90,5 +90,8 @@ extern void ts_continuous_agg_rename_view(char *old_schema, char *name, char *ne
 extern TSDLLEXPORT int32 ts_number_of_continuous_aggs(void);
 
 extern Oid ts_continuous_agg_get_user_view_oid(ContinuousAgg *agg);
+extern TSDLLEXPORT Dimension *
+ts_continous_agg_find_integer_now_func_by_materialization_id(int32 mat_htid);
+extern ContinuousAgg *ts_continuous_agg_find_userview_name(const char *schema, const char *name);
 
 #endif /* TIMESCALEDB_CONTINUOUS_AGG_H */

src/cross_module_fn.c

@@ -347,7 +347,8 @@ static void
 continuous_agg_drop_chunks_by_chunk_id_default(int32 raw_hypertable_id, Chunk **chunks,
 											   Size num_chunks, Datum older_than_datum,
 											   Datum newer_than_datum, Oid older_than_type,
-											   Oid newer_than_type, bool cascade, int32 log_level)
+											   Oid newer_than_type, bool cascade, int32 log_level,
+											   bool user_supplied_table_name)
 {
 	error_no_default_fn_community();
 }

src/cross_module_fn.h

@@ -71,7 +71,8 @@ typedef struct CrossModuleFunctions
 
 												   Datum older_than_datum, Datum newer_than_datum,
 												   Oid older_than_type, Oid newer_than_type,
-												   bool cascade, int32 log_level);
+												   bool cascade, int32 log_level,
+												   bool user_supplied_table_name);
 	PGFunction continuous_agg_trigfn;
 	void (*continuous_agg_update_options)(ContinuousAgg *cagg,
 										  WithClauseResult *with_clause_options);

src/hypertable.c

@@ -1987,7 +1987,14 @@ hypertable_tuple_match_name(TupleInfo *ti, void *data)
 	if (!OidIsValid(schema_oid))
 		return SCAN_CONTINUE;
 
-	relid = get_relname_relid(NameStr(fd.table_name), schema_oid);
+	if (accum->schema_name == NULL)
+	{
+		/* only user visible tables will be returned */
+		relid = RelnameGetRelid(NameStr(fd.table_name));
+	}
+	else
+		relid = get_relname_relid(NameStr(fd.table_name), schema_oid);
+
 	if (!OidIsValid(relid))
 		return SCAN_CONTINUE;
 
@@ -2000,13 +2007,12 @@ hypertable_tuple_match_name(TupleInfo *ti, void *data)
 										  NameGetDatum(accum->table_name),
 										  NameGetDatum(&fd.table_name)))))
 		accum->ht_oids = lappend_oid(accum->ht_oids, relid);
-
 	return SCAN_CONTINUE;
 }
 
 /*
  * Used for drop_chunks. Either name can be NULL, which indicates matching on
- * all possible names.
+ * all possible names visible in search path.
  */
 List *
 ts_hypertable_get_all_by_name(Name schema_name, Name table_name, MemoryContext mctx)

src/interval.c

@@ -95,7 +95,7 @@ ts_interval_from_sql_input(Oid relid, Datum interval, Oid interval_type, const c
 {
 	Hypertable *hypertable;
 	Cache *hcache;
-	FormData_ts_interval *invl = palloc0(sizeof(FormData_ts_interval));
+	FormData_ts_interval *invl;
 	Oid partitioning_type;
 	Dimension *open_dim;
 
@@ -109,8 +109,33 @@ ts_interval_from_sql_input(Oid relid, Datum interval, Oid interval_type, const c
 		elog(ERROR, "internal error: no open dimension found while parsing interval");
 
 	partitioning_type = ts_dimension_get_partition_type(open_dim);
+	if (IS_INTEGER_TYPE(partitioning_type))
+	{
+		if (strlen(NameStr(open_dim->fd.integer_now_func)) == 0 ||
+			strlen(NameStr(open_dim->fd.integer_now_func_schema)) == 0)
+			ereport(ERROR,
+					(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+					 errmsg("integer_now_func not set on hypertable \"%s\"", get_rel_name(relid))));
+	}
+	invl = ts_interval_from_sql_input_internal(open_dim,
+											   interval,
+											   interval_type,
+											   parameter_name,
+											   caller_name);
 	ts_cache_release(hcache);
+	return invl;
+}
 
+/* use this variant only if the open_dim needs to be
+ * inferred for the hypertable. This is the case for continuous aggr
+ * related materialization hypertables
+ */
+TSDLLEXPORT FormData_ts_interval *
+ts_interval_from_sql_input_internal(Dimension *open_dim, Datum interval, Oid interval_type,
+									const char *parameter_name, const char *caller_name)
+{
+	FormData_ts_interval *invl = palloc0(sizeof(FormData_ts_interval));
+	Oid partitioning_type = ts_dimension_get_partition_type(open_dim);
 	switch (interval_type)
 	{
 		case INTERVALOID:
@@ -134,12 +159,6 @@ ts_interval_from_sql_input(Oid relid, Datum interval, Oid interval_type, const c
 						 errhint("integer-based time duration cannot be used with hypertables with "
 								 "a timestamp-based time dimensions")));
 
-			if (strlen(NameStr(open_dim->fd.integer_now_func)) == 0 ||
-				strlen(NameStr(open_dim->fd.integer_now_func_schema)) == 0)
-				ereport(ERROR,
-						(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
-						 errmsg("integer_now_func not set on hypertable %s", get_rel_name(relid))));
-
 			invl->is_time_interval = false;
 			invl->integer_interval = ts_time_value_to_internal(interval, interval_type);
 
@@ -277,11 +296,15 @@ ts_interval_from_now_func_get_datum(int64 interval, Oid time_dim_type, Oid now_f
 						(errcode(ERRCODE_INTERVAL_FIELD_OVERFLOW), errmsg("ts_interval overflow")));
 			return Int32GetDatum(res);
 		case INT8OID:
-			res = DatumGetInt64(now) - interval;
-			if (res > DatumGetInt64(now))
+		{
+			bool overflow = pg_sub_s64_overflow(DatumGetInt64(now), interval, &res);
+			if (overflow)
+			{
 				ereport(ERROR,
 						(errcode(ERRCODE_INTERVAL_FIELD_OVERFLOW), errmsg("ts_interval overflow")));
+			}
 			return Int64GetDatum(res);
+		}
 		default:
 			pg_unreachable();
 	}

src/interval.h

@@ -21,5 +21,7 @@ TSDLLEXPORT bool ts_interval_equal(FormData_ts_interval *invl1, FormData_ts_inte
 TSDLLEXPORT void ts_interval_now_func_validate(Oid now_func_oid, Oid open_dim_type);
 TSDLLEXPORT Datum ts_interval_subtract_from_now(FormData_ts_interval *invl, Dimension *open_dim);
 TSDLLEXPORT int64 ts_get_now_internal(Dimension *open_dim);
-
+TSDLLEXPORT FormData_ts_interval *
+ts_interval_from_sql_input_internal(Dimension *open_dim, Datum interval, Oid interval_type,
+									const char *parameter_name, const char *caller_name);
 #endif /* TIMESCALEDB_INTERVAL */