We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
read_sql_table gives AttributeError when reading a database
When I try to read from a sql database:
`df_database = create_engine('sqlite:///' + database_path)`
`df = dd.read_sql_table('table', df_database, 'id_driver')`

I get this error: `AttributeError: 'Engine' object has no attribute '_instantiate_plugins'`

If I use the pandas `read_sql_query` function on the same db, it works just fine.

Here's a full trace:
> AttributeError Traceback (most recent call last)
> ipython-input-10-2da58b7e1773 in <module>()
>       3 t0 = time.time()
> ----> 4 df = dd.read_sql_table('table', df_database, 'id_driver')
>       5 t1 = time.time()
>       6 
> 
> ~/lib/python3.6/site-packages/dask/dataframe/io/sql.py in read_sql_table(table, uri, index_col, divisions, npartitions, limits, columns, bytes_per_chunk, **kwargs)
>      66     if index_col is None:
>      67         raise ValueError("Must specify index column to partition on")
> ---> 68     engine = sa.create_engine(uri)
>      69     meta = sa.MetaData()
>      70     if isinstance(table, six.string_types):
> 
> ~/lib/python3.6/site-packages/sqlalchemy/engine/__init__.py in create_engine(*args, **kwargs)
>     389     strategy = kwargs.pop('strategy', default_strategy)
>     390     strategy = strategies.strategies[strategy]
> --> 391     return strategy.create(*args, **kwargs)
>     392 
>     393 
> 
> ~/lib/python3.6/site-packages/sqlalchemy/engine/strategies.py in create(self, name_or_url, **kwargs)
>      50         u = url.make_url(name_or_url)
>      51 
> ---> 52         plugins = u._instantiate_plugins(kwargs)
>      53 
>      54         u.query.pop('plugin', None)
> 
> AttributeError: 'Engine' object has no attribute '_instantiate_plugins'

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `dask/dataframe/io/sql.py`
Content:
```
1 import numpy as np
2 import pandas as pd
3 
4 from ... import delayed
5 from ...compatibility import string_types
6 from .io import from_delayed, from_pandas
7 
8 
9 def read_sql_table(table, uri, index_col, divisions=None, npartitions=None,
10                    limits=None, columns=None, bytes_per_chunk=256 * 2**20,
11                    head_rows=5, schema=None, meta=None, **kwargs):
12     """
13     Create dataframe from an SQL table.
14 
15     If neither divisions or npartitions is given, the memory footprint of the
16     first few rows will be determined, and partitions of size ~256MB will
17     be used.
18 
19     Parameters
20     ----------
21     table : string or sqlalchemy expression
22         Select columns from here.
23     uri : string
24         Full sqlalchemy URI for the database connection
25     index_col : string
26         Column which becomes the index, and defines the partitioning. Should
27         be a indexed column in the SQL server, and any orderable type. If the
28         type is number or time, then partition boundaries can be inferred from
29         npartitions or bytes_per_chunk; otherwide must supply explicit
30         ``divisions=``.
31         ``index_col`` could be a function to return a value, e.g.,
32         ``sql.func.abs(sql.column('value')).label('abs(value)')``.
33         Labeling columns created by functions or arithmetic operations is
34         required.
35     divisions: sequence
36         Values of the index column to split the table by. If given, this will
37         override npartitions and bytes_per_chunk. The divisions are the value
38         boundaries of the index column used to define the partitions. For
39         example, ``divisions=list('acegikmoqsuwz')`` could be used to partition
40         a string column lexographically into 12 partitions, with the implicit
41         assumption that each partition contains similar numbers of records.
42     npartitions : int
43         Number of partitions, if divisions is not given. Will split the values
44         of the index column linearly between limits, if given, or the column
45         max/min. The index column must be numeric or time for this to work
46     limits: 2-tuple or None
47         Manually give upper and lower range of values for use with npartitions;
48         if None, first fetches max/min from the DB. Upper limit, if
49         given, is inclusive.
50     columns : list of strings or None
51         Which columns to select; if None, gets all; can include sqlalchemy
52         functions, e.g.,
53         ``sql.func.abs(sql.column('value')).label('abs(value)')``.
54         Labeling columns created by functions or arithmetic operations is
55         recommended.
56     bytes_per_chunk: int
57         If both divisions and npartitions is None, this is the target size of
58         each partition, in bytes
59     head_rows: int
60         How many rows to load for inferring the data-types, unless passing meta
61     meta: empty DataFrame or None
62         If provided, do not attempt to infer dtypes, but use these, coercing
63         all chunks on load
64     schema: str or None
65         If using a table name, pass this to sqlalchemy to select which DB
66         schema to use within the URI connection
67     kwargs : dict
68         Additional parameters to pass to `pd.read_sql()`
69 
70     Returns
71     -------
72     dask.dataframe
73 
74     Examples
75     --------
76     >>> df = dd.read_sql_table('accounts', 'sqlite:///path/to/bank.db',
77     ...                  npartitions=10, index_col='id')  # doctest: +SKIP
78     """
79     import sqlalchemy as sa
80     from sqlalchemy import sql
81     from sqlalchemy.sql import elements
82     if index_col is None:
83         raise ValueError("Must specify index column to partition on")
84     engine = sa.create_engine(uri)
85     m = sa.MetaData()
86     if isinstance(table, string_types):
87         table = sa.Table(table, m, autoload=True, autoload_with=engine,
88                          schema=schema)
89 
90     index = (table.columns[index_col] if isinstance(index_col, string_types)
91              else index_col)
92     if not isinstance(index_col, string_types + (elements.Label,)):
93         raise ValueError('Use label when passing an SQLAlchemy instance'
94                          ' as the index (%s)' % index)
95     if divisions and npartitions:
96         raise TypeError('Must supply either divisions or npartitions, not both')
97 
98     columns = ([(table.columns[c] if isinstance(c, string_types) else c)
99                 for c in columns]
100                if columns else list(table.columns))
101     if index_col not in columns:
102         columns.append(table.columns[index_col]
103                        if isinstance(index_col, string_types)
104                        else index_col)
105 
106     if isinstance(index_col, string_types):
107         kwargs['index_col'] = index_col
108     else:
109         # function names get pandas auto-named
110         kwargs['index_col'] = index_col.name
111 
112     if meta is None:
113         # derrive metadata from first few rows
114         q = sql.select(columns).limit(head_rows).select_from(table)
115         head = pd.read_sql(q, engine, **kwargs)
116 
117         if head.empty:
118             # no results at all
119             name = table.name
120             head = pd.read_sql_table(name, uri, index_col=index_col)
121             return from_pandas(head, npartitions=1)
122 
123         bytes_per_row = (head.memory_usage(deep=True, index=True)).sum() / 5
124         meta = head[:0]
125     else:
126         if divisions is None and npartitions is None:
127             raise ValueError('Must provide divisions or npartitions when'
128                              'using explicit meta.')
129 
130     if divisions is None:
131         if limits is None:
132             # calculate max and min for given index
133             q = sql.select([sql.func.max(index), sql.func.min(index)]
134                            ).select_from(table)
135             minmax = pd.read_sql(q, engine)
136             maxi, mini = minmax.iloc[0]
137             dtype = minmax.dtypes['max_1']
138         else:
139             mini, maxi = limits
140             dtype = pd.Series(limits).dtype
141         if npartitions is None:
142             q = sql.select([sql.func.count(index)]).select_from(table)
143             count = pd.read_sql(q, engine)['count_1'][0]
144             npartitions = round(count * bytes_per_row / bytes_per_chunk) or 1
145         if dtype.kind == "M":
146             divisions = pd.date_range(
147                 start=mini, end=maxi, freq='%iS' % (
148                     (maxi - mini) / npartitions).total_seconds()).tolist()
149             divisions[0] = mini
150             divisions[-1] = maxi
151         else:
152             divisions = np.linspace(mini, maxi, npartitions + 1).tolist()
153 
154     parts = []
155     lowers, uppers = divisions[:-1], divisions[1:]
156     for i, (lower, upper) in enumerate(zip(lowers, uppers)):
157         cond = index <= upper if i == len(lowers) - 1 else index < upper
158         q = sql.select(columns).where(sql.and_(index >= lower, cond)
159                                       ).select_from(table)
160         parts.append(delayed(_read_sql_chunk)(q, uri, meta, **kwargs))
161 
162     return from_delayed(parts, meta, divisions=divisions)
163 
164 
165 def _read_sql_chunk(q, uri, meta, **kwargs):
166     df = pd.read_sql(q, uri, **kwargs)
167     if df.empty:
168         return meta
169     else:
170         return df.astype(meta.dtypes.to_dict(), copy=False)
171
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/dask/dataframe/io/sql.py b/dask/dataframe/io/sql.py
--- a/dask/dataframe/io/sql.py
+++ b/dask/dataframe/io/sql.py
@@ -8,7 +8,7 @@
 
 def read_sql_table(table, uri, index_col, divisions=None, npartitions=None,
                    limits=None, columns=None, bytes_per_chunk=256 * 2**20,
-                   head_rows=5, schema=None, meta=None, **kwargs):
+                   head_rows=5, schema=None, meta=None, engine_kwargs=None, **kwargs):
     """
     Create dataframe from an SQL table.
 
@@ -53,17 +53,19 @@
         ``sql.func.abs(sql.column('value')).label('abs(value)')``.
         Labeling columns created by functions or arithmetic operations is
         recommended.
-    bytes_per_chunk: int
+    bytes_per_chunk : int
         If both divisions and npartitions is None, this is the target size of
         each partition, in bytes
-    head_rows: int
+    head_rows : int
         How many rows to load for inferring the data-types, unless passing meta
-    meta: empty DataFrame or None
+    meta : empty DataFrame or None
         If provided, do not attempt to infer dtypes, but use these, coercing
         all chunks on load
-    schema: str or None
+    schema : str or None
         If using a table name, pass this to sqlalchemy to select which DB
         schema to use within the URI connection
+    engine_kwargs : dict or None
+        Specific db engine parameters for sqlalchemy
     kwargs : dict
         Additional parameters to pass to `pd.read_sql()`
 
@@ -81,7 +83,8 @@
     from sqlalchemy.sql import elements
     if index_col is None:
         raise ValueError("Must specify index column to partition on")
-    engine = sa.create_engine(uri)
+    engine_kwargs = {} if engine_kwargs is None else engine_kwargs
+    engine = sa.create_engine(uri, **engine_kwargs)
     m = sa.MetaData()
     if isinstance(table, string_types):
         table = sa.Table(table, m, autoload=True, autoload_with=engine,

{"golden_diff": "diff --git a/dask/dataframe/io/sql.py b/dask/dataframe/io/sql.py\n--- a/dask/dataframe/io/sql.py\n+++ b/dask/dataframe/io/sql.py\n@@ -8,7 +8,7 @@\n \n def read_sql_table(table, uri, index_col, divisions=None, npartitions=None,\n                    limits=None, columns=None, bytes_per_chunk=256 * 2**20,\n-                   head_rows=5, schema=None, meta=None, **kwargs):\n+                   head_rows=5, schema=None, meta=None, engine_kwargs=None, **kwargs):\n     \"\"\"\n     Create dataframe from an SQL table.\n \n@@ -53,17 +53,19 @@\n         ``sql.func.abs(sql.column('value')).label('abs(value)')``.\n         Labeling columns created by functions or arithmetic operations is\n         recommended.\n-    bytes_per_chunk: int\n+    bytes_per_chunk : int\n         If both divisions and npartitions is None, this is the target size of\n         each partition, in bytes\n-    head_rows: int\n+    head_rows : int\n         How many rows to load for inferring the data-types, unless passing meta\n-    meta: empty DataFrame or None\n+    meta : empty DataFrame or None\n         If provided, do not attempt to infer dtypes, but use these, coercing\n         all chunks on load\n-    schema: str or None\n+    schema : str or None\n         If using a table name, pass this to sqlalchemy to select which DB\n         schema to use within the URI connection\n+    engine_kwargs : dict or None\n+        Specific db engine parameters for sqlalchemy\n     kwargs : dict\n         Additional parameters to pass to `pd.read_sql()`\n \n@@ -81,7 +83,8 @@\n     from sqlalchemy.sql import elements\n     if index_col is None:\n         raise ValueError(\"Must specify index column to partition on\")\n-    engine = sa.create_engine(uri)\n+    engine_kwargs = {} if engine_kwargs is None else engine_kwargs\n+    engine = sa.create_engine(uri, **engine_kwargs)\n     m = sa.MetaData()\n     if isinstance(table, string_types):\n         table = sa.Table(table, m, autoload=True, autoload_with=engine,\n", "issue": "read_sql_table gives AttributeError when reading a database\nWhen I try to read from a sql database:\r\n`df_database = create_engine('sqlite:///' + database_path)`\r\n`df = dd.read_sql_table('table', df_database, 'id_driver')`\r\n\r\nI get this error: `AttributeError: 'Engine' object has no attribute '_instantiate_plugins'`\r\n\r\nIf I use the pandas `read_sql_query` function on the same db, it works just fine.\r\n\r\nHere's a full trace:\r\n> AttributeError Traceback (most recent call last)\r\n> ipython-input-10-2da58b7e1773 in <module>()\r\n>       3 t0 = time.time()\r\n> ----> 4 df = dd.read_sql_table('table', df_database, 'id_driver')\r\n>       5 t1 = time.time()\r\n>       6 \r\n> \r\n> ~/lib/python3.6/site-packages/dask/dataframe/io/sql.py in read_sql_table(table, uri, index_col, divisions, npartitions, limits, columns, bytes_per_chunk, **kwargs)\r\n>      66     if index_col is None:\r\n>      67         raise ValueError(\"Must specify index column to partition on\")\r\n> ---> 68     engine = sa.create_engine(uri)\r\n>      69     meta = sa.MetaData()\r\n>      70     if isinstance(table, six.string_types):\r\n> \r\n> ~/lib/python3.6/site-packages/sqlalchemy/engine/__init__.py in create_engine(*args, **kwargs)\r\n>     389     strategy = kwargs.pop('strategy', default_strategy)\r\n>     390     strategy = strategies.strategies[strategy]\r\n> --> 391     return strategy.create(*args, **kwargs)\r\n>     392 \r\n>     393 \r\n> \r\n> ~/lib/python3.6/site-packages/sqlalchemy/engine/strategies.py in create(self, name_or_url, **kwargs)\r\n>      50         u = url.make_url(name_or_url)\r\n>      51 \r\n> ---> 52         plugins = u._instantiate_plugins(kwargs)\r\n>      53 \r\n>      54         u.query.pop('plugin', None)\r\n> \r\n> AttributeError: 'Engine' object has no attribute '_instantiate_plugins'\n", "before_files": [{"content": "import numpy as np\nimport pandas as pd\n\nfrom ... import delayed\nfrom ...compatibility import string_types\nfrom .io import from_delayed, from_pandas\n\n\ndef read_sql_table(table, uri, index_col, divisions=None, npartitions=None,\n                   limits=None, columns=None, bytes_per_chunk=256 * 2**20,\n                   head_rows=5, schema=None, meta=None, **kwargs):\n    \"\"\"\n    Create dataframe from an SQL table.\n\n    If neither divisions or npartitions is given, the memory footprint of the\n    first few rows will be determined, and partitions of size ~256MB will\n    be used.\n\n    Parameters\n    ----------\n    table : string or sqlalchemy expression\n        Select columns from here.\n    uri : string\n        Full sqlalchemy URI for the database connection\n    index_col : string\n        Column which becomes the index, and defines the partitioning. Should\n        be a indexed column in the SQL server, and any orderable type. If the\n        type is number or time, then partition boundaries can be inferred from\n        npartitions or bytes_per_chunk; otherwide must supply explicit\n        ``divisions=``.\n        ``index_col`` could be a function to return a value, e.g.,\n        ``sql.func.abs(sql.column('value')).label('abs(value)')``.\n        Labeling columns created by functions or arithmetic operations is\n        required.\n    divisions: sequence\n        Values of the index column to split the table by. If given, this will\n        override npartitions and bytes_per_chunk. The divisions are the value\n        boundaries of the index column used to define the partitions. For\n        example, ``divisions=list('acegikmoqsuwz')`` could be used to partition\n        a string column lexographically into 12 partitions, with the implicit\n        assumption that each partition contains similar numbers of records.\n    npartitions : int\n        Number of partitions, if divisions is not given. Will split the values\n        of the index column linearly between limits, if given, or the column\n        max/min. The index column must be numeric or time for this to work\n    limits: 2-tuple or None\n        Manually give upper and lower range of values for use with npartitions;\n        if None, first fetches max/min from the DB. Upper limit, if\n        given, is inclusive.\n    columns : list of strings or None\n        Which columns to select; if None, gets all; can include sqlalchemy\n        functions, e.g.,\n        ``sql.func.abs(sql.column('value')).label('abs(value)')``.\n        Labeling columns created by functions or arithmetic operations is\n        recommended.\n    bytes_per_chunk: int\n        If both divisions and npartitions is None, this is the target size of\n        each partition, in bytes\n    head_rows: int\n        How many rows to load for inferring the data-types, unless passing meta\n    meta: empty DataFrame or None\n        If provided, do not attempt to infer dtypes, but use these, coercing\n        all chunks on load\n    schema: str or None\n        If using a table name, pass this to sqlalchemy to select which DB\n        schema to use within the URI connection\n    kwargs : dict\n        Additional parameters to pass to `pd.read_sql()`\n\n    Returns\n    -------\n    dask.dataframe\n\n    Examples\n    --------\n    >>> df = dd.read_sql_table('accounts', 'sqlite:///path/to/bank.db',\n    ...                  npartitions=10, index_col='id')  # doctest: +SKIP\n    \"\"\"\n    import sqlalchemy as sa\n    from sqlalchemy import sql\n    from sqlalchemy.sql import elements\n    if index_col is None:\n        raise ValueError(\"Must specify index column to partition on\")\n    engine = sa.create_engine(uri)\n    m = sa.MetaData()\n    if isinstance(table, string_types):\n        table = sa.Table(table, m, autoload=True, autoload_with=engine,\n                         schema=schema)\n\n    index = (table.columns[index_col] if isinstance(index_col, string_types)\n             else index_col)\n    if not isinstance(index_col, string_types + (elements.Label,)):\n        raise ValueError('Use label when passing an SQLAlchemy instance'\n                         ' as the index (%s)' % index)\n    if divisions and npartitions:\n        raise TypeError('Must supply either divisions or npartitions, not both')\n\n    columns = ([(table.columns[c] if isinstance(c, string_types) else c)\n                for c in columns]\n               if columns else list(table.columns))\n    if index_col not in columns:\n        columns.append(table.columns[index_col]\n                       if isinstance(index_col, string_types)\n                       else index_col)\n\n    if isinstance(index_col, string_types):\n        kwargs['index_col'] = index_col\n    else:\n        # function names get pandas auto-named\n        kwargs['index_col'] = index_col.name\n\n    if meta is None:\n        # derrive metadata from first few rows\n        q = sql.select(columns).limit(head_rows).select_from(table)\n        head = pd.read_sql(q, engine, **kwargs)\n\n        if head.empty:\n            # no results at all\n            name = table.name\n            head = pd.read_sql_table(name, uri, index_col=index_col)\n            return from_pandas(head, npartitions=1)\n\n        bytes_per_row = (head.memory_usage(deep=True, index=True)).sum() / 5\n        meta = head[:0]\n    else:\n        if divisions is None and npartitions is None:\n            raise ValueError('Must provide divisions or npartitions when'\n                             'using explicit meta.')\n\n    if divisions is None:\n        if limits is None:\n            # calculate max and min for given index\n            q = sql.select([sql.func.max(index), sql.func.min(index)]\n                           ).select_from(table)\n            minmax = pd.read_sql(q, engine)\n            maxi, mini = minmax.iloc[0]\n            dtype = minmax.dtypes['max_1']\n        else:\n            mini, maxi = limits\n            dtype = pd.Series(limits).dtype\n        if npartitions is None:\n            q = sql.select([sql.func.count(index)]).select_from(table)\n            count = pd.read_sql(q, engine)['count_1'][0]\n            npartitions = round(count * bytes_per_row / bytes_per_chunk) or 1\n        if dtype.kind == \"M\":\n            divisions = pd.date_range(\n                start=mini, end=maxi, freq='%iS' % (\n                    (maxi - mini) / npartitions).total_seconds()).tolist()\n            divisions[0] = mini\n            divisions[-1] = maxi\n        else:\n            divisions = np.linspace(mini, maxi, npartitions + 1).tolist()\n\n    parts = []\n    lowers, uppers = divisions[:-1], divisions[1:]\n    for i, (lower, upper) in enumerate(zip(lowers, uppers)):\n        cond = index <= upper if i == len(lowers) - 1 else index < upper\n        q = sql.select(columns).where(sql.and_(index >= lower, cond)\n                                      ).select_from(table)\n        parts.append(delayed(_read_sql_chunk)(q, uri, meta, **kwargs))\n\n    return from_delayed(parts, meta, divisions=divisions)\n\n\ndef _read_sql_chunk(q, uri, meta, **kwargs):\n    df = pd.read_sql(q, uri, **kwargs)\n    if df.empty:\n        return meta\n    else:\n        return df.astype(meta.dtypes.to_dict(), copy=False)\n", "path": "dask/dataframe/io/sql.py"}], "after_files": [{"content": "import numpy as np\nimport pandas as pd\n\nfrom ... import delayed\nfrom ...compatibility import string_types\nfrom .io import from_delayed, from_pandas\n\n\ndef read_sql_table(table, uri, index_col, divisions=None, npartitions=None,\n                   limits=None, columns=None, bytes_per_chunk=256 * 2**20,\n                   head_rows=5, schema=None, meta=None, engine_kwargs=None, **kwargs):\n    \"\"\"\n    Create dataframe from an SQL table.\n\n    If neither divisions or npartitions is given, the memory footprint of the\n    first few rows will be determined, and partitions of size ~256MB will\n    be used.\n\n    Parameters\n    ----------\n    table : string or sqlalchemy expression\n        Select columns from here.\n    uri : string\n        Full sqlalchemy URI for the database connection\n    index_col : string\n        Column which becomes the index, and defines the partitioning. Should\n        be a indexed column in the SQL server, and any orderable type. If the\n        type is number or time, then partition boundaries can be inferred from\n        npartitions or bytes_per_chunk; otherwide must supply explicit\n        ``divisions=``.\n        ``index_col`` could be a function to return a value, e.g.,\n        ``sql.func.abs(sql.column('value')).label('abs(value)')``.\n        Labeling columns created by functions or arithmetic operations is\n        required.\n    divisions: sequence\n        Values of the index column to split the table by. If given, this will\n        override npartitions and bytes_per_chunk. The divisions are the value\n        boundaries of the index column used to define the partitions. For\n        example, ``divisions=list('acegikmoqsuwz')`` could be used to partition\n        a string column lexographically into 12 partitions, with the implicit\n        assumption that each partition contains similar numbers of records.\n    npartitions : int\n        Number of partitions, if divisions is not given. Will split the values\n        of the index column linearly between limits, if given, or the column\n        max/min. The index column must be numeric or time for this to work\n    limits: 2-tuple or None\n        Manually give upper and lower range of values for use with npartitions;\n        if None, first fetches max/min from the DB. Upper limit, if\n        given, is inclusive.\n    columns : list of strings or None\n        Which columns to select; if None, gets all; can include sqlalchemy\n        functions, e.g.,\n        ``sql.func.abs(sql.column('value')).label('abs(value)')``.\n        Labeling columns created by functions or arithmetic operations is\n        recommended.\n    bytes_per_chunk : int\n        If both divisions and npartitions is None, this is the target size of\n        each partition, in bytes\n    head_rows : int\n        How many rows to load for inferring the data-types, unless passing meta\n    meta : empty DataFrame or None\n        If provided, do not attempt to infer dtypes, but use these, coercing\n        all chunks on load\n    schema : str or None\n        If using a table name, pass this to sqlalchemy to select which DB\n        schema to use within the URI connection\n    engine_kwargs : dict or None\n        Specific db engine parameters for sqlalchemy\n    kwargs : dict\n        Additional parameters to pass to `pd.read_sql()`\n\n    Returns\n    -------\n    dask.dataframe\n\n    Examples\n    --------\n    >>> df = dd.read_sql_table('accounts', 'sqlite:///path/to/bank.db',\n    ...                  npartitions=10, index_col='id')  # doctest: +SKIP\n    \"\"\"\n    import sqlalchemy as sa\n    from sqlalchemy import sql\n    from sqlalchemy.sql import elements\n    if index_col is None:\n        raise ValueError(\"Must specify index column to partition on\")\n    engine_kwargs = {} if engine_kwargs is None else engine_kwargs\n    engine = sa.create_engine(uri, **engine_kwargs)\n    m = sa.MetaData()\n    if isinstance(table, string_types):\n        table = sa.Table(table, m, autoload=True, autoload_with=engine,\n                         schema=schema)\n\n    index = (table.columns[index_col] if isinstance(index_col, string_types)\n             else index_col)\n    if not isinstance(index_col, string_types + (elements.Label,)):\n        raise ValueError('Use label when passing an SQLAlchemy instance'\n                         ' as the index (%s)' % index)\n    if divisions and npartitions:\n        raise TypeError('Must supply either divisions or npartitions, not both')\n\n    columns = ([(table.columns[c] if isinstance(c, string_types) else c)\n                for c in columns]\n               if columns else list(table.columns))\n    if index_col not in columns:\n        columns.append(table.columns[index_col]\n                       if isinstance(index_col, string_types)\n                       else index_col)\n\n    if isinstance(index_col, string_types):\n        kwargs['index_col'] = index_col\n    else:\n        # function names get pandas auto-named\n        kwargs['index_col'] = index_col.name\n\n    if meta is None:\n        # derrive metadata from first few rows\n        q = sql.select(columns).limit(head_rows).select_from(table)\n        head = pd.read_sql(q, engine, **kwargs)\n\n        if head.empty:\n            # no results at all\n            name = table.name\n            head = pd.read_sql_table(name, uri, index_col=index_col)\n            return from_pandas(head, npartitions=1)\n\n        bytes_per_row = (head.memory_usage(deep=True, index=True)).sum() / 5\n        meta = head[:0]\n    else:\n        if divisions is None and npartitions is None:\n            raise ValueError('Must provide divisions or npartitions when'\n                             'using explicit meta.')\n\n    if divisions is None:\n        if limits is None:\n            # calculate max and min for given index\n            q = sql.select([sql.func.max(index), sql.func.min(index)]\n                           ).select_from(table)\n            minmax = pd.read_sql(q, engine)\n            maxi, mini = minmax.iloc[0]\n            dtype = minmax.dtypes['max_1']\n        else:\n            mini, maxi = limits\n            dtype = pd.Series(limits).dtype\n        if npartitions is None:\n            q = sql.select([sql.func.count(index)]).select_from(table)\n            count = pd.read_sql(q, engine)['count_1'][0]\n            npartitions = round(count * bytes_per_row / bytes_per_chunk) or 1\n        if dtype.kind == \"M\":\n            divisions = pd.date_range(\n                start=mini, end=maxi, freq='%iS' % (\n                    (maxi - mini) / npartitions).total_seconds()).tolist()\n            divisions[0] = mini\n            divisions[-1] = maxi\n        else:\n            divisions = np.linspace(mini, maxi, npartitions + 1).tolist()\n\n    parts = []\n    lowers, uppers = divisions[:-1], divisions[1:]\n    for i, (lower, upper) in enumerate(zip(lowers, uppers)):\n        cond = index <= upper if i == len(lowers) - 1 else index < upper\n        q = sql.select(columns).where(sql.and_(index >= lower, cond)\n                                      ).select_from(table)\n        parts.append(delayed(_read_sql_chunk)(q, uri, meta, **kwargs))\n\n    return from_delayed(parts, meta, divisions=divisions)\n\n\ndef _read_sql_chunk(q, uri, meta, **kwargs):\n    df = pd.read_sql(q, uri, **kwargs)\n    if df.empty:\n        return meta\n    else:\n        return df.astype(meta.dtypes.to_dict(), copy=False)\n", "path": "dask/dataframe/io/sql.py"}]}