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

Code before:
            session.check_privilege_for_drop_alter(schema_name, &**table)?;
            if table.table_type != TableType::Table || 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,

Review comment (suggestion):
The validation logic for sink backfill rate limit doesn't verify whether the sink actually supports backfill. While line 114 checks that the sink is not a "sink into table" for SINK_RATE_LIMIT, there's no corresponding validation for BACKFILL_RATE_LIMIT to ensure the sink has a backfill process. This could lead to accepting rate limit changes for sinks that don't have backfill operations, which may cause confusion or runtime errors.
```suggestion
                reader.get_any_sink_by_name(db_name, schema_path, &real_table_name)?;
            // Backfill rate limit is only applicable to streaming sinks, not sink-into-table.
            if sink.target_table.is_some() {
                return Err(ErrorCode::InvalidInputSyntax(format!(
                    "\"{table_name}\" is not a streaming sink and does not support backfill",
                ))
                .into());
            }
```

Code after:
        (PbThrottleTarget::Table, PbThrottleType::Backfill) => {
            let reader = session.env().catalog_reader().read_guard();
            let table =
                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
}