@@ -71,14 +71,16 @@
 
 use log::debug;
 use std::collections::HashMap;
-use url::{Position, Url};
+use url::Url;
 
 use object_store::{
     azure::{AzureConfigKey, MicrosoftAzureBuilder},
     path::Path,
     ObjectStore, ObjectStoreScheme,
 };
 
+use super::url_redaction::{display_authority, redacted, 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";
@@ -328,50 +330,6 @@ fn authority_host(authority: &str) -> String {
     authority.trim_end_matches('/').to_string()
 }
 
-// A URL's password and query (where a SAS token sits) can carry a credential, so no message
-// or URL handed to object_store, which quotes URLs in its errors, keeps them. The path is not
-// treated as secret.
-
-/// `url` without its password, query and fragment. The username stays: in an ABFS URL it is the
-/// container, which `MicrosoftAzureBuilder` reads.
-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`] without the username too, for a URL that may not be Azure, where the
-/// username can be an access key ID. `ObjectStoreScheme::parse` reads only the scheme, host and
-/// path.
-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
-}
-
-/// `scheme://[container@]host[:port]` of an ABFS URL, for messages.
-fn display_authority(url: &Url) -> String {
-    let safe = without_secrets(url);
-    format!(
-        "{}://{}",
-        safe.scheme(),
-        &safe[Position::BeforeUsername..Position::AfterPort]
-    )
-}
-
-/// `scheme://host[:port]` of a URL that may not be Azure, for messages.
-fn redacted(url: &Url) -> String {
-    format!(
-        "{}://{}",
-        url.scheme(),
-        &url[Position::BeforeHost..Position::AfterPort]
-    )
-}
-
 fn config_error(message: String) -> object_store::Error {
     object_store::Error::Generic {
         store: "MicrosoftAzure",
 
@@ -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;
@@ -77,8 +77,15 @@ 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, so only a valid scheme is named. `ParseError` names
+        // the problem without quoting the input.
+        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 +102,15 @@ pub(crate) fn normalize_object_store_url(
     })
 }
 
+/// The text before `://` in a URL that failed to parse, if it is a valid scheme.
+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 +319,29 @@ 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}");
+        // The text before `://` is not 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}");
+        // Starts like a scheme, but '@' is not a scheme character.
+        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");
 
@@ -0,0 +1,66 @@
+// 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 safe to put in messages, store keys and URLs handed to object_store, which quotes
+//! URLs in its errors. A URL's password and query (where a SAS token sits) can carry a
+//! credential. The path is not treated as secret.
+
+use std::borrow::Cow;
+use url::{Position, Url};
+
+/// `url` without its password, query and fragment. The username stays: 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`] without the username too, for a URL that may not be Azure, where the
+/// username can be an access key ID. `ObjectStoreScheme::parse` reads only the scheme, host and
+/// path.
+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
+}
+
+/// `[username@]host[:port]` of `url`, without the password.
+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())
+}
+
+/// `scheme://[container@]host[:port]` of an ABFS URL, for messages.
+pub(crate) fn display_authority(url: &Url) -> String {
+    format!("{}://{}", url.scheme(), authority_without_password(url))
+}
+
+/// `scheme://host[:port]` of a URL that may not be Azure, for messages.
+pub(crate) fn redacted(url: &Url) -> String {
+    format!(
+        "{}://{}",
+        url.scheme(),
+        &url[Position::BeforeHost..Position::AfterPort]
+    )
+}