clean up snapshots during data refresh based on snapshotValidityDays (#38)

AndrewNepogoda · web-flow · commit 8a9aeb660791 · 2023-10-17T19:56:43.000+02:00
* clean up snapshots during data refresh based on `snapshotValidityDays`

* added validation for loadNumberOfDays, snapshotValidity
diff --git a/cli-e2e-test/config/model/scenario7.json b/cli-e2e-test/config/model/scenario7.json
@@ -0,0 +1,24 @@
+{
+  "workflow": [
+    {
+      "type": "ConfigureSources",
+      "name": "ConfigureSources",
+      "configFiles": [
+        "config/config1.rel"
+      ],
+      "defaultContainer": "input",
+      "sources": [
+        {
+          "relation": "device_seen_snapshot",
+          "isChunkPartitioned": true,
+          "isDatePartitioned": true,
+          "relativePath": "device_seen_snapshot",
+          "inputFormat": "csv",
+          "loadsNumberOfDays": 1,
+          "offsetByNumberOfDays": 0,
+          "snapshotValidityDays": 1
+        }
+      ]
+    }
+  ]
+}
diff --git a/cli-e2e-test/config/model/scenario8.json b/cli-e2e-test/config/model/scenario8.json
@@ -0,0 +1,24 @@
+{
+  "workflow": [
+    {
+      "type": "ConfigureSources",
+      "name": "ConfigureSources",
+      "configFiles": [
+        "config/config1.rel"
+      ],
+      "defaultContainer": "input",
+      "sources": [
+        {
+          "relation": "device_seen_snapshot",
+          "isChunkPartitioned": true,
+          "isDatePartitioned": true,
+          "relativePath": "device_seen_snapshot",
+          "inputFormat": "csv",
+          "loadsNumberOfDays": 1,
+          "offsetByNumberOfDays": 0,
+          "snapshotValidityDays": 2
+        }
+      ]
+    }
+  ]
+}
diff --git a/cli-e2e-test/data/device_seen_snapshot/data_dt=20220103/part-1-3u1ec0b0-ebfd-a773-71d7-f71f42a2f066.csv b/cli-e2e-test/data/device_seen_snapshot/data_dt=20220103/part-1-3u1ec0b0-ebfd-a773-71d7-f71f42a2f066.csv
@@ -0,0 +1,3 @@
+device,last_seen
+IPhone1,2021-12-31
+IPhone2,2021-12-30
diff --git a/cli-e2e-test/data/device_seen_snapshot/data_dt=20220104/part-1-3p1ec0b0-ebfd-a773-71d7-f71f42a2f066.csv b/cli-e2e-test/data/device_seen_snapshot/data_dt=20220104/part-1-3p1ec0b0-ebfd-a773-71d7-f71f42a2f066.csv
@@ -0,0 +1,3 @@
+device,last_seen
+IPhone6,2021-11-16
+IPhone7,2021-11-17
diff --git a/cli-e2e-test/test_e2e.py b/cli-e2e-test/test_e2e.py
@@ -200,6 +200,63 @@ def test_scenario6_model_two_partitions_overriden_by_one(self):
         self.assertEqual(rsp_json, [{'partition': 1, 'relation': 'product_data'},
                                     {'partition': 2, 'relation': 'product_data'}])
 
+    def test_scenario7_model_1_day_snapshot_2_day_declared_1_day_out_of_range(self):
+        # when
+        test_args = ["--batch-config", "./config/model/scenario7.json"]
+        rsp = call(self.cmd_with_common_arguments + test_args + ["--end-date", "20220102", "--drop-db"])
+        # then
+        self.assertNotEqual(rsp, 1)
+        # and when
+        rsp = call(self.cmd_with_common_arguments + test_args + ["--end-date", "20220104"])
+        # then
+        self.assertNotEqual(rsp, 1)
+        rai_config = self.resource_manager.get_rai_config()
+        rsp_json = workflow.rai.execute_relation_json(self.logger, rai_config, RESOURCES_TO_DELETE_REL)
+        self.assertEqual(rsp_json, [{'partition': 2022010200001, 'relation': 'device_seen_snapshot'}])
+
+    def test_scenario7_model_1_day_snapshot_1_day_declared_1_day_out_of_range(self):
+        # when
+        test_args = ["--batch-config", "./config/model/scenario7.json"]
+        rsp = call(self.cmd_with_common_arguments + test_args + ["--start-date", "20220103", "--end-date", "20220104",
+                                                                 "--drop-db"])
+        # then
+        self.assertNotEqual(rsp, 1)
+        # and when
+        rsp = call(self.cmd_with_common_arguments + test_args + ["--end-date", "20220105"])
+        # then
+        self.assertNotEqual(rsp, 1)
+        rai_config = self.resource_manager.get_rai_config()
+        rsp_json = workflow.rai.execute_relation_json(self.logger, rai_config, RESOURCES_TO_DELETE_REL)
+        self.assertEqual(rsp_json, {})
+
+    def test_scenario7_model_1_day_snapshot_1_day_declared_0_days_out_of_range(self):
+        # when
+        test_args = ["--batch-config", "./config/model/scenario7.json"]
+        rsp = call(self.cmd_with_common_arguments + test_args + ["--end-date", "20220105", "--drop-db"])
+        # then
+        self.assertNotEqual(rsp, 1)
+        # and when
+        rsp = call(self.cmd_with_common_arguments + test_args + ["--end-date", "20220105"])
+        # then
+        self.assertNotEqual(rsp, 1)
+        rai_config = self.resource_manager.get_rai_config()
+        rsp_json = workflow.rai.execute_relation_json(self.logger, rai_config, RESOURCES_TO_DELETE_REL)
+        self.assertEqual(rsp_json, {})
+
+    def test_scenario8_model_2_day_snapshot_1_day_declared_1_days_out_of_range(self):
+        # when
+        test_args = ["--batch-config", "./config/model/scenario8.json"]
+        rsp = call(self.cmd_with_common_arguments + test_args + ["--end-date", "20220103", "--drop-db"])
+        # then
+        self.assertNotEqual(rsp, 1)
+        # and when
+        rsp = call(self.cmd_with_common_arguments + test_args + ["--start-date", "20220104", "--end-date", "20220105"])
+        # then
+        self.assertNotEqual(rsp, 1)
+        rai_config = self.resource_manager.get_rai_config()
+        rsp_json = workflow.rai.execute_relation_json(self.logger, rai_config, RESOURCES_TO_DELETE_REL)
+        self.assertEqual(rsp_json, [{'partition': 2022010300001, 'relation': 'device_seen_snapshot'}])
+
     @classmethod
     def setUpClass(cls) -> None:
         # Make sure output folder is empty since the folder share across repository. Remove README.md, other files left.
diff --git a/cli/README.md b/cli/README.md
@@ -6,7 +6,7 @@ This Command-Line Interface (CLI) is designed to provide an easy and interactive
 1. Create a batch configuration (ex. `poc.json`) file using the syntax and structure outlined in the [RAI Workflow Framework README](../workflow/README.md).
 2. Add `rai-workflow-manager` as dependency to your `requirements.txt` file:
 ```txt
-rai-workflow-manager==0.0.22
+rai-workflow-manager==0.0.23
 ```
 3. Build the project:
 ```bash
diff --git a/rel/source_configs/config.rel b/rel/source_configs/config.rel
@@ -419,6 +419,46 @@ def missing_resources_json(:[], i, :dates, :[], j, :resources, :[], k, :part_ind
     from s, r, d
 }
 
+/*
+ * Declared date partitioned sources
+ */
+module declared_date_partitioned_source
+
+    def path_on_date(rel, d, p) {
+        source_declares_resource(rel, _, p) and
+        r = ^RelationName[rel] . rel_name:identifies . relation:identifies . source:declares and
+        resource:id(r, ^URI[p]) and
+        uri:identifies(u, r) and
+        d = uri:parse[u, "date"]
+        from r, u
+    }
+
+    def date(s, d) = path_on_date(s, d, _)
+
+    def id(s) = date(s, _)
+
+    def index = enumerate[id]
+    def date_idx = enumerate[date]
+    def path_idx = enumerate[path_on_date]
+
+    def json(:[], idx, :source, s) {
+        index(idx, s)
+    }
+
+    def json(:[], idx, :dates, :[], d_idx, :date, d) {
+        index(idx, s) and
+        date_idx(d_idx, s, d)
+        from s
+    }
+
+    def json(:[], idx, :dates, :[], d_idx, :paths, :[], p_idx, p) {
+        index(idx, s) and
+        date_idx(d_idx, s, d) and
+        path_idx(p_idx, s, d, p)
+        from s, d
+    }
+end
+
 /**
  */
 
diff --git a/rel/source_configs/data_reload.rel b/rel/source_configs/data_reload.rel
@@ -3,13 +3,19 @@ bound force_reimport_not_chunk_partitioned
 bound resources_data_to_delete
 bound declared_sources_to_delete = String, String
 bound new_source_config_csv = RelName, FilePos, String
+bound expired_source_config_csv = RelName, FilePos, String
 
 def new_source_config[:syntax, :header_row] = -1
 def new_source_config[:syntax, :header] = (1, :Relation); (2, :Path); (3, :ChunkPartitioned)
 def new_source_config[:schema, :Relation] = "string"
 def new_source_config[:schema, :Path] = "string"
 def new_source_config[:schema, :ChunkPartitioned] = "string"
 
+def expired_source_config[:syntax, :header_row] = -1
+def expired_source_config[:syntax, :header] = (1, :Relation); (2, :Path)
+def expired_source_config[:schema, :Relation] = "string"
+def expired_source_config[:schema, :Path] = "string"
+
 def chunk_partitioned_sources(rel, path, p_idx) {
     new_source_config_csv(:Path, i, path) and
     new_source_config_csv(:Relation, i, rel) and
@@ -25,6 +31,12 @@ def simple_sources(rel, path) {
     from i
 }
 
+def expired_sources(rel, path) {
+    expired_source_config_csv(:Path, i, path) and
+    expired_source_config_csv(:Relation, i, rel)
+    from i
+}
+
 /*
  * All simple sources are affected if they match with declared sources.
  */
@@ -57,25 +69,30 @@ def potentially_affected_sources(rel, o_path, p_idx) {
     from n_path, res
 }
 /*
- * Identify sources for replacement.
+ * Identify sources for invalidation.
  */
-def part_resource_to_replace(rel, p_idx, path) {
+def part_resource_to_invalidate(rel, p_idx, path) {
     not force_reimport and
     potentially_affected_sources(rel, path, p_idx) and
     not chunk_partitioned_sources(rel, path, p_idx)
 }
 
-def part_resource_to_replace(rel, p_idx, path) {
+def part_resource_to_invalidate(rel, p_idx, path) {
     force_reimport and
     potentially_affected_sources(rel, path, p_idx)
 }
+// We support invalidation only for partitioned expired sources
+def part_resource_to_invalidate(rel, p_idx, path) {
+    expired_sources(rel, path) and
+    part_resource:parse_part_index[path](p_idx)
+}
 
-def resource_to_replace(rel, o_path) {
+def resource_to_invalidate(rel, o_path) {
     force_reimport_not_chunk_partitioned and
     simple_sources(rel, o_path)
 }
 
-def resource_to_replace(rel, o_path) {
+def resource_to_invalidate(rel, o_path) {
     force_reimport and
     simple_sources(rel, o_path)
 }
@@ -84,16 +101,16 @@ def resource_to_replace(rel, o_path) {
  * Save resources that we are marked for data reimport.
  */
 // PRODUCT_LIMITATION: Message: Argument for specializing is more complex than the current staging implementation supports.
-def reverse_part_resource_to_replace(path, p_idx, rel) = part_resource_to_replace(rel, p_idx, path)
+def reverse_part_resource_to_invalidate(path, p_idx, rel) = part_resource_to_invalidate(rel, p_idx, path)
 // resource partitions to delete
 def resources_to_delete(rel, val) {
-    rel = #(reverse_part_resource_to_replace[_, p_idx]) and
+    rel = #(reverse_part_resource_to_invalidate[_, p_idx]) and
     p_idx = ^PartIndex[val]
     from p_idx
 }
 // resources to delete
 def resources_to_delete(rel) {
-    rel = #(transpose[resource_to_replace][_])
+    rel = #(transpose[resource_to_invalidate][_])
 }
 
 def resources_data_to_delete_idx = enumerate[resources_data_to_delete]
diff --git a/test/test_cfg_src_step.py b/test/test_cfg_src_step.py
@@ -397,13 +397,103 @@ def test_inflate_sources_snapshot_1day_multiple_paths(self):
         ]
         self.assertEqual(expected_paths, test_src.paths)
 
+    def test_calculate_expired_sources_1_day_snapshot_1_day_declared_1_day_out_of_range(self):
+        # setup
+        test_src = _create_test_source(
+            snapshot_validity_days=1
+        )
+        declared_sources = {
+            "test": {
+                "source": "test",
+                "dates": [
+                    {
+                        "paths": [
+                            "/test/data_dt=20220104/part-1.csv"
+                        ],
+                        "date": "20220104"
+                    }
+                ]
+            }
+        }
+        paths_builder = Mock()
+        workflow_step = _create_cfg_sources_step([test_src], {"default": paths_builder}, None, "20220106")
+        # when
+        expired_source = workflow_step._calculate_expired_sources(self.logger, declared_sources)
+        # then
+        expected_sources = [("test", "/test/data_dt=20220104/part-1.csv")]
+        self.assertEqual(expected_sources, expired_source)
+
+    def test_calculate_expired_sources_1_day_snapshot_2_day_declared_1_day_out_of_range(self):
+        # setup
+        test_src = _create_test_source(
+            snapshot_validity_days=1
+        )
+        declared_sources = {
+            "test": {
+                "source": "test",
+                "dates": [
+                    {
+                        "paths": [
+                            "/test/data_dt=20220104/part-1.csv"
+                        ],
+                        "date": "20220104"
+                    },
+                    {
+                        "paths": [
+                            "/test/data_dt=20220105/part-1.csv"
+                        ],
+                        "date": "20220105"
+                    }
+                ]
+            }
+        }
+        paths_builder = Mock()
+        workflow_step = _create_cfg_sources_step([test_src], {"default": paths_builder}, None, "20220106")
+        # when
+        expired_source = workflow_step._calculate_expired_sources(self.logger, declared_sources)
+        # then
+        expected_sources = [("test", "/test/data_dt=20220104/part-1.csv")]
+        self.assertEqual(expected_sources, expired_source)
+
+    def test_calculate_expired_sources_1_day_snapshot_2_day_declared_0_day_out_of_range(self):
+        # setup
+        test_src = _create_test_source(
+            snapshot_validity_days=1
+        )
+        declared_sources = {
+            "test": {
+                "source": "test",
+                "dates": [
+                    {
+                        "paths": [
+                            "/test/data_dt=20220104/part-1.csv"
+                        ],
+                        "date": "20220104"
+                    },
+                    {
+                        "paths": [
+                            "/test/data_dt=20220105/part-1.csv"
+                        ],
+                        "date": "20220105"
+                    }
+                ]
+            }
+        }
+        paths_builder = Mock()
+        workflow_step = _create_cfg_sources_step([test_src], {"default": paths_builder}, None, "20220105")
+        # when
+        expired_source = workflow_step._calculate_expired_sources(self.logger, declared_sources)
+        # then
+        expected_sources = []
+        self.assertEqual(expected_sources, expired_source)
+
 
 def _create_test_source(is_chunk_partitioned: bool = True, is_date_partitioned: bool = True,
-                        loads_number_of_days: int = 1, offset_by_number_of_days: int = 0,
+                        loads_number_of_days: int = 1, offset_by_number_of_days: int = 0, relation="test",
                         snapshot_validity_days=None) -> Source:
     return Source(
         container=Container("default", ContainerType.LOCAL, {}),
-        relation="test",
+        relation=relation,
         relative_path="test",
         input_format="test",
         extensions=["test"],
diff --git a/workflow/__init__.py b/workflow/__init__.py
@@ -12,5 +12,5 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-__version_info__ = (0, 0, 22)
+__version_info__ = (0, 0, 23)
 __version__ = ".".join(map(str, __version_info__))
diff --git a/workflow/constants.py b/workflow/constants.py
@@ -29,6 +29,7 @@
 RESOURCES_TO_DELETE_REL = "resources_data_to_delete_json"
 WORKFLOW_JSON_REL = "workflow_json"
 BATCH_CONFIG_REL = "batch:config"
+DECLARED_DATE_PARTITIONED_SOURCE_REL = "declared_date_partitioned_source:json"
 
 FILE_LOAD_RELATION = {
     "CSV": "load_csv",
diff --git a/workflow/executor.py b/workflow/executor.py
diff --git a/workflow/query.py b/workflow/query.py

Original file line number	Diff line number	Diff line change
`@@ -0,0 +1,3 @@`
	`1`	`+device,last_seen`
	`2`	`+IPhone1,2021-12-31`
	`3`	`+IPhone2,2021-12-30`
Original file line number	Diff line number	Diff line change
`@@ -0,0 +1,3 @@`
	`1`	`+device,last_seen`
	`2`	`+IPhone6,2021-11-16`
	`3`	`+IPhone7,2021-11-17`