YACHT-1142: SLI Quality query finds also tables without backup table …

…in BQ even if datastore has information about backup.

Repairing predicate in SLITableNewerModificationPredicate - it was not using partiton_id from table_reference for partitioned tables which causes errors. Removed unnecessary code. Some fields are now NULLABLE in SLI_QUALITY_HISTORY_SCHEMA (as it is possible that some fields are nulls if there are any backup table).

src/slo/backup_quality/predicate/sli_table_newer_modification_predicate.py

@@ -14,9 +14,9 @@ def __init__(self, big_query):
     def is_modified_since_last_census_snapshot(self, sli_table_entry):
         table_reference = QualityQuerySpecification.to_table_reference(sli_table_entry)
         table = self.big_query.get_table(
-            project_id=table_reference.project_id,
-            dataset_id=table_reference.dataset_id,
-            table_id=table_reference.table_id)
+            project_id=table_reference.get_project_id(),
+            dataset_id=table_reference.get_dataset_id(),
+            table_id=table_reference.get_table_id_with_partition_id())
         table_metadata = BigQueryTableMetadata(table)
 
         is_table_modified = table_metadata.get_last_modified_datetime() > datetime.datetime.utcfromtimestamp(sli_table_entry["lastModifiedTime"])

src/slo/backup_quality/quality_violation_sli_service.py

@@ -1,7 +1,6 @@
 import logging
 
 from src.commons.big_query.big_query import BigQuery
-from src.slo.backup_quality.predicate.sli_bq_backup_exists_predicate import SLIBQBackupExistsPredicate
 from src.slo.predicate.sli_table_exists_predicate import \
     SLITableExistsPredicate
 from src.slo.backup_quality.quality_query_specification import \
@@ -24,7 +23,6 @@ def __init__(self):
         )
         self.table_newer_modification_predicate = SLITableNewerModificationPredicate(big_query)
         self.table_existence_predicate = SLITableExistsPredicate(big_query, QualityQuerySpecification)
-        self.bq_backup_existence_predicate = SLIBQBackupExistsPredicate()
 
     def check_and_stream_violation(self, json_table):
         if self.__should_stay_as_sli_violation(json_table):
@@ -33,8 +31,6 @@ def check_and_stream_violation(self, json_table):
 
     def __should_stay_as_sli_violation(self, table):
         try:
-            if not self.bq_backup_existence_predicate.exists(table):
-                return True
             if not self.table_existence_predicate.exists(table):
                 return False
             return not self.table_newer_modification_predicate.is_modified_since_last_census_snapshot(table)

terraform/SLI_backup_quality_filtered_table_schema.json

@@ -50,8 +50,8 @@
   {
     "name": "backupLastModifiedTime",
     "type": "TIMESTAMP",
-    "mode": "REQUIRED",
-    "description": "date of last modification (from GCP Census) that will be included in the Backup (copyJob start time - as it is atomic operation and every change before that point is included in copy)"
+    "mode": "NULLABLE",
+    "description": "date of last modification (from GCP Census) of backup table"
   },
   {
     "name": "backupEntityLastModifiedTime",
@@ -68,7 +68,7 @@
   {
     "name": "backupNumBytes",
     "type": "INTEGER",
-    "mode": "REQUIRED",
+    "mode": "NULLABLE",
     "description": "Number of bytes of table backup (from GCP Census)"
   },
   {
@@ -86,7 +86,7 @@
   {
     "name": "backupNumRows",
     "type": "INTEGER",
-    "mode": "REQUIRED",
+    "mode": "NULLABLE",
     "description": "Number of rows of table backup (from GCP Census)"
   }
 ]

tests/slo/backup_quality/test_quality_violation_sli_service.py

@@ -2,7 +2,6 @@
 
 from mock import patch
 
-from src.slo.backup_quality.predicate.sli_bq_backup_exists_predicate import SLIBQBackupExistsPredicate
 from src.slo.backup_quality.predicate.sli_table_newer_modification_predicate import \
     SLITableNewerModificationPredicate
 from src.slo.backup_quality.quality_violation_sli_service import QualityViolationSliService
@@ -24,11 +23,9 @@ def tearDown(self):
     @patch.object(SLITableNewerModificationPredicate, 'is_modified_since_last_census_snapshot')
     @patch.object(SLIResultsStreamer, 'stream')
     @patch.object(SLITableExistsPredicate, 'exists')
-    @patch.object(SLIBQBackupExistsPredicate, 'exists')
     def test_check_and_stream_violation_that_is_modified_since_last_census_snapshot_should_be_filtered_out(
-            self, backup_exists, table_exists, stream, is_modified_since_last_census_snapshot):
+            self, table_exists, stream, is_modified_since_last_census_snapshot):
         # given
-        backup_exists.return_value = True
         table_exists.return_value = True
         is_modified_since_last_census_snapshot.return_value = True
         payload = {"projectId": "p1", "datasetId": "d1",
@@ -43,11 +40,9 @@ def test_check_and_stream_violation_that_is_modified_since_last_census_snapshot_
     @patch.object(SLITableNewerModificationPredicate, 'is_modified_since_last_census_snapshot')
     @patch.object(SLIResultsStreamer, 'stream')
     @patch.object(SLITableExistsPredicate, 'exists')
-    @patch.object(SLIBQBackupExistsPredicate, 'exists')
     def test_check_and_stream_violation_that_is_not_modified_since_last_census_snapshot_should_not_be_filtered_out(
-            self, backup_exists, table_exists, stream, is_modified_since_last_census_snapshot):
+            self, table_exists, stream, is_modified_since_last_census_snapshot):
         # given
-        backup_exists.return_value = True
         table_exists.return_value = True
         is_modified_since_last_census_snapshot.return_value = False
         payload = {"projectId": "p1", "datasetId": "d1",
@@ -60,28 +55,11 @@ def test_check_and_stream_violation_that_is_not_modified_since_last_census_snaps
         stream.assert_called_with([{"projectId": "p1", "datasetId": "d1",
                                     "tableId": "t1", "partitionId": "part1"}])
 
-    @patch.object(SLIResultsStreamer, 'stream')
-    @patch.object(SLIBQBackupExistsPredicate, 'exists')
-    def test_check_and_stream_violation_that_doesnt_have_backup_table_in_bq(self, backup_exists,  stream):
-        # given
-        backup_exists.return_value = False
-        payload = {"projectId": "p1", "datasetId": "d1",
-                   "tableId": "t1", "partitionId": "part1"}
-
-        # when
-        QualityViolationSliService().check_and_stream_violation(payload)
-
-        # then
-        stream.assert_called_with([{"projectId": "p1", "datasetId": "d1",
-                                    "tableId": "t1", "partitionId": "part1"}])
-
     @patch.object(SLIResultsStreamer, 'stream')
     @patch.object(SLITableExistsPredicate, 'exists')
-    @patch.object(SLIBQBackupExistsPredicate, 'exists')
     def test_check_and_stream_violation_table_that_not_exists_should_be_filtered_out(
-            self, backup_exists, table_exists, stream):
+            self,  table_exists, stream):
         # given
-        backup_exists.return_value = True
         table_exists.return_value = False
         payload = {"projectId": "p1", "datasetId": "d1",
                    "tableId": "t1", "partitionId": "part1"}
@@ -95,11 +73,9 @@ def test_check_and_stream_violation_table_that_not_exists_should_be_filtered_out
     @patch.object(SLIResultsStreamer, 'stream')
     @patch.object(SLITableExistsPredicate, 'exists')
     @patch.object(SLITableNewerModificationPredicate, 'is_modified_since_last_census_snapshot')
-    @patch.object(SLIBQBackupExistsPredicate, 'exists')
     def test_check_and_stream_violation_table_that_caused_exception_in_modification_predicate_should_not_be_filtered_out(
-            self, backup_exists, is_modified_since_last_census_snapshot, table_exists, stream):
+            self, is_modified_since_last_census_snapshot, table_exists, stream):
         # given
-        backup_exists.return_value = True
         table_exists.return_value = True
         is_modified_since_last_census_snapshot.side_effect = Exception("An error")
         payload = {"projectId": "p1", "datasetId": "d1",