@spec close(t()) :: :ok

Release resources held by a stream.

For :parquet_multi streams this stops the backing Agent (and drops the open Parquet handle). Other backends are GC-safe and this is a no-op.

@spec from_adbc(ExArrow.ADBC.Statement.t()) :: {:ok, t()} | {:error, term()}

Execute an ADBC statement and return its result as a stream of record batches.

Accepts either a prepared ExArrow.ADBC.Statement.t() (built with ExArrow.ADBC.Statement.new/2 or new/3) or a {connection, sql} pair, in which case a one-shot statement is created, executed, and discarded.

The stream is tagged with source: {:adbc, sql} (or :statement when a pre-built statement is passed).

Examples

{:ok, stream} = ExArrow.Stream.from_adbc(stmt)

{:ok, stream} = ExArrow.Stream.from_adbc(conn, "SELECT * FROM events")

@spec from_flight(ExArrow.Flight.Client.t(), term()) :: {:ok, t()} | {:error, term()}

Retrieve data for ticket from a Flight server as a stream of record batches.

Delegates to ExArrow.Flight.Client.do_get/2 and emits a [:ex_arrow, :flight, :query] telemetry event on success. The stream is tagged with source: {:flight, ticket}.

Example

{:ok, stream} = ExArrow.Stream.from_flight(client, "sales_2024")

@spec from_flight_sql(ExArrow.FlightSQL.Client.t(), String.t()) ::
  {:ok, t()} | {:error, term()}

Execute a SQL query against a Flight SQL server and return a lazy stream of record batches.

Delegates to ExArrow.FlightSQL.Client.stream_query/2, emits a [:ex_arrow, :flight_sql, :query] telemetry event on success, and tags the stream with source: {:flight_sql, sql}.

Example

{:ok, stream} = ExArrow.Stream.from_flight_sql(client, "SELECT * FROM events")

@spec from_ipc(binary()) :: {:ok, t()} | {:error, String.t()}

Open an Arrow IPC stream from an in-memory binary.

Delegates to ExArrow.IPC.Reader.from_binary/1 and tags the resulting stream with source: {:ipc, :binary} for telemetry. Returns {:ok, stream} or {:error, message}.

Example

{:ok, stream} = ExArrow.Stream.from_ipc(ipc_bytes)

@spec from_ipc_file(Path.t()) :: {:ok, t()} | {:error, String.t()}

Open an Arrow IPC stream from a file at path.

Delegates to ExArrow.IPC.Reader.from_file/1 and tags the stream with source: {:ipc, path}. Returns {:ok, stream} or {:error, message}.

@spec from_parquet(
  Path.t(),
  keyword()
) :: {:ok, t()} | {:error, String.t()}

Open a Parquet file at path for lazy row-group streaming.

Delegates to ExArrow.Parquet.Reader.from_file/2. Pushdown opts (:columns, :row_groups, :filters) are forwarded. Emits [:ex_arrow, :parquet, :read] with row-group selection stats when available.

Example

{:ok, stream} =
  ExArrow.Stream.from_parquet("/data/events.parquet",
    columns: ["id"],
    filters: {:gt, "id", 10}
  )

@spec from_parquet_binary(
  binary(),
  keyword()
) :: {:ok, t()} | {:error, String.t()}

Open a Parquet stream from an in-memory binary.

See from_parquet/2 for pushdown opts.

@spec from_parquet_dir(
  Path.t(),
  keyword()
) :: {:ok, t()} | {:error, String.t()}

Stream all *.parquet files under dir (non-recursive), sorted lexicographically.

Options are the same as from_parquet/2.

@spec from_parquet_files(
  [Path.t()],
  keyword()
) :: {:ok, t()} | {:error, String.t()}

Lazily stream multiple Parquet files in lexicographic path order.

Files are opened one at a time — early Enum.take/2 does not open later files. Schema field names of each subsequent file must match the first file or an error is returned from next/1.

Pushdown opts are applied to every file.

Partially consumed multi-file streams hold an Agent and an open file handle; call close/1 when abandoning the stream early (e.g. after Enum.take/2) from a long-lived process.

@spec next(t()) ::
  ExArrow.RecordBatch.t()
  | nil
  | {:error, String.t()}
  | {:error, {atom(), non_neg_integer(), String.t()}}

Returns the next record batch from the stream, or nil when done. Returns {:error, message} on read error.

For :flight_sql streams, read errors carry a structured 3-tuple so callers can distinguish gRPC codes:

{:error, {code_atom, grpc_status_integer, message}} |
{:error, string_message}

Enum.* / Stream.* functions raise on any error shape. For recoverable error handling iterate with next/1 directly.

@spec schema(t()) :: {:ok, ExArrow.Schema.t()} | {:error, String.t()}

Returns the schema of this stream (without consuming it). Returns {:error, message} if the stream is invalid (e.g. poisoned lock).

@spec source(t()) :: term()

Returns the origin metadata attached to this stream by the from_*/ constructors. The value is backend-specific (e.g. {:parquet, path} or {:flight_sql, sql}) and is forwarded to telemetry events as :source.

@spec to_list(t()) :: [ExArrow.RecordBatch.t()]

Collects all remaining batches from the stream into a list.

Stops at the first error and raises. Returns an empty list for an already-exhausted stream.