src/infra/src/schema/history/sqlite.rs (1)

112-134: LGTM: Improved index naming and existence checking

The changes improve the code in two ways:

1. More descriptive index names enhance readability and maintainability.
2. The new logic for checking existing indices prevents redundant index creation, aligning with the PR objectives.

Minor suggestion: Consider extracting the index names into constants at the top of the file for easier maintenance and reuse.

const SCHEMA_HISTORY_ORG_IDX: &str = "schema_history_org_idx";
const SCHEMA_HISTORY_STREAM_IDX: &str = "schema_history_stream_idx";
const SCHEMA_HISTORY_STREAM_VERSION_IDX: &str = "schema_history_stream_version_idx";

src/infra/src/db/mod.rs (4)

245-257: LGTM: cache_indices_mysql function with suggestion.

The cache_indices_mysql function is well-implemented. It correctly queries the MySQL database for index information and handles potential errors.

Consider adding error logging when the query fails. This could help with debugging in production environments:

Err(e) => {
    log::error!("Failed to cache MySQL indices: {}", e);
    HashSet::new()
}

---

259-271: LGTM: cache_indices_pg function with suggestion.

The cache_indices_pg function is well-implemented. It correctly queries the PostgreSQL database for index information and handles potential errors.

Consider adding error logging when the query fails, similar to the suggestion for the MySQL function:

Err(e) => {
    log::error!("Failed to cache PostgreSQL indices: {}", e);
    HashSet::new()
}

---

273-285: LGTM: cache_indices_sqlite function with suggestion.

The cache_indices_sqlite function is well-implemented. It correctly queries the SQLite database for index information and handles potential errors.

Consider adding error logging when the query fails, similar to the suggestions for the MySQL and PostgreSQL functions:

Err(e) => {
    log::error!("Failed to cache SQLite indices: {}", e);
    HashSet::new()
}

---

239-285: Overall implementation aligns well with PR objectives.

The introduced changes effectively implement the caching mechanism for database indices across MySQL, PostgreSQL, and SQLite. This aligns well with the PR objective of checking for index existence before running create index commands.

Some points to consider:

1. The caching mechanism is implemented as described in the PR summary, operating during the initial startup of the application.
2. The implementation is consistent across all three database types, which is good for maintainability.
3. As mentioned in the PR summary, this approach may not capture indices created during initialization. Consider adding a comment in the code to highlight this limitation.

To address the potential issue of lock-ups in a multi-node scenario, consider implementing a distributed locking mechanism or exploring the use of IF NOT EXISTS in all relevant queries, as suggested in the PR summary.

src/infra/src/scheduler/postgres.rs (1)

85-95: Redundant index existence checks

The code checks for existing indices before creating them, and the SQL statements already include IF NOT EXISTS. This duplication might be unnecessary. Consider removing the code-level checks and relying solely on the SQL IF NOT EXISTS clause.

src/infra/src/scheduler/sqlite.rs (1)

95-95: Consider renaming variable idx to index_name for clarity

The variable idx represents an index name, not an index position. Renaming it to index_name or name will enhance readability and reduce potential confusion.

src/infra/src/db/postgres.rs (1)

Line range hint 614-629: Incorrect condition when creating index

In the add_start_dt_column function, the condition for creating the index meta_module_start_dt_idx is incorrect. Currently, it checks if the index exists and then proceeds to create it. The logic should be inverted to create the index only if it doesn't already exist.

Apply this diff to fix the condition:

-if indices.contains(&DBIndex {
+if !indices.contains(&DBIndex {
     name: "meta_module_start_dt_idx".into(),
     table: "meta".into(),
 }) {
     // Proceed to create the index
 }

This ensures that the index is only created if it does not already exist, aligning with the intended functionality.

src/infra/src/db/sqlite.rs (2)

Line range hint 743-751: Match specific error types instead of error message strings

In add_start_dt_column, the code checks for a duplicate column error by matching the error message string:

if !e.to_string().contains("duplicate column name")

Relying on error message strings is brittle and may fail if messages change or differ across database versions or locales. It's better to match on the specific error types or codes provided by sqlx or the database driver.

Apply this diff to improve error handling:

 if let Err(e) =
     sqlx::query(r#"ALTER TABLE meta ADD COLUMN start_dt INTEGER NOT NULL DEFAULT 0;"#)
         .execute(client)
         .await
 {
-    // Check if the error is about the duplicate column
-    if !e.to_string().contains("duplicate column name") {
-        // If the error is not about the duplicate column, return it
-        return Err(e.into());
+    if let sqlx::Error::Database(db_err) = &e {
+        if db_err.message().contains("duplicate column name") {
+            // Column already exists, ignore the error
+        } else {
+            // Return other database errors
+            return Err(e.into());
+        }
+    } else {
+        // Return non-database errors
+        return Err(e.into());
     }
 }

---

Line range hint 689-775: Ensure indices cache is updated after modifying indices

After creating or dropping indices in create_table and add_start_dt_column, the INDICES cache may become stale. To prevent inconsistencies, consider refreshing the INDICES cache after modifying indices.

Apply this diff to update the indices cache:

     }
 
+    // Refresh indices cache after modifying indices
+    *INDICES.get_mut().unwrap() = cache_indices_sqlite(client).await;
 
     Ok(())
 }

src/infra/src/db/mysql.rs (1)

593-593: Potential issue with indices caching mechanism.

Caching the indices at the beginning of the function may cause the INDICES cache to become outdated if indices are created or dropped afterward within the same runtime. Since the cache is in-memory and not refreshed, subsequent index operations may operate on stale data. Consider refreshing the INDICES cache after index creation or deletion to ensure it reflects the current state of the database.

Also applies to: 664-664

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

Line range hint 1254-1277: Incorrect logging condition in duplicate record deletion loop.

The condition if i / 1000 == 0 will always be true when i is less than 1000, causing excessive logging.

Update the condition to correctly log progress every 1000 iterations.

Apply this diff to fix the logging condition:

- if i / 1000 == 0 {
+ if i % 1000 == 0 {
     log::warn!("[POSTGRES] delete duplicate records: {}/{}", i, ret.len());
}

---

Line range hint 1254-1286: Inefficient deletion of duplicate records.

Deleting duplicates one by one can be time-consuming for large datasets.

Consider performing the deletion in a single SQL query using DELETE ... WHERE id IN (...) or using a CTE (Common Table Expression) to improve performance.

---

Line range hint 1249-1286: Lack of transactional integrity during index recreation.

The process of deleting duplicates and recreating the index is not enclosed in a transaction, which may lead to data inconsistency if an error occurs midway.

Wrap the deletion and index recreation within a transaction to ensure atomicity.

Apply this diff to include transaction handling:

+ let mut tx = pool.begin().await?;
  // delete duplicate records
  log::warn!("[POSTGRES] starting delete duplicate records");
  let ret = sqlx::query(
      r#"SELECT stream, date, file, min(id) as id FROM file_list GROUP BY stream, date, file HAVING COUNT(*) > 1;"#
  )
  .fetch_all(&mut *tx).await?;
  // ... existing deletion logic ...
  // create index again
  sqlx::query(unique_index_sql).execute(&mut *tx).await?;
+ tx.commit().await?;

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

1280-1283: Use specific error codes instead of parsing error messages

Relying on error message strings like "Duplicate key" is fragile and may not handle localization or changes in error messages. Consider checking for specific database error codes to handle duplicate key errors more robustly.

Configuration used: CodeRabbit UI
Review profile: CHILL

src/infra/src/scheduler/mysql.rs (1)

85-109: LGTM: Improved index creation implementation

The changes to the create_table_index function look good. The use of the create_index function improves efficiency by handling index existence checks internally. The indices being created are consistent with the previous implementation, and the unique constraint is correctly applied.

Consider extracting the table name "scheduled_jobs" into a constant to improve readability and maintainability. This would make it easier to update the table name if needed in the future.

const SCHEDULED_JOBS_TABLE: &str = "scheduled_jobs";

// Then use it in the function calls:
create_index(
    &pool,
    "scheduled_jobs_key_idx",
    SCHEDULED_JOBS_TABLE,
    false,
    &["module_key"],
)
.await?;
// ... and so on for the other create_index calls

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

606-617: Improved index management in add_start_dt_column function

The modifications to the add_start_dt_column function enhance consistency by using the new create_index and delete_index functions. This approach improves safety and maintainability.

Consider adding error handling for the create_index and delete_index calls to ensure the function fails gracefully if these operations encounter issues.

- create_index(
-     &pool,
-     "meta_module_start_dt_idx",
-     "meta",
-     true,
-     &["module", "key1", "key2", "start_dt"],
- )
- .await?;
- delete_index(&pool, "meta_module_key2_idx", "meta").await?;
+ if let Err(e) = create_index(
+     &pool,
+     "meta_module_start_dt_idx",
+     "meta",
+     true,
+     &["module", "key1", "key2", "start_dt"],
+ )
+ .await
+ {
+     log::error!("[POSTGRES] Error creating meta_module_start_dt_idx: {}", e);
+     return Err(e.into());
+ }
+ if let Err(e) = delete_index(&pool, "meta_module_key2_idx", "meta").await {
+     log::error!("[POSTGRES] Error deleting meta_module_key2_idx: {}", e);
+     return Err(e.into());
+ }

---

648-674: Well-implemented create_index function

The create_index function is efficiently implemented, using a cache to avoid unnecessary database queries and constructing a safe SQL query. The logging is appropriate and informative.

Consider adding more detailed error logging to provide context about which index creation failed:

- sqlx::query(&sql).execute(client).await?;
+ sqlx::query(&sql).execute(client).await.map_err(|e| {
+     log::error!("[POSTGRES] Failed to create index {}: {}", idx_name, e);
+     e
+ })?;

---

676-689: Well-implemented delete_index function

The delete_index function is efficiently implemented, using a cache to avoid unnecessary database queries and constructing a safe SQL query. The logging is appropriate and informative.

Similar to the create_index function, consider adding more detailed error logging:

- sqlx::query(&sql).execute(client).await?;
+ sqlx::query(&sql).execute(client).await.map_err(|e| {
+     log::error!("[POSTGRES] Failed to delete index {}: {}", idx_name, e);
+     e
+ })?;

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

54-67: Consider using a more descriptive variable name for the SQL query.

The variable name var_name for the SQL query string is not very descriptive. Consider using a more meaningful name like index_query or fetch_indices_sql to improve code readability.

-    let var_name = r#"SELECT INDEX_NAME,TABLE_NAME FROM information_schema.statistics;"#;
-    let sql = var_name;
+    let fetch_indices_sql = r#"SELECT INDEX_NAME,TABLE_NAME FROM information_schema.statistics;"#;
+    let sql = fetch_indices_sql;

---

Line range hint 659-690: LGTM: Improved index management in add_start_dt_column.

The modifications to use the new create_index and delete_index functions are well-implemented. However, there's an outdated comment that needs to be addressed.

Please update or remove the following comment as it's no longer relevant to the current implementation:

-        // Check for the specific MySQL error code for duplicate column
+        // Check for existing column to avoid duplicate column error

src/infra/src/db/sqlite.rs (5)

724-741: LGTM: Improved index creation using new create_index function

The changes to the create_table function look good. Using the new create_index function for creating indices improves code consistency and maintainability. The indices created ("meta_module_idx", "meta_module_key1_idx", and "meta_module_start_dt_idx") seem appropriate for the table structure.

Consider extracting the index names and table name into constants at the top of the file for easier maintenance and to avoid magic strings.

---

760-768: LGTM: Improved index management in add_start_dt_column

The changes to the add_start_dt_column function look good. Using the new create_index and delete_index functions improves code consistency. Creating a new index with the start_dt column and deleting the old one is a good practice for maintaining optimal database performance.

Consider adding a comment explaining why the "meta_module_key2_idx" index is being deleted, as it might not be immediately obvious to future maintainers.

---

797-823: LGTM: Well-implemented create_index function

The create_index function is well-implemented. It efficiently uses the INDICES cache to avoid unnecessary index creation attempts, handles both unique and non-unique indices, and includes appropriate logging.

Consider adding error logging in case the index creation fails. This would help with debugging in production environments.

 sqlx::query(&sql).execute(client).await?;
 log::info!("[SQLITE] index {} created successfully", idx_name);
+log::error!("[SQLITE] failed to create index {}: {}", idx_name, e);
 Ok(())

---

825-838: LGTM: Well-implemented delete_index function

The delete_index function is well-implemented. It efficiently uses the INDICES cache to avoid unnecessary index deletion attempts and includes appropriate logging.

Consider adding error logging in case the index deletion fails. Also, there's a minor typo in the success log message (missing space after "index").

 let sql = format!("DROP INDEX IF EXISTS {};", idx_name,);
-sqlx::query(&sql).execute(client).await?;
-log::info!("[SQLITE] index {}deleted successfully", idx_name);
+match sqlx::query(&sql).execute(client).await {
+    Ok(_) => log::info!("[SQLITE] index {} deleted successfully", idx_name),
+    Err(e) => log::error!("[SQLITE] failed to delete index {}: {}", idx_name, e),
+}
 Ok(())

---

Line range hint 100-838: Overall: Good improvements to index management, with room for minor enhancements

The changes introduce well-implemented functionality for managing database indices, including caching existing indices for performance optimization. The code is generally well-structured and follows good practices.

Consider creating a separate module or struct for index management, which could encapsulate the INDICES cache and provide methods for creating, deleting, and checking indices. This would further improve code organization and make it easier to maintain and potentially extend index-related functionality in the future.

Example structure:

mod index_manager {
    use once_cell::sync::Lazy;
    use sqlx::{Pool, Sqlite};
    use std::collections::HashSet;

    static INDICES: Lazy<RwLock<HashSet<DBIndex>>> = Lazy::new(|| RwLock::new(HashSet::new()));

    pub struct IndexManager;

    impl IndexManager {
        pub async fn init(client: &Pool<Sqlite>) {
            let indices = cache_indices(client).await;
            *INDICES.write().await = indices;
        }

        pub async fn create_index(/* ... */) -> Result<()> {
            // Implementation
        }

        pub async fn delete_index(/* ... */) -> Result<()> {
            // Implementation
        }

        pub async fn index_exists(/* ... */) -> bool {
            // Implementation
        }

        // Other index-related methods...
    }

    async fn cache_indices(client: &Pool<Sqlite>) -> HashSet<DBIndex> {
        // Implementation
    }
}

This structure would allow for better encapsulation of index-related functionality and make the code more modular and easier to maintain.

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

1247-1299: Special case handling for file_list table index

The implementation includes a special case for creating a unique index on the file_list table. It handles potential duplicate entries by deleting them before creating the index. This approach ensures data integrity but has some considerations:

1. The error message check on line 1257 might be too broad. Consider using a more specific error code check.
2. The duplicate record deletion process could potentially be optimized for large datasets.

Consider the following improvements:

1. Use a more specific error code check:

-if !e.to_string().contains("Duplicate entry") {
+if let Some(db_err) = e.as_database_error() {
+    if db_err.code() != Some("1062".into()) { // MySQL error code for duplicate entry
         return Err(e);
+    }
+} else {
+    return Err(e);
}

2. For large datasets, consider batching the delete operations:

let chunk_size = 1000;
for chunk in ret.chunks(chunk_size) {
    let mut query_builder: QueryBuilder<MySql> = QueryBuilder::new(
        "DELETE FROM file_list WHERE (stream, date, file, id) IN "
    );
    query_builder.push_tuples(chunk, |mut b, r| {
        b.push_bind(r.get::<String, _>("stream"))
         .push_bind(r.get::<String, _>("date"))
         .push_bind(r.get::<String, _>("file"))
         .push_bind(r.get::<i64, _>("id"));
    });
    query_builder.push(" AND id != VALUES(id)");
    query_builder.build().execute(&pool).await?;
    log::warn!("[MYSQL] Deleted {} duplicate records", chunk.len());
}

Configuration used: CodeRabbit UI
Review profile: CHILL

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

793-820: LGTM: Efficient index creation with room for minor improvement

The create_index function is well-implemented, efficiently checking for existing indices before creation and supporting unique indices. The use of proper logging is commendable.

Consider adding error logging in case the index creation fails. You can modify the last few lines like this:

-    sqlx::query(&sql).execute(&*client).await?;
-    log::info!("[SQLITE] index {} created successfully", idx_name);
-    Ok(())
+    match sqlx::query(&sql).execute(&*client).await {
+        Ok(_) => {
+            log::info!("[SQLITE] index {} created successfully", idx_name);
+            Ok(())
+        }
+        Err(e) => {
+            log::error!("[SQLITE] Failed to create index {}: {}", idx_name, e);
+            Err(e.into())
+        }
+    }

---

822-837: LGTM: Efficient index deletion with room for minor improvement

The delete_index function is well-implemented, efficiently checking for existing indices before deletion. The use of proper logging is commendable.

For consistency with the create_index function, consider adding error logging in case the index deletion fails. You can modify the last few lines like this:

-    sqlx::query(&sql).execute(&*client).await?;
-    log::info!("[SQLITE] index {}deleted successfully", idx_name);
-    Ok(())
+    match sqlx::query(&sql).execute(&*client).await {
+        Ok(_) => {
+            log::info!("[SQLITE] index {} deleted successfully", idx_name);
+            Ok(())
+        }
+        Err(e) => {
+            log::error!("[SQLITE] Failed to delete index {}: {}", idx_name, e);
+            Err(e.into())
+        }
+    }

---

693-694: LGTM: Improved data safety with backup creation

The changes to the add_start_dt_column method in the SqliteDb implementation improve data safety by creating a backup before modifying the table structure. The use of await ensures proper asynchronous execution.

Consider adding error handling to ensure that if the backup creation fails, the column addition is not performed. You can modify the code like this:

-        create_meta_backup().await?;
-        add_start_dt_column().await?;
+        create_meta_backup().await.map_err(|e| {
+            log::error!("[SQLITE] Failed to create backup before adding start_dt column: {}", e);
+            e
+        })?;
+        add_start_dt_column().await

This ensures that if the backup creation fails, an error is logged, and the column addition is not attempted.

---

Dropping the "meta_module_key2_idx" index may impact query performance.

Several queries in sqlite.rs, postgres.rs, and mysql.rs utilize the key2 field in their WHERE clauses, which may have previously benefited from the now-dropped index. This change could lead to slower query performance if these operations relied on the index for efficiency.

Analysis chain

Line range hint 737-762: Improved index management, but verify impact of dropped index

The changes to add_start_dt_column improve consistency by using the new create_index and delete_index functions. However, dropping the "meta_module_key2_idx" index might impact query performance if it was previously used.

Please verify that dropping the "meta_module_key2_idx" index doesn't negatively impact query performance. Run the following script to check for queries that might have used this index:

Scripts executed

The following scripts were executed for the analysis:

Script:

#!/bin/bash
# Description: Check for queries that might have used the dropped index
rg --type rust 'WHERE.*key2.*=' src/

Length of output: 2682

Configuration used: CodeRabbit UI
Review profile: CHILL