def fetch(
    method: str,
    tablename: str,
    json_output_folder: str,
    batch_size: int = 1 << 30,
    processes: int = 1,
    sparkSession: Optional['SparkSession'] = None,
    dbsql: Optional['Connection'] = None,
) -> None:
    """Fetch UC delta table with databricks-connect as JSONL.

    Args:
        method (str): dbconnect or dbsql
        tablename (str): catalog.scheme.tablename on UC
        json_output_folder (str): path to write the result json file to
        batch_size (int): number of rows that dbsql fetches each time to avoid OOM
        processes (int): max number of processes to use to parallelize the fetch
        sparkSession (pyspark.sql.sparksession): spark session
        dbsql (databricks.sql.connect): dbsql session
    """
    cursor = dbsql.cursor() if dbsql is not None else None
    try:
        # Get total rows
        nrows = get_total_rows(tablename, method, cursor, sparkSession)

        # Get columns info
        columns, order_by, columns_str = get_columns_info(
            tablename,
            method,
            cursor,
            sparkSession,
        )

        if method == 'dbconnect' and sparkSession is not None:
            log.info(f'{processes=}')
            df = sparkSession.table(tablename)

            # Running the query and collecting the data as arrow or json.
            signed, _, _ = df.collect_cf('arrow')  # pyright: ignore
            log.info(f'len(signed) = {len(signed)}')

            args = get_args(signed, json_output_folder, columns)

            # Stopping the SparkSession to avoid spilling connection state into the subprocesses.
            sparkSession.stop()

            with ProcessPoolExecutor(max_workers=processes) as executor:
                list(executor.map(download_starargs, args))

        elif method == 'dbsql' and cursor is not None:
            for start in range(0, nrows, batch_size):
                log.warning(f'batch {start}')
                end = min(start + batch_size, nrows)
                fetch_data(
                    method,
                    cursor,
                    sparkSession,
                    start,
                    end,
                    order_by,
                    tablename,
                    columns_str,
                    json_output_folder,
                )

    except Exception as e:
        import grpc
        import pyspark.errors.exceptions.connect as spark_errors
        from databricks.sql.exc import ServerOperationError
        from pyspark.errors import AnalysisException

        if isinstance(e, (AnalysisException, ServerOperationError)):
            error_message = str(e)
            if 'INSUFFICIENT_PERMISSIONS' in error_message:
                raise InsufficientPermissionsError(error_message) from e
            elif 'UC_NOT_ENABLED' in error_message:
                raise UCNotEnabledError() from e
            elif 'UNRESOLVED_COLUMN.WITH_SUGGESTION' in error_message:
                raise MalformedUCTableError(error_message) from e
            elif 'Delta table' in str(e) and "doesn't exist" in str(e):
                # Error processing `catalog`.`volume_name`.`table_name`:
                # Delta table `volume_name`.`table_name` doesn't exist.
                # ---
                parts = error_message.split('`')
                if len(parts) < 7:
                    # Failed to parse error, our codebase is brittle
                    # with respect to the string representations of
                    # errors in the spark library.
                    catalog_name, volume_name, table_name = ['unknown'] * 3
                else:
                    catalog_name = parts[1]
                    volume_name = parts[3]
                    table_name = parts[5]
                raise DeltaTableNotFoundError(
                    catalog_name,
                    volume_name,
                    table_name,
                ) from e
        elif isinstance(
            e,
            (
                spark_errors.SparkConnectGrpcException,
                grpc.RpcError,
                InsufficientPermissionsError,
            ),
        ):
            raise

        # For any other exception, raise a general error
        raise TableDownloadError(tablename, str(e))

    finally:
        if cursor is not None:
            cursor.close()
Fetch UC delta table with databricks-connect as JSONL.

    Args:
        method (str): dbconnect or dbsql
        tablename (str): catalog.scheme.tablename on UC
        json_output_folder (str): path to write the result json file to
        batch_size (int): number of rows that dbsql fetches each time to avoid OOM
        processes (int): max number of processes to use to parallelize the fetch
        sparkSession (pyspark.sql.sparksession): spark session
        dbsql (databricks.sql.connect): dbsql session
    
llmfoundry/command_utils/data_prep/convert_delta_to_json.py