Add SQL Support for ADBC Drivers

ci/deps/actions-310.yaml

@@ -59,5 +59,7 @@ dependencies:
   - zstandard>=0.17.0
 
   - pip:
+    - adbc_driver_postgresql>=0.5.1
+    - adbc_driver_sqlite>=0.5.1
     - pyqt5>=5.15.6
     - tzdata>=2022.1

ci/deps/actions-311-downstream_compat.yaml

@@ -73,5 +73,7 @@ dependencies:
   - pyyaml
   - py
   - pip:
+    - adbc_driver_postgresql>=0.5.1
+    - adbc_driver_sqlite>=0.5.1
     - pyqt5>=5.15.6
     - tzdata>=2022.1

ci/deps/actions-311.yaml

@@ -59,5 +59,7 @@ dependencies:
   - zstandard>=0.17.0
 
   - pip:
+    - adbc_driver_postgresql>=0.5.1
+    - adbc_driver_sqlite>=0.5.1
     - pyqt5>=5.15.6
     - tzdata>=2022.1

ci/deps/actions-39-minimum_versions.yaml

@@ -61,5 +61,7 @@ dependencies:
   - zstandard=0.17.0
 
   - pip:
+    - adbc_driver_postgresql==0.5.1
+    - adbc_driver_sqlite==0.5.1
     - pyqt5==5.15.6
     - tzdata==2022.1

ci/deps/actions-39.yaml

@@ -59,5 +59,7 @@ dependencies:
   - zstandard>=0.17.0
 
   - pip:
+    - adbc_driver_postgresql>=0.5.1
+    - adbc_driver_sqlite>=0.5.1
     - pyqt5>=5.15.6
     - tzdata>=2022.1

ci/deps/circle-310-arm64.yaml

@@ -58,3 +58,7 @@ dependencies:
   - xlrd>=2.0.1
   - xlsxwriter>=3.0.3
   - zstandard>=0.17.0
+
+  - pip:
+    - adbc_driver_postgresql>=0.5.1
+    - adbc_driver_sqlite>=0.5.1

doc/source/whatsnew/v2.1.0.rst

@@ -105,6 +105,7 @@ Other enhancements
 - Performance improvement in :func:`read_csv` (:issue:`52632`) with ``engine="c"``
 - :meth:`Categorical.from_codes` has gotten a ``validate`` parameter (:issue:`50975`)
 - :meth:`DataFrame.stack` gained the ``sort`` keyword to dictate whether the resulting :class:`MultiIndex` levels are sorted (:issue:`15105`)
+- :meth:`DataFrame.to_sql` and :func:`read_sql` now support ADBC drivers (:issue:`53869`)
 - :meth:`DataFrame.unstack` gained the ``sort`` keyword to dictate whether the resulting :class:`MultiIndex` levels are sorted (:issue:`15105`)
 - :meth:`DataFrameGroupby.agg` and :meth:`DataFrameGroupby.transform` now support grouping by multiple keys when the index is not a :class:`MultiIndex` for ``engine="numba"`` (:issue:`53486`)
 - :meth:`Series.explode` now supports pyarrow-backed list types (:issue:`53602`)

environment.yml

@@ -119,6 +119,8 @@ dependencies:
   - pygments # Code highlighting
 
   - pip:
+      - adbc_driver_postgresql
+      - adbc_driver_sqlite
       - sphinx-toggleprompt
       - typing_extensions; python_version<"3.11"
       - tzdata>=2022.1

pandas/compat/_optional.py

@@ -15,6 +15,8 @@
 # Update install.rst & setup.cfg when updating versions!
 
 VERSIONS = {
+    "adbc_driver_postgresql": "0.5.1",
+    "adbc_driver_sqlite": "0.5.1",
     "bs4": "4.11.1",
     "blosc": "1.21.0",
     "bottleneck": "1.3.4",

pandas/io/sql.py

@@ -629,6 +629,17 @@ def read_sql(
        int_column date_column
     0           0  2012-11-10
     1           1  2010-11-12
+
+    .. versionadded:: 2.1.0
+
+       pandas now supports reading via ADBC drivers
+
+    >>> from adbc_driver_postgresql import dbapi
+    >>> with dbapi.connect('postgres:///db_name') as conn:  # doctest:+SKIP
+    ...     pd.read_sql('SELECT int_column FROM test_data', conn)
+       int_column
+    0           0
+    1           1
     """
 
     check_dtype_backend(dtype_backend)
@@ -837,6 +848,10 @@ def pandasSQL_builder(
     if sqlalchemy is not None and isinstance(con, (str, sqlalchemy.engine.Connectable)):
         return SQLDatabase(con, schema, need_transaction)
 
+    adbc = import_optional_dependency("adbc_driver_manager.dbapi", errors="ignore")
+    if adbc and isinstance(con, adbc.Connection):
+        return ADBCDatabase(con)
+
     warnings.warn(
         "pandas only supports SQLAlchemy connectable (engine/connection) or "
         "database string URI or sqlite3 DBAPI2 connection. Other DBAPI2 "
@@ -2008,6 +2023,188 @@ def _create_sql_schema(
 
 
 # ---- SQL without SQLAlchemy ---
+
+
+class ADBCDatabase(PandasSQL):
+    """
+    This class enables conversion between DataFrame and SQL databases
+    using ADBC to handle DataBase abstraction.
+
+    Parameters
+    ----------
+    con : adbc_driver_manager.dbapi.Connection
+    """
+
+    def __init__(self, con) -> None:
+        self.con = con
+
+    def execute(self, sql: str | Select | TextClause, params=None):
+        with self.con.cursor() as cur:
+            return cur(sql)
+
+    def read_table(
+        self,
+        table_name: str,
+        index_col: str | list[str] | None = None,
+        coerce_float: bool = True,
+        parse_dates=None,
+        columns=None,
+        schema: str | None = None,
+        chunksize: int | None = None,
+        dtype_backend: DtypeBackend | Literal["numpy"] = "numpy",
+    ) -> DataFrame | Iterator[DataFrame]:
+        """
+        Read SQL database table into a DataFrame. Only keyword arguments used
+        are table_name and schema. The rest are silently discarded.
+
+        Parameters
+        ----------
+        table_name : str
+            Name of SQL table in database.
+        schema : string, default None
+            Name of SQL schema in database to read from
+
+        Returns
+        -------
+        DataFrame
+
+        See Also
+        --------
+        pandas.read_sql_table
+        SQLDatabase.read_query
+
+        """
+        if schema:
+            stmt = f"SELECT * FROM {schema}.{table_name}"
+        else:
+            stmt = f"SELECT * FROM {table_name}"
+
+        with self.con.cursor() as cur:
+            return cur(stmt).fetch_arrow_table().to_pandas()
+
+    def read_query(
+        self,
+        sql: str,
+        index_col: str | list[str] | None = None,
+        coerce_float: bool = True,
+        parse_dates=None,
+        params=None,
+        chunksize: int | None = None,
+        dtype: DtypeArg | None = None,
+        dtype_backend: DtypeBackend | Literal["numpy"] = "numpy",
+    ) -> DataFrame | Iterator[DataFrame]:
+        """
+        Read SQL query into a DataFrame. Keyword arguments are discarded.
+
+        Parameters
+        ----------
+        sql : str
+            SQL query to be executed.
+
+        Returns
+        -------
+        DataFrame
+
+        See Also
+        --------
+        read_sql_table : Read SQL database table into a DataFrame.
+        read_sql
+
+        """
+        with self.con.cursor() as cur:
+            return cur(sql).fetch_arrow_table().to_pandas()
+
+    read_sql = read_query
+
+    def to_sql(
+        self,
+        frame,
+        name: str,
+        if_exists: Literal["fail", "replace", "append"] = "fail",
+        index: bool = True,
+        index_label=None,
+        schema: str | None = None,
+        chunksize: int | None = None,
+        dtype: DtypeArg | None = None,
+        method: Literal["multi"] | Callable | None = None,
+        engine: str = "auto",
+        **engine_kwargs,
+    ) -> int | None:
+        """
+        Write records stored in a DataFrame to a SQL database.
+        Only frame, name, if_exists and schema are valid arguments.
+
+        Parameters
+        ----------
+        frame : DataFrame
+        name : string
+            Name of SQL table.
+        if_exists : {'fail', 'replace', 'append'}, default 'fail'
+            - fail: If table exists, do nothing.
+            - replace: If table exists, drop it, recreate it, and insert data.
+            - append: If table exists, insert data. Create if does not exist.
+        schema : string, default None
+            Name of SQL schema in database to write to (if database flavor
+            supports this). If specified, this overwrites the default
+            schema of the SQLDatabase object.
+        """
+        if schema:
+            table_name = f"{schema}.{name}"
+        else:
+            table_name = name
+
+        # TODO: pandas if_exists="append" will still create the
+        # table if it does not exist; ADBC has append/create
+        # as applicable modes, so the semantics get blurred across
+        # the libraries
+        mode = "create"
+        if self.has_table(name, schema):
+            if if_exists == "fail":
+                raise ValueError(f"Table '{table_name}' already exists.")
+            elif if_exists == "replace":
+                with self.con.cursor() as cur:
+                    cur.execute(f"DROP TABLE {table_name}")
+            elif if_exists == "append":
+                mode = "append"
+
+        import pyarrow as pa
+
+        tbl = pa.Table.from_pandas(frame)
+        with self.con.cursor() as cur:
+            total_inserted = cur.adbc_ingest(table_name, tbl, mode=mode)
+
+        self.con.commit()
+        return total_inserted
+
+    def has_table(self, name: str, schema: str | None = None) -> bool:
+        meta = self.con.adbc_get_objects(
+            db_schema_filter=schema, table_name_filter=name
+        ).read_all()
+
+        for catalog_schema in meta["catalog_db_schemas"].to_pylist():
+            if not catalog_schema:
+                continue
+            for schema_record in catalog_schema:
+                if not schema_record:
+                    continue
+
+                for table_record in schema_record["db_schema_tables"]:
+                    if table_record["table_name"] == name:
+                        return True
+
+        return False
+
+    def _create_sql_schema(
+        self,
+        frame: DataFrame,
+        table_name: str,
+        keys: list[str] | None = None,
+        dtype: DtypeArg | None = None,
+        schema: str | None = None,
+    ):
+        raise NotImplementedError("not implemented for adbc")
+
+
 # sqlite-specific sql strings and handler class
 # dictionary used for readability purposes
 _SQL_TYPES = {