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

Code before:
            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'",
                        SOURCE_REFRESH_INTERVAL_SEC_KEY, SOURCE_REFRESH_MODE_KEY
                    ))));
                }
                SourceRefreshMode {
                    refresh_mode: Some(RefreshMode::Streaming(SourceRefreshModeStreaming {})),
                }
            }
            "FULL_RECOMPUTE" => SourceRefreshMode {
                refresh_mode: Some(RefreshMode::FullRecompute(SourceRefreshModeFullRecompute {
                    refresh_interval_sec: source_refresh_interval_sec,

Review comment (bug):
The error message format is inconsistent between the parse error and validation error cases. When parsing fails, the error is shown in the `error` field (line 419), but when validation fails (value < 10), the `error` field shows `None` (line 428). This creates confusing error messages like "error: None" for validation failures.

Consider unifying the error message format:
```rust
if let Err(e) = maybe_pos_int {
    return Err(RwError::from(ErrorCode::InvalidParameterValue(format!(
        "`{}` must be a positive integer and larger than 10, but got: {}",
        SOURCE_REFRESH_INTERVAL_SEC_KEY, maybe_int
    ))));
}
if let Ok(some_int) = maybe_pos_int && some_int < 10 {
    return Err(RwError::from(ErrorCode::InvalidParameterValue(format!(
        "`{}` must be larger than 10, but got: {}",
        SOURCE_REFRESH_INTERVAL_SEC_KEY, some_int
    ))));
}
```
```suggestion
                return Err(expect_err(Some("value must be >= 10".to_string())));
```

Code after:
            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
                    ))));
                }
                SourceRefreshMode {
                    refresh_mode: Some(RefreshMode::Streaming(SourceRefreshModeStreaming {})),
                }
            }
            "FULL_RECOMPUTE" => SourceRefreshMode {
                refresh_mode: Some(RefreshMode::FullRecompute(SourceRefreshModeFullRecompute {
                    refresh_interval_sec: source_refresh_interval_sec,
                })),
            },