[OSC-1366] Skip on no changes (#40)

* skip if there's no changes

* check if data changed to avoid reloading

* ensure sync versions are valid before skipping

Author: Sean Budd

rdl/DataLoadManager.py

@@ -119,6 +119,13 @@ def start_single_import(self, model_file, requested_full_refresh, model_number,
         self.data_load_tracker_repository.create_execution_model(data_load_tracker)
         destination_table_manager.create_schema(model_config['target_schema'])
 
+        skip_incremental_reason, skip_incremental = DataLoadManager.check_skip_incremental(change_tracking_info)
+        if not full_refresh and skip_incremental:
+            self.logger.info(f"Skipping incremental refresh for reason '{skip_incremental_reason}'")
+            data_load_tracker.data_load_successful()
+            self.data_load_tracker_repository.save_execution_model(data_load_tracker)
+            return
+
         self.logger.debug(f"Recreating the staging table {model_config['target_schema']}."
                           f"{model_config['stage_table']}")
         destination_table_manager.create_table(model_config['target_schema'],
@@ -168,6 +175,15 @@ def start_single_import(self, model_file, requested_full_refresh, model_number,
                          f" COMPLETED {model_name}")
         self.data_load_tracker_repository.save_execution_model(data_load_tracker)
 
+    @staticmethod
+    def check_skip_incremental(change_tracking_info):
+        if change_tracking_info.last_sync_version and change_tracking_info.sync_version and (
+                change_tracking_info.last_sync_version == change_tracking_info.sync_version):
+            return Constants.IncrementalSkipReason.SYNC_VERSIONS_ARE_EQUAL, True
+        if not change_tracking_info.data_changed_since_last_sync:
+            return Constants.IncrementalSkipReason.NO_DATA_CHANGED, True
+        return Constants.IncrementalSkipReason.NOT_APPLICABLE, False
+
     @staticmethod
     def is_full_refresh(*,
                         user_requested,

rdl/data_sources/ChangeTrackingInfo.py

@@ -1,5 +1,6 @@
 class ChangeTrackingInfo:
-    def __init__(self, last_sync_version, sync_version, force_full_load):
+    def __init__(self, last_sync_version, sync_version, force_full_load, data_changed_since_last_sync):
         self.last_sync_version = last_sync_version
         self.sync_version = sync_version
         self.force_full_load = force_full_load
+        self.data_changed_since_last_sync = data_changed_since_last_sync

rdl/data_sources/CsvDataSource.py

@@ -76,4 +76,4 @@ def get_next_data_frame(
         return data_frame
 
     def init_change_tracking(self, table_config, last_sync_version):
-        return ChangeTrackingInfo(0, 0, False)
+        return ChangeTrackingInfo(0, 0, False, True)

rdl/data_sources/MsSqlDataSource.py

@@ -173,6 +173,8 @@ def init_change_tracking(self, table_config, last_known_sync_version):
         # c) whether a full refresh is needed - this is a derivative of the validity of the last known sync version
         #    because if the last known sync version is no longer valid, then our target data is in-an-invalid-state /
         #    out-of-sync and a full refresh must be forced to sync the data.
+        # d) whether any data has changed for the table synce the last sync - this is a derivative of the results of
+        #    the CHANGETABLE() call, which we perform later and join to the table to load in the changed rows.
         #
         # the following help us determining the above:
         # a) sync_version: the current version of change tracking at source database.
@@ -183,6 +185,8 @@ def init_change_tracking(self, table_config, last_known_sync_version):
         # c) min_valid_version: the minimum version that is valid for use in obtaining change tracking information from
         #                       the specified table.
         #                       it's value IS sourced from CHANGE_TRACKING_MIN_VALID_VERSION(OBJECT_ID(..)).
+        # d) data_changed_since_last_sync: whether or not data has changed in the table since last_known_sync_version
+        #                       it's value is sourced from CHANGETABLE(table, last_known_sync_version).
 
         get_change_tracking_info_sql = f"" \
             f"DECLARE @sync_version                     BIGINT  = CHANGE_TRACKING_CURRENT_VERSION(); \n" \
@@ -193,6 +197,7 @@ def init_change_tracking(self, table_config, last_known_sync_version):
             f" CASE WHEN @last_known_sync_version >= @min_valid_version THEN 1 ELSE 0 END; \n" \
             f"DECLARE @last_sync_version                BIGINT; \n" \
             f"DECLARE @force_full_load                  BIT; \n" \
+            f"DECLARE @data_changed_since_last_sync     BIT; \n" \
             f" \n" \
             f"IF @last_known_sync_version_is_valid = 1 \n" \
             f"BEGIN \n" \
@@ -205,17 +210,32 @@ def init_change_tracking(self, table_config, last_known_sync_version):
             f"    SET @last_sync_version = 0; \n" \
             f"END \n" \
             f" \n" \
-            f"SELECT @sync_version AS sync_version \n" \
-            f", @last_sync_version AS last_sync_version \n" \
-            f", @force_full_load   AS force_full_load; \n"
+            f"IF EXISTS ( " \
+            f"  SELECT 1" \
+            f"  FROM CHANGETABLE(CHANGES " \
+            f"      {table_config['schema']}.{table_config['name']}, {last_known_sync_version} ) " \
+            f"  as c )" \
+            f"BEGIN \n" \
+            f"    SET @data_changed_since_last_sync = 1; \n" \
+            f"END \n" \
+            f"ELSE \n" \
+            f"BEGIN \n" \
+            f"    SET @data_changed_since_last_sync = 0; \n" \
+            f"END \n" \
+            f" \n" \
+            f"SELECT @sync_version              AS sync_version \n" \
+            f", @last_sync_version              AS last_sync_version \n" \
+            f", @force_full_load                AS force_full_load \n" \
+            f", @data_changed_since_last_sync   AS data_changed_since_last_sync; \n"
 
         self.logger.debug(f"Getting ChangeTracking info for {table_config['schema']}.{table_config['name']}.\n"
                           f"{get_change_tracking_info_sql}")
 
         result = self.database_engine.execute(text(get_change_tracking_info_sql))
         row = result.fetchone()
 
-        return ChangeTrackingInfo(row["last_sync_version"], row["sync_version"], row["force_full_load"])
+        return ChangeTrackingInfo(  row["last_sync_version"], row["sync_version"],
+                                    row["force_full_load"], row["data_changed_since_last_sync"])
 
     @staticmethod
     def build_where_clause(batch_key_tracker, table_alias):

rdl/shared/Constants.py

@@ -10,6 +10,10 @@ class FullRefreshReason:
     MODEL_CHANGED = 'Model Changed'
     INVALID_CHANGE_TRACKING = 'Change Tracking Invalid'
 
+class IncrementalSkipReason:
+    NOT_APPLICABLE = 'N/A'
+    SYNC_VERSIONS_ARE_EQUAL = 'last_sync_version is the same as sync_version'
+    NO_DATA_CHANGED = 'No data has changed since last sync'
 
 class ExecutionStatus:
     STARTED = 'Started'