Add support for Server Side Cursors (a.k.a. stream results)

[jlynchMicron]

Is your feature request related to a problem? Please describe.
Please support the sqlalchemy stream_results cursor feature to break up large queries that would overwhelm a systems memory.

• sqlalchemy documentation: https://docs.sqlalchemy.org/en/14/core/connections.html#using-server-side-cursors-a-k-a-stream-results
• bigquery similar function: https://cloud.google.com/bigquery/docs/paging-results

Describe the solution you'd like
An implementation similar to this post: https://pythonspeed.com/articles/pandas-sql-chunking/

import pandas as pd
from sqlalchemy import create_engine

def process_sql_using_pandas():
    engine = create_engine(
        "postgresql://postgres:pass@localhost/example"
    )
    conn = engine.connect().execution_options(
        stream_results=True)

    for chunk_dataframe in pd.read_sql(
            "SELECT * FROM users", conn, chunksize=1000):
        print(f"Got dataframe w/{len(chunk_dataframe)} rows")
        # ... do something with dataframe ...

if __name__ == '__main__':
    process_sql_using_pandas()

Describe alternatives you've considered
Currently I have to perform an upload after every SELECT DISTINCT query I do to make sure my next query "chunk" will not grab the same row elements.

Related Issue in Pandas: pandas-dev/pandas#35689

[jlynchMicron]

It appears that this feature actually works and is implemented by this function:

python-bigquery-sqlalchemy/sqlalchemy_bigquery/base.py

Line 117 in 4b05d21

def create_cursor(self):

Someone please correct me if im wrong. I believe I have it working in a project I am using, but I do not have a good way to verify that my machine is not pulling down an entire query first, then just iterating through it.

[tswast]

The DB-API works as you suggest, pulling down only a page at a time as needed, but I'm unsure how this interacts with the SQLAlchemy connector. Can keep this open for further investigation.

[balajivenkatesh]

@jlynchMicron Would you be able to share an example of using create_cursor to read in chunks?

[jlynchMicron]

@balajivenkatesh This is what I think you roughly need to do and create_cursor happens under the hood, in this pandas example:

conn:sa.engine.Connection
    with engine.connect() as conn:
        conn.execution_options(stream_results=True)
        df:pd.DataFrame
        for df in pd.read_sql(sql, engine, chunksize=query_chunksize):
            #Read chunks here