diff --git a/native/core/src/execution/operators/parquet_writer.rs b/native/core/src/execution/operators/parquet_writer.rs
index 035adf754b8..f302420bb7c 100644
--- a/native/core/src/execution/operators/parquet_writer.rs
+++ b/native/core/src/execution/operators/parquet_writer.rs
@@ -33,9 +33,11 @@ use std::io::Cursor;
 
 use crate::jvm_bridge::OutputStreamSink;
 use crate::parquet::datetime_rebase::SPARK_VERSION_METADATA_KEY;
-use crate::parquet::parquet_support::is_hdfs_scheme;
+use crate::parquet::objectstore::s3_blob_fs_support::scheme_of_unparsed;
+use crate::parquet::objectstore::url_redaction::{without_credentials, without_secrets};
 #[cfg(feature = "hdfs-opendal")]
 use crate::parquet::parquet_support::{create_hdfs_operator, prepare_object_store_with_configs};
+use crate::parquet::parquet_support::{is_azure_scheme, is_hdfs_scheme};
 use arrow::datatypes::{Schema, SchemaRef};
 use arrow::record_batch::{RecordBatch, RecordBatchOptions};
 use async_trait::async_trait;
@@ -100,6 +102,7 @@ enum ParquetWriter {
         arrow_writer: Box<ArrowWriter<Cursor<Vec<u8>>>>,
         hdfs_writer: Option<opendal::Writer>,
         op: Box<Operator>,
+        /// The path inside the store, without the URL's scheme, host, userinfo or query.
         output_path: String,
     },
     /// Writer for a file the JVM created through Hadoop's FileSystem API (`HadoopFileOutput`),
@@ -256,6 +259,37 @@ impl ParquetWriter {
     }
 }
 
+/// `output_path` for a message, the plan or the log. A path with no password, query, fragment or
+/// username (outside ABFS, where the username is the container) is shown exactly as given; one
+/// with any of them shows only its scheme, host and path. A string that does not parse is shown
+/// as given when it has no `://`, a local path with no userinfo, and as its scheme otherwise.
+fn display_output_path(output_path: &str) -> String {
+    let url = match Url::parse(output_path) {
+        Ok(url) => url,
+        Err(_) if !output_path.contains("://") => return output_path.to_string(),
+        Err(_) => {
+            return match scheme_of_unparsed(output_path) {
+                Some(scheme) => format!("{scheme}://<unparseable>"),
+                None => "<unparseable>".to_string(),
+            }
+        }
+    };
+    let keeps_username = is_azure_scheme(url.scheme());
+    let has_secret = url.password().is_some()
+        || url.query().is_some()
+        || url.fragment().is_some()
+        || (!url.username().is_empty() && !keeps_username);
+    if !has_secret {
+        return output_path.to_string();
+    }
+    let shown = if keeps_username {
+        without_secrets(&url)
+    } else {
+        without_credentials(&url)
+    };
+    shown.to_string()
+}
+
 /// The task's error for `e`, a Parquet write that failed. A failed write to the JVM output stream
 /// comes back as the stream's own error, which keeps the Java exception it carries, so the JVM
 /// rethrows that exception as it was. Any other failure is reported with `what` and its text.
@@ -470,7 +504,11 @@ impl ParquetWriterExec {
 
         // Parse URL and match on storage scheme directly
         let url = Url::parse(output_file_path).map_err(|e| {
-            DataFusionError::Execution(format!("Failed to parse URL '{}': {}", output_file_path, e))
+            DataFusionError::Execution(format!(
+                "Failed to parse URL '{}': {}",
+                display_output_path(output_file_path),
+                e
+            ))
         })?;
 
         if is_hdfs_scheme(&url, object_store_options) {
@@ -485,7 +523,8 @@ impl ParquetWriterExec {
                 .map_err(|e| {
                     DataFusionError::Execution(format!(
                         "Failed to prepare object store for '{}': {}",
-                        output_file_path, e
+                        display_output_path(output_file_path),
+                        e
                     ))
                 })?;
 
@@ -501,7 +540,8 @@ impl ParquetWriterExec {
                 let op = create_hdfs_operator(&url).map_err(|e| {
                     DataFusionError::Execution(format!(
                         "Failed to create HDFS operator for '{}': {}",
-                        output_file_path, e
+                        display_output_path(output_file_path),
+                        e
                     ))
                 })?;
 
@@ -566,7 +606,7 @@ impl ParquetWriterExec {
             // Unsupported storage scheme
             Err(DataFusionError::Execution(format!(
                 "Unsupported storage scheme in path: {}",
-                output_file_path
+                display_output_path(output_file_path)
             )))
         }
     }
@@ -579,7 +619,8 @@ impl DisplayAs for ParquetWriterExec {
                 write!(
                     f,
                     "ParquetWriterExec: path={}, compression={:?}",
-                    self.output_path, self.compression
+                    display_output_path(&self.output_path),
+                    self.compression
                 )
             }
             DisplayFormatType::TreeRender => unimplemented!(),
@@ -772,7 +813,7 @@ impl ExecutionPlan for ParquetWriterExec {
 
             log::debug!(
                 "Wrote Parquet file: path={}, size={}, rows={}",
-                part_file,
+                display_output_path(&part_file),
                 file_size,
                 total_rows
             );
@@ -1418,6 +1459,175 @@ mod tests {
         );
     }
 
+    /// Userinfo, query and fragment markers the test paths carry, none of which may be shown.
+    const MARKERS: [&str; 4] = ["alice", "SECRET", "SAS-MARKER", "FRAG-MARKER"];
+
+    fn assert_no_marker(shown: &str) {
+        for marker in MARKERS {
+            assert!(!shown.contains(marker), "{marker} echoed: {shown}");
+        }
+    }
+
+    #[test]
+    fn display_output_path_keeps_scheme_host_and_path() {
+        let cases = [
+            (
+                "hdfs://alice:SECRET@nn:8020/t/p.parquet?sig=SAS-MARKER#FRAG-MARKER",
+                "hdfs://nn:8020/t/p.parquet",
+            ),
+            // Each part that can carry a credential, alone.
+            (
+                "hdfs://:SECRET@nn:8020/t/p.parquet",
+                "hdfs://nn:8020/t/p.parquet",
+            ),
+            (
+                "hdfs://nn:8020/t/p.parquet?sig=SAS-MARKER",
+                "hdfs://nn:8020/t/p.parquet",
+            ),
+            (
+                "hdfs://nn:8020/t/p.parquet#FRAG-MARKER",
+                "hdfs://nn:8020/t/p.parquet",
+            ),
+            (
+                "hdfs://alice@nn:8020/t/p.parquet",
+                "hdfs://nn:8020/t/p.parquet",
+            ),
+            (
+                "s3a://alice:SECRET@bucket/t/p.parquet?sig=SAS-MARKER",
+                "s3a://bucket/t/p.parquet",
+            ),
+            // An ABFS username is the container, so only the password goes.
+            (
+                "abfss://data:SECRET@acct.dfs.core.windows.net/t/p.parquet?sig=SAS-MARKER",
+                "abfss://data@acct.dfs.core.windows.net/t/p.parquet",
+            ),
+            (
+                "hdfs://alice:SECRET@nn:notaport/t/p.parquet?sig=SAS-MARKER",
+                "hdfs://<unparseable>",
+            ),
+            (
+                "1x://alice:SECRET@nn/t/p.parquet?sig=SAS-MARKER",
+                "<unparseable>",
+            ),
+            // Nothing to hide: shown as Spark gave it, not re-serialized.
+            ("hdfs://nn:8020/t/p.parquet", "hdfs://nn:8020/t/p.parquet"),
+            (
+                "abfss://data@acct.dfs.core.windows.net/my dir/p.parquet",
+                "abfss://data@acct.dfs.core.windows.net/my dir/p.parquet",
+            ),
+            ("file:///tmp/t/p.parquet", "file:///tmp/t/p.parquet"),
+            ("file:/tmp/t/p.parquet", "file:/tmp/t/p.parquet"),
+            ("file:/tmp/my dir/p.parquet", "file:/tmp/my dir/p.parquet"),
+            ("/tmp/t/p.parquet", "/tmp/t/p.parquet"),
+        ];
+        for (output_path, expected) in cases {
+            let shown = display_output_path(output_path);
+            assert_no_marker(&shown);
+            assert_eq!(shown, expected);
+        }
+    }
+
+    #[test]
+    fn create_arrow_writer_errors_do_not_echo_credentials() -> Result<()> {
+        let cases = [
+            (
+                "s3a://alice:SECRET@bucket/t/part-0.parquet?sig=SAS-MARKER#FRAG-MARKER",
+                "Unsupported storage scheme in path: s3a://bucket/t/part-0.parquet",
+            ),
+            // `Url::parse` fails before the scheme is looked at, so no HDFS is needed.
+            (
+                "hdfs://alice:SECRET@nn:notaport/t/part-0.parquet?sig=SAS-MARKER",
+                "Failed to parse URL 'hdfs://<unparseable>': invalid port number",
+            ),
+        ];
+        let schema = Arc::new(Schema::new(vec![Field::new("a", DataType::Int32, true)]));
+        for (output_path, expected) in cases {
+            let error = match ParquetWriterExec::create_arrow_writer(
+                output_path,
+                Arc::clone(&schema),
+                WriterProperties::default(),
+                RuntimeEnvBuilder::new().build_arc()?,
+                &HashMap::new(),
+                None,
+            ) {
+                Ok(_) => panic!("no writer expected for {output_path}"),
+                Err(e) => e.to_string(),
+            };
+            assert_no_marker(&error);
+            assert!(error.contains(expected), "{error}");
+        }
+        Ok(())
+    }
+
+    #[test]
+    fn plan_display_does_not_echo_credentials() -> Result<()> {
+        let writer = writer_of(
+            vec![create_test_record_batch(1)?],
+            "hdfs://alice:SECRET@nn:8020/t/p.parquet?sig=SAS-MARKER#FRAG-MARKER".to_string(),
+            ParquetCompression::Snappy,
+        )?;
+        let shown = datafusion::physical_plan::displayable(&writer)
+            .one_line()
+            .to_string();
+        assert_no_marker(&shown);
+        assert!(
+            shown.contains("ParquetWriterExec: path=hdfs://nn:8020/t/p.parquet,"),
+            "{shown}"
+        );
+        Ok(())
+    }
+
+    /// Keeps the lines this module logs, for checking what the writer logs.
+    struct WriterLog(std::sync::Mutex<Vec<String>>);
+
+    impl log::Log for WriterLog {
+        fn enabled(&self, metadata: &log::Metadata) -> bool {
+            metadata.target() == module_path!().trim_end_matches("::tests")
+        }
+
+        fn log(&self, record: &log::Record) {
+            if self.enabled(record.metadata()) {
+                self.0.lock().unwrap().push(record.args().to_string());
+            }
+        }
+
+        fn flush(&self) {}
+    }
+
+    // `log::set_logger` succeeds once per process, so this must stay the only logger the test
+    // binary installs.
+    static WRITER_LOG: WriterLog = WriterLog(std::sync::Mutex::new(Vec::new()));
+
+    #[tokio::test]
+    async fn test_parquet_writer_log_does_not_echo_credentials() -> Result<()> {
+        log::set_logger(&WRITER_LOG).expect("no other logger is installed in tests");
+        log::set_max_level(log::LevelFilter::Debug);
+        let sink = Arc::new(RecordingSink::new(4096));
+        write_batches(
+            vec![create_test_record_batch(1)?],
+            "s3a://alice:SECRET@log-bucket/t/part-0.parquet?sig=SAS-MARKER#FRAG-MARKER".to_string(),
+            Some(sink as Arc<dyn OutputStreamSink>),
+        )
+        .await?;
+
+        let lines: Vec<String> = WRITER_LOG
+            .0
+            .lock()
+            .unwrap()
+            .iter()
+            .filter(|line| line.contains("log-bucket"))
+            .cloned()
+            .collect();
+        assert_eq!(lines.len(), 1, "{lines:?}");
+        assert_no_marker(&lines[0]);
+        assert!(
+            lines[0].starts_with("Wrote Parquet file: path=s3a://log-bucket/t/part-0.parquet,"),
+            "{}",
+            lines[0]
+        );
+        Ok(())
+    }
+
     #[tokio::test]
     #[cfg(feature = "hdfs-opendal")]
     #[ignore = "This test requires a running HDFS cluster"]
diff --git a/native/core/src/parquet/objectstore/azure.rs b/native/core/src/parquet/objectstore/azure.rs
index 8b18ac70250..d0f8f598c98 100644
--- a/native/core/src/parquet/objectstore/azure.rs
+++ b/native/core/src/parquet/objectstore/azure.rs
@@ -78,7 +78,7 @@
 
 use log::debug;
 use std::collections::HashMap;
-use url::{Position, Url};
+use url::Url;
 
 use object_store::{
     azure::{AzureConfigKey, MicrosoftAzureBuilder},
@@ -86,6 +86,8 @@ use object_store::{
     ObjectStore, ObjectStoreScheme,
 };
 
+use super::url_redaction::{display_authority, without_credentials, without_secrets};
+
 const MARKER_RESOLUTION: &str = "comet.azure.resolution";
 const MARKER_AUTH_TYPE: &str = "comet.azure.auth.type";
 const MARKER_OAUTH_PROVIDER_CLASS: &str = "comet.azure.oauth.provider.class";
@@ -127,9 +129,12 @@ pub fn create_store(
     url: &Url,
     configs: &HashMap<String, String>,
 ) -> Result<(Box<dyn ObjectStore>, Path), object_store::Error> {
-    let (scheme, path) = ObjectStoreScheme::parse(url)?;
+    let (scheme, path) = ObjectStoreScheme::parse(&without_credentials(url))?;
     if scheme != ObjectStoreScheme::MicrosoftAzure {
-        return Err(config_error(format!("Scheme of URL is not Azure: {url}")));
+        return Err(config_error(format!(
+            "Scheme of URL is not Azure: {}",
+            url.scheme()
+        )));
     }
     let path = Path::parse(path)?;
     let store = builder_for(url, configs)?.build()?;
@@ -159,17 +164,21 @@ fn builder_for(
     let marker = |key: &str| configs.get(key).map(String::as_str);
     match marker(MARKER_RESOLUTION) {
         Some("none") => {
-            debug!("Azure authentication for {url}: none configured in Hadoop, reading AZURE_*");
-            Ok(MicrosoftAzureBuilder::from_env().with_url(url.to_string()))
+            debug!(
+                "Azure authentication for {}: none configured in Hadoop, reading AZURE_*",
+                display_authority(url)
+            );
+            Ok(MicrosoftAzureBuilder::from_env().with_url(without_secrets(url).to_string()))
         }
         Some("resolved") => {
             let settings = resolved_settings(configs)?;
             debug!(
-                "Azure authentication for {url}: {} with keys {:?}",
+                "Azure authentication for {}: {} with keys {:?}",
+                display_authority(url),
                 marker(MARKER_AUTH_TYPE).unwrap_or_default(),
                 settings.iter().map(|(k, _)| k.as_ref()).collect::<Vec<_>>()
             );
-            let builder = MicrosoftAzureBuilder::new().with_url(url.to_string());
+            let builder = MicrosoftAzureBuilder::new().with_url(without_secrets(url).to_string());
             Ok(settings
                 .into_iter()
                 .fold(builder, |b, (key, value)| b.with_config(key, value)))
@@ -184,9 +193,8 @@ fn builder_for(
             ))
         }
         Some("error") => Err(config_error(format!(
-            "Hadoop ABFS authentication failed on the driver for {}://{}: {}: {}",
-            url.scheme(),
-            &url[Position::BeforeUsername..Position::AfterPort],
+            "Hadoop ABFS authentication failed on the driver for {}: {}: {}",
+            display_authority(url),
             marker(MARKER_ERROR_CLASS).unwrap_or("<unknown>"),
             marker(MARKER_ERROR_MESSAGE).unwrap_or("<unknown>"),
         ))),
@@ -570,6 +578,54 @@ mod tests {
         assert!(err.contains("Scheme of URL is not Azure"), "{err}");
     }
 
+    #[test]
+    fn create_store_errors_do_not_echo_userinfo_or_query() {
+        let configs = with(shared_key(), HADOOP_KEY, KEY);
+        let error_for = |raw: String| {
+            let url = Url::parse(&raw).unwrap();
+            let err = create_store(&url, &configs).unwrap_err().to_string();
+            for leaked in ["alice", SECRET, "SAS-MARKER"] {
+                assert!(!err.contains(leaked), "{leaked} echoed: {err}");
+            }
+            err
+        };
+        let not_azure = error_for(format!(
+            "s3://alice:{SECRET}@bucket/dir/file.parquet?sig=SAS-MARKER"
+        ));
+        assert!(
+            not_azure.ends_with("Scheme of URL is not Azure: s3"),
+            "{not_azure}"
+        );
+        assert!(!not_azure.contains("bucket"), "host echoed: {not_azure}");
+        error_for(format!(
+            "foo://alice:{SECRET}@host/dir/file.parquet?sig=SAS-MARKER"
+        ));
+    }
+
+    #[test]
+    fn builder_errors_do_not_echo_the_password_or_query() {
+        let configs = with(shared_key(), HADOOP_KEY, KEY);
+        // object_store recognises only the public cloud and Fabric hosts, and quotes the URL.
+        let url = Url::parse(&format!(
+            "abfss://data:{SECRET}@myacct.dfs.core.chinacloudapi.cn/path/file.parquet?sig=SAS-MARKER"
+        ))
+        .unwrap();
+        let err = create_store(&url, &configs).unwrap_err().to_string();
+        for leaked in [SECRET, "SAS-MARKER"] {
+            assert!(!err.contains(leaked), "{leaked} echoed: {err}");
+        }
+    }
+
+    #[test]
+    fn create_store_keeps_the_container_of_a_url_with_a_query() {
+        let configs = with(shared_key(), HADOOP_KEY, KEY);
+        let url =
+            Url::parse("abfss://data@myacct.dfs.core.windows.net/path/file.parquet?sig=SAS-MARKER")
+                .unwrap();
+        let (_store, path) = create_store(&url, &configs).expect("store builds");
+        assert_eq!(path.as_ref(), "path/file.parquet");
+    }
+
     #[test]
     fn store_error_answers_whether_create_store_builds() {
         let shared_key = with(shared_key(), HADOOP_KEY, KEY);
@@ -923,6 +979,19 @@ mod tests {
              org.apache.hadoop.fs.azurebfs.contracts.exceptions.KeyProviderException: \
              Failure to initialize configuration"
         );
+        let with_password = Url::parse(&format!(
+            "abfss://data:{SECRET}@myacct.dfs.core.windows.net/path/file.parquet"
+        ))
+        .unwrap();
+        let err = builder_for(&with_password, &configs)
+            .err()
+            .map(|e| e.to_string())
+            .expect("an error");
+        assert!(!err.contains(SECRET), "password echoed: {err}");
+        assert!(
+            err.contains("for abfss://data@myacct.dfs.core.windows.net: "),
+            "{err}"
+        );
     }
 
     #[test]
diff --git a/native/core/src/parquet/objectstore/mod.rs b/native/core/src/parquet/objectstore/mod.rs
index fe7f38040e9..e1b71610139 100644
--- a/native/core/src/parquet/objectstore/mod.rs
+++ b/native/core/src/parquet/objectstore/mod.rs
@@ -20,3 +20,4 @@ pub(crate) mod http_metrics;
 pub mod location_scoped;
 pub mod s3;
 pub mod s3_blob_fs_support;
+pub(crate) mod url_redaction;
diff --git a/native/core/src/parquet/objectstore/s3_blob_fs_support.rs b/native/core/src/parquet/objectstore/s3_blob_fs_support.rs
index ab09e7d24c3..1a1e9a0dfcb 100644
--- a/native/core/src/parquet/objectstore/s3_blob_fs_support.rs
+++ b/native/core/src/parquet/objectstore/s3_blob_fs_support.rs
@@ -77,8 +77,14 @@ pub(crate) fn normalize_object_store_url(
     url_str: &str,
     object_store_configs: &HashMap<String, String>,
 ) -> Result<NormalizedObjectStoreUrl, ExecutionError> {
-    let url = Url::parse(url_str)
-        .map_err(|e| ExecutionError::GeneralError(format!("Error parsing URL {url_str}: {e}")))?;
+    let url = Url::parse(url_str).map_err(|e| {
+        // `url_str` may carry credentials; `ParseError` does not quote it.
+        let message = match scheme_of_unparsed(url_str) {
+            Some(scheme) => format!("Error parsing {scheme} URL: {e}"),
+            None => format!("Error parsing URL: {e}"),
+        };
+        ExecutionError::GeneralError(message)
+    })?;
     if is_hdfs_scheme(&url, object_store_configs) {
         return Ok(NormalizedObjectStoreUrl { url, is_hdfs: true });
     }
@@ -95,6 +101,15 @@ pub(crate) fn normalize_object_store_url(
     })
 }
 
+/// The text before `://` in a URL that failed to parse, if it is a valid scheme.
+pub(crate) fn scheme_of_unparsed(url_str: &str) -> Option<&str> {
+    let (scheme, _) = url_str.split_once("://")?;
+    let mut chars = scheme.chars();
+    let is_scheme = chars.next().is_some_and(|c| c.is_ascii_alphabetic())
+        && chars.all(|c| c.is_ascii_alphanumeric() || matches!(c, '+' | '-' | '.'));
+    is_scheme.then_some(scheme)
+}
+
 /// True if `scheme` is a configured s3-compliant alias (`fs.comet.s3Compliant.schemes`; empty or
 /// unset means none). These route to `AmazonS3` but are not recognized by
 /// `ObjectStoreScheme::parse`. `s3a` is excluded -- object_store knows it and callers special-case
@@ -303,6 +318,28 @@ mod tests {
         configs
     }
 
+    #[test]
+    fn parse_errors_name_only_a_valid_scheme() {
+        let empty = HashMap::new();
+        let error = |url: &str| match normalize_object_store_url(url, &empty) {
+            Err(e) => e.to_string(),
+            Ok(_) => panic!("{url} parsed"),
+        };
+        let err = error("s3a://alice:SECRET@bucket:notaport/x?sig=SAS-MARKER");
+        assert!(
+            err.contains("Error parsing s3a URL: invalid port number"),
+            "{err}"
+        );
+        assert!(!err.contains("alice") && !err.contains("SECRET"), "{err}");
+        // A leading digit, then an '@': neither is a scheme, so none of the input is echoed.
+        let err = error("1alice:SECRET@host://x");
+        assert!(err.contains("Error parsing URL: "), "{err}");
+        assert!(!err.contains("alice") && !err.contains("SECRET"), "{err}");
+        let err = error("alice@host://x");
+        assert!(err.contains("Error parsing URL: "), "{err}");
+        assert!(!err.contains("alice") && !err.contains("host"), "{err}");
+    }
+
     #[test]
     fn test_normalize_object_store_url() {
         let blob = configs_with("blob");
diff --git a/native/core/src/parquet/objectstore/url_redaction.rs b/native/core/src/parquet/objectstore/url_redaction.rs
new file mode 100644
index 00000000000..c94bb06e5ee
--- /dev/null
+++ b/native/core/src/parquet/objectstore/url_redaction.rs
@@ -0,0 +1,51 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! URL forms for messages, store keys and object_store, which quotes URLs in its errors. The
+//! password and query (a SAS token) can carry a credential; the path is not treated as secret.
+
+use std::borrow::Cow;
+use url::{Position, Url};
+
+/// Keeps the username: in an ABFS URL it is the container, which `MicrosoftAzureBuilder` reads.
+pub(crate) fn without_secrets(url: &Url) -> Url {
+    let mut stripped = url.clone();
+    stripped.set_query(None);
+    stripped.set_fragment(None);
+    // Fails only for a URL that cannot have userinfo, so there is nothing to clear.
+    let _ = stripped.set_password(None);
+    stripped
+}
+
+/// [`without_secrets`] minus the username, which outside Azure can be an access key ID.
+pub(crate) fn without_credentials(url: &Url) -> Url {
+    let mut stripped = without_secrets(url);
+    // Fails only for a URL that cannot have userinfo, so there is nothing to clear.
+    let _ = stripped.set_username("");
+    stripped
+}
+
+pub(crate) fn authority_without_password(url: &Url) -> Cow<'_, str> {
+    if url.password().is_none() {
+        return Cow::Borrowed(&url[Position::BeforeUsername..Position::AfterPort]);
+    }
+    Cow::Owned(without_secrets(url)[Position::BeforeUsername..Position::AfterPort].to_string())
+}
+
+pub(crate) fn display_authority(url: &Url) -> String {
+    format!("{}://{}", url.scheme(), authority_without_password(url))
+}
diff --git a/native/core/src/parquet/parquet_support.rs b/native/core/src/parquet/parquet_support.rs
index 42f0bab544f..f3f5ce64f33 100644
--- a/native/core/src/parquet/parquet_support.rs
+++ b/native/core/src/parquet/parquet_support.rs
@@ -47,6 +47,7 @@ use log::debug;
 use object_store::path::Path;
 use object_store::{parse_url, ObjectStore, ObjectStoreScheme};
 use parquet::arrow::PARQUET_FIELD_ID_META_KEY;
+use std::borrow::Cow;
 use std::collections::HashMap;
 use std::sync::OnceLock;
 use std::time::Duration;
@@ -62,6 +63,7 @@ use super::objectstore;
 use super::objectstore::s3_blob_fs_support::{
     normalize_object_store_url, NormalizedObjectStoreUrl,
 };
+use super::objectstore::url_redaction::{authority_without_password, without_credentials};
 
 pub(crate) fn duplicate_parquet_field_error(name: &str) -> DataFusionError {
     DataFusionError::Execution(format!("Found duplicate Parquet field name '{name}'"))
@@ -737,7 +739,7 @@ pub fn is_hdfs_scheme(url: &Url, object_store_configs: &HashMap<String, String>)
 }
 
 /// Check if the scheme is an Azure ABFS URL.
-fn is_azure_scheme(scheme: &str) -> bool {
+pub(crate) fn is_azure_scheme(scheme: &str) -> bool {
     matches!(scheme, "abfs" | "abfss")
 }
 
@@ -747,14 +749,13 @@ fn uses_azure_container_key(scheme: &str) -> bool {
         || scheme.starts_with("abfss+comet-")
 }
 
-pub(crate) fn object_store_authority(url: &Url) -> &str {
-    let start = if uses_azure_container_key(url.scheme()) {
+pub(crate) fn object_store_authority(url: &Url) -> Cow<'_, str> {
+    if uses_azure_container_key(url.scheme()) {
         // ABFS URLs encode the container in the userinfo.
-        url::Position::BeforeUsername
+        authority_without_password(url)
     } else {
-        url::Position::BeforeHost
-    };
-    &url[start..url::Position::AfterPort]
+        Cow::Borrowed(&url[url::Position::BeforeHost..url::Position::AfterPort])
+    }
 }
 
 fn registry_url_key(url: &Url) -> String {
@@ -793,7 +794,10 @@ impl ObjectStoreRegistry for CometObjectStoreRegistry {
                 .unwrap_or_else(PoisonError::into_inner)
                 .remove(&registry_url_key(url))
                 .ok_or_else(|| {
-                    DataFusionError::Internal(format!("No suitable object store found for {url}"))
+                    DataFusionError::Internal(format!(
+                        "No suitable object store found for {}",
+                        registry_url_key(url)
+                    ))
                 })
         } else {
             self.default.deregister_store(url)
@@ -969,8 +973,9 @@ fn object_store_backend(url: &Url, is_hdfs: bool) -> Result<ObjectStoreBackend,
         // Custom libhdfs schemes may look like cloud URLs but retain their own range-read API.
         return Ok(ObjectStoreBackend::Other);
     }
-    let (scheme, _) =
-        ObjectStoreScheme::parse(url).map_err(|e| ExecutionError::GeneralError(e.to_string()))?;
+    // The parser quotes the URL it rejects.
+    let (scheme, _) = ObjectStoreScheme::parse(&without_credentials(url))
+        .map_err(|e| ExecutionError::GeneralError(e.to_string()))?;
     Ok(match scheme {
         ObjectStoreScheme::Local => ObjectStoreBackend::Local,
         ObjectStoreScheme::AmazonS3
@@ -1198,6 +1203,27 @@ mod tests {
             ("gs://bucket/k.parquet", "", "", "gs://bucket", false),
             (abfss_path.as_str(), "", "", abfss, false),
             (abfs_path.as_str(), "", "", abfs, false),
+            (
+                "abfss://container:secret@account.dfs.core.windows.net/k.parquet",
+                "",
+                "",
+                abfss,
+                false,
+            ),
+            (
+                "abfss://container:secret@account.dfs.core.windows.net:8443/k.parquet",
+                "",
+                "",
+                "abfss://container@account.dfs.core.windows.net:8443",
+                false,
+            ),
+            (
+                "abfss://:secret@account.dfs.core.windows.net/k.parquet",
+                "",
+                "",
+                "abfss://account.dfs.core.windows.net",
+                false,
+            ),
             (
                 "wasbs://container@account.blob.core.windows.net/k.parquet",
                 "",
@@ -1428,6 +1454,10 @@ mod tests {
             "http://example.com/path",
             "https://example.com/path",
             "https://account.blob.core.windows.net/container/path",
+            // Stripping the credentials before parsing must not change the class.
+            "s3://alice:SECRET@bucket/path?sig=SAS-MARKER",
+            "abfss://data:SECRET@acct.dfs.core.windows.net/path?sig=SAS-MARKER",
+            "https://alice:SECRET@account.blob.core.windows.net/container/path?sig=SAS-MARKER",
         ] {
             let url = url::Url::parse(address).unwrap();
             assert_eq!(
@@ -2042,6 +2072,154 @@ mod tests {
         ));
     }
 
+    #[test]
+    fn object_store_authority_drops_only_the_abfs_password() {
+        for (url, expected) in [
+            (
+                "abfss://data:SECRET@acct.dfs.core.windows.net/p?sig=SAS-MARKER",
+                "data@acct.dfs.core.windows.net",
+            ),
+            (
+                "abfs://data@acct.dfs.core.windows.net:8443/p",
+                "data@acct.dfs.core.windows.net:8443",
+            ),
+            (
+                "abfss://:SECRET@acct.dfs.core.windows.net/p",
+                "acct.dfs.core.windows.net",
+            ),
+            (
+                "abfss+comet-0123456789abcdef-native://data:SECRET@acct.dfs.core.windows.net/",
+                "data@acct.dfs.core.windows.net",
+            ),
+            (
+                "s3://alice:SECRET@bucket:9000/k?sig=SAS-MARKER",
+                "bucket:9000",
+            ),
+            ("hdfs://nn:8020/t", "nn:8020"),
+        ] {
+            let parsed = url::Url::parse(url).unwrap();
+            assert_eq!(super::object_store_authority(&parsed), expected, "{url}");
+        }
+    }
+
+    #[test]
+    fn deregister_error_does_not_echo_the_password_or_query() {
+        use datafusion::execution::object_store::ObjectStoreRegistry;
+
+        let url = url::Url::parse("abfss://data:SECRET@acct.dfs.core.windows.net/p?sig=SAS-MARKER")
+            .unwrap();
+        let err = super::CometObjectStoreRegistry::default()
+            .deregister_store(&url)
+            .expect_err("nothing is registered")
+            .to_string();
+        for leaked in ["SECRET", "SAS-MARKER"] {
+            assert!(!err.contains(leaked), "{leaked} echoed: {err}");
+        }
+        assert!(
+            err.contains("abfss://data@acct.dfs.core.windows.net"),
+            "{err}"
+        );
+    }
+
+    #[test]
+    fn prepare_errors_do_not_echo_userinfo_or_query() {
+        let runtime = Arc::new(RuntimeEnv::default());
+        let options = HashMap::from([
+            ("comet.azure.resolution".into(), "resolved".into()),
+            ("comet.azure.auth.type".into(), "SharedKey".into()),
+            ("fs.azure.account.key".into(), "c2VjcmV0".into()),
+        ]);
+        let errors: Vec<(String, &str)> = [
+            // The Azure builder recognises only public cloud and Fabric hosts.
+            (
+                "abfss://data:SECRET@acct.dfs.core.chinacloudapi.cn/x?sig=SAS-MARKER",
+                "URL did not match any known pattern",
+            ),
+            (
+                "foo://alice:SECRET@host/x?sig=SAS-MARKER",
+                "Unable to recognise URL",
+            ),
+            (
+                "s3a://alice:SECRET@bucket:notaport/x?sig=SAS-MARKER",
+                "invalid port number",
+            ),
+        ]
+        .into_iter()
+        .map(|(url, reason)| {
+            let err = prepare_object_store_with_configs(Arc::clone(&runtime), url.into(), &options)
+                .expect_err("the URL is rejected")
+                .to_string();
+            (err, reason)
+        })
+        .collect();
+        for (err, reason) in &errors {
+            assert!(err.contains(reason), "{err}");
+        }
+        assert!(
+            errors[2].0.contains("Error parsing s3a URL"),
+            "{}",
+            errors[2].0
+        );
+        let leaks: Vec<&String> = errors
+            .iter()
+            .map(|(err, _)| err)
+            .filter(|err| {
+                ["alice", "SECRET", "SAS-MARKER"]
+                    .iter()
+                    .any(|s| err.contains(s))
+            })
+            .collect();
+        assert!(leaks.is_empty(), "credentials echoed: {leaks:#?}");
+    }
+
+    /// Credentials come from the configuration, so the key keeps the container but no password.
+    #[test]
+    fn abfs_password_stays_out_of_store_identity() {
+        use datafusion::execution::runtime_env::RuntimeEnvBuilder;
+
+        let host = "comet-password-identity.dfs.core.windows.net";
+        let options = HashMap::from([
+            ("comet.azure.resolution".into(), "resolved".into()),
+            ("comet.azure.auth.type".into(), "SharedKey".into()),
+            ("fs.azure.account.key".into(), "c2VjcmV0".into()),
+        ]);
+        let runtime = RuntimeEnvBuilder::new()
+            .with_object_store_registry(Arc::new(super::CometObjectStoreRegistry::default()))
+            .build_arc()
+            .unwrap();
+        let register = |password: &str| {
+            let (url, _, _) = prepare_object_store_with_configs(
+                Arc::clone(&runtime),
+                format!("abfss://data:{password}@{host}/p/file.parquet?sig=SAS-MARKER"),
+                &options,
+            )
+            .unwrap();
+            (url.clone(), runtime.object_store(&url).unwrap())
+        };
+
+        let (url, store) = register("SECRET");
+        for leaked in ["SECRET", "SAS-MARKER"] {
+            assert!(!url.as_str().contains(leaked), "{leaked} in {url}");
+        }
+        assert!(url.as_str().contains(&format!("://data@{host}")), "{url}");
+        let (other_url, other_store) = register("OTHER");
+        assert_eq!(other_url, url);
+        assert!(Arc::ptr_eq(&other_store, &store));
+
+        let keys: Vec<String> = object_store_cache()
+            .read()
+            .unwrap()
+            .keys()
+            .filter(|(key, _, _)| key.contains(host))
+            .map(|(key, _, _)| key.clone())
+            .collect();
+        object_store_cache()
+            .write()
+            .unwrap()
+            .retain(|(key, _, _), _| !key.contains(host));
+        assert_eq!(keys, vec![format!("abfss://data@{host}")]);
+    }
+
     /// Parses the url, registers the object store, and returns a tuple of the object store url and object store path
     #[cfg(not(feature = "hdfs-opendal"))]
     pub(crate) fn prepare_object_store(
diff --git a/spark/src/main/java/org/apache/comet/parquet/CometFileKeyUnwrapper.java b/spark/src/main/java/org/apache/comet/parquet/CometFileKeyUnwrapper.java
index 79e32400bad..433b8b4b0ea 100644
--- a/spark/src/main/java/org/apache/comet/parquet/CometFileKeyUnwrapper.java
+++ b/spark/src/main/java/org/apache/comet/parquet/CometFileKeyUnwrapper.java
@@ -154,6 +154,43 @@ static String normalizeS3Scheme(final String filePath, final Set<String> s3Compl
     return "s3a://" + StringUtils.stripStart(filePath.substring(schemeEnd + 1), "/");
   }
 
+  /**
+   * The ABFS path native looks keys up by: the scheme lowercased and the password dropped from the
+   * authority, where the user ends at the first {@code :} and the host follows the last {@code @},
+   * as native {@code authority_without_password} reads it. Non-ABFS paths are returned unchanged.
+   */
+  static String withoutAbfsPassword(final String filePath) {
+    final int schemeEnd = filePath.indexOf(':');
+    if (schemeEnd <= 0) {
+      return filePath;
+    }
+    final String scheme = filePath.substring(0, schemeEnd).toLowerCase(Locale.ROOT);
+    if (!scheme.equals("abfs") && !scheme.equals("abfss")) {
+      return filePath;
+    }
+    if (!filePath.startsWith("//", schemeEnd + 1)) {
+      return scheme + filePath.substring(schemeEnd);
+    }
+    final int start = schemeEnd + 3;
+    int end = start;
+    while (end < filePath.length() && "/?#".indexOf(filePath.charAt(end)) < 0) {
+      end++;
+    }
+    final String authority = filePath.substring(start, end);
+    final int at = authority.lastIndexOf('@');
+    final String userInfo = at < 0 ? "" : authority.substring(0, at);
+    final int colon = userInfo.indexOf(':');
+    final String user = colon < 0 ? userInfo : userInfo.substring(0, colon);
+    final String host = authority.substring(at + 1);
+    final String key = user.isEmpty() ? host : user + "@" + host;
+    return scheme + "://" + key + filePath.substring(end);
+  }
+
+  /** The retriever cache key: S3 aliases folded to s3a and the ABFS password dropped. */
+  private String cacheKey(final String filePath) {
+    return withoutAbfsPassword(normalizeS3Scheme(filePath));
+  }
+
   /**
    * Creates and stores a DecryptionKeyRetriever instance for the given file path.
    *
@@ -175,13 +212,13 @@ public void storeDecryptionKeyRetriever(final String filePath, final Configurati
       // Check the assumption that all files have the same hadoopConf and thus same Factory
       assert (conf == hadoopConf);
     }
-    final String normalizedPath = normalizeS3Scheme(filePath);
+    final String key = cacheKey(filePath);
     Path path = new Path(filePath);
     FileDecryptionProperties decryptionProperties =
         factory.getFileDecryptionProperties(hadoopConf, path);
 
     DecryptionKeyRetriever keyRetriever = decryptionProperties.getKeyRetriever();
-    retrieverCache.put(normalizedPath, keyRetriever);
+    retrieverCache.put(key, keyRetriever);
   }
 
   /**
@@ -195,11 +232,10 @@ public void storeDecryptionKeyRetriever(final String filePath, final Configurati
    */
   public byte[] getKey(final String filePath, final byte[] keyMetadata)
       throws ParquetCryptoRuntimeException {
-    final String normalizedPath = normalizeS3Scheme(filePath);
-    DecryptionKeyRetriever keyRetriever = retrieverCache.get(normalizedPath);
+    DecryptionKeyRetriever keyRetriever = retrieverCache.get(cacheKey(filePath));
     if (keyRetriever == null) {
       throw new ParquetCryptoRuntimeException(
-          "Failed to find DecryptionKeyRetriever for path: " + filePath);
+          "Failed to find DecryptionKeyRetriever for path: " + cacheKey(filePath));
     }
     return keyRetriever.getKey(keyMetadata);
   }
diff --git a/spark/src/main/scala/org/apache/comet/objectstore/AbfsAuthResolver.scala b/spark/src/main/scala/org/apache/comet/objectstore/AbfsAuthResolver.scala
index 3f3812bb6ea..03cd9dfd8c3 100644
--- a/spark/src/main/scala/org/apache/comet/objectstore/AbfsAuthResolver.scala
+++ b/spark/src/main/scala/org/apache/comet/objectstore/AbfsAuthResolver.scala
@@ -20,6 +20,7 @@
 package org.apache.comet.objectstore
 
 import java.net.URI
+import java.util.Arrays
 
 import scala.collection.mutable.ArrayBuffer
 import scala.util.{Failure, Success, Try}
@@ -146,7 +147,8 @@ private[comet] object AbfsAuthResolver extends Logging {
   }
 
   private val REDACTED = "<redacted>"
-  // Shorter values (`true`, a port) would mangle messages without protecting anything.
+  // Configured values shorter than this (`true`, a port) are not recorded as secrets: hiding them
+  // would mangle messages without protecting anything. URI secrets are hidden at any length.
   private val MIN_REDACTED_LENGTH = 8
   // `SimpleKeyProvider` (3.4.1+) puts the raw account key in its message as `key ="..."`.
   private val QUOTED_KEY_PATTERN = "key =\"[^\"]*\"".r
@@ -211,7 +213,7 @@ private[comet] object AbfsAuthResolver extends Logging {
       hadoopConf: Configuration,
       uri: URI,
       handles: Try[Handles]): Outcome = {
-    val secrets = ArrayBuffer[String]()
+    val secrets = ArrayBuffer[String](uriSecrets(uri): _*)
     // Set once the AbfsConfiguration exists, so a failure can read the credentials it skipped.
     var openSession: Option[Session] = None
     try {
@@ -260,17 +262,17 @@ private[comet] object AbfsAuthResolver extends Logging {
   private[objectstore] def authorityParts(uri: URI): Either[Failed, (String, String)] = {
     Option(uri.getRawAuthority) match {
       case None =>
-        Left(Failed(Exceptions.INVALID_URI_AUTHORITY, s"$uri has invalid authority."))
+        Left(Failed(Exceptions.INVALID_URI_AUTHORITY, s"${display(uri)} has invalid authority."))
       case Some(authority) if !authority.contains("@") =>
-        Left(Failed(Exceptions.INVALID_URI_AUTHORITY, s"$uri has invalid authority."))
+        Left(Failed(Exceptions.INVALID_URI_AUTHORITY, s"${display(uri)} has invalid authority."))
       case Some(authority) =>
         val parts = authority.split("@", 2)
         if (parts.length < 2 || parts(0).isEmpty) {
           Left(
             Failed(
               Exceptions.INVALID_URI,
-              s"Invalid URI '$uri' has a malformed authority, expected container name. " +
-                "Authority takes the form abfs://[<container name>@]<account name>"))
+              s"Invalid URI '${display(uri)}' has a malformed authority, expected container " +
+                "name. Authority takes the form abfs://[<container name>@]<account name>"))
         } else {
           Right((parts(0), parts(1)))
         }
@@ -354,7 +356,9 @@ private[comet] object AbfsAuthResolver extends Logging {
   private def sharedKey(session: Session, uri: URI, account: String): Outcome = {
     // Store.initializeClient checks the account name before fetching the key.
     if (account.indexOf('.') <= 0) {
-      Failed(Exceptions.INVALID_URI, s"Invalid URI $uri - account name is not fully qualified.")
+      Failed(
+        Exceptions.INVALID_URI,
+        s"Invalid URI ${display(uri)} - account name is not fully qualified.")
     } else {
       val key = session.storageAccountKey()
       if (key.isEmpty) {
@@ -481,11 +485,8 @@ private[comet] object AbfsAuthResolver extends Logging {
   def nativeFallbackReason(uri: URI, outcome: Outcome): Option[String] = {
     // The container and host only: a password in the user info stays out of the reason.
     val authority = Option(uri.getRawAuthority).getOrElse("")
-    val at = authority.lastIndexOf('@')
-    val safeAuthority =
-      if (at < 0) authority
-      else authority.substring(0, at).takeWhile(_ != ':') + authority.substring(at)
-    val location = s"${uri.getScheme}://$safeAuthority"
+    val safeAuthority = NativeConfig.authorityWithoutPassword(uri)
+    val location = display(uri)
     outcome match {
       case Declined(authType, providerClass, assertionProviderClass) =>
         val classes = providerClass.map(c => s" with provider $c").getOrElse("") +
@@ -536,16 +537,46 @@ private[comet] object AbfsAuthResolver extends Logging {
     if (authType == AuthTypes.SAS) Markers.SAS_PROVIDER_CLASS else Markers.OAUTH_PROVIDER_CLASS
 
   /**
-   * Hides every `secret` of at least [[MIN_REDACTED_LENGTH]] characters in `message`, and the raw
-   * key a `SimpleKeyProvider` message quotes.
+   * Hides every non-empty `secret` in `message`, and the raw key a `SimpleKeyProvider` message
+   * quotes. Secrets are matched in the same text, so one that overlaps or contains another is
+   * hidden whole, and each run of hidden characters becomes one marker.
    */
   private[objectstore] def redact(message: String, secrets: Iterable[String]): String = {
-    val withoutQuotedKey = QUOTED_KEY_PATTERN.replaceAllIn(message, "key =\"" + REDACTED + "\"")
-    secrets
-      .filter(_.length >= MIN_REDACTED_LENGTH)
-      .foldLeft(withoutQuotedKey)((text, secret) => text.replace(secret, REDACTED))
+    val text = QUOTED_KEY_PATTERN.replaceAllIn(message, "key =\"" + REDACTED + "\"")
+    val hidden = new Array[Boolean](text.length)
+    secrets.filter(_.nonEmpty).foreach { secret =>
+      var i = text.indexOf(secret)
+      while (i >= 0) {
+        Arrays.fill(hidden, i, i + secret.length, true)
+        i = text.indexOf(secret, i + 1)
+      }
+    }
+    val out = new StringBuilder(text.length)
+    for (i <- 0 until text.length) {
+      if (!hidden(i)) out += text(i)
+      else if (i == 0 || !hidden(i - 1)) out ++= REDACTED
+    }
+    out.toString
   }
 
+  /**
+   * The parts of `uri` that can carry a credential: the password and the raw query. The password
+   * is read from the raw authority, which has one even when `getRawUserInfo` is null (an `_` in
+   * the host makes the authority registry-based). It is the text after the first `:` of the user
+   * part, which ends at the last `@` as in [[NativeConfig.authorityWithoutPassword]] and at the
+   * first `@` as in [[authorityParts]].
+   */
+  private[objectstore] def uriSecrets(uri: URI): Seq[String] = {
+    val authority = Option(uri.getRawAuthority).getOrElse("")
+    val passwords = Seq(authority.lastIndexOf('@'), authority.indexOf('@'))
+      .filter(_ >= 0)
+      .map(at => authority.take(at).dropWhile(_ != ':').drop(1))
+    (passwords ++ Option(uri.getRawQuery)).filter(_.nonEmpty).distinct
+  }
+
+  // Messages reach native errors and logs, so they omit the password, path and query.
+  private def display(uri: URI): String = NativeConfig.displayAuthority(uri)
+
   private def failed(uri: URI, e: Throwable, secrets: Iterable[String]): Failed = {
     val chain = causeChain(e)
     val thrown = chain.head
@@ -554,8 +585,8 @@ private[comet] object AbfsAuthResolver extends Logging {
       if (verbatim) s"$name: ${redact(Option(t.getMessage).getOrElse(""), secrets)}" else name
     }
     logWarning(
-      s"Hadoop ABFS authentication for ${uri.getScheme}://${uri.getRawAuthority} failed on the " +
-        s"driver: ${chain.map(describe(_, verbatim = true)).mkString(", caused by ")}")
+      s"Hadoop ABFS authentication for ${display(uri)} failed on the driver: " +
+        s"${chain.map(describe(_, verbatim = true)).mkString(", caused by ")}")
 
     val forwarded = chain.map(t => describe(t, isVerbatim(t)))
     val hidden = chain.exists(t => !isVerbatim(t))
diff --git a/spark/src/main/scala/org/apache/comet/objectstore/NativeConfig.scala b/spark/src/main/scala/org/apache/comet/objectstore/NativeConfig.scala
index b6f63ac77ef..8d7b08e38f8 100644
--- a/spark/src/main/scala/org/apache/comet/objectstore/NativeConfig.scala
+++ b/spark/src/main/scala/org/apache/comet/objectstore/NativeConfig.scala
@@ -132,6 +132,26 @@ object NativeConfig {
     rawAuthority.substring(rawAuthority.lastIndexOf('@') + 1)
   }
 
+  /**
+   * As in native, the user ends at the first `:` and the host follows the last `@`. An authority
+   * with no password is returned as is.
+   */
+  private[objectstore] def authorityWithoutPassword(uri: URI): String = {
+    val rawAuthority = Option(uri.getRawAuthority).getOrElse("")
+    val at = rawAuthority.lastIndexOf('@')
+    if (at < 0 || (at > 0 && rawAuthority.lastIndexOf(':', at - 1) < 0)) {
+      rawAuthority
+    } else {
+      val user = rawAuthority.take(at).takeWhile(_ != ':')
+      val host = rawAuthority.substring(at + 1)
+      if (user.isEmpty) host else s"$user@$host"
+    }
+  }
+
+  /** `scheme://[user@]host[:port]` of `uri`, for messages: no password, path or query. */
+  private[comet] def displayAuthority(uri: URI): String =
+    s"${lowerScheme(uri).getOrElse("file")}://${authorityWithoutPassword(uri)}"
+
   /**
    * The schemes native reads through libhdfs: the `fs.comet.libhdfs.schemes` list, or only `hdfs`
    * when it is unset or blank, as native `is_hdfs_scheme` decides.
@@ -198,7 +218,7 @@ object NativeConfig {
     val scheme = lowerScheme(uri).getOrElse("file")
     val isLibhdfs = libhdfsSchemes.contains(scheme)
     val authority =
-      if (scheme == "abfs" || scheme == "abfss") Option(uri.getRawAuthority).getOrElse("")
+      if (scheme == "abfs" || scheme == "abfss") authorityWithoutPassword(uri)
       else hostAndPort(uri)
     val isS3Alias = !isLibhdfs &&
       (scheme == "s3a" ||
@@ -421,9 +441,12 @@ object NativeConfig {
             s"(${first.getScheme} and ${uri.getScheme})")
       }
       if (firstIsAbfs && uri.getRawAuthority != first.getRawAuthority) {
+        // Hadoop reads the whole user info, password too, as the container.
+        val (a, b) = (displayAuthority(first), displayAuthority(uri))
+        val stores = if (a == b) s"$a, with different user info" else s"$a and $b"
         return Left(
           "Native scan does not support ABFS files from more than one container or account " +
-            s"(${first.getRawAuthority} and ${uri.getRawAuthority})")
+            s"($stores)")
       }
     }
     if (firstIsAbfs) {
diff --git a/spark/src/test/java/org/apache/comet/parquet/TestCometFileKeyUnwrapper.java b/spark/src/test/java/org/apache/comet/parquet/TestCometFileKeyUnwrapper.java
index 621e68ff98d..c1fcaae1a3d 100644
--- a/spark/src/test/java/org/apache/comet/parquet/TestCometFileKeyUnwrapper.java
+++ b/spark/src/test/java/org/apache/comet/parquet/TestCometFileKeyUnwrapper.java
@@ -19,14 +19,23 @@
 
 package org.apache.comet.parquet;
 
+import java.nio.charset.StandardCharsets;
 import java.util.Collections;
 import java.util.HashSet;
 import java.util.Set;
 
 import org.junit.Test;
 
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.parquet.crypto.DecryptionPropertiesFactory;
+import org.apache.parquet.crypto.FileDecryptionProperties;
+import org.apache.parquet.crypto.ParquetCryptoRuntimeException;
+
 import static java.util.Arrays.asList;
+import static org.junit.Assert.assertArrayEquals;
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.fail;
 
 /**
  * Regression guard for {@link CometFileKeyUnwrapper#normalizeS3Scheme(String, Set)}. The put side
@@ -95,4 +104,75 @@ public void aliasSingleAndTripleSlashPromoteBucketToAuthority() {
     assertEquals(canonical, norm("blob:/bucket/foo/part-0.parquet", BLOB));
     assertEquals(canonical, norm("blob:///bucket/foo/part-0.parquet", BLOB));
   }
+
+  @Test
+  public void abfsKeyDropsThePasswordAsNativeDoes() {
+    // Native looks keys up by the URL it read the file through, whose ABFS authority has no
+    // password and whose scheme the url crate lowercased.
+    String host = "acct.dfs.core.windows.net";
+    String[][] cases = {
+      {"abfss://data:secret@" + host + "/t/r.parquet", "abfss://data@" + host + "/t/r.parquet"},
+      {
+        "abfs://data:secret@" + host + ":8443/t/r.parquet",
+        "abfs://data@" + host + ":8443/t/r.parquet"
+      },
+      {"abfss://:secret@" + host + "/t/r.parquet", "abfss://" + host + "/t/r.parquet"},
+      {"abfss://data@" + host + "/t/r.parquet", "abfss://data@" + host + "/t/r.parquet"},
+      {"ABFSS://data:secret@" + host + "/t/r.parquet", "abfss://data@" + host + "/t/r.parquet"},
+      {"abfss://data@" + host + "/a@b:c/r.parquet", "abfss://data@" + host + "/a@b:c/r.parquet"},
+      {"abfss:/t/r.parquet", "abfss:/t/r.parquet"},
+      {"abfss://" + host + "/p", "abfss://" + host + "/p"},
+      {"abfss://data:se:cret@" + host + "/p", "abfss://data@" + host + "/p"},
+      // The url crate drops the `:` of an empty password.
+      {"abfss://data:@" + host + "/p", "abfss://data@" + host + "/p"},
+      {"abfss://data:secret@" + host + "?q", "abfss://data@" + host + "?q"},
+      {"abfss://data:secret@" + host + "#f", "abfss://data@" + host + "#f"},
+      // Only ABFS is rewritten here; S3 user info is left as it is.
+      {"s3a://user:secret@bucket/t/r.parquet", "s3a://user:secret@bucket/t/r.parquet"},
+      {"hdfs://nn:8020/t/r.parquet", "hdfs://nn:8020/t/r.parquet"},
+    };
+    for (String[] c : cases) {
+      assertEquals(c[0], c[1], CometFileKeyUnwrapper.withoutAbfsPassword(c[0]));
+    }
+  }
+
+  @Test
+  public void abfsKeyStoredWithPasswordIsFoundWithout() {
+    Configuration conf = new Configuration(false);
+    conf.set(
+        DecryptionPropertiesFactory.CRYPTO_FACTORY_CLASS_PROPERTY_NAME,
+        FixedKeyFactory.class.getName());
+    CometFileKeyUnwrapper unwrapper = new CometFileKeyUnwrapper();
+    String stored = "abfss://data:fixture@acct.dfs.core.windows.net/t/r.parquet";
+    unwrapper.storeDecryptionKeyRetriever(stored, conf);
+    byte[] metadata = new byte[0];
+    assertArrayEquals(
+        FixedKeyFactory.KEY,
+        unwrapper.getKey("abfss://data@acct.dfs.core.windows.net/t/r.parquet", metadata));
+    assertArrayEquals(FixedKeyFactory.KEY, unwrapper.getKey(stored, metadata));
+  }
+
+  @Test
+  public void missingAbfsKeyErrorOmitsThePassword() {
+    CometFileKeyUnwrapper unwrapper = new CometFileKeyUnwrapper();
+    try {
+      unwrapper.getKey("abfss://data:fixture@acct.dfs.core.windows.net/t/r.parquet", new byte[0]);
+      fail("expected a missing retriever");
+    } catch (ParquetCryptoRuntimeException e) {
+      assertEquals(
+          "Failed to find DecryptionKeyRetriever for path: "
+              + "abfss://data@acct.dfs.core.windows.net/t/r.parquet",
+          e.getMessage());
+    }
+  }
+
+  /** Hands every file a retriever that returns {@link #KEY}. */
+  public static class FixedKeyFactory implements DecryptionPropertiesFactory {
+    static final byte[] KEY = "0123456789abcdef".getBytes(StandardCharsets.US_ASCII);
+
+    @Override
+    public FileDecryptionProperties getFileDecryptionProperties(Configuration conf, Path path) {
+      return FileDecryptionProperties.builder().withKeyRetriever(metadata -> KEY).build();
+    }
+  }
 }
diff --git a/spark/src/test/scala/org/apache/comet/objectstore/AbfsAuthResolverSuite.scala b/spark/src/test/scala/org/apache/comet/objectstore/AbfsAuthResolverSuite.scala
index 4d69c2e01a3..60ca6aece03 100644
--- a/spark/src/test/scala/org/apache/comet/objectstore/AbfsAuthResolverSuite.scala
+++ b/spark/src/test/scala/org/apache/comet/objectstore/AbfsAuthResolverSuite.scala
@@ -20,6 +20,10 @@
 package org.apache.comet.objectstore
 
 import java.net.URI
+import java.nio.charset.StandardCharsets.UTF_8
+import java.nio.file.{Files, Path, Paths}
+import java.security.KeyStore
+import javax.crypto.spec.SecretKeySpec
 
 import scala.collection.mutable.ArrayBuffer
 import scala.util.{Failure, Success, Try}
@@ -32,6 +36,11 @@ import org.apache.hadoop.fs.azurebfs.AbfsConfiguration
 import org.apache.hadoop.fs.azurebfs.extensions.SASTokenProvider
 import org.apache.hadoop.fs.azurebfs.oauth2.{AccessTokenProvider, AzureADToken}
 import org.apache.hadoop.fs.azurebfs.services.KeyProvider
+import org.apache.hadoop.security.alias.CredentialProviderFactory
+import org.apache.logging.log4j.{Level, LogManager}
+import org.apache.logging.log4j.core.{LogEvent, Logger => CoreLogger}
+import org.apache.logging.log4j.core.appender.AbstractAppender
+import org.apache.logging.log4j.core.config.Property
 
 import org.apache.comet.objectstore.AbfsAuthResolver._
 import org.apache.comet.objectstore.AbfsReflection.Handles
@@ -77,6 +86,49 @@ class AbfsAuthResolverSuite extends AnyFunSuite with Matchers {
   private def cyclic(key: String, other: String): Seq[(String, String)] =
     Seq(key -> ("${" + other + "}"), other -> ("${" + key + "}"))
 
+  /** A JCEKS keystore Hadoop can open whose one entry, `alias`, it cannot recover. */
+  private def unrecoverableKeystore(alias: String): Path = {
+    val storePassword = sys.env.getOrElse("HADOOP_CREDSTORE_PASSWORD", "none").toCharArray
+    val keyStore = KeyStore.getInstance("JCEKS")
+    keyStore.load(null, storePassword)
+    keyStore.setEntry(
+      alias,
+      new KeyStore.SecretKeyEntry(new SecretKeySpec("value".getBytes(UTF_8), "AES")),
+      new KeyStore.PasswordProtection("not-the-store-password".toCharArray))
+    val tmp = Files.createDirectories(Paths.get(System.getProperty("java.io.tmpdir")))
+    val path = Files.createTempFile(tmp, "abfs-unrecoverable", ".jceks")
+    path.toFile.deleteOnExit()
+    val out = Files.newOutputStream(path)
+    try keyStore.store(out, storePassword)
+    finally out.close()
+    path
+  }
+
+  /** The resolver's WARN messages logged while `f` runs, one per line. */
+  private def capturedWarnings(f: => Unit): String = {
+    val logger = LogManager
+      .getLogger(AbfsAuthResolver.getClass.getName.stripSuffix("$"))
+      .asInstanceOf[CoreLogger]
+    val messages = ArrayBuffer[String]()
+    val appender =
+      new AbstractAppender("AbfsAuthResolverSuite", null, null, false, Property.EMPTY_ARRAY) {
+        override def append(event: LogEvent): Unit = messages.synchronized {
+          if (event.getLevel == Level.WARN) messages += event.getMessage.getFormattedMessage
+        }
+      }
+    appender.start()
+    val originalLevel = logger.getLevel
+    logger.addAppender(appender)
+    logger.setLevel(Level.WARN)
+    try f
+    finally {
+      logger.removeAppender(appender)
+      logger.setLevel(originalLevel)
+      appender.stop()
+    }
+    messages.synchronized(messages.mkString("\n"))
+  }
+
   private def assertNoAzureKeys(options: Map[String, String]): Unit = {
     val leaked = options.keys.filter(_.startsWith("fs."))
     assert(leaked.isEmpty, s"fs.* keys forwarded: $leaked")
@@ -96,20 +148,23 @@ class AbfsAuthResolverSuite extends AnyFunSuite with Matchers {
   }
 
   test("authorityParts - no '@', empty container and null authority fail as Hadoop does") {
-    val noAt = "abfss://myacct.dfs.core.windows.net/p"
-    authorityParts(new URI(noAt)) shouldBe
-      Left(Failed(invalidUriAuthorityException, s"$noAt has invalid authority."))
+    // Messages name the URI by scheme and authority only, so a SAS query never reaches them.
+    val noAt = "abfss://myacct.dfs.core.windows.net/p?sig=SAS-MARKER"
+    authorityParts(new URI(noAt)) shouldBe Left(
+      Failed(
+        invalidUriAuthorityException,
+        "abfss://myacct.dfs.core.windows.net has invalid authority."))
 
-    val emptyContainer = "abfss://@myacct.dfs.core.windows.net/p"
+    val emptyContainer = "abfss://@myacct.dfs.core.windows.net/p?sig=SAS-MARKER"
     authorityParts(new URI(emptyContainer)) shouldBe Left(
       Failed(
         invalidUriException,
-        s"Invalid URI '$emptyContainer' has a malformed authority, expected container name. " +
-          "Authority takes the form abfs://[<container name>@]<account name>"))
+        "Invalid URI 'abfss://myacct.dfs.core.windows.net' has a malformed authority, expected " +
+          "container name. Authority takes the form abfs://[<container name>@]<account name>"))
 
-    val nullAuthority = "abfss:///p"
+    val nullAuthority = "abfss:///p?sig=SAS-MARKER"
     authorityParts(new URI(nullAuthority)) shouldBe
-      Left(Failed(invalidUriAuthorityException, s"$nullAuthority has invalid authority."))
+      Left(Failed(invalidUriAuthorityException, "abfss:// has invalid authority."))
   }
 
   // ---- reflection ----
@@ -295,6 +350,18 @@ class AbfsAuthResolverSuite extends AnyFunSuite with Matchers {
     failed.message should not include secret
   }
 
+  test("Failed - a credential that contains the URI's password is redacted whole") {
+    val secret = "oauth-client-secret-4f9a2c"
+    val secretUri = new URI(s"abfss://data:client-secret@$account/path/file.parquet")
+    val outcome =
+      resolve(conf(acct(Keys.AUTH_TYPE) -> secret, acct(Keys.CLIENT_SECRET) -> secret), secretUri)
+    outcome shouldBe a[Failed]
+    val failed = outcome.asInstanceOf[Failed]
+    failed.message should include("No enum constant")
+    Seq("oauth-", "-4f9a2c", "client-secret").foreach(part =>
+      failed.message should not include part)
+  }
+
   test("Failed - an unknown auth type value is Hadoop's enum error") {
     val outcome = resolveWith(acct(Keys.AUTH_TYPE) -> "oauth")
     outcome shouldBe a[Failed]
@@ -335,7 +402,86 @@ class AbfsAuthResolverSuite extends AnyFunSuite with Matchers {
     val shortUri = new URI("abfss://data@myacct/path")
     val outcome = resolve(conf("fs.azure.account.key.myacct" -> "c2VjcmV0"), shortUri)
     outcome shouldBe
-      Failed(invalidUriException, s"Invalid URI $shortUri - account name is not fully qualified.")
+      Failed(
+        invalidUriException,
+        "Invalid URI abfss://data@myacct - account name is not fully qualified.")
+  }
+
+  test("Failed - the forwarded message drops the URI's password, path and query") {
+    val shortUri = new URI("abfss://data:SECRET@myacct/path?sig=SAS-MARKER")
+    val outcome = resolve(conf("fs.azure.account.key.myacct" -> "c2VjcmV0"), shortUri)
+    outcome shouldBe a[Failed]
+    val message = toOptions(outcome)(Markers.ERROR_MESSAGE)
+    Seq("SECRET", "SAS-MARKER", "/path").foreach(leaked => message should not include leaked)
+    message should include("abfss://data@myacct")
+  }
+
+  test("Failed - a forwarded message is scrubbed of the URI's password and query") {
+    // Forwarded verbatim, so it stands in for a Hadoop message that quotes the URI.
+    val secretUri = new URI(s"abfss://data:the-uri-password@$account/p?sig=the-sas-signature")
+    val thrown = new ClassNotFoundException(s"no class for $secretUri")
+    val message = toOptions(resolve(conf(), secretUri, Failure(thrown)))(Markers.ERROR_MESSAGE)
+    message should include(s"no class for abfss://data:<redacted>@$account/p?<redacted>")
+    Seq("the-uri-password", "the-sas-signature").foreach(s => message should not include s)
+  }
+
+  test("Failed - a URI password and query shorter than eight characters are scrubbed too") {
+    val shortUri = new URI(s"abfss://data:pw12345@$account/p?sig=abc")
+    val thrown = new ClassNotFoundException(s"no class for $shortUri")
+    val message = toOptions(resolve(conf(), shortUri, Failure(thrown)))(Markers.ERROR_MESSAGE)
+    message should include(s"no class for abfss://data:<redacted>@$account/p?<redacted>")
+    Seq("pw12345", "sig=abc").foreach(s => message should not include s)
+  }
+
+  /**
+   * The driver warnings from an OAuth client-credentials resolution of a URI on `host` with
+   * `password` in its user info, where Hadoop cannot read the container-scoped client secret from
+   * the keystore. The outcome must be a [[Failed]] that does not contain the password.
+   */
+  private def keystoreAliasWarnings(host: String, password: String): String = {
+    // On 3.4.2+ the container-scoped alias embeds the raw container, user info included, and
+    // Hadoop names the alias when it cannot recover the entry.
+    val secretUri = new URI(s"abfss://$container:$password@$host/path/file.parquet")
+    val keystore = unrecoverableKeystore(s"${Keys.CLIENT_SECRET}.$container:$password.$host")
+    val providerPath = s"jceks://file${keystore.toUri.getRawPath}"
+    def key(name: String): String = s"$name.$host"
+    capturedWarnings {
+      val outcome = resolve(
+        conf(
+          CredentialProviderFactory.CREDENTIAL_PROVIDER_PATH -> providerPath,
+          key(Keys.AUTH_TYPE) -> "OAuth",
+          key(Keys.OAUTH_PROVIDER_TYPE) -> Providers.CLIENT_CREDS,
+          key(Keys.CLIENT_ENDPOINT) -> "https://login.microsoftonline.com/tenant/oauth2/token",
+          key(Keys.CLIENT_ID) -> "client-123"),
+        secretUri)
+      outcome shouldBe a[Failed]
+      toOptions(outcome).values.foreach(v => v should not include password)
+    }
+  }
+
+  test("Failed - a short URI password in a keystore alias stays out of the driver warning") {
+    assume(handles.hasContainerScopedConfiguration, "no container-scoped keys before 3.4.2")
+    val warnings = keystoreAliasWarnings(account, "pw12345")
+    warnings should include(s"${Keys.CLIENT_SECRET}.$container:<redacted>.$account")
+    warnings should not include "pw12345"
+  }
+
+  test("Failed - a URI password is hidden when the host makes the authority registry-based") {
+    // An underscore in the host leaves java.net.URI without user info, but Hadoop still takes
+    // the container, password included, from the raw authority.
+    assume(handles.hasContainerScopedConfiguration, "no container-scoped keys before 3.4.2")
+    val host = "my_account.dfs.core.windows.net"
+    assert(new URI(s"abfss://$container:pw12345@$host/p").getRawUserInfo == null)
+    val warnings = keystoreAliasWarnings(host, "pw12345")
+    warnings should include(s"${Keys.CLIENT_SECRET}.$container:<redacted>.$host")
+    warnings should not include "pw12345"
+  }
+
+  test("Failed - a configured value shorter than eight characters is left readable") {
+    // `true` read as a client secret is too short to hide, so Hadoop's enum error still names it.
+    val outcome = resolveWith(acct(Keys.AUTH_TYPE) -> "true", acct(Keys.CLIENT_SECRET) -> "true")
+    outcome shouldBe a[Failed]
+    outcome.asInstanceOf[Failed].message should include("AuthType.true")
   }
 
   test("Failed - a LinkageError from the reflection layer is an error marker, not a crash") {
@@ -355,10 +501,32 @@ class AbfsAuthResolverSuite extends AnyFunSuite with Matchers {
     failed.message should include(AbfsReflection.ClassNames.ABFS_CONFIGURATION)
   }
 
-  test("redact - hides quoted keys and long secrets, leaves short values alone") {
-    val message = "Failure for acct key =\"c2VjcmV0c2VjcmV0\": token=the-secret-value port=8080"
-    redact(message, Seq("the-secret-value", "8080")) shouldBe
-      "Failure for acct key =\"<redacted>\": token=<redacted> port=8080"
+  test("uriSecrets - the password comes from the raw authority, at either `@` split") {
+    uriSecrets(new URI(s"abfss://data:pw@$account/p?sig=x")) shouldBe Seq("pw", "sig=x")
+    uriSecrets(new URI("abfss://data:pw@my_acct/p")) shouldBe Seq("pw")
+    uriSecrets(new URI("abfss://data:a@b@acct/p")) shouldBe Seq("a@b", "a")
+    uriSecrets(new URI(s"abfss://data@$account/p")) shouldBe empty
+    uriSecrets(new URI("abfss://data:@my_acct/p?")) shouldBe empty
+  }
+
+  test("redact - hides quoted keys and every non-empty secret, however short") {
+    val message = "Failure for acct key =\"c2VjcmV0c2VjcmV0\": token=the-secret-value pw=1234567"
+    redact(message, Seq("the-secret-value", "1234567", "")) shouldBe
+      "Failure for acct key =\"<redacted>\": token=<redacted> pw=<redacted>"
+  }
+
+  test("redact - overlapping, contained and adjacent secrets become one marker") {
+    // Each secret is matched in the original text, so the order they arrive in does not matter.
+    val cases = Seq(
+      ("x oauth-client-secret-4f9a2c y", Seq("client-secret", "oauth-client-secret-4f9a2c")),
+      ("x first-secret-tail y", Seq("first-secret", "secret-tail")),
+      ("x first-secretsecond-secret y", Seq("first-secret", "second-secret")),
+      ("x abababababab y", Seq("abababab")))
+    for ((message, secrets) <- cases; ordered <- Seq(secrets, secrets.reverse)) {
+      withClue(s"$message with $ordered: ") {
+        redact(message, ordered) shouldBe "x <redacted> y"
+      }
+    }
   }
 
   test("redact - a padded secret is hidden in its trimmed form too") {
diff --git a/spark/src/test/scala/org/apache/comet/objectstore/NativeConfigSuite.scala b/spark/src/test/scala/org/apache/comet/objectstore/NativeConfigSuite.scala
index 1246fbac154..34d885aa319 100644
--- a/spark/src/test/scala/org/apache/comet/objectstore/NativeConfigSuite.scala
+++ b/spark/src/test/scala/org/apache/comet/objectstore/NativeConfigSuite.scala
@@ -554,6 +554,20 @@ class NativeConfigSuite extends AnyFunSuite with Matchers {
       // ABFS keeps the container from the user info; WASB does not.
       Case(s"abfss://container@$account/k.parquet", "", "", s"abfss://container@$account", false),
       Case(s"abfs://container@$account/k.parquet", "", "", s"abfs://container@$account", false),
+      // The password never joins the key; credentials come from the configuration.
+      Case(
+        s"abfss://container:secret@$account/k.parquet",
+        "",
+        "",
+        s"abfss://container@$account",
+        hdfs = false),
+      Case(s"abfss://:secret@$account/k.parquet", "", "", s"abfss://$account", hdfs = false),
+      Case(
+        s"abfss://container:secret@$account:8443/k.parquet",
+        "",
+        "",
+        s"abfss://container@$account:8443",
+        hdfs = false),
       Case(
         "wasbs://container@account.blob.core.windows.net/k.parquet",
         "",
@@ -716,6 +730,46 @@ class NativeConfigSuite extends AnyFunSuite with Matchers {
       Seq(abfsFile("data", "accta", "a.parquet"), abfsFile("data", "acctb", "b.parquet")))
   }
 
+  test("extractScanObjectStoreOptions - a decline names ABFS stores without password or query") {
+    val hadoopConf = new Configuration()
+    hadoopConf.set("fs.azure.account.key.accta.dfs.core.windows.net", "a2V5LWE=")
+    val uris = Seq("SECRET", "OTHER").map { password =>
+      new URI(s"abfss://data:$password@accta.dfs.core.windows.net/path/a.parquet?sig=SAS-MARKER")
+    }
+
+    def declineReason(uris: Seq[URI]): String =
+      NativeConfig.extractScanObjectStoreOptions(hadoopConf, uris) match {
+        case Left(reason) =>
+          Seq("SECRET", "OTHER", "SAS-MARKER", "/path").foreach { leaked =>
+            reason should not include leaked
+          }
+          reason
+        case Right(options) => fail(s"expected a decline, got $options")
+      }
+
+    declineReason(uris) should include(
+      "(abfss://data@accta.dfs.core.windows.net, with different user info)")
+    val otherContainer =
+      new URI("abfss://logs:OTHER@accta.dfs.core.windows.net/path/b.parquet?sig=SAS-MARKER")
+    declineReason(Seq(uris.head, otherContainer)) should include(
+      "(abfss://data@accta.dfs.core.windows.net and abfss://logs@accta.dfs.core.windows.net)")
+  }
+
+  test("authorityWithoutPassword - an authority without a password is returned as is") {
+    for (authority <- Seq("container@acct.dfs.core.windows.net", "container@acct:8443")) {
+      val uri = new URI(s"abfss://$authority/k.parquet")
+      NativeConfig.authorityWithoutPassword(uri) should be theSameInstanceAs uri.getRawAuthority
+    }
+    NativeConfig.authorityWithoutPassword(new URI("/tmp/t/k.parquet")) shouldBe ""
+  }
+
+  test("displayAuthority - lowercase scheme and no password, path or query") {
+    NativeConfig.displayAuthority(
+      new URI("ABFSS://data:SECRET@acct.dfs.core.windows.net:8443/p?sig=SAS-MARKER")) shouldBe
+      "abfss://data@acct.dfs.core.windows.net:8443"
+    NativeConfig.displayAuthority(new URI("/tmp/t/k.parquet")) shouldBe "file://"
+  }
+
   test("extractScanObjectStoreOptions - files in one container keep the first file's options") {
     val hadoopConf = new Configuration()
     hadoopConf.set("fs.azure.account.key.accta.dfs.core.windows.net", "a2V5LWE=")
@@ -824,6 +878,17 @@ class NativeConfigSuite extends AnyFunSuite with Matchers {
       "auth type UserboundSASWithOAuth")
   }
 
+  test("nativeFallbackReason - renders the location as the resolver's other messages do") {
+    val uri =
+      new URI("ABFSS://data:pw-secret@acct.dfs.core.windows.net/dir/f.parquet?sig=SAS-MARKER")
+    val reason =
+      AbfsAuthResolver.nativeFallbackReason(uri, AbfsAuthResolver.Declined("Custom", None)).get
+    reason should include(NativeConfig.displayAuthority(uri))
+    reason should include("abfss://data@acct.dfs.core.windows.net")
+    reason should not include "pw-secret"
+    reason should not include "SAS-MARKER"
+  }
+
   test("nativeStoreError - passes native's answer through") {
     val options = Map("a" -> "1", "b" -> "2")
     var seen = Map.empty[String, String]