<h3 dir="auto">Apache Airflow version</h3>
<p dir="auto">2.6.1</p>
<h3 dir="auto">What happened</h3>
<p dir="auto">When i use <em>do_xcom_push=True</em> in <strong>DatabricksSqlOperator</strong> the an exception with following stack trace is thrown:</p>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="[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 &quot;/home/airflow/.local/lib/python3.10/site-packages/airflow/utils/session.py&quot;, line 73, in wrapper
    return func(*args, **kwargs)
  File &quot;/home/airflow/.local/lib/python3.10/site-packages/airflow/models/taskinstance.py&quot;, line 2354, in xcom_push
    XCom.set(
  File &quot;/home/airflow/.local/lib/python3.10/site-packages/airflow/utils/session.py&quot;, line 73, in wrapper
    return func(*args, **kwargs)
  File &quot;/home/airflow/.local/lib/python3.10/site-packages/airflow/models/xcom.py&quot;, line 237, in set
    value = cls.serialize_value(
  File &quot;/home/airflow/.local/lib/python3.10/site-packages/airflow/models/xcom.py&quot;, line 632, in serialize_value
    return json.dumps(value, cls=XComEncoder).encode(&quot;UTF-8&quot;)
  File &quot;/usr/local/lib/python3.10/json/__init__.py&quot;, line 238, in dumps
    **kw).encode(obj)
  File &quot;/home/airflow/.local/lib/python3.10/site-packages/airflow/utils/json.py&quot;, line 102, in encode
    o = self.default(o)
  File &quot;/home/airflow/.local/lib/python3.10/site-packages/airflow/utils/json.py&quot;, line 91, in default
    return serialize(o)
  File &quot;/home/airflow/.local/lib/python3.10/site-packages/airflow/serialization/serde.py&quot;, line 144, in serialize
    return encode(classname, version, serialize(data, depth + 1))
  File &quot;/home/airflow/.local/lib/python3.10/site-packages/airflow/serialization/serde.py&quot;, line 123, in serialize
    return [serialize(d, depth + 1) for d in o]
  File &quot;/home/airflow/.local/lib/python3.10/site-packages/airflow/serialization/serde.py&quot;, line 123, in &lt;listcomp&gt;
    return [serialize(d, depth + 1) for d in o]
  File &quot;/home/airflow/.local/lib/python3.10/site-packages/airflow/serialization/serde.py&quot;, line 132, in serialize
    qn = qualname(o)
  File &quot;/home/airflow/.local/lib/python3.10/site-packages/airflow/utils/module_loading.py&quot;, line 47, in qualname
    return f&quot;{o.__module__}.{o.__name__}&quot;
  File &quot;/home/airflow/.local/lib/python3.10/site-packages/databricks/sql/types.py&quot;, line 161, in __getattr__
    raise AttributeError(item)
AttributeError: __name__. Did you mean: '__ne__'?"><pre class="notranslate"><code class="notranslate">[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 &lt;listcomp&gt;
    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__'?
</code></pre></div>
<h3 dir="auto">What you think should happen instead</h3>
<p dir="auto">In _process_output() if self._output_path is False a list of tuples is returned:</p>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content=" def _process_output(self, results: list[Any], descriptions: list[Sequence[Sequence] | None]) -&gt; list[Any]:
    if not self._output_path:
        return list(zip(descriptions, results))"><pre class="notranslate"><code class="notranslate"> def _process_output(self, results: list[Any], descriptions: list[Sequence[Sequence] | None]) -&gt; list[Any]:
    if not self._output_path:
        return list(zip(descriptions, results))
</code></pre></div>
<p dir="auto">I suspect this breaks the serialization somehow which might be related to my own meta database(postgres).</p>
<p dir="auto">Replacing the Databricks SQL Operator with simple <strong>PythonOperator</strong> and <strong>DatabricksSqlHook</strong> works just fine:</p>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="def get_max_id(ti):
    hook = DatabricksSqlHook(databricks_conn_id=databricks_sql_conn_id, sql_endpoint_name='sql_endpoint')
    sql = &quot;SELECT cast(max(id) as STRING) FROM prod.unified.sessions&quot;
    return str(hook.get_first(sql)[0])"><pre class="notranslate"><code class="notranslate">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])
</code></pre></div>
<h3 dir="auto">How to reproduce</h3>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="    get_max_id_task = DatabricksSqlOperator(
        databricks_conn_id=databricks_sql_conn_id,
        sql_endpoint_name='sql_endpoint',
        task_id='get_max_id',
        sql=&quot;SELECT cast(max(id) as STRING) FROM prod.unified.sessions&quot;,
        do_xcom_push=True
    )"><pre class="notranslate"><code class="notranslate">    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
    )
</code></pre></div>
<h3 dir="auto">Operating System</h3>
<p dir="auto">Debian GNU/Linux 11 (bullseye) docker image, python 3.10</p>
<h3 dir="auto">Versions of Apache Airflow Providers</h3>
<p dir="auto">apache-airflow-providers-common-sql==1.5.1<br>
databricks-sql-connector==2.5.2<br>
apache-airflow-providers-databricks==4.2.0</p>
<h3 dir="auto">Deployment</h3>
<p dir="auto">Docker-Compose</p>
<h3 dir="auto">Deployment details</h3>
<p dir="auto">Using extended Airflow image, LocalExecutor, Postgres 13 meta db as container in the same stack.<br>
docker-compose version 1.29.2, build 5becea4c<br>
Docker version 23.0.5, build bc4487a</p>
<h3 dir="auto">Anything else</h3>
<p dir="auto"><em>No response</em></p>
<h3 dir="auto">Are you willing to submit PR?</h3>
<ul class="contains-task-list">
<li class="task-list-item"><input type="checkbox" id="" disabled="" class="task-list-item-checkbox"> Yes I am willing to submit a PR!</li>
</ul>
<h3 dir="auto">Code of Conduct</h3>
<ul class="contains-task-list">
<li class="task-list-item"><input type="checkbox" id="" disabled="" class="task-list-item-checkbox" checked=""> I agree to follow this project's <a href="https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md">Code of Conduct</a></li>
</ul>
<h3 dir="auto">Apache Airflow version</h3>
<p dir="auto">2.6.2</p>
<h3 dir="auto">What happened</h3>
<p dir="auto">I installed <code class="notranslate">apache-airflow</code> and attempted to set up the db via:</p>
<p dir="auto"><code class="notranslate">airflow db init</code></p>
<p dir="auto">but it fails with the following pydantic error:</p>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="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 &quot;Users/username/Desktop/project_root/.venv/bin/airflow&quot;, line 8, in &lt;module&gt;
    sys.exit(main())
  File &quot;Users/username/Desktop/project_root/.venv/lib/python3.9/site-packages/airflow/__main__.py&quot;, line 48, in main
    args.func(args)
  File &quot;Users/username/Desktop/project_root/.venv/lib/python3.9/site-packages/airflow/cli/cli_config.py&quot;, line 52, in command
    return func(*args, **kwargs)
  File &quot;Users/username/Desktop/project_root/.venv/lib/python3.9/site-packages/airflow/cli/commands/db_command.py&quot;, line 37, in initdb
    db.initdb()
  File &quot;Users/username/Desktop/project_root/.venv/lib/python3.9/site-packages/airflow/utils/session.py&quot;, line 76, in wrapper
    return func(*args, session=session, **kwargs)
  File &quot;Users/username/Desktop/project_root/.venv/lib/python3.9/site-packages/airflow/utils/db.py&quot;, line 711, in initdb
    import_all_models()
  File &quot;Users/username/Desktop/project_root/.venv/lib/python3.9/site-packages/airflow/models/__init__.py&quot;, line 60, in import_all_models
    __getattr__(name)
  File &quot;Users/username/Desktop/project_root/.venv/lib/python3.9/site-packages/airflow/models/__init__.py&quot;, line 78, in __getattr__
    val = import_string(f&quot;{path}.{name}&quot;)
  File &quot;/Users/username/Desktop/project_root/.venv/lib/python3.9/site-packages/airflow/utils/module_loading.py&quot;, line 36, in import_string
    module = import_module(module_path)
  File &quot;/opt/homebrew/Cellar/python@3.9/3.9.17/Frameworks/Python.framework/Versions/3.9/lib/python3.9/importlib/__init__.py&quot;, line 127, in import_module
    return _bootstrap._gcd_import(name[level:], package, level)
  File &quot;&lt;frozen importlib._bootstrap&gt;&quot;, line 1030, in _gcd_import
  File &quot;&lt;frozen importlib._bootstrap&gt;&quot;, line 1007, in _find_and_load
  File &quot;&lt;frozen importlib._bootstrap&gt;&quot;, line 986, in _find_and_load_unlocked
  File &quot;&lt;frozen importlib._bootstrap&gt;&quot;, line 680, in _load_unlocked
  File &quot;&lt;frozen importlib._bootstrap_external&gt;&quot;, line 850, in exec_module
  File &quot;&lt;frozen importlib._bootstrap&gt;&quot;, line 228, in _call_with_frames_removed
  File &quot;/Users/username/Desktop/project_root/.venv/lib/python3.9/site-packages/airflow/models/skipmixin.py&quot;, line 25, in &lt;module&gt;
    from airflow.serialization.pydantic.dag_run import DagRunPydantic
  File &quot;/Users/username/Desktop/project_root/.venv/lib/python3.9/site-packages/airflow/serialization/pydantic/dag_run.py&quot;, line 24, in &lt;module&gt;
    from airflow.serialization.pydantic.dataset import DatasetEventPydantic
  File &quot;/Users/username/Desktop/project_root/.venv/lib/python3.9/site-packages/airflow/serialization/pydantic/dataset.py&quot;, line 40, in &lt;module&gt;
    class TaskOutletDatasetReferencePydantic(BaseModelPydantic):
  File &quot;/Users/username/Desktop/project_root/.venv/lib/python3.9/site-packages/pydantic/_internal/_model_construction.py&quot;, line 95, in __new__
    private_attributes = inspect_namespace(
  File &quot;/Users/username/Desktop/project_root/.venv/lib/python3.9/site-packages/pydantic/_internal/_model_construction.py&quot;, line 328, in inspect_namespace
    raise PydanticUserError(
pydantic.errors.PydanticUserError: A non-annotated attribute was detected: `dag_id = &lt;class 'str'&gt;`. 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']`."><pre class="notranslate"><code class="notranslate">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 &lt;module&gt;
    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 "&lt;frozen importlib._bootstrap&gt;", line 1030, in _gcd_import
  File "&lt;frozen importlib._bootstrap&gt;", line 1007, in _find_and_load
  File "&lt;frozen importlib._bootstrap&gt;", line 986, in _find_and_load_unlocked
  File "&lt;frozen importlib._bootstrap&gt;", line 680, in _load_unlocked
  File "&lt;frozen importlib._bootstrap_external&gt;", line 850, in exec_module
  File "&lt;frozen importlib._bootstrap&gt;", 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 &lt;module&gt;
    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 &lt;module&gt;
    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 &lt;module&gt;
    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 = &lt;class 'str'&gt;`. 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']`.
</code></pre></div>
<h3 dir="auto">How to reproduce</h3>
<ol dir="auto">
<li>install <code class="notranslate">apache-airflow</code> via pip</li>
<li>open terminal, cd to <code class="notranslate">~/airflow</code> folder</li>
<li>run <code class="notranslate">airflow db init</code> command</li>
</ol>
<h3 dir="auto">Operating System</h3>
<p dir="auto">MacOS Ventura 13.3.1.</p>
<h3 dir="auto">Versions of Apache Airflow Providers</h3>
<p dir="auto"><strong>apache-airflow-providers-common-sql==1.5.2<br>
apache-airflow-providers-ftp==3.4.2<br>
apache-airflow-providers-http==4.4.2<br>
apache-airflow-providers-imap==3.2.2<br>
apache-airflow-providers-sqlite==3.4.2</strong></p>
<h3 dir="auto">Are you willing to submit PR?</h3>
<ul class="contains-task-list">
<li class="task-list-item"><input type="checkbox" id="" disabled="" class="task-list-item-checkbox"> Yes I am willing to submit a PR!</li>
</ul>
<h3 dir="auto">Code of Conduct</h3>
<ul class="contains-task-list">
<li class="task-list-item"><input type="checkbox" id="" disabled="" class="task-list-item-checkbox" checked=""> I agree to follow this project's <a href="https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md">Code of Conduct</a></li>
</ul>