Add support for changing chunk schema and name

Previously, chunks could be renamed and have their schema changed, but
the metadata in the TimescaleDB catalog was not updated in a
corresponding way. Further, renaming a chunk column was possible,
which could break functionality on the hypertable.

Then catalog metadata is now properly updated on name and schema
changes applied to chunks. Renaming chunk columns have been blocked
with an informational error message.

Author: erimatnor

src/chunk.c

@@ -1142,3 +1142,57 @@ chunk_recreate_all_constraints_for_dimension(Hyperspace *hs, int32 dimension_id)
 	chunk_scan_ctx_foreach_chunk(&chunkctx, chunk_recreate_constraint, 0);
 	chunk_scan_ctx_destroy(&chunkctx);
 }
+
+static bool
+chunk_tuple_update(TupleInfo *ti, void *data)
+{
+	HeapTuple	tuple = heap_copytuple(ti->tuple);
+	FormData_chunk *form = (FormData_chunk *) GETSTRUCT(tuple);
+	FormData_chunk *update = data;
+	CatalogSecurityContext sec_ctx;
+
+	namecpy(&form->schema_name, &update->schema_name);
+	namecpy(&form->table_name, &update->table_name);
+
+	catalog_become_owner(catalog_get(), &sec_ctx);
+	catalog_update(ti->scanrel, tuple);
+	catalog_restore_user(&sec_ctx);
+
+	heap_freetuple(tuple);
+
+	return false;
+}
+
+static bool
+chunk_update_form(FormData_chunk *form)
+{
+	ScanKeyData scankey[1];
+
+	ScanKeyInit(&scankey[0], Anum_chunk_idx_id, BTEqualStrategyNumber,
+				F_INT4EQ, form->id);
+
+	return chunk_scan_internal(CHUNK_ID_INDEX,
+							   scankey,
+							   1,
+							   chunk_tuple_update,
+							   form,
+							   0,
+							   AccessShareLock,
+							   CurrentMemoryContext) > 0;
+}
+
+bool
+chunk_set_name(Chunk *chunk, const char *newname)
+{
+	namestrcpy(&chunk->fd.table_name, newname);
+
+	return chunk_update_form(&chunk->fd);
+}
+
+bool
+chunk_set_schema(Chunk *chunk, const char *newschema)
+{
+	namestrcpy(&chunk->fd.schema_name, newschema);
+
+	return chunk_update_form(&chunk->fd);
+}

src/chunk.h

@@ -77,6 +77,8 @@ extern void chunk_recreate_all_constraints_for_dimension(Hyperspace *hs, int32 d
 extern int	chunk_delete_by_relid(Oid chunk_oid);
 extern int	chunk_delete_by_hypertable_id(int32 hypertable_id);
 extern int	chunk_delete_by_name(const char *schema, const char *table);
+extern bool chunk_set_name(Chunk *chunk, const char *newname);
+extern bool chunk_set_schema(Chunk *chunk, const char *newschema);
 
 #define chunk_get_by_name(schema_name, table_name, num_constraints, fail_if_not_found) \
 	chunk_get_by_name_with_memory_context(schema_name, table_name, num_constraints, \

src/process_utility.c

@@ -141,7 +141,7 @@ relation_not_only(RangeVar *rv)
 				 errmsg("ONLY option not supported on hypertable operations")));
 }
 
-/* Change the schema of a hypertable */
+/* Change the schema of a hypertable or a chunk */
 static void
 process_alterobjectschema(Node *parsetree)
 {
@@ -161,7 +161,14 @@ process_alterobjectschema(Node *parsetree)
 	hcache = hypertable_cache_pin();
 	ht = hypertable_cache_get_entry(hcache, relid);
 
-	if (ht != NULL)
+	if (ht == NULL)
+	{
+		Chunk	   *chunk = chunk_get_by_relid(relid, 0, false);
+
+		if (NULL != chunk)
+			chunk_set_schema(chunk, alterstmt->newschema);
+	}
+	else
 		hypertable_set_schema(ht, alterstmt->newschema);
 
 	cache_release(hcache);
@@ -626,12 +633,22 @@ process_reindex(Node *parsetree)
 	return ret;
 }
 
+/*
+ * Rename a hypertable or a chunk.
+ */
 static void
 process_rename_table(Cache *hcache, Oid relid, RenameStmt *stmt)
 {
 	Hypertable *ht = hypertable_cache_get_entry(hcache, relid);
 
-	if (NULL != ht)
+	if (NULL == ht)
+	{
+		Chunk	   *chunk = chunk_get_by_relid(relid, 0, false);
+
+		if (NULL != chunk)
+			chunk_set_name(chunk, stmt->newname);
+	}
+	else
 		hypertable_set_name(ht, stmt->newname);
 }
 
@@ -642,7 +659,17 @@ process_rename_column(Cache *hcache, Oid relid, RenameStmt *stmt)
 	Dimension  *dim;
 
 	if (NULL == ht)
+	{
+		Chunk	   *chunk = chunk_get_by_relid(relid, 0, false);
+
+		if (NULL != chunk)
+			ereport(ERROR,
+					(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
+					 errmsg("cannot rename column \"%s\" of hypertable chunk \"%s\"",
+							stmt->subname, get_rel_name(relid)),
+					 errhint("Rename the hypertable column instead.")));
 		return;
+	}
 
 	dim = hyperspace_get_dimension_by_name(ht->space, DIMENSION_TYPE_ANY, stmt->subname);
 

test/expected/alter.out

@@ -130,3 +130,32 @@ SELECT * FROM alter_after;
  Tue Aug 22 09:19:14 2017 | 12.5 |       3 |  7
 (7 rows)
 
+-- Need superuser to ALTER chunks in _timescaledb_internal schema
+\c single :ROLE_SUPERUSER
+SELECT * FROM _timescaledb_catalog.chunk WHERE id = 2;
+ id | hypertable_id |      schema_name      |    table_name    
+----+---------------+-----------------------+------------------
+  2 |             2 | _timescaledb_internal | _hyper_2_2_chunk
+(1 row)
+
+-- Rename chunk
+ALTER TABLE _timescaledb_internal._hyper_2_2_chunk RENAME TO new_chunk_name;
+SELECT * FROM _timescaledb_catalog.chunk WHERE id = 2;
+ id | hypertable_id |      schema_name      |   table_name   
+----+---------------+-----------------------+----------------
+  2 |             2 | _timescaledb_internal | new_chunk_name
+(1 row)
+
+-- Set schema
+ALTER TABLE _timescaledb_internal.new_chunk_name SET SCHEMA public;
+SELECT * FROM _timescaledb_catalog.chunk WHERE id = 2;
+ id | hypertable_id | schema_name |   table_name   
+----+---------------+-------------+----------------
+  2 |             2 | public      | new_chunk_name
+(1 row)
+
+-- Test that we cannot rename chunk columns
+\set ON_ERROR_STOP 0
+ALTER TABLE public.new_chunk_name RENAME COLUMN time TO newtime;
+ERROR:  cannot rename column "time" of hypertable chunk "new_chunk_name"
+\set ON_ERROR_STOP 1

test/sql/alter.sql

@@ -56,3 +56,20 @@ AND a.attnum > 0
 ORDER BY c.relname, a.attnum;
 
 SELECT * FROM alter_after;
+
+-- Need superuser to ALTER chunks in _timescaledb_internal schema
+\c single :ROLE_SUPERUSER
+SELECT * FROM _timescaledb_catalog.chunk WHERE id = 2;
+
+-- Rename chunk
+ALTER TABLE _timescaledb_internal._hyper_2_2_chunk RENAME TO new_chunk_name;
+SELECT * FROM _timescaledb_catalog.chunk WHERE id = 2;
+
+-- Set schema
+ALTER TABLE _timescaledb_internal.new_chunk_name SET SCHEMA public;
+SELECT * FROM _timescaledb_catalog.chunk WHERE id = 2;
+
+-- Test that we cannot rename chunk columns
+\set ON_ERROR_STOP 0
+ALTER TABLE public.new_chunk_name RENAME COLUMN time TO newtime;
+\set ON_ERROR_STOP 1