<commit_before>

CREATE OR REPLACE FUNCTION "system"."version"()
    RETURNS system.version_tuple
AS $$
SELECT (5,1,6)::system.version_tuple;
$$ LANGUAGE sql IMMUTABLE;


DROP FUNCTION "relation_directory"."remove"(name);

CREATE FUNCTION "relation_directory"."remove"(name)
    RETURNS text
AS $$
DECLARE
  result text;
BEGIN
  SELECT name FROM relation_directory.type WHERE name = $1 INTO result;
  PERFORM public.action(format('DROP MATERIALIZED VIEW IF EXISTS relation.%I', $1));
  DELETE FROM relation_directory.type WHERE name = $1;
  RETURN result;
END;
$$ LANGUAGE plpgsql VOLATILE;


DROP FUNCTION "trend_directory"."assure_table_trends_exist"(trend_directory.trend_store_part, trend_directory.trend_descr[], trend_directory.generated_trend_descr[]);

DROP FUNCTION "trend_directory"."add_trends"(trend_directory.trend_store, trend_directory.trend_store_part_descr[]);

DROP FUNCTION "trend_directory"."get_trend_if_defined"(trend_directory.table_trend, trend_directory.trend_descr[], text);

DROP FUNCTION "trend_directory"."remove_trend_if_extraneous"(trend_directory.table_trend, trend_directory.trend_store_part_descr[]);

DROP FUNCTION "trend_directory"."remove_extra_trends"(trend_directory.trend_store, trend_directory.trend_store_part_descr[]);

DROP FUNCTION "trend_directory"."change_table_trend_data_unsafe"(trend_directory.table_trend, text, text, text);

DROP FUNCTION "trend_directory"."change_table_trend_data_safe"(trend_directory.table_trend, text, text, text);

DROP FUNCTION "trend_directory"."change_trend_data_unsafe"(trend_directory.table_trend, trend_directory.trend_descr[], text);

DROP FUNCTION "trend_directory"."change_trend_data_safe"(trend_directory.table_trend, trend_directory.trend_descr[], text);

DROP FUNCTION "trend_directory"."change_all_trend_data"(trend_directory.trend_store, trend_directory.trend_store_part_descr[]);

DROP FUNCTION "trend_directory"."change_trend_data_upward"(trend_directory.trend_store, trend_directory.trend_store_part_descr[]);

DROP FUNCTION "trend_directory"."change_trendstore_strong"(trend_directory.trend_store, trend_directory.trend_store_part_descr[]);

DROP FUNCTION "trend_directory"."change_trendstore_weak"(trend_directory.trend_store, trend_directory.trend_store_part_descr[]);

ALTER TABLE "trend_directory"."materialization_state" ADD COLUMN "max_modified" timestamp with time zone;


CREATE FUNCTION "trend_directory"."cleanup_for_materialization"()
    RETURNS trigger
AS $$
BEGIN
  EXECUTE format(
    'DROP FUNCTION trend.%I(timestamp with time zone)',
    trend_directory.fingerprint_function_name(OLD)
  );

  RETURN OLD;
END;
$$ LANGUAGE plpgsql VOLATILE;


CREATE FUNCTION "trend_directory"."cleanup_for_view_materialization"()
    RETURNS trigger
AS $$
BEGIN
    EXECUTE format('DROP VIEW %s', OLD.src_view);

    RETURN OLD;
END;
$$ LANGUAGE plpgsql VOLATILE;


CREATE FUNCTION "trend_directory"."undefine_materialization"("name" name)
    RETURNS void
AS $$
DELETE FROM trend_directory.materialization
WHERE materialization::text = $1;
$$ LANGUAGE sql VOLATILE;

COMMENT ON FUNCTION "trend_directory"."undefine_materialization"("name" name) IS 'Undefine and remove a materialization';


CREATE FUNCTION "trend_directory"."cleanup_for_function_materialization"()
    RETURNS trigger
AS $$
BEGIN
    EXECUTE format('DROP FUNCTION %s', OLD.src_function);

    RETURN OLD;
END;
$$ LANGUAGE plpgsql VOLATILE;


CREATE FUNCTION "trend_directory"."max_modified"(trend_directory.materialization, timestamp with time zone)
    RETURNS timestamp with time zone
AS $$
SELECT max(last) FROM trend_directory.modified
  WHERE trend_store_part_id = $1.dst_trend_store_part_id
  AND timestamp < $2;
$$ LANGUAGE sql STABLE;


CREATE FUNCTION "trend_directory"."fingerprint_function_name"(trend_directory.materialization)
    RETURNS name
AS $$
SELECT format('%s_fingerprint', $1::text)::name;
$$ LANGUAGE sql VOLATILE;


CREATE FUNCTION "trend_directory"."get_partition_size"(trend_directory.trend_store_part)
    RETURNS interval
AS $$
SELECT partition_size FROM trend_directory.trend_store WHERE trend_store.id = $1.trend_store_id;
$$ LANGUAGE sql STABLE;


CREATE FUNCTION "trend_directory"."create_partition"(trend_directory.trend_store_part, timestamp with time zone)
    RETURNS trend_directory.partition
AS $$
SELECT trend_directory.create_partition($1, trend_directory.timestamp_to_index(trend_directory.get_partition_size($1), $2));
$$ LANGUAGE sql VOLATILE;


CREATE FUNCTION "trend_directory"."get_trend_store_id"(trend_directory.trend_store)
    RETURNS integer
AS $$
SELECT $1.id;
$$ LANGUAGE sql VOLATILE;


CREATE FUNCTION "trend_directory"."delete_trend_store"("data_source_name" text, "entity_type_name" text, "granularity" interval)
    RETURNS void
AS $$
SELECT trend_directory.delete_trend_store((trend_directory.get_trend_store($1, $2, $3)).id);
$$ LANGUAGE sql STABLE;


CREATE FUNCTION "trend_directory"."get_trend_store_part_id"(trend_directory.trend_store_part)
    RETURNS integer
AS $$
SELECT $1.id;
$$ LANGUAGE sql VOLATILE;


CREATE FUNCTION "trend_directory"."assure_table_trends_exist"(trend_directory.trend_store_part, trend_directory.trend_descr[], trend_directory.generated_trend_descr[])
    RETURNS text[]
AS $$
DECLARE
  result text[];
BEGIN
  CREATE TEMP TABLE missing_trends(trend trend_directory.trend_descr);
  CREATE TEMP TABLE missing_generated_trends(trend trend_directory.generated_trend_descr);

  -- Normal trends
  INSERT INTO missing_trends SELECT trend_directory.missing_table_trends($1, $2);

  IF EXISTS (SELECT * FROM missing_trends LIMIT 1) THEN
    PERFORM trend_directory.create_table_trends($1, ARRAY(SELECT trend FROM missing_trends));
  END IF;

  -- Generated trends
  INSERT INTO missing_generated_trends SELECT trend_directory.missing_generated_table_trends($1, $3);

  IF EXISTS (SELECT * FROM missing_generated_trends LIMIT 1) THEN
    PERFORM trend_directory.create_generated_table_trends($1, missing_generated_trends);
  END IF;

  SELECT ARRAY(SELECT (mt).trend.name FROM missing_trends mt UNION SELECT (mt).trend.name FROM missing_generated_trends mt) INTO result;
  DROP TABLE missing_trends;
  DROP TABLE missing_generated_trends;

  RETURN result;
END;
$$ LANGUAGE plpgsql VOLATILE;


CREATE FUNCTION "trend_directory"."add_trends"(trend_directory.trend_store, "parts" trend_directory.trend_store_part_descr[])
    RETURNS text[]
AS $$
SELECT trend_directory.assure_table_trends_exist(
  trend_directory.get_or_create_trend_store_part($1.id, name),
  trends,
  generated_trends
)
FROM unnest($2);
$$ LANGUAGE sql VOLATILE;


CREATE FUNCTION "trend_directory"."get_trend_if_defined"("trend" trend_directory.table_trend, "trends" trend_directory.trend_descr[])
    RETURNS name
AS $$
SELECT t.name FROM trend_directory.table_trend t JOIN unnest($2) t2
  ON t.name = t2.name WHERE t.id = $1.id
$$ LANGUAGE sql VOLATILE;

COMMENT ON FUNCTION "trend_directory"."get_trend_if_defined"("trend" trend_directory.table_trend, "trends" trend_directory.trend_descr[]) IS 'Return the trend, but only if it is a trend defined by trends';


CREATE FUNCTION "trend_directory"."remove_trend_if_extraneous"("trend" trend_directory.table_trend, "trends" trend_directory.trend_descr[])
    RETURNS text
AS $$
DECLARE
  result text;
  defined_trend name;
BEGIN
  SELECT trend_directory.get_trend_if_defined($1, $2) INTO defined_trend;
  IF defined_trend IS NULL THEN
    SELECT $1.name INTO result;
    PERFORM trend_directory.remove_table_trend($1);
  END IF;
  RETURN result;
END;
$$ LANGUAGE plpgsql VOLATILE;

COMMENT ON FUNCTION "trend_directory"."remove_trend_if_extraneous"("trend" trend_directory.table_trend, "trends" trend_directory.trend_descr[]) IS 'Remove the trend if it is not one that is described by trends';


CREATE FUNCTION "trend_directory"."remove_extra_trends"(trend_directory.trend_store_part, trend_directory.trend_descr[])
    RETURNS text[]
AS $$
DECLARE
  trend trend_directory.table_trend;
  removal_result text;
  result text[];
BEGIN
  FOR trend IN SELECT * FROM trend_directory.get_trends_for_trend_store_part($1)
  LOOP
    SELECT trend_directory.remove_trend_if_extraneous(trend, $2) INTO removal_result;
    IF removal_result IS NOT NULL THEN
      SELECT result || removal_result INTO result;
    END IF;
  END LOOP;
  RETURN result;
END;
$$ LANGUAGE plpgsql VOLATILE;


CREATE FUNCTION "trend_directory"."remove_extra_trends"(trend_directory.trend_store, "parts" trend_directory.trend_store_part_descr[])
    RETURNS text[]
AS $$
DECLARE
  result text[];
  partresult text[];
BEGIN
  FOR partresult IN
    SELECT trend_directory.remove_extra_trends(
      trend_directory.get_trend_store_part($1.id, name), trends)
    FROM unnest($2)
  LOOP
    SELECT result || partresult INTO result;
  END LOOP;
  RETURN result;
END;
$$ LANGUAGE plpgsql VOLATILE;


CREATE FUNCTION "trend_directory"."change_table_trend_data_unsafe"(trend_directory.table_trend, "data_type" text, "entity_aggregation" text, "time_aggregation" text)
    RETURNS text
AS $$
DECLARE
  result text;
BEGIN
  IF $1.data_type <> $2 OR $1.entity_aggregation <> $3 OR $1.time_aggregation <> $4
  THEN
    UPDATE trend_directory.table_trend SET
      data_type = $2,
      entity_aggregation = $3,
      time_aggregation = $4
    WHERE id = $1.id;
    EXECUTE format('ALTER TABLE trend.%I ALTER %I TYPE %s USING CAST(%I AS %s)',
      trend_directory.trend_store_part_name_for_trend($1),
      $1.name,
      $2,
      $1.name,
      $2);
    SELECT $1.name INTO result;
  END IF;
  RETURN result;
END;
$$ LANGUAGE plpgsql VOLATILE;


CREATE FUNCTION "trend_directory"."change_table_trend_data_safe"(trend_directory.table_trend, "data_type" text, "entity_aggregation" text, "time_aggregation" text)
    RETURNS text
AS $$
SELECT trend_directory.change_table_trend_data_unsafe(
  $1,
  trend_directory.greatest_data_type($2, $1.data_type),
  $3,
  $4);
$$ LANGUAGE sql VOLATILE;


CREATE FUNCTION "trend_directory"."change_trend_data_unsafe"("trend" trend_directory.table_trend, "trends" trend_directory.trend_descr[], "partname" text)
    RETURNS text
AS $$
SELECT trend_directory.change_table_trend_data_unsafe($1, t.data_type, t.entity_aggregation, t.time_aggregation)
  FROM unnest($2) t
  WHERE t.name = $1.name AND trend_directory.trend_store_part_name_for_trend($1) = $3;
$$ LANGUAGE sql VOLATILE;


CREATE FUNCTION "trend_directory"."change_trend_data_safe"("trend" trend_directory.table_trend, "trends" trend_directory.trend_descr[], "partname" text)
    RETURNS text
AS $$
SELECT trend_directory.change_table_trend_data_safe($1, t.data_type, t.entity_aggregation, t.time_aggregation)
  FROM unnest($2) t
  WHERE t.name = $1.name AND trend_directory.trend_store_part_name_for_trend($1) = $3;
$$ LANGUAGE sql VOLATILE;


CREATE FUNCTION "trend_directory"."change_all_trend_data"(trend_directory.trend_store, "parts" trend_directory.trend_store_part_descr[])
    RETURNS text[]
AS $$
DECLARE
  result text[];
  partresult text;
BEGIN
  FOR partresult IN
    SELECT trend_directory.change_trend_data_unsafe(
      trend_directory.get_trends_for_trend_store($1), trends, name)
    FROM unnest($2)
  LOOP
    IF partresult IS NOT null THEN
      SELECT result || partresult INTO result;
    END IF;
  END LOOP;
  RETURN result;
END;
$$ LANGUAGE plpgsql VOLATILE;


CREATE FUNCTION "trend_directory"."change_trend_data_upward"(trend_directory.trend_store, "parts" trend_directory.trend_store_part_descr[])
    RETURNS text[]
AS $$
DECLARE
  result text[];
  partresult text;
BEGIN
  FOR partresult IN
    SELECT trend_directory.change_trend_data_safe(
      trend_directory.get_trends_for_trend_store($1), trends, name)
    FROM unnest($2)
  LOOP
    IF partresult IS NOT null THEN
      SELECT result || partresult INTO result;
    END IF;
  END LOOP;
  RETURN result;
END;
$$ LANGUAGE plpgsql VOLATILE;


CREATE FUNCTION "trend_directory"."change_trendstore_strong"(trend_directory.trend_store, "parts" trend_directory.trend_store_part_descr[])
    RETURNS text[]
AS $$
DECLARE
  result text[];
  partresult text[];
BEGIN
  SELECT trend_directory.add_trends($1, $2) INTO partresult;
  IF array_ndims(partresult) > 0
  THEN
    SELECT result || ARRAY['added trends:'] || partresult INTO result;
  ELSE
    SELECT result || ARRAY['no trends added'] INTO result;
  END IF;
  
  SELECT trend_directory.remove_extra_trends($1, $2) INTO partresult;
  IF array_ndims(partresult) > 0
  THEN
    SELECT result || ARRAY['removed trends:'] || partresult INTO result;
  ELSE
    SELECT result || ARRAY['no trends removed'] INTO result;
  END IF;

  SELECT trend_directory.change_all_trend_data($1, $2) INTO partresult;
  IF array_ndims(partresult) > 0
  THEN
    SELECT result || ARRAY['changed trends:'] || partresult INTO result;
  ELSE
    SELECT result || ARRAY['no trends changed'] INTO result;
  END IF;
  RETURN result;
END;
$$ LANGUAGE plpgsql VOLATILE;


CREATE FUNCTION "trend_directory"."change_trendstore_weak"(trend_directory.trend_store, "parts" trend_directory.trend_store_part_descr[])
    RETURNS text[]
AS $$
DECLARE
  result text[];
  partresult text[];
BEGIN
  SELECT trend_directory.add_trends($1, $2) INTO partresult;
  IF array_ndims(partresult) > 0
  THEN
    SELECT result || ARRAY['added trends:'] || partresult INTO result;
  ELSE
    SELECT result || ARRAY['no trends added'] INTO result;
  END IF;
  
  SELECT trend_directory.remove_extra_trends($1, $2) INTO partresult;
  IF array_ndims(partresult) > 0
  THEN
    SELECT result || ARRAY['removed trends:'] || partresult INTO result;
  ELSE
    SELECT result || ARRAY['no trends removed'] INTO result;
  END IF;

  SELECT trend_directory.change_trend_data_upward($1, $2) INTO partresult;
  IF array_ndims(partresult) > 0
  THEN
    SELECT result || ARRAY['changed trends:'] || partresult INTO result;
  ELSE
    SELECT result || ARRAY['no trends changed'] INTO result;
  END IF;
  RETURN result;
END;
$$ LANGUAGE plpgsql VOLATILE;


CREATE FUNCTION "trend_directory"."get_table_name_for_trend"("trend" text, "entity" text, "granularity" interval)
    RETURNS name
AS $$
SELECT tsp.name FROM trend_directory.table_trend t
  JOIN trend_directory.trend_store_part tsp ON tsp.id = t.trend_store_part_id
  JOIN trend_directory.trend_store ts ON ts.id = tsp.trend_store_id
  JOIN directory.entity_type et ON et.id = ts.entity_type_id
  WHERE t.name = $1
    AND ts.granularity = $3
    AND et.name = $2;
$$ LANGUAGE sql STABLE;


CREATE OR REPLACE FUNCTION "trend_directory"."update_source_fingerprint"(trend_directory.materialization, timestamp with time zone)
    RETURNS void
AS $$
INSERT INTO trend_directory.materialization_state(materialization_id, timestamp, source_fingerprint, max_modified, processed_fingerprint, job_id)
VALUES ($1.id, $2, (trend_directory.source_fingerprint($1, $2)).body, trend_directory.max_modified($1, $2), null, null)
ON CONFLICT ON CONSTRAINT materialization_state_pkey DO UPDATE SET source_fingerprint = (trend_directory.source_fingerprint($1, $2)).body, max_modified = trend_directory.max_modified($1, $2);
$$ LANGUAGE sql VOLATILE;

COMMENT ON FUNCTION "trend_directory"."update_source_fingerprint"(trend_directory.materialization, timestamp with time zone) IS 'Update the fingerprint of the sources in the materialization_state table.';


CREATE OR REPLACE FUNCTION "trend_directory"."create_base_table_sql"(trend_directory.trend_store_part)
    RETURNS text[]
AS $$
SELECT ARRAY[
    format(
        'CREATE TABLE %I.%I ('
        'entity_id integer NOT NULL, '
        '"timestamp" timestamp with time zone NOT NULL, '
        'created timestamp with time zone NOT NULL, '
        '%s'
        ') PARTITION BY RANGE ("timestamp");',
        trend_directory.base_table_schema(),
        trend_directory.base_table_name($1),
        array_to_string(ARRAY['job_id bigint NOT NULL'] || trend_directory.column_specs($1), ',')
    ),
    format(
        'ALTER TABLE %I.%I ADD PRIMARY KEY (entity_id, "timestamp");',
        trend_directory.base_table_schema(),
        trend_directory.base_table_name($1)
    ),
    format(
        'CREATE INDEX ON %I.%I USING btree (job_id)',
        trend_directory.base_table_schema(),
        trend_directory.base_table_name($1)
    ),
    format(
        'CREATE INDEX ON %I.%I USING btree (timestamp);',
        trend_directory.base_table_schema(),
        trend_directory.base_table_name($1)
    ),
    format(
        'GRANT SELECT ON TABLE %I.%I TO minerva;',
        trend_directory.base_table_schema(),
        trend_directory.base_table_name($1)
    ),
    format(
        'GRANT INSERT,DELETE,UPDATE ON TABLE %I.%I TO minerva_writer;',
        trend_directory.base_table_schema(),
        trend_directory.base_table_name($1)
    )
];
$$ LANGUAGE sql STABLE STRICT;


CREATE OR REPLACE FUNCTION "trend_directory"."create_staging_table_sql"(trend_directory.trend_store_part)
    RETURNS text[]
AS $$
SELECT ARRAY[
    format(
        'CREATE UNLOGGED TABLE %I.%I (entity_id integer, "timestamp" timestamp with time zone, created timestamp with time zone, job_id integer%s);',
        trend_directory.staging_table_schema(),
        trend_directory.staging_table_name($1),
        (
            SELECT string_agg(format(', %I %s', t.name, t.data_type), ' ')
            FROM trend_directory.table_trend t
            WHERE t.trend_store_part_id = $1.id
        )
    ),
    format(
        'ALTER TABLE ONLY %I.%I ADD PRIMARY KEY (entity_id, "timestamp");',
        trend_directory.staging_table_schema(),
        trend_directory.staging_table_name($1)
    ),
    format(
        'GRANT SELECT ON TABLE %I.%I TO minerva;',
        trend_directory.staging_table_schema(),
        trend_directory.staging_table_name($1)
    ),
    format(
        'GRANT INSERT,DELETE,UPDATE ON TABLE %I.%I TO minerva_writer;',
        trend_directory.staging_table_schema(),
        trend_directory.staging_table_name($1)
    )
];
$$ LANGUAGE sql STABLE;


CREATE OR REPLACE FUNCTION "trend_directory"."get_trend_store_parts"("trend_store_id" integer)
    RETURNS trend_directory.trend_store_part
AS $$
SELECT trend_store_part FROM trend_directory.trend_store_part WHERE trend_store_id = $1;
$$ LANGUAGE sql VOLATILE;


CREATE OR REPLACE FUNCTION "trend_directory"."alter_trend_name"(trend_directory.trend_store_part, "trend_name" name, "new_name" name)
    RETURNS trend_directory.trend_store_part
AS $$
UPDATE trend_directory.table_trend
  SET name = $3
  WHERE trend_store_part_id = $1.id AND name = $2
  RETURNING $1;
$$ LANGUAGE sql VOLATILE;


CREATE OR REPLACE FUNCTION "trend_directory"."get_trends_for_trend_store"("trend_store_id" integer)
    RETURNS SETOF trend_directory.table_trend
AS $$
SELECT table_trend
  FROM trend_directory.table_trend
  LEFT JOIN trend_directory.trend_store_part
  ON table_trend.trend_store_part_id = trend_store_part.id
  WHERE trend_store_part.trend_store_id = $1;
$$ LANGUAGE sql STABLE;


CREATE OR REPLACE FUNCTION "trend_directory"."transfer"("materialization" trend_directory.view_materialization, "timestamp" timestamp with time zone)
    RETURNS integer
AS $$
DECLARE
    columns_part text;
    row_count integer;
    job_id integer;
BEGIN
    SELECT logging.start_job(format('{"view_materialization": "%s", "timestamp": "%s"}', m::text, $2::text)::jsonb) INTO job_id
    FROM trend_directory.materialization m WHERE id = $1.materialization_id;

    SELECT trend_directory.columns_part($1) INTO columns_part;

    EXECUTE format(
        'INSERT INTO trend.%I (entity_id, timestamp, created, job_id, %s) SELECT entity_id, timestamp, now(), %s, %s FROM %s WHERE timestamp = $1',
        (trend_directory.dst_trend_store_part($1)).name,
        columns_part,
        job_id,
        columns_part,
        $1.src_view::name
    ) USING timestamp;

    PERFORM logging.end_job(job_id);

    GET DIAGNOSTICS row_count = ROW_COUNT;

    RETURN row_count;
END;
$$ LANGUAGE plpgsql VOLATILE;

COMMENT ON FUNCTION "trend_directory"."transfer"("materialization" trend_directory.view_materialization, "timestamp" timestamp with time zone) IS 'Transfer all records of the specified timestamp from the view to the target trend store of the materialization.';


CREATE OR REPLACE FUNCTION "trend_directory"."transfer"("materialization" trend_directory.function_materialization, "timestamp" timestamp with time zone)
    RETURNS integer
AS $$
DECLARE
    columns_part text;
    row_count integer;
    job_id integer;
BEGIN
    SELECT logging.start_job(format('{"function_materialization": "%s", "timestamp": "%s"}', m::text, $2::text)::jsonb) INTO job_id
    FROM trend_directory.materialization m WHERE id = $1.materialization_id;

    SELECT trend_directory.columns_part($1) INTO columns_part;

    EXECUTE format(
        'INSERT INTO trend.%I (entity_id, timestamp, created, job_id, %s) SELECT entity_id, timestamp, now(), %s, %s FROM %s($1)',
        (trend_directory.dst_trend_store_part($1)).name,
        columns_part,
        job_id,
        columns_part,
        $1.src_function::regproc
    ) USING timestamp;

    PERFORM logging.end_job(job_id);

    GET DIAGNOSTICS row_count = ROW_COUNT;

    RETURN row_count;
END;
$$ LANGUAGE plpgsql VOLATILE;


CREATE OR REPLACE FUNCTION "trend_directory"."changes_on_trend_update"()
    RETURNS trigger
AS $$
DECLARE
    base_table_name text;
BEGIN
    IF NEW.name <> OLD.name THEN
        FOR base_table_name IN
            SELECT trend_directory.base_table_name(trend_store_part)
            FROM trend_directory.table_trend
            JOIN trend_directory.trend_store_part ON table_trend.trend_store_part_id = trend_store_part.id
            WHERE table_trend.id = NEW.id
        LOOP
            EXECUTE format('ALTER TABLE trend.%I RENAME COLUMN %I TO %I', base_table_name, OLD.name, NEW.name);
        END LOOP;
    END IF;

    RETURN NEW;
END;
$$ LANGUAGE plpgsql VOLATILE;


DROP FUNCTION "attribute_directory"."greatest_data_type"(varchar, varchar);

DROP FUNCTION "attribute_directory"."update_compacted"(integer, timestamp with time zone);

DROP FUNCTION "attribute_directory"."store_compacted"(integer, timestamp with time zone);

DROP FUNCTION "attribute_directory"."mark_compacted"(integer, timestamp with time zone);

DROP FUNCTION "attribute_directory"."mark_compacted"(integer);

ALTER TABLE "attribute_directory"."attribute_store_compacted" DROP COLUMN "compacted";
ALTER TABLE "attribute_directory"."attribute_store_compacted" ADD COLUMN "compacted" integer NOT NULL;


CREATE FUNCTION "attribute_directory"."get_attribute_store"("data_source" text, "entity_type" text)
    RETURNS attribute_directory.attribute_store
AS $$
SELECT attribute_store
FROM attribute_directory.attribute_store
LEFT JOIN directory.data_source
  ON data_source_id = data_source.id
LEFT JOIN directory.entity_type<commit_msg>Fix migration<commit_after>

CREATE OR REPLACE FUNCTION "system"."version"()
    RETURNS system.version_tuple
AS $$
SELECT (5,1,6)::system.version_tuple;
$$ LANGUAGE sql IMMUTABLE;


DROP FUNCTION "relation_directory"."remove"(name);

CREATE FUNCTION "relation_directory"."remove"(name)
    RETURNS text
AS $$
DECLARE
  result text;
BEGIN
  SELECT name FROM relation_directory.type WHERE name = $1 INTO result;
  PERFORM public.action(format('DROP MATERIALIZED VIEW IF EXISTS relation.%I', $1));
  DELETE FROM relation_directory.type WHERE name = $1;
  RETURN result;
END;
$$ LANGUAGE plpgsql VOLATILE;


DROP FUNCTION "trend_directory"."assure_table_trends_exist"(trend_directory.trend_store_part, trend_directory.trend_descr[], trend_directory.generated_trend_descr[]);

DROP FUNCTION "trend_directory"."add_trends"(trend_directory.trend_store, trend_directory.trend_store_part_descr[]);

DROP FUNCTION "trend_directory"."get_trend_if_defined"(trend_directory.table_trend, trend_directory.trend_descr[], text);

DROP FUNCTION "trend_directory"."remove_trend_if_extraneous"(trend_directory.table_trend, trend_directory.trend_store_part_descr[]);

DROP FUNCTION "trend_directory"."remove_extra_trends"(trend_directory.trend_store, trend_directory.trend_store_part_descr[]);

DROP FUNCTION "trend_directory"."change_table_trend_data_unsafe"(trend_directory.table_trend, text, text, text);

DROP FUNCTION "trend_directory"."change_table_trend_data_safe"(trend_directory.table_trend, text, text, text);

DROP FUNCTION "trend_directory"."change_trend_data_unsafe"(trend_directory.table_trend, trend_directory.trend_descr[], text);

DROP FUNCTION "trend_directory"."change_trend_data_safe"(trend_directory.table_trend, trend_directory.trend_descr[], text);

DROP FUNCTION "trend_directory"."change_all_trend_data"(trend_directory.trend_store, trend_directory.trend_store_part_descr[]);

DROP FUNCTION "trend_directory"."change_trend_data_upward"(trend_directory.trend_store, trend_directory.trend_store_part_descr[]);

DROP FUNCTION "trend_directory"."change_trendstore_strong"(trend_directory.trend_store, trend_directory.trend_store_part_descr[]);

DROP FUNCTION "trend_directory"."change_trendstore_weak"(trend_directory.trend_store, trend_directory.trend_store_part_descr[]);

ALTER TABLE "trend_directory"."materialization_state" ADD COLUMN "max_modified" timestamp with time zone;


CREATE FUNCTION "trend_directory"."cleanup_for_materialization"()
    RETURNS trigger
AS $$
BEGIN
  EXECUTE format(
    'DROP FUNCTION trend.%I(timestamp with time zone)',
    trend_directory.fingerprint_function_name(OLD)
  );

  RETURN OLD;
END;
$$ LANGUAGE plpgsql VOLATILE;


CREATE FUNCTION "trend_directory"."cleanup_for_view_materialization"()
    RETURNS trigger
AS $$
BEGIN
    EXECUTE format('DROP VIEW %s', OLD.src_view);

    RETURN OLD;
END;
$$ LANGUAGE plpgsql VOLATILE;


CREATE FUNCTION "trend_directory"."undefine_materialization"("name" name)
    RETURNS void
AS $$
DELETE FROM trend_directory.materialization
WHERE materialization::text = $1;
$$ LANGUAGE sql VOLATILE;

COMMENT ON FUNCTION "trend_directory"."undefine_materialization"("name" name) IS 'Undefine and remove a materialization';


CREATE FUNCTION "trend_directory"."cleanup_for_function_materialization"()
    RETURNS trigger
AS $$
BEGIN
    EXECUTE format('DROP FUNCTION %s', OLD.src_function);

    RETURN OLD;
END;
$$ LANGUAGE plpgsql VOLATILE;


CREATE FUNCTION "trend_directory"."max_modified"(trend_directory.materialization, timestamp with time zone)
    RETURNS timestamp with time zone
AS $$
SELECT max(last) FROM trend_directory.modified
  WHERE trend_store_part_id = $1.dst_trend_store_part_id
  AND timestamp < $2;
$$ LANGUAGE sql STABLE;


CREATE FUNCTION "trend_directory"."fingerprint_function_name"(trend_directory.materialization)
    RETURNS name
AS $$
SELECT format('%s_fingerprint', $1::text)::name;
$$ LANGUAGE sql VOLATILE;


CREATE FUNCTION "trend_directory"."get_partition_size"(trend_directory.trend_store_part)
    RETURNS interval
AS $$
SELECT partition_size FROM trend_directory.trend_store WHERE trend_store.id = $1.trend_store_id;
$$ LANGUAGE sql STABLE;


CREATE FUNCTION "trend_directory"."create_partition"(trend_directory.trend_store_part, timestamp with time zone)
    RETURNS trend_directory.partition
AS $$
SELECT trend_directory.create_partition($1, trend_directory.timestamp_to_index(trend_directory.get_partition_size($1), $2));
$$ LANGUAGE sql VOLATILE;


CREATE FUNCTION "trend_directory"."get_trend_store_id"(trend_directory.trend_store)
    RETURNS integer
AS $$
SELECT $1.id;
$$ LANGUAGE sql VOLATILE;


CREATE FUNCTION "trend_directory"."delete_trend_store"("data_source_name" text, "entity_type_name" text, "granularity" interval)
    RETURNS void
AS $$
SELECT trend_directory.delete_trend_store((trend_directory.get_trend_store($1, $2, $3)).id);
$$ LANGUAGE sql STABLE;


CREATE FUNCTION "trend_directory"."get_trend_store_part_id"(trend_directory.trend_store_part)
    RETURNS integer
AS $$
SELECT $1.id;
$$ LANGUAGE sql VOLATILE;


CREATE FUNCTION "trend_directory"."assure_table_trends_exist"(trend_directory.trend_store_part, trend_directory.trend_descr[], trend_directory.generated_trend_descr[])
    RETURNS text[]
AS $$
DECLARE
  result text[];
BEGIN
  CREATE TEMP TABLE missing_trends(trend trend_directory.trend_descr);
  CREATE TEMP TABLE missing_generated_trends(trend trend_directory.generated_trend_descr);

  -- Normal trends
  INSERT INTO missing_trends SELECT trend_directory.missing_table_trends($1, $2);

  IF EXISTS (SELECT * FROM missing_trends LIMIT 1) THEN
    PERFORM trend_directory.create_table_trends($1, ARRAY(SELECT trend FROM missing_trends));
  END IF;

  -- Generated trends
  INSERT INTO missing_generated_trends SELECT trend_directory.missing_generated_table_trends($1, $3);

  IF EXISTS (SELECT * FROM missing_generated_trends LIMIT 1) THEN
    PERFORM trend_directory.create_generated_table_trends($1, missing_generated_trends);
  END IF;

  SELECT ARRAY(SELECT (mt).trend.name FROM missing_trends mt UNION SELECT (mt).trend.name FROM missing_generated_trends mt) INTO result;
  DROP TABLE missing_trends;
  DROP TABLE missing_generated_trends;

  RETURN result;
END;
$$ LANGUAGE plpgsql VOLATILE;


CREATE FUNCTION "trend_directory"."add_trends"(trend_directory.trend_store, "parts" trend_directory.trend_store_part_descr[])
    RETURNS text[]
AS $$
SELECT trend_directory.assure_table_trends_exist(
  trend_directory.get_or_create_trend_store_part($1.id, name),
  trends,
  generated_trends
)
FROM unnest($2);
$$ LANGUAGE sql VOLATILE;


CREATE FUNCTION "trend_directory"."get_trend_if_defined"("trend" trend_directory.table_trend, "trends" trend_directory.trend_descr[])
    RETURNS name
AS $$
SELECT t.name FROM trend_directory.table_trend t JOIN unnest($2) t2
  ON t.name = t2.name WHERE t.id = $1.id
$$ LANGUAGE sql VOLATILE;

COMMENT ON FUNCTION "trend_directory"."get_trend_if_defined"("trend" trend_directory.table_trend, "trends" trend_directory.trend_descr[]) IS 'Return the trend, but only if it is a trend defined by trends';


CREATE FUNCTION "trend_directory"."remove_trend_if_extraneous"("trend" trend_directory.table_trend, "trends" trend_directory.trend_descr[])
    RETURNS text
AS $$
DECLARE
  result text;
  defined_trend name;
BEGIN
  SELECT trend_directory.get_trend_if_defined($1, $2) INTO defined_trend;
  IF defined_trend IS NULL THEN
    SELECT $1.name INTO result;
    PERFORM trend_directory.remove_table_trend($1);
  END IF;
  RETURN result;
END;
$$ LANGUAGE plpgsql VOLATILE;

COMMENT ON FUNCTION "trend_directory"."remove_trend_if_extraneous"("trend" trend_directory.table_trend, "trends" trend_directory.trend_descr[]) IS 'Remove the trend if it is not one that is described by trends';


CREATE FUNCTION "trend_directory"."remove_extra_trends"(trend_directory.trend_store_part, trend_directory.trend_descr[])
    RETURNS text[]
AS $$
DECLARE
  trend trend_directory.table_trend;
  removal_result text;
  result text[];
BEGIN
  FOR trend IN SELECT * FROM trend_directory.get_trends_for_trend_store_part($1)
  LOOP
    SELECT trend_directory.remove_trend_if_extraneous(trend, $2) INTO removal_result;
    IF removal_result IS NOT NULL THEN
      SELECT result || removal_result INTO result;
    END IF;
  END LOOP;
  RETURN result;
END;
$$ LANGUAGE plpgsql VOLATILE;


CREATE FUNCTION "trend_directory"."remove_extra_trends"(trend_directory.trend_store, "parts" trend_directory.trend_store_part_descr[])
    RETURNS text[]
AS $$
DECLARE
  result text[];
  partresult text[];
BEGIN
  FOR partresult IN
    SELECT trend_directory.remove_extra_trends(
      trend_directory.get_trend_store_part($1.id, name), trends)
    FROM unnest($2)
  LOOP
    SELECT result || partresult INTO result;
  END LOOP;
  RETURN result;
END;
$$ LANGUAGE plpgsql VOLATILE;


CREATE FUNCTION "trend_directory"."change_table_trend_data_unsafe"(trend_directory.table_trend, "data_type" text, "entity_aggregation" text, "time_aggregation" text)
    RETURNS text
AS $$
DECLARE
  result text;
BEGIN
  IF $1.data_type <> $2 OR $1.entity_aggregation <> $3 OR $1.time_aggregation <> $4
  THEN
    UPDATE trend_directory.table_trend SET
      data_type = $2,
      entity_aggregation = $3,
      time_aggregation = $4
    WHERE id = $1.id;
    EXECUTE format('ALTER TABLE trend.%I ALTER %I TYPE %s USING CAST(%I AS %s)',
      trend_directory.trend_store_part_name_for_trend($1),
      $1.name,
      $2,
      $1.name,
      $2);
    SELECT $1.name INTO result;
  END IF;
  RETURN result;
END;
$$ LANGUAGE plpgsql VOLATILE;


CREATE FUNCTION "trend_directory"."change_table_trend_data_safe"(trend_directory.table_trend, "data_type" text, "entity_aggregation" text, "time_aggregation" text)
    RETURNS text
AS $$
SELECT trend_directory.change_table_trend_data_unsafe(
  $1,
  trend_directory.greatest_data_type($2, $1.data_type),
  $3,
  $4);
$$ LANGUAGE sql VOLATILE;


CREATE FUNCTION "trend_directory"."change_trend_data_unsafe"("trend" trend_directory.table_trend, "trends" trend_directory.trend_descr[], "partname" text)
    RETURNS text
AS $$
SELECT trend_directory.change_table_trend_data_unsafe($1, t.data_type, t.entity_aggregation, t.time_aggregation)
  FROM unnest($2) t
  WHERE t.name = $1.name AND trend_directory.trend_store_part_name_for_trend($1) = $3;
$$ LANGUAGE sql VOLATILE;


CREATE FUNCTION "trend_directory"."change_trend_data_safe"("trend" trend_directory.table_trend, "trends" trend_directory.trend_descr[], "partname" text)
    RETURNS text
AS $$
SELECT trend_directory.change_table_trend_data_safe($1, t.data_type, t.entity_aggregation, t.time_aggregation)
  FROM unnest($2) t
  WHERE t.name = $1.name AND trend_directory.trend_store_part_name_for_trend($1) = $3;
$$ LANGUAGE sql VOLATILE;


CREATE FUNCTION "trend_directory"."change_all_trend_data"(trend_directory.trend_store, "parts" trend_directory.trend_store_part_descr[])
    RETURNS text[]
AS $$
DECLARE
  result text[];
  partresult text;
BEGIN
  FOR partresult IN
    SELECT trend_directory.change_trend_data_unsafe(
      trend_directory.get_trends_for_trend_store($1), trends, name)
    FROM unnest($2)
  LOOP
    IF partresult IS NOT null THEN
      SELECT result || partresult INTO result;
    END IF;
  END LOOP;
  RETURN result;
END;
$$ LANGUAGE plpgsql VOLATILE;


CREATE FUNCTION "trend_directory"."change_trend_data_upward"(trend_directory.trend_store, "parts" trend_directory.trend_store_part_descr[])
    RETURNS text[]
AS $$
DECLARE
  result text[];
  partresult text;
BEGIN
  FOR partresult IN
    SELECT trend_directory.change_trend_data_safe(
      trend_directory.get_trends_for_trend_store($1), trends, name)
    FROM unnest($2)
  LOOP
    IF partresult IS NOT null THEN
      SELECT result || partresult INTO result;
    END IF;
  END LOOP;
  RETURN result;
END;
$$ LANGUAGE plpgsql VOLATILE;


CREATE FUNCTION "trend_directory"."change_trendstore_strong"(trend_directory.trend_store, "parts" trend_directory.trend_store_part_descr[])
    RETURNS text[]
AS $$
DECLARE
  result text[];
  partresult text[];
BEGIN
  SELECT trend_directory.add_trends($1, $2) INTO partresult;
  IF array_ndims(partresult) > 0
  THEN
    SELECT result || ARRAY['added trends:'] || partresult INTO result;
  ELSE
    SELECT result || ARRAY['no trends added'] INTO result;
  END IF;
  
  SELECT trend_directory.remove_extra_trends($1, $2) INTO partresult;
  IF array_ndims(partresult) > 0
  THEN
    SELECT result || ARRAY['removed trends:'] || partresult INTO result;
  ELSE
    SELECT result || ARRAY['no trends removed'] INTO result;
  END IF;

  SELECT trend_directory.change_all_trend_data($1, $2) INTO partresult;
  IF array_ndims(partresult) > 0
  THEN
    SELECT result || ARRAY['changed trends:'] || partresult INTO result;
  ELSE
    SELECT result || ARRAY['no trends changed'] INTO result;
  END IF;
  RETURN result;
END;
$$ LANGUAGE plpgsql VOLATILE;


CREATE FUNCTION "trend_directory"."change_trendstore_weak"(trend_directory.trend_store, "parts" trend_directory.trend_store_part_descr[])
    RETURNS text[]
AS $$
DECLARE
  result text[];
  partresult text[];
BEGIN
  SELECT trend_directory.add_trends($1, $2) INTO partresult;
  IF array_ndims(partresult) > 0
  THEN
    SELECT result || ARRAY['added trends:'] || partresult INTO result;
  ELSE
    SELECT result || ARRAY['no trends added'] INTO result;
  END IF;
  
  SELECT trend_directory.remove_extra_trends($1, $2) INTO partresult;
  IF array_ndims(partresult) > 0
  THEN
    SELECT result || ARRAY['removed trends:'] || partresult INTO result;
  ELSE
    SELECT result || ARRAY['no trends removed'] INTO result;
  END IF;

  SELECT trend_directory.change_trend_data_upward($1, $2) INTO partresult;
  IF array_ndims(partresult) > 0
  THEN
    SELECT result || ARRAY['changed trends:'] || partresult INTO result;
  ELSE
    SELECT result || ARRAY['no trends changed'] INTO result;
  END IF;
  RETURN result;
END;
$$ LANGUAGE plpgsql VOLATILE;


CREATE FUNCTION "trend_directory"."get_table_name_for_trend"("trend" text, "entity" text, "granularity" interval)
    RETURNS name
AS $$
SELECT tsp.name FROM trend_directory.table_trend t
  JOIN trend_directory.trend_store_part tsp ON tsp.id = t.trend_store_part_id
  JOIN trend_directory.trend_store ts ON ts.id = tsp.trend_store_id
  JOIN directory.entity_type et ON et.id = ts.entity_type_id
  WHERE t.name = $1
    AND ts.granularity = $3
    AND et.name = $2;
$$ LANGUAGE sql STABLE;


CREATE OR REPLACE FUNCTION "trend_directory"."update_source_fingerprint"(trend_directory.materialization, timestamp with time zone)
    RETURNS void
AS $$
INSERT INTO trend_directory.materialization_state(materialization_id, timestamp, source_fingerprint, max_modified, processed_fingerprint, job_id)
VALUES ($1.id, $2, (trend_directory.source_fingerprint($1, $2)).body, trend_directory.max_modified($1, $2), null, null)
ON CONFLICT ON CONSTRAINT materialization_state_pkey DO UPDATE SET source_fingerprint = (trend_directory.source_fingerprint($1, $2)).body, max_modified = trend_directory.max_modified($1, $2);
$$ LANGUAGE sql VOLATILE;

COMMENT ON FUNCTION "trend_directory"."update_source_fingerprint"(trend_directory.materialization, timestamp with time zone) IS 'Update the fingerprint of the sources in the materialization_state table.';


CREATE OR REPLACE FUNCTION "trend_directory"."create_base_table_sql"(trend_directory.trend_store_part)
    RETURNS text[]
AS $$
SELECT ARRAY[
    format(
        'CREATE TABLE %I.%I ('
        'entity_id integer NOT NULL, '
        '"timestamp" timestamp with time zone NOT NULL, '
        'created timestamp with time zone NOT NULL, '
        '%s'
        ') PARTITION BY RANGE ("timestamp");',
        trend_directory.base_table_schema(),
        trend_directory.base_table_name($1),
        array_to_string(ARRAY['job_id bigint NOT NULL'] || trend_directory.column_specs($1), ',')
    ),
    format(
        'ALTER TABLE %I.%I ADD PRIMARY KEY (entity_id, "timestamp");',
        trend_directory.base_table_schema(),
        trend_directory.base_table_name($1)
    ),
    format(
        'CREATE INDEX ON %I.%I USING btree (job_id)',
        trend_directory.base_table_schema(),
        trend_directory.base_table_name($1)
    ),
    format(
        'CREATE INDEX ON %I.%I USING btree (timestamp);',
        trend_directory.base_table_schema(),
        trend_directory.base_table_name($1)
    ),
    format(
        'GRANT SELECT ON TABLE %I.%I TO minerva;',
        trend_directory.base_table_schema(),
        trend_directory.base_table_name($1)
    ),
    format(
        'GRANT INSERT,DELETE,UPDATE ON TABLE %I.%I TO minerva_writer;',
        trend_directory.base_table_schema(),
        trend_directory.base_table_name($1)
    )
];
$$ LANGUAGE sql STABLE STRICT;


CREATE OR REPLACE FUNCTION "trend_directory"."create_staging_table_sql"(trend_directory.trend_store_part)
    RETURNS text[]
AS $$
SELECT ARRAY[
    format(
        'CREATE UNLOGGED TABLE %I.%I (entity_id integer, "timestamp" timestamp with time zone, created timestamp with time zone, job_id integer%s);',
        trend_directory.staging_table_schema(),
        trend_directory.staging_table_name($1),
        (
            SELECT string_agg(format(', %I %s', t.name, t.data_type), ' ')
            FROM trend_directory.table_trend t
            WHERE t.trend_store_part_id = $1.id
        )
    ),
    format(
        'ALTER TABLE ONLY %I.%I ADD PRIMARY KEY (entity_id, "timestamp");',
        trend_directory.staging_table_schema(),
        trend_directory.staging_table_name($1)
    ),
    format(
        'GRANT SELECT ON TABLE %I.%I TO minerva;',
        trend_directory.staging_table_schema(),
        trend_directory.staging_table_name($1)
    ),
    format(
        'GRANT INSERT,DELETE,UPDATE ON TABLE %I.%I TO minerva_writer;',
        trend_directory.staging_table_schema(),
        trend_directory.staging_table_name($1)
    )
];
$$ LANGUAGE sql STABLE;


CREATE OR REPLACE FUNCTION "trend_directory"."get_trend_store_parts"("trend_store_id" integer)
    RETURNS trend_directory.trend_store_part
AS $$
SELECT trend_store_part FROM trend_directory.trend_store_part WHERE trend_store_id = $1;
$$ LANGUAGE sql VOLATILE;


CREATE OR REPLACE FUNCTION "trend_directory"."alter_trend_name"(trend_directory.trend_store_part, "trend_name" name, "new_name" name)
    RETURNS trend_directory.trend_store_part
AS $$
UPDATE trend_directory.table_trend
  SET name = $3
  WHERE trend_store_part_id = $1.id AND name = $2
  RETURNING $1;
$$ LANGUAGE sql VOLATILE;


CREATE OR REPLACE FUNCTION "trend_directory"."get_trends_for_trend_store"("trend_store_id" integer)
    RETURNS SETOF trend_directory.table_trend
AS $$
SELECT table_trend
  FROM trend_directory.table_trend
  LEFT JOIN trend_directory.trend_store_part
  ON table_trend.trend_store_part_id = trend_store_part.id
  WHERE trend_store_part.trend_store_id = $1;
$$ LANGUAGE sql STABLE;


CREATE OR REPLACE FUNCTION "trend_directory"."transfer"("materialization" trend_directory.view_materialization, "timestamp" timestamp with time zone)
    RETURNS integer
AS $$
DECLARE
    columns_part text;
    row_count integer;
    job_id integer;
BEGIN
    SELECT logging.start_job(format('{"view_materialization": "%s", "timestamp": "%s"}', m::text, $2::text)::jsonb) INTO job_id
    FROM trend_directory.materialization m WHERE id = $1.materialization_id;

    SELECT trend_directory.columns_part($1) INTO columns_part;

    EXECUTE format(
        'INSERT INTO trend.%I (entity_id, timestamp, created, job_id, %s) SELECT entity_id, timestamp, now(), %s, %s FROM %s WHERE timestamp = $1',
        (trend_directory.dst_trend_store_part($1)).name,
        columns_part,
        job_id,
        columns_part,
        $1.src_view::name
    ) USING timestamp;

    PERFORM logging.end_job(job_id);

    GET DIAGNOSTICS row_count = ROW_COUNT;

    RETURN row_count;
END;
$$ LANGUAGE plpgsql VOLATILE;

COMMENT ON FUNCTION "trend_directory"."transfer"("materialization" trend_directory.view_materialization, "timestamp" timestamp with time zone) IS 'Transfer all records of the specified timestamp from the view to the target trend store of the materialization.';


CREATE OR REPLACE FUNCTION "trend_directory"."transfer"("materialization" trend_directory.function_materialization, "timestamp" timestamp with time zone)
    RETURNS integer
AS $$
DECLARE
    columns_part text;
    row_count integer;
    job_id integer;
BEGIN
    SELECT logging.start_job(format('{"function_materialization": "%s", "timestamp": "%s"}', m::text, $2::text)::jsonb) INTO job_id
    FROM trend_directory.materialization m WHERE id = $1.materialization_id;

    SELECT trend_directory.columns_part($1) INTO columns_part;

    EXECUTE format(
        'INSERT INTO trend.%I (entity_id, timestamp, created, job_id, %s) SELECT entity_id, timestamp, now(), %s, %s FROM %s($1)',
        (trend_directory.dst_trend_store_part($1)).name,
        columns_part,
        job_id,
        columns_part,
        $1.src_function::regproc
    ) USING timestamp;

    PERFORM logging.end_job(job_id);

    GET DIAGNOSTICS row_count = ROW_COUNT;

    RETURN row_count;
END;
$$ LANGUAGE plpgsql VOLATILE;


CREATE OR REPLACE FUNCTION "trend_directory"."changes_on_trend_update"()
    RETURNS trigger
AS $$
DECLARE
    base_table_name text;
BEGIN
    IF NEW.name <> OLD.name THEN
        FOR base_table_name IN
            SELECT trend_directory.base_table_name(trend_store_part)
            FROM trend_directory.table_trend
            JOIN trend_directory.trend_store_part ON table_trend.trend_store_part_id = trend_store_part.id
            WHERE table_trend.id = NEW.id
        LOOP
            EXECUTE format('ALTER TABLE trend.%I RENAME COLUMN %I TO %I', base_table_name, OLD.name, NEW.name);
        END LOOP;
    END IF;

    RETURN NEW;
END;
$$ LANGUAGE plpgsql VOLATILE;


DROP FUNCTION "attribute_directory"."greatest_data_type"(varchar, varchar);

DROP FUNCTION "attribute_directory"."update_compacted"(integer, timestamp with time zone);

DROP FUNCTION "attribute_directory"."store_compacted"(integer, timestamp with time zone);

DROP FUNCTION "attribute_directory"."mark_compacted"(integer, timestamp with time zone);

DROP FUNCTION "attribute_directory"."mark_compacted"(integer);

ALTER TABLE "attribute_directory"."attribute_store_compacted" DROP COLUMN "compacted";
ALTER TABLE "attribute_directory"."attribute_store_compacted" ADD COLUMN "compacted" integer NOT NULL DEFAULT 0;


CREATE FUNCTION "attribute_directory"."get_attribute_store"("data_source" text, "entity_type" text)
    RETURNS attribute_directory.attribute_store
AS $$
SELECT attribute_store
FROM attribute_directory.attribute_store
LEFT JOIN directory.data_source
  ON data_source_id = data_source.id
LEFT JOIN directory.entity_type
{'fraction_non_alphanumeric': 0.09843626460511289, 'fraction_numerical': 0.006039708336993762, 'mean_word_length': 5.115095353209777, 'pattern_counts': {'":': 8, '<': 13, '<?xml version=': 0, '>': 23, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '944274', 'n_tokens_mistral': 15971, 'n_tokens_neox': 15169, 'n_words': 3458}