DataChain

def listings(
    session: Session | None = None,
    in_memory: bool = False,
    column: str = "listing",
    **kwargs,
) -> "DataChain":
    """Generate chain with list of cached listings.
    Listing is a special kind of dataset which has directory listing data of
    some underlying storage (e.g S3 bucket).

    Example:
        ```py
        import datachain as dc
        dc.listings().show()
        ```
    """
    session = Session.get(session, in_memory=in_memory)
    catalog = kwargs.get("catalog") or session.catalog

    return read_values(
        session=session,
        in_memory=in_memory,
        output={column: ListingInfo},
        **{column: catalog.listings()},  # type: ignore[arg-type]
    )

def read_csv(
    path: str | os.PathLike[str] | list[str] | list[os.PathLike[str]],
    delimiter: str | None = None,
    header: bool = True,
    output: OutputType = None,
    column: str = "",
    model_name: str = "",
    source: bool = True,
    nrows: int | None = None,
    session: Session | None = None,
    settings: dict | None = None,
    column_types: dict[str, "str | ArrowDataType"] | None = None,
    parse_options: dict[str, str | bool | Callable] | None = None,
    **kwargs,
) -> "DataChain":
    """Generate chain from csv files.

    Parameters:
        path: Storage URI with directory. URI must start with storage prefix such
            as `s3://`, `gs://`, `az://` or "file:///".
        delimiter: Character for delimiting columns. Takes precedence if also
            specified in `parse_options`. Defaults to ",".
        header: Whether the files include a header row.
        output: Dictionary or feature class defining column names and their
            corresponding types. List of column names is also accepted, in which
            case types will be inferred.
        column: Created column name.
        model_name: Generated model name.
        source: Whether to include info about the source file.
        nrows: Optional row limit.
        session: Session to use for the chain.
        settings: Settings to use for the chain.
        column_types: Dictionary of column names and their corresponding types.
            It is passed to CSV reader and for each column specified type auto
            inference is disabled.
        parse_options: Tells the parser how to process lines.
            See https://arrow.apache.org/docs/python/generated/pyarrow.csv.ParseOptions.html

    Example:
        Reading a csv file:
        ```py
        import datachain as dc
        chain = dc.read_csv("s3://mybucket/file.csv")
        ```

        Reading csv files from a directory as a combined dataset:
        ```py
        import datachain as dc
        chain = dc.read_csv("s3://mybucket/dir")
        ```
    """
    from pandas._libs.parsers import STR_NA_VALUES
    from pyarrow.csv import ConvertOptions, ParseOptions, ReadOptions
    from pyarrow.dataset import CsvFileFormat
    from pyarrow.lib import type_for_alias

    from .storage import read_storage

    parse_options = parse_options or {}
    if "delimiter" not in parse_options:
        parse_options["delimiter"] = ","
    if delimiter:
        parse_options["delimiter"] = delimiter

    if column_types:
        column_types = {
            name: type_for_alias(typ) if isinstance(typ, str) else typ
            for name, typ in column_types.items()
        }
    else:
        column_types = {}

    chain = read_storage(path, session=session, settings=settings, **kwargs)

    column_names = None
    if not header:
        if not output:
            msg = "error parsing csv - provide output if no header"
            raise DatasetPrepareError(chain.name, msg)
        if isinstance(output, Sequence):
            column_names = output  # type: ignore[assignment]
        elif isinstance(output, dict):
            column_names = list(output.keys())
        elif (fr := ModelStore.to_pydantic(output)) is not None:
            column_names = list(fr.model_fields.keys())
        else:
            msg = f"error parsing csv - incompatible output type {type(output)}"
            raise DatasetPrepareError(chain.name, msg)

    parse_options = ParseOptions(**parse_options)
    read_options = ReadOptions(column_names=column_names)
    convert_options = ConvertOptions(
        strings_can_be_null=True,
        null_values=STR_NA_VALUES,
        column_types=column_types,
    )
    format = CsvFileFormat(
        parse_options=parse_options,
        read_options=read_options,
        convert_options=convert_options,
    )
    return chain.parse_tabular(
        output=output,
        column=column,
        model_name=model_name,
        source=source,
        nrows=nrows,
        format=format,
        parse_options=parse_options,
    )

def read_dataset(
    name: str,
    namespace: str | None = None,
    project: str | None = None,
    version: str | int | None = None,
    session: Session | None = None,
    settings: dict | None = None,
    delta: bool | None = False,
    delta_on: str | Sequence[str] | None = (
        "file.path",
        "file.etag",
        "file.version",
    ),
    delta_result_on: str | Sequence[str] | None = None,
    delta_compare: str | Sequence[str] | None = None,
    delta_retry: bool | str | None = None,
    delta_unsafe: bool = False,
    update: bool = False,
) -> "DataChain":
    """Get data from a saved Dataset. It returns the chain itself.
    If dataset or version is not found locally, it will try to pull it from Studio.

    Parameters:
        name: The dataset name, which can be a fully qualified name including the
            namespace and project. Alternatively, it can be a regular name, in which
            case the explicitly defined namespace and project will be used if they are
            set; otherwise, default values will be applied. The name can also include
            a version using the ``name@version`` format (e.g. ``"my_dataset@1.0.0"``).
            If ``version`` is also provided explicitly, it takes priority.
        namespace: optional name of namespace in which dataset to read is created
        project: optional name of project in which dataset to read is created
        version: dataset version. Supports:
            - Exact version strings: "1.2.3"
            - Legacy integer versions: 1, 2, 3 (finds latest major version)
            - Version specifiers (PEP 440): ">=1.0.0,<2.0.0", "~=1.4.2", "==1.2.*", etc.
        session: Session to use for the chain.
        settings: Settings to use for the chain.
        delta: If True, only process new or changed files instead of reprocessing
            everything. This saves time by skipping files that were already processed in
            previous versions. The optimization is working when a new version of the
            dataset is created.
            Default is False.
        delta_on: Field(s) that uniquely identify each record in the source data.
            Used to detect which records are new or changed.
            Default is ("file.path", "file.etag", "file.version").
        delta_result_on: Field(s) in the result dataset that match `delta_on` fields.
            Only needed if you rename the identifying fields during processing.
            Default is None.
        delta_compare: Field(s) used to detect if a record has changed.
            If not specified, all fields except `delta_on` fields are used.
            Default is None.
        delta_retry: Controls retry behavior for failed records:
            - String (field name): Reprocess records where this field is not empty
              (error mode)
            - True: Reprocess records missing from the result dataset (missing mode)
            - None: No retry processing (default)
        update: If True always checks for newer versions available on Studio, even if
            some version of the dataset exists locally already. If False (default), it
            will only fetch the dataset from Studio if it is not found locally.
        delta_unsafe: Allow restricted ops in delta: merge, union, subtract,
            diff, file_diff, agg, group_by, distinct. When multiple delta
            sources participate in one composed query, this must be enabled on
            every participating delta source.


    Example:
        ```py
        import datachain as dc
        chain = dc.read_dataset("my_cats")
        ```

        ```py
        import datachain as dc
        chain = dc.read_dataset("dev.animals.my_cats")
        ```

        ```py
        chain = dc.read_dataset("my_cats", version="1.0.0")
        ```

        ```py
        # Version can also be embedded in the name using the @ syntax
        chain = dc.read_dataset("my_cats@1.0.0")
        ```

        ```py
        # Using version specifiers (PEP 440)
        chain = dc.read_dataset("my_cats", version=">=1.0.0,<2.0.0")
        ```

        ```py
        # Legacy integer version support (finds latest in major version)
        chain = dc.read_dataset("my_cats", version=1)  # Latest 1.x.x version
        ```

        ```py
        # Always check for newer versions matching a version specifier from Studio
        chain = dc.read_dataset("my_cats", version=">=1.0.0", update=True)
        ```

        ```py
        session = Session.get(client_config={"aws_endpoint_url": "<minio-url>"})
        settings = {
            "cache": True,
            "parallel": 4,
            "workers": 4,
            "min_task_size": 1000,
            "prefetch": 10,
        }
        chain = dc.read_dataset(
            name="my_cats",
            version="1.0.0",
            session=session,
            settings=settings,
        )
        ```
    """
    from datachain.telemetry import telemetry

    from .datachain import DataChain

    name, name_version = parse_dataset_with_version(name)
    if version is None:
        version = name_version

    telemetry.send_event_once("class", "datachain_init", name=name, version=version)

    session = Session.get(session)
    catalog = session.catalog

    namespace_name, project_name, name = catalog.get_full_dataset_name(
        name,
        project_name=project,
        namespace_name=namespace,
    )

    if version is not None:
        dataset = session.catalog.get_dataset_with_remote_fallback(
            name,
            namespace_name,
            project_name,
            update=update,
            include_incomplete=False,  # Never include incomplete datasets
        )

        # Convert legacy integer versions to version specifiers
        # For backward compatibility we still allow users to put version as integer
        # in which case we convert it to a version specifier that finds the latest
        # version where major part is equal to that input version.
        # For example if user sets version=2, we convert it to ">=2.0.0,<3.0.0"
        # which will find something like 2.4.3 (assuming 2.4.3 is the biggest among
        # all 2.* dataset versions)
        if isinstance(version, int):
            version_spec = f">={version}.0.0,<{version + 1}.0.0"
        else:
            version_spec = str(version)

        from packaging.specifiers import InvalidSpecifier, SpecifierSet

        try:
            # Try to parse as version specifier
            SpecifierSet(version_spec)
            # If it's a valid specifier set, find the latest compatible version
            latest_compatible = dataset.latest_compatible_version(version_spec)
            if not latest_compatible:
                raise DatasetVersionNotFoundError(
                    f"No dataset {name} version matching specifier {version_spec}"
                )
            version = latest_compatible
        except InvalidSpecifier:
            # If not a valid specifier, treat as exact version string
            # This handles cases like "1.2.3" which are exact versions, not specifiers
            pass

    if settings:
        _settings = Settings(**settings)
    else:
        _settings = Settings()

    query = DatasetQuery(
        name=name,
        project_name=project_name,
        namespace_name=namespace_name,
        version=version,  #  type: ignore[arg-type]
        session=session,
        update=update,
    )

    signals_schema = SignalSchema({"sys": Sys})
    if query.feature_schema:
        signals_schema |= SignalSchema.deserialize(query.feature_schema)
    else:
        signals_schema |= SignalSchema.from_column_types(query.column_types or {})

    if delta:
        signals_schema = signals_schema.clone_without_sys_signals()

    chain = DataChain(query, _settings, signals_schema)

    if delta:
        if delta_on is None:
            raise ValueError("'delta on' fields must be defined")
        query.delta_spec = DeltaSpec(
            on=delta_on,
            right_on=delta_result_on,
            compare=delta_compare,
            delta_retry=delta_retry,
            delta_unsafe=delta_unsafe,
        )

    return chain

def read_hf(
    dataset: "HFDatasetType",
    *args: Any,
    session: Session | None = None,
    settings: dict | None = None,
    column: str = "",
    model_name: str = "",
    limit: int = 0,
    **kwargs: Any,
) -> "DataChain":
    """Generate chain from Hugging Face Hub dataset.

    Parameters:
        dataset: Path or name of the dataset to read from Hugging Face Hub,
            or an instance of `datasets.Dataset`-like object.
        args: Additional positional arguments to pass to `datasets.load_dataset`.
        session: Session to use for the chain.
        settings: Settings to use for the chain.
        column: Generated object column name.
        model_name: Generated model name.
        limit: The maximum number of items to read from the HF dataset.
            Applies `take(limit)` to `datasets.load_dataset`.
            Defaults to 0 (no limit).
        kwargs: Parameters to pass to `datasets.load_dataset`.

    Example:
        Load from Hugging Face Hub:
        ```py
        import datachain as dc
        chain = dc.read_hf("beans", split="train")
        ```

        Generate chain from loaded dataset:
        ```py
        from datasets import load_dataset
        ds = load_dataset("beans", split="train")
        import datachain as dc
        chain = dc.read_hf(ds)
        ```

        Streaming with limit, for large datasets:
        ```py
        import datachain as dc
        ds = dc.read_hf("beans", split="train", streaming=True, limit=10)
        ```

        or use HF split syntax (not supported if streaming is enabled):
        ```py
        import datachain as dc
        ds = dc.read_hf("beans", split="train[%10]")
        ```
    """
    from datachain.lib.hf import HFGenerator, get_output_schema, stream_splits

    from .values import read_values

    output: dict[str, DataType] = {}
    ds_dict = stream_splits(dataset, *args, **kwargs)
    if len(ds_dict) > 1:
        output = {"split": str}

    model_name = model_name or column or ""
    hf_features = next(iter(ds_dict.values())).features
    hf_output, normalized_names = get_output_schema(hf_features, list(output.keys()))
    output = output | hf_output
    model = dict_to_data_model(model_name, output, list(normalized_names.values()))
    if column:
        output = {column: model}

    chain = read_values(split=list(ds_dict.keys()), session=session, settings=settings)
    return chain.gen(HFGenerator(dataset, model, limit, *args, **kwargs), output=output)

def read_json(
    path: str | os.PathLike[str],
    type: FileType = "text",
    spec: DataType | None = None,
    schema_from: str | None = "auto",
    jmespath: str | None = None,
    column: str | None = "",
    model_name: str | None = None,
    format: str | None = "json",
    nrows: int | None = None,
    **kwargs,
) -> "DataChain":
    """Get data from JSON. It returns the chain itself.

    Parameters:
        path: storage URI with directory. URI must start with storage prefix such
            as `s3://`, `gs://`, `az://` or "file:///"
        type: read file as "binary", "text", or "image" data. Default is "text".
        spec: optional Data Model
        schema_from: path to sample to infer spec (if schema not provided)
        column: generated column name
        model_name: optional generated model name
        format: "json", "jsonl"
        jmespath: optional JMESPATH expression to reduce JSON
        nrows: optional row limit for jsonl and JSON arrays

    Example:
        infer JSON schema from data, reduce using JMESPATH
        ```py
        import datachain as dc
        chain = dc.read_json("gs://json", jmespath="key1.key2")
        ```

        infer JSON schema from a particular path
        ```py
        import datachain as dc
        chain = dc.read_json("gs://json_ds", schema_from="gs://json/my.json")
        ```
    """
    from .storage import read_storage

    if schema_from == "auto":
        schema_from = os.fspath(path)

    def jmespath_to_name(s: str):
        name_end = re.search(r"\W", s).start() if re.search(r"\W", s) else len(s)  # type: ignore[union-attr]
        return s[:name_end]

    if (not column) and jmespath:
        column = jmespath_to_name(jmespath)
    if not column:
        column = format
    chain = read_storage(uri=path, type=type, **kwargs)
    signal_dict = {
        column: meta_formats.read_meta(
            schema_from=schema_from,
            format=format,
            spec=spec,
            model_name=model_name,
            jmespath=jmespath,
            nrows=nrows,
            session=kwargs.get("session"),
        ),
        "params": {"file": File},
    }
    # disable prefetch if nrows is set
    settings = {"prefetch": 0} if nrows else {}

    cloudpickle.register_pickle_by_value(meta_formats)

    return chain.settings(**settings).gen(**signal_dict)  # type: ignore[misc, arg-type]

def read_pandas(  # type: ignore[override]
    df: "pd.DataFrame",
    name: str = "",
    session: Session | None = None,
    settings: dict | None = None,
    in_memory: bool = False,
    column: str = "",
) -> "DataChain":
    """Generate chain from pandas data-frame.

    Example:
        ```py
        import pandas as pd
        import datachain as dc

        df = pd.DataFrame({"fib": [1, 2, 3, 5, 8]})
        dc.read_pandas(df)
        ```
    """
    from .utils import DatasetPrepareError

    def get_col_name(col):
        if isinstance(col, tuple):
            # Join tuple elements with underscore for MultiIndex columns
            return "_".join(map(str, col)).lower()
        # Handle regular string column names
        return str(col).lower()

    fr_map = {get_col_name(col): df[col].tolist() for col in df.columns}

    for c in fr_map:
        if not c.isidentifier():
            raise DatasetPrepareError(
                name,
                f"import from pandas error - '{c}' cannot be a column name",
            )

    return read_values(
        name,
        session,
        settings=settings,
        column=column,
        in_memory=in_memory,
        **fr_map,
    )

def read_parquet(
    path: str | os.PathLike[str] | list[str] | list[os.PathLike[str]],
    partitioning: Any = "hive",
    output: dict[str, DataType] | None = None,
    column: str = "",
    model_name: str = "",
    source: bool = True,
    session: Session | None = None,
    settings: dict | None = None,
    **kwargs,
) -> "DataChain":
    """Generate chain from parquet files.

    Parameters:
        path: Storage path(s) or URI(s). Can be a local path or start with a
            storage prefix like `s3://`, `gs://`, `az://`, `hf://` or "file:///".
            Supports glob patterns:
              - `*` : wildcard
              - `**` : recursive wildcard
              - `?` : single character
              - `{a,b}` : brace expansion list
              - `{1..9}` : brace numeric or alphabetic range
        partitioning: Any pyarrow partitioning schema.
        output: Dictionary defining column names and their corresponding types.
        column: Created column name.
        model_name: Generated model name.
        source: Whether to include info about the source file.
        session: Session to use for the chain.
        settings: Settings to use for the chain.

    Example:
        Reading a single file:
        ```py
        import datachain as dc
        dc.read_parquet("s3://mybucket/file.parquet")
        ```

        All files from a directory:
        ```py
        dc.read_parquet("s3://mybucket/dir/")
        ```

        Only parquet files from a directory, and all it's subdirectories:
        ```py
        dc.read_parquet("s3://mybucket/dir/**/*.parquet")
        ```

        Using filename patterns - numeric, list, starting with zeros:
        ```py
        dc.read_parquet("s3://mybucket/202{1..4}/{yellow,green}-{01..12}.parquet")
        ```
    """
    from .storage import read_storage

    chain = read_storage(path, session=session, settings=settings, **kwargs)
    return chain.parse_tabular(
        output=output,
        column=column,
        model_name=model_name,
        source=source,
        format="parquet",
        partitioning=partitioning,
    )

def read_records(
    to_insert: dict | Iterable[dict] | None,
    schema: dict[str, DataType],
    session: Session | None = None,
    settings: dict | None = None,
    in_memory: bool = False,
) -> "DataChain":
    """Create a DataChain from the provided records. This is a low-level function
    that directly inserts records into the database. Unlike convenience functions
    like `read_values()` or `read_csv()`, you have to provide the schema and records
    explicitly.

    Compare it with `read_values()` which infers schema automatically and is using
    higher-level abstractions which makes it less efficient. E.g. `read_values()` cannot
    handle large datasets efficiently since it needs to load all data into memory.

    Parameters:
        to_insert: records to insert (empty list / None to create an empty chain). Can
                    be a list, iterator, or generator. Iterators are processed lazily
                    without loading all records into memory at once.

                    Each record must be a dictionary with keys matching the schema.
                    Dictionary values can be:
                    - Primitive types (str, int, etc.)
                    - DataModel objects (automatically flattened to match schema)
                    - Raw flattened data (e.g., {"person__name": "Alice", ...})
        schema: describes chain signals and their corresponding types.

    Example:
        ```py
        import datachain as dc
        from datachain import DataModel

        # Simple records with primitive types
        records = [
            {"name": "Alice", "age": 30},
            {"name": "Bob", "age": 25}
        ]
        chain = dc.read_records(records, schema={"name": str, "age": int})

        # Complex records with DataModel objects (automatically flattened)
        class Person(DataModel):
            name: str
            age: int
            city: str

        people = [
            Person(name="Alice", age=30, city="NYC"),
            Person(name="Bob", age=25, city="LA"),
        ]
        records = [{"person": p} for p in people]
        chain = dc.read_records(records, schema={"person": Person})

        # Raw pre-flattened data (also works)
        records = [
            {"person__name": "Alice", "person__age": 30, "person__city": "NYC"},
            {"person__name": "Bob", "person__age": 25, "person__city": "LA"},
        ]
        chain = dc.read_records(records, schema={"person": Person})

        # Using an iterator/generator for memory efficiency
        def generate_records():
            for i in range(1000000):
                yield {"id": i, "value": i * 2}

        chain = dc.read_records(generate_records(), schema={"id": int, "value": int})
        ```

    Notes:
        This call blocks until all records are inserted, but iterators are processed
        in batches to avoid loading all data into memory at once.
    """
    if isinstance(to_insert, dict):
        to_insert = [to_insert]
    elif not to_insert:
        to_insert = []

    signal_schema = SignalSchema(schema)
    flat: Iterable[dict] = (_flatten_record(rec, signal_schema) for rec in to_insert)
    content_hash: str | None = None
    if isinstance(to_insert, (list, tuple)):
        flat = list(flat)
        content_hash = _content_hash(flat, signal_schema)

    return create_records_dataset(
        flat,
        schema,
        content_hash,
        session=session,
        settings=settings,
        in_memory=in_memory,
    )

def read_storage(  # noqa: PLR0913
    uri: str | os.PathLike[str] | list[str] | list[os.PathLike[str]],
    *,
    type: FileType = "binary",
    session: Session | None = None,
    settings: dict | None = None,
    in_memory: bool = False,
    recursive: bool | None = True,
    column: str = "file",
    update: bool = False,
    anon: bool | None = None,
    versions: bool = True,
    delta: bool | None = False,
    delta_on: str | Sequence[str] | None = (
        "file.path",
        "file.etag",
        "file.version",
    ),
    delta_result_on: str | Sequence[str] | None = None,
    delta_compare: str | Sequence[str] | None = None,
    delta_retry: bool | str | None = None,
    delta_unsafe: bool = False,
    client_config: dict | None = None,
) -> "DataChain":
    """Get data from storage(s) as a list of file with all file attributes.
    It returns the chain itself as usual.

    Parameters:
        uri: Storage path(s) or URI(s). Can be a local path or start with a
            storage prefix like `s3://`, `gs://`, `az://`, `hf://` or "file:///".
            Supports glob patterns:
              - `*` : wildcard
              - `**` : recursive wildcard
              - `?` : single character
              - `{a,b}` : brace expansion list
              - `{1..9}` : brace numeric or alphabetic range
        type: read file as "binary", "text", or "image" data. Default is "binary".
        recursive: search recursively for the given path.
        column: Column name that will contain File objects. Default is "file".
        update: force storage reindexing. Default is False.
        anon: If True, we will treat cloud bucket as public one.
        versions: If False, list only current objects, without versions.
        client_config: Optional client configuration for the storage client.
        delta: If True, only process new or changed files instead of reprocessing
            everything. This saves time by skipping files that were already processed in
            previous versions. The optimization is working when a new version of the
            dataset is created.
            Default is False.
        delta_on: Field(s) that uniquely identify each record in the source data.
            Used to detect which records are new or changed.
            Default is ("file.path", "file.etag", "file.version").
        delta_result_on: Field(s) in the result dataset that match `delta_on` fields.
            Only needed if you rename the identifying fields during processing.
            Default is None.
        delta_compare: Field(s) used to detect if a record has changed.
            If not specified, all fields except `delta_on` fields are used.
            Default is None.
        delta_retry: Controls retry behavior for failed records:
            - String (field name): Reprocess records where this field is not empty
              (error mode)
            - True: Reprocess records missing from the result dataset (missing mode)
            - None: No retry processing (default)
        delta_unsafe: Allow restricted ops in delta: merge, union, subtract,
            diff, file_diff, agg, group_by, distinct. When multiple delta
            sources participate in one composed query, this must be enabled on
            every participating delta source. Caller must ensure datasets are
            consistent and not partially updated.

    Returns:
        DataChain: A DataChain object containing the file information.

    Examples:
        Simple call from s3:
        ```python
        import datachain as dc
        dc.read_storage("s3://my-bucket/my-dir")
        ```

        Match all .json files recursively using glob pattern
        ```py
        dc.read_storage("gs://bucket/meta/**/*.json")
        ```

        Match image file extensions for directories with pattern
        ```py
        dc.read_storage("s3://bucket/202?/**/*.{jpg,jpeg,png}")
        ```

        By ranges in filenames:
        ```py
        dc.read_storage("s3://bucket/202{1..4}/**/*.{jpg,jpeg,png}")
        ```

        Multiple URIs:
        ```python
        dc.read_storage(["s3://my-bkt/dir1", "s3://bucket2/dir2/dir3"])
        ```

        Current objects only, without the history of a versioned bucket:
        ```python
        dc.read_storage("s3://my-bucket/my-dir", versions=False)
        ```

        With AWS S3-compatible storage:
        ```python
        dc.read_storage(
            "s3://my-bucket/my-dir",
            client_config = {"aws_endpoint_url": "<minio-endpoint-url>"}
        )
        ```
    """
    from .datasets import read_dataset
    from .records import create_records_dataset, read_records

    file_type = get_file_type(type)

    uris = uri if isinstance(uri, (list, tuple)) else [uri]

    if not uris:
        raise ValueError("No URIs provided")

    for single_uri in uris:
        validate_cloud_bucket_name(str(single_uri))

    probe_config = client_config or (
        session.catalog.client_config if session is not None else None
    )

    if (
        anon is None
        and not _backends_have_credentials(uris, probe_config)
        and _all_buckets_anonymous(uris, probe_config)
    ):
        anon = True

    if anon is not None:
        client_config = (client_config or {}) | {"anon": anon}
    session = Session.get(session, client_config=client_config, in_memory=in_memory)
    catalog = session.catalog
    cache = catalog.cache
    client_config = session.catalog.client_config
    if anon is not None:
        # Session.get discards our client_config when an existing session is
        # passed. Re-apply anon locally for the listing path without mutating
        # the caller's session.
        client_config = client_config | {"anon": anon}
    if not versions:
        client_config = client_config | {"version_aware": False}
    listing_namespace_name = catalog.metastore.system_namespace_name
    listing_project_name = catalog.metastore.listing_project_name

    # Then expand all URIs that contain brace patterns
    expanded_uris = []
    for single_uri in uris:
        uri_str = str(single_uri)
        expanded_uris.extend(expand_brace_pattern(uri_str))

    # Now process each expanded URI
    chains = []
    listed_ds_name = set()
    file_values = []

    updated_uris = set()

    for single_uri in expanded_uris:
        # Check if URI contains glob patterns and split them
        base_uri, glob_pattern = split_uri_pattern(single_uri)

        # If a pattern is found, use the base_uri for listing
        # The pattern will be used for filtering later
        list_uri_to_use = base_uri if glob_pattern else single_uri

        # Avoid double updates for the same URI
        update_single_uri = False
        if update and (list_uri_to_use not in updated_uris):
            updated_uris.add(list_uri_to_use)
            update_single_uri = True

        list_ds_name, list_uri, list_path, _ = get_listing(
            list_uri_to_use, session, update=update_single_uri, versions=versions
        )

        # list_ds_name is None if object is a file, we don't want to use cache
        # or do listing in that case - just read that single object
        if not list_ds_name:
            file_values.append(
                get_file_info(list_uri, cache, client_config=client_config)
            )
            continue

        dc = read_dataset(
            list_ds_name,
            namespace=listing_namespace_name,
            project=listing_project_name,
            session=session,
            settings=settings,
            delta=delta,
            delta_on=delta_on,
            delta_result_on=delta_result_on,
            delta_compare=delta_compare,
            delta_retry=delta_retry,
            delta_unsafe=delta_unsafe,
        )
        dc._query.update = update
        dc.signals_schema = dc.signals_schema.mutate({f"{column}": file_type})

        def lst_fn(ds_name, lst_uri):
            # Seed for .gen() iteration. content_hash=None because hash_callable
            # doesn't capture list_func's closure (which holds `lst_uri`, `cache`,
            # `client_config`) -- auto-hashing the seed would let the UDF
            # checkpoint cache return a stale listing across URIs/runs.
            (
                create_records_dataset(
                    [{"seed": 0}],
                    schema={"seed": int},
                    content_hash=None,
                    session=session,
                    settings=settings,
                    in_memory=in_memory,
                )
                .settings(
                    prefetch=0,
                    namespace=listing_namespace_name,
                    project=listing_project_name,
                )
                .gen(
                    list_bucket(lst_uri, cache, client_config=client_config),
                    output={f"{column}": file_type},
                )
                # for internal listing datasets, we always bump major version
                .save(ds_name, listing=True, update_version="major")
            )

        # Always attach listing_fn so resolve_listing can refresh stale listings.
        dc._query.set_listing_fn(
            lambda ds_name=list_ds_name, lst_uri=list_uri: lst_fn(ds_name, lst_uri)
        )

        # If a glob pattern was detected, use it for filtering
        # Otherwise, use the original list_path from get_listing
        if glob_pattern:
            # Determine if we should use recursive listing based on the pattern
            use_recursive = should_use_recursion(glob_pattern, recursive or False)

            # Apply glob filter - no need for brace expansion here as it's done above
            chain = apply_glob_filter(
                dc, glob_pattern, list_path, use_recursive, column
            )
            chains.append(chain)
        else:
            # No glob pattern detected, use normal ls behavior
            chains.append(ls(dc, list_path, recursive=recursive, column=column))

        listed_ds_name.add(list_ds_name)

    storage_chain = None if not chains else reduce(lambda x, y: x.union(y), chains)

    if file_values:
        # Use read_records directly (not read_values) so the chain hash is
        # derived from the flattened File records (deterministic across runs) —
        # needed for checkpoint reuse on single-file read_storage /
        # read_csv / read_parquet.
        file_chain = read_records(
            [{"file": f} for f in file_values],
            schema={"file": file_type},
            session=session,
            settings=settings,
            in_memory=in_memory,
        )
        file_chain.signals_schema = file_chain.signals_schema.mutate(
            {f"{column}": file_type}
        )
        storage_chain = storage_chain.union(file_chain) if storage_chain else file_chain

    assert storage_chain is not None

    return storage_chain