direct_compress_insert.sql

-- This file and its contents are licensed under the Timescale License.
-- Please see the included NOTICE for copyright information and
-- LICENSE-TIMESCALE for a copy of the license.

CREATE TABLE metrics (time TIMESTAMPTZ NOT NULL, device TEXT, value float) WITH (tsdb.hypertable, tsdb.orderby='time');

-- first try without the GUCs
BEGIN;
INSERT INTO metrics SELECT '2025-01-01'::timestamptz + (i || ' hour')::interval, 'd1', i::float FROM generate_series(0,100) i;
EXPLAIN (ANALYZE, BUFFERS OFF, COSTS OFF, SUMMARY OFF, TIMING OFF) SELECT * FROM metrics;
-- should have no status aka normal uncompressed chunk
SELECT DISTINCT _timescaledb_functions.chunk_status_text(chunk) FROM show_chunks('metrics') chunk;
ROLLBACK;

SET timescaledb.enable_direct_compress_insert = true;
SET timescaledb.enable_direct_compress_insert_sort_batches = true;
SET timescaledb.enable_direct_compress_insert_client_sorted = false;

-- EXPLAIN with too small batch
EXPLAIN (BUFFERS OFF, COSTS OFF, SUMMARY OFF, TIMING OFF) INSERT INTO metrics SELECT '2025-01-01'::timestamptz, 'd1', i::float FROM generate_series(0,5) i;
-- EXPLAIN with large enough batch
EXPLAIN (BUFFERS OFF, COSTS OFF, SUMMARY OFF, TIMING OFF) INSERT INTO metrics SELECT '2025-01-01'::timestamptz, 'd1', i::float FROM generate_series(0,500) i;

-- EXPLAIN with debug GUC disabled
SET timescaledb.enable_optimizations TO OFF;
EXPLAIN (BUFFERS OFF, COSTS OFF, SUMMARY OFF, TIMING OFF) INSERT INTO metrics SELECT '2025-01-01'::timestamptz, 'd1', i::float FROM generate_series(0,500) i;
RESET timescaledb.enable_optimizations;

-- simple test with compressed insert enabled
BEGIN;
INSERT INTO metrics SELECT '2025-01-01'::timestamptz + (i || ' minute')::interval, 'd1', i::float FROM generate_series(0,3000) i;
EXPLAIN (ANALYZE, BUFFERS OFF, COSTS OFF, SUMMARY OFF, TIMING OFF) SELECT * FROM metrics;
SELECT first(time,rn), last(time,rn) FROM (SELECT ROW_NUMBER() OVER () as rn, time FROM metrics) sub;
-- since the chunks are new status should be COMPRESSED, UNORDERED
SELECT DISTINCT _timescaledb_functions.chunk_status_text(chunk) FROM show_chunks('metrics') chunk;
ROLLBACK;

-- simple test with compressed insert enabled and reversed order
BEGIN;
INSERT INTO metrics SELECT '2025-01-01'::timestamptz - (i || ' minute')::interval, 'd1', i::float FROM generate_series(0,3000) i;
EXPLAIN (ANALYZE, BUFFERS OFF, COSTS OFF, SUMMARY OFF, TIMING OFF) SELECT * FROM metrics;
SELECT first(time,rn), last(time,rn) FROM (SELECT ROW_NUMBER() OVER () as rn, time FROM metrics) sub;
-- since the chunks are new status should be COMPRESSED, UNORDERED
SELECT DISTINCT _timescaledb_functions.chunk_status_text(chunk) FROM show_chunks('metrics') chunk;
ROLLBACK;

SET timescaledb.enable_direct_compress_insert_sort_batches = false;
-- simple test with compressed insert enabled and without batch sorting
BEGIN;
INSERT INTO metrics SELECT '2025-01-01'::timestamptz + (i || ' minute')::interval, 'd1', i::float FROM generate_series(0,3000) i;
EXPLAIN (ANALYZE, BUFFERS OFF, COSTS OFF, SUMMARY OFF, TIMING OFF) SELECT * FROM metrics;
SELECT first(time,rn), last(time,rn) FROM (SELECT ROW_NUMBER() OVER () as rn, time FROM metrics) sub;
-- since the chunks are new status should be COMPRESSED, UNORDERED
SELECT DISTINCT _timescaledb_functions.chunk_status_text(chunk) FROM show_chunks('metrics') chunk;
ROLLBACK;

-- simple test with compressed insert enabled and reversed order and no batch sorting
BEGIN;
INSERT INTO metrics SELECT '2025-01-01'::timestamptz - (i || ' minute')::interval, 'd1', i::float FROM generate_series(0,3000) i;
EXPLAIN (ANALYZE, BUFFERS OFF, COSTS OFF, SUMMARY OFF, TIMING OFF) SELECT * FROM metrics;
SELECT first(time,rn), last(time,rn) FROM (SELECT ROW_NUMBER() OVER () as rn, time FROM metrics) sub;
-- since the chunks are new status should be COMPRESSED, UNORDERED
SELECT DISTINCT _timescaledb_functions.chunk_status_text(chunk) FROM show_chunks('metrics') chunk;
ROLLBACK;

-- test compressing into uncompressed chunk
RESET timescaledb.enable_direct_compress_insert;
RESET timescaledb.enable_direct_compress_insert_sort_batches;
RESET timescaledb.enable_direct_compress_insert_client_sorted;

BEGIN;
INSERT INTO metrics SELECT '2025-01-01'::timestamptz + (i || ' minute')::interval, 'd1', i::float FROM generate_series(0,3000) i;
SET timescaledb.enable_direct_compress_insert = true;
SET timescaledb.enable_direct_compress_insert_client_sorted = true;
INSERT INTO metrics SELECT '2025-01-01'::timestamptz + (i || ' minute')::interval, 'd1', i::float FROM generate_series(0,3000) i;
EXPLAIN (ANALYZE, BUFFERS OFF, COSTS OFF, SUMMARY OFF, TIMING OFF) SELECT * FROM metrics;
-- since the chunks are new status should be COMPRESSED, PARTIAL
SELECT DISTINCT _timescaledb_functions.chunk_status_text(chunk) FROM show_chunks('metrics') chunk;
ROLLBACK;

-- simple test with compressed insert enabled and reversed order
BEGIN;
SET timescaledb.enable_direct_compress_insert = true;
SET timescaledb.enable_direct_compress_insert_client_sorted = true;
INSERT INTO metrics SELECT '2025-01-01'::timestamptz - (i || ' minute')::interval, 'd1', i::float FROM generate_series(0,3000) i;
EXPLAIN (ANALYZE, BUFFERS OFF, COSTS OFF, SUMMARY OFF, TIMING OFF) SELECT * FROM metrics;
-- since the chunks are new status should be COMPRESSED
SELECT DISTINCT _timescaledb_functions.chunk_status_text(chunk) FROM show_chunks('metrics') chunk;
ROLLBACK;

-- simple test with compressed insert enabled and no presorted
BEGIN;
SET timescaledb.enable_direct_compress_insert = true;
SET timescaledb.enable_direct_compress_insert_client_sorted = false;
INSERT INTO metrics SELECT '2025-01-01'::timestamptz - (i || ' minute')::interval, 'd1', i::float FROM generate_series(0,3000) i;
EXPLAIN (ANALYZE, BUFFERS OFF, COSTS OFF, SUMMARY OFF, TIMING OFF) SELECT * FROM metrics;
-- since the chunks are new status should be COMPRESSED, UNORDERED
SELECT DISTINCT _timescaledb_functions.chunk_status_text(chunk) FROM show_chunks('metrics') chunk;
ROLLBACK;

-- simple test with compressed insert enabled and no presorted and with uncompressed data
BEGIN;
SET timescaledb.enable_direct_compress_insert = true;
SET timescaledb.enable_direct_compress_insert_client_sorted = false;
INSERT INTO metrics SELECT '2025-01-01'::timestamptz, 'd1', 0;
INSERT INTO metrics SELECT '2025-01-01'::timestamptz - (i || ' minute')::interval, 'd1', i::float FROM generate_series(0,3000) i;
EXPLAIN (ANALYZE, BUFFERS OFF, COSTS OFF, SUMMARY OFF, TIMING OFF) SELECT * FROM metrics;
-- since the chunks are new status should be COMPRESSED, UNORDERED, PARTIAL
SELECT DISTINCT _timescaledb_functions.chunk_status_text(chunk) FROM show_chunks('metrics') chunk;
ROLLBACK;

-- simple test with compressed insert enabled and no presorted with partial and compressed chunks
BEGIN;
SET timescaledb.enable_direct_compress_insert = true;
SET timescaledb.enable_direct_compress_insert_client_sorted = false;
INSERT INTO metrics SELECT '2025-01-01'::timestamptz, 'd1', 0;
INSERT INTO metrics SELECT '2025-01-01'::timestamptz + (i || ' minute')::interval, 'd1', i::float FROM generate_series(0,3000) i;
INSERT INTO metrics SELECT '2025-01-02'::timestamptz + (i || ' minute')::interval, 'd1', i::float FROM generate_series(0,3000) i;
EXPLAIN (ANALYZE, BUFFERS OFF, COSTS OFF, SUMMARY OFF, TIMING OFF) SELECT * FROM metrics;
SELECT first(time,rn), last(time,rn) FROM (SELECT ROW_NUMBER() OVER () as rn, time FROM metrics) sub;
SELECT compress_relid::text AS "COMPRESSED_CHUNK" FROM _timescaledb_catalog.compression_settings WHERE compress_relid IS NOT NULL order by 1 desc limit 1 \gset
-- should see overlapping batches
select _ts_meta_count, _ts_meta_min_1, _ts_meta_max_1 from :COMPRESSED_CHUNK order by 2;
-- since the chunks are new status should be COMPRESSED, UNORDERED, PARTIAL and COMPRESSED, UNORDERED
SELECT DISTINCT _timescaledb_functions.chunk_status_text(chunk) FROM show_chunks('metrics') chunk order by 1;
ROLLBACK;

-- simple test with compressed insert enabled and presorted with partial and compressed chunks
BEGIN;
SET timescaledb.enable_direct_compress_insert = true;
SET timescaledb.enable_direct_compress_insert_client_sorted = true;
INSERT INTO metrics SELECT '2025-01-01'::timestamptz, 'd1', 0;
INSERT INTO metrics SELECT '2025-01-01'::timestamptz + (i || ' minute')::interval, 'd1', i::float FROM generate_series(0,3000) i;
INSERT INTO metrics SELECT '2025-01-05'::timestamptz + (i || ' minute')::interval, 'd1', i::float FROM generate_series(0,3000) i;
EXPLAIN (ANALYZE, BUFFERS OFF, COSTS OFF, SUMMARY OFF, TIMING OFF) SELECT * FROM metrics;
SELECT first(time,rn), last(time,rn) FROM (SELECT ROW_NUMBER() OVER () as rn, time FROM metrics) sub;
SELECT compress_relid::text AS "COMPRESSED_CHUNK" FROM _timescaledb_catalog.compression_settings WHERE compress_relid IS NOT NULL order by 1 desc limit 1 \gset
-- should not see overlapping batches
select _ts_meta_count, _ts_meta_min_1, _ts_meta_max_1 from :COMPRESSED_CHUNK order by 2;
-- since the chunks are new status should be COMPRESSED, UNORDERED, PARTIAL and COMPRESSED, UNORDERED
SELECT DISTINCT _timescaledb_functions.chunk_status_text(chunk) FROM show_chunks('metrics') chunk order by 1;
ROLLBACK;

-- test with segmentby
BEGIN;
ALTER TABLE metrics SET (tsdb.segmentby = 'device');
SET timescaledb.enable_direct_compress_insert = true;
SET timescaledb.enable_direct_compress_insert_client_sorted = true;
INSERT INTO metrics SELECT '2025-01-01'::timestamptz - (i || ' minute')::interval, floor(i), i::float FROM generate_series(0.0,9.8,0.2) i;
EXPLAIN (ANALYZE, BUFFERS OFF, COSTS OFF, SUMMARY OFF, TIMING OFF) SELECT * FROM metrics;
SELECT compress_relid::text AS "COMPRESSED_CHUNK" FROM _timescaledb_catalog.compression_settings WHERE compress_relid IS NOT NULL \gset
-- should have 10 batches
SELECT count(*) FROM :COMPRESSED_CHUNK;
-- since the chunks are new status should be COMPRESSED
SELECT DISTINCT _timescaledb_functions.chunk_status_text(chunk) FROM show_chunks('metrics') chunk;
ROLLBACK;

-- segmentby with overlapping batches
BEGIN;
ALTER TABLE metrics SET (tsdb.segmentby = 'device');
SET timescaledb.enable_direct_compress_insert = true;
SET timescaledb.enable_direct_compress_insert_client_sorted = false;
INSERT INTO metrics SELECT '2025-01-01'::timestamptz + (i || ' minute')::interval, 'd'||i%2, i::float FROM generate_series(0,3000) i;
INSERT INTO metrics SELECT '2025-01-02'::timestamptz + (i || ' minute')::interval, 'd'||i%2, i::float FROM generate_series(0,3000) i;
EXPLAIN (ANALYZE, BUFFERS OFF, COSTS OFF, SUMMARY OFF, TIMING OFF) SELECT * FROM metrics;
SELECT compress_relid::text AS "COMPRESSED_CHUNK" FROM _timescaledb_catalog.compression_settings WHERE compress_relid IS NOT NULL order by 1 desc limit 1 \gset
-- should see overlapping batches per device
select _ts_meta_count, _ts_meta_min_1, _ts_meta_max_1, device from :COMPRESSED_CHUNK order by 4, 2;
-- since the chunks are new status should be COMPRESSED, UNORDERED
SELECT DISTINCT _timescaledb_functions.chunk_status_text(chunk) FROM show_chunks('metrics') chunk;
ROLLBACK;

-- multikey orderby
BEGIN;
ALTER TABLE metrics SET (tsdb.orderby = 'device desc,time');
SET timescaledb.enable_direct_compress_insert = true;
SET timescaledb.enable_direct_compress_insert_client_sorted = false;
INSERT INTO metrics SELECT '2025-01-01'::timestamptz + (i || ' minute')::interval, 'd'||i%3, i::float FROM generate_series(0,3000) i;
INSERT INTO metrics SELECT '2025-01-02'::timestamptz - (i || ' minute')::interval, 'd'||i%3, i::float FROM generate_series(0,3000) i;
EXPLAIN (ANALYZE, BUFFERS OFF, COSTS OFF, SUMMARY OFF, TIMING OFF) SELECT * FROM metrics;
SELECT compress_relid::text AS "COMPRESSED_CHUNK" FROM _timescaledb_catalog.compression_settings WHERE compress_relid IS NOT NULL order by 1 limit 1 \gset
-- should see overlapping batches
select _ts_meta_count, _ts_meta_min_1, _ts_meta_max_1, _ts_meta_min_2, _ts_meta_max_2 from :COMPRESSED_CHUNK order by 2, 4;
-- since the chunks are new status should be COMPRESSED, UNORDERED
SELECT DISTINCT _timescaledb_functions.chunk_status_text(chunk) FROM show_chunks('metrics') chunk;
ROLLBACK;

-- test unique constraints prevent direct compress
BEGIN;
SET timescaledb.enable_direct_compress_insert = true;
ALTER TABLE metrics ADD CONSTRAINT unique_time_device UNIQUE (time, device);
INSERT INTO metrics SELECT '2025-01-01'::timestamptz + (i || ' minute')::interval, 'd1', i::float FROM generate_series(0,100) i;
EXPLAIN (ANALYZE, BUFFERS OFF, COSTS OFF, SUMMARY OFF, TIMING OFF) SELECT * FROM metrics;
SELECT DISTINCT status FROM _timescaledb_catalog.chunk WHERE relid IN (SELECT relid FROM _timescaledb_catalog.compression_settings WHERE compress_relid IS NOT NULL);
ROLLBACK;

-- test triggers prevent direct compress
BEGIN;
SET timescaledb.enable_direct_compress_insert = true;
CREATE OR REPLACE FUNCTION test_trigger() RETURNS TRIGGER AS $$ BEGIN RETURN NEW; END; $$ LANGUAGE plpgsql;
CREATE TRIGGER metrics_trigger BEFORE INSERT OR UPDATE ON metrics FOR EACH ROW EXECUTE FUNCTION test_trigger();
INSERT INTO metrics SELECT '2025-01-01'::timestamptz + (i || ' minute')::interval, 'd1', i::float FROM generate_series(0,100) i;
EXPLAIN (ANALYZE, BUFFERS OFF, COSTS OFF, SUMMARY OFF, TIMING OFF) SELECT * FROM metrics;
SELECT DISTINCT status FROM _timescaledb_catalog.chunk WHERE relid IN (SELECT relid FROM _timescaledb_catalog.compression_settings WHERE compress_relid IS NOT NULL);
ROLLBACK;

-- test caggs with direct compress
BEGIN;
SET timescaledb.enable_direct_compress_insert = true;
CREATE MATERIALIZED VIEW metrics_cagg WITH (tsdb.continuous) AS SELECT time_bucket('1 hour', time) AS bucket, device, avg(value) AS avg_value FROM metrics GROUP BY bucket, device WITH NO DATA;
INSERT INTO metrics SELECT '2025-01-01'::timestamptz + (i || ' minute')::interval, 'd1', i::float FROM generate_series(0,100) i;
EXPLAIN (ANALYZE, BUFFERS OFF, COSTS OFF, SUMMARY OFF, TIMING OFF) SELECT * FROM metrics;
SELECT DISTINCT status FROM _timescaledb_catalog.chunk WHERE relid IN (SELECT relid FROM _timescaledb_catalog.compression_settings WHERE compress_relid IS NOT NULL);
ROLLBACK;

-- cagg + direct compress where the cagg time column is segmentby.
-- segmentby columns don't get _ts_meta_min/max columns, so no MINMAX builder
-- is created for them; the flush path must still derive an invalidation range
-- from the segmentby value rather than crashing.
BEGIN;
CREATE TABLE metrics_segmentby_time (time TIMESTAMPTZ NOT NULL, device TEXT, value float)
    WITH (tsdb.hypertable);
CREATE MATERIALIZED VIEW metrics_segmentby_time_cagg WITH (tsdb.continuous) AS
    SELECT time_bucket('1 hour', time) AS bucket, device, avg(value) AS avg_value
    FROM metrics_segmentby_time GROUP BY bucket, device WITH NO DATA;
ALTER TABLE metrics_segmentby_time SET (
    timescaledb.compress,
    timescaledb.compress_segmentby = 'time'
);
INSERT INTO metrics_segmentby_time
SELECT '2025-01-01'::timestamptz + (i * interval '5 minutes'), 'd1', random()
FROM generate_series(1, 50) i;
SET timescaledb.enable_direct_compress_insert = true;
INSERT INTO metrics_segmentby_time
SELECT '2025-02-01'::timestamptz + (i * interval '1 second'), 'd1', random()
FROM generate_series(1, 5000) i;
-- The new chunk should be in COMPRESSED state, confirming the flush path ran.
SELECT _timescaledb_functions.chunk_status_text(chunk)
  FROM show_chunks('metrics_segmentby_time') chunk
 WHERE chunk::text LIKE '%hyper_%'
 ORDER BY chunk::text;
ROLLBACK;

-- test chunk status handling
CREATE TABLE metrics_status(time timestamptz) WITH (tsdb.hypertable,tsdb.partition_column='time');

INSERT INTO metrics_status SELECT '2025-01-01';
-- normal insert should result in chunk status 0
SELECT _timescaledb_functions.chunk_status_text(chunk) FROM show_chunks('metrics_status') chunk;

SET timescaledb.enable_direct_compress_insert = true;

BEGIN;
INSERT INTO metrics_status SELECT '2025-01-01' FROM generate_series(1,9);
-- small insert batches should not result in compressed chunk
SELECT _timescaledb_functions.chunk_status_text(chunk) FROM show_chunks('metrics_status') chunk;
ROLLBACK;

BEGIN;
INSERT INTO metrics_status SELECT '2025-01-01' FROM generate_series(1,10);
-- status should be COMPRESSED, UNORDERED, PARTIAL since we have more than 10 rows in the chunk
SELECT _timescaledb_functions.chunk_status_text(chunk) FROM show_chunks('metrics_status') chunk;
ROLLBACK;

BEGIN;
-- compressed sorted copy into uncompressed chunk should result in chunk status 9 (compressed,partial)
SET timescaledb.enable_direct_compress_insert = true;
SET timescaledb.enable_direct_compress_insert_client_sorted = true;
INSERT INTO metrics_status SELECT '2025-01-01'::timestamptz + (i || ' minute')::interval FROM generate_series(0,100) i;
-- status should be COMPRESSED, PARTIAL
SELECT _timescaledb_functions.chunk_status_text(chunk) FROM show_chunks('metrics_status') chunk;
ROLLBACK;

TRUNCATE metrics_status;

BEGIN;
-- compressed insert into new chunk should result in chunk status 3 (compressed,unordered)
SET timescaledb.enable_direct_compress_insert = true;
SET timescaledb.enable_direct_compress_insert_client_sorted = false;
INSERT INTO metrics_status SELECT '2025-01-01'::timestamptz + (i || ' minute')::interval FROM generate_series(0,100) i;
-- status should be COMPRESSED, UNORDERED
SELECT _timescaledb_functions.chunk_status_text(chunk) FROM show_chunks('metrics_status') chunk;
ROLLBACK;

BEGIN;
-- compressed sorted copy into new chunk should result in chunk status 1 (compressed)
SET timescaledb.enable_direct_compress_insert = true;
SET timescaledb.enable_direct_compress_insert_client_sorted = true;
INSERT INTO metrics_status SELECT '2025-01-01'::timestamptz + (i || ' minute')::interval FROM generate_series(0,100) i;
-- status should be COMPRESSED
SELECT _timescaledb_functions.chunk_status_text(chunk) FROM show_chunks('metrics_status') chunk;
ROLLBACK;

SET timescaledb.enable_direct_compress_insert = false;
SET timescaledb.enable_direct_compress_insert_client_sorted = false;
INSERT INTO metrics_status SELECT '2025-01-01';
-- no status aka normal uncompressed chunk
SELECT _timescaledb_functions.chunk_status_text(chunk) FROM show_chunks('metrics_status') chunk;
SELECT compress_chunk(show_chunks('metrics_status'));
-- status should be COMPRESSED
SELECT _timescaledb_functions.chunk_status_text(chunk) FROM show_chunks('metrics_status') chunk;

BEGIN;
-- compressed insert into fully compressed chunk should result in chunk status 3 (compressed,unordered)
SET timescaledb.enable_direct_compress_insert = true;
SET timescaledb.enable_direct_compress_insert_client_sorted = false;
INSERT INTO metrics_status SELECT '2025-01-01'::timestamptz + (i || ' minute')::interval FROM generate_series(0,100) i;
-- status should be COMPRESSED, UNORDERED
SELECT _timescaledb_functions.chunk_status_text(chunk) FROM show_chunks('metrics_status') chunk;
ROLLBACK;

BEGIN;
-- compressed insert new chunk should result in chunk status 1 (compressed)
SET timescaledb.enable_direct_compress_insert = true;
SET timescaledb.enable_direct_compress_insert_client_sorted = true;
INSERT INTO metrics_status SELECT '2025-01-01'::timestamptz + (i || ' minute')::interval FROM generate_series(0,100) i;
-- status should be COMPRESSED
SELECT _timescaledb_functions.chunk_status_text(chunk) FROM show_chunks('metrics_status') chunk;
ROLLBACK;

-- test direct compress into chunk directly
CREATE TABLE metrics_chunk(time timestamptz) WITH (tsdb.hypertable,tsdb.partition_column='time');
SET timescaledb.enable_direct_compress_insert = true;

-- create uncompressed chunk
INSERT INTO metrics_chunk SELECT '2025-01-01';
-- status should be normal uncompressed chunk since it was single tuple insert
SELECT _timescaledb_functions.chunk_status_text(chunk) FROM show_chunks('metrics_chunk') chunk;
SELECT show_chunks('metrics_chunk') AS "CHUNK" \gset

EXPLAIN (costs off,summary off,timing off) INSERT INTO :CHUNK SELECT '2025-01-01'::timestamptz + (i || ' minute')::interval FROM generate_series(0,100) i;

BEGIN;
INSERT INTO :CHUNK SELECT '2025-01-01'::timestamptz + (i || ' minute')::interval FROM generate_series(0,100) i;
-- status should be COMPRESSED, UNORDERED, PARTIAL
SELECT _timescaledb_functions.chunk_status_text(chunk) FROM show_chunks('metrics_chunk') chunk;
-- delete should propagate to compressed chunk
SELECT count(*) FROM :CHUNK;
EXPLAIN (analyze,buffers off,costs off,summary off,timing off) DELETE FROM :CHUNK WHERE time > '2025-01-01'::timestamptz;
SELECT count(*) FROM :CHUNK;
ROLLBACK;

BEGIN;
SET timescaledb.enable_direct_compress_insert_client_sorted = true;
INSERT INTO :CHUNK SELECT '2025-01-01'::timestamptz + (i || ' minute')::interval FROM generate_series(0,100) i;
-- status should be COMPRESSED, PARTIAL
SELECT _timescaledb_functions.chunk_status_text(chunk) FROM show_chunks('metrics_chunk') chunk;
ROLLBACK;

-- simple test with compressed insert enabled and sorting limited to 500
-- batches should be limited to that amount so we have more compressed batches
-- dataset is tweaked to fall into single chunk with exactly 3 batches
BEGIN;
SET timescaledb.enable_direct_compress_insert = true;
SET timescaledb.direct_compress_insert_tuple_sort_limit = 1000;
INSERT INTO metrics SELECT '2025-01-02'::timestamptz + (i || ' minute')::interval, 'd1', i::float FROM generate_series(1,3000) i;
EXPLAIN (ANALYZE, BUFFERS OFF, COSTS OFF, SUMMARY OFF, TIMING OFF) SELECT * FROM metrics;
SELECT first(time,rn), last(time,rn) FROM (SELECT ROW_NUMBER() OVER () as rn, time FROM metrics) sub;
-- since the chunks are new status should be COMPRESSED, UNORDERED
SELECT DISTINCT _timescaledb_functions.chunk_status_text(chunk) FROM show_chunks('metrics') chunk;
ROLLBACK;

-- test sort limit with TEXT segmentby triggers multiple flush cycles
BEGIN;
ALTER TABLE metrics SET (tsdb.segmentby = 'device');
SET timescaledb.enable_direct_compress_insert = true;
SET timescaledb.enable_direct_compress_insert_sort_batches = true;
SET timescaledb.enable_direct_compress_insert_client_sorted = false;
SET timescaledb.direct_compress_insert_tuple_sort_limit = 50;
INSERT INTO metrics
SELECT t, 'location_' || (i % 3), random()
FROM generate_series('2025-01-02'::timestamptz,
                     '2025-01-02 00:30:00'::timestamptz,
                     '1 minute') t,
     generate_series(1, 3) i;
SELECT count(*) FROM metrics;
EXPLAIN (ANALYZE, BUFFERS OFF, COSTS OFF, SUMMARY OFF, TIMING OFF) SELECT * FROM metrics;
SELECT DISTINCT _timescaledb_functions.chunk_status_text(chunk) FROM show_chunks('metrics') chunk;
ROLLBACK;

-- test INSERT SELECT from compressed source into compressed target
-- this tests that reading from ColumnarScan correctly copies all rows
SET timescaledb.enable_direct_compress_insert = true;
SET timescaledb.enable_direct_compress_insert_client_sorted = true;

CREATE TABLE compress_src (
    time TIMESTAMPTZ NOT NULL,
    device_id TEXT NOT NULL,
    value FLOAT NOT NULL
) WITH (
    tsdb.hypertable,
    tsdb.partition_column = 'time',
    tsdb.chunk_interval = '4 hours',
    tsdb.orderby = 'time asc',
    tsdb.segment_by = 'device_id'
);

CREATE TABLE compress_tgt (
    time TIMESTAMPTZ NOT NULL,
    device_id TEXT NOT NULL,
    value FLOAT NOT NULL
) WITH (
    tsdb.hypertable,
    tsdb.partition_column = 'time',
    tsdb.chunk_interval = '4 hours',
    tsdb.orderby = 'time asc',
    tsdb.segment_by = 'device_id'
);

-- Insert with device_id, time ordering to create multi-row compressed blocks
INSERT INTO compress_src (time, device_id, value)
SELECT t.time, s.device_id, random()
FROM (SELECT generate_series('2025-01-01'::timestamptz, '2025-01-02'::timestamptz, INTERVAL '1 minute') AS time) t
CROSS JOIN (SELECT generate_series(1, 5)::TEXT AS device_id) s
ORDER BY s.device_id, t.time;

SELECT count(*) FROM compress_src;

INSERT INTO compress_tgt SELECT * FROM compress_src ORDER BY device_id, time;

SELECT count(*) FROM compress_tgt;

RESET timescaledb.enable_direct_compress_insert;
RESET timescaledb.enable_direct_compress_insert_client_sorted;

DROP TABLE compress_src;
DROP TABLE compress_tgt;

-- simple test to check default segmentby does not get set for direct compress
BEGIN;
RESET timescaledb.enable_direct_compress_insert;
CREATE TABLE test_segmentby_stats (
    time TIMESTAMPTZ NOT NULL,
    device_id INT NOT NULL,
    value FLOAT
) WITH (tsdb.hypertable, tsdb.compress);

INSERT INTO test_segmentby_stats
SELECT t, (i % 5), random()
FROM generate_series('2024-01-01'::timestamptz, '2024-01-02'::timestamptz, '1 second') t,
     generate_series(1, 5) i;

ANALYZE test_segmentby_stats;
SELECT compress_chunk(c) FROM show_chunks('test_segmentby_stats') c;

-- will have devide_id as default segmentby for compressed chunk;
SELECT * FROM _timescaledb_catalog.compression_settings ORDER BY relid::text COLLATE "C";

SET timescaledb.enable_direct_compress_insert = true;
INSERT INTO test_segmentby_stats
SELECT t, (i % 5), random()
FROM generate_series('2024-01-06'::timestamptz, '2024-01-07'::timestamptz, '1 second') t,
     generate_series(1, 5) i;
ANALYZE test_segmentby_stats;
-- will have default segmentby set for a newly created direct compressed chunk (should observe a skipped chunk id);
SELECT * FROM _timescaledb_catalog.compression_settings ORDER BY relid::text COLLATE "C";
ROLLBACK;

BEGIN;
RESET timescaledb.enable_direct_compress_insert;
CREATE TABLE test_segmentby_stats (
    time TIMESTAMPTZ NOT NULL,
    device_id INT NOT NULL,
    value FLOAT
) WITH (tsdb.hypertable, tsdb.compress, tsdb.segmentby='device_id');

SET timescaledb.enable_direct_compress_insert = true;
INSERT INTO test_segmentby_stats
SELECT t, (i % 5), random()
FROM generate_series('2024-01-06'::timestamptz, '2024-01-07'::timestamptz, '1 second') t,
     generate_series(1, 5) i;
ANALYZE test_segmentby_stats;
-- will have device_id by as configured segmentby for direct compressed chunk (should not observe skipped chunk id);
SELECT * FROM _timescaledb_catalog.compression_settings ORDER BY relid::text COLLATE "C";
ROLLBACK;

-- Direct compress on a hypertable with a dropped column before the time
-- dimension. Chunks created after the drop have a different TupleDesc than
-- the hypertable, so the slot must be converted to chunk format before being
-- passed to the compressor. Cover both INSERT and COPY entry points.
BEGIN;
RESET timescaledb.enable_direct_compress_insert;
CREATE TABLE test_dropcol(col1 int, time timestamptz NOT NULL, device text, value float);
SELECT create_hypertable('test_dropcol', 'time');
ALTER TABLE test_dropcol DROP COLUMN col1;
INSERT INTO test_dropcol(time, device, value)
SELECT '2025-01-01'::timestamptz + (i * interval '5 minutes'), 'd1', i::float
FROM generate_series(1, 50) i;
ALTER TABLE test_dropcol SET (timescaledb.compress, timescaledb.compress_segmentby = 'device');
SELECT count(compress_chunk(c)) FROM show_chunks('test_dropcol') c;
SET timescaledb.enable_direct_compress_insert = true;
INSERT INTO test_dropcol(time, device, value)
SELECT '2025-02-01'::timestamptz + (i * interval '1 second'), 'd2', i::float
FROM generate_series(1, 5000) i;
SELECT count(*), min(time), max(time), min(value), max(value) FROM test_dropcol WHERE device = 'd2';
RESET timescaledb.enable_direct_compress_insert;
SET timescaledb.enable_direct_compress_copy = true;
COPY test_dropcol(time, device, value) FROM STDIN WITH (FORMAT csv);
2025-03-01 00:00:01+00,d3,1.0
2025-03-01 00:00:02+00,d3,2.0
2025-03-01 00:00:03+00,d3,3.0
\.
SELECT count(*), min(time), max(time), min(value), max(value) FROM test_dropcol WHERE device = 'd3';
ROLLBACK;

-- Test orderby columns are not selected by DC segmentby default
BEGIN;
RESET timescaledb.enable_direct_compress_insert;
CREATE TABLE test_orderby_not_segmentby(time timestamptz NOT NULL, device_id int NOT NULL,
    value float) WITH (tsdb.hypertable);
CREATE INDEX ON test_orderby_not_segmentby(device_id);
INSERT INTO test_orderby_not_segmentby SELECT '2024-01-01'::timestamptz + (i||' min')::interval,
    (i%5)+1, random() FROM generate_series(1,500) i;
ANALYZE test_orderby_not_segmentby;
ALTER TABLE test_orderby_not_segmentby SET (timescaledb.compress);
SET timescaledb.enable_direct_compress_insert = on;
INSERT INTO test_orderby_not_segmentby SELECT '2024-06-01'::timestamptz + (i||' min')::interval,
    (i%5)+1, random() FROM generate_series(1,2000) i;
SELECT * FROM _timescaledb_catalog.compression_settings ORDER BY relid::text COLLATE "C";
ROLLBACK;

-- Direct compress must still enforce CHECK and NOT NULL constraints
CREATE TABLE metrics_check(time timestamptz NOT NULL, value int NOT NULL CHECK (value > 0))
    WITH (tsdb.hypertable, tsdb.partition_column='time');
SET timescaledb.enable_direct_compress_insert = true;
\set ON_ERROR_STOP 0
-- CHECK violation should be reported
INSERT INTO metrics_check SELECT '2025-01-01'::timestamptz + (i || ' minute')::interval, CASE WHEN i = 50 THEN -1 ELSE i END FROM generate_series(1,100) i;
-- NOT NULL violation should be reported
INSERT INTO metrics_check SELECT '2025-01-01'::timestamptz + (i || ' minute')::interval, CASE WHEN i = 50 THEN NULL ELSE i END FROM generate_series(1,100) i;
\set ON_ERROR_STOP 1
-- valid rows insert and compress without error
INSERT INTO metrics_check SELECT '2025-01-01'::timestamptz + (i || ' minute')::interval, i FROM generate_series(1,100) i;
SELECT count(*), min(value), max(value) FROM metrics_check;
DROP TABLE metrics_check;
RESET timescaledb.enable_direct_compress_insert;

-- Direct compress must still enforce a view's WITH CHECK OPTION
CREATE TABLE metrics_view(time timestamptz NOT NULL, value int)
    WITH (tsdb.hypertable, tsdb.partition_column='time');
CREATE VIEW metrics_view_pos AS SELECT * FROM metrics_view WHERE value > 0 WITH CHECK OPTION;
SET timescaledb.enable_direct_compress_insert = true;
\set ON_ERROR_STOP 0
-- row not satisfying the view qual should be reported
INSERT INTO metrics_view_pos SELECT '2025-01-01'::timestamptz + (i || ' minute')::interval, CASE WHEN i = 50 THEN -1 ELSE i END FROM generate_series(1,100) i;
\set ON_ERROR_STOP 1
-- valid rows insert without error
INSERT INTO metrics_view_pos SELECT '2025-01-01'::timestamptz + (i || ' minute')::interval, i FROM generate_series(1,100) i;
SELECT count(*), min(value), max(value) FROM metrics_view;
RESET timescaledb.enable_direct_compress_insert;
DROP VIEW metrics_view_pos;
DROP TABLE metrics_view;

-- Direct compress must be disabled when the table has an exclusion constraint
CREATE TABLE dc_excl(time timestamptz NOT NULL, device text NOT NULL, value float,
    EXCLUDE USING btree (time WITH =, device WITH =))
    WITH (tsdb.hypertable, tsdb.partition_column='time');
ALTER TABLE dc_excl SET (timescaledb.compress, timescaledb.compress_segmentby='device');
SET timescaledb.enable_direct_compress_insert = true;
-- insert warns, disables direct compress and falls back to a normal insert
INSERT INTO dc_excl SELECT '2025-01-01'::timestamptz + (i || ' minute')::interval, 'd1', i::float FROM generate_series(1,100) i;
-- chunk stays uncompressed since direct compress was disabled
SELECT DISTINCT _timescaledb_functions.chunk_status_text(chunk) FROM show_chunks('dc_excl') chunk;
-- exclusion constraint is still enforced
\set ON_ERROR_STOP 0
INSERT INTO dc_excl VALUES ('2025-01-01'::timestamptz + INTERVAL '1 minute', 'd1', 99);
\set ON_ERROR_STOP 1
SELECT count(*) FROM dc_excl;
RESET timescaledb.enable_direct_compress_insert;
DROP TABLE dc_excl;

-- Direct compress must return rows for a RETURNING clause
CREATE TABLE dc_ret(time timestamptz NOT NULL, device text NOT NULL, val int NOT NULL)
    WITH (tsdb.hypertable, tsdb.partition_column='time');
ALTER TABLE dc_ret SET (timescaledb.compress, timescaledb.compress_segmentby='device');
SET timescaledb.enable_direct_compress_insert = true;
-- confirm the direct compress path is used for this insert
EXPLAIN (COSTS OFF, SUMMARY OFF, TIMING OFF)
INSERT INTO dc_ret SELECT '2025-01-01'::timestamptz + (i * INTERVAL '1 minute'), 'd' || (i % 2)::text, i
    FROM generate_series(1, 20) i RETURNING val;
-- RETURNING should return one row per inserted tuple
WITH inserted AS (
    INSERT INTO dc_ret SELECT '2025-01-01'::timestamptz + (i * INTERVAL '1 minute'), 'd' || (i % 2)::text, i
        FROM generate_series(1, 20) i RETURNING val, device
)
SELECT count(*), min(val), max(val), sum(val), count(DISTINCT device) FROM inserted;
-- RETURNING with an expression referencing multiple columns
WITH inserted AS (
    INSERT INTO dc_ret SELECT '2025-01-01'::timestamptz + (i * INTERVAL '1 hour'), 'd' || (i % 2)::text, i
        FROM generate_series(1, 20) i RETURNING device || ':' || val AS label
)
SELECT count(*), min(label), max(label) FROM inserted;
-- RETURNING a system column falls back to the normal insert path, except
-- tableoid which the returning projection sets explicitly.
EXPLAIN (COSTS OFF, SUMMARY OFF, TIMING OFF)
INSERT INTO dc_ret SELECT '2025-01-01'::timestamptz + (i * INTERVAL '1 minute'), 'd' || (i % 2)::text, i
    FROM generate_series(1, 20) i RETURNING ctid;
EXPLAIN (COSTS OFF, SUMMARY OFF, TIMING OFF)
INSERT INTO dc_ret SELECT '2025-01-01'::timestamptz + (i * INTERVAL '1 minute'), 'd' || (i % 2)::text, i
    FROM generate_series(1, 20) i RETURNING xmin;
EXPLAIN (COSTS OFF, SUMMARY OFF, TIMING OFF)
INSERT INTO dc_ret SELECT '2025-01-01'::timestamptz + (i * INTERVAL '1 minute'), 'd' || (i % 2)::text, i
    FROM generate_series(1, 20) i RETURNING cmin;
EXPLAIN (COSTS OFF, SUMMARY OFF, TIMING OFF)
INSERT INTO dc_ret SELECT '2025-01-01'::timestamptz + (i * INTERVAL '1 minute'), 'd' || (i % 2)::text, i
    FROM generate_series(1, 20) i RETURNING xmax;
EXPLAIN (COSTS OFF, SUMMARY OFF, TIMING OFF)
INSERT INTO dc_ret SELECT '2025-01-01'::timestamptz + (i * INTERVAL '1 minute'), 'd' || (i % 2)::text, i
    FROM generate_series(1, 20) i RETURNING cmax;
-- tableoid stays on the direct compress path
EXPLAIN (COSTS OFF, SUMMARY OFF, TIMING OFF)
INSERT INTO dc_ret SELECT '2025-01-01'::timestamptz + (i * INTERVAL '1 minute'), 'd' || (i % 2)::text, i
    FROM generate_series(1, 20) i RETURNING tableoid;
-- ctid falls back and returns real tuple ids
WITH inserted AS (
    INSERT INTO dc_ret SELECT '2025-01-01'::timestamptz + (i * INTERVAL '1 day'), 'd' || (i % 2)::text, i
        FROM generate_series(1, 20) i RETURNING ctid AS c, val
)
SELECT count(*), bool_and(c IS NOT NULL), min(val), max(val) FROM inserted;
-- tableoid on the direct compress path holds the target chunk (all 20 rows
-- land in one chunk, so every RETURNING row reports the same chunk)
INSERT INTO dc_ret SELECT '2025-06-01'::timestamptz + (i * INTERVAL '1 minute'), 'd' || (i % 2)::text, i
    FROM generate_series(1, 20) i RETURNING tableoid::regclass AS chunk;
-- rows were actually inserted
SELECT count(*) FROM dc_ret;
RESET timescaledb.enable_direct_compress_insert;
DROP TABLE dc_ret;