Repository: risingwavelabs/risingwave
File: src/frontend/src/handler/alter_streaming_rate_limit.rs (Rust)

Code before:
                    "\"{table_name}\" is not a CDC table",
                ))
                .into());
            }
            (StatementType::ALTER_TABLE, table.id.as_raw_id())
        }
        (PbThrottleTarget::Sink, PbThrottleType::Sink) => {
            let reader = session.env().catalog_reader().read_guard();
            let (sink, schema_name) =
                reader.get_any_sink_by_name(db_name, schema_path, &real_table_name)?;
            if sink.target_table.is_some() {
                bail!("ALTER SINK_RATE_LIMIT is not for sink into table")
            }
            session.check_privilege_for_drop_alter(schema_name, &**sink)?;
            (StatementType::ALTER_SINK, sink.id.as_raw_id())
        }
        (PbThrottleTarget::Sink, PbThrottleType::Backfill) => {
            let reader = session.env().catalog_reader().read_guard();
            let (sink, schema_name) =
                reader.get_any_sink_by_name(db_name, schema_path, &real_table_name)?;
            session.check_privilege_for_drop_alter(schema_name, &**sink)?;
            if sink.target_table.is_some() {
                return Err(ErrorCode::InvalidInputSyntax(format!(
                    "ALTER BACKFILL_RATE_LIMIT is unsupported for sink into table",
                ))
                .into());
            }
            (StatementType::ALTER_SINK, sink.id.as_raw_id())
        }
        _ => bail!(
            "Unsupported throttle target: {:?} and throttle type: {:?}",
            throttle_target,
            throttle_type
        ),
    };
    execute_with_long_running_notification(
        handle_alter_streaming_rate_limit_by_id(
            &session,
            throttle_target,
            throttle_type,
            id,
            rate_limit,
            stmt_type,
        ),
        &session,
        "ALTER STREAMING RATE LIMIT",
        LongRunningNotificationAction::SuggestRecover,
    )
    .await
}


Review comment (suggestion):
The error handling for checking sink_into_table is inconsistent between lines 114-116 and 125-130. Line 115 uses `bail!` macro while line 126 uses `return Err(ErrorCode::InvalidInputSyntax(...).into())`. Both patterns work but for maintainability, they should use the same approach.
```suggestion
                bail!(ErrorCode::InvalidInputSyntax(
                    "ALTER BACKFILL_RATE_LIMIT is unsupported for sink into table".into(),
                ));
```

Code after:
                check_table_mismatch(&reader, &session, db_name, schema_path, &real_table_name)?;
            if table.cdc_table_type.is_none() {
                return Err(ErrorCode::InvalidInputSyntax(format!(
                    "\"{table_name}\" is not a CDC table",
                ))
                .into());
            }
            (StatementType::ALTER_TABLE, table.id.as_raw_id())
        }
        (PbThrottleTarget::Sink, PbThrottleType::Sink) => {
            let reader = session.env().catalog_reader().read_guard();
            let (sink, schema_name) =
                reader.get_any_sink_by_name(db_name, schema_path, &real_table_name)?;
            if sink.target_table.is_some() {
                bail!("ALTER SINK_RATE_LIMIT is not for sink into table")
            }
            session.check_privilege_for_drop_alter(schema_name, &**sink)?;
            (StatementType::ALTER_SINK, sink.id.as_raw_id())
        }
        (PbThrottleTarget::Sink, PbThrottleType::Backfill) => {
            let reader = session.env().catalog_reader().read_guard();
            let (sink, schema_name) =
                reader.get_any_sink_by_name(db_name, schema_path, &real_table_name)?;
            session.check_privilege_for_drop_alter(schema_name, &**sink)?;
            (StatementType::ALTER_SINK, sink.id.as_raw_id())
        }
        _ => bail!(
            "Unsupported throttle target: {:?} and throttle type: {:?}",
            throttle_target,
            throttle_type
        ),
    };
    execute_with_long_running_notification(
        handle_alter_streaming_rate_limit_by_id(
            &session,
            throttle_target,
            throttle_type,
            id,
            rate_limit,
            stmt_type,
        ),
        &session,
        "ALTER STREAMING RATE LIMIT",
        LongRunningNotificationAction::SuggestRecover,
    )
    .await
}

pub async fn handle_alter_streaming_rate_limit_by_id(
    session: &SessionImpl,
    throttle_target: PbThrottleTarget,