feat(puffin): complete dir support

[zhongzc]

I hereby agree to the terms of the GreptimeDB CLA.

Refer to a related PR or issue link (optional)

close #4193

What's changed and what's your intention?

as @coderabbitai

Checklist

• I have written the necessary rustdoc comments.
• I have added the necessary unit tests and integration tests.
• This PR requires documentation updates.

Summary by CodeRabbit

• New Features

  • Introduced CachedPuffinManager for managing cached readers and writers.
  • Added FsPuffinManager for handling puffin data in the filesystem.
  • Introduced test functionalities for CachedPuffinManager with various compression codecs.

• Bug Fixes

  • Updated comments for better clarity in reader functions.

• Refactor

  • Renamed CachedPuffinReader to FsPuffinReader.
  • Renamed CachedPuffinWriter to FsPuffinWriter.
  • Improved compression handling logic.

• Documentation

  • Enhanced comments for public methods to improve understanding.

• Chores

  • Removed unused DELETE_QUEUE_SIZE constant.

[coderabbitai]

Note

Reviews paused

Use the following commands to manage reviews:

• @coderabbitai resume to resume automatic reviews.
• @coderabbitai review to trigger a single review.

Walkthrough

The changes to the Puffin repository revolve around enhancing the PuffinManager by introducing directory support and data compression mechanisms. This includes updating reader and writer structures, implementing a cache manager, and adding comprehensive test cases. These modifications ensure efficient file handling, caching, and compression capabilities within the Puffin system.

Changes

File(s)	Change Summary
src/puffin/src/file_format/reader.rs, .../fs_puffin_manager/reader.rs, .../fs_puffin_manager/writer.rs	Updated comments for clarity, added data compression comment, and renamed structs/functions to establish uniform naming.
.../puffin_manager/cached_puffin_manager/cached_puffin_manager.rs, .../cache_manager/moka_cache_manager.rs	Introduced CachedPuffinManager, removed DELETE_QUEUE_SIZE constant, implemented new struct methods and trait extensions.
src/puffin/src/puffin_manager/tests.rs	Added new test functions for CachedPuffinManager covering directory and file operations with various compression codecs.
src/puffin/src/puffin_manager.rs	Introduced new test module and updated BlobGuard trait to include the Unpin requirement for Reader.
src/puffin/src/puffin_manager/fs_puffin_manager.rs	Introduced FsPuffinManager, providing filesystem-oriented puffin data management with reading and writing functionalities.

Sequence Diagram(s)

sequenceDiagram
    participant User
    participant CachedPuffinManager
    participant Cache
    participant FsPuffinWriter
    participant FsPuffinReader

    User->>CachedPuffinManager: Put directory
    CachedPuffinManager->>FsPuffinWriter: Write data with compression
    FsPuffinWriter->>Cache: Store compressed data
    User->>CachedPuffinManager: Get directory
    CachedPuffinManager->>Cache: Retrieve data
    Cache->>FsPuffinReader: Fetch compressed data
    FsPuffinReader->>cachedPuffinManager: Read data
    CachedPuffinManager->>User: Return data

Loading

Assessment against linked issues

Objective	Addressed	Explanation
Support footer lz4 compression (#4193)	✅
Introduce puffin manager trait (#4193)	✅
Introduce writer for puffin manager (#4193)	✅
Introduce cache manager (#4193)	✅
Introduce reader for puffin manager (#4193)	✅

Poem

In a warren of code so tight,
Puffin files now take flight,
With caches swift and data compressed,
Our changes stand the test,
Bringing joy, both day and night.
🐇✨

---

Tip

Early access features: enabled

We are currently testing the following features in early access:

• OpenAI gpt-4o model for code reviews and chat: OpenAI claims that this model is better at understanding and generating code than the previous models. We seek your feedback over the next few weeks before making it generally available.

Note:

• You can enable or disable early access features from the CodeRabbit UI or by updating the CodeRabbit configuration file.
• Please join our Discord Community to provide feedback and report issues.
• OSS projects are currently opted into early access features by default.

Thank you for using CodeRabbit. We offer it for free to the OSS community and would appreciate your support in helping us grow. If you find it useful, would you consider giving us a shout-out on your favorite social media?

Share

• X
• Mastodon
• Reddit
• LinkedIn

Tips

Chat

There are 3 ways to chat with CodeRabbit:

• Review comments: Directly reply to a review comment made by CodeRabbit. Example:
  • I pushed a fix in commit <commit_id>.
  • Generate unit testing code for this file.
  • Open a follow-up GitHub issue for this discussion.
• Files and specific lines of code (under the "Files changed" tab): Tag @coderabbitai in a new review comment at the desired location with your query. Examples:
  • @coderabbitai generate unit testing code for this file.
  • @coderabbitai modularize this function.
• PR comments: Tag @coderabbitai in a new PR comment to ask questions about the PR branch. For the best results, please provide a very specific query, as very limited context is provided in this mode. Examples:
  • @coderabbitai generate interesting stats about this repository and render them as a table.
  • @coderabbitai show all the console.log statements in this repository.
  • @coderabbitai read src/utils.ts and generate unit testing code.
  • @coderabbitai read the files in the src/scheduler package and generate a class diagram using mermaid and a README in the markdown format.
  • @coderabbitai help me debug CodeRabbit configuration file.

Note: Be mindful of the bot's finite context window. It's strongly recommended to break down tasks such as reading entire modules into smaller chunks. For a focused discussion, use review comments to chat about specific files and their changes, instead of using the PR comments.

CodeRabbit Commands (invoked as PR comments)

• @coderabbitai pause to pause the reviews on a PR.
• @coderabbitai resume to resume the paused reviews.
• @coderabbitai review to trigger an incremental review. This is useful when automatic reviews are disabled for the repository.
• @coderabbitai full review to do a full review from scratch and review all the files again.
• @coderabbitai summary to regenerate the summary of the PR.
• @coderabbitai resolve resolve all the CodeRabbit review comments.
• @coderabbitai configuration to show the current CodeRabbit configuration for the repository.
• @coderabbitai help to get help.

Additionally, you can add @coderabbitai ignore anywhere in the PR description to prevent this PR from being reviewed.

CodeRabbit Configration File (.coderabbit.yaml)

• You can programmatically configure CodeRabbit by adding a .coderabbit.yaml file to the root of your repository.
• Please see the configuration documentation for more information.
• If your editor has YAML language server enabled, you can add the path at the top of this file to enable auto-completion and validation: # yaml-language-server: $schema=https://coderabbit.ai/integrations/schema.v2.json

Documentation and Community

• Visit our Documentation for detailed information on how to use CodeRabbit.
• Join our Discord Community to get help, request features, and share feedback.
• Follow us on X/Twitter for updates and announcements.

[coderabbitai]

Actionable comments posted: 0

Review details

Configuration used: CodeRabbit UI
Review profile: CHILL

Commits

Files that changed from the base of the PR and between e531326 and 172cbb4.

Files ignored due to path filters (1)

• Cargo.lock is excluded by !**/*.lock

Files selected for processing (11)

• src/puffin/Cargo.toml (1 hunks)
• src/puffin/src/blob_metadata.rs (2 hunks)
• src/puffin/src/file_format/reader/file.rs (1 hunks)
• src/puffin/src/file_format/writer/file.rs (4 hunks)
• src/puffin/src/puffin_manager.rs (2 hunks)
• src/puffin/src/puffin_manager/cache_manager.rs (1 hunks)
• src/puffin/src/puffin_manager/cache_manager/moka_cache_manager.rs (1 hunks)
• src/puffin/src/puffin_manager/cached_puffin_manager.rs (1 hunks)
• src/puffin/src/puffin_manager/cached_puffin_manager/reader.rs (1 hunks)
• src/puffin/src/puffin_manager/cached_puffin_manager/writer.rs (5 hunks)
• src/puffin/src/puffin_manager/tests.rs (1 hunks)

Files skipped from review due to trivial changes (4)

• src/puffin/Cargo.toml
• src/puffin/src/puffin_manager/cache_manager.rs
• src/puffin/src/puffin_manager/cache_manager/moka_cache_manager.rs
• src/puffin/src/puffin_manager/cached_puffin_manager/reader.rs

Additional comments not posted (10)

src/puffin/src/puffin_manager/cached_puffin_manager.rs (3)

31-37: Struct definition looks good.

The CachedPuffinManager struct is well-defined with appropriate fields.

---

39-49: Constructor method looks good.

The new method correctly initializes the struct fields.

---

51-77: Trait implementation looks good.

The reader and writer methods are well-defined and follow async conventions.

src/puffin/src/puffin_manager.rs (1)

95-95: Trait modification looks good.

The addition of the Unpin trait to the BlobGuard trait's associated type Reader is appropriate for async operations.

src/puffin/src/file_format/reader/file.rs (1)

Line range hint 73-76: Document the removal of decompression support.

The blob_reader method no longer supports decompression. Consider documenting this change to inform users.

src/puffin/src/file_format/writer/file.rs (1)

60-67: Method modification looks good.

The addition of the compression_codec parameter to the create_blob_metadata method is appropriate and well-implemented.

src/puffin/src/puffin_manager/cached_puffin_manager/writer.rs (2)

173-201: LGTM!

The handle_compress function correctly handles different compression codecs and falls back to no compression if none is specified.

---

82-84: LGTM! But verify the method usage in the codebase.

The code changes are approved.

However, ensure that all function calls to handle_compress are correct and match the new signature.

Verification successful

Let's examine the handle_compress function definition to confirm its parameters and ensure that the calls on lines 82 and 100 match the new signature.

---

Let's use rg to search for the handle_compress function definition and verify its parameters to ensure the calls match the new signature.

---

The method usage for handle_compress is correct.

Both calls to handle_compress match the new function signature:

• key.to_string(), file_key.clone() -> key: String
• raw_data, reader -> raw_data: impl AsyncRead + Send
• options.compression -> compression: Option<CompressionCodec>

No issues found with the method usage.

Scripts executed

The following scripts were executed for the analysis:

---

Script:

#!/bin/bash
# Description: Verify all method calls to `handle_compress` match the new signature.

# Test: Search for the method usage. Expect: Only occurrences of the new signature.
rg --type rust $'handle_compress'

Length of output: 405

---

---

Script:

#!/bin/bash
# Extract the `handle_compress` function definition to verify its parameters
ast-grep --lang rust --pattern $'async fn handle_compress($_, $_, $_) {
  $$$
}'

Length of output: 81

---

---

Script:

#!/bin/bash
# Search for the `handle_compress` function definition to verify its parameters
rg 'async fn handle_compress' src/puffin/src/puffin_manager/cached_puffin_manager/writer.rs -A 5

Length of output: 279

src/puffin/src/blob_metadata.rs (1)

59-59: LGTM! But verify the field usage in the codebase.

The code changes are approved.

However, ensure that all instances of compression_codec are handled correctly in the codebase.

src/puffin/src/puffin_manager/tests.rs (1)

1-361: LGTM!

The new tests are comprehensive and cover various scenarios for putting and getting files and directories with different compression codecs.

[coderabbitai]

Actionable comments posted: 9

Review details

Configuration used: CodeRabbit UI
Review profile: CHILL

Commits

Files that changed from the base of the PR and between 172cbb4 and ddaf716.

Files selected for processing (2)

• src/puffin/src/puffin_manager/cached_puffin_manager.rs (1 hunks)
• src/puffin/src/puffin_manager/tests.rs (1 hunks)

Files skipped from review as they are similar to previous changes (1)

• src/puffin/src/puffin_manager/cached_puffin_manager.rs

Additional comments not posted (1)

src/puffin/src/puffin_manager/tests.rs (1)

15-31: Consider removing redundant imports.

The import futures::AsyncReadExt as _; seems redundant as tokio::io::AsyncReadExt as _; is already imported.

- use futures::AsyncReadExt as _;

[codecov]

Codecov Report

Attention: Patch coverage is 98.82629% with 5 lines in your changes missing coverage. Please review.

Project coverage is 84.73%. Comparing base (e531326) to head (aaa5ea1).
Report is 3 commits behind head on main.

Additional details and impacted files

@@            Coverage Diff             @@
##             main    #4240      +/-   ##
==========================================
- Coverage   84.84%   84.73%   -0.11%     
==========================================
  Files        1049     1051       +2     
  Lines      186267   186627     +360     
==========================================
+ Hits       158029   158134     +105     
- Misses      28238    28493     +255     

[killme2008]

LGTM

[coderabbitai]

Actionable comments posted: 8

Review details

Configuration used: CodeRabbit UI
Review profile: CHILL

Commits

Files that changed from the base of the PR and between ddaf716 and 3e5efcc.

Files selected for processing (4)

• src/puffin/src/file_format/reader.rs (2 hunks)
• src/puffin/src/puffin_manager/cache_manager/moka_cache_manager.rs (2 hunks)
• src/puffin/src/puffin_manager/cached_puffin_manager/writer.rs (5 hunks)
• src/puffin/src/puffin_manager/tests.rs (1 hunks)

Files skipped from review due to trivial changes (1)

• src/puffin/src/file_format/reader.rs

Files skipped from review as they are similar to previous changes (2)

• src/puffin/src/puffin_manager/cache_manager/moka_cache_manager.rs
• src/puffin/src/puffin_manager/cached_puffin_manager/writer.rs

[coderabbitai]

Actionable comments posted: 19

Outside diff range and nitpick comments (3)

src/puffin/src/puffin_manager/fs_puffin_manager/writer.rs (1)

Line range hint 67-84: Handle potential errors in put_blob.

The put_blob function should handle potential errors from handle_compress instead of unwrapping.

let written_bytes = self
    .handle_compress(key.to_string(), raw_data, options.compression)
    .await?;

src/puffin/src/puffin_manager/fs_puffin_manager/reader.rs (2)

Line range hint 101-105: Consider adding error handling for init_blob_to_cache.

The function init_blob_to_cache should handle potential errors more gracefully and provide meaningful error messages.

fn init_blob_to_cache(
    puffin_file_name: String,
    key: String,
    mut writer: BoxWriter,
    accessor: PuffinFileAccessorRef<AR, AW>,
) -> BoxFuture<'static, Result<u64>> {
    Box::pin(async move {
        let reader = accessor.reader(&puffin_file_name).await?;
        let mut file = PuffinFileReader::new(reader);

        let metadata = file.metadata().await?;
        let blob_metadata = metadata
            .blobs
            .iter()
            .find(|m| m.blob_type == key.as_str())
            .context(BlobNotFoundSnafu { blob: key })?;
        let reader = file.blob_reader(blob_metadata)?;

        let compression = blob_metadata.compression_codec;
        let size = Self::handle_decompress(reader, &mut writer, compression).await?;

        Ok(size)
    })
}

---

Line range hint 137-146: Consider adding error handling for init_dir_to_cache.

The function init_dir_to_cache should handle potential errors more gracefully and provide meaningful error messages.

fn init_dir_to_cache(
    puffin_file_name: String,
    key: String,
    writer_provider: DirWriterProviderRef,
    accessor: PuffinFileAccessorRef<AR, AW>,
) -> BoxFuture<'static, Result<u64>> {
    Box::pin(async move {
        let reader = accessor.reader(&puffin_file_name).await?;
        let mut file = PuffinFileReader::new(reader);

        let puffin_metadata = file.metadata().await?;
        let blob_metadata = puffin_metadata
            .blobs
            .iter()
            .find(|m| m.blob_type == key.as_str())
            .context(BlobNotFoundSnafu { blob: key })?;

        let mut reader = file.blob_reader(blob_metadata)?;
        let mut buf = vec![];
        reader.read_to_end(&mut buf).await.context(ReadSnafu)?;
        let dir_meta: DirMetadata =
            serde_json::from_slice(buf.as_slice()).context(DeserializeJsonSnafu)?;

        let mut size = 0;
        for file_meta in dir_meta.files {
            let blob_meta = puffin_metadata.blobs.get(file_meta.blob_index).context(
                BlobIndexOutOfBoundSnafu {
                    index: file_meta.blob_index,
                    max_index: puffin_metadata.blobs.len(),
                },
            )?;
            ensure!(
                blob_meta.blob_type == file_meta.key,
                FileKeyNotMatchSnafu {
                    expected: file_meta.key,
                    actual: &blob_meta.blob_type,
                }
            );

            let reader = file.blob_reader(blob_meta)?;
            let writer = writer_provider.writer(&file_meta.relative_path).await?;

            let compression = blob_meta.compression_codec;
            size += Self::handle_decompress(reader, writer, compression).await?;
        }

        Ok(size)
    })
}

Review details

Configuration used: CodeRabbit UI
Review profile: CHILL

Commits

Files that changed from the base of the PR and between 3e5efcc and 8b15588.

Files selected for processing (5)

• src/puffin/src/puffin_manager.rs (2 hunks)
• src/puffin/src/puffin_manager/fs_puffin_manager.rs (1 hunks)
• src/puffin/src/puffin_manager/fs_puffin_manager/reader.rs (4 hunks)
• src/puffin/src/puffin_manager/fs_puffin_manager/writer.rs (7 hunks)
• src/puffin/src/puffin_manager/tests.rs (1 hunks)

Files skipped from review as they are similar to previous changes (1)

• src/puffin/src/puffin_manager.rs

Additional comments not posted (5)

src/puffin/src/puffin_manager/fs_puffin_manager.rs (1)

29-36: Ensure fields are documented.

The fields cache_manager and puffin_file_accessor should have more detailed documentation to explain their roles within the FsPuffinManager.

pub struct FsPuffinManager<B, D, AR, AW> {
    /// The cache manager for managing cached operations.
    cache_manager: CacheManagerRef<B, D>,

    /// The puffin file accessor for file operations.
    puffin_file_accessor: PuffinFileAccessorRef<AR, AW>,
}

src/puffin/src/puffin_manager/fs_puffin_manager/writer.rs (2)

Line range hint 36-50: Ensure fields are documented.

The fields puffin_file_name, cache_manager, puffin_file_writer, and blob_keys should have more detailed documentation to explain their roles within the FsPuffinWriter.

pub struct FsPuffinWriter<B, D, W> {
    /// The name of the puffin file.
    puffin_file_name: String,

    /// The cache manager for managing cached operations.
    cache_manager: CacheManagerRef<B, D>,

    /// The underlying Puffin file writer.
    puffin_file_writer: PuffinFileWriter<W>,

    /// Set of written blob keys to avoid duplicates.
    blob_keys: HashSet<String>,
}

---

127-131: Ensure cross-platform compatibility.

The code correctly handles path separators for different platforms. Ensure that this logic is thoroughly tested on both Windows and Unix-based systems.

let unified_rel_path = if cfg!(windows) {
    relative_path.to_string_lossy().replace('\\', "/")
} else {
    relative_path.to_string_lossy().to_string()
};

src/puffin/src/puffin_manager/fs_puffin_manager/reader.rs (1)

Line range hint 33-44: Ensure fields are documented.

The fields puffin_file_name, cache_manager, and puffin_file_accessor should have more detailed documentation to explain their roles within the FsPuffinReader.

pub struct FsPuffinReader<B, G, AR, AW> {
    /// The name of the puffin file.
    puffin_file_name: String,

    /// The cache manager for managing cached operations.
    cache_manager: CacheManagerRef<B, G>,

    /// The puffin file accessor for file operations.
    puffin_file_accessor: PuffinFileAccessorRef<AR, AW>,
}

src/puffin/src/puffin_manager/tests.rs (1)

34-40: Handle potential errors from create_temp_dir and MokaCacheManager::new.

The function should handle potential errors instead of unwrapping them.

let cache_dir = create_temp_dir(prefix).expect("Failed to create temp directory");

Arc::new(MokaCacheManager::new(path, cache_size).await.expect("Failed to create MokaCacheManager")),

Likely invalid or redundant comment.

[evenyag]

https://github.com/GreptimeTeam/greptimedb/actions/runs/9755006232/job/26922900846?pr=4240

Is this related to #4243? @WenyXu

thread '<unnamed>' panicked at tests-fuzz/targets/fuzz_insert_logical_table.rs:307:35:
fuzz test must be succeed: 0: Failed to execute query: CREATE TABLE `CupidiTaTE`(
ts TIMESTAMP(3) TIME INDEX,
val DOUBLE,
`eT` STRING,
`VoLUPTaTe` STRING,
`ASPerNatur` STRING,
`ExPLiCabo` STRING,
PRIMARY KEY(`VoLUPTaTe`, `ASPerNatur`, `eT`, `ExPLiCabo`)
)
ENGINE=metric with ("on_physical_table" = "doLORuM");, at tests-fuzz/targets/fuzz_insert_logical_table.rs:256:14
1: Database(MySqlDatabaseError { code: Some("HY000"), number: 1815, message: "Region `4647154614272(1082, 0)` already exists" })

[WenyXu]

https://github.com/GreptimeTeam/greptimedb/actions/runs/9755006232/job/26922900846?pr=4240

Is this related to #4243? @WenyXu

thread '<unnamed>' panicked at tests-fuzz/targets/fuzz_insert_logical_table.rs:307:35:
fuzz test must be succeed: 0: Failed to execute query: CREATE TABLE `CupidiTaTE`(
ts TIMESTAMP(3) TIME INDEX,
val DOUBLE,
`eT` STRING,
`VoLUPTaTe` STRING,
`ASPerNatur` STRING,
`ExPLiCabo` STRING,
PRIMARY KEY(`VoLUPTaTe`, `ASPerNatur`, `eT`, `ExPLiCabo`)
)
ENGINE=metric with ("on_physical_table" = "doLORuM");, at tests-fuzz/targets/fuzz_insert_logical_table.rs:256:14
1: Database(MySqlDatabaseError { code: Some("HY000"), number: 1815, message: "Region `4647154614272(1082, 0)` already exists" })

No, it seems something else. Please trigger test again, and I’ll check the failed action ASAP.

[evenyag]

@coderabbitai pause

[coderabbitai]

Actions performed

Reviews paused.