ronantakizawa/codereview-bench huggingface dataset:
Repository: risingwavelabs/risingwave
File: src/frontend/src/optimizer/plan_visitor/datafusion_execute_checker.rs (Rust)

Code before:
            supported: true,
            have_iceberg_scan: false,
        }
    }

    fn visit_logical_dedup(
        &mut self,
        _: &crate::optimizer::plan_node::LogicalDedup,
    ) -> Self::Result {
        CheckResult {
            supported: true,
            have_iceberg_scan: false,
        }
    }

    fn visit_logical_iceberg_scan(
        &mut self,
        _: &crate::optimizer::plan_node::LogicalIcebergScan,
    ) -> Self::Result {
        CheckResult {
            supported: true,
            have_iceberg_scan: true,
        }
    }
}

#[easy_ext::ext(DataFusionExecuteCheckerExt)]
pub impl LogicalPlanRef {
    /// Returns `true` if this plan is able to be executed by DataFusion.
    fn able_to_run_by_datafusion(&self) -> bool {
        let result = DataFusionExecuteChecker.visit(self.clone());
        result.supported && result.have_iceberg_scan
    }
}


Review comment (suggestion):
The visitor methods in DataFusionExecuteChecker are not visiting their child nodes. When you override a visitor method, the default behavior (which automatically visits all children via plan.inputs()) is bypassed. This means the checker only examines the root node and misses checking the entire plan subtree.

Each visit method should call self.visit() on the plan's inputs and merge the results with the current node's check. For example, visit_logical_agg should visit plan.input() and combine that result with the grouping sets check.

Code after:
        self.check_inputs(plan)
    }

    fn visit_logical_over_window(
        &mut self,
        plan: &crate::optimizer::plan_node::LogicalOverWindow,
    ) -> Self::Result {
        self.check_inputs(plan)
    }

    fn visit_logical_dedup(
        &mut self,
        plan: &crate::optimizer::plan_node::LogicalDedup,
    ) -> Self::Result {
        self.check_inputs(plan)
    }

    fn visit_logical_iceberg_scan(
        &mut self,
        _: &crate::optimizer::plan_node::LogicalIcebergScan,
    ) -> Self::Result {
        CheckResult {
            supported: true,
            have_iceberg_scan: true,
        }
    }
}

#[easy_ext::ext(DataFusionExecuteCheckerExt)]
pub impl LogicalPlanRef {
    /// Returns `CheckResult` indicating if this plan can be executed by DataFusion.
    fn check_for_datafusion(&self) -> CheckResult {
        DataFusionExecuteChecker.visit(self.clone())
    }
}
Repository: risingwavelabs/risingwave
File: grafana/dashboard/common.py (Python)

Code before:
        "25": ["0.25", "25"],
        "50": ["0.5", "50"],
        "60": ["0.6", "60"],
        "75": ["0.75", "75"],
        "90": ["0.9", "90"],
        "99": ["0.99", "99"],
        "999": ["0.999", "999"],
        "100": ["1.0", "100"],
        "max": ["1.0", "max"],
    }
    return list(
        map(lambda p: f(quantile_map[str(p)][0], quantile_map[str(p)][1]), percentiles)
    )


def epoch_to_unix_millis(epoch_expr):
    # UNIX_RISINGWAVE_DATE_SEC
    return f"(1617235200000+({epoch_expr} != 0)/65536)"


def alert_when(expr: str) -> str:
    """Return 1 when expr > 0, and drop the series when expr <= 0."""
    return f"(({expr}) > bool 0) > 0"


def alert_threshold(expr: str, threshold, comparator: str = ">=") -> str:
    """Return 1 when expr compares true against threshold, and drop the series otherwise."""
    return f"(({expr}) {comparator} bool {threshold}) > 0"


Review comment (suggestion):
The `threshold` parameter is missing a type hint. For consistency with the `expr` and `comparator` parameters and to improve code clarity, consider adding a type hint such as `threshold: float` or `threshold: Union[int, float]`.

Code after:
    if node_filter_enabled:
        filters.append(f'{COMPONENT_LABEL}=~"${COMPONENT_VARIABLE}"')
        filters.append(f'{NODE_LABEL}=~"${NODE_VARIABLE}"')
    if filters:
        return f"{name}{{{','.join(filters)}}}"
    else:
        return name


def table_metric(name, filter=None):
    return metric(name, filter, True, True)


def quantile(f, percentiles):
    quantile_map = {
        "10": ["0.1", "10"],
        "25": ["0.25", "25"],
        "50": ["0.5", "50"],
        "60": ["0.6", "60"],
        "75": ["0.75", "75"],
        "90": ["0.9", "90"],
        "99": ["0.99", "99"],
        "999": ["0.999", "999"],
        "100": ["1.0", "100"],
        "max": ["1.0", "max"],
    }
    return list(
        map(lambda p: f(quantile_map[str(p)][0], quantile_map[str(p)][1]), percentiles)
    )


def epoch_to_unix_millis(epoch_expr):
    # UNIX_RISINGWAVE_DATE_SEC
    return f"(1617235200000+({epoch_expr} != 0)/65536)"


def alert_when(expr: str) -> str:
    """Return 1 when expr > 0, and drop the series when expr <= 0."""
    return f"(({expr}) > bool 0) > 0"


def alert_threshold(expr: str, threshold: Union[int, float], comparator: str = ">=") -> str:
    """Return 1 when expr compares true against threshold, and drop the series otherwise."""
    return f"(({expr}) {comparator} bool {threshold}) > 0"
Repository: risingwavelabs/risingwave
File: src/tests/simulation/src/client.rs (Rust)

Code before:
    fn drop(&mut self) {
        self.task.abort();
    }
}

fn parse_risedev_ctl_args(command: &std::process::Command) -> Option<Vec<String>> {
    let program = command.get_program().to_str()?;
    if program != "bash" && !program.ends_with("/bash") {
        return None;
    }

    let mut args = command.get_args();
    if args.next()?.to_str()? != "-c" {
        return None;
    }
    let script = args.next()?.to_str()?;
    if args.next().is_some() {
        return None;
    }

    let mut parts = split(script).ok()?;
    if parts.len() < 2 || parts[0] != "./risedev" || parts[1] != "ctl" {
        return None;
    }
    Some(parts.split_off(2))
}

fn success_output() -> std::process::Output {
    #[cfg(unix)]
    {
        use std::os::unix::process::ExitStatusExt;

        std::process::Output {
            status: std::process::ExitStatus::from_raw(0),
            stdout: vec![],
            stderr: vec![],
        }
    }

    #[cfg(not(unix))]
    {
        unimplemented!("simulation mode does not support non-unix platforms")
    }
}

#[async_trait::async_trait]
impl sqllogictest::AsyncDB for RisingWave {
    type ColumnType = DefaultColumnType;
    type Error = tokio_postgres::error::Error;

    async fn run(&mut self, sql: &str) -> Result<DBOutput<Self::ColumnType>, Self::Error> {

Review comment (refactor):
Consider adding documentation for this function to explain its purpose, expected input format, and return value. The function parses commands in the format "bash -c './risedev ctl <args>'" and extracts the ctl arguments. This is important for maintainability and understanding the contract this function enforces.

Code after:
    fn drop(&mut self) {
        self.task.abort();
    }
}

fn parse_risedev_ctl_args(command: &std::process::Command) -> Option<Vec<String>> {
    let program = command.get_program().to_str()?;
    if program != "bash" && !program.ends_with("/bash") {
        return None;
    }

    let mut args = command.get_args();
    if args.next()?.to_str()? != "-c" {
        return None;
    }
    let script = args.next()?.to_str()?;
    if args.next().is_some() {
        return None;
    }

    let mut parts = split(script).ok()?;
    if parts.len() < 2 || parts[0] != "./risedev" || parts[1] != "ctl" {
        return None;
    }
    Some(parts.split_off(2))
}

fn command_output(exit_code: i32, stderr: Vec<u8>) -> std::process::Output {
    #[cfg(unix)]
    {
        use std::os::unix::process::ExitStatusExt;

        std::process::Output {
            status: std::process::ExitStatus::from_raw(exit_code << 8),
            stdout: vec![],
            stderr,
        }
    }

    #[cfg(not(unix))]
    {
        unimplemented!("simulation mode does not support non-unix platforms")
    }
}

#[async_trait::async_trait]
impl sqllogictest::AsyncDB for RisingWave {
    type ColumnType = DefaultColumnType;
    type Error = tokio_postgres::error::Error;

    async fn run(&mut self, sql: &str) -> Result<DBOutput<Self::ColumnType>, Self::Error> {
Repository: risingwavelabs/risingwave
File: src/tests/simulation/src/client.rs (Rust)

Code before:
    }

    let mut parts = split(script).ok()?;
    if parts.len() < 2 || parts[0] != "./risedev" || parts[1] != "ctl" {
        return None;
    }
    Some(parts.split_off(2))
}

fn success_output() -> std::process::Output {
    #[cfg(unix)]
    {
        use std::os::unix::process::ExitStatusExt;

        std::process::Output {
            status: std::process::ExitStatus::from_raw(0),
            stdout: vec![],
            stderr: vec![],
        }
    }

    #[cfg(not(unix))]
    {
        unimplemented!("simulation mode does not support non-unix platforms")
    }
}

#[async_trait::async_trait]
impl sqllogictest::AsyncDB for RisingWave {
    type ColumnType = DefaultColumnType;
    type Error = tokio_postgres::error::Error;

    async fn run(&mut self, sql: &str) -> Result<DBOutput<Self::ColumnType>, Self::Error> {
        use sqllogictest::DBOutput;

        if self.client.is_closed() {
            // connection error, reset the client
            self.reconnect().await?;
        }

        if sql.trim_start().to_lowercase().starts_with("set") {
            self.set_stmts.push(sql);
        }

        let mut output = vec![];

        let rows = self.client.simple_query(sql).await?;
        let mut cnt = 0;
        for row in rows {
            let mut row_vec = vec![];
            match row {

Review comment (suggestion):
Consider adding documentation for this function to explain why it creates an empty output with exit status 0, and under what circumstances it should be used. This helps future maintainers understand the contract and platform-specific behavior.

Code after:
    }

    let mut parts = split(script).ok()?;
    if parts.len() < 2 || parts[0] != "./risedev" || parts[1] != "ctl" {
        return None;
    }
    Some(parts.split_off(2))
}

fn command_output(exit_code: i32, stderr: Vec<u8>) -> std::process::Output {
    #[cfg(unix)]
    {
        use std::os::unix::process::ExitStatusExt;

        std::process::Output {
            status: std::process::ExitStatus::from_raw(exit_code << 8),
            stdout: vec![],
            stderr,
        }
    }

    #[cfg(not(unix))]
    {
        unimplemented!("simulation mode does not support non-unix platforms")
    }
}

#[async_trait::async_trait]
impl sqllogictest::AsyncDB for RisingWave {
    type ColumnType = DefaultColumnType;
    type Error = tokio_postgres::error::Error;

    async fn run(&mut self, sql: &str) -> Result<DBOutput<Self::ColumnType>, Self::Error> {
        use sqllogictest::DBOutput;

        if self.client.is_closed() {
            // connection error, reset the client
            self.reconnect().await?;
        }

        if sql.trim_start().to_lowercase().starts_with("set") {
            self.set_stmts.push(sql);
        }

        let mut output = vec![];

        let rows = self.client.simple_query(sql).await?;
        let mut cnt = 0;
        for row in rows {
            let mut row_vec = vec![];
            match row {
Repository: risingwavelabs/risingwave
File: src/tests/simulation/src/client.rs (Rust)

Code before:
    fn drop(&mut self) {
        self.task.abort();
    }
}

fn parse_risedev_ctl_args(command: &std::process::Command) -> Option<Vec<String>> {
    let program = command.get_program().to_str()?;
    if program != "bash" && !program.ends_with("/bash") {
        return None;
    }

    let mut args = command.get_args();
    if args.next()?.to_str()? != "-c" {
        return None;
    }
    let script = args.next()?.to_str()?;
    if args.next().is_some() {
        return None;
    }

    let mut parts = split(script).ok()?;
    if parts.len() < 2 || parts[0] != "./risedev" || parts[1] != "ctl" {
        return None;
    }
    Some(parts.split_off(2))
}

fn success_output() -> std::process::Output {
    #[cfg(unix)]
    {
        use std::os::unix::process::ExitStatusExt;

        std::process::Output {
            status: std::process::ExitStatus::from_raw(0),
            stdout: vec![],
            stderr: vec![],
        }
    }

    #[cfg(not(unix))]
    {
        unimplemented!("simulation mode does not support non-unix platforms")
    }
}

#[async_trait::async_trait]
impl sqllogictest::AsyncDB for RisingWave {
    type ColumnType = DefaultColumnType;
    type Error = tokio_postgres::error::Error;

    async fn run(&mut self, sql: &str) -> Result<DBOutput<Self::ColumnType>, Self::Error> {

Review comment (bug):
Consider adding unit tests for the parse_risedev_ctl_args function to verify it correctly handles various input cases, including edge cases like empty scripts, malformed commands, scripts with extra arguments, and commands that don't match the expected pattern. Following the pattern in slt.rs which has a tests module, you could add a similar module here to test this parsing logic.

Code after:
    fn drop(&mut self) {
        self.task.abort();
    }
}

fn parse_risedev_ctl_args(command: &std::process::Command) -> Option<Vec<String>> {
    let program = command.get_program().to_str()?;
    if program != "bash" && !program.ends_with("/bash") {
        return None;
    }

    let mut args = command.get_args();
    if args.next()?.to_str()? != "-c" {
        return None;
    }
    let script = args.next()?.to_str()?;
    if args.next().is_some() {
        return None;
    }

    let mut parts = split(script).ok()?;
    if parts.len() < 2 || parts[0] != "./risedev" || parts[1] != "ctl" {
        return None;
    }
    Some(parts.split_off(2))
}

fn command_output(exit_code: i32, stderr: Vec<u8>) -> std::process::Output {
    #[cfg(unix)]
    {
        use std::os::unix::process::ExitStatusExt;

        std::process::Output {
            status: std::process::ExitStatus::from_raw(exit_code << 8),
            stdout: vec![],
            stderr,
        }
    }

    #[cfg(not(unix))]
    {
        unimplemented!("simulation mode does not support non-unix platforms")
    }
}

#[async_trait::async_trait]
impl sqllogictest::AsyncDB for RisingWave {
    type ColumnType = DefaultColumnType;
    type Error = tokio_postgres::error::Error;

    async fn run(&mut self, sql: &str) -> Result<DBOutput<Self::ColumnType>, Self::Error> {
Repository: risingwavelabs/risingwave
File: src/tests/simulation/src/client.rs (Rust)

Code before:
        if output.is_empty() {
            Ok(DBOutput::StatementComplete(cnt))
        } else {
            Ok(DBOutput::Rows {
                types: vec![DefaultColumnType::Any; output[0].len()],
                rows: output,
            })
        }
    }

    async fn shutdown(&mut self) {}

    fn engine_name(&self) -> &str {
        "risingwave"
    }

    async fn sleep(dur: Duration) {
        tokio::time::sleep(dur).await
    }

    async fn run_command(command: std::process::Command) -> std::io::Result<std::process::Output> {
        if let Some(ctl_args) = parse_risedev_ctl_args(&command) {
            start_ctl(ctl_args)
                .await
                .map_err(|err| std::io::Error::other(err.to_string()))?;
            return Ok(success_output());
        }
        unimplemented!("spawning process is not supported in simulation mode")
    }
}


Review comment (suggestion):
Ideally the stdout of `ctl` should also be preserved, as some commands are intended to produce output with stdout.

Code after:
        if output.is_empty() {
            Ok(DBOutput::StatementComplete(cnt))
        } else {
            Ok(DBOutput::Rows {
                types: vec![DefaultColumnType::Any; output[0].len()],
                rows: output,
            })
        }
    }

    async fn shutdown(&mut self) {}

    fn engine_name(&self) -> &str {
        "risingwave"
    }

    async fn sleep(dur: Duration) {
        tokio::time::sleep(dur).await
    }

    async fn run_command(command: std::process::Command) -> std::io::Result<std::process::Output> {
        if let Some(ctl_args) = parse_risedev_ctl_args(&command) {
            let output = match start_ctl(ctl_args).await {
                Ok(()) => command_output(0, vec![]),
                Err(err) => command_output(1, format!("{err:#}\n").into_bytes()),
            };
            return Ok(output);
        }
        unimplemented!("spawning process is not supported in simulation mode")
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn command_output_uses_exit_code_and_stderr() {
        let output = command_output(1, b"ctl failed\n".to_vec());

        assert!(!output.status.success());
        assert_eq!(output.status.code(), Some(1));
        assert_eq!(output.stderr, b"ctl failed\n");
    }
}
Repository: roboflow/supervision
File: test/classification/test_core.py (Python)

Code before:
    'class_id, confidence, expected_result, exception',
    [
        (
            [0, 1, 2, 3, 4],
            [0.1, 0.2, 0.9, 0.4, 0.5],
            (np.array([2, 4, 3, 1, 0]), np.array([0.9, 0.5, 0.4, 0.2, 0.1])),
            DoesNotRaise()
        ),  # class_id with 5 numbers and 5 confidences
        (
            [0, 1, 2, 3, 4],
            [0.1, 0.2, 0.3, 0.4],
            None,
            pytest.raises(ValueError)
        ),  # class_id with 5 numbers and 4 confidences
    ]
)
def test_top_k(
    class_id: List[T],
    confidence: Optional[List[T]],
    expected_result: Optional[Tuple[List[T], List[T]]],
    exception: Exception
) -> None:
    with exception:
        result = Classifications(class_id=np.array(class_id), confidence=np.array(confidence)).get_top_k(len(class_id))

        assert result[0].tolist() == expected_result[0].tolist()
        assert result[1].tolist() == expected_result[1].tolist()


Review comment (suggestion):
Please use `np.array_equal(result[0], expected_result[0])` not `result[0].tolist() == expected_result[0].tolist()`.

Code after:
        (
            np.array([0, 1, 2, 3, 4]),
            np.array([0.1, 0.2, 0.9, 0.4, 0.5]),
            5,
            (np.array([2, 4, 3, 1, 0]), np.array([0.9, 0.5, 0.4, 0.2, 0.1])),
            DoesNotRaise(),
        ),  # class_id with 5 numbers and 5 confidences
        (
            np.array([5, 1, 2, 3, 4]),
            np.array([0.1, 0.2, 0.9, 0.4, 0.5]),
            1,
            (np.array([2]), np.array([0.9])),
            DoesNotRaise(),
        ),  # class_id with 5 numbers and 5 confidences, retrieve where k = 1
        (
            np.array([4, 1, 2, 3, 6, 5]),
            np.array([0.8, 0.2, 0.9, 0.4, 0.5, 0.1]),
            2,
            (np.array([2, 4]), np.array([0.9, 0.8])),
            DoesNotRaise(),
        ),  # class_id with 5 numbers and 5 confidences, retrieve where k = 3
        (
            np.array([0, 1, 2, 3, 4]),
            np.array([]),
            5,
            None,
            pytest.raises(ValueError),
        ),  # class_id with 5 numbers and 0 confidences
        (
            [0, 1, 2, 3, 4],
            [0.1, 0.2, 0.3, 0.4],
            5,
            None,
            pytest.raises(ValueError),
        ),  # class_id with 5 numbers and 4 confidences
    ],
)
def test_top_k(
    class_id: np.ndarray,
    confidence: Optional[np.ndarray],
    k: int,
    expected_result: Optional[Tuple[np.ndarray, np.ndarray]],
    exception: Exception,
) -> None:
    with exception:
        result = Classifications(
            class_id=np.array(class_id), confidence=np.array(confidence)
        ).get_top_k(k)

        assert np.array_equal(result[0], expected_result[0])
        assert np.array_equal(result[1], expected_result[1])
Repository: roboflow/supervision
File: supervision/detection/core.py (Python)

Code before:
            AssertionError: If `confidence` is None and class_agnostic is False.
                If `class_id` is None and class_agnostic is False.
        """
        if len(self) == 0:
            return self

        assert (
            self.confidence is not None
        ), "Detections confidence must be given for NMS to be executed."

        if class_agnostic:
            predictions = np.hstack((self.xyxy, self.confidence.reshape(-1, 1)))
        else:
            assert self.class_id is not None, (
                "Detections class_id must be given for NMS to be executed. If you"
                " intended to perform class agnostic NMS set class_agnostic=True."
            )
            predictions = np.hstack(
                (
                    self.xyxy,
                    self.confidence.reshape(-1, 1),
                    self.class_id.reshape(-1, 1),
                )
            )

        if hasattr(self, "mask") and self.mask is not None:
            indices = mask_non_max_suppression(
                predictions=predictions, masks=self.mask, iou_threshold=threshold
            )
        else:
            indices = box_non_max_suppression(
                predictions=predictions, iou_threshold=threshold
            )

        return self[indices]


Review comment (refactor):
`Detections` always has an attribute `mask`, but it can be equal to `None` or `np.ndarray`. So we can simplify this condition to `self.mask is not None`.

Code after:
            AssertionError: If `confidence` is None and class_agnostic is False.
                If `class_id` is None and class_agnostic is False.
        """
        if len(self) == 0:
            return self

        assert (
            self.confidence is not None
        ), "Detections confidence must be given for NMS to be executed."

        if class_agnostic:
            predictions = np.hstack((self.xyxy, self.confidence.reshape(-1, 1)))
        else:
            assert self.class_id is not None, (
                "Detections class_id must be given for NMS to be executed. If you"
                " intended to perform class agnostic NMS set class_agnostic=True."
            )
            predictions = np.hstack(
                (
                    self.xyxy,
                    self.confidence.reshape(-1, 1),
                    self.class_id.reshape(-1, 1),
                )
            )

        if self.mask is not None:
            indices = mask_non_max_suppression(
                predictions=predictions, masks=self.mask, iou_threshold=threshold
            )
        else:
            indices = box_non_max_suppression(
                predictions=predictions, iou_threshold=threshold
            )

        return self[indices]