[DOP-11682] Cache result of connection.check()

[dolfinus]

Change Summary

connection.check() is being called only once inside .run() method of DBReader/FileDownloader/... object, to avoid printing same connection detauls into logs again and again.

Before:

|onETL| Using IncrementalBatchStrategy as a strategy
        step = 10
|IncrementalBatchStrategy| First iteration
=================================== DBReader.run() starts ===================================
|DBReader| Getting Spark type for HWM expression: 'hwm_int'
|Postgres| Fetching schema of table 'onetl.test_b47b282cf4' ...
|Postgres| Schema fetched.
|DBReader| Got Spark field: StructField('hwm_int', LongType(), True)
|DBReader| Detected HWM type: 'ColumnIntHWM'
|IncrementalBatchStrategy| Fetching HWM from MemoryHWMStore:
        name = 'd5930cc53b'
|IncrementalBatchStrategy| Fetched HWM:
        hwm = ColumnIntHWM(
            name = 'd5930cc53b',
            entity = 'onetl.test_b47b282cf4',
            expression = 'hwm_int',
        )
|Postgres| -> |Spark| Reading DataFrame from source using parameters:
        source = 'onetl.test_b47b282cf4'
        columns = [
            '*',
        ]
        hwm = AutoDetectHWM(
            name = 'd5930cc53b',
            entity = 'onetl.test_b47b282cf4',
            expression = 'hwm_int',
        )
        options = {
            'fetchsize': 100000,
            'numPartitions': 1,
            'partitioningMode': 'range',
        }
|Postgres| Checking connection availability...
|Postgres| Using connection parameters:
        user = 'onetl'
        password = SecretStr('**********')
        host = 'localhost'
        port = 5432
        database = 'onetl'
        extra = {}
        jdbc_url = 'jdbc:postgresql://localhost:5432/onetl?ApplicationName=onetl'
|Postgres| Connection is available.
|Postgres| Getting min and max values for expression 'hwm_int' ...
|Postgres| Executing SQL query (on driver):
        SELECT
               MIN(hwm_int) AS "min",
               MAX(hwm_int) AS "max"
        FROM
               onetl.test_b47b282cf4
|Postgres| Received values:
        MIN(hwm_int) = 0
        MAX(hwm_int) = 50
|Postgres| Executing SQL query (on executor):
        SELECT
               *
        FROM
               onetl.test_b47b282cf4
        WHERE
               (hwm_int >= 0)
          AND
               (hwm_int <= 10)
|Spark| DataFrame successfully created from SQL statement 
------------------------------------ DBReader.run() ends ------------------------------------
|IncrementalBatchStrategy| Next iteration
|IncrementalBatchStrategy| Saving HWM to 'MemoryHWMStore':
        hwm = ColumnIntHWM(
            name = 'd5930cc53b',
            entity = 'onetl.test_b47b282cf4',
            expression = 'hwm_int',
            value = 10,
        )
|IncrementalBatchStrategy| HWM has been saved
=================================== DBReader.run() starts ===================================
|Postgres| -> |Spark| Reading DataFrame from source using parameters:
        source = 'onetl.test_b47b282cf4'
        columns = [
            '*',
        ]
        hwm = AutoDetectHWM(
            name = 'd5930cc53b',
            entity = 'onetl.test_b47b282cf4',
            expression = 'hwm_int',
        )
        options = {
            'fetchsize': 100000,
            'numPartitions': 1,
            'partitioningMode': 'range',
        }
|Postgres| Checking connection availability...
|Postgres| Using connection parameters:
        user = 'onetl'
        password = SecretStr('**********')
        host = 'localhost'
        port = 5432
        database = 'onetl'
        extra = {}
        jdbc_url = 'jdbc:postgresql://localhost:5432/onetl?ApplicationName=onetl'
|Postgres| Connection is available.
|Postgres| Executing SQL query (on executor):
        SELECT
               *
        FROM
               onetl.test_b47b282cf4
        WHERE
               (hwm_int >  10)
          AND
               (hwm_int <= 20)
|Spark| DataFrame successfully created from SQL statement 
------------------------------------ DBReader.run() ends ------------------------------------
|IncrementalBatchStrategy| Next iteration
|IncrementalBatchStrategy| Saving HWM to 'MemoryHWMStore':
        hwm = ColumnIntHWM(
            name = 'd5930cc53b',
            entity = 'onetl.test_b47b282cf4',
            expression = 'hwm_int',
            value = 20,
        )
|IncrementalBatchStrategy| HWM has been saved
=================================== DBReader.run() starts ===================================
|Postgres| -> |Spark| Reading DataFrame from source using parameters:
        source = 'onetl.test_b47b282cf4'
        columns = [
            '*',
        ]
        hwm = AutoDetectHWM(
            name = 'd5930cc53b',
            entity = 'onetl.test_b47b282cf4',
            expression = 'hwm_int',
        )
        options = {
            'fetchsize': 100000,
            'numPartitions': 1,
            'partitioningMode': 'range',
        }
|Postgres| Checking connection availability...
|Postgres| Using connection parameters:
        user = 'onetl'
        password = SecretStr('**********')
        host = 'localhost'
        port = 5432
        database = 'onetl'
        extra = {}
        jdbc_url = 'jdbc:postgresql://localhost:5432/onetl?ApplicationName=onetl'
|Postgres| Connection is available.
|Postgres| Executing SQL query (on executor):
        SELECT
               *
        FROM
               onetl.test_b47b282cf4
        WHERE
               (hwm_int >  20)
          AND
               (hwm_int <= 30)
|Spark| DataFrame successfully created from SQL statement 
------------------------------------ DBReader.run() ends ------------------------------------

After:

|onETL| Using IncrementalBatchStrategy as a strategy
        step = 10
|IncrementalBatchStrategy| First iteration
=================================== DBReader.run() starts ===================================
|DBReader| Getting Spark type for HWM expression: 'hwm_int'
|Postgres| Fetching schema of table 'onetl.test_a0662ca113' ...
|Postgres| Schema fetched.
|DBReader| Got Spark field: StructField('hwm_int', LongType(), True)
|DBReader| Detected HWM type: 'ColumnIntHWM'
|IncrementalBatchStrategy| Fetching HWM from MemoryHWMStore:
        name = 'e71fa4d9b1'
|IncrementalBatchStrategy| Fetched HWM:
        hwm = ColumnIntHWM(
            name = 'e71fa4d9b1',
            entity = 'onetl.test_a0662ca113',
            expression = 'hwm_int',
        )
|Postgres| Getting min and max values for expression 'hwm_int' ...
|Postgres| Executing SQL query (on driver):
        SELECT
               MIN(hwm_int) AS "min",
               MAX(hwm_int) AS "max"
        FROM
               onetl.test_a0662ca113
|Postgres| Received values:
        MIN(hwm_int) = 0
        MAX(hwm_int) = 50
|Postgres| Executing SQL query (on executor):
        SELECT
               *
        FROM
               onetl.test_a0662ca113
        WHERE
               (hwm_int >= 0)
          AND
               (hwm_int <= 10)
|Spark| DataFrame successfully created from SQL statement 
------------------------------------ DBReader.run() ends ------------------------------------
|IncrementalBatchStrategy| Saving HWM to 'MemoryHWMStore':
        hwm = ColumnIntHWM(
            name = 'e71fa4d9b1',
            entity = 'onetl.test_a0662ca113',
            expression = 'hwm_int',
            value = 10,
        )
|IncrementalBatchStrategy| HWM has been saved
|IncrementalBatchStrategy| Next iteration
=================================== DBReader.run() starts ===================================
|Postgres| Executing SQL query (on executor):
        SELECT
               *
        FROM
               onetl.test_a0662ca113
        WHERE
               (hwm_int >  10)
          AND
               (hwm_int <= 20)
|Spark| DataFrame successfully created from SQL statement 
------------------------------------ DBReader.run() ends ------------------------------------
|IncrementalBatchStrategy| Saving HWM to 'MemoryHWMStore':
        hwm = ColumnIntHWM(
            name = 'e71fa4d9b1',
            entity = 'onetl.test_a0662ca113',
            expression = 'hwm_int',
            value = 20,
        )
|IncrementalBatchStrategy| HWM has been saved
|IncrementalBatchStrategy| Next iteration
=================================== DBReader.run() starts ===================================
|Postgres| Executing SQL query (on executor):
        SELECT
               *
        FROM
               onetl.test_a0662ca113
        WHERE
               (hwm_int >  20)
          AND
               (hwm_int <= 30)
|Spark| DataFrame successfully created from SQL statement
------------------------------------ DBReader.run() ends ------------------------------------
|IncrementalBatchStrategy| Saving HWM to 'MemoryHWMStore':
        hwm = ColumnIntHWM(
            name = 'e71fa4d9b1',
            entity = 'onetl.test_a0662ca113',
            expression = 'hwm_int',
            value = 30,
        )
|IncrementalBatchStrategy| HWM has been saved

Related issue number

Checklist

• Commit message and PR title is comprehensive
• Keep the change as small as possible
• Unit and integration tests for the changes exist
• Tests pass on CI and coverage does not decrease
• Documentation reflects the changes where applicable
• docs/changelog/next_release/<pull request or issue id>.<change type>.rst file added describing change
  (see CONTRIBUTING.rst for details.)
• My PR is ready to review.

[codecov]

Codecov Report

Attention: 8 lines in your changes are missing coverage. Please review.

Comparison is base (26e8768) 94.27% compared to head (824de5a) 94.21%.

Files	Patch %	Lines
onetl/file/file_mover/file_mover.py	84.21%	1 Missing and 2 partials ⚠️
onetl/file/file_uploader/file_uploader.py	81.25%	1 Missing and 2 partials ⚠️
onetl/file/file_downloader/file_downloader.py	90.00%	1 Missing and 1 partial ⚠️

Additional details and impacted files

@@             Coverage Diff             @@
##           develop     #190      +/-   ##
===========================================
- Coverage    94.27%   94.21%   -0.06%     
===========================================
  Files          204      204              
  Lines         7704     7732      +28     
  Branches      1355     1370      +15     
===========================================
+ Hits          7263     7285      +22     
- Misses         327      329       +2     
- Partials       114      118       +4     

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.