1
0
mirror of https://github.com/timescale/timescaledb.git synced 2025-05-17 19:13:16 +08:00

Fix FK constraints for compressed chunks

When foreign key support for compressed chunks was added we moved
the FK constraint from the uncompressed chunk to the compressed chunk as
part of compress_chunk and moved it back as part of decompress_chunk.
With the addition of partially compressed chunks in 2.10.x this approach
was no longer sufficient and the FK constraint needs to be present on
both the uncompressed and the compressed chunk.

While this patch will fix future compressed chunks a migration has to be
run after upgrading timescaledb to migrate existing chunks affected by
this.

The following code will fix any affected hypertables:
```
CREATE OR REPLACE FUNCTION pg_temp.constraint_columns(regclass, int2[]) RETURNS text[] AS
$$
  SELECT array_agg(attname) FROM unnest($2) un(attnum) LEFT JOIN pg_attribute att ON att.attrelid=$1 AND att.attnum = un.attnum;
$$ LANGUAGE SQL SET search_path TO pg_catalog, pg_temp;

DO $$
DECLARE
  ht_id int;
  ht regclass;
  chunk regclass;
  con_oid oid;
  con_frelid regclass;
  con_name text;
  con_columns text[];
  chunk_id int;

BEGIN

  -- iterate over all hypertables that have foreign key constraints
  FOR ht_id, ht in
    SELECT
      ht.id,
      format('%I.%I',ht.schema_name,ht.table_name)::regclass
    FROM _timescaledb_catalog.hypertable ht
    WHERE
      EXISTS (
        SELECT FROM pg_constraint con
        WHERE
          con.contype='f' AND
          con.conrelid=format('%I.%I',ht.schema_name,ht.table_name)::regclass
      )
  LOOP
    RAISE NOTICE 'Hypertable % has foreign key constraint', ht;

    -- iterate over all foreign key constraints on the hypertable
    -- and check that they are present on every chunk
    FOR con_oid, con_frelid, con_name, con_columns IN
      SELECT con.oid, con.confrelid, con.conname, pg_temp.constraint_columns(con.conrelid,con.conkey)
      FROM pg_constraint con
      WHERE
        con.contype='f' AND
        con.conrelid=ht
    LOOP
      RAISE NOTICE 'Checking constraint % %', con_name, con_columns;
      -- check that the foreign key constraint is present on the chunk

      FOR chunk_id, chunk IN
        SELECT
          ch.id,
          format('%I.%I',ch.schema_name,ch.table_name)::regclass
        FROM _timescaledb_catalog.chunk ch
        WHERE
          ch.hypertable_id=ht_id
      LOOP
        RAISE NOTICE 'Checking chunk %', chunk;
        IF NOT EXISTS (
          SELECT FROM pg_constraint con
          WHERE
            con.contype='f' AND
            con.conrelid=chunk AND
            con.confrelid=con_frelid  AND
            pg_temp.constraint_columns(con.conrelid,con.conkey) = con_columns
        ) THEN
          RAISE WARNING 'Restoring constraint % on chunk %', con_name, chunk;
          PERFORM _timescaledb_functions.constraint_clone(con_oid, chunk);
          INSERT INTO _timescaledb_catalog.chunk_constraint(chunk_id, dimension_slice_id, constraint_name, hypertable_constraint_name) VALUES (chunk_id, NULL, con_name, con_name);
        END IF;

      END LOOP;
    END LOOP;

  END LOOP;

END
$$;

DROP FUNCTION pg_temp.constraint_columns(regclass, int2[]);
```
This commit is contained in:
Sven Klemm 2024-03-31 09:40:52 +02:00 committed by Sven Klemm
parent f6b2215560
commit ecf6beae5d
8 changed files with 88 additions and 72 deletions

1
.unreleased/pr_6797 Normal file

@ -0,0 +1 @@
Fixes: #6797 Fix foreign key constraint handling on compressed hypertables

@ -3222,58 +3222,6 @@ ts_chunk_recreate_all_constraints_for_dimension(Hypertable *ht, int32 dimension_
chunk_scan_ctx_destroy(&chunkctx);
}
/*
* Drops all FK constraints on a given chunk.
* Currently it is used only for chunks, which have been compressed and
* contain no data.
*/
void
ts_chunk_drop_fks(const Chunk *const chunk)
{
Relation rel;
List *fks;
ListCell *lc;
ASSERT_IS_VALID_CHUNK(chunk);
rel = table_open(chunk->table_id, AccessShareLock);
fks = copyObject(RelationGetFKeyList(rel));
table_close(rel, AccessShareLock);
foreach (lc, fks)
{
const ForeignKeyCacheInfo *const fk = lfirst_node(ForeignKeyCacheInfo, lc);
ts_chunk_constraint_delete_by_constraint_name(chunk->fd.id,
get_constraint_name(fk->conoid),
true,
true);
}
}
/*
* Recreates all FK constraints on a chunk by using the constraints on the parent hypertable
* as a template. Currently it is used only during chunk decompression, since FK constraints
* are dropped during compression.
*/
void
ts_chunk_create_fks(const Hypertable *ht, const Chunk *const chunk)
{
Relation rel;
List *fks;
ListCell *lc;
ASSERT_IS_VALID_CHUNK(chunk);
rel = table_open(chunk->hypertable_relid, AccessShareLock);
fks = copyObject(RelationGetFKeyList(rel));
table_close(rel, AccessShareLock);
foreach (lc, fks)
{
ForeignKeyCacheInfo *fk = lfirst_node(ForeignKeyCacheInfo, lc);
ts_chunk_constraint_create_on_chunk(ht, chunk, fk->conoid);
}
}
/*
* Chunk catalog updates are done in three steps.
* This is achieved by following this sequence:

@ -187,8 +187,6 @@ extern bool ts_chunk_exists_relid(Oid relid);
extern TSDLLEXPORT int ts_chunk_num_of_chunks_created_after(const Chunk *chunk);
extern TSDLLEXPORT bool ts_chunk_exists_with_compression(int32 hypertable_id);
extern void ts_chunk_recreate_all_constraints_for_dimension(Hypertable *ht, int32 dimension_id);
extern TSDLLEXPORT void ts_chunk_drop_fks(const Chunk *const chunk);
extern TSDLLEXPORT void ts_chunk_create_fks(const Hypertable *ht, const Chunk *const chunk);
extern int ts_chunk_delete_by_hypertable_id(int32 hypertable_id);
extern int ts_chunk_delete_by_name(const char *schema, const char *table, DropBehavior behavior);
extern bool ts_chunk_set_name(Chunk *chunk, const char *newname);

@ -457,11 +457,6 @@ compress_chunk_impl(Oid hypertable_relid, Oid chunk_relid)
before_size = ts_relation_size_impl(cxt.srcht_chunk->table_id);
cstat = compress_chunk(cxt.srcht_chunk->table_id, compress_ht_chunk->table_id, insert_options);
/* Drop all FK constraints on the uncompressed chunk. This is needed to allow
* cascading deleted data in FK-referenced tables, while blocking deleting data
* directly on the hypertable or chunks.
*/
ts_chunk_drop_fks(cxt.srcht_chunk);
after_size = ts_relation_size_impl(compress_ht_chunk->table_id);
if (new_compressed_chunk)
@ -596,9 +591,6 @@ decompress_chunk_impl(Chunk *uncompressed_chunk, bool if_compressed)
decompress_chunk(compressed_chunk->table_id, uncompressed_chunk->table_id);
/* Recreate FK constraints, since they were dropped during compression. */
ts_chunk_create_fks(uncompressed_hypertable, uncompressed_chunk);
/* Delete the compressed chunk */
ts_compression_chunk_size_delete(uncompressed_chunk->fd.id);
ts_chunk_clear_compressed_chunk(uncompressed_chunk);
@ -670,12 +662,6 @@ tsl_create_compressed_chunk(PG_FUNCTION_ARGS)
ts_chunk_constraints_create(cxt.compress_ht, compress_ht_chunk);
ts_trigger_create_all_on_chunk(compress_ht_chunk);
/* Drop all FK constraints on the uncompressed chunk. This is needed to allow
* cascading deleted data in FK-referenced tables, while blocking deleting data
* directly on the hypertable or chunks.
*/
ts_chunk_drop_fks(cxt.srcht_chunk);
/* Insert empty stats to compression_chunk_size */
compression_chunk_size_catalog_insert(cxt.srcht_chunk->fd.id,
&uncompressed_size,

@ -1054,9 +1054,10 @@ SELECT constraint_schema, constraint_name, table_schema, table_name, constraint_
FROM information_schema.table_constraints
WHERE table_name = :'CHUNK_NAME' AND constraint_type = 'FOREIGN KEY'
ORDER BY constraint_name;
constraint_schema | constraint_name | table_schema | table_name | constraint_type
-------------------+-----------------+--------------+------------+-----------------
(0 rows)
constraint_schema | constraint_name | table_schema | table_name | constraint_type
-----------------------+---------------------------+-----------------------+--------------------+-----------------
_timescaledb_internal | 42_6_hyper_device_id_fkey | _timescaledb_internal | _hyper_18_42_chunk | FOREIGN KEY
(1 row)
-- Delete data from compressed chunk directly fails
\set ON_ERROR_STOP 0
@ -1094,7 +1095,7 @@ WHERE table_name = :'CHUNK_NAME' AND constraint_type = 'FOREIGN KEY'
ORDER BY constraint_name;
constraint_schema | constraint_name | table_schema | table_name | constraint_type
-----------------------+---------------------------+-----------------------+--------------------+-----------------
_timescaledb_internal | 42_8_hyper_device_id_fkey | _timescaledb_internal | _hyper_18_42_chunk | FOREIGN KEY
_timescaledb_internal | 42_6_hyper_device_id_fkey | _timescaledb_internal | _hyper_18_42_chunk | FOREIGN KEY
(1 row)
-- create hypertable with 2 chunks

@ -0,0 +1,46 @@
-- 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 foreign key constraints with compression
CREATE TABLE keys(time timestamptz unique);
CREATE TABLE ht_with_fk(time timestamptz);
SELECT create_hypertable('ht_with_fk','time');
NOTICE: adding not-null constraint to column "time"
create_hypertable
-------------------------
(1,public,ht_with_fk,t)
(1 row)
ALTER TABLE ht_with_fk ADD CONSTRAINT keys FOREIGN KEY (time) REFERENCES keys(time) ON DELETE CASCADE;
ALTER TABLE ht_with_fk SET (timescaledb.compress,timescaledb.compress_segmentby='time');
NOTICE: default order by for hypertable "ht_with_fk" is set to ""
-- no keys added yet so any insert into ht_with_fk should fail
\set ON_ERROR_STOP 0
INSERT INTO ht_with_fk SELECT '2000-01-01';
ERROR: insert or update on table "_hyper_1_1_chunk" violates foreign key constraint "1_1_keys"
\set ON_ERROR_STOP 1
-- create a key in the referenced table
INSERT INTO keys SELECT '2000-01-01';
-- now the insert should succeed
INSERT INTO ht_with_fk SELECT '2000-01-01';
SELECT compress_chunk(ch) FROM show_chunks('ht_with_fk') ch;
compress_chunk
----------------------------------------
_timescaledb_internal._hyper_1_2_chunk
(1 row)
-- insert should still succeed after compression
INSERT INTO ht_with_fk SELECT '2000-01-01';
-- inserting key not present in keys should fail
\set ON_ERROR_STOP 0
INSERT INTO ht_with_fk SELECT '2000-01-01 0:00:01';
ERROR: insert or update on table "_hyper_1_2_chunk" violates foreign key constraint "2_2_keys"
\set ON_ERROR_STOP 1
SELECT conrelid::regclass,conname,confrelid::regclass FROM pg_constraint WHERE contype = 'f' AND confrelid = 'keys'::regclass ORDER BY conrelid::regclass::text COLLATE "C",conname;
conrelid | conname | confrelid
------------------------------------------------+----------+-----------
_timescaledb_internal._hyper_1_2_chunk | 2_2_keys | keys
_timescaledb_internal.compress_hyper_2_3_chunk | keys | keys
ht_with_fk | keys | keys
(3 rows)

@ -22,6 +22,7 @@ set(TEST_FILES
compression_create_compressed_table.sql
compression_conflicts.sql
compression_defaults.sql
compression_fks.sql
compression_insert.sql
compression_policy.sql
compression_qualpushdown.sql

@ -0,0 +1,35 @@
-- 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 foreign key constraints with compression
CREATE TABLE keys(time timestamptz unique);
CREATE TABLE ht_with_fk(time timestamptz);
SELECT create_hypertable('ht_with_fk','time');
ALTER TABLE ht_with_fk ADD CONSTRAINT keys FOREIGN KEY (time) REFERENCES keys(time) ON DELETE CASCADE;
ALTER TABLE ht_with_fk SET (timescaledb.compress,timescaledb.compress_segmentby='time');
-- no keys added yet so any insert into ht_with_fk should fail
\set ON_ERROR_STOP 0
INSERT INTO ht_with_fk SELECT '2000-01-01';
\set ON_ERROR_STOP 1
-- create a key in the referenced table
INSERT INTO keys SELECT '2000-01-01';
-- now the insert should succeed
INSERT INTO ht_with_fk SELECT '2000-01-01';
SELECT compress_chunk(ch) FROM show_chunks('ht_with_fk') ch;
-- insert should still succeed after compression
INSERT INTO ht_with_fk SELECT '2000-01-01';
-- inserting key not present in keys should fail
\set ON_ERROR_STOP 0
INSERT INTO ht_with_fk SELECT '2000-01-01 0:00:01';
\set ON_ERROR_STOP 1
SELECT conrelid::regclass,conname,confrelid::regclass FROM pg_constraint WHERE contype = 'f' AND confrelid = 'keys'::regclass ORDER BY conrelid::regclass::text COLLATE "C",conname;