Do a full refresh on model change

Author: davidames

integration_tests/mssql_source/config/LargeTableTest.json

@@ -1,5 +1,4 @@
 {
-
   "source_table": {
     "name": "LargeTable",
     "schema": "dbo",

integration_tests/mssql_source/source_database_setup/create_database.sql

@@ -2,5 +2,4 @@ IF NOT EXISTS (SELECT * FROM sys.databases WHERE Name = 'RelationalDataLoaderInt
 	CREATE DATABASE	 RelationalDataLoaderIntegrationTestSource
 
 ALTER DATABASE RelationalDataLoaderIntegrationTestSource
-SET CHANGE_TRACKING = ON
-(CHANGE_RETENTION = 2 DAYS, AUTO_CLEANUP = ON)
+SET CHANGE_TRACKING = ON (CHANGE_RETENTION = 2 DAYS, AUTO_CLEANUP = ON)

modules/DataLoadManager.py

@@ -2,6 +2,8 @@
 import json
 import uuid
 import logging
+import hashlib
+
 from modules.BatchDataLoader import BatchDataLoader
 from modules.DestinationTableManager import DestinationTableManager
 from modules.data_load_tracking.DataLoadTracker import DataLoadTracker
@@ -25,35 +27,54 @@ def start_single_import(self, target_engine, model_name, requested_full_refresh)
 
         config_file = os.path.abspath(self.configuration_path + model_name)
         self.logger.debug("Using configuration file : {0}".format(config_file))
-        with open(config_file) as json_data:
-            pipeline_configuration = json.load(json_data)
+        with open(config_file) as json_file:
+            model_checksum = hashlib.md5(json_file.read().encode('utf-8')).hexdigest()
+            json_file.seek(0)
+            pipeline_configuration = json.load(json_file)
 
         self.logger.info("Execute Starting for: {0} requested_full_refresh: {1}".format(model_name, requested_full_refresh))
 
         destination_table_manager = DestinationTableManager(target_engine)
 
+        full_refresh_reason = "Command Line Argument" if requested_full_refresh else "N/A"
         full_refresh = requested_full_refresh
         if not requested_full_refresh and not destination_table_manager.table_exists(pipeline_configuration['target_schema'],
                                                                                      pipeline_configuration['load_table']):
             self.logger.warning("The load table {0}.{1} does not exist. Swapping to full-refresh mode".format(pipeline_configuration['target_schema'],
                                                                                                               pipeline_configuration['load_table']))
+
+            full_refresh_reason = "Destination table does not exist"
             full_refresh = True
 
         self.data_source.assert_data_source_is_valid(pipeline_configuration['source_table'],
                                                      pipeline_configuration['columns'])
 
-        last_sync_version = self.data_load_tracker_repository.get_last_sync_version(model_name)
+        last_successful_data_load_execution = self.data_load_tracker_repository.get_last_successful_data_load_execution(model_name)
+
+        if last_successful_data_load_execution is None:
+            last_sync_version = 0
+            full_refresh_reason = "First Execution"
+            full_refresh = True,
+        else:
+            self.logger.debug("Previous Checksum {0}. Current Checksum {1}".format(last_successful_data_load_execution.model_checksum, model_checksum))
+            last_sync_version = last_successful_data_load_execution.next_sync_version
+            if not full_refresh and last_successful_data_load_execution.model_checksum != model_checksum:
+                self.logger.info("A model checksum change has forced this to be a full load")
+                full_refresh = True
+                full_refresh_reason = "Model Change"
 
         change_tracking_info = self.data_source.init_change_tracking(pipeline_configuration['source_table'],
                                                                      last_sync_version)
 
 
-        data_load_tracker = DataLoadTracker(model_name, json_data, full_refresh, change_tracking_info, self.correlation_id)
 
-        self.logger.debug(" Change Tracking: this_sync_version: {0} next_sync_version: {1} force_full_load:{2}    : ".format(change_tracking_info.this_sync_version, change_tracking_info.next_sync_version, change_tracking_info.force_full_load))
         if not full_refresh and change_tracking_info.force_full_load:
             self.logger.info("Change tracking has forced this to be a full load")
             full_refresh = True
+            full_refresh_reason = "Change Tracking Invalid"
+
+        data_load_tracker = DataLoadTracker(model_name, model_checksum, json_file, full_refresh, change_tracking_info,
+                                            self.correlation_id, full_refresh_reason)
 
         columns = pipeline_configuration['columns']
         destination_table_manager.create_schema(pipeline_configuration['target_schema'])

modules/data_load_tracking/DataLoadExecution.py

@@ -17,4 +17,5 @@ class DataLoadExecution(Base):
     rows_processed = Column(Integer, nullable=False)
     correlation_id = Column(String(250), nullable=True)
     status = Column(String(25), nullable=False)
-
+    model_checksum = Column(String(100), nullable=False)
+    full_refresh_reason = Column(String(100), nullable=False)

modules/data_load_tracking/DataLoadTracker.py

@@ -14,15 +14,18 @@ class DataLoadTracker:
     total_row_count = 0
     rows_per_second = 0
     correlation_id = None,
+    full_refresh_reason = "N/A"
 
-    def __init__(self, model_name, configuration, is_full_refresh, change_tracking_info, correlation_id):
+    def __init__(self, model_name, model_checksum, configuration, is_full_refresh, change_tracking_info, correlation_id, full_refresh_reason):
         self.model_name = model_name
+        self.model_checksum = model_checksum
         self.configuration = configuration
         self.is_full_refresh = is_full_refresh
         self.started = datetime.now()
         self.status = "Not Started"
         self.change_tracking_info = change_tracking_info
         self.correlation_id = correlation_id
+        self.full_refresh_reason = full_refresh_reason
 
     def start_batch(self):
         batch = self.Batch()

modules/data_load_tracking/DataLoadTrackerRepository.py

@@ -12,13 +12,9 @@ def create_tables(self, engine):
         engine.execute("CREATE SCHEMA IF NOT EXISTS {0}".format("data_pipeline"))
         Base.metadata.create_all(engine)
 
-    def get_last_sync_version(self, model_name):
+    def get_last_successful_data_load_execution(self, model_name):
         session = self.session_maker()
-        result = session.query(DataLoadExecution).filter_by(model_name=model_name, status="Completed Successfully").order_by(desc(DataLoadExecution.completed_on)).first()
-
-        if result is None:
-            return 0
-        return result.next_sync_version
+        return session.query(DataLoadExecution).filter_by(model_name=model_name, status="Completed Successfully").order_by(desc(DataLoadExecution.completed_on)).first()
 
 
     def save(self, data_load_tracker):
@@ -30,9 +26,9 @@ def save(self, data_load_tracker):
                                                 next_sync_version=data_load_tracker.change_tracking_info.next_sync_version,
                                                 execution_time_ms=int(data_load_tracker.total_execution_time.total_seconds() * 1000),
                                                 rows_processed=data_load_tracker.total_row_count,
-                                                status=data_load_tracker.status)
-
-
+                                                status=data_load_tracker.status,
+                                                model_checksum=data_load_tracker.model_checksum,
+                                                full_refresh_reason = data_load_tracker.full_refresh_reason)
 
         session = self.session_maker()
         session.add(data_load_execution)