[OSC-1317] remove alter table calls to source db (#43)

* remove alter table calls to source db

* add change tracking in integration test create table scripts

* add change tracking in unit test create table scripts

Author: Sean Budd

rdl/DataLoadManager.py

@@ -95,7 +95,7 @@ def start_single_import(self, model_file, requested_full_refresh, model_number,
             last_sync_version = last_successful_data_load_execution.sync_version
 
         destination_table_manager = DestinationTableManager(self.target_db)
-        change_tracking_info = self.source_db.init_change_tracking(model_config['source_table'], last_sync_version)
+        change_tracking_info = self.source_db.get_change_tracking_info(model_config['source_table'], last_sync_version)
 
         last_successful_execution_exists = last_successful_data_load_execution is not None
         model_changed = (not last_successful_execution_exists) or \

rdl/data_sources/CsvDataSource.py

@@ -75,5 +75,5 @@ def get_next_data_frame(
         batch_tracker.extract_completed_successfully(len(data_frame))
         return data_frame
 
-    def init_change_tracking(self, table_config, last_sync_version):
+    def get_change_tracking_info(self, table_config, last_sync_version):
         return ChangeTrackingInfo(0, 0, False, True)

rdl/data_sources/MsSqlDataSource.py

@@ -149,23 +149,11 @@ def get_next_data_frame(self, table_config, columns, batch_config, batch_tracker
 
         return data_frame
 
-    def init_change_tracking(self, table_config, last_known_sync_version):
+    def get_change_tracking_info(self, table_config, last_known_sync_version):
 
         if last_known_sync_version is None:
             last_known_sync_version = 'NULL'
 
-        init_change_tracking_sql = "IF NOT EXISTS(SELECT 1 FROM sys.change_tracking_tables " \
-            f"WHERE object_id = OBJECT_ID('{table_config['schema']}.{table_config['name']}'))\n" \
-                                   "BEGIN\n" \
-            f"ALTER TABLE {table_config['schema']}.{table_config['name']} " \
-            f"ENABLE CHANGE_TRACKING WITH(TRACK_COLUMNS_UPDATED=OFF);\n" \
-                                   "END\n"
-
-        self.logger.debug(f"Initializing ChangeTracking for "
-                          f"{table_config['schema']}.{table_config['name']}:\n"
-                          f"{init_change_tracking_sql}")
-        self.database_engine.execute(text(init_change_tracking_sql).execution_options(autocommit=True))
-
         # in the following we determine:
         # a) the current sync version - sourced straight up from the source db.
         # b) the last valid sync version - derived from the last known sync version and its validity based on the

tests/integration_tests/mssql_source/source_database_setup/create_compound_pk.sql

@@ -17,4 +17,8 @@ SELECT	1,2
 UNION ALL
 SELECT	2,2
 UNION ALL
-SELECT	2,1
+SELECT	2,1;
+
+
+ALTER TABLE CompoundPk
+ENABLE CHANGE_TRACKING WITH(TRACK_COLUMNS_UPDATED=OFF);

tests/integration_tests/mssql_source/source_database_setup/create_large_table.sql

@@ -42,4 +42,8 @@ SELECT    n,
         CASE WHEN n % 13 = 0 THEN NULL ELSE newid() END,
         CASE WHEN n % 3 = 0 THEN NULL ELSE 1 END
 
-FROM    Numbers
+FROM    Numbers;
+
+
+ALTER TABLE LargeTable
+ENABLE CHANGE_TRACKING WITH(TRACK_COLUMNS_UPDATED=OFF);

tests/unit_tests/setup/create_simple_table.sql

@@ -16,3 +16,6 @@ UNION ALL
 SELECT	2,'Walker'
 UNION ALL
 SELECT	3,'National';
+
+ALTER TABLE SimpleTest
+ENABLE CHANGE_TRACKING WITH(TRACK_COLUMNS_UPDATED=OFF);

tests/unit_tests/test_MsSqlDataSource.py

@@ -59,19 +59,19 @@ def tearDownClass(cls):
         TestMsSqlDataSource.data_source.database_engine.execute(text(TEAR_DOWN_STRING))
         TestMsSqlDataSource.data_source = None
 
-    def test_init_change_tracking(self):
+    def test_get_change_tracking_info(self):
 
         last_sync_version = None
         for table in TestMsSqlDataSource.table_configs:
             print("TESTING ON TABLE: " + table["source_table"]["name"])
             print("FIRST TEST: INITIALISE TABLE")
-            results = TestMsSqlDataSource.data_source.init_change_tracking(
+            results = TestMsSqlDataSource.data_source.get_change_tracking_info(
                 table["source_table"], last_sync_version)
             self.assertEqual(results.force_full_load, True)
 
             print("SECOND TEST: NO CHANGES")
             last_sync_version = results.sync_version
-            results = TestMsSqlDataSource.data_source.init_change_tracking(
+            results = TestMsSqlDataSource.data_source.get_change_tracking_info(
                 table["source_table"], last_sync_version)
             self.assertEqual(results.force_full_load, False)
 
@@ -82,20 +82,20 @@ def test_init_change_tracking(self):
                 TestMsSqlDataSource.data_source.database_engine.execute(
                     text(operation_string).execution_options(autocommit=True))
 
-                results = TestMsSqlDataSource.data_source.init_change_tracking(
+                results = TestMsSqlDataSource.data_source.get_change_tracking_info(
                     table["source_table"], last_sync_version)
                 self.assertEqual(results.force_full_load, False, msg="Failed on: " + operation_string)
                 last_sync_version = results.sync_version
 
             print("EXTRA TEST: NO CHANGES")
             last_sync_version = results.sync_version
-            results = TestMsSqlDataSource.data_source.init_change_tracking(
+            results = TestMsSqlDataSource.data_source.get_change_tracking_info(
                 table["source_table"], last_sync_version)
             self.assertEqual(results.force_full_load, False)
 
             print("EXTRA TEST: LOST TRACK")
             last_sync_version = -1
-            results = TestMsSqlDataSource.data_source.init_change_tracking(table["source_table"], last_sync_version)
+            results = TestMsSqlDataSource.data_source.get_change_tracking_info(table["source_table"], last_sync_version)
             self.assertEqual(results.force_full_load, True)
 
     def test_can_handle_connection_string(self):