Fixed CSV regressions

Author: davidames

modules/data_sources/CsvDataSource.py

@@ -3,13 +3,15 @@
 import os.path
 from modules.ColumnTypeResolver import ColumnTypeResolver
 from pathlib import Path
+from modules.data_sources.ChangeTrackingInfo import ChangeTrackingInfo
 
 
 class CsvDataSource(object):
     def __init__(self, connection_string, logger=None):
         self.logger = logger or logging.getLogger(__name__)
         self.source_path = Path(connection_string[len(self.connection_string_prefix()):])
         self.column_type_resolver = ColumnTypeResolver()
+
     @staticmethod
     def can_handle_connection_string(connection_string):
         return connection_string.startswith(CsvDataSource.connection_string_prefix())
@@ -41,7 +43,7 @@ def assert_column_exists(self, column_name, data_frame, csv_file):
 
 
     # For now, the CSV data sources will get all rows in the CSV regardless of batch size. - Ie, they don't currently support paging.
-    def get_next_data_frame(self, table_configuration, columns, batch_configuration, batch_tracker, previous_batch_key):
+    def get_next_data_frame(self, table_configuration, columns, batch_configuration, batch_tracker, previous_batch_key, full_refresh, change_tracking_info):
 
         if previous_batch_key > 0:
             return None
@@ -55,9 +57,11 @@ def get_next_data_frame(self, table_configuration, columns, batch_configuration,
 
         self.logger.debug("Starting read of file: {0}".format(csv_file))
 
-
         data_frame = pandas.read_csv(csv_file)
         self.logger.debug("Completed read")
 
         batch_tracker.extract_completed_successfully(len(data_frame))
         return data_frame
+
+    def init_change_tracking(self, table_configuration, last_sync_version):
+        return ChangeTrackingInfo(0,0)

modules/data_sources/MsSqlDataSource.py

@@ -34,8 +34,6 @@ def prefix_column(column_name, full_refresh, primary_key_column_name):
     def build_select_statement(self, table_configuration, columns, batch_configuration, previous_batch_key, full_refresh, change_tracking_info):
         column_array = list(map(lambda cfg: self.prefix_column(cfg['source_name'], full_refresh, table_configuration['primary_key']), columns))
         column_names = ", ".join(column_array)
-
-        #This line below is temp until we have a proper storage log of what ran - then data_pipeline_next_change_minimum_version will be stored there.
         column_names = "{0}, {1} as data_pipeline_next_change_minimum_version".format(column_names, change_tracking_info.next_sync_version)
         if full_refresh:
             return "SELECT TOP ({0}) {1} FROM {2}.{3} t WHERE t.{4} > {5} ORDER BY t.{4}".format(batch_configuration['size'],
@@ -86,7 +84,7 @@ def get_table_columns(self, table_configuration):
         return list(map(lambda column: column.name, table.columns))
 
 
-    def get_next_data_frame(self, table_configuration, columns, batch_configuration, batch_tracker, previous_batch_key, full_refresh, change_tracking_info,):
+    def get_next_data_frame(self, table_configuration, columns, batch_configuration, batch_tracker, previous_batch_key, full_refresh, change_tracking_info):
         sql = self.build_select_statement(table_configuration, columns, batch_configuration, previous_batch_key, full_refresh, change_tracking_info,)
         self.logger.debug("Starting read of SQL Statement: {0}".format(sql))
         data_frame = pandas.read_sql_query(sql, self.database_engine)
@@ -110,7 +108,6 @@ def init_change_tracking(self, table_configuration, last_sync_version):
 
         self.database_engine.execute(text(sql_builder.getvalue()).execution_options(autocommit=True))
 
-
         sql_builder = io.StringIO()
         sql_builder.write("DECLARE @last_sync_version bigint = {0}; \n".format(last_sync_version))
         sql_builder.write("DECLARE @this_sync_version bigint = 0; \n")
@@ -125,9 +122,6 @@ def init_change_tracking(self, table_configuration, last_sync_version):
 
         result = self.database_engine.execute(sql_builder.getvalue())
         row = result.fetchone()
-        return_value = ChangeTrackingInfo(row["this_sync_version"], row["next_sync_version"])
-
         sql_builder.close()
 
-        return return_value
-
+        return ChangeTrackingInfo(row["this_sync_version"], row["next_sync_version"])

test_full_refresh_from_mssql.cmd

@@ -1,9 +1,9 @@
 @echo off
-REM IF /I "%APPVEYOR%"=="TRUE" py rdl.py mssql+pyodbc://(local)\SQL2016/RelationalDataLoaderIntegrationTestSource?driver=SQL+Server+Native+Client+11.0 postgresql+psycopg2://postgres:there_is_no_password_due_to_pg_trust@localhost/relational_data_loader_integration_tests ./integration_tests/mssql_source/config/ --log-level DEBUG --full-refresh yes
+IF /I "%APPVEYOR%"=="TRUE" py rdl.py mssql+pyodbc://(local)\SQL2016/RelationalDataLoaderIntegrationTestSource?driver=SQL+Server+Native+Client+11.0 postgresql+psycopg2://postgres:there_is_no_password_due_to_pg_trust@localhost/relational_data_loader_integration_tests ./integration_tests/mssql_source/config/ --log-level DEBUG --full-refresh yes
 IF /I NOT "%APPVEYOR%"=="TRUE" py rdl.py mssql+pyodbc://(local)/RelationalDataLoaderIntegrationTestSource?driver=SQL+Server+Native+Client+11.0 postgresql+psycopg2://postgres:there_is_no_password_due_to_pg_trust@localhost/relational_data_loader_integration_tests ./integration_tests/mssql_source/config/ --log-level DEBUG --full-refresh no
-REM if %errorlevel% neq 0 exit /b %errorlevel%
-REM psql -U postgres -d relational_data_loader_integration_tests -a -v ON_ERROR_STOP=1 -f ./integration_tests/mssql_source/assertions/large_table_test_full_refresh_assertions.sql
-REM if %errorlevel% neq 0 exit /b %errorlevel%
+if %errorlevel% neq 0 exit /b %errorlevel%
+psql -U postgres -d relational_data_loader_integration_tests -a -v ON_ERROR_STOP=1 -f ./integration_tests/mssql_source/assertions/large_table_test_full_refresh_assertions.sql
+if %errorlevel% neq 0 exit /b %errorlevel%