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

Code before:
        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 (suggestion):
The use of `.unwrap()` here is safe due to the conditional check on line 425, but it could be written more idiomatically to avoid the unwrap pattern. Consider:

```rust
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 10, but got: {}",
                SOURCE_REFRESH_INTERVAL_SEC_KEY, maybe_int
            )))
        })?;
        if some_int < 10 {
            return Err(RwError::from(ErrorCode::InvalidParameterValue(format!(
                "`{}` must be larger than 10, but got: {}",
                SOURCE_REFRESH_INTERVAL_SEC_KEY, some_int
            ))));
        }
        Some(some_int)
    } else {
        None
    }
};
```

This approach avoids the `.unwrap()` and provides clearer error messages.
```suggestion
            let some_int = maybe_int.parse::<i64>().map_err(|e| {
                RwError::from(ErrorCode::InvalidParameterValue(format!(
                    "`{}` must be a positive integer and larger than 10, but got: {}",
                    SOURCE_REFRESH_INTERVAL_SEC_KEY, maybe_int
                )))
            })?;
            if some_int < 10 {
                return Err(RwError::from(ErrorCode::InvalidParameterValue(format!(
                    "`{}` must be larger than 10, but got: {}",
                    SOURCE_REFRESH_INTERVAL_SEC_KEY, some_int
                ))));
            }
            Some(some_int)
```

Code after:
        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,
                })),
            },
            _ => {
                return Err(RwError::from(ErrorCode::InvalidParameterValue(format!(