<p dir="auto"><strong>Apache Airflow version</strong>: 1.10.15 with postgres db in docker container</p>
<p dir="auto"><strong>Kubernetes version (if you are using kubernetes)</strong> (use <code class="notranslate">kubectl version</code>): NA</p>
<p dir="auto"><strong>Environment</strong>: DEV</p>
<ul dir="auto">
<li><strong>Cloud provider or hardware configuration</strong>:</li>
<li><strong>OS</strong> (e.g. from /etc/os-release): MacOS 10.15.7</li>
<li><strong>Kernel</strong> (e.g. <code class="notranslate">uname -a</code>): Darwin Kernel</li>
<li><strong>Install tools</strong>:</li>
<li><strong>Others</strong>:</li>
</ul>
<p dir="auto"><strong>What happened</strong>: Running <code class="notranslate">airflow upgrade_check</code> returns the following error:</p>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="Traceback (most recent call last):
  File &quot;/Users/abagri/Workspace/service-workflows/venv/bin/airflow&quot;, line 37, in &lt;module&gt;
    args.func(args)
  File &quot;/Users/abagri/Workspace/service-workflows/venv/lib/python3.7/site-packages/airflow/upgrade/checker.py&quot;, line 118, in run
    all_problems = check_upgrade(formatter, rules)
  File &quot;/Users/abagri/Workspace/service-workflows/venv/lib/python3.7/site-packages/airflow/upgrade/checker.py&quot;, line 38, in check_upgrade
    rule_status = RuleStatus.from_rule(rule)
  File &quot;/Users/abagri/Workspace/service-workflows/venv/lib/python3.7/site-packages/airflow/upgrade/problem.py&quot;, line 44, in from_rule
    result = rule.check()
  File &quot;/Users/abagri/Workspace/service-workflows/venv/lib/python3.7/site-packages/airflow/utils/db.py&quot;, line 74, in wrapper
    return func(*args, **kwargs)
  File &quot;/Users/abagri/Workspace/service-workflows/venv/lib/python3.7/site-packages/airflow/upgrade/rules/postgres_mysql_sqlite_version_upgrade_check.py&quot;, line 50, in check
    installed_postgres_version = Version(session.execute('SHOW server_version;').scalar())
  File &quot;/Users/abagri/Workspace/service-workflows/venv/lib/python3.7/site-packages/packaging/version.py&quot;, line 298, in __init__
    raise InvalidVersion(&quot;Invalid version: '{0}'&quot;.format(version))
packaging.version.InvalidVersion: Invalid version: '12.3 (Debian 12.3-1.pgdg100+1)'"><pre class="notranslate"><code class="notranslate">Traceback (most recent call last):
  File "/Users/abagri/Workspace/service-workflows/venv/bin/airflow", line 37, in &lt;module&gt;
    args.func(args)
  File "/Users/abagri/Workspace/service-workflows/venv/lib/python3.7/site-packages/airflow/upgrade/checker.py", line 118, in run
    all_problems = check_upgrade(formatter, rules)
  File "/Users/abagri/Workspace/service-workflows/venv/lib/python3.7/site-packages/airflow/upgrade/checker.py", line 38, in check_upgrade
    rule_status = RuleStatus.from_rule(rule)
  File "/Users/abagri/Workspace/service-workflows/venv/lib/python3.7/site-packages/airflow/upgrade/problem.py", line 44, in from_rule
    result = rule.check()
  File "/Users/abagri/Workspace/service-workflows/venv/lib/python3.7/site-packages/airflow/utils/db.py", line 74, in wrapper
    return func(*args, **kwargs)
  File "/Users/abagri/Workspace/service-workflows/venv/lib/python3.7/site-packages/airflow/upgrade/rules/postgres_mysql_sqlite_version_upgrade_check.py", line 50, in check
    installed_postgres_version = Version(session.execute('SHOW server_version;').scalar())
  File "/Users/abagri/Workspace/service-workflows/venv/lib/python3.7/site-packages/packaging/version.py", line 298, in __init__
    raise InvalidVersion("Invalid version: '{0}'".format(version))
packaging.version.InvalidVersion: Invalid version: '12.3 (Debian 12.3-1.pgdg100+1)'
</code></pre></div>

<p dir="auto"><strong>What you expected to happen</strong>: commands runs through and prints helpful messages</p>

<p dir="auto">Running <code class="notranslate">'SHOW server_version;</code> against the postgres db returns "12.3 (Debian 12.3-1.pgdg100+1)"' which is possibly not a valid value for Version class <strong>init</strong> function because of the <code class="notranslate">(Debian 12.3-1.pgdg100+1)</code> ending?<br>
<strong>How to reproduce it</strong>: Run airflow upgrade_check again.</p>

<p dir="auto"><strong>Anything else we need to know</strong>:</p>
<p dir="auto"><strong>Description</strong></p>
<p dir="auto">Currently, the <a href="https://github.com/apache/airflow/blob/main/airflow/providers/postgres/operators/postgres.py">PostgresOperator</a> defines the database hook in a fixed way:<br>
<code class="notranslate">self.hook = PostgresHook(postgres_conn_id=self.postgres_conn_id, schema=self.database)</code><br>
This makes the operator inflexible for certain tech stacks. For example, long queries in Redshift will sometimes fail (in my experience) with <code class="notranslate">psycopg2.operationalerror: SSL SYSCALL error: EOF detected</code> because the hook timed out on the Redshift side of things. The workaround (described in this <a href="https://stackoverflow.com/a/63130830/10131478" rel="nofollow">StackExchange post</a>) that worked for me is to specify some additional arguments in the Hook (see below) and then write my own custom operator.</p>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="PostgresHook(
        postgres_conn_id=postgres_conn_id,
        extra={
            &quot;keepalives&quot;: 1,
            &quot;keepalives_idle&quot;: 30,
            &quot;keepalives_interval&quot;: 5,
            &quot;keepalives_count&quot;: 5,
        }
    )"><pre class="notranslate"><code class="notranslate">PostgresHook(
        postgres_conn_id=postgres_conn_id,
        extra={
            "keepalives": 1,
            "keepalives_idle": 30,
            "keepalives_interval": 5,
            "keepalives_count": 5,
        }
    )
</code></pre></div>
<p dir="auto">Now I don't think the default implementation of the PostgresOperator should have these extra parameters, I just think it should be modified to be more flexible so that users can pass arguments to the hook as well. My proposed solution (and I will open a PR after some community input on this solution) would be either</p>
<p dir="auto">OPTION 1: (User-specified hook)</p>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="    def __init__(
        self,
        *,
        sql: str,
        postgres_conn_id: str = 'postgres_default',
        autocommit: bool = False,
        parameters: Optional[Union[Mapping, Iterable]] = None,
        database: Optional[str] = None,
        hook = PostgresHook(postgres_conn_id=self.postgres_conn_id, schema=self.database),
        **kwargs,
    ) -&gt; None:
        super().__init__(**kwargs)
        self.sql = sql
        self.postgres_conn_id = postgres_conn_id
        self.autocommit = autocommit
        self.parameters = parameters
        self.database = database
        self.hook = hook

    def execute(self, context):
        self.log.info('Executing: %s', self.sql)
        self.hook.run(self.sql, self.autocommit, parameters=self.parameters)
        for output in self.hook.conn.notices:
            self.log.info(output)"><pre class="notranslate"><code class="notranslate">    def __init__(
        self,
        *,
        sql: str,
        postgres_conn_id: str = 'postgres_default',
        autocommit: bool = False,
        parameters: Optional[Union[Mapping, Iterable]] = None,
        database: Optional[str] = None,
        hook = PostgresHook(postgres_conn_id=self.postgres_conn_id, schema=self.database),
        **kwargs,
    ) -&gt; None:
        super().__init__(**kwargs)
        self.sql = sql
        self.postgres_conn_id = postgres_conn_id
        self.autocommit = autocommit
        self.parameters = parameters
        self.database = database
        self.hook = hook

    def execute(self, context):
        self.log.info('Executing: %s', self.sql)
        self.hook.run(self.sql, self.autocommit, parameters=self.parameters)
        for output in self.hook.conn.notices:
            self.log.info(output)
</code></pre></div>
<p dir="auto">OPTION 2: (Pass extra params to hook)</p>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="    def __init__(
        self,
        *,
        sql: str,
        postgres_conn_id: str = 'postgres_default',
        autocommit: bool = False,
        parameters: Optional[Union[Mapping, Iterable]] = None,
        database: Optional[str] = None,
        **kwargs,
    ) -&gt; None:
        super().__init__(**kwargs)
        self.sql = sql
        self.postgres_conn_id = postgres_conn_id
        self.autocommit = autocommit
        self.parameters = parameters
        self.database = database
        self.extra = kwargs.pop(&quot;extra&quot;, None)
        self.hook = None

    def execute(self, context):
        self.log.info('Executing: %s', self.sql)
        self.hook = PostgresHook(
            postgres_conn_id=self.postgres_conn_id,
            schema=self.database,
            extra=self.extra
        )
        self.hook.run(self.sql, self.autocommit, parameters=self.parameters)
        for output in self.hook.conn.notices:
            self.log.info(output)"><pre class="notranslate"><code class="notranslate">    def __init__(
        self,
        *,
        sql: str,
        postgres_conn_id: str = 'postgres_default',
        autocommit: bool = False,
        parameters: Optional[Union[Mapping, Iterable]] = None,
        database: Optional[str] = None,
        **kwargs,
    ) -&gt; None:
        super().__init__(**kwargs)
        self.sql = sql
        self.postgres_conn_id = postgres_conn_id
        self.autocommit = autocommit
        self.parameters = parameters
        self.database = database
        self.extra = kwargs.pop("extra", None)
        self.hook = None

    def execute(self, context):
        self.log.info('Executing: %s', self.sql)
        self.hook = PostgresHook(
            postgres_conn_id=self.postgres_conn_id,
            schema=self.database,
            extra=self.extra
        )
        self.hook.run(self.sql, self.autocommit, parameters=self.parameters)
        for output in self.hook.conn.notices:
            self.log.info(output)
</code></pre></div>
<p dir="auto">I slightly favor Option 2 because I don't think one should be using the PostgresOperator with a hook other than the PostgresHook, so really all we want is for users to be able to pass the named parameters through.</p>
<p dir="auto"><strong>Use case / motivation</strong></p>
<p dir="auto">This feature will make the PostgresOperator more flexible, allowing users to pass more arguments to the PostgresHook to make better use of its features.</p>
<p dir="auto"><strong>Are you willing to submit a PR?</strong></p>
<p dir="auto">Yes, I would be happy to submit a PR.</p>
<p dir="auto"><strong>Related Issues</strong></p>
<p dir="auto">I couldn't find a related issue, but those more familiar with the project might identify this as a duplicate and I would be happy to delete if that is the case.</p>