Add SQL and physical planner support for MERGE INTO

.github/workflows/audit.yml

@@ -45,7 +45,7 @@ jobs:
     steps:
       - uses: actions/checkout@df4cb1c069e1874edd31b4311f1884172cec0e10  # v6.0.3
       - name: Install cargo-audit
-        uses: taiki-e/install-action@0631aa6515c7d545823c67cfae7ef4fc7f490154  # v2.81.8
+        uses: taiki-e/install-action@15449e3094499af05d8d964a1c884208e4b8b595  # v2.81.11
         with:
           tool: cargo-audit
       - name: Run audit check

.github/workflows/breaking_changes_detector.yml

@@ -89,7 +89,7 @@ jobs:
 
       - name: Install cargo-semver-checks
         if: steps.changed_crates.outputs.packages != ''
-        uses: taiki-e/install-action@0631aa6515c7d545823c67cfae7ef4fc7f490154  # v2.81.8
+        uses: taiki-e/install-action@15449e3094499af05d8d964a1c884208e4b8b595  # v2.81.11
         with:
           tool: cargo-semver-checks
 

.github/workflows/dependencies.yml

@@ -50,9 +50,9 @@ jobs:
         with:
           rust-version: stable
       - name: Check dependencies
+        working-directory: dev/depcheck
         run: |
-          cd dev/depcheck
-          cargo run
+          cargo run --locked
 
   detect-unused-dependencies:
     name: Detect Unused Dependencies

.github/workflows/dev.yml

@@ -64,7 +64,7 @@ jobs:
           source ci/scripts/utils/tool_versions.sh
           echo "LYCHEE_VERSION=${LYCHEE_VERSION}" >> "$GITHUB_ENV"
       - name: Install lychee
-        uses: taiki-e/install-action@0631aa6515c7d545823c67cfae7ef4fc7f490154  # v2.81.8
+        uses: taiki-e/install-action@15449e3094499af05d8d964a1c884208e4b8b595  # v2.81.11
         with:
           tool: lychee@${{ env.LYCHEE_VERSION }}
       - name: Run markdown link check

.github/workflows/rust.yml

@@ -429,7 +429,7 @@ jobs:
           sudo apt-get update -qq
           sudo apt-get install -y -qq clang
       - name: Setup wasm-pack
-        uses: taiki-e/install-action@0631aa6515c7d545823c67cfae7ef4fc7f490154  # v2.81.8
+        uses: taiki-e/install-action@15449e3094499af05d8d964a1c884208e4b8b595  # v2.81.11
         with:
           tool: wasm-pack
       - name: Run tests with headless mode
@@ -773,7 +773,7 @@ jobs:
       - name: Setup Rust toolchain
         uses: ./.github/actions/setup-builder
       - name: Install cargo-msrv
-        uses: taiki-e/install-action@0631aa6515c7d545823c67cfae7ef4fc7f490154  # v2.81.8
+        uses: taiki-e/install-action@15449e3094499af05d8d964a1c884208e4b8b595  # v2.81.11
         with:
           tool: cargo-msrv
 

datafusion-cli/Cargo.toml

@@ -75,7 +75,7 @@ workspace = true
 [dev-dependencies]
 ctor = { workspace = true }
 insta = { workspace = true }
-insta-cmd = "0.6.0"
+insta-cmd = "0.7.0"
 rstest = { workspace = true }
 testcontainers-modules = { workspace = true, features = ["minio"] }
 # Makes sure `test_display_pg_json` behaves in a consistent way regardless of

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