Repository: risingwavelabs/risingwave
File: src/connector/src/connector_common/iceberg/mod.rs (Rust)

Code before:
                    let (bucket, _) = {
                        let is_s3_tables = warehouse_path.starts_with("arn:aws:s3tables");
                        // BigLake catalog federation uses bq:// prefix for BigQuery-managed Iceberg tables
                        let is_bq_catalog_federation = warehouse_path.starts_with("bq://");
                        let url = Url::parse(warehouse_path);
                        if (url.is_err() || is_s3_tables || is_bq_catalog_federation)
                            && (catalog_type == "rest" || catalog_type == "rest_rust")
                        {
                            // If the warehouse path is not a valid URL, it could be:
                            // - A warehouse name in REST catalog
                            // - An S3 Tables path (arn:aws:s3tables:...)
                            // - A BigLake path (bq://projects/...) for Google Cloud BigQuery integration
                            // We allow these to pass through for REST catalogs.
                            (None, None)
                        } else {
                            let url = url.with_context(|| {
                                format!("Invalid warehouse path: {}", warehouse_path)
                            })?;
                            let bucket = url
                                .host_str()
                                .with_context(|| {
                                    format!(
                                        "Invalid s3 path: {}, bucket is missing",
                                        warehouse_path
                                    )
                                })?
                                .to_owned();
                            let root = url.path().trim_start_matches('/').to_owned();
                            (Some(bucket), Some(root))
                        }
                    };

                    if let Some(bucket) = bucket {
                        iceberg_configs.insert("iceberg.table.io.bucket".to_owned(), bucket);
                    }
                }
                None => {
                    if catalog_type != "rest" && catalog_type != "rest_rust" {
                        bail!("`warehouse.path` must be set in {} catalog", &catalog_type);
                    }
                }
            }
            iceberg_configs.insert(
                S3_DISABLE_CONFIG_LOAD.to_owned(),
                (!enable_config_load).to_string(),
            );

            iceberg_configs.insert(
                GCS_DISABLE_CONFIG_LOAD.to_owned(),
                (!enable_config_load).to_string(),
            );

Review comment (bug):
The comment should be updated to reflect the new logic. The comment now includes BigLake paths but should also mention that for BigLake paths (bq://), the URL parsing will fail, which is why they're included in the condition. The phrase "if the warehouse path is not a valid URL" should be clarified to explicitly state that BigLake paths starting with "bq://" are intentionally not valid URLs.
```suggestion
                            // If the warehouse path is not a standard URL that `Url::parse` can handle, it could be:
                            // - A warehouse name in REST catalog
                            // - An S3 Tables path (arn:aws:s3tables:...)
                            // - A BigLake path (bq://projects/...) for Google Cloud BigQuery integration, which
                            //   intentionally uses the non-URL `bq://` scheme so that URL parsing fails
                            // In these cases, we treat the value as a logical warehouse identifier and allow it
                            // to pass through for REST catalogs without extracting bucket/root from a URL.
```

Code after:
            if let (Some(account_name), Some(account_key)) = (
                self.adlsgen2_account_name.as_ref(),
                self.adlsgen2_account_key.as_ref(),
            ) {
                iceberg_configs.insert(ADLS_ACCOUNT_NAME.to_owned(), account_name.clone());
                iceberg_configs.insert(ADLS_ACCOUNT_KEY.to_owned(), account_key.clone());
                if catalog_type != "rest" && catalog_type != "rest_rust" {
                    bail!("adlsgen2 unsupported in {} catalog", &catalog_type);
                }
            }

            match &self.warehouse_path {
                Some(warehouse_path) => {
                    let (bucket, _) = {
                        let is_s3_tables = warehouse_path.starts_with("arn:aws:s3tables");
                        // BigLake catalog federation uses bq:// prefix for BigQuery-managed Iceberg tables
                        let is_bq_catalog_federation = warehouse_path.starts_with("bq://");
                        let url = Url::parse(warehouse_path);
                        if (url.is_err() || is_s3_tables || is_bq_catalog_federation)
                            && (catalog_type == "rest" || catalog_type == "rest_rust")
                        {
                            // If the warehouse path is not a valid URL, it could be:
                            // - A warehouse name in REST catalog
                            // - An S3 Tables path (arn:aws:s3tables:...)
                            // - A BigLake path (bq://projects/...) for Google Cloud BigQuery integration
                            // We allow these to pass through for REST catalogs.
                            (None, None)
                        } else {
                            let url = url.with_context(|| {
                                format!("Invalid warehouse path: {}", warehouse_path)
                            })?;
                            let bucket = url
                                .host_str()
                                .with_context(|| {
                                    format!(
                                        "Invalid s3 path: {}, bucket is missing",
                                        warehouse_path
                                    )
                                })?
                                .to_owned();
                            let root = url.path().trim_start_matches('/').to_owned();
                            (Some(bucket), Some(root))
                        }
                    };

                    if let Some(bucket) = bucket {
                        iceberg_configs.insert("iceberg.table.io.bucket".to_owned(), bucket);
                    }
                }
                None => {
                    if catalog_type != "rest" && catalog_type != "rest_rust" {