text: -- 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.
  
  ---------------------------------------------------
  -- Test compression on a distributed hypertable
  ---------------------------------------------------
  \c :TEST_DBNAME :ROLE_CLUSTER_SUPERUSER
  
  \set DATA_NODE_1 :TEST_DBNAME _1
  \set DATA_NODE_2 :TEST_DBNAME _2
  \set DATA_NODE_3 :TEST_DBNAME _3
  
  \ir include/remote_exec.sql
  \ir include/compression_utils.sql
  
  SELECT (add_data_node (name, host => 'localhost', DATABASE => name)).*
  FROM (VALUES (:'DATA_NODE_1'), (:'DATA_NODE_2'), (:'DATA_NODE_3')) v (name);
  
  GRANT USAGE ON FOREIGN SERVER :DATA_NODE_1, :DATA_NODE_2, :DATA_NODE_3 TO :ROLE_1;
  SET ROLE :ROLE_1;
  
  CREATE TABLE compressed(time timestamptz, device int, temp float);
  -- Replicate twice to see that compress_chunk compresses all replica chunks
  SELECT create_distributed_hypertable('compressed', 'time', 'device', replication_factor => 2);
  INSERT INTO compressed SELECT t, (abs(timestamp_hash(t::timestamp)) % 10) + 1, random()*80
  FROM generate_series('2018-03-02 1:00'::TIMESTAMPTZ, '2018-03-04 1:00', '1 hour') t;
  ALTER TABLE compressed SET (timescaledb.compress, timescaledb.compress_segmentby='device', timescaledb.compress_orderby = 'time DESC');
  
  SELECT table_name, compression_state, compressed_hypertable_id
  FROM _timescaledb_catalog.hypertable
  ORDER BY 1;
  
  SELECT * FROM timescaledb_information.compression_settings order by attname;
  \x
  SELECT * FROM _timescaledb_catalog.hypertable
  WHERE table_name = 'compressed';
  \x
  
  SELECT test.remote_exec(NULL, $$
  SELECT table_name, compressed_hypertable_id
  FROM _timescaledb_catalog.hypertable
  WHERE table_name = 'compressed';
  $$);
  
  -- There should be no compressed chunks
  SELECT * from chunk_compression_stats( 'compressed')
  ORDER BY chunk_name, node_name;
  
  -- Test that compression is rolled back on aborted transaction
  BEGIN;
  SELECT compress_chunk(chunk)
  FROM show_chunks('compressed') AS chunk
  ORDER BY chunk
  LIMIT 1;
  
  -- Data nodes should now report compressed chunks
  SELECT * from chunk_compression_stats( 'compressed')
  ORDER BY chunk_name, node_name;
  -- Abort the transaction
  ROLLBACK;
  
  -- No compressed chunks since we rolled back
  SELECT * from chunk_compression_stats( 'compressed')
  ORDER BY chunk_name, node_name;
  
  -- Compress for real this time
  SELECT compress_chunk(chunk)
  FROM show_chunks('compressed') AS chunk
  ORDER BY chunk
  LIMIT 1;
  
  -- Check that one chunk, and its replica, is compressed
  SELECT * from chunk_compression_stats( 'compressed')
  ORDER BY chunk_name, node_name;
  select * from hypertable_compression_stats('compressed');
  
  --- Decompress the chunk and replica
  SELECT decompress_chunk(chunk)
  FROM show_chunks('compressed') AS chunk
  ORDER BY chunk
  LIMIT 1;
  
  -- Compress odd numbered chunks
  SELECT compress_chunk(chunk) FROM
  (
    SELECT *, row_number() OVER () AS rownum
    FROM show_chunks('compressed') AS chunk
    ORDER BY chunk
  ) AS t
  WHERE rownum % 2 = 1;
  SELECT * from chunk_compression_stats('compressed')
  ORDER BY chunk_name, node_name;
  
  -- Compress twice to notice idempotent operation
  SELECT compress_chunk(chunk, if_not_compressed => true)
  FROM show_chunks('compressed') AS chunk
  ORDER BY chunk;
  
  -- Compress again to verify errors are ignored
  SELECT compress_chunk(chunk, if_not_compressed => true)
  FROM show_chunks('compressed') AS chunk
  ORDER BY chunk;
  
  -- There should be no uncompressed chunks
  SELECT * from chunk_compression_stats('compressed')
  ORDER BY chunk_name, node_name;
  SELECT test.remote_exec(NULL, $$
  SELECT * FROM _timescaledb_catalog.chunk ORDER BY id;
  $$);
  
  -- Decompress the chunks and replicas
  SELECT decompress_chunk(chunk)
  FROM show_chunks('compressed') AS chunk
  ORDER BY chunk;
  
  -- Should now be decompressed
  SELECT * from chunk_compression_stats('compressed')
  ORDER BY chunk_name, node_name;
  
  -- Decompress twice to generate NOTICE that the chunk is already decompressed
  SELECT decompress_chunk(chunk, if_compressed => true)
  FROM show_chunks('compressed') AS chunk
  ORDER BY chunk
  LIMIT 1;
  
  \x
  SELECT * FROM timescaledb_information.hypertables
  WHERE hypertable_name = 'compressed';
  SELECT * from timescaledb_information.chunks
  ORDER BY hypertable_name, chunk_name;
  SELECT * from timescaledb_information.dimensions
  ORDER BY hypertable_name, dimension_number;
  \x
  
  SELECT * FROM chunks_detailed_size('compressed'::regclass)
  ORDER BY chunk_name, node_name;
  SELECT * FROM hypertable_detailed_size('compressed'::regclass) ORDER BY node_name;
  
  -- Disable compression on distributed table tests
  ALTER TABLE compressed SET (timescaledb.compress = false);
  
  SELECT table_name, compression_state, compressed_hypertable_id
  FROM _timescaledb_catalog.hypertable
  ORDER BY 1;
  
  SELECT * FROM timescaledb_information.compression_settings order by attname;
  
  --Now re-enable compression
  ALTER TABLE compressed SET (timescaledb.compress, timescaledb.compress_segmentby='device');
  SELECT table_name, compression_state, compressed_hypertable_id
  FROM _timescaledb_catalog.hypertable
  ORDER BY 1;
  SELECT * FROM timescaledb_information.compression_settings order by attname;
  SELECT compress_chunk(chunk, if_not_compressed => true)
  FROM show_chunks('compressed') AS chunk
  ORDER BY chunk
  LIMIT 1;
  
  SELECT chunk_name, node_name, compression_status
  FROM chunk_compression_stats('compressed')
  ORDER BY 1, 2;
  
  -- ALTER TABLE on distributed compressed hypertable
  ALTER TABLE compressed ADD COLUMN new_coli integer;
  ALTER TABLE compressed ADD COLUMN new_colv varchar(30);
  
  SELECT * FROM _timescaledb_catalog.hypertable_compression
  ORDER BY attname;
  
  SELECT count(*) from compressed where new_coli is not null;
  
  --insert data into new chunk
  INSERT INTO compressed
  SELECT '2019-08-01 00:00',  100, 100, 1, 'newcolv' ;
  
  SELECT COUNT(*) AS count_compressed
  FROM
  (
  SELECT compress_chunk(chunk.schema_name|| '.' || chunk.table_name, true)
  FROM _timescaledb_catalog.chunk chunk
  INNER JOIN _timescaledb_catalog.hypertable hypertable ON (chunk.hypertable_id = hypertable.id)
  WHERE hypertable.table_name like 'compressed' and chunk.compressed_chunk_id IS NULL ORDER BY chunk.id
  )
  AS sub;
  
  SELECT * from compressed where new_coli is not null;
  
  -- Test ALTER TABLE rename column on distributed hypertables
  ALTER TABLE compressed RENAME new_coli TO new_intcol  ;
  ALTER TABLE compressed RENAME device TO device_id  ;
  
  SELECT * FROM test.remote_exec( NULL,
      $$ SELECT * FROM _timescaledb_catalog.hypertable_compression
         WHERE attname = 'device_id' OR attname = 'new_intcol'  and
         hypertable_id = (SELECT id from _timescaledb_catalog.hypertable
                         WHERE table_name = 'compressed' ) ORDER BY attname; $$ );
  
  -- TEST insert data into compressed chunk
  INSERT INTO compressed
  SELECT '2019-08-01 01:00',  300, 300, 3, 'newcolv' ;
  SELECT * from compressed where new_intcol = 3;
  
  -- We're done with the table, so drop it.
  DROP TABLE IF EXISTS compressed CASCADE;
  
  ------------------------------------------------------
  -- Test compression policy on a distributed hypertable
  ------------------------------------------------------
  
  CREATE TABLE conditions (
        time        TIMESTAMPTZ       NOT NULL,
        location    TEXT              NOT NULL,
        location2    char(10)              NOT NULL,
        temperature DOUBLE PRECISION  NULL,
        humidity    DOUBLE PRECISION  NULL
      );
  SELECT create_distributed_hypertable('conditions', 'time', chunk_time_interval => '31days'::interval, replication_factor => 2);
  
  --TEST 1--
  --cannot set policy without enabling compression --
  \set ON_ERROR_STOP 0
  select add_compression_policy('conditions', '60d'::interval);
  \set ON_ERROR_STOP 1
  
  -- TEST2 --
  --add a policy to compress chunks --
  alter table conditions set (timescaledb.compress, timescaledb.compress_segmentby = 'location', timescaledb.compress_orderby = 'time');
  insert into conditions
  select generate_series('2018-12-01 00:00'::timestamp, '2018-12-31 00:00'::timestamp, '1 day'), 'POR', 'klick', 55, 75;
  
  select add_compression_policy('conditions', '60d'::interval) AS compressjob_id
  \gset
  
  select * from _timescaledb_config.bgw_job where id = :compressjob_id;
  select * from alter_job(:compressjob_id, schedule_interval=>'1s');
  select * from _timescaledb_config.bgw_job where id >= 1000 ORDER BY id;
  -- we want only 1 chunk to be compressed --
  SELECT alter_job(id,config:=jsonb_set(config,'{maxchunks_to_compress}', '1'))
  FROM _timescaledb_config.bgw_job WHERE id = :compressjob_id;
  
  insert into conditions
  select now()::timestamp, 'TOK', 'sony', 55, 75;
  
  -- TEST3 --
  --only the old chunks will get compressed when policy is executed--
  CALL run_job(:compressjob_id);
  select chunk_name, node_name, pg_size_pretty(before_compression_total_bytes) before_total,
  pg_size_pretty( after_compression_total_bytes)  after_total
  from chunk_compression_stats('conditions') where compression_status like 'Compressed' order by chunk_name;
  SELECT * FROM _timescaledb_catalog.chunk ORDER BY id;
  
  -- TEST 4 --
  --cannot set another policy
  \set ON_ERROR_STOP 0
  select add_compression_policy('conditions', '60d'::interval, if_not_exists=>true);
  select add_compression_policy('conditions', '60d'::interval);
  select add_compression_policy('conditions', '30d'::interval, if_not_exists=>true);
  \set ON_ERROR_STOP 1
  
  --TEST 5 --
  -- drop the policy --
  select remove_compression_policy('conditions');
  select count(*) from _timescaledb_config.bgw_job WHERE id>=1000;
  
  --TEST 6 --
  -- try to execute the policy after it has been dropped --
  \set ON_ERROR_STOP 0
  CALL run_job(:compressjob_id);
  \set ON_ERROR_STOP 1
  
  -- We're done with the table, so drop it.
  DROP TABLE IF EXISTS conditions CASCADE;
  
  --TEST 7
  --compression policy for smallint, integer or bigint based partition hypertable
  --smallint tests
  CREATE TABLE test_table_smallint(time smallint, val int);
  SELECT create_distributed_hypertable('test_table_smallint', 'time', chunk_time_interval => 1, replication_factor => 2);
  
  CREATE OR REPLACE FUNCTION dummy_now_smallint() RETURNS SMALLINT LANGUAGE SQL IMMUTABLE as  'SELECT 5::SMALLINT';
  CALL distributed_exec($$
  CREATE OR REPLACE FUNCTION dummy_now_smallint() RETURNS SMALLINT LANGUAGE SQL IMMUTABLE as  'SELECT 5::SMALLINT'
  $$);
  SELECT set_integer_now_func('test_table_smallint', 'dummy_now_smallint');
  INSERT INTO test_table_smallint SELECT generate_series(1,5), 10;
  ALTER TABLE test_table_smallint SET (timescaledb.compress);
  SELECT add_compression_policy('test_table_smallint', 2::int) AS compressjob_id \gset
  
  SELECT * FROM _timescaledb_config.bgw_job WHERE id = :compressjob_id;
  
  CALL run_job(:compressjob_id);
  CALL run_job(:compressjob_id);
  
  SELECT chunk_name, node_name, before_compression_total_bytes, after_compression_total_bytes
  FROM chunk_compression_stats('test_table_smallint')
  WHERE compression_status LIKE 'Compressed'
  ORDER BY chunk_name;
  
  --integer tests
  CREATE TABLE test_table_integer(time int, val int);
  SELECT create_distributed_hypertable('test_table_integer', 'time', chunk_time_interval => 1, replication_factor => 2);
  
  CREATE OR REPLACE FUNCTION dummy_now_integer() RETURNS INTEGER LANGUAGE SQL IMMUTABLE as  'SELECT 5::INTEGER';
  CALL distributed_exec($$
  CREATE OR REPLACE FUNCTION dummy_now_integer() RETURNS INTEGER LANGUAGE SQL IMMUTABLE as  'SELECT 5::INTEGER'
  $$);
  SELECT set_integer_now_func('test_table_integer', 'dummy_now_integer');
  INSERT INTO test_table_integer SELECT generate_series(1,5), 10;
  ALTER TABLE test_table_integer SET (timescaledb.compress);
  SELECT add_compression_policy('test_table_integer', 2::int) AS compressjob_id \gset
  
  SELECT * FROM _timescaledb_config.bgw_job WHERE id = :compressjob_id;
  
  CALL run_job(:compressjob_id);
  CALL run_job(:compressjob_id);
  
  SELECT chunk_name, node_name, before_compression_total_bytes, after_compression_total_bytes
  FROM chunk_compression_stats('test_table_integer')
  WHERE compression_status LIKE 'Compressed'
  ORDER BY chunk_name;
  
  --bigint tests
  CREATE TABLE test_table_bigint(time bigint, val int);
  SELECT create_distributed_hypertable('test_table_bigint', 'time', chunk_time_interval => 1, replication_factor => 2);
  
  CREATE OR REPLACE FUNCTION dummy_now_bigint() RETURNS BIGINT LANGUAGE SQL IMMUTABLE as  'SELECT 5::BIGINT';
  CALL distributed_exec($$
  CREATE OR REPLACE FUNCTION dummy_now_bigint() RETURNS BIGINT LANGUAGE SQL IMMUTABLE as  'SELECT 5::BIGINT'
  $$);
  SELECT set_integer_now_func('test_table_bigint', 'dummy_now_bigint');
  INSERT INTO test_table_bigint SELECT generate_series(1,5), 10;
  ALTER TABLE test_table_bigint SET (timescaledb.compress);
  SELECT add_compression_policy('test_table_bigint', 2::int) AS compressjob_id \gset
  
  SELECT * FROM _timescaledb_config.bgw_job WHERE id = :compressjob_id;
  
  CALL run_job(:compressjob_id);
  CALL run_job(:compressjob_id);
  
  SELECT chunk_name, node_name, before_compression_total_bytes, after_compression_total_bytes
  FROM chunk_compression_stats('test_table_bigint')
  WHERE compression_status LIKE 'Compressed'
  ORDER BY chunk_name;
  
  --TEST8 insert into compressed chunks on dist. hypertable
  CREATE TABLE test_recomp_int(time bigint, val int);
  SELECT create_distributed_hypertable('test_recomp_int', 'time', chunk_time_interval => 20);
  
  CREATE OR REPLACE FUNCTION dummy_now() RETURNS BIGINT LANGUAGE SQL IMMUTABLE as  'SELECT 100::BIGINT';
  CALL distributed_exec($$
  CREATE OR REPLACE FUNCTION dummy_now() RETURNS BIGINT LANGUAGE SQL IMMUTABLE as  'SELECT 100::BIGINT'
  $$);
  select set_integer_now_func('test_recomp_int', 'dummy_now');
  insert into test_recomp_int select generate_series(1,5), 10;
  alter table test_recomp_int set (timescaledb.compress);
  
  CREATE VIEW test_recomp_int_chunk_status as
  SELECT
     c.table_name as chunk_name,
     c.status as chunk_status
  FROM _timescaledb_catalog.hypertable h, _timescaledb_catalog.chunk c
  WHERE h.id = c.hypertable_id and h.table_name = 'test_recomp_int';
  
  --compress chunks
  SELECT compress_chunk(chunk)
  FROM show_chunks('test_recomp_int') AS chunk
  ORDER BY chunk;
  
  --check the status
  SELECT * from test_recomp_int_chunk_status ORDER BY 1;
  
  -- insert into compressed chunks of test_recomp_int (using same value for val)--
  insert into test_recomp_int select 10, 10;
  SELECT count(*) from test_recomp_int where val = 10;
  
  --chunk status should change ---
  SELECT * from test_recomp_int_chunk_status ORDER BY 1;
  
  SELECT
  c.schema_name || '.' || c.table_name as "CHUNK_NAME"
  FROM _timescaledb_catalog.hypertable h, _timescaledb_catalog.chunk c
  WHERE h.id = c.hypertable_id and h.table_name = 'test_recomp_int' \gset
  
  --call recompress_chunk directly on distributed chunk
  CALL recompress_chunk(:'CHUNK_NAME'::regclass);
  
  --check chunk status now, should be compressed
  SELECT * from test_recomp_int_chunk_status ORDER BY 1;
  SELECT count(*) from test_recomp_int;
  
  --add a policy--
  select add_compression_policy('test_recomp_int', 1::int) AS compressjob_id
  \gset
  --once again add data to the compressed chunk
  insert into test_recomp_int select generate_series(5,7), 10;
  SELECT * from test_recomp_int_chunk_status ORDER BY 1;
  --run the compression policy job, it will recompress chunks that are unordered
  CALL run_job(:compressjob_id);
  SELECT count(*) from test_recomp_int;
  
  -- check with per datanode queries disabled
  SET timescaledb.enable_per_data_node_queries TO false;
  SELECT count(*) from test_recomp_int;
  RESET timescaledb.enable_per_data_node_queries;
  
  SELECT * from test_recomp_int_chunk_status ORDER BY 1;
  
  ---run copy tests
  --copy data into existing chunk + for a new chunk
  COPY test_recomp_int  FROM STDIN WITH DELIMITER ',';
  11, 11
  12, 12
  13, 13
  100, 100
  101, 100
  102, 100
  \.
  
  SELECT time_bucket(20, time ), count(*)
  FROM test_recomp_int
  GROUP BY time_bucket( 20, time) ORDER BY 1;
  
  --another new chunk
  INSERT INTO test_recomp_int VALUES( 65, 10);
  SELECT * from test_recomp_int_chunk_status ORDER BY 1;
  
  --compress all 3 chunks ---
  --check status, unordered chunk status will not change
  SELECT compress_chunk(chunk, true)
  FROM show_chunks('test_recomp_int') AS chunk
  ORDER BY chunk;
  SELECT * from test_recomp_int_chunk_status ORDER BY 1;
  
  --interleave copy into 3 different chunks and check status--
  COPY test_recomp_int  FROM STDIN WITH DELIMITER ',';
  14, 14
  103, 100
  66, 66
  15, 15
  104, 100
  70, 70
  \.
  
  SELECT * from test_recomp_int_chunk_status ORDER BY 1;
  SELECT time_bucket(20, time), count(*)
  FROM test_recomp_int
  GROUP BY time_bucket(20, time) ORDER BY 1;
  
  -- check with per datanode queries disabled
  SET timescaledb.enable_per_data_node_queries TO false;
  SELECT time_bucket(20, time), count(*)
  FROM test_recomp_int
  GROUP BY time_bucket(20, time) ORDER BY 1;
  RESET timescaledb.enable_per_data_node_queries;
  
  --check compression_status afterwards--
  CALL recompress_all_chunks('test_recomp_int', 2, true);
  SELECT * from test_recomp_int_chunk_status ORDER BY 1;
  
  CALL run_job(:compressjob_id);
  SELECT * from test_recomp_int_chunk_status ORDER BY 1;
  
  --verify that there are no errors if the policy/recompress_chunk is executed again
  --on previously compressed chunks
  CALL run_job(:compressjob_id);
  CALL recompress_all_chunks('test_recomp_int', true);
  
  --decompress and recompress chunk
  \set ON_ERROR_STOP 0
  SELECT decompress_chunk(chunk, true) FROM
  ( SELECT chunk FROM show_chunks('test_recomp_int') AS chunk ORDER BY chunk LIMIT 1 )q;
  CALL recompress_all_chunks('test_recomp_int', 1, false);
  \set ON_ERROR_STOP 1
  
  -- test alter column type with distributed hypertable
  \set ON_ERROR_STOP 0
  ALTER TABLE test_table_smallint ALTER COLUMN val TYPE float;
  ALTER TABLE test_table_integer ALTER COLUMN val TYPE float;
  ALTER TABLE test_table_bigint ALTER COLUMN val TYPE float;
  \set ON_ERROR_STOP 1
  
  --create a cont agg view on the ht with compressed chunks as well 
  SELECT compress_chunk(chunk, true) FROM
  ( SELECT chunk FROM show_chunks('test_recomp_int') AS chunk ORDER BY chunk LIMIT 1 )q;
  
  CREATE MATERIALIZED VIEW test_recomp_int_cont_view
  WITH (timescaledb.continuous,
        timescaledb.materialized_only=true)
  AS SELECT time_bucket(BIGINT '5', "time"), SUM(val)
     FROM test_recomp_int
     GROUP BY 1 WITH NO DATA;
  SELECT add_continuous_aggregate_policy('test_recomp_int_cont_view', NULL, BIGINT '5', '1 day'::interval);
  CALL refresh_continuous_aggregate('test_recomp_int_cont_view', NULL, NULL);
  SELECT * FROM test_recomp_int ORDER BY 1;
  SELECT * FROM test_recomp_int_cont_view ORDER BY 1;
  
  --TEST cagg triggers work on distributed hypertables when we insert into 
  -- compressed chunks.
  SELECT 
  CASE WHEN compress_chunk(chunk, true) IS NOT NULL THEN 'compress' END AS ch
   FROM
  ( SELECT chunk FROM show_chunks('test_recomp_int') AS chunk ORDER BY chunk )q;
  
  SELECT * FROM test_recomp_int_cont_view 
  WHERE time_bucket = 0 or time_bucket = 50 ORDER BY 1;
  
  --insert into an existing compressed chunk and a new chunk
  SET timescaledb.enable_distributed_insert_with_copy=false;
  
  INSERT INTO test_recomp_int VALUES (1, 10), (2,10), (3, 10);
  INSERT INTO test_recomp_int VALUES(51, 10);
   
  --refresh the cagg and verify the new results
  CALL refresh_continuous_aggregate('test_recomp_int_cont_view', NULL, 100);
  SELECT * FROM test_recomp_int_cont_view 
  WHERE time_bucket = 0 or time_bucket = 50 ORDER BY 1;
  
  --repeat test with copy setting turned to true
  SET timescaledb.enable_distributed_insert_with_copy=true;
  INSERT INTO test_recomp_int VALUES (4, 10);
  
  CALL refresh_continuous_aggregate('test_recomp_int_cont_view', NULL, 100);
  SELECT * FROM test_recomp_int_cont_view 
  WHERE time_bucket = 0 or time_bucket = 50 ORDER BY 1;
  
  --TEST drop one of the compressed chunks in test_recomp_int. The catalog
  --tuple for the chunk will be preserved since we have a cagg.
  -- Verify that status is accurate.
  SELECT
     c.table_name as chunk_name,
     c.status as chunk_status, c.dropped, c.compressed_chunk_id as comp_id
  FROM _timescaledb_catalog.hypertable h, _timescaledb_catalog.chunk c
  WHERE h.id = c.hypertable_id and h.table_name = 'test_recomp_int' ORDER BY 1;
  SELECT drop_chunks('test_recomp_int', older_than=> 20::bigint );
  SELECT
     c.table_name as chunk_name,
     c.status as chunk_status, c.dropped, c.compressed_chunk_id as comp_id
  FROM _timescaledb_catalog.hypertable h, _timescaledb_catalog.chunk c
  WHERE h.id = c.hypertable_id and h.table_name = 'test_recomp_int' ORDER BY 1;
  
  -- TEST drop should nuke everything
  DROP TABLE test_recomp_int CASCADE;
  
  -- test compression default handling
  CREATE TABLE test_defaults(time timestamptz NOT NULL, device_id int);
  SELECT create_distributed_hypertable('test_defaults','time');
  ALTER TABLE test_defaults SET (timescaledb.compress,timescaledb.compress_segmentby='device_id');
  
  -- create 2 chunks
  INSERT INTO test_defaults SELECT '2000-01-01', 1;
  INSERT INTO test_defaults SELECT '2001-01-01', 1;
  
  -- compress first chunk
  SELECT compress_chunk(show_chunks) AS compressed_chunk FROM show_chunks('test_defaults') ORDER BY show_chunks::text LIMIT 1;
  
  SELECT * FROM test_defaults ORDER BY 1;
  ALTER TABLE test_defaults ADD COLUMN c1 int;
  ALTER TABLE test_defaults ADD COLUMN c2 int NOT NULL DEFAULT 42;
  SELECT * FROM test_defaults ORDER BY 1,2;
  
  -- try insert into compressed and recompress
  INSERT INTO test_defaults SELECT '2000-01-01', 2;
  SELECT * FROM test_defaults ORDER BY 1,2;
  CALL recompress_all_chunks('test_defaults', 1, false);
  SELECT * FROM test_defaults ORDER BY 1,2;
  
  -- test dropping columns from compressed
  CREATE TABLE test_drop(f1 text, f2 text, f3 text, time timestamptz, device int, o1 text, o2 text);
  SELECT create_distributed_hypertable('test_drop','time');
  ALTER TABLE test_drop SET (timescaledb.compress,timescaledb.compress_segmentby='device',timescaledb.compress_orderby='o1,o2');
  
  -- switch to WARNING only to suppress compress_chunk NOTICEs
  SET client_min_messages TO WARNING;
  
  -- create some chunks each with different physical layout
  ALTER TABLE test_drop DROP COLUMN f1;
  INSERT INTO test_drop SELECT NULL,NULL,'2000-01-01',1,'o1','o2';
  SELECT count(compress_chunk(chunk,true)) FROM show_chunks('test_drop') chunk;
  
  ALTER TABLE test_drop DROP COLUMN f2;
  -- test non-existant column
  \set ON_ERROR_STOP 0
  ALTER TABLE test_drop DROP COLUMN f10;
  \set ON_ERROR_STOP 1
  ALTER TABLE test_drop DROP COLUMN IF EXISTS f10;
  
  INSERT INTO test_drop SELECT NULL,'2001-01-01',2,'o1','o2';
  SELECT count(compress_chunk(chunk,true)) FROM show_chunks('test_drop') chunk;
  ALTER TABLE test_drop DROP COLUMN f3;
  INSERT INTO test_drop SELECT '2003-01-01',3,'o1','o2';
  SELECT count(compress_chunk(chunk,true)) FROM show_chunks('test_drop') chunk;
  ALTER TABLE test_drop ADD COLUMN c1 TEXT;
  ALTER TABLE test_drop ADD COLUMN c2 TEXT;
  INSERT INTO test_drop SELECT '2004-01-01',4,'o1','o2','c1','c2-4';
  SELECT count(compress_chunk(chunk,true)) FROM show_chunks('test_drop') chunk;
  ALTER TABLE test_drop DROP COLUMN c1;
  INSERT INTO test_drop SELECT '2005-01-01',5,'o1','o2','c2-5';
  SELECT count(compress_chunk(chunk,true)) FROM show_chunks('test_drop') chunk;
  
  RESET client_min_messages;
  SELECT * FROM test_drop ORDER BY 1;
  
  -- check dropped columns got removed from catalog
  -- only c2 should be left in metadata
  SELECT attname
  FROM _timescaledb_catalog.hypertable_compression htc
  INNER JOIN _timescaledb_catalog.hypertable ht
    ON ht.id=htc.hypertable_id AND ht.table_name='test_drop'
  WHERE attname NOT IN ('time','device','o1','o2')
  ORDER BY 1;
  
  