273 patch partitioningspartidparents

.github/CODEOWNERS

@@ -1 +1 @@
-* @benedeki @lsulak @TebaleloS @Zejnilovic @dk1844 @salamonpavel
+* @benedeki @lsulak @TebaleloS @Zejnilovic @dk1844 @salamonpavel @ABLL526

database/README.md

@@ -32,3 +32,10 @@ to remove them or
 sbt flywayBaseline 
 ```
 to set the current state as the baseline.
+
+```zsh
+docker kill atum_db
+docker rm atum_db
+docker run --name=atum_db -e POSTGRES_PASSWORD=postgres -e POSTGRES_DB=atum_db -p 5432:5432 -d postgres:16
+sbt flywayMigrate
+```

database/src/main/postgres/runs/V0.3.0.1__get_ancestors.sql

@@ -0,0 +1,121 @@
+/*
+ * Copyright 2021 ABSA Group Limited
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+CREATE OR REPLACE FUNCTION runs.get_ancestors(
+    IN i_id_partitioning    BIGINT,
+    IN i_limit              INT DEFAULT 5,
+    IN i_offset             BIGINT DEFAULT 0,
+    OUT status              INTEGER,
+    OUT status_text         TEXT,
+    OUT ancestorid         BIGINT,
+    OUT partitioning        JSONB,
+    OUT author              TEXT,
+    OUT has_more            BOOLEAN
+)  RETURNS SETOF record AS
+$$
+    -------------------------------------------------------------------------------
+--
+-- Function: runs.get_ancestors(3)
+--      Returns Ancestors' partition ID for the given id
+--
+-- Parameters:
+--      i_id_partitioning   - id that we asking the Ancestors for
+--      i_limit             - (optional) maximum number of partitionings to return, default is 5
+--      i_offset            - (optional) offset to use for pagination, default is 0
+--
+-- Returns:
+--      status              - Status code
+--      status_text         - Status message
+--      ancestorid         - ID of Ancestor partition
+--      partitioning        - partitioning data of ancestor
+--      author              - author of the Ancestor partitioning
+--      has_more            - Flag indicating if there are more partitionings available
+
+-- Status codes:
+--      11 - OK
+--      41 - Partitioning not found
+--      42 - Ancestor Partitioning not found
+--
+-------------------------------------------------------------------------------
+DECLARE
+    partitionCreateAt TIMESTAMP;
+    _has_more BOOLEAN;
+
+BEGIN
+    -- Check if the partitioning exists
+    PERFORM 1 FROM runs.partitionings WHERE id_partitioning = i_id_partitioning;
+    IF NOT FOUND THEN
+        status := 41;
+        status_text := 'Partitioning not found';
+        RETURN NEXT;
+        RETURN;
+    END IF;
+
+    -- Get the creation timestamp of the partitioning
+    SELECT created_at
+    FROM runs.partitionings
+    WHERE id_partitioning = i_id_partitioning
+    INTO partitionCreateAt;
+
+    -- Check if there are more partitionings than the limit
+    SELECT count(*) > i_limit
+    FROM flows.partitioning_to_flow PTF
+    WHERE PTF.fk_flow IN (
+        SELECT fk_flow
+        FROM flows.partitioning_to_flow
+        WHERE fk_partitioning = i_id_partitioning
+    )
+    LIMIT i_limit + 1 OFFSET i_offset
+    INTO _has_more;
+
+    -- Return the ancestors
+    RETURN QUERY
+        SELECT
+            11 AS status,
+            'OK' AS status_text,
+            P.id_partitioning AS ancestorid,
+            P.partitioning AS partitioning,
+            P.created_by AS author,
+            _has_more AS has_more
+        FROM
+            runs.partitionings P
+                INNER JOIN flows.partitioning_to_flow PF ON PF.fk_partitioning = P.id_partitioning
+                INNER JOIN flows.partitioning_to_flow PF2 ON PF2.fk_flow = PF.fk_flow
+        WHERE
+            PF2.fk_partitioning = i_id_partitioning
+            AND
+            P.created_at < partitionCreateAt
+        GROUP BY P.id_partitioning
+        ORDER BY P.id_partitioning, P.created_at DESC
+        LIMIT i_limit
+        OFFSET i_offset;
+
+    IF FOUND THEN
+        status := 11;
+        status_text := 'OK';
+    ELSE
+        status := 42;
+        status_text := 'Ancestor Partitioning not found';
+    END IF;
+    RETURN NEXT;
+    RETURN;
+
+END;
+$$
+    LANGUAGE plpgsql VOLATILE SECURITY DEFINER;
+
+ALTER FUNCTION runs.get_ancestors(BIGINT, INT, BIGINT) OWNER TO atum_owner;
+GRANT EXECUTE ON FUNCTION runs.get_ancestors(BIGINT, INT, BIGINT) TO atum_user;

database/src/main/postgres/runs/V0.3.0.2__add_parent.sql

@@ -0,0 +1,112 @@
+/*
+ * Copyright 2021 ABSA Group Limited
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+CREATE OR REPLACE FUNCTION runs.add_parent(
+    IN i_id_partitioning          BIGINT,
+    IN i_id_parent_partitioning   BIGINT,
+    IN i_by_user                  TEXT,
+    IN i_copy_measurements        BOOLEAN DEFAULT true,
+    IN i_copy_additional_data     BOOLEAN DEFAULT true,
+    OUT status                    INTEGER,
+    OUT status_text               TEXT
+) RETURNS record AS
+$$
+    -------------------------------------------------------------------------------
+--
+-- Function: runs.add_parent(5)
+--      Add the Parent Partition for a given partition ID and copies measurements and additional data from parent.
+--
+-- Parameters:
+--      i_id_partitioning                - id of the partition to be changed
+--      i_id_parent_partitioning         - id of the new parent of the partition,
+--      i_by_user                        - user behind the change
+--      i_copy_measurements              - copies measurements
+--      i_copy_additional_data           - copies additional data
+--
+-- Returns:
+--      status              - Status code
+--      status_text         - Status message
+--
+-- Status codes:
+--      11 - OK
+--      41 - Child Partitioning not found
+--      42 - Parent Partitioning not found
+--
+-------------------------------------------------------------------------------
+DECLARE
+    Additional_Data HSTORE;
+
+BEGIN
+
+    PERFORM 1 FROM runs.partitionings WHERE id_partitioning = i_id_partitioning;
+    IF NOT FOUND THEN
+        status := 41;
+        status_text := 'Child Partitioning not found';
+        RETURN;
+    END IF;
+
+    PERFORM 1 FROM runs.partitionings WHERE id_partitioning = i_id_parent_partitioning;
+    IF NOT FOUND THEN
+        status := 42;
+        status_text := 'Parent Partitioning not found';
+        RETURN;
+    END IF;
+
+--    SELECT F.id_flow as mainFlow
+--    FROM runs.get_partitioning_main_flow(i_partitioning_id) AS F
+--    INTO mainFlow;
+
+--    flow_id := array(
+--            SELECT fk_flow AS flow_id
+--            FROM flows.partitioning_to_flow
+--            WHERE fk_partitioning = i_partitioning_id
+--            AND fk_flow != mainFlow
+--    );
+
+--    FOREACH var IN ARRAY flow_id LOOP
+--        DELETE FROM flows.partitioning_to_flow AS PTF
+--        WHERE PTF.fk_partitioning = i_partitioning_id
+--        AND PTF.fk_flow = var;
+--    END LOOP;
+
+    IF i_copy_additional_data THEN
+        SELECT
+            hstore(array_agg(PAD.ad_name), array_agg(PAD.ad_value)) AS Additional_Data
+        FROM
+            runs.get_partitioning_additional_data(i_id_parent_partitioning) AS PAD
+        INTO
+            Additional_Data;
+        PERFORM 1 FROM runs.create_or_update_additional_data(i_id_partitioning, Additional_Data, i_by_user);
+    END IF;
+
+    IF i_copy_measurements THEN
+        INSERT INTO runs.measure_definitions (fk_partitioning, measure_name, measured_columns, created_by)
+        SELECT i_id_partitioning, PMI.measure_name, PMI.measured_columns, i_by_user
+            FROM
+                runs.get_partitioning_measures_by_id(i_id_parent_partitioning) AS PMI;
+    END IF;
+
+    PERFORM 1 FROM flows._add_to_parent_flows(i_id_parent_partitioning, i_id_partitioning, i_by_user);
+    status := 11;
+    status_text := 'Parent Updated';
+    RETURN;
+
+END;
+$$
+    LANGUAGE plpgsql VOLATILE SECURITY DEFINER;
+
+ALTER FUNCTION runs.add_parent(BIGINT, BIGINT, TEXT, BOOLEAN, BOOLEAN) OWNER TO atum_owner;
+GRANT EXECUTE ON FUNCTION runs.add_parent(BIGINT, BIGINT, TEXT, BOOLEAN, BOOLEAN) TO atum_user;