Files that changed from the base of the PR and between 9964626 and e397079.

• src/config/src/config.rs (2 hunks)
• src/config/src/meta/sql.rs (1 hunks)
• src/config/src/utils/time.rs (1 hunks)
• src/infra/src/file_list/mod.rs (4 hunks)
• src/infra/src/file_list/mysql.rs (3 hunks)
• src/infra/src/file_list/postgres.rs (4 hunks)
• src/infra/src/file_list/sqlite.rs (3 hunks)
• src/job/compactor.rs (4 hunks)
• src/service/compact/merge.rs (9 hunks)
• src/service/compact/mod.rs (4 hunks)
• web/src/plugins/logs/SearchBar.vue (2 hunks)

src/infra/src/file_list/mod.rs (1)

118-129: Ensure that the new methods related to merge job management (add_job, get_pending_jobs, set_job_done, update_running_jobs, check_running_jobs, clean_done_jobs) handle exceptions and errors appropriately, especially in asynchronous contexts.

src/config/src/utils/time.rs (1)

37-40: The function now_micros is a useful addition for high-precision time requirements. Ensure that it is used consistently across the project where such precision is necessary.

src/infra/src/file_list/mysql.rs (2)

626-701: Optimize transaction handling for better performance.
[REFACTOR_SUGGESTion]
This method involves multiple database operations within a transaction, which could be optimized. Consider consolidating the database calls or using batch operations where possible to reduce the overhead of transaction management and improve performance.

---

741-751: Validate the logic for cleaning up jobs.

The clean_done_jobs method deletes jobs based on the updated_at timestamp and job status. Verify that this logic correctly identifies only the intended jobs to prevent accidental deletion of active jobs.

Verification successful

---

The gathered context provides a clear view of how the clean_done_jobs function is utilized. The run_clean_done_jobs function in src/job/compactor.rs runs in a loop, periodically invoking clean_done_jobs with a calculated updated_at timestamp, ensuring that only jobs older than a certain threshold are targeted. This logic appears sound and should prevent the accidental deletion of active jobs, as it relies on the updated_at timestamp and the job status.

• src/job/compactor.rs:

  • run_clean_done_jobs calculates an updated_at timestamp based on a configured wait time.
  • It calls clean_done_jobs with this timestamp in a loop, ensuring periodic cleanup.

• src/infra/src/file_list/mod.rs:

  • The clean_done_jobs function is defined and called through the CLIENT interface.

The logic for cleaning up jobs based on the updated_at timestamp and job status appears to be correctly implemented.

Scripts executed

The following scripts were executed for the analysis:

---

Script:

#!/bin/bash
# Check for any active jobs that might be incorrectly marked for deletion.
rg --type rust "clean_done_jobs"

Length of output: 887

---

---

Script:

#!/bin/bash
# Extract the implementation of run_clean_done_jobs from src/job/compactor.rs
ast-grep --lang rust --pattern 'async fn run_clean_done_jobs() -> Result<$_, $_> { $$$ }' 

# Extract the relevant parts of src/infra/src/file_list/mod.rs where clean_done_jobs is defined and called
rg -A 10 'async fn clean_done_jobs' src/infra/src/file_list/mod.rs

Length of output: 1526

src/infra/src/file_list/postgres.rs (4)

720-728: Optimize the set_job_done method by caching the database connection.
[PERFORMANCE]
The method set_job_done retrieves a new database connection on every call. This could be optimized by reusing a connection or using a connection pool. Consider modifying the method to accept a database connection as a parameter or use a global/static connection pool.

---

731-738: Review the necessity of frequent time updates in update_running_jobs.
[PERFORMANCE]
The method update_running_jobs updates the updated_at timestamp every time it's called. If this method is called frequently, it could lead to performance issues due to the constant updates. Consider if there's a way to reduce the frequency of these updates or batch them.

---

630-718: Ensure proper transaction handling in get_pending_jobs.

The method get_pending_jobs uses a transaction which is good for consistency. However, the rollback scenarios are scattered and might be missed in future modifications. Consider using a transaction management pattern or a context manager that ensures the rollback in all error scenarios.

Verification successful

---

The extracted code for the get_pending_jobs function reveals that transaction rollbacks are indeed handled in various error scenarios. Each Err(e) block includes a tx.rollback().await call, ensuring that transactions are rolled back upon encountering errors. Here are the key points:

• Rollback is handled after failing to acquire the advisory lock.
• Rollback is handled after failing to fetch pending jobs.
• Rollback is handled if there are no pending jobs to update.
• Rollback is handled after failing to update job statuses.
• Rollback is handled after failing to fetch jobs by IDs.

Given this thorough handling of transaction rollbacks, the initial review comment's concern about scattered rollbacks is mitigated.

Scripts executed

The following scripts were executed for the analysis:

---

Script:

#!/bin/bash
# Verify that all error paths in get_pending_jobs handle transactions correctly
rg --type rust "tx.rollback().await" "src/infra/src/file_list/postgres.rs"

Length of output: 74

---

---

Script:

#!/bin/bash
# Extract the `get_pending_jobs` function from `postgres.rs` for manual review
ast-grep --lang rust --pattern $'impl $_ {
    $$$
    async fn get_pending_jobs(&self, $_, $_) -> $_ {
        $$$
    }
    $$$
}' "src/infra/src/file_list/postgres.rs"

Length of output: 169

---

---

Script:

#!/bin/bash
# Extract lines 630-718 from `postgres.rs` for manual review
sed -n '630,718p' src/infra/src/file_list/postgres.rs

Length of output: 3356

---

938-953: Ensure the new file_list_jobs table is indexed properly.

The creation of the file_list_jobs table includes setting up indexes. It's important to verify that these indexes are appropriate for the queries run against this table to ensure optimal performance.

Verification successful

---

The file_list_jobs table is indexed with the file_list_jobs_stream_status_idx index on the status and stream columns in the src/infra/src/file_list/postgres.rs file. This confirms that an index is created for the table.

• Index Creation: src/infra/src/file_list/postgres.rs: "CREATE INDEX IF NOT EXISTS file_list_jobs_stream_status_idx on file_list_jobs (status, stream);"

Scripts executed

The following scripts were executed for the analysis:

---

Script:

#!/bin/bash
# Verify that indexes on file_list_jobs are used in queries
rg --type rust "FROM file_list_jobs" "src/infra/src/file_list/postgres.rs" | rg "INDEX"

Length of output: 85

---

---

Script:

#!/bin/bash
# Search for index creation commands related to file_list_jobs table
rg --type sql "CREATE INDEX" | rg "file_list_jobs"

Length of output: 199

---

---

Script:

#!/bin/bash
# Search for index creation commands related to file_list_jobs table without file type filter
rg "CREATE INDEX" | rg "file_list_jobs"

Length of output: 461

src/infra/src/file_list/sqlite.rs (4)

718-728: Ensure proper error handling and logging in set_job_done.

Currently, the function does not log any information on success or failure, which could hinder debugging. Adding logging could improve observability.

sqlx::query(r#"UPDATE file_list_jobs SET status = $1, updated_at = $2 WHERE id = $3;"#)
    .bind(super::FileListJobStatus::Done)
    .bind(config::utils::time::now_micros())
    .bind(id)
    .execute(&*client)
    .await
    .map_err(|e| {
        log::error!("Failed to set job done: {}", e);
        e.into()
    })?;
log::info!("Job {} marked as done", id);
Ok(())

---

945-961: Ensure the new table creation adheres to schema requirements.

The creation of the file_list_jobs table should be verified to ensure it meets all schema and business logic requirements. This includes checking constraints, types, and indexes.

sqlx::query(r#"CREATE TABLE IF NOT EXISTS file_list_jobs (...) "#).execute(&*client).await?;
log::info("Table file_list_jobs created successfully.");
Ok(())

---

1020-1026: Check the uniqueness and indexing strategy for file_list_jobs.

The unique index on stream and offsets might cause issues if the same stream can have jobs with identical offsets. Verify that this indexing strategy aligns with the expected usage patterns.

Verification successful

---

The search results provide extensive information on the usage and constraints related to the file_list_jobs table across different database systems (SQLite, MySQL, Postgres). Key observations include:

• The file_list_jobs table schema consistently includes stream and offsets as non-nullable fields.
• There are unique indexes created on the combination of stream and offsets in all database systems.
• Multiple instances of INSERT INTO file_list_jobs statements suggest that stream and offsets are frequently used together, and there are mechanisms to handle unique constraint violations.
• Queries and updates frequently interact with the file_list_jobs table, indicating its active role in job management.

Given these observations, the unique index on stream and offsets seems intentional and aligns with the expected usage patterns. There are no immediate indications of issues with this indexing strategy.

• src/infra/src/file_list/sqlite.rs
• src/infra/src/file_list/mysql.rs
• src/infra/src/file_list/postgres.rs

Scripts executed

The following scripts were executed for the analysis:

---

Script:

#!/bin/bash
# Check for potential issues with unique index on stream and offsets
echo "Add verification logic here"

Length of output: 64

---

---

Script:

#!/bin/bash
# Search for the definition of the file_list_jobs table and any associated constraints or indexes
rg "CREATE TABLE.*file_list_jobs" -A 20

Length of output: 3761

---

---

Script:

#!/bin/bash
# Search for comments or documentation related to the usage patterns of stream and offsets
rg -i "stream.*offsets" -A 5

# Search for queries or code that interact with the file_list_jobs table
rg -i "file_list_jobs" -A 10

Length of output: 35474

---

759-771: Validate the logic for cleaning done jobs.

The function deletes jobs based on status and update time. It's crucial to ensure that this does not accidentally delete jobs that are still needed. A verification step or additional logging could be beneficial.

Verification successful

---

The FileListJobStatus enum is defined with three variants: Pending, Running, and Done. The clean_done_jobs function deletes jobs with the status Done that were updated before a specified date. This logic appears to be sound, as it targets only jobs that are marked as Done and have not been updated recently.

• The FileListJobStatus enum definition is located in src/infra/src/file_list/mod.rs:

  pub enum FileListJobStatus {
      #[default]
      Pending,
      Running,
      Done,
  }

Given this context, the function's logic should be verified to ensure it does not accidentally delete essential jobs. However, based on the current information, the implementation seems appropriate for its intended purpose.

Scripts executed

The following scripts were executed for the analysis:

---

Script:

#!/bin/bash
# Verify that no essential jobs are deleted unintentionally
# This script should check the job statuses and their last update times to ensure correctness.
echo "Add verification logic here"

Length of output: 64

---

---

Script:

#!/bin/bash
# Find the definition and usage of FileListJobStatus::Done
ast-grep --lang rust --pattern 'enum FileListJobStatus { Done, $$$ }'

Length of output: 71

---

---

Script:

#!/bin/bash
# Find the definition of FileListJobStatus enum
ast-grep --lang rust --pattern 'enum FileListJobStatus { $$$ }'

Length of output: 65

---

---

Script:

#!/bin/bash
# Search for the term FileListJobStatus in the codebase
rg 'FileListJobStatus'

Length of output: 1858

---

---

Script:

#!/bin/bash
# View the definition of FileListJobStatus in src/infra/src/file_list/mod.rs
cat src/infra/src/file_list/mod.rs | grep -A 10 'pub enum FileListJobStatus'

Length of output: 160

src/config/src/meta/sql.rs (1)

265-265: Updated error message for Group conversion improves clarity and specificity.

src/config/src/config.rs (1)

873-890: Ensure the new configuration fields for the compactor module are well-documented and the default values are appropriate.

The new settings batch_size, job_run_timeout, and job_clean_wait_time are crucial for managing job operations efficiently in the compactor module. The default values seem reasonable, but you might want to ensure they align with expected operational parameters in production environments.

Files that changed from the base of the PR and between e397079 and 65974e1.

• src/service/compact/merge.rs (9 hunks)

Files that changed from the base of the PR and between 65974e1 and 4ade497.

• src/service/traces/otlp_http.rs (1 hunks)

src/service/traces/otlp_http.rs (1)

46-54: Improved error handling by using a match statement to return a detailed BadRequest response if decoding fails.

Files that changed from the base of the PR and between 4ade497 and d88a905.

• src/config/src/config.rs (5 hunks)
• src/config/src/utils/inverted_index.rs (1 hunks)
• src/config/src/utils/mod.rs (1 hunks)
• src/ingester/src/writer.rs (1 hunks)
• src/job/files/parquet.rs (7 hunks)
• src/service/search/cluster/mod.rs (2 hunks)
• src/wal/src/lib.rs (1 hunks)
• web/src/plugins/logs/SearchBar.vue (2 hunks)

• src/config/src/config.rs
• web/src/plugins/logs/SearchBar.vue

src/config/src/utils/mod.rs (1)

23-23: The addition of the inverted_index module aligns with the enhancements described in the PR summary.

src/wal/src/lib.rs (1)

28-29: Standardizing the file type identifier with constants improves code maintainability and readability.

src/ingester/src/writer.rs (1)

287-287: Using FILE_TYPE_IDENTIFIER_LEN improves the clarity of the condition. Ensure the complex logical condition accurately implements the intended checks.

src/job/files/parquet.rs (1)

Line range hint 767-818: Ensure consistency and reliability in generate_index_on_compactor.

• Verify that the function consistently handles edge cases and large datasets.
• Consider adding more comprehensive tests to cover various scenarios and data sizes.

src/service/search/cluster/mod.rs (2)

30-30: The import of split_token is correctly placed and used effectively in the search logic.

---

121-121: Usage of split_token function is correctly implemented for processing search terms in a flexible and efficient manner.

Files that changed from the base of the PR and between d88a905 and 071586d.

• src/config/src/meta/stream.rs (1 hunks)
• src/config/src/utils/time.rs (1 hunks)
• src/infra/src/file_list/mod.rs (4 hunks)
• src/infra/src/file_list/mysql.rs (3 hunks)
• src/infra/src/file_list/postgres.rs (4 hunks)
• src/infra/src/file_list/sqlite.rs (3 hunks)
• src/job/compactor.rs (2 hunks)
• src/job/files/idx.rs (1 hunks)
• src/service/compact/merge.rs (8 hunks)
• src/service/compact/mod.rs (5 hunks)

• src/config/src/utils/time.rs
• src/infra/src/file_list/mod.rs
• src/infra/src/file_list/sqlite.rs

src/job/files/idx.rs (1)

81-81: Improved log message format enhances clarity and traceability.

src/service/compact/mod.rs (3)

44-88: Refactored run_retention function improves handling based on stream-specific retention settings and enhances error logging with detailed stream information.

---

Line range hint 138-202: Updated run_generate_job function now includes checks for stream-specific settings and detailed error logging, improving reliability and maintainability.

---

211-343: The run_merge function has been enhanced to handle jobs based on stream settings and partition levels more efficiently, with improved error logging.

src/config/src/meta/stream.rs (1)

34-42: Adding ALL_STREAM_TYPES as a constant array is a good practice for managing multiple stream types centrally. This allows for easier maintenance and scalability.

src/infra/src/file_list/mysql.rs (2)

934-949: Creation of file_list_jobs table is implemented correctly.

The SQL statement for creating the file_list_jobs table is well-formed and includes all necessary fields.

---

1010-1016: Ensure the uniqueness and efficiency of the new indexes on file_list_jobs.

The creation of unique and non-unique indexes on the file_list_jobs table is appropriate for optimizing queries and ensuring data integrity.

src/infra/src/file_list/postgres.rs (3)

630-718: Review the transaction handling in get_pending_jobs to ensure atomicity and error recovery.

async fn get_pending_jobs(&self, node: &str, limit: i64) -> Result<Vec<super::MergeJobRecord>> {
    let pool = CLIENT.clone();
    let mut tx = pool.begin().await?;
    let lock_key = "file_list_jobs:get_pending_jobs";
    let lock_id = config::utils::hash::gxhash::new().sum64(lock_key);
    let lock_id = if lock_id > i64::MAX as u64 {
        (lock_id >> 1) as i64
    } else {
        lock_id as i64
    };
    let lock_sql = format!("SELECT pg_advisory_xact_lock({lock_id})");
    sqlx::query(&lock_sql).execute(&mut *tx).await?;
    
    let sql = r#"
        SELECT stream, max(id) as id, COUNT(*)::BIGINT AS num
        FROM file_list_jobs 
        WHERE status = $1 
        GROUP BY stream 
        ORDER BY num DESC 
        LIMIT $2;
    "#;
    let pending_jobs = sqlx::query_as::<_, super::MergeJobPendingRecord>(sql)
        .bind(super::FileListJobStatus::Pending)
        .bind(limit)
        .fetch_all(&mut *tx)
        .await?;
    
    let ids = pending_jobs.iter().map(|r| r.id.to_string()).collect::<Vec<_>>();
    if ids.is_empty() {
        tx.rollback().await?;
        return Ok(Vec::new());
    }
    
    let update_sql = format!(
        "UPDATE file_list_jobs SET status = $1, node = $2, updated_at = $3 WHERE id IN ({});",
        ids.join(",")
    );
    sqlx::query(&update_sql)
        .bind(super::FileListJobStatus::Running)
        .bind(node)
        .bind(config::utils::time::now_micros())
        .execute(&mut *tx)
        .await?;
    
    let select_sql = format!(
        "SELECT * FROM file_list_jobs WHERE id IN ({});",
        ids.join(",")
    );
    let jobs = sqlx::query_as::<_, super::MergeJobRecord>(&select_sql)
        .fetch_all(&mut *tx)
        .await?;
    
    tx.commit().await?;
    Ok(jobs)
}

Ensure that the transaction is properly rolled back on errors and committed only when all operations are successful. This script will verify the transaction management logic.

---

953-968: Ensure the new table file_list_jobs is appropriately indexed for performance.

The creation of the file_list_jobs table with necessary fields and types is correctly implemented. The fields are well-defined, and the SQL syntax is correct.

---

1029-1035: Check the uniqueness and indexing strategy on the file_list_jobs table.

The unique index on stream and offsets ensures that there are no duplicate job entries for the same stream and offset, which is crucial for the integrity of job processing. The status index will aid in efficient querying of jobs based on their current state.