Create compressed chunks at insert time

To avoid lock contention during compression operation, we are moving
compressed chunk creation together with uncompressed chunk creation
during insert time. Now compressed chunks live while the uncompressed
chunk is alive, we don't remove them during decompression but rather
truncate them. This moves lock contention over compressed hypertable
to coincide with lock contention over uncompressed hypertable.

.unreleased/PR_5849

@@ -0,0 +1 @@
+Implements: #5849 Create compressed chunks at insert time

sql/maintenance_utils.sql

@@ -32,6 +32,10 @@ CREATE OR REPLACE FUNCTION _timescaledb_functions.create_compressed_chunk(
     numrows_post_compression BIGINT
 ) RETURNS REGCLASS AS '@MODULE_PATHNAME@', 'ts_create_compressed_chunk' LANGUAGE C STRICT VOLATILE;
 
+CREATE OR REPLACE FUNCTION _timescaledb_functions.create_compressed_chunks_for_hypertable(
+    hypertable REGCLASS
+) RETURNS VOID AS '@MODULE_PATHNAME@', 'ts_create_compressed_chunks_for_hypertable' LANGUAGE C STRICT VOLATILE;
+
 CREATE OR REPLACE FUNCTION @[email protected]_chunk(
     uncompressed_chunk REGCLASS,
     if_not_compressed BOOLEAN = false

sql/size_utils.sql

@@ -515,10 +515,10 @@ SELECT
     srcht.table_name AS hypertable_name,
     srcch.schema_name AS chunk_schema,
     srcch.table_name AS chunk_name,
-    CASE WHEN srcch.compressed_chunk_id IS NULL THEN
-        'Uncompressed'::text
-    ELSE
+    CASE WHEN srcch.status & 1 > 0 THEN
         'Compressed'::text
+    ELSE
+        'Uncompressed'::text
     END AS compression_status,
     map.uncompressed_heap_size,
     map.uncompressed_index_size,

sql/updates/latest-dev.sql

@@ -167,3 +167,6 @@ ALTER FUNCTION _timescaledb_internal.finalize_agg_ffunc(internal,text,name,name,
 ALTER FUNCTION _timescaledb_internal.finalize_agg_sfunc(internal,text,name,name,name[],bytea,anyelement) SET SCHEMA _timescaledb_functions;
 ALTER FUNCTION _timescaledb_internal.partialize_agg(anyelement) SET SCHEMA _timescaledb_functions;
 
+CREATE OR REPLACE FUNCTION _timescaledb_functions.create_compressed_chunks_for_hypertable(
+    hypertable REGCLASS
+) RETURNS VOID AS '@MODULE_PATHNAME@', 'ts_create_compressed_chunks_for_hypertable' LANGUAGE C STRICT VOLATILE;

sql/updates/post-update.sql

@@ -194,3 +194,17 @@ BEGIN
 END;
 $$;
 
+-- create compressed chunks for all uncompressed chunks for all hypertables that have compression enabled
+DO $$
+DECLARE
+  schema name;
+  ht_name name;
+BEGIN
+  FOR schema, ht_name IN
+  SELECT schema_name, table_name FROM _timescaledb_catalog.hypertable
+  WHERE compressed_hypertable_id IS NOT NULL
+  LOOP
+    PERFORM _timescaledb_functions.create_compressed_chunks_for_hypertable(schema || '.' || ht_name);
+  END LOOP;
+END;
+$$;

sql/updates/reverse-dev.sql

@@ -273,3 +273,27 @@ ALTER FUNCTION _timescaledb_functions.finalize_agg_sfunc(internal,text,name,name
 ALTER FUNCTION _timescaledb_functions.partialize_agg(anyelement) SET SCHEMA _timescaledb_internal;
 ALTER AGGREGATE _timescaledb_functions.finalize_agg(text,name,name,name[][],bytea,anyelement) SET SCHEMA _timescaledb_internal;
 
+-- remove compressed chunks for all uncompressed chunks for all hypertables that have compression enabled
+DO $$
+DECLARE
+  schema name;
+  chunk_name name;
+BEGIN
+  FOR schema, chunk_name IN
+  SELECT c.schema_name, c.table_name FROM _timescaledb_catalog.hypertable h
+  INNER JOIN _timescaledb_catalog.chunk c ON c.hypertable_id = h.id
+  WHERE h.compressed_hypertable_id IS NOT NULL
+  AND c.status & 1 = 0
+  LOOP
+    EXECUTE format('DROP TABLE %s.%s', schema, chunk_name);
+  END LOOP;
+
+  UPDATE _timescaledb_catalog.chunk SET compressed_chunk_id = NULL
+  FROM _timescaledb_catalog.chunk c
+  INNER JOIN _timescaledb_catalog.hypertable h ON c.hypertable_id = h.id
+  WHERE h.compressed_hypertable_id IS NOT NULL
+  AND c.status & 1 = 0;
+END;
+$$;
+
+DROP FUNCTION IF EXISTS _timescaledb_functions.create_compressed_chunks_for_hypertable(REGCLASS);