-
Notifications
You must be signed in to change notification settings - Fork 433
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
chore: restructure crate, make it feature
- Loading branch information
1 parent
ceed6f7
commit a65f72c
Showing
10 changed files
with
146 additions
and
98 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,87 @@ | ||
| use async_trait::async_trait; | ||
| use deltalake_core::{ | ||
| logstore::LogStoreRef, operations::CustomExecuteHandler, DeltaResult, DeltaTableError, | ||
| }; | ||
| use tracing::debug; | ||
| use uuid::Uuid; | ||
|
|
||
| use crate::logstore::LakeFSLogStore; | ||
|
|
||
| pub struct LakeFSCustomExecuteHandler {} | ||
|
|
||
| #[async_trait] | ||
| impl CustomExecuteHandler for LakeFSCustomExecuteHandler { | ||
| // LakeFS Log store pre execution of delta operation (create branch, logs object store and transaction) | ||
| async fn pre_execute(&self, log_store: &LogStoreRef, operation_id: Uuid) -> DeltaResult<()> { | ||
| debug!("Running LakeFS pre execution inside delta operation"); | ||
| if let Some(lakefs_store) = log_store.clone().as_any().downcast_ref::<LakeFSLogStore>() { | ||
| lakefs_store.pre_execute(operation_id).await | ||
| } else { | ||
| Err(DeltaTableError::generic( | ||
| "LakeFSPreEcuteHandler is used, but no LakeFSLogStore has been found", | ||
| )) | ||
| } | ||
| } | ||
| // Not required for LakeFS | ||
| async fn post_execute(&self, log_store: &LogStoreRef, operation_id: Uuid) -> DeltaResult<()> { | ||
| debug!("Running LakeFS post execution inside delta operation"); | ||
| if let Some(lakefs_store) = log_store.clone().as_any().downcast_ref::<LakeFSLogStore>() { | ||
| let (repo, _, _) = lakefs_store | ||
| .client | ||
| .decompose_url(lakefs_store.config.location.to_string()); | ||
| let result = lakefs_store | ||
| .client | ||
| .delete_branch(repo, lakefs_store.client.get_transaction(operation_id)?) | ||
| .await | ||
| .map_err(|e| DeltaTableError::Transaction { source: e }); | ||
| lakefs_store.client.clear_transaction(operation_id); | ||
| result | ||
| } else { | ||
| Err(DeltaTableError::generic( | ||
| "LakeFSPreEcuteHandler is used, but no LakeFSLogStore has been found", | ||
| )) | ||
| } | ||
| } | ||
|
|
||
| // Execute arbitrary code at the start of the post commit hook | ||
| async fn before_post_commit_hook( | ||
| &self, | ||
| log_store: &LogStoreRef, | ||
| file_operations: bool, | ||
| operation_id: Uuid, | ||
| ) -> DeltaResult<()> { | ||
| if file_operations { | ||
| debug!("Running LakeFS pre execution inside post_commit_hook"); | ||
| if let Some(lakefs_store) = log_store.clone().as_any().downcast_ref::<LakeFSLogStore>() | ||
| { | ||
| lakefs_store.pre_execute(operation_id).await | ||
| } else { | ||
| Err(DeltaTableError::generic( | ||
| "LakeFSPreEcuteHandler is used, but no LakeFSLogStore has been found", | ||
| )) | ||
| }?; | ||
| } | ||
| Ok(()) | ||
| } | ||
|
|
||
| // Execute arbitrary code at the end of the post commit hook | ||
| async fn after_post_commit_hook( | ||
| &self, | ||
| log_store: &LogStoreRef, | ||
| file_operations: bool, | ||
| operation_id: Uuid, | ||
| ) -> DeltaResult<()> { | ||
| if file_operations { | ||
| debug!("Running LakeFS post execution inside post_commit_hook"); | ||
| if let Some(lakefs_store) = log_store.clone().as_any().downcast_ref::<LakeFSLogStore>() | ||
| { | ||
| lakefs_store.commit_merge(operation_id).await | ||
| } else { | ||
| Err(DeltaTableError::generic( | ||
| "LakeFSPreEcuteHandler is used, but no LakeFSLogStore has been found", | ||
| )) | ||
| }?; | ||
| } | ||
| Ok(()) | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,45 @@ | ||
| // #![cfg(feature = "integration_test")] | ||
| use deltalake_lakefs::register_handlers; | ||
| use deltalake_test::utils::*; | ||
| use std::{ | ||
| collections::HashSet, | ||
| process::{Command, ExitStatus}, | ||
| }; | ||
|
|
||
| use which::which; | ||
|
|
||
| pub struct LakeFSIntegration {} | ||
|
|
||
| impl Default for LakeFSIntegration { | ||
| fn default() -> Self { | ||
| register_handlers(None); | ||
| Self {} | ||
| } | ||
| } | ||
|
|
||
| impl StorageIntegration for LakeFSIntegration { | ||
| fn prepare_env(&self) { | ||
| println!("Preparing env"); | ||
|
|
||
| set_env_if_not_set("endpoint", "http://127.0.0.1:8000"); | ||
| set_env_if_not_set("access_key_id", "LAKEFSID"); | ||
| set_env_if_not_set("secret_access_key", "LAKEFSKEY"); | ||
| set_env_if_not_set("allow_http", "true"); | ||
| } | ||
|
|
||
| fn create_bucket(&self) -> std::io::Result<ExitStatus> { | ||
| Ok(()) | ||
| } | ||
|
|
||
| fn bucket_name(&self) -> String { | ||
| "bronze" | ||
| } | ||
|
|
||
| fn root_uri(&self) -> String { | ||
| format!("lakefs://{}/main", self.bucket_name()) | ||
| } | ||
|
|
||
| fn copy_directory(&self, source: &str, destination: &str) -> std::io::Result<ExitStatus> { | ||
|
|
||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters