Issue 1:
Apache Airflow version: 1.10.15 with postgres db in docker container

Kubernetes version (if you are using kubernetes) (use kubectl version): NA

Environment: DEV

Cloud provider or hardware configuration:

OS (e.g. from /etc/os-release): MacOS 10.15.7

Kernel (e.g. uname -a): Darwin Kernel

Install tools:

Others:

What happened: Running airflow upgrade_check returns the following error:

Traceback (most recent call last):
File "/Users/abagri/Workspace/service-workflows/venv/bin/airflow", line 37, in <module>
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)'

What you expected to happen: commands runs through and prints helpful messages

Running 'SHOW server_version; against the postgres db returns "12.3 (Debian 12.3-1.pgdg100+1)"' which is possibly not a valid value for Version class init function because of the (Debian 12.3-1.pgdg100+1) ending?

How to reproduce it: Run airflow upgrade_check again.

Anything else we need to know:

Issue 2:
Description

Currently, the PostgresOperator defines the database hook in a fixed way:

self.hook = PostgresHook(postgres_conn_id=self.postgres_conn_id, schema=self.database)

This makes the operator inflexible for certain tech stacks. For example, long queries in Redshift will sometimes fail (in my experience) with psycopg2.operationalerror: SSL SYSCALL error: EOF detected because the hook timed out on the Redshift side of things. The workaround (described in this StackExchange post) that worked for me is to specify some additional arguments in the Hook (see below) and then write my own custom operator.

PostgresHook(
postgres_conn_id=postgres_conn_id,
extra={
"keepalives": 1,
"keepalives_idle": 30,
"keepalives_interval": 5,
"keepalives_count": 5,
}
)

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

OPTION 1: (User-specified hook)

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,
) -> 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)

OPTION 2: (Pass extra params to hook)

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,
) -> 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)

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.

Use case / motivation

This feature will make the PostgresOperator more flexible, allowing users to pass more arguments to the PostgresHook to make better use of its features.

Are you willing to submit a PR?

Yes, I would be happy to submit a PR.

Related Issues

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.