I/O#

@PublicAPI
def from_arrow(
    data: Union["pa.Table", list["pa.Table"], Iterable["pa.Table"], ArrowStreamExportable],
) -> "DataFrame":
    """Creates a DataFrame from Arrow data.

    Accepts pyarrow Tables, lists/iterables of pyarrow Tables, or any object
    implementing the `Arrow PyCapsule Interface <https://arrow.apache.org/docs/format/CDataInterface/PyCapsuleInterface>`
    (i.e. has an ``__arrow_c_stream__`` method). This includes pyarrow RecordBatchReaders,
    pandas DataFrames (2.2+), nanoarrow arrays, and other Arrow-compatible libraries.

    Args:
        data: Arrow data to convert into a Daft DataFrame.

    Returns:
        DataFrame: DataFrame created from the provided Arrow data.

    Examples:
        >>> import pyarrow as pa
        >>> import daft
        >>> t = pa.table({"a": [1, 2, 3], "b": ["foo", "bar", "baz"]})
        >>> df = daft.from_arrow(t)
        >>> df.show()
        ╭───────┬────────╮
        │ a     ┆ b      │
        │ ---   ┆ ---    │
        │ Int64 ┆ String │
        ╞═══════╪════════╡
        │ 1     ┆ foo    │
        ├╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌┤
        │ 2     ┆ bar    │
        ├╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌┤
        │ 3     ┆ baz    │
        ╰───────┴────────╯
        <BLANKLINE>
        (Showing first 3 of 3 rows)
    """
    from daft import DataFrame

    # pa.Table implements __arrow_c_stream__ but we prefer the pyarrow-aware path
    # because it handles types the Rust FFI stream cannot (e.g. extension types, Decimal256).
    if not isinstance(data, pa.Table) and isinstance(data, ArrowStreamExportable):
        return DataFrame._from_arrow_stream(data)

    return DataFrame._from_arrow(data)

@PublicAPI
def from_dask_dataframe(ddf: "dask.DataFrame") -> "DataFrame":
    """Creates a Daft DataFrame from a Dask DataFrame.

    The provided Dask DataFrame must have been created using [Dask-on-Ray](https://docs.ray.io/en/latest/ray-more-libs/dask-on-ray.html).

    Args:
        ddf: The Dask DataFrame to create a Daft DataFrame from.

    Returns:
        DataFrame: Daft DataFrame created from the provided Dask DataFrame.

    Note:
        This function can only work if Daft is running using the RayRunner

    Examples:
        >>> import dask.dataframe as dd
        >>> import pandas as pd
        >>> import daft
        >>> import ray
        >>>
        >>> daft.set_runner_ray()  # doctest: +SKIP
        >>>
        >>> ddf = dd.from_pandas(pd.DataFrame({"a": [1, 2], "b": ["foo", "bar"]}), npartitions=2)  # doctest: +SKIP
        >>> df = daft.from_dask_dataframe(ddf)  # doctest: +SKIP
        >>> df.show()  # doctest: +SKIP
        ╭───────┬────────╮
        │ a     ┆ b      │
        │ ---   ┆ ---    │
        │ Int64 ┆ String │
        ╞═══════╪════════╡
        │ 1     ┆ foo    │
        ├╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌┤
        │ 2     ┆ bar    │
        ╰───────┴────────╯
        <BLANKLINE>
        (Showing first 2 of 2 rows)

    """
    from daft import DataFrame

    return DataFrame._from_dask_dataframe(ddf)

@PublicAPI
def from_glob_path(path: str | list[str], io_config: IOConfig | None = None) -> DataFrame:
    """Creates a DataFrame of file paths and other metadata from a glob path.

    This method supports wildcards:

    1. `*` matches any number of any characters including none
    2. `?` matches any single character
    3. `[...]` matches any single character in the brackets
    4. `**` recursively matches any number of layers of directories

    The returned DataFrame will have the following columns:

    1. path: the path to the file/directory
    2. size: size of the object in bytes
    3. rows: the total rows of parquet object, it's None for other formats.

    Args:
        path (str|list): Path to files on disk (allows wildcards).
        io_config (IOConfig): Configuration to use when running IO with remote services

    Returns:
        DataFrame: DataFrame containing the path to each file as a row, along with other metadata parsed from the provided filesystem.

    Note:
        If no files match the glob pattern(s), an empty DataFrame is returned instead of raising an error.

    Examples:
        >>> df = daft.from_glob_path("/path/to/files/*.jpeg")
        >>> df = daft.from_glob_path("/path/to/files/**/*.jpeg")
        >>> df = daft.from_glob_path("/path/to/files/**/image-?.jpeg")
        >>> df = daft.from_glob_path(["/path/to/files/*.jpeg", "/path/to/others/*.jpeg"])
    """
    if isinstance(path, str):
        path = [path]

    if len(path) == 0:
        raise ValueError("Must specify at least one glob path")

    context = get_context()
    io_config = context.daft_planning_config.default_io_config if io_config is None else io_config

    builder = LogicalPlanBuilder.from_glob_scan(
        glob_paths=path,
        io_config=io_config,
    )
    return DataFrame(builder)

@PublicAPI
def from_pandas(data: Union["pd.DataFrame", list["pd.DataFrame"]]) -> "DataFrame":
    """Creates a Daft DataFrame from a pandas DataFrame.

    Args:
        data: pandas DataFrame(s) that we wish to convert into a Daft DataFrame.

    Returns:
        DataFrame: Daft DataFrame created from the provided pandas DataFrame.

    Examples:
        >>> import pandas as pd
        >>> import daft
        >>> pd_df = pd.DataFrame({"a": [1, 2, 3], "b": ["foo", "bar", "baz"]})
        >>> df = daft.from_pandas(pd_df)
        >>> df.show()
        ╭───────┬────────╮
        │ a     ┆ b      │
        │ ---   ┆ ---    │
        │ Int64 ┆ String │
        ╞═══════╪════════╡
        │ 1     ┆ foo    │
        ├╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌┤
        │ 2     ┆ bar    │
        ├╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌┤
        │ 3     ┆ baz    │
        ╰───────┴────────╯
        <BLANKLINE>
        (Showing first 3 of 3 rows)
    """
    from daft import DataFrame

    return DataFrame._from_pandas(data)

@PublicAPI
def from_pydict(data: dict[str, InputListType]) -> "DataFrame":
    """Creates a DataFrame from a Python dictionary.

    Args:
        data: Key -> Sequence[item] of data. Each Key is created as a column, and must have a value that is
            a Python list, Numpy array or PyArrow array. Values must be equal in length across all keys.

    Returns:
        DataFrame: DataFrame created from dictionary of columns

    Examples:
        >>> import daft
        >>> df = daft.from_pydict({"foo": [1, 2]})
        >>> df.show()
        ╭───────╮
        │ foo   │
        │ ---   │
        │ Int64 │
        ╞═══════╡
        │ 1     │
        ├╌╌╌╌╌╌╌┤
        │ 2     │
        ╰───────╯
        <BLANKLINE>
        (Showing first 2 of 2 rows)
    """
    from daft import DataFrame

    return DataFrame._from_pydict(data)

@PublicAPI
def from_pylist(data: list[dict[str, Any]]) -> "DataFrame":
    """Creates a DataFrame from a list of dictionaries.

    Args:
        data: List of dictionaries, where each key is a column name.

    Returns:
        DataFrame: DataFrame created from list of dictionaries.

    Examples:
        >>> import daft
        >>> df = daft.from_pylist([{"foo": 1}, {"foo": 2}])
        >>> df.show()
        ╭───────╮
        │ foo   │
        │ ---   │
        │ Int64 │
        ╞═══════╡
        │ 1     │
        ├╌╌╌╌╌╌╌┤
        │ 2     │
        ╰───────╯
        <BLANKLINE>
        (Showing first 2 of 2 rows)
    """
    from daft import DataFrame

    return DataFrame._from_pylist(data)

@PublicAPI
def from_ray_dataset(ds: "RayDataset") -> "DataFrame":
    """Creates a DataFrame from a Ray Dataset.

    Args:
        ds: The Ray Dataset to create a Daft DataFrame from.

    Returns:
        DataFrame: Daft DataFrame created from the provided Ray dataset.

    Note:
        This function can only work if Daft is running using the RayRunner

    Examples:
        >>> import ray
        >>> import daft
        >>>
        >>> daft.set_runner_ray()  # doctest: +SKIP
        >>>
        >>> ds = ray.data.from_items([{"a": 1, "b": "foo"}, {"a": 2, "b": "bar"}])  # doctest: +SKIP
        >>> df = daft.from_ray_dataset(ds)  # doctest: +SKIP
        >>> df.show()  # doctest: +SKIP
        ╭───────┬────────╮
        │ a     ┆ b      │
        │ ---   ┆ ---    │
        │ Int64 ┆ String │
        ╞═══════╪════════╡
        │ 1     ┆ foo    │
        ├╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌┤
        │ 2     ┆ bar    │
        ╰───────┴────────╯
        <BLANKLINE>
        (Showing first 2 of 2 rows)

    """
    from daft import DataFrame

    return DataFrame._from_ray_dataset(ds)

@PublicAPI
def read_csv(
    path: str | list[str],
    infer_schema: bool = True,
    schema: dict[str, DataType] | None = None,
    has_headers: bool = True,
    delimiter: str | None = None,
    double_quote: bool = True,
    quote: str | None = None,
    escape_char: str | None = None,
    comment: str | None = None,
    allow_variable_columns: bool = False,
    io_config: IOConfig | None = None,
    file_path_column: str | None = None,
    hive_partitioning: bool = False,
    _buffer_size: int | None = None,
    _chunk_size: int | None = None,
    checkpoint: "CheckpointConfig | None" = None,
) -> DataFrame:
    """Creates a DataFrame from CSV file(s).

    Args:
        path (str): Path to CSV (allows for wildcards; supports remote URLs to object stores such as ``s3://`` or ``gs://``)
        infer_schema (bool): Whether to infer the schema of the CSV, defaults to True.
        schema (dict[str, DataType]): A schema that is used as the definitive schema for the CSV if infer_schema is False, otherwise it is used as a schema hint that is applied after the schema is inferred (overriding the types of inferred columns, and appending any new columns not found during inference).
        has_headers (bool): Whether the CSV has a header or not, defaults to True
        delimiter (Str): Delimiter used in the CSV, defaults to ","
        double_quote (bool): Whether to support double quote escapes, defaults to True
        escape_char (str): Character to use as the escape character for double quotes, or defaults to `"`
        comment (str): Character to treat as the start of a comment line, or None to not support comments
        allow_variable_columns (bool): Whether to allow for variable number of columns in the CSV, defaults to False. If set to True, Daft will append nulls to rows with less columns than the schema, and ignore extra columns in rows with more columns
        io_config (IOConfig): Config to be used with the native downloader
        file_path_column: Include the source path(s) as a column with this name. Defaults to None.
        hive_partitioning: Whether to infer hive_style partitions from file paths and include them as columns in the Dataframe. Defaults to False.
        checkpoint: Optional :class:`daft.CheckpointConfig` for progress tracking across runs. Bundles the
            checkpoint store, the source key column (``on=``), and optional anti-join tuning. Rows whose key
            already exists in the store are skipped on re-run. Requires the Ray runner.

    Returns:
        DataFrame: parsed DataFrame

    Examples:
        Read a CSV file from a local path:
        >>> df = daft.read_csv("/path/to/file.csv")
        >>> df = daft.read_csv("/path/to/directory")
        >>> df = daft.read_csv("/path/to/files-*.csv")

        Read a CSV file from a public S3 bucket:
        >>> from daft.io import S3Config, IOConfig
        >>> io_config = IOConfig(s3=S3Config(region="us-west-2", anonymous=True))
        >>> df = daft.read_csv("s3://path/to/files-*.csv", io_config=io_config)
        >>> df.show()
    """
    if isinstance(path, list) and len(path) == 0:
        raise ValueError("Cannot read DataFrame from empty list of CSV filepaths")

    if not infer_schema and schema is None:
        raise ValueError(
            "Cannot read DataFrame with infer_schema=False and schema=None, please provide a schema or set infer_schema=True"
        )

    io_config = context.get_context().daft_planning_config.default_io_config if io_config is None else io_config

    csv_config = CsvSourceConfig(
        delimiter=delimiter,
        has_headers=has_headers,
        double_quote=double_quote,
        quote=quote,
        escape_char=escape_char,
        comment=comment,
        allow_variable_columns=allow_variable_columns,
        buffer_size=_buffer_size,
        chunk_size=_chunk_size,
    )
    file_format_config = FileFormatConfig.from_csv_config(csv_config)
    storage_config = StorageConfig(True, io_config)

    builder = get_tabular_files_scan(
        path=path,
        infer_schema=infer_schema,
        schema=schema,
        file_format_config=file_format_config,
        storage_config=storage_config,
        file_path_column=file_path_column,
        hive_partitioning=hive_partitioning,
    )
    builder = attach_checkpoint(builder, checkpoint)
    return DataFrame(builder)

@PublicAPI
def read_deltalake(
    table: Union[str, "UnityCatalogTable"],
    version: Union[int, str, "datetime"] | None = None,
    io_config: IOConfig | None = None,
    ignore_deletion_vectors: bool = False,
    _multithreaded_io: bool | None = None,
) -> DataFrame:
    """Create a DataFrame from a Delta Lake table.

    Args:
        table: Either a URI for the Delta Lake table (supports remote URLs to object stores such as ``s3://`` or ``gs://``)
            or a ``UnityCatalogTable`` instance from a Unity Catalog client.
        version (optional): If int is passed, read the table with specified version number. Otherwise if string or datetime,
            read the timestamp version of the table. Strings must be RFC 3339 and ISO 8601 date and time format.
            Datetimes are assumed to be UTC timezone unless specified. By default, read the latest version of the table.
        io_config (optional): A custom :class:`~daft.daft.IOConfig` to use when accessing Delta Lake object storage data. Defaults to None.
        ignore_deletion_vectors (optional): Whether to skip checking for deletion vectors when reading the table. Defaults to False.
        _multithreaded_io (optional): Whether to use multithreading for IO threads. Setting this to False can be helpful in reducing
            the amount of system resources (number of connections and thread contention) when running in the Ray runner.
            Defaults to None, which will let Daft decide based on the runner it is currently using.

    Returns:
        DataFrame: A DataFrame with the schema converted from the specified Delta Lake table.

    Note:
        This function requires the use of [deltalake](https://delta-io.github.io/delta-rs/), a Python library for interacting with Delta Lake.

    Examples:
        Read a Delta Lake table from a local path:
        >>> df = daft.read_deltalake("some-table-uri")
        >>>
        >>> # Filters on this dataframe can now be pushed into the read operation from Delta Lake.
        >>> df = df.where(df["foo"] > 5)
        >>> df.show()

        Read a Delta Lake table from a public S3 bucket:
        >>> from daft.io import S3Config, IOConfig
        >>> io_config = IOConfig(s3=S3Config(region="us-west-2", anonymous=True))
        >>> df = daft.read_deltalake("s3://daft-oss-public-data/test_fixtures/delta_table/", io_config=io_config)
        >>> df.show()
    """
    from daft.io.delta_lake.delta_lake_scan import DeltaLakeScanOperator

    # If running on Ray, we want to limit the amount of concurrency and requests being made.
    # This is because each Ray worker process receives its own pool of thread workers and connections
    multithreaded_io = (
        (runners.get_or_create_runner().name != "ray") if _multithreaded_io is None else _multithreaded_io
    )

    io_config = context.get_context().daft_planning_config.default_io_config if io_config is None else io_config
    storage_config = StorageConfig(multithreaded_io, io_config)

    if isinstance(table, str):
        table_uri = os.path.expanduser(table)
    elif unity_catalog.module_available() and isinstance(table, unity_catalog.UnityCatalogTable):
        table_uri = table.table_uri

        # Override the storage_config with the one provided by Unity catalog
        recordbatch_io_config = table.io_config
        if recordbatch_io_config is not None:
            storage_config = StorageConfig(multithreaded_io, recordbatch_io_config)
    else:
        raise ValueError(
            f"table argument must be a table URI string or UnityCatalogTable instance, but got: {type(table)}, {table}"
        )
    delta_lake_operator = DeltaLakeScanOperator(
        table_uri, storage_config=storage_config, version=version, ignore_deletion_vectors=ignore_deletion_vectors
    )

    handle = ScanOperatorHandle.from_python_scan_operator(delta_lake_operator)
    builder = LogicalPlanBuilder.from_tabular_scan(scan_operator=handle)
    return DataFrame(builder)

@PublicAPI
def read_hudi(
    table_uri: str,
    io_config: IOConfig | None = None,
    checkpoint: "CheckpointConfig | None" = None,
) -> DataFrame:
    """Create a DataFrame from a Hudi table.

    Args:
        table_uri: URI to the Hudi table (supports remote URLs to object stores such as ``s3://`` or ``gs://``).
        io_config: A custom IOConfig to use when accessing Hudi table object storage data. Defaults to None.
        checkpoint: Optional :class:`daft.CheckpointConfig` for progress tracking across runs. Bundles the
            checkpoint store, the source key column (``on=``), and optional anti-join tuning. Rows whose key
            already exists in the store are skipped on re-run. Requires the Ray runner.

    Returns:
        DataFrame: A DataFrame with the schema converted from the specified Hudi table.

    Note:
        This function requires the use of Apache Hudi. To ensure that this is installed with Daft, you may install: ``pip install -U daft[hudi]``

    Examples:
        Read a Hudi table from a local path:
        >>> df = daft.read_hudi("some-table-uri")
        >>> df = df.where(df["foo"] > 5)
        >>> df.show()

        Read a Hudi table from a public S3 bucket:
        >>> from daft.io import S3Config, IOConfig
        >>> io_config = IOConfig(s3=S3Config(region="us-west-2", anonymous=True))
        >>> df = daft.read_hudi("s3://bucket/path/to/hudi_table/", io_config=io_config)
        >>> df.show()
    """
    from daft.io.hudi.hudi_scan import HudiScanOperator

    io_config = context.get_context().daft_planning_config.default_io_config if io_config is None else io_config

    multithreaded_io = runners.get_or_create_runner().name != "ray"
    storage_config = StorageConfig(multithreaded_io, io_config)

    hudi_operator = HudiScanOperator(table_uri, storage_config=storage_config)

    handle = ScanOperatorHandle.from_python_scan_operator(hudi_operator)
    builder = LogicalPlanBuilder.from_tabular_scan(scan_operator=handle)
    builder = attach_checkpoint(builder, checkpoint)
    return DataFrame(builder)

@PublicAPI
def read_iceberg(
    table: Union[str, "PyIcebergTable"],
    snapshot_id: int | None = None,
    io_config: IOConfig | None = None,
    checkpoint: "CheckpointConfig | None" = None,
) -> DataFrame:
    """Create a DataFrame from an Iceberg table.

    Args:
        table (str or pyiceberg.table.Table): A path to an Iceberg metadata file (supports remote URLs to object stores
            such as ``s3://`` or ``gs://``) or a [PyIceberg Table](https://py.iceberg.apache.org/reference/pyiceberg/table/#pyiceberg.table.Table)
            created using the PyIceberg library.
        snapshot_id (int, optional): Snapshot ID of the table to query
        io_config (IOConfig, optional): A custom IOConfig to use when accessing Iceberg object storage data. If provided, configurations set in `table` are ignored.
        checkpoint: Optional :class:`daft.CheckpointConfig` for progress tracking across runs. Bundles the
            checkpoint store, the source key column (``on=``), and optional anti-join tuning. Rows whose key
            already exists in the store are skipped on re-run. Requires the Ray runner.

    Returns:
        DataFrame: a DataFrame with the schema converted from the specified Iceberg table

    Note:
        This function requires the use of [PyIceberg](https://py.iceberg.apache.org/), which is the Apache Iceberg's
        official project for Python.

    Examples:
        Read an Iceberg table from a PyIceberg table:
        >>> import pyiceberg
        >>>
        >>> table = pyiceberg.Table(...)
        >>> df = daft.read_iceberg(table)
        >>>
        >>> # Filters on this dataframe can now be pushed into the read operation from Iceberg
        >>> df = df.where(df["foo"] > 5)
        >>> df.show()

        Read an Iceberg table from S3 using IOConfig:
        >>> from daft.io import S3Config, IOConfig
        >>> io_config = IOConfig(s3=S3Config(region="us-west-2", anonymous=True))
        >>> df = daft.read_iceberg("s3://bucket/path/to/iceberg/metadata.json", io_config=io_config)
        >>> df.show()
    """
    from pyiceberg.table import StaticTable

    from daft.io.iceberg.iceberg_scan import IcebergScanOperator

    # support for read_iceberg('path/to/metadata.json')
    if isinstance(table, str):
        table = StaticTable.from_metadata(metadata_location=table)

    io_config = (
        _convert_iceberg_file_io_properties_to_io_config(table.io.properties) if io_config is None else io_config
    )
    io_config = context.get_context().daft_planning_config.default_io_config if io_config is None else io_config

    multithreaded_io = runners.get_or_create_runner().name != "ray"
    storage_config = StorageConfig(multithreaded_io, io_config)

    iceberg_operator = IcebergScanOperator(table, snapshot_id=snapshot_id, storage_config=storage_config)

    handle = ScanOperatorHandle.from_python_scan_operator(iceberg_operator)
    builder = LogicalPlanBuilder.from_tabular_scan(scan_operator=handle)
    builder = attach_checkpoint(builder, checkpoint)
    return DataFrame(builder)

@PublicAPI
def read_json(
    path: str | list[str],
    infer_schema: bool = True,
    schema: dict[str, DataType] | None = None,
    io_config: IOConfig | None = None,
    file_path_column: str | None = None,
    hive_partitioning: bool = False,
    skip_empty_files: bool = False,
    _buffer_size: int | None = None,
    _chunk_size: int | None = None,
    checkpoint: "CheckpointConfig | None" = None,
) -> DataFrame:
    """Creates a DataFrame from line-delimited JSON file(s).

    Args:
        path (str): Path to JSON files (allows for wildcards; supports remote URLs to object stores such as ``s3://`` or ``gs://``)
        infer_schema (bool): Whether to infer the schema of the JSON, defaults to True.
        schema (dict[str, DataType]): A schema that is used as the definitive schema for the JSON if infer_schema is False, otherwise it is used as a schema hint that is applied after the schema is inferred (overriding the types of inferred columns, and appending any new columns not found during inference).
        io_config (IOConfig): Config to be used with the native downloader
        file_path_column: Include the source path(s) as a column with this name. Defaults to None.
        hive_partitioning: Whether to infer hive_style partitions from file paths and include them as columns in the Dataframe. Defaults to False.
        skip_empty_files: Whether to skip empty files when reading. Defaults to False.
        checkpoint: Optional :class:`daft.CheckpointConfig` for progress tracking across runs. Bundles the
            checkpoint store, the source key column (``on=``), and optional anti-join tuning. Rows whose key
            already exists in the store are skipped on re-run. Requires the Ray runner.

    Returns:
        DataFrame: parsed DataFrame

    Examples:
        Read a JSON file from a local path:
        >>> df = daft.read_json("/path/to/file.json")
        >>> df = daft.read_json("/path/to/directory")
        >>> df = daft.read_json("/path/to/files-*.json")

        Read a JSON file from a public S3 bucket:
        >>> from daft.io import S3Config, IOConfig
        >>> io_config = IOConfig(s3=S3Config(region="us-west-2", anonymous=True))
        >>> df = daft.read_json("s3://path/to/files-*.json", io_config=io_config)
        >>> df.show()
    """
    if isinstance(path, list) and len(path) == 0:
        raise ValueError("Cannot read DataFrame from empty list of JSON filepaths")

    if not infer_schema and schema is None:
        raise ValueError(
            "Cannot read DataFrame with infer_schema=False and schema=None, please provide a schema or set infer_schema=True"
        )

    io_config = context.get_context().daft_planning_config.default_io_config if io_config is None else io_config

    json_config = JsonSourceConfig(buffer_size=_buffer_size, chunk_size=_chunk_size, skip_empty_files=skip_empty_files)
    file_format_config = FileFormatConfig.from_json_config(json_config)
    storage_config = StorageConfig(True, io_config)

    builder = get_tabular_files_scan(
        path=path,
        infer_schema=infer_schema,
        schema=schema,
        file_format_config=file_format_config,
        storage_config=storage_config,
        file_path_column=file_path_column,
        hive_partitioning=hive_partitioning,
    )
    builder = attach_checkpoint(builder, checkpoint)
    return DataFrame(builder)

@PublicAPI
def read_kafka(
    bootstrap_servers: str | Sequence[str],
    topics: str | Sequence[str],
    *,
    start: object = _KIND_EARLIEST,
    end: object = _KIND_LATEST,
    group_id: str = "daft-bounded-kafka-reader",
    partitions: Sequence[int] | None = None,
    chunk_size: int = 1024,
    kafka_client_config: Mapping[str, object] | None = None,
    timeout_ms: int = 10_000,
) -> DataFrame:
    """Creates a DataFrame by reading messages from Kafka topic(s).

    .. warning::

        This API is **experimental** and may change in future releases. Currently only bounded
        batch reads are supported — there is no streaming/unbounded mode and no offset commit
        management.

    This function reads bounded ranges of messages from one or more Kafka topics. It supports
    multiple ways to specify the start and end bounds: earliest/latest, timestamp, or explicit
    partition offsets.

    Args:
        bootstrap_servers (str | Sequence[str]): Kafka bootstrap server(s) to connect to.
            Can be a single server string (e.g., "localhost:9092") or a sequence of servers.
        topics (str | Sequence[str]): Kafka topic(s) to read from. Can be a single topic string
            or a sequence of topics.
        start (object): The start bound for reading messages. Defaults to "earliest".
            Supported values:
            - "earliest": Start from the earliest available offset for each partition.
            - "latest": Start from the latest offset for each partition.
            - int: Timestamp in milliseconds since epoch.
            - datetime: A timezone-aware or naive datetime (naive datetimes are assumed UTC).
            - str: An ISO-8601 timestamp string (e.g., "2024-01-01T00:00:00Z").
            - dict: For single topic: ``{partition: offset}``. For multiple topics: ``{topic: {partition: offset}}``.
        end (object): The end bound for reading messages. Defaults to "latest".
            Supports the same value types as ``start``. The end offset is exclusive.
        group_id (str): Consumer group ID used for the Kafka consumer. Defaults to
            "daft-bounded-kafka-reader".
        partitions (Sequence[int] | None): Optional sequence of partition IDs to read from.
            If None, reads from all partitions of the specified topic(s). Defaults to None.
        chunk_size (int): Maximum number of messages per RecordBatch. Defaults to 1024.
        kafka_client_config (Mapping[str, object] | None): Optional additional configuration
            options passed directly to the underlying Kafka consumer. These are merged with
            the default configuration. Defaults to None.
        timeout_ms (int): Timeout in milliseconds for Kafka operations (metadata queries,
            message consumption, etc.). Defaults to 10_000 (10 seconds).

    Returns:
        DataFrame: A DataFrame with the following schema:
            - topic (string): The topic the message was read from.
            - partition (int32): The partition ID within the topic.
            - offset (int64): The offset of the message within the partition.
            - timestamp_ms (int64): The timestamp of the message in milliseconds since epoch,
              or null if not available.
            - key (binary): The message key as raw bytes, or null if not present.
            - value (binary): The message value as raw bytes.

    Examples:
        Read from a single topic with default bounds (earliest to latest):
        >>> df = daft.read_kafka("localhost:9092", "my-topic")

        Read from multiple topics:
        >>> df = daft.read_kafka("localhost:9092", ["topic-a", "topic-b"])

        Read from specific partitions:
        >>> df = daft.read_kafka("localhost:9092", "my-topic", partitions=[0, 1])

        Read from a timestamp range:
        >>> from datetime import datetime, timezone
        >>> start_dt = datetime(2024, 1, 1, tzinfo=timezone.utc)
        >>> end_dt = datetime(2024, 1, 2, tzinfo=timezone.utc)
        >>> df = daft.read_kafka("localhost:9092", "my-topic", start=start_dt, end=end_dt)

        Read from specific partition offsets:
        >>> df = daft.read_kafka("localhost:9092", "my-topic", start={0: 100, 1: 200})

        Read from multiple topics with per-topic offsets:
        >>> df = daft.read_kafka(
        ...     "localhost:9092",
        ...     ["topic-a", "topic-b"],
        ...     start={"topic-a": {0: 100}, "topic-b": {0: 50}},
        ... )

        Configure Kafka client options:
        >>> df = daft.read_kafka(
        ...     "localhost:9092",
        ...     "my-topic",
        ...     kafka_client_config={"enable.partition.eof": True, "session.timeout.ms": 30000},
        ... )

    Note:
        This function requires the ``confluent-kafka`` package. Install it with:
        ``pip install daft[kafka]`` or ``pip install confluent-kafka``

        Timestamp bounds use Kafka message timestamps. If your cluster uses CreateTime, producers can publish
        late/out-of-order timestamps; timestamp bounds are not a safe exactly-once checkpoint. Prefer offset-based
        checkpoints (e.g., partition offset maps or committed consumer offsets)
    """
    if isinstance(bootstrap_servers, str):
        bootstrap_servers_str = bootstrap_servers
    else:
        bootstrap_servers_str = ",".join(str(s) for s in bootstrap_servers)

    if isinstance(topics, str):
        topics = [topics]
    else:
        topics = list(topics)
    topics = list(dict.fromkeys(topics))

    if not topics:
        raise ValueError("[read_kafka] topics must be non-empty")

    if timeout_ms <= 0:
        raise ValueError("[read_kafka] timeout_ms must be > 0")

    if chunk_size <= 0:
        raise ValueError("[read_kafka] chunk_size must be > 0")

    start_kind, start_timestamp_ms, start_topic_partition_offsets = _normalize_bound(start, topics)
    end_kind, end_timestamp_ms, end_topic_partition_offsets = _normalize_bound(end, topics)

    return KafkaSource(
        bootstrap_servers=bootstrap_servers_str,
        group_id=group_id,
        topics=topics,
        start_kind=start_kind,
        start_timestamp_ms=start_timestamp_ms,
        start_topic_partition_offsets=start_topic_partition_offsets,
        end_kind=end_kind,
        end_timestamp_ms=end_timestamp_ms,
        end_topic_partition_offsets=end_topic_partition_offsets,
        partitions=partitions,
        kafka_client_config=kafka_client_config,
        timeout_ms=timeout_ms,
        chunk_size=chunk_size,
    ).read()

@PublicAPI
def read_lance(
    uri: str | os.PathLike[str],
    io_config: Any = None,
    version: Any = None,
    asof: Any = None,
    block_size: Any = None,
    commit_lock: Any = None,
    index_cache_size: Any = None,
    default_scan_options: Any = None,
    metadata_cache_size_bytes: Any = None,
    fragment_group_size: Any = None,
    include_fragment_id: Any = None,
    checkpoint: Any = None,
) -> Any:
    """Create a DataFrame from a LanceDB table.

    Args:
        uri: The URI of the Lance table to read from. Accepts a local path or an
            object-store URI like "s3://bucket/path".
        io_config: A custom IOConfig to use when accessing LanceDB data. Defaults to None.
        version : optional, int | str
            If specified, load a specific version of the Lance dataset. Else, loads the
            latest version. A version number (`int`) or a tag (`str`) can be provided.
        asof : optional, datetime or str
            If specified, find the latest version created on or earlier than the given
            argument value. If a version is already specified, this arg is ignored.
        block_size : optional, int
            Block size in bytes. Provide a hint for the size of the minimal I/O request.
        commit_lock : optional, lance.commit.CommitLock
            A custom commit lock.  Only needed if your object store does not support
            atomic commits.  See the user guide for more details.
        index_cache_size : optional, int
            Index cache size. Index cache is a LRU cache with TTL. This number specifies the
            number of index pages, for example, IVF partitions, to be cached in
            the host memory. Default value is ``256``.

            Roughly, for an ``IVF_PQ`` partition with ``n`` rows, the size of each index
            page equals the combination of the pq code (``np.array([n,pq], dtype=uint8))``
            Approximately, ``n = Total Rows / number of IVF partitions``.
            ``pq = number of PQ sub-vectors``.
        default_scan_options : optional, dict
            Default scan options that are used when scanning the dataset.  This accepts
            the same arguments described in :py:meth:`lance.LanceDataset.scanner`.  The
            arguments will be applied to any scan operation.

            This can be useful to supply defaults for common parameters such as
            ``batch_size``.

            It can also be used to create a view of the dataset that includes meta
            fields such as ``_rowid`` or ``_rowaddr``.  If ``default_scan_options`` is
            provided then the schema returned by :py:meth:`lance.LanceDataset.schema` will
            include these fields if the appropriate scan options are set.
            like this:
            default_scan_options = {"with_row_address": True, "with_row_id" : True,  "batch_size": 1024}
            more see: https://lance-format.github.io/lance-python-doc/dataset.html
        metadata_cache_size_bytes : optional, int
            Size of the metadata cache in bytes. This cache is used to store metadata
            information about the dataset, such as schema and statistics. If not specified,
            a default size will be used.
        fragment_group_size : optional, int
            Number of fragments to group together in a single scan task. If None or <= 1,
            each fragment will be processed individually (default behavior).
        include_fragment_id : Optional, bool
            Whether to display fragment_id.
            if you have the behavior of 'merge_columns_df' or 'write_lance(mode = 'merge')', the `include_fragment_id` must be set to True
        checkpoint: Optional :class:`daft.CheckpointConfig` for progress tracking across runs. Bundles the
            checkpoint store, the source key column (``on=``), and optional anti-join tuning. Rows whose key
            already exists in the store are skipped on re-run. Requires the Ray runner.

    Returns:
        DataFrame: a DataFrame with the schema converted from the specified LanceDB table

        This function requires the use of [LanceDB](https://lancedb.github.io/lancedb/), which is the Python library for the LanceDB project.
        To ensure that this is installed with Daft, you may install: `pip install daft[lance]`

    Examples:
        Read a local LanceDB table:
        >>> df = daft.read_lance("/path/to/lance/data/")
        >>> df.show()

        Read a LanceDB table and specify a version:
        >>> df = daft.read_lance("/path/to/lance/data/", version=1)
        >>> df.show()

        Read a LanceDB table with fragment grouping:
        >>> df = daft.read_lance("/path/to/lance/data/", fragment_group_size=5)
        >>> df.show()

        Read a LanceDB table from a public S3 bucket:
        >>> from daft.io import S3Config, IOConfig
        >>> io_config = IOConfig(s3=S3Config(region="us-west-2", anonymous=True))
        >>> df = daft.read_lance("s3://daft-oss-public-data/lance/words-test-dataset", io_config=io_config)
        >>> df.show()
    """
    return _daft_lance.read_lance(
        uri,
        io_config=io_config,
        version=version,
        asof=asof,
        block_size=block_size,
        commit_lock=commit_lock,
        index_cache_size=index_cache_size,
        default_scan_options=default_scan_options,
        metadata_cache_size_bytes=metadata_cache_size_bytes,
        fragment_group_size=fragment_group_size,
        include_fragment_id=include_fragment_id,
        checkpoint=checkpoint,
    )

@PublicAPI
def read_parquet(
    path: str | list[str],
    row_groups: list[list[int]] | None = None,
    infer_schema: bool = True,
    schema: dict[str, DataType] | None = None,
    io_config: IOConfig | None = None,
    file_path_column: str | None = None,
    hive_partitioning: bool = False,
    coerce_int96_timestamp_unit: str | TimeUnit | None = None,
    _multithreaded_io: bool | None = None,
    _chunk_size: int | None = None,  # A hidden parameter for testing purposes.
    checkpoint: "CheckpointConfig | None" = None,
) -> DataFrame:
    """Creates a DataFrame from Parquet file(s).

    Args:
        path (str): Path to Parquet file (allows for wildcards; supports remote URLs to object stores such as ``s3://`` or ``gs://``)
        row_groups (List[int] or List[List[int]]): List of row groups to read corresponding to each file.
        infer_schema (bool): Whether to infer the schema of the Parquet, defaults to True.
        schema (dict[str, DataType]): A schema that is used as the definitive schema for the Parquet file if infer_schema is False, otherwise it is used as a schema hint that is applied after the schema is inferred (overriding the types of inferred columns, and appending any new columns not found during inference).
        io_config (IOConfig): Config to be used with the native downloader
        file_path_column: Include the source path(s) as a column with this name. Defaults to None.
        hive_partitioning: Whether to infer hive_style partitions from file paths and include them as columns in the Dataframe. Defaults to False.
        coerce_int96_timestamp_unit: TimeUnit to coerce Int96 TimeStamps to. e.g.: [ns, us, ms], Defaults to None.
        _multithreaded_io: Whether to use multithreading for IO threads. Setting this to False can be helpful in reducing
            the amount of system resources (number of connections and thread contention) when running in the Ray runner.
            Defaults to None, which will let Daft decide based on the runner it is currently using.
        checkpoint: Optional :class:`daft.CheckpointConfig` for progress tracking across runs. Bundles the
            checkpoint store, the source key column (``on=``), and optional anti-join tuning. Rows whose key
            already exists in the store are skipped on re-run. Requires the Ray runner.

    Returns:
        DataFrame: parsed DataFrame

    Examples:
        Read a Parquet file from a local path:
        >>> df = daft.read_parquet("/path/to/file.parquet")
        >>> df = daft.read_parquet("/path/to/directory")
        >>> df = daft.read_parquet("/path/to/files-*.parquet")

        Read a Parquet file from a public S3 bucket:
        >>> from daft.io import S3Config, IOConfig
        >>> io_config = IOConfig(s3=S3Config(region="us-west-2", anonymous=True))
        >>> df = daft.read_parquet("s3://path/to/files-*.parquet", io_config=io_config)
        >>> df.show()
    """
    io_config = context.get_context().daft_planning_config.default_io_config if io_config is None else io_config

    if isinstance(path, list) and len(path) == 0:
        raise ValueError("Cannot read DataFrame from empty list of Parquet filepaths")

    # If running on Ray, we want to limit the amount of concurrency and requests being made.
    # This is because each Ray worker process receives its own pool of thread workers and connections
    multithreaded_io = (
        (runners.get_or_create_runner().name != "ray") if _multithreaded_io is None else _multithreaded_io
    )

    if isinstance(coerce_int96_timestamp_unit, str):
        coerce_int96_timestamp_unit = TimeUnit.from_str(coerce_int96_timestamp_unit)

    pytimeunit = coerce_int96_timestamp_unit._timeunit if coerce_int96_timestamp_unit is not None else None

    if isinstance(path, list) and row_groups is not None and len(path) != len(row_groups):
        raise ValueError("row_groups must be the same length as the list of paths provided.")
    if isinstance(row_groups, list) and not isinstance(path, list):
        raise ValueError("row_groups are only supported when reading multiple non-globbed/wildcarded files")

    file_format_config = FileFormatConfig.from_parquet_config(
        ParquetSourceConfig(coerce_int96_timestamp_unit=pytimeunit, row_groups=row_groups, chunk_size=_chunk_size)
    )
    storage_config = StorageConfig(multithreaded_io, io_config)

    builder = get_tabular_files_scan(
        path=path,
        infer_schema=infer_schema,
        schema=schema,
        file_format_config=file_format_config,
        storage_config=storage_config,
        file_path_column=file_path_column,
        hive_partitioning=hive_partitioning,
    )

    builder = attach_checkpoint(builder, checkpoint)

    return DataFrame(builder)