[DOP-22133] Implement increment for transfers with file sources

Author: Ilyas Gasanov

.env.docker

@@ -38,6 +38,14 @@ SYNCMASTER__AUTH__ACCESS_TOKEN__SECRET_KEY=generate_another_random_string
 # Scheduler options
 SYNCMASTER__SCHEDULER__TRANSFER_FETCHING_TIMEOUT_SECONDS=200
 
+# HWM Store
+SYNCMASTER__HWM_STORE__TYPE=horizon
+SYNCMASTER__HWM_STORE__URL=http://horizon:8000
+SYNCMASTER__HWM_STORE__NAMESPACE=syncmaster_namespace
+SYNCMASTER__HWM_STORE__USER=admin
+SYNCMASTER__HWM_STORE__PASSWORD=123UsedForTestOnly@!
+
+# Tests-only
 TEST_S3_HOST_FOR_CONFTEST=test-s3
 TEST_S3_PORT_FOR_CONFTEST=9000
 TEST_S3_HOST_FOR_WORKER=test-s3

.env.local

@@ -38,6 +38,13 @@ export SYNCMASTER__AUTH__ACCESS_TOKEN__SECRET_KEY=generate_another_random_string
 # Scheduler options
 export SYNCMASTER__SCHEDULER__TRANSFER_FETCHING_TIMEOUT_SECONDS=200
 
+# HWM Store
+export SYNCMASTER__HWM_STORE__TYPE=horizon
+export SYNCMASTER__HWM_STORE__URL=http://localhost:8020
+export SYNCMASTER__HWM_STORE__NAMESPACE=syncmaster_namespace
+export SYNCMASTER__HWM_STORE__USER=admin
+export SYNCMASTER__HWM_STORE__PASSWORD=123UsedForTestOnly@!
+
 # Tests-only
 export TEST_S3_HOST_FOR_CONFTEST=localhost
 export TEST_S3_PORT_FOR_CONFTEST=9010

docker-compose.test.yml

@@ -129,7 +129,51 @@ services:
         condition: service_completed_successfully
       rabbitmq:
         condition: service_healthy
-    profiles: [worker, scheduler, s3, oracle, hdfs, hive, clickhouse, mysql, mssql, sftp, ftp, ftps, samba, webdav, all]
+    profiles: [worker, scheduler, s3, hdfs, hive, oracle, clickhouse, mysql, mssql, sftp, ftp, ftps, samba, webdav, all]
+
+  horizon:
+    image: mtsrus/horizon-backend:develop
+    restart: unless-stopped
+    env_file: .env.docker
+    environment:
+      # list here usernames which should be assigned SUPERADMIN role on application start
+      HORIZON__ENTRYPOINT__ADMIN_USERS: admin
+      HORIZON__DATABASE__URL: postgresql+asyncpg://horizon:changeme@horizon-db:5432/horizon
+      HORIZON__AUTH__ACCESS_TOKEN__SECRET_KEY: generate_another_random_string
+      HORIZON__AUTH__PROVIDER: horizon.backend.providers.auth.dummy.DummyAuthProvider
+      HORIZON__SERVER__LOGGING__PRESET: colored
+      # PROMETHEUS_MULTIPROC_DIR is required for multiple workers, see:
+      # https://prometheus.github.io/client_python/multiprocess/
+      PROMETHEUS_MULTIPROC_DIR: /tmp/prometheus-metrics
+    # tmpfs dir is cleaned up each container restart
+    tmpfs:
+      - /tmp/prometheus-metrics
+    ports:
+      - 8020:8000
+    depends_on:
+      horizon-db:
+        condition: service_healthy
+    profiles: [horizon, s3, hdfs, hive, oracle, clickhouse, mysql, mssql, sftp, ftp, ftps, samba, webdav, all]
+
+  horizon-db:
+    image: postgres
+    restart: unless-stopped
+    environment:
+      TZ: UTC
+      POSTGRES_DB: horizon
+      POSTGRES_USER: horizon
+      POSTGRES_PASSWORD: changeme
+    ports:
+      - 5434:5432
+    volumes:
+      - horizon_test_data:/var/lib/postgresql/data
+    healthcheck:
+      test: pg_isready
+      start_period: 5s
+      interval: 30s
+      timeout: 5s
+      retries: 3
+    profiles: [horizon, s3, hdfs, hive, oracle, clickhouse, mysql, mssql, sftp, ftp, ftps, samba, webdav, all]
 
   test-postgres:
     image: postgres
@@ -147,7 +191,7 @@ services:
       interval: 30s
       timeout: 5s
       retries: 3
-    profiles: [s3, oracle, clickhouse, mysql, mssql, hdfs, hive, sftp, ftp, ftps, samba, webdav, all]
+    profiles: [s3, hdfs, hive, oracle, clickhouse, mysql, mssql, sftp, ftp, ftps, samba, webdav, all]
 
   test-s3:
     image: bitnami/minio:latest
@@ -366,4 +410,5 @@ services:
 volumes:
   postgres_test_data:
   rabbitmq_test_data:
+  horizon_test_data:
   keycloak_data:

docs/changelog/next_release/209.feature.rst

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

poetry.lock

@@ -71,6 +71,7 @@ apscheduler = { version = "^3.10.4", optional = true }
 starlette-exporter = {version = "^0.23.0", optional = true}
 itsdangerous = {version = "*", optional = true}
 python-keycloak = {version = ">=4.7,<6.0", optional = true}
+horizon-hwm-store = {version = ">=1.0.2", optional = true }
 
 [tool.poetry.extras]
 server = [
@@ -109,6 +110,7 @@ worker = [
     "coloredlogs",
     "python-json-logger",
     "pyyaml",
+    "horizon-hwm-store",
 ]
 
 scheduler = [

pyproject.toml

@@ -6,6 +6,8 @@
 
 from onetl.file.format import CSV, JSON, ORC, XML, Excel, JSONLine, Parquet
 
+from syncmaster.dto.transfers_strategy import FullStrategy, IncrementalStrategy
+
 
 @dataclass
 class TransferDTO:
@@ -14,14 +16,24 @@ class TransferDTO:
 
 @dataclass
 class DBTransferDTO(TransferDTO):
+    id: int
     table_name: str
+    strategy: FullStrategy | IncrementalStrategy
     transformations: list[dict] | None = None
+    options: dict | None = None
+
+    def __post_init__(self):
+        if self.options is None:
+            self.options = {}
+        self.options.setdefault("if_exists", "replace_entire_table")
 
 
 @dataclass
 class FileTransferDTO(TransferDTO):
+    id: int
     directory_path: str
     file_format: CSV | JSONLine | JSON | Excel | XML | ORC | Parquet
+    strategy: FullStrategy | IncrementalStrategy
     options: dict
     file_name_template: str | None = None
     df_schema: dict | None = None
@@ -43,7 +55,7 @@ def __post_init__(self):
         if isinstance(self.df_schema, str):
             self.df_schema = json.loads(self.df_schema)
 
-        self.options.setdefault("if_exists", "replace_entire_directory")  # TODO: use "append" for incremental strategy
+        self.options.setdefault("if_exists", "replace_overlapping_partitions")
 
     def _get_file_format(self, file_format: dict) -> CSV | JSONLine | JSON | Excel | XML | ORC | Parquet:
         file_type = file_format.pop("type", None)
@@ -87,6 +99,10 @@ class MySQLTransferDTO(DBTransferDTO):
 class HiveTransferDTO(DBTransferDTO):
     type: ClassVar[str] = "hive"
 
+    def __post_init__(self):
+        super().__post_init__()
+        self.options.setdefault("if_exists", "replace_overlapping_partitions")
+
 
 @dataclass
 class S3TransferDTO(FileTransferDTO):

syncmaster/dto/transfers.py

@@ -0,0 +1,33 @@
+# SPDX-FileCopyrightText: 2023-2024 MTS PJSC
+# SPDX-License-Identifier: Apache-2.0
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+
+@dataclass
+class Strategy:
+    type: str
+
+    @classmethod
+    def from_dict(cls, data: dict) -> Strategy:
+        strategy_classes = {
+            "full": FullStrategy,
+            "incremental": IncrementalStrategy,
+        }
+
+        strategy_type = data.get("type")
+        if strategy_type not in strategy_classes:
+            raise ValueError(f"Unknown strategy type: {strategy_type}")
+
+        return strategy_classes[strategy_type](**data)
+
+
+@dataclass
+class FullStrategy(Strategy):
+    pass
+
+
+@dataclass
+class IncrementalStrategy(Strategy):
+    increment_by: str

syncmaster/dto/transfers_strategy.py

@@ -255,8 +255,18 @@ def validate_increment_by(cls, values):
         source_type = values.source_params.type
         increment_by = values.strategy_params.increment_by
 
-        if source_type in FILE_CONNECTION_TYPES and increment_by != "modified_since":
-            raise ValueError("Field 'increment_by' must be equal to 'modified_since' for file source types")
+        if source_type in FILE_CONNECTION_TYPES and increment_by not in ("file_modified_since", "file_name"):
+            raise ValueError(
+                "Field 'increment_by' must be equal to 'file_modified_since' or 'file_name' for file source types",
+            )
+
+        return values
+
+    @model_validator(mode="after")
+    def validate_strategy(cls, values):
+
+        if values.source_params.type in ("s3", "hdfs") and isinstance(values.strategy_params, IncrementalStrategy):
+            raise ValueError("S3 and HDFS sources do not support incremental strategy for now")
 
         return values
 

syncmaster/schemas/v1/transfers/__init__.py

@@ -94,6 +94,9 @@ def validate_file_name_template(cls, value: str) -> str:
         if missing_keys:
             raise ValueError(f"Missing required placeholders: {', '.join(missing_keys)}")
 
+        if "{run_id}" not in value and "{run_created_at}" not in value:
+            raise ValueError("At least one of placeholders must be present: {run_id} or {run_created_at}")
+
         try:
             value.format(index="", extension="", run_created_at="", run_id="")
         except KeyError as e: