code: def _read_sql_chunk(q, uri, meta, engine_kwargs=None, **kwargs):
      import sqlalchemy as sa
  
      engine_kwargs = engine_kwargs or {}
      engine = sa.create_engine(uri, **engine_kwargs)
      df = pd.read_sql(q, engine, **kwargs)
      engine.dispose()
      if len(df) == 0:
          return meta
      elif len(meta.dtypes.to_dict()) == 0:
          # only index column in loaded
          # required only for pandas < 1.0.0
          return df
      else:
          return df.astype(meta.dtypes.to_dict(), copy=False)
  
file_path: dask/dask/dataframe/io/sql.py
imports: from __future__ import annotations
  import warnings
  import numpy
  import pandas
  from dask.base import compute
  from dask.dataframe import methods
  from dask.dataframe.io.io import from_delayed, from_pandas
  from dask.dataframe.utils import pyarrow_strings_enabled
  from dask.delayed import delayed, tokenize
  from dask.utils import parse_bytes
label_desc: Use image node_id 4 for calling a global function with example usage: _read_sql_chunk(q, uri, meta, engine_kwargs) and returns: meta, df, df