Add api to associate hypertable with custom jobs

sql/job_api.sql

@@ -28,3 +28,9 @@ CREATE OR REPLACE FUNCTION @extschema@.alter_job(
 RETURNS TABLE (job_id INTEGER, schedule_interval INTERVAL, max_runtime INTERVAL, max_retries INTEGER, retry_period INTERVAL, scheduled BOOL, config JSONB, next_start TIMESTAMPTZ)
 AS '@MODULE_PATHNAME@', 'ts_job_alter'
 LANGUAGE C VOLATILE;
+
+CREATE OR REPLACE FUNCTION _timescaledb_internal.alter_job_set_hypertable_id(
+    job_id INTEGER,
+    hypertable REGCLASS )
+RETURNS INTEGER AS '@MODULE_PATHNAME@', 'ts_job_alter_set_hypertable_id'
+LANGUAGE C VOLATILE;

sql/updates/reverse-dev.sql

@@ -16,3 +16,5 @@ CREATE FUNCTION @extschema@.detach_data_node(
 AS '@MODULE_PATHNAME@', 'ts_data_node_detach' LANGUAGE C VOLATILE;
 
 DROP FUNCTION _timescaledb_internal.attach_osm_table_chunk( hypertable REGCLASS, chunk REGCLASS);
+DROP FUNCTION _timescaledb_internal.alter_job_set_hypertable_id( job_id INTEGER, hypertable REGCLASS );
+

src/bgw/job.c

@@ -635,6 +635,15 @@ bgw_job_tuple_update_by_id(TupleInfo *ti, void *const data)
 	else
 		isnull[AttrNumberGetAttrOffset(Anum_bgw_job_config)] = true;
 
+	if (updated_job->fd.hypertable_id != 0)
+	{
+		values[AttrNumberGetAttrOffset(Anum_bgw_job_hypertable_id)] =
+			Int32GetDatum(updated_job->fd.hypertable_id);
+		repl[AttrNumberGetAttrOffset(Anum_bgw_job_hypertable_id)] = true;
+	}
+	else
+		isnull[AttrNumberGetAttrOffset(Anum_bgw_job_hypertable_id)] = true;
+
 	new_tuple = heap_modify_tuple(tuple, ts_scanner_get_tupledesc(ti), values, isnull, repl);
 
 	ts_catalog_update(ti->scanrel, new_tuple);

src/cross_module_fn.c

@@ -42,6 +42,7 @@ CROSSMODULE_WRAPPER(job_add);
 CROSSMODULE_WRAPPER(job_delete);
 CROSSMODULE_WRAPPER(job_run);
 CROSSMODULE_WRAPPER(job_alter);
+CROSSMODULE_WRAPPER(job_alter_set_hypertable_id);
 
 CROSSMODULE_WRAPPER(reorder_chunk);
 CROSSMODULE_WRAPPER(move_chunk);
@@ -390,6 +391,7 @@ TSDLLEXPORT CrossModuleFunctions ts_cm_functions_default = {
 
 	.job_add = error_no_default_fn_pg_community,
 	.job_alter = error_no_default_fn_pg_community,
+	.job_alter_set_hypertable_id = error_no_default_fn_pg_community,
 	.job_delete = error_no_default_fn_pg_community,
 	.job_run = error_no_default_fn_pg_community,
 	.job_execute = job_execute_default_fn,

src/cross_module_fn.h

@@ -55,6 +55,7 @@ typedef struct CrossModuleFunctions
 
 	PGFunction job_add;
 	PGFunction job_alter;
+	PGFunction job_alter_set_hypertable_id;
 	PGFunction job_delete;
 	PGFunction job_run;
 

tsl/src/bgw_policy/job_api.c

@@ -15,6 +15,7 @@
 
 #include "job.h"
 #include "job_api.h"
+#include "hypertable_cache.h"
 
 /* Default max runtime for a custom job is unlimited for now */
 #define DEFAULT_MAX_RUNTIME 0
@@ -263,3 +264,59 @@ job_alter(PG_FUNCTION_ARGS)
 	tuple = heap_form_tuple(tupdesc, values, nulls);
 	return HeapTupleGetDatum(tuple);
 }
+
+static Hypertable *
+get_hypertable_from_oid(Cache **hcache, Oid table_oid)
+{
+	Hypertable *hypertable = NULL;
+	hypertable = ts_hypertable_cache_get_cache_and_entry(table_oid, CACHE_FLAG_MISSING_OK, hcache);
+	if (!hypertable)
+	{
+		const char *view_name = get_rel_name(table_oid);
+
+		if (!view_name)
+			ereport(ERROR,
+					(errcode(ERRCODE_UNDEFINED_OBJECT),
+					 errmsg("relation is not a hypertable or continuous aggregate")));
+		else
+		{
+			ContinuousAgg *ca = ts_continuous_agg_find_by_relid(table_oid);
+
+			if (!ca)
+				ereport(ERROR,
+						(errcode(ERRCODE_UNDEFINED_OBJECT),
+						 errmsg("relation \"%s\" is not a hypertable or continuous aggregate",
+								view_name)));
+			hypertable = ts_hypertable_get_by_id(ca->data.mat_hypertable_id);
+		}
+	}
+	Assert(hypertable != NULL);
+	return hypertable;
+}
+
+Datum
+job_alter_set_hypertable_id(PG_FUNCTION_ARGS)
+{
+	int32 job_id = PG_GETARG_INT32(0);
+	Oid table_oid = PG_GETARG_OID(1);
+	Cache *hcache = NULL;
+	Hypertable *ht = NULL;
+
+	TS_PREVENT_FUNC_IF_READ_ONLY();
+	BgwJob *job = find_job(job_id, PG_ARGISNULL(0), false /* missing_ok */);
+	if (job == NULL)
+		PG_RETURN_NULL();
+	ts_bgw_job_permission_check(job);
+
+	if (!PG_ARGISNULL(1))
+	{
+		ht = get_hypertable_from_oid(&hcache, table_oid);
+		ts_hypertable_permissions_check(ht->main_table_relid, GetUserId());
+	}
+
+	job->fd.hypertable_id = (ht != NULL ? ht->fd.id : 0);
+	ts_bgw_job_update_by_id(job_id, job);
+	if (hcache)
+		ts_cache_release(hcache);
+	PG_RETURN_INT32(job_id);
+}

tsl/src/bgw_policy/job_api.h

@@ -13,5 +13,6 @@ extern Datum job_add(PG_FUNCTION_ARGS);
 extern Datum job_alter(PG_FUNCTION_ARGS);
 extern Datum job_delete(PG_FUNCTION_ARGS);
 extern Datum job_run(PG_FUNCTION_ARGS);
+extern Datum job_alter_set_hypertable_id(PG_FUNCTION_ARGS);
 
 #endif /* TSL_BGW_POLICY_JOB_API_H */

tsl/src/init.c

@@ -108,6 +108,7 @@ CrossModuleFunctions tsl_cm_functions = {
 
 	.job_add = job_add,
 	.job_alter = job_alter,
+	.job_alter_set_hypertable_id = job_alter_set_hypertable_id,
 	.job_delete = job_delete,
 	.job_run = job_run,
 	.job_execute = job_execute,

tsl/test/expected/bgw_custom.out

@@ -525,6 +525,48 @@ SELECT count(*) FROM conditions_summary_daily;
     62
 (1 row)
 
+-- TESTs for alter_job_set_hypertable_id API
+SELECT _timescaledb_internal.alter_job_set_hypertable_id( :job_id_5, NULL);
+ alter_job_set_hypertable_id 
+-----------------------------
+                        1004
+(1 row)
+
+SELECT id, proc_name, hypertable_id 
+FROM _timescaledb_config.bgw_job WHERE id = :job_id_5;
+  id  |  proc_name   | hypertable_id 
+------+--------------+---------------
+ 1004 | custom_proc5 |              
+(1 row)
+
+-- error case, try to associate with a PG relation
+\set ON_ERROR_STOP 0
+SELECT _timescaledb_internal.alter_job_set_hypertable_id( :job_id_5, 'custom_log');
+ERROR:  relation "custom_log" is not a hypertable or continuous aggregate
+\set ON_ERROR_STOP 1
+-- TEST associate the cagg with the job
+SELECT _timescaledb_internal.alter_job_set_hypertable_id( :job_id_5, 'conditions_summary_daily'::regclass);
+ alter_job_set_hypertable_id 
+-----------------------------
+                        1004
+(1 row)
+
+SELECT id, proc_name, hypertable_id 
+FROM _timescaledb_config.bgw_job WHERE id = :job_id_5;
+  id  |  proc_name   | hypertable_id 
+------+--------------+---------------
+ 1004 | custom_proc5 |             3
+(1 row)
+
+--verify that job is dropped when cagg is dropped
+DROP MATERIALIZED VIEW conditions_summary_daily;
+NOTICE:  drop cascades to table _timescaledb_internal._hyper_3_10_chunk
+SELECT id, proc_name, hypertable_id 
+FROM _timescaledb_config.bgw_job WHERE id = :job_id_5;
+ id | proc_name | hypertable_id 
+----+-----------+---------------
+(0 rows)
+
 -- Stop Background Workers
 SELECT _timescaledb_internal.stop_background_workers();
  stop_background_workers 

tsl/test/shared/expected/extension.out

@@ -19,6 +19,7 @@ FROM pg_proc p
     e.oid = d.refobjid
 WHERE proname <> 'get_telemetry_report'
 ORDER BY pronamespace::regnamespace::text COLLATE "C", p.oid::regprocedure::text COLLATE "C";
+ _timescaledb_internal.alter_job_set_hypertable_id(integer,regclass)
  _timescaledb_internal.attach_osm_table_chunk(regclass,regclass)
  _timescaledb_internal.bookend_deserializefunc(bytea,internal)
  _timescaledb_internal.bookend_finalfunc(internal,anyelement,"any")

tsl/test/sql/bgw_custom.sql

@@ -292,5 +292,28 @@ SELECT add_job('custom_proc5', '1h', config := '{"type":"procedure"}'::jsonb, in
 SELECT wait_for_job_to_run(:job_id_5, 1);
 SELECT count(*) FROM conditions_summary_daily;
 
+-- TESTs for alter_job_set_hypertable_id API
+
+SELECT _timescaledb_internal.alter_job_set_hypertable_id( :job_id_5, NULL);
+SELECT id, proc_name, hypertable_id 
+FROM _timescaledb_config.bgw_job WHERE id = :job_id_5;
+
+-- error case, try to associate with a PG relation
+\set ON_ERROR_STOP 0
+SELECT _timescaledb_internal.alter_job_set_hypertable_id( :job_id_5, 'custom_log');
+\set ON_ERROR_STOP 1
+
+-- TEST associate the cagg with the job
+SELECT _timescaledb_internal.alter_job_set_hypertable_id( :job_id_5, 'conditions_summary_daily'::regclass);
+
+SELECT id, proc_name, hypertable_id 
+FROM _timescaledb_config.bgw_job WHERE id = :job_id_5;
+
+--verify that job is dropped when cagg is dropped
+DROP MATERIALIZED VIEW conditions_summary_daily;
+
+SELECT id, proc_name, hypertable_id 
+FROM _timescaledb_config.bgw_job WHERE id = :job_id_5;
+
 -- Stop Background Workers
 SELECT _timescaledb_internal.stop_background_workers();