[DOP-22132] Implement increment for transfers with DB source

Author: Ilyas Gasanov

docs/changelog/next_release/211.feature.rst

@@ -0,0 +1 @@
+Implement increment for transfers with database sources

syncmaster/worker/controller.py

@@ -42,6 +42,7 @@
 )
 from syncmaster.dto.transfers_strategy import Strategy
 from syncmaster.exceptions.connection import ConnectionTypeNotRecognizedError
+from syncmaster.schemas.v1.connection_types import FILE_CONNECTION_TYPES
 from syncmaster.worker.handlers.base import Handler
 from syncmaster.worker.handlers.db.clickhouse import ClickhouseHandler
 from syncmaster.worker.handlers.db.hive import HiveHandler
@@ -242,11 +243,15 @@ def _perform_incremental_transfer(self) -> None:
         ).force_create_namespace() as hwm_store:
 
             with IncrementalStrategy():
+                if self.source_handler.connection_dto.type in FILE_CONNECTION_TYPES:
+                    hwm_name_suffix = self.source_handler.transfer_dto.directory_path
+                else:
+                    hwm_name_suffix = self.source_handler.transfer_dto.table_name
                 hwm_name = "_".join(
                     [
                         str(self.source_handler.transfer_dto.id),
                         self.source_handler.connection_dto.type,
-                        self.source_handler.transfer_dto.directory_path,
+                        hwm_name_suffix,
                     ],
                 )
                 hwm = hwm_store.get_hwm(hwm_name)

syncmaster/worker/handlers/db/base.py

@@ -36,11 +36,21 @@ class DBHandler(Handler):
     }
 
     def read(self) -> DataFrame:
+        reader_params = {}
+        if self.transfer_dto.strategy.type == "incremental":
+            self.transfer_dto.strategy.increment_by = self._quote_field(self.transfer_dto.strategy.increment_by)
+            hwm_name = f"{self.transfer_dto.id}_{self.connection_dto.type}_{self.transfer_dto.table_name}"
+            reader_params["hwm"] = DBReader.AutoDetectHWM(
+                name=hwm_name,
+                expression=self.transfer_dto.strategy.increment_by,
+            )
+
         reader = DBReader(
             connection=self.connection,
             table=self.transfer_dto.table_name,
             where=self._get_rows_filter_expression(),
             columns=self._get_columns_filter_expressions(),
+            **reader_params,
         )
         return reader.run()
 

tests/test_integration/test_run_transfer/connection_fixtures/__init__.py

@@ -123,6 +123,7 @@
     full_strategy,
     incremental_strategy_by_file_modified_since,
     incremental_strategy_by_file_name,
+    incremental_strategy_by_number_column,
 )
 from tests.test_integration.test_run_transfer.connection_fixtures.webdav_fixtures import (
     prepare_webdav,

tests/test_integration/test_run_transfer/connection_fixtures/filters_fixtures.py

@@ -50,16 +50,26 @@ def expected_dataframe_rows_filter():
 
 @pytest.fixture
 def dataframe_columns_filter_transformations(source_type: str):
-    string_types_per_source_type = {
-        "postgres": "VARCHAR(10)",
-        "oracle": "VARCHAR2(10)",
-        "clickhouse": "VARCHAR(10)",
+    string_type_per_source_type = {
+        "postgres": "VARCHAR(30)",
+        "oracle": "VARCHAR2(30)",
+        "clickhouse": "VARCHAR(30)",
         "mysql": "CHAR",
-        "mssql": "VARCHAR(10)",
-        "hive": "VARCHAR(10)",
+        "mssql": "VARCHAR(30)",
+        "hive": "VARCHAR(30)",
         "s3": "STRING",
         "hdfs": "STRING",
     }
+    int_type_per_source_type = {
+        "postgres": "INTEGER",
+        "oracle": "NUMBER",
+        "clickhouse": "Int32",
+        "mysql": "INT",
+        "mssql": "INT",
+        "hive": "INT",
+        "s3": "INT",
+        "hdfs": "INT",
+    }
     return [
         {
             "type": "dataframe_columns_filter",
@@ -74,7 +84,7 @@ def dataframe_columns_filter_transformations(source_type: str):
                 },
                 {
                     "type": "include",
-                    "field": "PHONE_NUMBER",
+                    "field": "NUMBER",
                 },
                 {
                     "type": "rename",
@@ -83,8 +93,8 @@ def dataframe_columns_filter_transformations(source_type: str):
                 },
                 {
                     "type": "cast",
-                    "field": "NUMBER",
-                    "as_type": string_types_per_source_type[source_type],
+                    "field": "ACCOUNT_BALANCE",
+                    "as_type": int_type_per_source_type[source_type],
                 },
                 {
                     "type": "include",
@@ -100,9 +110,9 @@ def expected_dataframe_columns_filter():
     return lambda df, source_type: df.selectExpr(
         "ID",
         "REGION",
-        "PHONE_NUMBER",
+        "NUMBER",
         "REGION AS REGION2",
-        "CAST(NUMBER AS STRING) AS NUMBER",
+        "CAST(ACCOUNT_BALANCE AS INT) AS ACCOUNT_BALANCE",
         "REGISTERED_AT",
     )
 

tests/test_integration/test_run_transfer/connection_fixtures/s3_fixtures.py

@@ -82,7 +82,7 @@ def s3_file_connection(s3_server):
     return s3_connection
 
 
-@pytest.fixture(scope="session")
+@pytest.fixture
 def s3_file_connection_with_path(request, s3_file_connection):
     connection = s3_file_connection
     source = PurePosixPath("/data")
@@ -99,7 +99,7 @@ def finalizer():
     return connection, source
 
 
-@pytest.fixture(scope="session")
+@pytest.fixture
 def s3_file_df_connection_with_path(s3_file_connection_with_path, s3_file_df_connection):
     _, root = s3_file_connection_with_path
     return s3_file_df_connection, root
@@ -123,7 +123,7 @@ def s3_file_df_connection(s3_file_connection, spark, s3_server):
     )
 
 
-@pytest.fixture(scope="session")
+@pytest.fixture
 def prepare_s3(
     resource_path: PosixPath,
     s3_file_connection: S3,

tests/test_integration/test_run_transfer/connection_fixtures/strategy_fixtures.py

@@ -22,3 +22,11 @@ def incremental_strategy_by_file_name():
         "type": "incremental",
         "increment_by": "file_name",
     }
+
+
+@pytest.fixture
+def incremental_strategy_by_number_column():
+    return {
+        "type": "incremental",
+        "increment_by": "NUMBER",
+    }