We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
read_sql_table creates delayed objects without provided sqlalchemy connection arguments
I am trying to load a table from presto using Dask `read_sql_table`. The presto server runs on `https`. As per the [pyhive docs](https://github.com/dropbox/PyHive#passing-session-configuration) if server is on SSL you have to pass `{'engine_kwargs': {'connect_args': {'protocol': 'https'}}}`. 

Code:
```python
import dask.dataframe as dd
df = read_sql_table('test_extension', 'presto://server_url/hive',
                      index_col='col_name', divisions=[0, 30000, 60000], head_rows=100,
                       engine_kwargs={'connect_args': {'protocol': 'https'}})
# till here everything is fine it extracts metadata , creates correct number of partitions.
df.head()
```

`df.head` fails saying an http request was made to an https endpoint. I checked this using a debugger as well and the request is made with http protocol. 

I suspect the problem is at https://github.com/dask/dask/blob/6fd2eb57158dbf260e29311e84f6a389904f0ef2/dask/dataframe/io/sql.py#L191, this does not pass engine kwargs to https://github.com/dask/dask/blob/6fd2eb57158dbf260e29311e84f6a389904f0ef2/dask/dataframe/io/sql.py#L196

If the server is on http and I change protocol param to http then everything works.

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

{"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@@ -188,13 +188,21 @@\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)).select_from(table)\n-        parts.append(delayed(_read_sql_chunk)(q, uri, meta, **kwargs))\n+        parts.append(\n+            delayed(_read_sql_chunk)(\n+                q, uri, meta, engine_kwargs=engine_kwargs, **kwargs\n+            )\n+        )\n \n     return from_delayed(parts, meta, divisions=divisions)\n \n \n-def _read_sql_chunk(q, uri, meta, **kwargs):\n-    df = pd.read_sql(q, uri, **kwargs)\n+def _read_sql_chunk(q, uri, meta, engine_kwargs=None, **kwargs):\n+    import sqlalchemy as sa\n+\n+    engine_kwargs = engine_kwargs or {}\n+    conn = sa.create_engine(uri, **engine_kwargs)\n+    df = pd.read_sql(q, conn, **kwargs)\n     if df.empty:\n         return meta\n     else:\n", "issue": "read_sql_table creates delayed objects without provided sqlalchemy connection arguments\nI am trying to load a table from presto using Dask `read_sql_table`. The presto server runs on `https`. As per the [pyhive docs](https://github.com/dropbox/PyHive#passing-session-configuration) if server is on SSL you have to pass `{'engine_kwargs': {'connect_args': {'protocol': 'https'}}}`. \r\n\r\nCode:\r\n```python\r\nimport dask.dataframe as dd\r\ndf = read_sql_table('test_extension', 'presto://server_url/hive',\r\n                      index_col='col_name', divisions=[0, 30000, 60000], head_rows=100,\r\n                       engine_kwargs={'connect_args': {'protocol': 'https'}})\r\n# till here everything is fine it extracts metadata , creates correct number of partitions.\r\ndf.head()\r\n```\r\n\r\n`df.head` fails saying an http request was made to an https endpoint. I checked this using a debugger as well and the request is made with http protocol. \r\n\r\nI suspect the problem is at https://github.com/dask/dask/blob/6fd2eb57158dbf260e29311e84f6a389904f0ef2/dask/dataframe/io/sql.py#L191, this does not pass engine kwargs to https://github.com/dask/dask/blob/6fd2eb57158dbf260e29311e84f6a389904f0ef2/dask/dataframe/io/sql.py#L196\r\n\r\nIf the server is on http and I change protocol param to http then everything works.\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(\n    table,\n    uri,\n    index_col,\n    divisions=None,\n    npartitions=None,\n    limits=None,\n    columns=None,\n    bytes_per_chunk=256 * 2 ** 20,\n    head_rows=5,\n    schema=None,\n    meta=None,\n    engine_kwargs=None,\n    **kwargs\n):\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\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, schema=schema)\n\n    index = (\n        table.columns[index_col] if isinstance(index_col, string_types) else index_col\n    )\n    if not isinstance(index_col, string_types + (elements.Label,)):\n        raise ValueError(\n            \"Use label when passing an SQLAlchemy instance as the index (%s)\" % index\n        )\n    if divisions and npartitions:\n        raise TypeError(\"Must supply either divisions or npartitions, not both\")\n\n    columns = (\n        [(table.columns[c] if isinstance(c, string_types) else c) for c in columns]\n        if columns\n        else list(table.columns)\n    )\n    if index_col not in columns:\n        columns.append(\n            table.columns[index_col]\n            if isinstance(index_col, string_types)\n            else index_col\n        )\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            schema = table.schema\n            head = pd.read_sql_table(name, uri, schema=schema, index_col=index_col)\n            return from_pandas(head, npartitions=1)\n\n        bytes_per_row = (head.memory_usage(deep=True, index=True)).sum() / head_rows\n        meta = head[:0]\n    else:\n        if divisions is None and npartitions is None:\n            raise ValueError(\n                \"Must provide divisions or npartitions when using explicit meta.\"\n            )\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)]).select_from(\n                table\n            )\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\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 = int(round(count * bytes_per_row / bytes_per_chunk)) or 1\n        if dtype.kind == \"M\":\n            divisions = pd.date_range(\n                start=mini,\n                end=maxi,\n                freq=\"%iS\" % ((maxi - mini).total_seconds() / npartitions),\n            ).tolist()\n            divisions[0] = mini\n            divisions[-1] = maxi\n        elif dtype.kind in [\"i\", \"u\", \"f\"]:\n            divisions = np.linspace(mini, maxi, npartitions + 1).tolist()\n        else:\n            raise TypeError(\n                'Provided index column is of type \"{}\".  If divisions is not provided the '\n                \"index column type must be numeric or datetime.\".format(dtype)\n            )\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)).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(\n    table,\n    uri,\n    index_col,\n    divisions=None,\n    npartitions=None,\n    limits=None,\n    columns=None,\n    bytes_per_chunk=256 * 2 ** 20,\n    head_rows=5,\n    schema=None,\n    meta=None,\n    engine_kwargs=None,\n    **kwargs\n):\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\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, schema=schema)\n\n    index = (\n        table.columns[index_col] if isinstance(index_col, string_types) else index_col\n    )\n    if not isinstance(index_col, string_types + (elements.Label,)):\n        raise ValueError(\n            \"Use label when passing an SQLAlchemy instance as the index (%s)\" % index\n        )\n    if divisions and npartitions:\n        raise TypeError(\"Must supply either divisions or npartitions, not both\")\n\n    columns = (\n        [(table.columns[c] if isinstance(c, string_types) else c) for c in columns]\n        if columns\n        else list(table.columns)\n    )\n    if index_col not in columns:\n        columns.append(\n            table.columns[index_col]\n            if isinstance(index_col, string_types)\n            else index_col\n        )\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            schema = table.schema\n            head = pd.read_sql_table(name, uri, schema=schema, index_col=index_col)\n            return from_pandas(head, npartitions=1)\n\n        bytes_per_row = (head.memory_usage(deep=True, index=True)).sum() / head_rows\n        meta = head[:0]\n    else:\n        if divisions is None and npartitions is None:\n            raise ValueError(\n                \"Must provide divisions or npartitions when using explicit meta.\"\n            )\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)]).select_from(\n                table\n            )\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\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 = int(round(count * bytes_per_row / bytes_per_chunk)) or 1\n        if dtype.kind == \"M\":\n            divisions = pd.date_range(\n                start=mini,\n                end=maxi,\n                freq=\"%iS\" % ((maxi - mini).total_seconds() / npartitions),\n            ).tolist()\n            divisions[0] = mini\n            divisions[-1] = maxi\n        elif dtype.kind in [\"i\", \"u\", \"f\"]:\n            divisions = np.linspace(mini, maxi, npartitions + 1).tolist()\n        else:\n            raise TypeError(\n                'Provided index column is of type \"{}\".  If divisions is not provided the '\n                \"index column type must be numeric or datetime.\".format(dtype)\n            )\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)).select_from(table)\n        parts.append(\n            delayed(_read_sql_chunk)(\n                q, uri, meta, engine_kwargs=engine_kwargs, **kwargs\n            )\n        )\n\n    return from_delayed(parts, meta, divisions=divisions)\n\n\ndef _read_sql_chunk(q, uri, meta, engine_kwargs=None, **kwargs):\n    import sqlalchemy as sa\n\n    engine_kwargs = engine_kwargs or {}\n    conn = sa.create_engine(uri, **engine_kwargs)\n    df = pd.read_sql(q, conn, **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"}]}