Repository: libsql/libsql
File: libsql-server/src/namespace/configurator/helpers.rs (Rust)

Code before:
            replicator.set_generation(gen);
        }
    }

    Ok((replicator, did_recover))
}

async fn run_periodic_compactions(logger: Arc<ReplicationLogger>) -> anyhow::Result<()> {
    // calling `ReplicationLogger::maybe_compact()` is cheap if the compaction does not actually
    // take place, so we can afford to poll it very often for simplicity
    let mut interval = tokio::time::interval(tokio::time::Duration::from_millis(1000));
    interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);

    loop {
        interval.tick().await;
        let handle = BLOCKING_RT.spawn_blocking(enclose! {(logger) move || {
            logger.maybe_compact()
        }});
        handle
            .await
            .expect("Compaction task crashed")
            .context("Compaction failed")?;
    }
}

fn tokenize_sql_keywords(text: &str) -> Vec<String> {
    let mut tokens = Vec::new();
    let mut chars = text.chars().peekable();
    let mut current_token = String::new();
    let mut in_string_literal = false;
    let mut string_delimiter = '\0';

    while let Some(ch) = chars.next() {
        match ch {
            '\'' | '"' => {
                if !in_string_literal {
                    in_string_literal = true;
                    string_delimiter = ch;
                } else if ch == string_delimiter {
                    in_string_literal = false;
                }
            }
            c if c.is_whitespace() || "(){}[];,".contains(c) => {
                if in_string_literal {
                    continue;
                }
                if !current_token.is_empty() {
                    tokens.push(current_token.to_uppercase());
                    current_token.clear();
                }
            }

Review comment (bug):
I think the risk here is that the tokenizer has some subtle  bugs. Did you consider the sqlite-parser we already use elsewhere in the server?

Code after:
            replicator.set_generation(gen);
        }
    }

    Ok((replicator, did_recover))
}

async fn run_periodic_compactions(logger: Arc<ReplicationLogger>) -> anyhow::Result<()> {
    // calling `ReplicationLogger::maybe_compact()` is cheap if the compaction does not actually
    // take place, so we can afford to poll it very often for simplicity
    let mut interval = tokio::time::interval(tokio::time::Duration::from_millis(1000));
    interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);

    loop {
        interval.tick().await;
        let handle = BLOCKING_RT.spawn_blocking(enclose! {(logger) move || {
            logger.maybe_compact()
        }});
        handle
            .await
            .expect("Compaction task crashed")
            .context("Compaction failed")?;
    }
}

async fn load_dump<S>(dump: S, conn: PrimaryConnection) -> crate::Result<(), LoadDumpError>
where
    S: Stream<Item = std::io::Result<Bytes>> + Unpin,
{
    let mut reader = tokio::io::BufReader::new(StreamReader::new(dump));
    let mut dump_content = String::new();
    reader
        .read_to_string(&mut dump_content)
        .await
        .map_err(|e| LoadDumpError::Internal(format!("Failed to read dump content: {}", e)))?;

    if dump_content.to_lowercase().contains("attach") {
        return Err(LoadDumpError::InvalidSqlInput(
            "attach statements are not allowed in dumps".to_string(),
        ));
    }

    let mut parser = Box::new(Parser::new(dump_content.as_bytes()));
    let mut skipped_wasm_table = false;
    let mut n_stmt = 0;

    loop {
        match parser.next() {
            Ok(Some(cmd)) => {
                n_stmt += 1;