laminar_connectors/lakehouse/delta_io/
catalog.rs

1//! Catalog-specific table URI and storage-option resolution.
2
3#[cfg(feature = "delta-lake-glue")]
4use super::info;
5use super::{ConnectorError, HashMap};
6#[cfg(feature = "delta-lake-glue")]
7use laminar_core::storage_location::StorageProvider;
8
9#[cfg(feature = "delta-lake-glue")]
10fn glue_init_error_category(error: &deltalake_catalog_glue::GlueError) -> &'static str {
11    match error {
12        deltalake_catalog_glue::GlueError::MissingMetadata { .. } => "missing-metadata",
13        deltalake_catalog_glue::GlueError::AWSError { source } => aws_glue_error_category(source),
14    }
15}
16
17#[cfg(feature = "delta-lake-glue")]
18fn aws_glue_error_category(error: &aws_sdk_glue::Error) -> &'static str {
19    use aws_sdk_glue::Error;
20
21    match error {
22        Error::AccessDeniedException(_)
23        | Error::KmsKeyNotAccessibleFault(_)
24        | Error::PermissionTypeMismatchException(_) => "authorization",
25        Error::EntityNotFoundException(_)
26        | Error::IntegrationNotFoundFault(_)
27        | Error::ResourceNotFoundException(_)
28        | Error::TargetResourceNotFound(_) => "not-found",
29        Error::ConcurrentRunsExceededException(_)
30        | Error::IntegrationQuotaExceededFault(_)
31        | Error::ResourceNumberLimitExceededException(_)
32        | Error::ThrottlingException(_) => "throttled",
33        Error::OperationTimeoutException(_) => "service-timeout",
34        Error::FederationSourceRetryableException(_)
35        | Error::InternalServerException(_)
36        | Error::InternalServiceException(_)
37        | Error::ResourceNotReadyException(_) => "service-unavailable",
38        Error::InvalidInputException(_) | Error::ValidationException(_) => "invalid-request",
39        Error::AlreadyExistsException(_)
40        | Error::ConcurrentModificationException(_)
41        | Error::ConflictException(_)
42        | Error::VersionMismatchException(_) => "catalog-conflict",
43        _ => aws_glue_pre_service_error_category(error).unwrap_or("unclassified-service-error"),
44    }
45}
46
47#[cfg(feature = "delta-lake-glue")]
48fn aws_glue_pre_service_error_category(error: &aws_sdk_glue::Error) -> Option<&'static str> {
49    type GetTableSdkError =
50        aws_sdk_glue::error::SdkError<aws_sdk_glue::operation::get_table::GetTableError>;
51
52    let mut source = std::error::Error::source(error);
53    while let Some(current) = source {
54        if current
55            .downcast_ref::<aws_sdk_glue::error::BuildError>()
56            .is_some()
57        {
58            return Some("client-input");
59        }
60        if let Some(sdk_error) = current.downcast_ref::<GetTableSdkError>() {
61            return match sdk_error {
62                GetTableSdkError::ConstructionFailure(_) => Some("client-input"),
63                GetTableSdkError::TimeoutError(_) => Some("client-timeout"),
64                GetTableSdkError::DispatchFailure(_) => Some("transport"),
65                GetTableSdkError::ResponseError(_) => Some("invalid-response"),
66                GetTableSdkError::ServiceError(_) => None,
67                _ => Some("unclassified-sdk-error"),
68            };
69        }
70        source = current.source();
71    }
72
73    None
74}
75
76#[cfg(feature = "delta-lake-glue")]
77fn glue_lookup_error_category(error: &deltalake::data_catalog::DataCatalogError) -> &'static str {
78    match error {
79        deltalake::data_catalog::DataCatalogError::Generic { source, .. } => source
80            .downcast_ref::<deltalake_catalog_glue::GlueError>()
81            .map_or("catalog-provider", glue_init_error_category),
82        deltalake::data_catalog::DataCatalogError::InvalidDataCatalog { .. } => "invalid-catalog",
83        deltalake::data_catalog::DataCatalogError::UnknownConfigKey { .. } => {
84            "invalid-catalog-configuration"
85        }
86        deltalake::data_catalog::DataCatalogError::RequestError { .. } => "catalog-request",
87    }
88}
89
90/// Whether a catalog-resolved table is opened for reading or for commits.
91#[derive(Debug, Clone, Copy, PartialEq, Eq)]
92pub enum CatalogAccess {
93    /// Snapshot, change-feed, reference, or lookup reads.
94    Read,
95    /// Sink writes and table preparation.
96    Write,
97}
98
99/// A table location resolved through its catalog.
100#[derive(Clone)]
101pub struct ResolvedCatalogTable {
102    /// Storage URI of the Delta table root.
103    pub location: String,
104    /// Configured storage options plus any credentials the catalog vended for this access.
105    pub storage_options: HashMap<String, String>,
106    /// When vended credentials expire; `None` for credentials that do not expire here.
107    pub credentials_expire_at: Option<std::time::SystemTime>,
108}
109
110impl std::fmt::Debug for ResolvedCatalogTable {
111    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
112        f.debug_struct("ResolvedCatalogTable")
113            .field("location", &self.location)
114            .field(
115                "storage_options",
116                &crate::storage::SecretMasker::display_map(&self.storage_options),
117            )
118            .field("credentials_expire_at", &self.credentials_expire_at)
119            .finish()
120    }
121}
122
123/// Vended credentials are renewed once less than this remains before they expire.
124const CREDENTIAL_REFRESH_MARGIN: std::time::Duration = std::time::Duration::from_secs(10 * 60);
125
126/// When credentials expiring at `expire_at` are due for renewal.
127#[must_use]
128pub fn credentials_refresh_at(
129    expire_at: Option<std::time::SystemTime>,
130) -> Option<std::time::SystemTime> {
131    expire_at.map(|at| {
132        at.checked_sub(CREDENTIAL_REFRESH_MARGIN)
133            .unwrap_or(std::time::UNIX_EPOCH)
134    })
135}
136
137/// Whether credentials expiring at `expire_at` must be renewed before the next storage I/O.
138#[must_use]
139pub fn credentials_need_refresh(expire_at: Option<std::time::SystemTime>) -> bool {
140    credentials_refresh_at(expire_at).is_some_and(|at| std::time::SystemTime::now() >= at)
141}
142
143/// Resolves catalog-aware table URI and merges catalog-specific storage options.
144///
145/// - `None`: returns table path and storage options as-is.
146/// - `Glue`: calls AWS Glue API to resolve the table's S3 location.
147/// - `Unity`: looks the table up through the Unity Catalog API, rejects kinds that cannot serve
148///   `access`, and adds table-scoped vended credentials unless configured credentials are chosen.
149///
150/// # Errors
151///
152/// Returns `ConnectorError` if catalog resolution fails.
153#[cfg(feature = "delta-lake")]
154#[allow(clippy::implicit_hasher, clippy::unused_async)]
155pub async fn resolve_catalog_options(
156    catalog: &super::super::delta_config::DeltaCatalogType,
157    #[allow(unused_variables)] catalog_database: Option<&str>,
158    #[allow(unused_variables)] catalog_name: Option<&str>,
159    _catalog_schema: Option<&str>,
160    table_path: &str,
161    base_storage_options: &HashMap<String, String>,
162    #[allow(unused_variables)] access: CatalogAccess,
163) -> Result<ResolvedCatalogTable, ConnectorError> {
164    use super::super::delta_config::DeltaCatalogType;
165
166    match catalog {
167        DeltaCatalogType::None => Ok(ResolvedCatalogTable {
168            location: table_path.to_string(),
169            storage_options: base_storage_options.clone(),
170            credentials_expire_at: None,
171        }),
172        #[cfg(feature = "delta-lake-glue")]
173        DeltaCatalogType::Glue => {
174            use deltalake::DataCatalog;
175            let database = catalog_database.ok_or_else(|| {
176                ConnectorError::ConfigurationError(
177                    "Glue catalog requires 'catalog.database'".into(),
178                )
179            })?;
180            let glue = deltalake_catalog_glue::GlueDataCatalog::from_env()
181                .await
182                .map_err(|error| {
183                    ConnectorError::ConnectionFailed(
184                        format!(
185                            "failed to initialize Glue catalog using the downstream AWS credential chain ({}); verify AWS region and credential-chain setup",
186                            glue_init_error_category(&error)
187                        ),
188                    )
189                })?;
190            let resolved = glue
191                .get_table_storage_location(catalog_name.map(String::from), database, table_path)
192                .await
193                .map_err(|error| {
194                    ConnectorError::ConnectionFailed(
195                        format!(
196                            "Glue catalog lookup failed ({}); verify catalog identifiers, AWS region, credentials, and glue:GetTable permission",
197                            glue_lookup_error_category(&error)
198                        ),
199                    )
200                })?;
201            info!(
202                glue_database = database,
203                storage_provider =
204                    StorageProvider::detect_uri(&resolved).map_or("unknown", StorageProvider::name),
205                "resolved table path via Glue catalog"
206            );
207            Ok(ResolvedCatalogTable {
208                location: resolved,
209                storage_options: base_storage_options.clone(),
210                credentials_expire_at: None,
211            })
212        }
213        #[cfg(not(feature = "delta-lake-glue"))]
214        DeltaCatalogType::Glue => Err(ConnectorError::ConfigurationError(
215            "Glue catalog requires the 'delta-lake-glue' feature. \
216             Build with: cargo build --features delta-lake-glue"
217                .into(),
218        )),
219        #[cfg(feature = "delta-lake-unity")]
220        DeltaCatalogType::Unity(unity) => {
221            let resolved =
222                super::super::unity_catalog::resolve_table(unity, table_path, access).await?;
223            let mut storage_options = base_storage_options.clone();
224            storage_options.extend(resolved.vended_options);
225            Ok(ResolvedCatalogTable {
226                location: resolved.storage_location,
227                storage_options,
228                credentials_expire_at: resolved.expires_at,
229            })
230        }
231        #[cfg(not(feature = "delta-lake-unity"))]
232        DeltaCatalogType::Unity(_) => Err(ConnectorError::ConfigurationError(
233            "Unity catalog requires the 'delta-lake-unity' feature. \
234             Build with: cargo build --features delta-lake-unity"
235                .into(),
236        )),
237    }
238}
239
240#[cfg(all(test, feature = "delta-lake-glue"))]
241mod tests {
242    use super::*;
243
244    #[test]
245    fn glue_categories_are_actionable_without_provider_details() {
246        let error: deltalake::data_catalog::DataCatalogError =
247            deltalake_catalog_glue::GlueError::MissingMetadata {
248                metadata: "secret-table-location".into(),
249            }
250            .into();
251        assert_eq!(glue_lookup_error_category(&error), "missing-metadata");
252        assert!(!glue_lookup_error_category(&error).contains("secret"));
253
254        let error = deltalake_catalog_glue::GlueError::AWSError {
255            source: aws_sdk_glue::Error::AccessDeniedException(
256                aws_sdk_glue::types::error::AccessDeniedException::builder()
257                    .message("private provider detail")
258                    .build(),
259            ),
260        };
261        assert_eq!(glue_init_error_category(&error), "authorization");
262        assert!(!glue_init_error_category(&error).contains("private"));
263
264        let unhandled = aws_sdk_glue::types::TableInput::builder()
265            .build()
266            .expect_err("missing table name must produce a client-side build error");
267        let error = deltalake_catalog_glue::GlueError::AWSError {
268            source: unhandled.into(),
269        };
270        assert_eq!(glue_init_error_category(&error), "client-input");
271
272        let timeout: aws_sdk_glue::error::SdkError<
273            aws_sdk_glue::operation::get_table::GetTableError,
274        > = aws_sdk_glue::error::SdkError::timeout_error(std::io::Error::other(
275            "private transport detail",
276        ));
277        let error = deltalake_catalog_glue::GlueError::AWSError {
278            source: timeout.into(),
279        };
280        assert_eq!(glue_init_error_category(&error), "client-timeout");
281
282        let no_code_service_error = aws_sdk_glue::operation::get_table::GetTableError::unhandled(
283            std::io::Error::other("private provider response"),
284        );
285        let error = deltalake_catalog_glue::GlueError::AWSError {
286            source: no_code_service_error.into(),
287        };
288        assert_eq!(
289            glue_init_error_category(&error),
290            "unclassified-service-error"
291        );
292    }
293}
294
295// ============================================================================
296// Integration tests (require delta-lake feature)
297// ============================================================================
laminar_connectors/lakehouse/delta_io/catalog.rs

laminar_connectors/lakehouse/delta_io/
catalog.rs
