Merge pull request #39 from sendbird/release-1.2.1

Release 1.2.1

Author: jjh-kim

src/sbosc/controller/controller.py

@@ -61,8 +61,17 @@ def create_bulk_import_chunks(self):
         # Remove old chunks
         self.redis_data.remove_all_chunks()
 
+        # Get max pk
         metadata = self.redis_data.metadata
-        max_pk = metadata.max_pk
+        with self.db.cursor(role="reader") as cursor:
+            cursor.execute('''
+                  SELECT MIN(%s), MAX(%s) FROM %s.%s
+              ''' % (metadata.pk_column, metadata.pk_column, metadata.source_db, metadata.source_table))
+            max_pk = cursor.fetchone()[0]
+            if max_pk is None:
+                raise Exception("No data in source table")
+            metadata.max_pk = max_pk
+        self.logger.info("Saved total rows to Redis")
 
         # chunk_count is determined by min_chunk_size and max_chunk_count
         # Each chunk will have min_chunk_size rows and the number of chunks should not exceed max_chunk_count

src/sbosc/controller/initializer.py

@@ -185,16 +185,6 @@ def fetch_metadata(self, redis_data):
             metadata.pk_column = f"`{cursor.fetchone()[0]}`"
             self.logger.info("Saved primary key column to Redis")
 
-            # Get max PK
-            cursor.execute('''
-                SELECT MAX(%s) FROM %s.%s
-            ''' % (metadata.pk_column, metadata.source_db, metadata.source_table))
-            max_pk = cursor.fetchone()[0]
-            if max_pk is None:
-                raise Exception("No data in source table")
-            metadata.max_pk = max_pk
-            self.logger.info("Saved total rows to Redis")
-
         metadata.start_datetime = datetime.now()
         if not config.DISABLE_EVENTHANDLER:
             redis_data.set_current_stage(Stage.START_EVENT_HANDLER)

tests/test_controller.py

@@ -104,7 +104,8 @@ def test_chunk_creation(controller, setup_table, redis_data):
 
 @pytest.mark.parametrize('setup_table', ['with_data'], indirect=True)
 @pytest.mark.parametrize('controller', ['object'], indirect=True)
-def test_bulk_import_validation(controller: Controller, setup_table, cursor, override_operation_class):
+def test_bulk_import_validation(controller: Controller, setup_table, cursor, override_operation_class, redis_data):
+    redis_data.metadata.max_pk = TABLE_SIZE
     assert controller.validator.bulk_import_validation()
     delete_pks = random.sample(range(1, TABLE_SIZE), 10)
     cursor.execute(f'''