feat: support reading from stdin in datafusion-cli (#22839)

## Which issue does this PR close?

- Closes #9430.

## Rationale for this change

Users frequently want to pipe data into the CLI, e.g. `cat data.csv |
datafusion-cli`, but pointing `LOCATION` at `/dev/stdin` did not work:

- CSV failed with `Illegal seek` (a pipe is not seekable).
- Parquet failed with `file size of 0 is less than footer` (a pipe
reports size 0).
- JSON silently returned 0 rows.

This PR makes reading from standard input work for CSV, JSON, and
Parquet.

## What changes are included in this PR?

stdin is exposed as a `stdin://` object store, dispatched alongside the
other schemes (`s3`, `gs`, `http`, ...) in `get_object_store` —
conceptually similar to DuckDB's `PipeFileSystem`.

- `rewrite_stdin_location` maps the well-known stdin pseudo-paths
(`/dev/stdin`, `/dev/fd/0`, `/proc/self/fd/0`) to a canonical
`stdin:///stdin.<ext>` URL, so they flow through the normal
object-store/listing code path. The extension matches the declared
`STORED AS` format because the listing layer filters candidate files by
extension.
- The `stdin://` store reads all of standard input into an in-memory
object store. Buffering up front is required because a pipe is not
seekable and Parquet stores its metadata at the end of the file.

Known scope/limitations (left as potential follow-ups):
- Only `CREATE EXTERNAL TABLE` is supported (not dynamic `SELECT * FROM
'/dev/stdin'`).
- Input is fully buffered in memory, so it must fit in memory.
- stdin can only be consumed once per session.
- Unix-only (`/dev/stdin` does not exist on Windows); writing to
`/dev/stdout` is out of scope.

## Are these changes tested?

Yes:
- Unit tests in `object_storage.rs` cover `rewrite_stdin_location` and
end-to-end reads for CSV, JSON, and Parquet via the in-memory store.
- A `#[cfg(unix)]` integration test in `cli_integration.rs` drives the
real binary through an actual pipe, exercising the real stdin read.
- Manually verified all three formats via real pipes, and confirmed
normal local-file reads are unaffected.

## Are there any user-facing changes?

Yes — reading from stdin via `LOCATION '/dev/stdin'` is now supported.
Documented in `docs/source/user-guide/cli/datasources.md` (new "Reading
from standard input" section). No breaking changes.

Author: huan233usc

datafusion-cli/src/exec.rs

@@ -23,7 +23,7 @@ use crate::print_format::PrintFormat;
 use crate::{
     command::{Command, OutputFormat},
     helper::CliHelper,
-    object_storage::get_object_store,
+    object_storage::{get_object_store, stdin::StdinUtils},
     print_options::{MaxRows, PrintOptions},
 };
 use datafusion::common::instant::Instant;
@@ -417,9 +417,14 @@ async fn create_plan(
     // Note that cmd is a mutable reference so that create_external_table function can remove all
     // datafusion-cli specific options before passing through to datafusion. Otherwise, datafusion
     // will raise Configuration errors.
-    if let LogicalPlan::Ddl(DdlStatement::CreateExternalTable(cmd)) = &plan {
+    if let LogicalPlan::Ddl(DdlStatement::CreateExternalTable(cmd)) = &mut plan {
         // To support custom formats, treat error as None
         let format = config_file_type_from_str(&cmd.file_type);
+
+        // Expose stdin (e.g. `cat data.csv | datafusion-cli`) as a `stdin://`
+        // object store, registered like any other scheme in `get_object_store`.
+        cmd.location = StdinUtils::rewrite_location(&cmd.location, format.as_ref());
+
         register_object_store_and_config_extensions(
             ctx,
             &cmd.location,

datafusion-cli/src/main.rs

@@ -37,6 +37,7 @@ use datafusion_cli::functions::{
 use datafusion_cli::object_storage::instrumented::{
     InstrumentedObjectStoreMode, InstrumentedObjectStoreRegistry,
 };
+use datafusion_cli::object_storage::{StdinCarriesCommands, is_stdin_location};
 use datafusion_cli::{
     DATAFUSION_CLI_VERSION, exec,
     pool_type::PoolType,
@@ -158,6 +159,23 @@ struct Args {
     object_store_profiling: InstrumentedObjectStoreMode,
 }
 
+impl Args {
+    /// Without -c/-f the CLI enters the REPL, which reads its SQL from
+    /// stdin — interactively or piped.
+    fn repl_mode(&self) -> bool {
+        self.command.is_empty() && self.file.is_empty()
+    }
+
+    /// Whether the CLI consumes stdin for its own SQL input. This covers the
+    /// REPL (no -c/-f, reading SQL interactively or piped) as well as an
+    /// explicit `-f /dev/stdin` (or the other stdin pseudo-paths), where the
+    /// SQL file *is* stdin. In either case stdin is already spoken for and
+    /// cannot also back a `LOCATION '/dev/stdin'` table.
+    fn reads_sql_from_stdin(&self) -> bool {
+        self.repl_mode() || self.file.iter().any(|f| is_stdin_location(f))
+    }
+}
+
 #[tokio::main]
 /// Calls [`main_inner`], then handles printing errors and returning the correct exit code
 pub async fn main() -> ExitCode {
@@ -268,6 +286,7 @@ async fn main_inner() -> Result<()> {
         instrumented_registry: Arc::clone(&instrumented_registry),
     };
 
+    let repl_mode = args.repl_mode();
     let commands = args.command;
     let files = args.file;
     let rc = match args.rc {
@@ -285,7 +304,7 @@ async fn main_inner() -> Result<()> {
         }
     };
 
-    if commands.is_empty() && files.is_empty() {
+    if repl_mode {
         if !rc.is_empty() {
             exec::exec_from_files(&ctx, rc, &print_options).await?;
         }
@@ -330,8 +349,16 @@ fn get_session_config(args: &Args) -> Result<SessionConfig> {
         config_options.format.null = String::from("NULL");
     }
 
-    let session_config =
+    let mut session_config =
         SessionConfig::from(config_options).with_information_schema(true);
+
+    if args.reads_sql_from_stdin() {
+        // When stdin carries the session's SQL — the REPL (including any rc
+        // file run before it) or an explicit `-f /dev/stdin` — it cannot also
+        // serve as a data source for `LOCATION '/dev/stdin'`.
+        session_config = session_config.with_extension(Arc::new(StdinCarriesCommands));
+    }
+
     Ok(session_config)
 }
 

datafusion-cli/src/object_storage.rs

@@ -16,6 +16,9 @@
 // under the License.
 
 pub mod instrumented;
+pub(crate) mod stdin;
+
+pub use stdin::{StdinCarriesCommands, is_stdin_location};
 
 use async_trait::async_trait;
 use aws_config::BehaviorVersion;
@@ -564,6 +567,9 @@ pub(crate) async fn get_object_store(
                 .with_url(url.origin().ascii_serialization())
                 .build()?,
         ),
+        _ if scheme == stdin::StdinUtils::SCHEME => {
+            stdin::StdinUtils::get_or_create(state, url).await?
+        }
         _ => {
             // For other types, try to get from `object_store_registry`:
             state