Repository: risingwavelabs/risingwave
File: src/frontend/src/utils/with_options.rs (Rust)

Code before:
    for (key, secret_ref) in secret_refs {
        let (schema_name, secret_name) =
            Binder::resolve_schema_qualified_name(db_name, &secret_ref.secret_name)?;
        let secret_catalog = session.get_secret_by_name(schema_name, &secret_name)?;
        let ref_as = match secret_ref.ref_as {
            SecretRefAsType::Text => PbRefAsType::Text,
            SecretRefAsType::File => PbRefAsType::File,
        };
        let pb_secret_ref = PbSecretRef {
            secret_id: secret_catalog.id,
            ref_as: ref_as.into(),
        };
        resolved_secret_refs.insert(key.clone(), pb_secret_ref);
    }
    Ok(resolved_secret_refs)
}

pub(crate) fn resolve_source_refresh_mode_in_with_option(
    with_options: &mut WithOptions,
) -> RwResult<Option<SourceRefreshMode>> {
    let source_refresh_interval_sec = {
        if let Some(maybe_int) = with_options.remove(SOURCE_REFRESH_INTERVAL_SEC_KEY) {
            let maybe_pos_int = maybe_int.parse::<i64>();
            let expect_err = |e: Option<String>| {
                RwError::from(ErrorCode::InvalidParameterValue(format!(
                    "`{}` must be a positive integer and larger than 10, but got: {}: error: {:?}",
                    SOURCE_REFRESH_INTERVAL_SEC_KEY, maybe_int, e
                )))
            };
            if let Err(e) = maybe_pos_int {
                return Err(expect_err(Some(e.to_report_string())));
            }
            if let Ok(some_int) = maybe_pos_int
                && some_int < 10
            {
                return Err(expect_err(None));
            }
            Some(maybe_pos_int.unwrap())
        } else {
            None
        }
    };

    let source_refresh_mode = if let Some(source_refresh_mode_str) =
        with_options.remove(SOURCE_REFRESH_MODE_KEY)
    {
        match source_refresh_mode_str.to_uppercase().as_str() {
            "STREAMING" => {
                if source_refresh_interval_sec.is_some() {
                    return Err(RwError::from(ErrorCode::InvalidParameterValue(format!(
                        "`{}` is not allowed when `{}` is 'STREAMING'",

Review comment (suggestion):
The error message says "must be a positive integer and larger than 10" but the validation check `some_int < 10` actually allows the value 10. The error message is misleading. It should say either:
- "must be at least 10" (if 10 is the minimum allowed value), or  
- Change the check to `some_int <= 10` and keep the message as "larger than 10"

Based on the CI config setting `refresh_scheduler_interval_sec = 10` and the test case using `'11'`, it appears the intended minimum is 10 (inclusive), so the message should be "must be at least 10".
```suggestion
                    "`{}` must be a positive integer and at least 10, but got: {}: error: {:?}",
```

Code after:
    for (key, secret_ref) in secret_refs {
        let (schema_name, secret_name) =
            Binder::resolve_schema_qualified_name(db_name, &secret_ref.secret_name)?;
        let secret_catalog = session.get_secret_by_name(schema_name, &secret_name)?;
        let ref_as = match secret_ref.ref_as {
            SecretRefAsType::Text => PbRefAsType::Text,
            SecretRefAsType::File => PbRefAsType::File,
        };
        let pb_secret_ref = PbSecretRef {
            secret_id: secret_catalog.id,
            ref_as: ref_as.into(),
        };
        resolved_secret_refs.insert(key.clone(), pb_secret_ref);
    }
    Ok(resolved_secret_refs)
}

pub(crate) fn resolve_source_refresh_mode_in_with_option(
    with_options: &mut WithOptions,
) -> RwResult<Option<SourceRefreshMode>> {
    let source_refresh_interval_sec =
        {
            if let Some(maybe_int) = with_options.remove(SOURCE_REFRESH_INTERVAL_SEC_KEY) {
                let some_int = maybe_int.parse::<i64>().map_err(|e| {
                RwError::from(ErrorCode::InvalidParameterValue(format!(
                    "`{}` must be a positive integer and larger than 0, but got: {} (error: {})",
                    SOURCE_REFRESH_INTERVAL_SEC_KEY, maybe_int, e.as_report()
                )))
            })?;
                if some_int <= 0 {
                    return Err(RwError::from(ErrorCode::InvalidParameterValue(format!(
                        "`{}` must be larger than 0, but got: {}",
                        SOURCE_REFRESH_INTERVAL_SEC_KEY, some_int
                    ))));
                }
                Some(some_int)
            } else {
                None
            }
        };

    let source_refresh_mode = if let Some(source_refresh_mode_str) =
        with_options.remove(SOURCE_REFRESH_MODE_KEY)
    {
        match source_refresh_mode_str.to_uppercase().as_str() {
            "STREAMING" => {
                if source_refresh_interval_sec.is_some() {
                    return Err(RwError::from(ErrorCode::InvalidParameterValue(format!(
                        "`{}` is not allowed when `{}` is 'STREAMING'",
                        SOURCE_REFRESH_INTERVAL_SEC_KEY, SOURCE_REFRESH_MODE_KEY
                    ))));