Issue 1:
Apache Airflow version

2.6.1

What happened

When i use do_xcom_push=True in DatabricksSqlOperator the an exception with following stack trace is thrown:

[2023-06-06, 08:52:24 UTC] {sql.py:375} INFO - Running statement: SELECT cast(max(id) as STRING) FROM prod.unified.sessions, parameters: None
[2023-06-06, 08:52:25 UTC] {taskinstance.py:1824} ERROR - Task failed with exception
Traceback (most recent call last):
File "/home/airflow/.local/lib/python3.10/site-packages/airflow/utils/session.py", line 73, in wrapper
return func(*args, **kwargs)
File "/home/airflow/.local/lib/python3.10/site-packages/airflow/models/taskinstance.py", line 2354, in xcom_push
XCom.set(
File "/home/airflow/.local/lib/python3.10/site-packages/airflow/utils/session.py", line 73, in wrapper
return func(*args, **kwargs)
File "/home/airflow/.local/lib/python3.10/site-packages/airflow/models/xcom.py", line 237, in set
value = cls.serialize_value(
File "/home/airflow/.local/lib/python3.10/site-packages/airflow/models/xcom.py", line 632, in serialize_value
return json.dumps(value, cls=XComEncoder).encode("UTF-8")
File "/usr/local/lib/python3.10/json/__init__.py", line 238, in dumps
**kw).encode(obj)
File "/home/airflow/.local/lib/python3.10/site-packages/airflow/utils/json.py", line 102, in encode
o = self.default(o)
File "/home/airflow/.local/lib/python3.10/site-packages/airflow/utils/json.py", line 91, in default
return serialize(o)
File "/home/airflow/.local/lib/python3.10/site-packages/airflow/serialization/serde.py", line 144, in serialize
return encode(classname, version, serialize(data, depth + 1))
File "/home/airflow/.local/lib/python3.10/site-packages/airflow/serialization/serde.py", line 123, in serialize
return [serialize(d, depth + 1) for d in o]
File "/home/airflow/.local/lib/python3.10/site-packages/airflow/serialization/serde.py", line 123, in <listcomp>
return [serialize(d, depth + 1) for d in o]
File "/home/airflow/.local/lib/python3.10/site-packages/airflow/serialization/serde.py", line 132, in serialize
qn = qualname(o)
File "/home/airflow/.local/lib/python3.10/site-packages/airflow/utils/module_loading.py", line 47, in qualname
return f"{o.__module__}.{o.__name__}"
File "/home/airflow/.local/lib/python3.10/site-packages/databricks/sql/types.py", line 161, in __getattr__
raise AttributeError(item)
AttributeError: __name__. Did you mean: '__ne__'?

What you think should happen instead

In _process_output() if self._output_path is False a list of tuples is returned:

def _process_output(self, results: list[Any], descriptions: list[Sequence[Sequence] | None]) -> list[Any]:
if not self._output_path:
return list(zip(descriptions, results))

I suspect this breaks the serialization somehow which might be related to my own meta database(postgres).

Replacing the Databricks SQL Operator with simple PythonOperator and DatabricksSqlHook works just fine:

def get_max_id(ti):
hook = DatabricksSqlHook(databricks_conn_id=databricks_sql_conn_id, sql_endpoint_name='sql_endpoint')
sql = "SELECT cast(max(id) as STRING) FROM prod.unified.sessions"
return str(hook.get_first(sql)[0])

How to reproduce

get_max_id_task = DatabricksSqlOperator(
databricks_conn_id=databricks_sql_conn_id,
sql_endpoint_name='sql_endpoint',
task_id='get_max_id',
sql="SELECT cast(max(id) as STRING) FROM prod.unified.sessions",
do_xcom_push=True
)

Operating System

Debian GNU/Linux 11 (bullseye) docker image, python 3.10

Versions of Apache Airflow Providers

apache-airflow-providers-common-sql==1.5.1

databricks-sql-connector==2.5.2

apache-airflow-providers-databricks==4.2.0

Deployment

Docker-Compose

Deployment details

Using extended Airflow image, LocalExecutor, Postgres 13 meta db as container in the same stack.

docker-compose version 1.29.2, build 5becea4c

Docker version 23.0.5, build bc4487a

Anything else

No response

Are you willing to submit PR?

Yes I am willing to submit a PR!

Code of Conduct

I agree to follow this project's Code of Conduct

Issue 2:
Apache Airflow version

2.6.2

What happened

I installed apache-airflow and attempted to set up the db via:

airflow db init

but it fails with the following pydantic error:

DB: sqlite:////Users/username/airflow/airflow.db
/Users/username/Desktop/project_root/.venv/lib/python3.9/site-packages/pydantic/_internal/_config.py:257 UserWarning: Valid config keys have changed in V2:
* 'orm_mode' has been renamed to 'from_attributes'
Traceback (most recent call last):
File "Users/username/Desktop/project_root/.venv/bin/airflow", line 8, in <module>
sys.exit(main())
File "Users/username/Desktop/project_root/.venv/lib/python3.9/site-packages/airflow/__main__.py", line 48, in main
args.func(args)
File "Users/username/Desktop/project_root/.venv/lib/python3.9/site-packages/airflow/cli/cli_config.py", line 52, in command
return func(*args, **kwargs)
File "Users/username/Desktop/project_root/.venv/lib/python3.9/site-packages/airflow/cli/commands/db_command.py", line 37, in initdb
db.initdb()
File "Users/username/Desktop/project_root/.venv/lib/python3.9/site-packages/airflow/utils/session.py", line 76, in wrapper
return func(*args, session=session, **kwargs)
File "Users/username/Desktop/project_root/.venv/lib/python3.9/site-packages/airflow/utils/db.py", line 711, in initdb
import_all_models()
File "Users/username/Desktop/project_root/.venv/lib/python3.9/site-packages/airflow/models/__init__.py", line 60, in import_all_models
__getattr__(name)
File "Users/username/Desktop/project_root/.venv/lib/python3.9/site-packages/airflow/models/__init__.py", line 78, in __getattr__
val = import_string(f"{path}.{name}")
File "/Users/username/Desktop/project_root/.venv/lib/python3.9/site-packages/airflow/utils/module_loading.py", line 36, in import_string
module = import_module(module_path)
File "/opt/homebrew/Cellar/python@3.9/3.9.17/Frameworks/Python.framework/Versions/3.9/lib/python3.9/importlib/__init__.py", line 127, in import_module
return _bootstrap._gcd_import(name[level:], package, level)
File "<frozen importlib._bootstrap>", line 1030, in _gcd_import
File "<frozen importlib._bootstrap>", line 1007, in _find_and_load
File "<frozen importlib._bootstrap>", line 986, in _find_and_load_unlocked
File "<frozen importlib._bootstrap>", line 680, in _load_unlocked
File "<frozen importlib._bootstrap_external>", line 850, in exec_module
File "<frozen importlib._bootstrap>", line 228, in _call_with_frames_removed
File "/Users/username/Desktop/project_root/.venv/lib/python3.9/site-packages/airflow/models/skipmixin.py", line 25, in <module>
from airflow.serialization.pydantic.dag_run import DagRunPydantic
File "/Users/username/Desktop/project_root/.venv/lib/python3.9/site-packages/airflow/serialization/pydantic/dag_run.py", line 24, in <module>
from airflow.serialization.pydantic.dataset import DatasetEventPydantic
File "/Users/username/Desktop/project_root/.venv/lib/python3.9/site-packages/airflow/serialization/pydantic/dataset.py", line 40, in <module>
class TaskOutletDatasetReferencePydantic(BaseModelPydantic):
File "/Users/username/Desktop/project_root/.venv/lib/python3.9/site-packages/pydantic/_internal/_model_construction.py", line 95, in __new__
private_attributes = inspect_namespace(
File "/Users/username/Desktop/project_root/.venv/lib/python3.9/site-packages/pydantic/_internal/_model_construction.py", line 328, in inspect_namespace
raise PydanticUserError(
pydantic.errors.PydanticUserError: A non-annotated attribute was detected: `dag_id = <class 'str'>`. All model fields require a type annotation; if `dag_id` is not meant to be a field, you may be able to resolve this error by annotating it as a `ClassVar` or updating `model_config['ignored_types']`.

How to reproduce

install apache-airflow via pip

open terminal, cd to ~/airflow folder

run airflow db init command

Operating System

MacOS Ventura 13.3.1.

Versions of Apache Airflow Providers

apache-airflow-providers-common-sql==1.5.2

apache-airflow-providers-ftp==3.4.2

apache-airflow-providers-http==4.4.2

apache-airflow-providers-imap==3.2.2

apache-airflow-providers-sqlite==3.4.2

Are you willing to submit PR?

Yes I am willing to submit a PR!

Code of Conduct

I agree to follow this project's Code of Conduct