Repository: risingwavelabs/risingwave
File: src/frontend/src/optimizer/plan_visitor/datafusion_plan_converter.rs (Rust)

Code before:
        &mut self,
        plan: &crate::optimizer::plan_node::LogicalTopN,
    ) -> Self::Result {
        let rw_input = plan.input();
        let df_input = self.visit(rw_input.clone())?;

        let TopNLimit::Simple(limit) = plan.limit_attr() else {
            bail_not_implemented!(
                "DataFusionPlanConverter: LogicalTopN with_ties is not supported"
            );
        };
        if !plan.group_key().is_empty() {
            bail_not_implemented!(
                "DataFusionPlanConverter: LogicalTopN with a non-empty group key is not supported. This may arise from DISTINCT ON, ORDER BY with DISTINCT, or correlated joins."
            );
        }
        let offset = plan.offset();

        let input_columns = InputColumns::new(df_input.schema().as_ref(), rw_input.schema());
        let sort_expr = plan
            .topn_order()
            .column_orders
            .iter()
            .map(|order| convert_column_order(order, &input_columns))
            .collect_vec();
        let fetch = offset.saturating_add(limit).min(usize::MAX as _) as usize;
        let sort = datafusion::logical_expr::Sort {
            expr: sort_expr,
            input: df_input,
            fetch: Some(fetch),
        };
        let mut result = Arc::new(LogicalPlan::Sort(sort));

        if offset > 0 {
            let limit = datafusion::logical_expr::Limit {
                skip: Some(Box::new(lit(offset))),
                fetch: Some(Box::new(lit(limit))),
                input: result,
            };
            result = Arc::new(LogicalPlan::Limit(limit));
        }

        Ok(result)
    }

    fn visit_logical_iceberg_scan(
        &mut self,
        plan: &crate::optimizer::plan_node::LogicalIcebergScan,
    ) -> Self::Result {
        // DataFusion requires unique table names for each scan, so we can't use actual table name here.
        let table_name = format!("iceberg_scan#{}", plan.base.id().0);

Review comment (suggestion):
The expression `usize::MAX as _` is ambiguous and relies on type inference. It should be explicitly cast to `u64` for clarity and to ensure the correct type is used in the `min` comparison. Change this to `usize::MAX as u64`.
```suggestion
        let fetch = offset.saturating_add(limit).min(usize::MAX as u64) as usize;
```

Code after:
        &mut self,
        plan: &crate::optimizer::plan_node::LogicalTopN,
    ) -> Self::Result {
        let rw_input = plan.input();
        let df_input = self.visit(rw_input.clone())?;

        let TopNLimit::Simple(limit) = plan.limit_attr() else {
            bail_not_implemented!(
                "DataFusionPlanConverter: LogicalTopN with_ties is not supported"
            );
        };
        if !plan.group_key().is_empty() {
            bail_not_implemented!(
                "DataFusionPlanConverter: LogicalTopN with a non-empty group key is not supported. This may arise from DISTINCT ON or correlated joins."
            );
        }
        let offset = plan.offset();

        let input_columns = InputColumns::new(df_input.schema().as_ref(), rw_input.schema());
        let sort_expr = plan
            .topn_order()
            .column_orders
            .iter()
            .map(|order| convert_column_order(order, &input_columns))
            .collect_vec();
        let fetch = offset.saturating_add(limit).min(usize::MAX as _) as usize;
        let sort = datafusion::logical_expr::Sort {
            expr: sort_expr,
            input: df_input,
            fetch: Some(fetch),
        };
        let mut result = Arc::new(LogicalPlan::Sort(sort));

        if offset > 0 {
            let limit = datafusion::logical_expr::Limit {
                skip: Some(Box::new(lit(offset))),
                fetch: Some(Box::new(lit(limit))),
                input: result,
            };
            result = Arc::new(LogicalPlan::Limit(limit));
        }

        Ok(result)
    }

    fn visit_logical_iceberg_scan(
        &mut self,
        plan: &crate::optimizer::plan_node::LogicalIcebergScan,
    ) -> Self::Result {
        // DataFusion requires unique table names for each scan, so we can't use actual table name here.
        let table_name = format!("iceberg_scan#{}", plan.base.id().0);