Repository: risingwavelabs/risingwave
File: src/meta/src/rpc/ddl_controller.rs (Rust)

Code before:
        let (object_id, object_type) = match job_id {
            StreamingJobId::MaterializedView(id) => (id.as_raw_id() as _, ObjectType::Table),
            StreamingJobId::Sink(id) => (id as _, ObjectType::Sink),
            StreamingJobId::Table(_, id) => (id.as_raw_id() as _, ObjectType::Table),
            StreamingJobId::Index(idx) => (idx as _, ObjectType::Index),
        };

        let job_status = self
            .metadata_manager
            .catalog_controller
            .get_streaming_job_status(job_id.id())
            .await?;
        let version = match job_status {
            JobStatus::Initial => {
                unreachable!(
                    "Job with Initial status should not notify frontend and therefore should not arrive here"
                );
            }
            JobStatus::Creating => {
                let canceled_jobs = self
                    .stream_manager
                    .cancel_streaming_jobs(vec![job_id.id()])
                    .await?;
                if canceled_jobs.is_empty() {
                    tracing::warn!(job_id = %job_id.id(), "failed to cancel streaming job");
                }
                IGNORED_NOTIFICATION_VERSION
            }
            JobStatus::Created => {
                let version = self.drop_object(object_type, object_id, drop_mode).await?;
                #[cfg(not(madsim))]
                if let StreamingJobId::Sink(sink_id) = job_id {
                    // delete system table for exactly once iceberg sink
                    // todo(wcy-fdu): optimize the logic to be Iceberg unique.
                    let db = self.env.meta_store_ref().conn.clone();
                    clean_all_rows_by_sink_id(&db, sink_id).await?;
                }
                version
            }
        };

        Ok(version)
    }

    /// Resolve the parallelism of the stream job based on the given information.
    ///
    /// Returns error if user specifies a parallelism that cannot be satisfied.
    fn resolve_stream_parallelism(
        &self,
        specified: Option<NonZeroUsize>,
        max: NonZeroUsize,

Review comment (refactor):
After the refactoring, when `cancel_streaming_jobs()` returns an error, it will be propagated up via the `?` operator at line 1549, making the check at line 1550-1552 unreachable. Since errors are now propagated instead of silently returning an empty vector, this check should be removed to avoid dead code.
```suggestion

```

Code after:
        let (object_id, object_type) = match job_id {
            StreamingJobId::MaterializedView(id) => (id.as_raw_id() as _, ObjectType::Table),
            StreamingJobId::Sink(id) => (id as _, ObjectType::Sink),
            StreamingJobId::Table(_, id) => (id.as_raw_id() as _, ObjectType::Table),
            StreamingJobId::Index(idx) => (idx as _, ObjectType::Index),
        };

        let job_status = self
            .metadata_manager
            .catalog_controller
            .get_streaming_job_status(job_id.id())
            .await?;
        let version = match job_status {
            JobStatus::Initial => {
                unreachable!(
                    "Job with Initial status should not notify frontend and therefore should not arrive here"
                );
            }
            JobStatus::Creating => {
                self.stream_manager
                    .cancel_streaming_jobs(vec![job_id.id()])
                    .await?;
                IGNORED_NOTIFICATION_VERSION
            }
            JobStatus::Created => {
                let version = self.drop_object(object_type, object_id, drop_mode).await?;
                #[cfg(not(madsim))]
                if let StreamingJobId::Sink(sink_id) = job_id {
                    // delete system table for exactly once iceberg sink
                    // todo(wcy-fdu): optimize the logic to be Iceberg unique.
                    let db = self.env.meta_store_ref().conn.clone();
                    clean_all_rows_by_sink_id(&db, sink_id).await?;
                }
                version
            }
        };

        Ok(version)
    }

    /// Resolve the parallelism of the stream job based on the given information.
    ///
    /// Returns error if user specifies a parallelism that cannot be satisfied.
    fn resolve_stream_parallelism(
        &self,
        specified: Option<NonZeroUsize>,
        max: NonZeroUsize,
        cluster_info: &StreamingClusterInfo,
        resource_group: String,
    ) -> MetaResult<NonZeroUsize> {
        let available = cluster_info.parallelism(&resource_group);