Add rebuild_sparse_index function

Rebuild sparse index metadata columns on compressed
chunks in-place without full recompression.

Adds _timescaledb_functions.rebuild_sparse_index(chunk,
force) which diffs chunk vs hypertable sparse index
settings, drops/adds metadata columns, then populates
them by decompressing each batch through the existing
RowDecompressor and BatchMetadataBuilder pipeline.

Author: Poroma-Banerjee

.unreleased/pr_9938

@@ -0,0 +1 @@
+Implements: #9938 Add rebuild_sparse_index function

sql/maintenance_utils.sql

@@ -59,6 +59,11 @@ CREATE OR REPLACE PROCEDURE _timescaledb_functions.rebuild_columnstore(
     chunk REGCLASS
 ) AS '@MODULE_PATHNAME@', 'ts_rebuild_columnstore' LANGUAGE C;
 
+CREATE OR REPLACE FUNCTION _timescaledb_functions.rebuild_sparse_index(
+    chunk REGCLASS,
+    force BOOLEAN = false
+) RETURNS VOID AS '@MODULE_PATHNAME@', 'ts_rebuild_sparse_index' LANGUAGE C VOLATILE;
+
 CREATE OR REPLACE PROCEDURE _timescaledb_functions.chunk_rewrite_cleanup()
 LANGUAGE C AS '@MODULE_PATHNAME@', 'ts_chunk_rewrite_cleanup';
 

sql/updates/reverse-dev.sql

@@ -188,3 +188,4 @@ DROP FUNCTION IF EXISTS @extschema@.create_hypertable(relation REGCLASS, time_co
 -- Restore the chunk_target_size check constraint dropped in the forward path.
 ALTER TABLE _timescaledb_catalog.hypertable
     ADD CONSTRAINT hypertable_chunk_target_size_check CHECK (chunk_target_size >= 0);
+DROP FUNCTION IF EXISTS _timescaledb_functions.rebuild_sparse_index(REGCLASS, BOOLEAN);

src/cross_module_fn.c

@@ -82,6 +82,7 @@ CROSSMODULE_WRAPPER(create_compressed_chunk);
 CROSSMODULE_WRAPPER(compress_chunk);
 CROSSMODULE_WRAPPER(decompress_chunk);
 CROSSMODULE_WRAPPER(rebuild_columnstore);
+CROSSMODULE_WRAPPER(rebuild_sparse_index);
 CROSSMODULE_WRAPPER(bloom1_contains);
 CROSSMODULE_WRAPPER(bloom1_contains_any);
 CROSSMODULE_WRAPPER(bloom1_contains_any_hashes);
@@ -353,6 +354,7 @@ TSDLLEXPORT CrossModuleFunctions ts_cm_functions_default = {
 	.compress_chunk = error_no_default_fn_pg_community,
 	.decompress_chunk = error_no_default_fn_pg_community,
 	.rebuild_columnstore = error_no_default_fn_pg_community,
+	.rebuild_sparse_index = error_no_default_fn_pg_community,
 	.compressed_data_decompress_forward = error_no_default_fn_pg_community,
 	.compressed_data_decompress_reverse = error_no_default_fn_pg_community,
 	.compressed_data_column_size = error_no_default_fn_pg_community,

src/cross_module_fn.h

@@ -124,6 +124,7 @@ typedef struct CrossModuleFunctions
 	PGFunction compress_chunk;
 	PGFunction decompress_chunk;
 	PGFunction rebuild_columnstore;
+	PGFunction rebuild_sparse_index;
 	void (*decompress_batches_for_insert)(ChunkInsertState *state, TupleTableSlot *slot);
 	void (*init_decompress_state_for_insert)(ChunkInsertState *state, TupleTableSlot *slot);
 	bool (*decompress_target_segments)(ModifyHypertableState *ht_state);

src/ts_catalog/compression_settings.c

@@ -28,8 +28,6 @@ static HeapTuple compression_settings_formdata_make_tuple(const FormData_compres
 														  TupleDesc desc);
 static Bitmapset *resolve_columns_to_attnos(List *column_names, Oid relid);
 static bool sparse_index_values_equal(List *left, List *right);
-static bool sparse_index_object_equal(SparseIndexSettingsObject *left,
-									  SparseIndexSettingsObject *right);
 
 /*
  * Compare two compression settings for equality
@@ -90,12 +88,56 @@ sparse_index_values_equal(List *left, List *right)
 	return true;
 }
 
+bool
+ts_sparse_index_object_get_type_and_columns(SparseIndexSettingsObject *obj, const char **type_out,
+											List **columns_out)
+{
+	const char *type = NULL;
+	List *columns = NIL;
+
+	foreach_ptr(SparseIndexSettingsPair, pair, obj->pairs)
+	{
+		if (strcmp(pair->key, ts_sparse_index_common_keys[SparseIndexKeyType]) == 0)
+		{
+			Assert(list_length(pair->values) == 1);
+			type = (const char *) lfirst(list_head(pair->values));
+		}
+		else if (strcmp(pair->key, ts_sparse_index_common_keys[SparseIndexKeyCol]) == 0)
+		{
+			columns = pair->values;
+		}
+	}
+
+	if (!type || !columns)
+	{
+		return false;
+	}
+
+	*type_out = type;
+	*columns_out = columns;
+	return true;
+}
+
+bool
+ts_sparse_index_is_orderby_source(SparseIndexSettingsObject *obj)
+{
+	foreach_ptr(SparseIndexSettingsPair, pair, obj->pairs)
+	{
+		if (strcmp(pair->key, ts_sparse_index_common_keys[SparseIndexKeySource]) == 0)
+		{
+			const char *source = (const char *) lfirst(list_head(pair->values));
+			return strcmp(source, ts_sparse_index_source_names[_SparseIndexSourceEnumOrderby]) == 0;
+		}
+	}
+	return false;
+}
+
 /*
  * Compare two sparse index objects for equality.
  * Two objects are equal if they have the same pairs with the same values.
  */
-static bool
-sparse_index_object_equal(SparseIndexSettingsObject *left, SparseIndexSettingsObject *right)
+bool
+ts_sparse_index_object_equal(SparseIndexSettingsObject *left, SparseIndexSettingsObject *right)
 {
 	if (list_length(left->pairs) != list_length(right->pairs))
 	{
@@ -165,7 +207,7 @@ ts_sparse_index_equal(const Jsonb *left, const Jsonb *right)
 		bool found = false;
 		foreach_ptr(SparseIndexSettingsObject, robj, right_settings->objects)
 		{
-			if (!already_found[ri] && sparse_index_object_equal(lobj, robj))
+			if (!already_found[ri] && ts_sparse_index_object_equal(lobj, robj))
 			{
 				already_found[ri] = true;
 				found = true;

src/ts_catalog/compression_settings.h

@@ -154,6 +154,12 @@ TSDLLEXPORT bool ts_compression_settings_equal(const CompressionSettings *left,
 TSDLLEXPORT bool ts_compression_settings_equal_with_defaults(const CompressionSettings *ht,
 															 const CompressionSettings *chunk);
 TSDLLEXPORT bool ts_sparse_index_equal(const Jsonb *left, const Jsonb *right);
+TSDLLEXPORT bool ts_sparse_index_object_equal(SparseIndexSettingsObject *left,
+											  SparseIndexSettingsObject *right);
+TSDLLEXPORT bool ts_sparse_index_object_get_type_and_columns(SparseIndexSettingsObject *obj,
+															 const char **type_out,
+															 List **columns_out);
+TSDLLEXPORT bool ts_sparse_index_is_orderby_source(SparseIndexSettingsObject *obj);
 
 TSDLLEXPORT int ts_compression_settings_update(CompressionSettings *settings);
 TSDLLEXPORT void ts_compression_settings_rename_column_cascade(Oid parent_relid, const char *old,

tsl/src/compression/api.c

@@ -1071,6 +1071,74 @@ tsl_rebuild_columnstore(PG_FUNCTION_ARGS)
 	PG_RETURN_VOID();
 }
 
+Datum
+tsl_rebuild_sparse_index(PG_FUNCTION_ARGS)
+{
+	Oid chunk_relid = PG_ARGISNULL(0) ? InvalidOid : PG_GETARG_OID(0);
+	bool force = PG_ARGISNULL(1) ? false : PG_GETARG_BOOL(1);
+
+	ts_feature_flag_check(FEATURE_HYPERTABLE_COMPRESSION);
+
+	TS_PREVENT_FUNC_IF_READ_ONLY();
+
+	if (!OidIsValid(chunk_relid))
+	{
+		ereport(ERROR, (errcode(ERRCODE_INVALID_PARAMETER_VALUE), errmsg("invalid chunk OID")));
+	}
+
+	Chunk *chunk = ts_chunk_get_by_relid(chunk_relid, true);
+	ts_hypertable_permissions_check(chunk->hypertable_relid, GetUserId());
+
+	if (!ts_chunk_is_compressed(chunk) || ts_chunk_is_frozen(chunk))
+	{
+		ereport(NOTICE,
+				(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+				 errmsg("chunk \"%s.%s\" is uncompressed or frozen, skipping",
+						NameStr(chunk->fd.schema_name),
+						NameStr(chunk->fd.table_name))));
+		PG_RETURN_VOID();
+	}
+
+	CompressionSettings *chunk_settings = ts_compression_settings_get(chunk->table_id);
+	CompressionSettings *ht_settings = ts_compression_settings_get(chunk->hypertable_relid);
+
+	if (ht_settings->fd.index == NULL)
+	{
+		ereport(NOTICE,
+				(errmsg("no sparse index configured on hypertable \"%s\" for chunk \"%s.%s\", "
+						"skipping",
+						get_rel_name(chunk->hypertable_relid),
+						NameStr(chunk->fd.schema_name),
+						NameStr(chunk->fd.table_name))));
+		PG_RETURN_VOID();
+	}
+
+	/* Orderby changes require recompression since batch data is physically sorted by orderby */
+	if (!ts_array_equal(chunk_settings->fd.orderby, ht_settings->fd.orderby))
+	{
+		ereport(NOTICE,
+				(errmsg("orderby settings for chunk \"%s.%s\" differ from hypertable \"%s\"",
+						NameStr(chunk->fd.schema_name),
+						NameStr(chunk->fd.table_name),
+						get_rel_name(chunk->hypertable_relid)),
+				 errhint("Use compress_chunk(chunk, recompress => true) to recompress.")));
+		PG_RETURN_VOID();
+	}
+
+	if (!force && ts_sparse_index_equal(chunk_settings->fd.index, ht_settings->fd.index))
+	{
+		ereport(NOTICE,
+				(errmsg("sparse index settings for chunk \"%s.%s\" already match hypertable, "
+						"skipping (use force => true to override)",
+						NameStr(chunk->fd.schema_name),
+						NameStr(chunk->fd.table_name))));
+		PG_RETURN_VOID();
+	}
+
+	rebuild_sparse_index_impl(chunk, force);
+	PG_RETURN_VOID();
+}
+
 /*
  * This is hacky but it doesn't matter. We just want to check for the existence of such an index
  * on the compressed chunk.

tsl/src/compression/api.h

@@ -16,6 +16,7 @@ extern Datum tsl_create_compressed_chunk(PG_FUNCTION_ARGS);
 extern Datum tsl_compress_chunk(PG_FUNCTION_ARGS);
 extern Datum tsl_decompress_chunk(PG_FUNCTION_ARGS);
 extern Datum tsl_rebuild_columnstore(PG_FUNCTION_ARGS);
+extern Datum tsl_rebuild_sparse_index(PG_FUNCTION_ARGS);
 extern Oid tsl_compress_chunk_wrapper(Chunk *chunk, bool if_not_compressed, bool recompress);
 extern Chunk *tsl_compression_chunk_create(Hypertable *compressed_ht, Chunk *src_chunk);
 

tsl/src/compression/batch_metadata_builder.h

@@ -8,8 +8,6 @@
 #include <postgres.h>
 #include "funcapi.h" /* for PGFunction, FmgrInfo */
 
-typedef struct RowCompressor RowCompressor;
-
 enum BatchMetadataBuilderType
 {
 	METADATA_BUILDER_MINMAX,
@@ -20,8 +18,9 @@ enum BatchMetadataBuilderType
 typedef struct BatchMetadataBuilder
 {
 	void (*update_row)(void *builder, TupleTableSlot *slot);
-	void (*insert_to_compressed_row)(void *builder, RowCompressor *compressor);
-	void (*reset)(void *builder, RowCompressor *compressor);
+	void (*insert_to_compressed_row)(void *builder, Datum *compressed_values,
+									 bool *compressed_is_null);
+	void (*reset)(void *builder, Datum *compressed_values, bool *compressed_is_null);
 	enum BatchMetadataBuilderType builder_type;
 } BatchMetadataBuilder;
 
@@ -54,7 +53,8 @@ uint64 batch_metadata_builder_bloom1_calculate_hash(PGFunction hash_function, Fm
 void batch_metadata_builder_bloom1_update_bloom_filter_with_hash(void *varlena_ptr, uint64 hash);
 void batch_metadata_builder_bloom1_insert_bloom_filter_to_compressed_row(void *bloom_varlena,
 																		 int16 bloom_attr_offset,
-																		 RowCompressor *compressor);
+																		 Datum *compressed_values,
+																		 bool *compressed_is_null);
 
 /* Returns true if the hash is maybe present in a bloom filter, if the bloom filter data is
  * NULL, it returns true, because we cannot be sure if the hash is present or not. */