chore: lakefs rust integration test

Signed-off-by: Ion Koutsouris <[email protected]>

.github/workflows/build.yml

@@ -139,3 +139,40 @@ jobs:
       - name: Run tests with native-tls
         run: |
           cargo test --no-default-features --features integration_test,s3-native-tls,datafusion
+
+  integration_test_lakefs:
+    name: Integration Tests (LakeFS v1.48)
+    runs-on: ubuntu-latest
+    env:
+      CARGO_INCREMENTAL: 0
+      # Disable full debug symbol generation to speed up CI build and keep memory down
+      # <https://doc.rust-lang.org/cargo/reference/profiles.html>
+      RUSTFLAGS: "-C debuginfo=line-tables-only"
+      # https://github.com/rust-lang/cargo/issues/10280
+      CARGO_NET_GIT_FETCH_WITH_CLI: "true"
+      RUST_BACKTRACE: "1"
+      RUST_LOG: debug
+
+    steps:
+      - uses: actions/checkout@v3
+
+      - name: Install minimal stable with clippy and rustfmt
+        uses: actions-rs/toolchain@v1
+        with:
+          profile: default
+          toolchain: '1.81'
+          override: true
+
+      - name: Download Lakectl
+        run: |
+          wget -q https://github.com/treeverse/lakeFS/releases/download/v1.48.0/lakeFS_1.48.0_Linux_x86_64.tar.gz
+          tar -xf lakeFS_1.48.0_Linux_x86_64.tar.gz -C $GITHUB_WORKSPACE
+          echo "$GITHUB_WORKSPACE/lakectl" >> $GITHUB_PATH
+
+      - name: Start emulated services
+        run: docker compose -f docker-compose-lakefs.yml up -d
+
+      - name: Run tests with rustls (default)
+        run: |
+          cargo test --features integration_test,lakefs,datafusion
+

.github/workflows/codecov.yml

@@ -26,7 +26,7 @@ jobs:
         uses: taiki-e/install-action@cargo-llvm-cov
       - uses: Swatinem/rust-cache@v2
       - name: Generate code coverage
-        run: cargo llvm-cov --features ${DEFAULT_FEATURES} --workspace --codecov --output-path codecov.json -- --skip read_table_version_hdfs
+        run: cargo llvm-cov --features ${DEFAULT_FEATURES} --workspace --codecov --output-path codecov.json -- --skip read_table_version_hdfs --skip test_read_tables_lakefs
       - name: Upload coverage to Codecov
         uses: codecov/codecov-action@v4
         with:

crates/core/src/operations/optimize.rs

@@ -748,7 +748,7 @@ impl MergePlan {
                 #[cfg(not(feature = "datafusion"))]
                 let exec_context = Arc::new(zorder::ZOrderExecContext::new(
                     zorder_columns,
-                    log_store.object_store(),
+                    log_store.object_store(Some(operation_id)),
                     // If there aren't enough bins to use all threads, then instead
                     // use threads within the bins. This is important for the case where
                     // the table is un-partitioned, in which case the entire table is just

crates/lakefs/Cargo.toml

@@ -39,6 +39,7 @@ serial_test = "3"
 deltalake-test = { path = "../test" }
 pretty_env_logger = "0.5.0"
 rand = "0.8"
+which = "7"
 
 
 [features]

crates/lakefs/tests/context.rs

@@ -1,10 +1,7 @@
-// #![cfg(feature = "integration_test")]
+#![cfg(feature = "integration_test")]
 use deltalake_lakefs::register_handlers;
 use deltalake_test::utils::*;
-use std::{
-    collections::HashSet,
-    process::{Command, ExitStatus},
-};
+use std::process::{Command, ExitStatus};
 
 use which::which;
 
@@ -25,21 +22,53 @@ impl StorageIntegration for LakeFSIntegration {
         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");
+
+        set_env_if_not_set("LAKECTL_CREDENTIALS_ACCESS_KEY_ID", "LAKEFSID");
+        set_env_if_not_set("LAKECTL_CREDENTIALS_SECRET_ACCESS_KEY", "LAKEFSKEY");
+        set_env_if_not_set("LAKECTL_SERVER_ENDPOINT_URL", "http://127.0.0.1:8000");
     }
 
     fn create_bucket(&self) -> std::io::Result<ExitStatus> {
-        Ok(())
+        // Bucket is already created in docker-compose
+        Ok(ExitStatus::default())
     }
 
     fn bucket_name(&self) -> String {
-        "bronze"
+        "bronze".to_string()
     }
 
     fn root_uri(&self) -> String {
+        // Default branch is always main
         format!("lakefs://{}/main", self.bucket_name())
     }
 
     fn copy_directory(&self, source: &str, destination: &str) -> std::io::Result<ExitStatus> {
-
+        println!(
+            "Copy directory called with {} {}",
+            source,
+            &format!("{}/{}", self.root_uri(), destination)
+        );
+        let lakectl = which("lakectl").expect("Failed to find lakectl executable");
+
+        // Upload files to branch
+        Command::new(lakectl.clone())
+            .args([
+                "fs",
+                "upload",
+                "-r",
+                "--source",
+                &format!("{}/", source),
+                &format!("{}/{}/", self.root_uri(), destination),
+            ])
+            .status()?;
+
+        // Commit changes
+        Command::new(lakectl)
+            .args([
+                "commit",
+                &format!("{}/", self.root_uri()),
+                "--allow-empty-message",
+            ])
+            .status()
     }
-}
+}

crates/lakefs/tests/integration.rs

@@ -0,0 +1,16 @@
+#![cfg(feature = "integration_test")]
+use deltalake_test::{test_read_tables, IntegrationContext, TestResult};
+use serial_test::serial;
+
+mod context;
+use context::*;
+//
+#[tokio::test]
+#[serial]
+async fn test_read_tables_lakefs() -> TestResult {
+    let context = IntegrationContext::new(Box::<LakeFSIntegration>::default())?;
+
+    test_read_tables(&context).await?;
+
+    Ok(())
+}