basic incremental loads now working

Author: davidames

modules/BatchDataLoader.py

@@ -6,7 +6,7 @@
 
 class BatchDataLoader(object):
     def __init__(self, data_source, source_table_configuration, target_schema, target_table, columns, data_load_tracker,
-                 batch_configuration, target_engine, logger=None):
+                 batch_configuration, target_engine, full_refresh, change_tracking_info, logger=None):
         self.logger = logger or logging.getLogger(__name__)
         self.source_table_configuration = source_table_configuration
         self.columns = columns
@@ -16,15 +16,18 @@ def __init__(self, data_source, source_table_configuration, target_schema, targe
         self.data_load_tracker = data_load_tracker
         self.batch_configuration = batch_configuration
         self.target_engine = target_engine
+        self.full_refresh = full_refresh
+        self.change_tracking_info = change_tracking_info
 
     # Imports rows, returns True if >0 rows were found
     def load_batch(self, previous_batch_key):
         batch_tracker = self.data_load_tracker.start_batch()
 
-        self.logger.debug("ImportBatch Starting from previous_batch_key: {0}".format(previous_batch_key))
+        self.logger.debug("ImportBatch Starting from previous_batch_key: {0}. Full Refresh: {1} this_sync_version: {2}".format(previous_batch_key, self.full_refresh, self.change_tracking_info.this_sync_version))
 
         data_frame = self.data_source.get_next_data_frame(self.source_table_configuration, self.columns,
-                                                          self.batch_configuration, batch_tracker, previous_batch_key)
+                                                          self.batch_configuration, batch_tracker, previous_batch_key,
+                                                          self.full_refresh, self.change_tracking_info)
 
         if data_frame is None or len(data_frame) == 0:
             self.logger.debug("There are no rows to import, returning -1")
@@ -44,7 +47,7 @@ def write_data_frame_to_table(self, data_frame):
         qualified_target_table = "{0}.{1}".format(self.target_schema, self.target_table)
         self.logger.debug("Starting write to table {0}".format(qualified_target_table))
         data = StringIO()
-
+        print(data_frame)
         data_frame.to_csv(data, header=False, index=False, na_rep='', float_format='%.16g')
         # Float_format is used to truncate any insignificant digits. Unfortunately it gives us an artificial limitation
 
@@ -58,6 +61,7 @@ def write_data_frame_to_table(self, data_frame):
 
         sql = "COPY {0}({1}) FROM STDIN with csv".format(qualified_target_table, column_list)
         self.logger.debug("Writing to table using command {0}".format(sql))
+
         curs.copy_expert(sql=sql, file=data)
 
         self.logger.debug("Completed write to table {0}".format(qualified_target_table))
@@ -70,6 +74,10 @@ def get_destination_column_name(self, source_column_name):
             if column['source_name'] == source_column_name:
                 return column['destination']['name']
 
+        # Internal columns - map them straight through
+        if source_column_name.startswith("data_pipeline_"):
+            return source_column_name;
+
         message = 'A source column with name {0} was not found in the column configuration'.format(source_column_name)
         raise ValueError(message)
 

modules/DataLoadManager.py

@@ -38,14 +38,23 @@ def start_single_import(self, target_engine, configuration_name, requested_full_
                                                                                                               pipeline_configuration['load_table']))
             full_refresh = True
 
-        #Ask the source to check that ChangeTrackingIsEnabled and valid
-        #Maybe this returns a flag?
+        self.data_source.assert_data_source_is_valid(pipeline_configuration['source_table'],
+                                                     pipeline_configuration['columns'])
 
+        last_sync_version = destination_table_manager.get_last_sync_version(pipeline_configuration['target_schema'],
+                                                                            pipeline_configuration['load_table'])
 
-        data_load_tracker = DataLoadTracker(configuration_name, json_data, full_refresh)
+        change_tracking_info = self.data_source.init_change_tracking(pipeline_configuration['source_table'],
+                                                                     last_sync_version)
 
-        self.data_source.assert_data_source_is_valid(pipeline_configuration['source_table'],
-                                                     pipeline_configuration['columns'])
+
+        data_load_tracker = DataLoadTracker(configuration_name, json_data, full_refresh, change_tracking_info)
+
+
+        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
 
         columns = pipeline_configuration['columns']
         destination_table_manager.create_schema(pipeline_configuration['target_schema'])
@@ -64,7 +73,9 @@ def start_single_import(self, target_engine, configuration_name, requested_full_
                                          columns,
                                          data_load_tracker,
                                          pipeline_configuration['batch'],
-                                         target_engine)
+                                         target_engine,
+                                         full_refresh,
+                                         change_tracking_info)
 
         previous_unique_column_value = 0
         while previous_unique_column_value > -1:
@@ -86,5 +97,5 @@ def start_single_import(self, target_engine, configuration_name, requested_full_
             destination_table_manager.drop_table(pipeline_configuration['target_schema'],
                                                  pipeline_configuration['stage_table'])
         data_load_tracker.completed_successfully()
-        self.logger.info("Import for configuration: {0} Complete. {1}".format(configuration_name, data_load_tracker.get_statistics()))
+        self.logger.info("Import Complete for: {0}. {1}".format(configuration_name, data_load_tracker.get_statistics()))
 

modules/DataLoadTracker.py

@@ -14,12 +14,13 @@ class DataLoadTracker:
     total_row_count = 0
     rows_per_second = 0
 
-    def __init__(self, configuration_name, configuration, is_full_load):
+    def __init__(self, configuration_name, configuration, is_full_load, change_tracking_info):
         self.configuration_name = configuration_name
         self.configuration = configuration
         self.is_full_load = is_full_load
         self.started = datetime.now()
         self.status = "Not Started"
+        self.change_tracking_info = change_tracking_info
 
     def start_batch(self):
         batch = self.Batch()
@@ -36,9 +37,11 @@ def completed_successfully(self):
         self.rows_per_second = self.total_row_count / self.total_execution_time.total_seconds()
 
     def get_statistics(self):
-        return "Rows: {0}, Total Execution Time: {1}. ({2:.2f} rows per second)".format(self.total_row_count,
-                                                                                            self.total_execution_time,
-                                                                                            self.rows_per_second)
+        load_type = 'full' if self.is_full_load else "incremental from version {0} ".format(self.change_tracking_info.this_sync_version)
+        return "Rows: {0} ({1}), Total Execution Time: {2}. ({3:.2f} rows per second) ".format(self.total_row_count,
+                                                                                               load_type,
+                                                                                               self.total_execution_time,
+                                                                                               self.rows_per_second)
 
     class Batch:
         row_count = 0

modules/DestinationTableManager.py

@@ -3,14 +3,16 @@
 import logging
 from modules.ColumnTypeResolver import ColumnTypeResolver
 
-from sqlalchemy import MetaData, DateTime, Boolean
+from sqlalchemy import MetaData, DateTime, Boolean, BigInteger
 from sqlalchemy.schema import Column, Table
 from sqlalchemy.sql import func
 
 
 class DestinationTableManager(object):
     TIMESTAMP_COLUMN_NAME = "data_pipeline_timestamp"
     IS_DELETED_COLUMN_NAME = "data_pipeline_is_deleted"
+    CHANGE_VERSION_COLUMN_NAME = "data_pipeline_change_version"
+    NEXT_CHANGE_MINIMUM_VERSION_COLUMN_NAME = "data_pipeline_next_change_minimum_version"
 
     def __init__(self, target_engine, logger=None):
         self.logger = logger or logging.getLogger(__name__)
@@ -52,9 +54,11 @@ def create_table(self, schema_name, table_name, columns_configuration, drop_firs
         table.append_column(
             Column(self.IS_DELETED_COLUMN_NAME, Boolean, server_default='f', default=False))
 
+        table.append_column(
+            Column(self.CHANGE_VERSION_COLUMN_NAME, BigInteger))
 
-
-
+        table.append_column(
+            Column(self.NEXT_CHANGE_MINIMUM_VERSION_COLUMN_NAME, BigInteger))
 
         if drop_first:
             self.logger.debug(
@@ -63,16 +67,29 @@ def create_table(self, schema_name, table_name, columns_configuration, drop_firs
             self.logger.debug(
                 "Dropped table {0}.{1}".format(schema_name, table_name))
 
+
         self.logger.debug("Creating table {0}.{1}".format(schema_name, table_name))
         table.create(self.target_engine, checkfirst=False)
         self.logger.debug("Created table {0}.{1}".format(schema_name, table_name))
+
         return
 
     def create_column(self, configuration):
         return Column(configuration['name'], self.column_type_resolver.resolve_postgres_type(configuration),
                       primary_key=configuration.get("primary_key", False),
                       nullable=configuration['nullable'])
 
+    # TODO: this should come from a different log table which is only written to at the end of a successful load load.
+    def get_last_sync_version(self, schema_name, table_name):
+        sql = "SELECT max({0}) as version FROM {1}.{2}".format(self.NEXT_CHANGE_MINIMUM_VERSION_COLUMN_NAME, schema_name, table_name)
+
+        result = self.target_engine.execute(sql)
+        row = result.fetchone()
+        if row["version"] is None:
+            return 0
+        return row["version"]
+
+
     def rename_table(self, schema_name, source_table_name, target_table_name):
 
         # Steps to efficiently rename a table.
@@ -116,6 +133,10 @@ def upsert_table(self, schema_name, source_table_name, target_table_name, column
         column_array = list(map(lambda column: column['destination']['name'], columns_configuration))
         column_list = ','.join(map(str, column_array))
         column_list = column_list + ",{0}".format(self.TIMESTAMP_COLUMN_NAME)
+        column_list = column_list + ",{0}".format(self.IS_DELETED_COLUMN_NAME)
+        column_list = column_list + ",{0}".format(self.CHANGE_VERSION_COLUMN_NAME)
+        column_list = column_list + ",{0}".format(self.NEXT_CHANGE_MINIMUM_VERSION_COLUMN_NAME)
+
 
         primary_key_column_array = [column_configuration['destination']['name'] for column_configuration in columns_configuration if 'primary_key' in column_configuration['destination'] and column_configuration['destination']['primary_key']]
 
@@ -132,7 +153,14 @@ def upsert_table(self, schema_name, source_table_name, target_table_name, column
             sql_builder.write("{0} = EXCLUDED.{0},".format(column_configuratiomn['destination']['name']))
             sql_builder.write(os.linesep)
 
-        sql_builder.write("{0} = EXCLUDED.{0}".format(self.TIMESTAMP_COLUMN_NAME))
+        sql_builder.write("{0} = EXCLUDED.{0},".format(self.TIMESTAMP_COLUMN_NAME))
+        sql_builder.write(os.linesep)
+        sql_builder.write("{0} = EXCLUDED.{0},".format(self.IS_DELETED_COLUMN_NAME))
+        sql_builder.write(os.linesep)
+        sql_builder.write("{0} = EXCLUDED.{0},".format(self.CHANGE_VERSION_COLUMN_NAME))
+        sql_builder.write(os.linesep)
+        sql_builder.write("{0} = EXCLUDED.{0}".format(self.NEXT_CHANGE_MINIMUM_VERSION_COLUMN_NAME))
+        sql_builder.write(os.linesep)
 
         self.logger.debug("Upsert executing {0}".format(sql_builder.getvalue()))
         self.target_engine.execute(sql_builder.getvalue())

modules/data_sources/ChangeTrackingInfo.py

@@ -0,0 +1,13 @@
+
+class ChangeTrackingInfo:
+    this_sync_version = 0
+    next_sync_version = 0
+
+    def __init__(self, this_sync_version, next_sync_version):
+        self.this_sync_version = this_sync_version
+        self.next_sync_version = next_sync_version
+
+    def force_full_load(self):
+        return bool(self.next_sync_version == 0)
+
+