Repository: risingwavelabs/risingwave
File: src/meta/src/controller/fragment.rs (Rust)

Code before:
    ) -> MetaResult<FragmentRenderMap> {
        let adaptive_parallelism_strategy = {
            let system_params_reader = self.env.system_params_reader().await;
            system_params_reader.adaptive_parallelism_strategy()
        };

        let inner = self.inner.read().await;
        let txn = inner.db.begin().await?;

        let database_fragment_infos = load_fragment_info(
            &txn,
            self.env.actor_id_generator(),
            database_id,
            worker_nodes,
            adaptive_parallelism_strategy,
        )
        .await?;

        tracing::trace!(?database_fragment_infos, "reload all actors");

        Ok(database_fragment_infos)
    }

    #[await_tree::instrument]
    pub async fn fill_snapshot_backfill_epoch(
        &self,
        fragment_ids: impl Iterator<Item = FragmentId>,
        snapshot_backfill_info: Option<&SnapshotBackfillInfo>,
        cross_db_snapshot_backfill_info: &SnapshotBackfillInfo,
    ) -> MetaResult<()> {
        let inner = self.inner.write().await;
        let txn = inner.db.begin().await?;
        for fragment_id in fragment_ids {
            let fragment = FragmentModel::find_by_id(fragment_id)
                .one(&txn)
                .await?
                .context(format!("fragment {} not found", fragment_id))?;
            let mut node = fragment.stream_node.to_protobuf();
            if crate::stream::fill_snapshot_backfill_epoch(
                &mut node,
                snapshot_backfill_info,
                cross_db_snapshot_backfill_info,
            )? {
                let node = StreamNode::from(&node);
                FragmentModel::update(fragment::ActiveModel {
                    fragment_id: Set(fragment_id),
                    stream_node: Set(node),
                    ..Default::default()
                })
                .exec(&txn)
                .await?;

Review comment (suggestion):
Using `.unwrap()` on `NonZeroUsize::new()` can cause a panic if `compute_node_parallelism()` returns 0. While there's validation at compute node startup, there's no guarantee this value is always non-zero at runtime. Consider handling this more gracefully with proper error handling or validation before creating the WorkerInfo, such as filtering out workers with zero parallelism or returning a descriptive error.
```suggestion
            .filter_map(|worker| {
                match NonZeroUsize::new(worker.compute_node_parallelism()) {
                    Some(parallelism) => Some((
                        worker.id,
                        WorkerInfo {
                            parallelism,
                            resource_group: worker.resource_group(),
                        },
                    )),
                    None => {
                        tracing::warn!(
                            worker_id = worker.id,
                            "skipping worker with zero compute node parallelism"
                        );
                        None
                    }
                }
```

Code after:
        database_id: Option<DatabaseId>,
        worker_nodes: &ActiveStreamingWorkerNodes,
    ) -> MetaResult<FragmentRenderMap> {
        let loaded = self.load_fragment_context(database_id).await?;

        if loaded.is_empty() {
            return Ok(HashMap::new());
        }

        let adaptive_parallelism_strategy = {
            let system_params_reader = self.env.system_params_reader().await;
            system_params_reader.adaptive_parallelism_strategy()
        };

        let available_workers: BTreeMap<_, _> = worker_nodes
            .current()
            .values()
            .filter(|worker| worker.is_streaming_schedulable())
            .map(|worker| {
                (
                    worker.id,
                    WorkerInfo {
                        parallelism: NonZeroUsize::new(worker.compute_node_parallelism()).unwrap(),
                        resource_group: worker.resource_group(),
                    },
                )
            })
            .collect();

        let RenderedGraph { fragments, .. } = render_actor_assignments(
            self.env.actor_id_generator(),
            &available_workers,
            adaptive_parallelism_strategy,
            &loaded,
        )?;

        tracing::trace!(?fragments, "reload all actors");

        Ok(fragments)
    }

    /// Async load stage: collects all metadata required for rendering actor assignments.
    pub async fn load_fragment_context(
        &self,
        database_id: Option<DatabaseId>,
    ) -> MetaResult<LoadedFragmentContext> {
        let inner = self.inner.read().await;
        let txn = inner.db.begin().await?;

        let mut query = StreamingJob::find()
            .select_only()