def to_cf(self: 'SparkConnectClient',
          plan: 'pb2.Plan',
          type: str = 'json') -> tuple[list[Result], int, bool]:
    """Executes the query plans and return as presigned URLS for cloud fetch.

    It can handle the current output formats that are supported by the server.
    In contrast to the regular API methods of the client, this method does not
    return the schema and drops all other responses.

    Args:
        self (SparkConnectClient): The SparkConnectClient we are processing.
        plan (pb2.Plan): The plan object to be executed by spark.
        type (str): The output format of the result, supported formats are 'json', 'csv', and 'arrow'.

    Returns:
        Tuple[List[Result], int, bool]: A tuple containing:
            - A list of Result namedtuples, each containing a URL, row count, compressed size,
                and uncompressed size of the part of the result.
            - Total row count of all parts of the result.
            - A boolean indicating whether the result has been truncated.
    """
    req = self._execute_plan_request_with_metadata()
    req.plan.CopyFrom(plan)

    import pyspark.sql.connect.proto as pb2
    import pyspark.sql.connect.proto.cloud_pb2 as cloud_pb2

    # Add the request options
    if type == 'json':
        format = cloud_pb2.ResultOptions.CloudOptions.FORMAT_JSON
    elif type == 'csv':
        format = cloud_pb2.ResultOptions.CloudOptions.FORMAT_CSV
    elif type == 'arrow':
        format = cloud_pb2.ResultOptions.CloudOptions.FORMAT_ARROW
    else:
        raise ValueError(
            f'Only formats json, csv, and arrow are supported. Got invalid type {type}',
        )

    ro = cloud_pb2.ResultOptions(
        type=cloud_pb2.ResultOptions.TYPE_CLOUD,
        cloudOptions=cloud_pb2.ResultOptions.CloudOptions(
            format=format,
            useCompression=False,
        ),
    )
    cloud_option = any_pb2.Any()
    cloud_option.Pack(ro)
    req.request_options.append(
        pb2.ExecutePlanRequest.RequestOption(extension=cloud_option),
    )

    # Create the iterator
    from pyspark.sql.connect.client.reattach import (
        ExecutePlanResponseReattachableIterator,
    )
    iterator = ExecutePlanResponseReattachableIterator(
        req,
        self._stub,
        self._retry_policy,
        self._builder.metadata(),
    )
    # Iterate over the response
    result = []
    row_count = 0
    is_overflow = False

    for response in iterator:
        if response.HasField('extension') and response.extension.Is(
            cloud_pb2.CloudResultBatch.DESCRIPTOR,
        ):
            batch = cloud_pb2.CloudResultBatch()
            if not response.extension.Is(cloud_pb2.CloudResultBatch.DESCRIPTOR):
                raise ValueError(
                    'Response extension is not of type CloudResultBatch.',
                )
            response.extension.Unpack(batch)
            result += [
                Result(
                    b.url,
                    b.row_count,
                    b.compressed_size,
                    b.uncompressed_size,
                ) for b in batch.results
            ]
            row_count += sum(result.row_count for result in batch.results)
            is_overflow |= batch.truncated
    return result, row_count, is_overflow
Executes the query plans and return as presigned URLS for cloud fetch.

    It can handle the current output formats that are supported by the server.
    In contrast to the regular API methods of the client, this method does not
    return the schema and drops all other responses.

    Args:
        self (SparkConnectClient): The SparkConnectClient we are processing.
        plan (pb2.Plan): The plan object to be executed by spark.
        type (str): The output format of the result, supported formats are 'json', 'csv', and 'arrow'.

    Returns:
        Tuple[List[Result], int, bool]: A tuple containing:
            - A list of Result namedtuples, each containing a URL, row count, compressed size,
                and uncompressed size of the part of the result.
            - Total row count of all parts of the result.
            - A boolean indicating whether the result has been truncated.
    
llmfoundry/command_utils/data_prep/convert_delta_to_json.py