ronantakizawa/codereview-bench huggingface dataset:
Repository: tikv/tikv
File: components/tidb_query_expr/src/impl_time.rs (Rust)

Code before:

#[rpn_fn(capture = [ctx])]
#[inline]
pub fn unix_timestamp_int(ctx: &mut EvalContext, time: &DateTime) -> Result<Option<i64>> {
    let timestamp_res = get_micro_timestamp(time, &ctx.cfg.tz);
    let timestamp = match timestamp_res {
        Ok(val) => val,
        Err(err) => return Err(err),
    };

    let res: std::result::Result<i64, Error> =
        unix_timestamp_to_mysql_unix_timestamp(ctx, timestamp, 1)?
            .as_i64()
            .into();
    Ok(Some(res?))
}

#[rpn_fn(capture = [ctx, extra])]
#[inline]
pub fn unix_timestamp_decimal(
    ctx: &mut EvalContext,
    extra: &RpnFnCallExtra,
    time: &DateTime,
) -> Result<Option<Decimal>> {
    let timestamp_res = get_micro_timestamp(time, &ctx.cfg.tz);
    let timestamp = match timestamp_res {
        Ok(val) => val,
        Err(err) => return Err(err),
    };

    let res = unix_timestamp_to_mysql_unix_timestamp(
        ctx,
        timestamp,
        extra.ret_field_type.get_decimal() as i8,
    )?;
    return Ok(Some(res));
}

#[cfg(test)]
mod tests {
    use std::{str::FromStr, sync::Arc};

    use tidb_query_datatype::{
        builder::FieldTypeBuilder,
        codec::{
            batch::LazyBatchColumnVec,
            data_type::*,
            error::ERR_TRUNCATE_WRONG_VALUE,
            mysql::{Time, MAX_FSP},
        },
        expr::EvalConfig,

Review comment (suggestion):
```suggestion
    let timestamp = match get_micro_timestamp(time, &ctx.cfg.tz) {
```

Code after:

#[rpn_fn(capture = [ctx])]
#[inline]
pub fn unix_timestamp_int(ctx: &mut EvalContext, time: &DateTime) -> Result<Option<i64>> {
    let timestamp = match get_micro_timestamp(time, &ctx.cfg.tz) {
        Ok(val) => val,
        Err(err) => return Err(err),
    };

    let res: std::result::Result<i64, Error> =
        unix_timestamp_to_mysql_unix_timestamp(ctx, timestamp, 1)?
            .as_i64()
            .into();
    Ok(Some(res?))
}

#[rpn_fn(capture = [ctx, extra])]
#[inline]
pub fn unix_timestamp_decimal(
    ctx: &mut EvalContext,
    extra: &RpnFnCallExtra,
    time: &DateTime,
) -> Result<Option<Decimal>> {
    let timestamp = match get_micro_timestamp(time, &ctx.cfg.tz) {
        Ok(val) => val,
        Err(err) => return Err(err),
    };

    let res = unix_timestamp_to_mysql_unix_timestamp(
        ctx,
        timestamp,
        extra.ret_field_type.get_decimal() as i8,
    )?;
    Ok(Some(res))
}

fn build_timestamp_diff_meta(expr: &mut Expr) -> Result<IntervalUnit> {
    let children = expr.mut_children();
    if children.len() != 3 {
        return Err(box_err!(
            "wrong timestamp_diff expr size {}",
            children.len()
        ));
    }
    let unit_str = match children[0].get_tp() {
        ExprType::Bytes | ExprType::String => {
            std::str::from_utf8(children[0].get_val()).map_err(Error::Encoding)?
        }
        _ => return Err(box_err!("unknown unit type {:?}", children[0].get_tp())),
    };
    let unit = IntervalUnit::from_str(unit_str)?;
Repository: tikv/tikv
File: components/tikv_util/src/lib.rs (Rust)

Code before:
    /// Indicates whether a sufficient number of Raft peers have caught up
    /// applying logs.
    pub raft_peers_caught_up: AtomicBool,
}

impl ServerReadiness {
    /// Checks if the server is ready.
    ///
    /// All conditions must be met for the server to be considered ready.
    pub fn is_ready(&self) -> bool {
        self.raft_peers_caught_up.load(Ordering::SeqCst)
            && self.connected_to_pd.load(Ordering::SeqCst)
    }

    pub fn to_json(&self) -> String {
        let json_result = serde_json::to_string_pretty(&self);
        match json_result {
            Ok(json) => json,
            Err(e) => {
                format!("failed to serialize ServerReadiness: {}", e)
            }
        }
    }
}

impl Default for ServerReadiness {
    fn default() -> Self {
        ServerReadiness {
            raft_peers_caught_up: AtomicBool::new(false),
            connected_to_pd: AtomicBool::new(false),
        }
    }
}

#[cfg(test)]
mod tests {
    use std::{
        io::Read,
        rc::Rc,
        sync::atomic::{AtomicBool, Ordering},
        *,
    };

    use tempfile::Builder;

    use super::*;

    #[test]
    fn test_panic_hook() {
        use gag::BufferRedirect;
        use slog::{self, Drain, OwnedKVList, Record};

Review comment (suggestion):
It can be replaced with `derive(Default)` since the default value of `AtomicBool` is `AtomicBool::new(false)`

Code after:
        }
    }))
}

/// Checks environment variables that affect TiKV.
pub fn check_environment_variables() {
    if cfg!(unix) && env::var("TZ").is_err() {
        env::set_var("TZ", ":/etc/localtime");
        warn!("environment variable `TZ` is missing, using `/etc/localtime`");
    }

    if let Ok(var) = env::var("GRPC_POLL_STRATEGY") {
        info!(
            "environment variable is present";
            "GRPC_POLL_STRATEGY" => var
        );
    }

    for proxy in &["http_proxy", "https_proxy"] {
        if let Ok(var) = env::var(proxy) {
            info!("environment variable is present";
                *proxy => var
            );
        }
    }
}

/// Create a child process and wait to get its exit code.
pub fn run_and_wait_child_process(child: impl Fn()) -> Result<i32, String> {
    match unsafe { fork() } {
        Ok(ForkResult::Parent { .. }) => match wait().unwrap() {
            WaitStatus::Exited(_, status) => Ok(status),
            v => Err(format!("{:?}", v)),
        },
        Ok(ForkResult::Child) => {
            child();
            std::process::exit(0);
        }
        Err(e) => Err(format!("Fork failed: {}", e)),
    }
}

#[inline]
pub fn is_zero_duration(d: &Duration) -> bool {
    d.as_secs() == 0 && d.subsec_nanos() == 0
}

pub fn empty_shared_slice<T>() -> Arc<[T]> {
    Vec::new().into()
}
Repository: tikv/tikv
File: components/compact-log-backup/src/util.rs (Rust)

Code before:
}

/// A wrapper that make a `u64` always be displayed as {:016X}.
#[derive(Debug)]
struct HexU64(u64);

impl Display for HexU64 {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "{:016X}", self.0)
    }
}

pub fn aligned_u64(v: u64) -> impl Display {
    HexU64(v)
}

pub fn compression_type_to_u8(c: SstCompressionType) -> u8 {
    match c {
        SstCompressionType::Lz4 => 0,
        SstCompressionType::Snappy => 1,
        SstCompressionType::Zstd => 2,
    }
}

pub fn redact(k: &Vec<u8>) -> log_wrappers::Value<'_> {
    log_wrappers::Value::key(k.as_slice())
}

#[derive(Eq, PartialEq)]
pub struct EndKey<'a>(pub &'a [u8]);

impl PartialOrd for EndKey<'_> {
    fn partial_cmp(&self, other: &Self) -> Option<std::cmp::Ordering> {
        use std::cmp::Ordering::*;
        match (self, other) {
            (EndKey(b""), EndKey(b"")) => Some(Equal),
            (EndKey(b""), _) => Some(Greater),
            (_, EndKey(b"")) => Some(Less),
            (EndKey(a), EndKey(b)) => Some(a.cmp(b)),
        }
    }
}

impl Ord for EndKey<'_> {
    fn cmp(&self, other: &Self) -> std::cmp::Ordering {
        self.partial_cmp(other).unwrap()
    }
}

/// Get the storage URL string.
/// Once an error occurred, return the error message.

Review comment (suggestion):
The parameter type should be &[u8] instead of &Vec<u8> to be more flexible and follow Rust best practices. Taking a reference to a Vec unnecessarily restricts the function to only accept Vec references, when it could accept any byte slice.
```suggestion
pub fn redact(k: &[u8]) -> log_wrappers::Value<'_> {
    log_wrappers::Value::key(k)
```

Code after:
}

/// A wrapper that make a `u64` always be displayed as {:016X}.
#[derive(Debug)]
struct HexU64(u64);

impl Display for HexU64 {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "{:016X}", self.0)
    }
}

pub fn aligned_u64(v: u64) -> impl Display {
    HexU64(v)
}

pub fn compression_type_to_u8(c: SstCompressionType) -> u8 {
    match c {
        SstCompressionType::Lz4 => 0,
        SstCompressionType::Snappy => 1,
        SstCompressionType::Zstd => 2,
    }
}

pub fn redact(k: &[u8]) -> log_wrappers::Value<'_> {
    log_wrappers::Value::key(k)
}

#[derive(Eq, PartialEq)]
pub struct EndKey<'a>(pub &'a [u8]);

impl PartialOrd for EndKey<'_> {
    fn partial_cmp(&self, other: &Self) -> Option<std::cmp::Ordering> {
        use std::cmp::Ordering::*;
        match (self, other) {
            (EndKey(b""), EndKey(b"")) => Some(Equal),
            (EndKey(b""), _) => Some(Greater),
            (_, EndKey(b"")) => Some(Less),
            (EndKey(a), EndKey(b)) => Some(a.cmp(b)),
        }
    }
}

impl Ord for EndKey<'_> {
    fn cmp(&self, other: &Self) -> std::cmp::Ordering {
        self.partial_cmp(other).unwrap()
    }
}

/// Get the storage URL string.
/// Once an error occurred, return the error message.
Repository: tikv/tikv
File: components/tikv_util/src/timer.rs (Rust)

Code before:
                        last_now_ms = last_now;
                    }
                }
            } else {
                return self.start + Duration::from_millis(last_now_ms);
            }
        }
    }
}

impl<T: Now> RatchetClock<T> {
    fn new(clock: T) -> Self {
        RatchetClock {
            clock,
            start: std::time::Instant::now(),
            last_now_ms: AtomicU64::new(0),
        }
    }
}

/// Start a timer thread with specific thread name and clock source.
fn start_timer_thread<T: Now>(name: &str, clock: T) -> Handle {
    let (tx, rx) = mpsc::channel();
    let props = crate::thread_group::current_properties();
    let ratchet_clock = RatchetClock::new(clock);
    let clock = Clock::new_with_now(ratchet_clock);
    Builder::new()
        .name(thd_name!(name))
        .spawn_wrapper(move || {
            crate::thread_group::set_properties(props);

            let mut timer = tokio_timer::Timer::new_with_now(ParkThread::new(), clock);
            tx.send(timer.handle()).unwrap();
            loop {
                timer.turn(None).unwrap();
            }
        })
        .unwrap();
    rx.recv().unwrap()
}

#[cfg(test)]
mod tests {
    use std::sync::atomic::AtomicBool;

    use futures::{compat::Future01CompatExt, executor::block_on};

    use super::*;

    #[derive(Debug, PartialEq, Copy, Clone)]
    enum Task {

Review comment (suggestion):
I if understand correctly, the timer thread is a single-threaded, and the clock is only used by the timer thread itself, it's seems unnecessary to let the clock to be thread safe. 

Code after:
}

impl<T: Now> RatchetClock<T> {
    fn new(clock: T) -> Self {
        RatchetClock {
            state: Mutex::new(RatchetClockState {
                clock,
                start: std::time::Instant::now(),
                last_now_ms: 0,
            }),
        }
    }
}

/// Start a timer thread with specific thread name and clock source.
fn start_timer_thread<T: Now>(name: &str, clock: T) -> Handle {
    let (tx, rx) = mpsc::channel();
    let props = crate::thread_group::current_properties();
    let ratchet_clock = RatchetClock::new(clock);
    let clock = Clock::new_with_now(ratchet_clock);
    Builder::new()
        .name(thd_name!(name))
        .spawn_wrapper(move || {
            crate::thread_group::set_properties(props);

            let mut timer = tokio_timer::Timer::new_with_now(ParkThread::new(), clock);
            tx.send(timer.handle()).unwrap();
            loop {
                timer.turn(None).unwrap();
            }
        })
        .unwrap();
    rx.recv().unwrap()
}

#[cfg(test)]
mod tests {
    use std::sync::atomic::{AtomicBool, Ordering as AtomicOrdering};

    use futures::{compat::Future01CompatExt, executor::block_on};

    use super::*;

    #[derive(Debug, PartialEq, Copy, Clone)]
    enum Task {
        A,
        B,
        C,
    }

    #[test]
Repository: tikv/tikv
File: components/raftstore/src/store/msg.rs (Rust)

Code before:
            self,
            PeerMsg::SignificantMsg(box SignificantMsg::CaptureChange { .. })
        )
    }
}

#[derive(EnumCount, EnumVariantNames)]
pub enum StoreMsg<EK>
where
    EK: KvEngine,
{
    RaftMessage(Box<InspectedRaftMessage>),

    StoreUnreachable {
        store_id: u64,
    },

    // Clear region size and keys for all regions in the range, so we can force them to
    // re-calculate their size later.
    ClearRegionSizeInRange {
        start_key: Vec<u8>,
        end_key: Vec<u8>,
    },

    // Compaction finished event
    CompactedEvent(EK::CompactedEvent),
    Tick(StoreTick),
    Start {
        store: metapb::Store,
    },

    /// Asks the store to update replication mode.
    UpdateReplicationMode(ReplicationStatus),

    /// Inspect the latency of raftstore.
    LatencyInspect {
        factor: InspectFactor,
        send_time: Instant,
        inspector: LatencyInspector,
    },

    UnsafeRecoveryReport(pdpb::StoreReport),
    UnsafeRecoveryCreatePeer {
        syncer: UnsafeRecoveryExecutePlanSyncer,
        create: metapb::Region,
    },

    GcSnapshotFinish,

    AwakenRegions {
        abnormal_stores: Vec<u64>,

Review comment (suggestion):
`CompactedEvent` should be moved up as well? 

```
impl<EK: KvEngine> StoreMsg<EK> {
    pub fn discriminant(&self) -> usize {
        match self {
            StoreMsg::RaftMessage(_) => 0,
            StoreMsg::StoreUnreachable { .. } => 1,
            StoreMsg::CompactedEvent(_) => 2,
            StoreMsg::ClearRegionSizeInRange { .. } => 3,
            StoreMsg::Tick(_) => 4,
```

Code after:
            self,
            PeerMsg::SignificantMsg(box SignificantMsg::CaptureChange { .. })
        )
    }
}

#[derive(EnumCount, EnumVariantNames)]
pub enum StoreMsg<EK>
where
    EK: KvEngine,
{
    RaftMessage(Box<InspectedRaftMessage>),

    StoreUnreachable {
        store_id: u64,
    },

    // Compaction finished event
    CompactedEvent(EK::CompactedEvent),

    // Clear region size and keys for all regions in the range, so we can force them to
    // re-calculate their size later.
    ClearRegionSizeInRange {
        start_key: Vec<u8>,
        end_key: Vec<u8>,
    },

    Tick(StoreTick),
    Start {
        store: metapb::Store,
    },

    /// Asks the store to update replication mode.
    UpdateReplicationMode(ReplicationStatus),

    /// Inspect the latency of raftstore.
    LatencyInspect {
        factor: InspectFactor,
        send_time: Instant,
        inspector: LatencyInspector,
    },

    UnsafeRecoveryReport(pdpb::StoreReport),
    UnsafeRecoveryCreatePeer {
        syncer: UnsafeRecoveryExecutePlanSyncer,
        create: metapb::Region,
    },

    GcSnapshotFinish,

    AwakenRegions {
Repository: tikv/tikv
File: components/cloud/aws/src/s3.rs (Rust)

Code before:
// Copyright 2019 TiKV Project Authors. Licensed under Apache-2.0.
use std::{
    error::Error as StdError,
    io,
    pin::Pin,
    time::{Duration, SystemTime},
};

use async_trait::async_trait;
use aws_config::{sts::AssumeRoleProvider, BehaviorVersion, Region, SdkConfig};
use aws_credential_types::{provider::ProvideCredentials, Credentials};
use aws_sdk_s3::{
<<<<<<< HEAD
    config::HttpClient,
=======
    Client,
    config::{HttpClient, StalledStreamProtectionConfig},
>>>>>>> 219816571a (aws: disable stalled stream protection for S3Storage (#18875))
    operation::get_object::GetObjectError,
    types::{CompletedMultipartUpload, CompletedPart},
    Client,
};
use bytes::Bytes;
use cloud::{
    blob::{
        none_to_empty, BlobConfig, BlobObject, BlobStorage, BucketConf, DeletableStorage,
        IterableStorage, PutResource, StringNonEmpty,
    },
    metrics::CLOUD_REQUEST_HISTOGRAM_VEC,
};
use fail::fail_point;
use futures::{executor::block_on, stream::Stream};
use futures_util::{
    future::{FutureExt, LocalBoxFuture},
    io::{AsyncRead, AsyncReadExt},
    stream::TryStreamExt,
    StreamExt,
};
pub use kvproto::brpb::S3 as InputConfig;
use thiserror::Error;
use tikv_util::{
    debug,
    stream::{error_stream, RetryError},

Review comment (suggestion):
```suggestion
    config::{HttpClient, StalledStreamProtectionConfig},
```

Code after:
// Copyright 2019 TiKV Project Authors. Licensed under Apache-2.0.
use std::{
    error::Error as StdError,
    io,
    pin::Pin,
    time::{Duration, SystemTime},
};

use async_trait::async_trait;
use aws_config::{sts::AssumeRoleProvider, BehaviorVersion, Region, SdkConfig};
use aws_credential_types::{provider::ProvideCredentials, Credentials};
use aws_sdk_s3::{
    config::{HttpClient, StalledStreamProtectionConfig},
    operation::get_object::GetObjectError,
    types::{CompletedMultipartUpload, CompletedPart},
    Client,
};
use bytes::Bytes;
use cloud::{
    blob::{
        none_to_empty, BlobConfig, BlobObject, BlobStorage, BucketConf, DeletableStorage,
        IterableStorage, PutResource, StringNonEmpty,
    },
    metrics::CLOUD_REQUEST_HISTOGRAM_VEC,
};
use fail::fail_point;
use futures::{executor::block_on, stream::Stream};
use futures_util::{
    future::{FutureExt, LocalBoxFuture},
    io::{AsyncRead, AsyncReadExt},
    stream::TryStreamExt,
    StreamExt,
};
pub use kvproto::brpb::S3 as InputConfig;
use thiserror::Error;
use tikv_util::{
    debug,
    stream::{error_stream, RetryError},
    time::Instant,
};
use tokio::time::{sleep, timeout};
use tokio_util::io::ReaderStream;
Repository: tikv/tikv
File: components/raftstore/src/store/worker/split_check.rs (Rust)

Code before:
    ChangeConfig(ConfigChange),
    // Compaction finished event
    CompactedEvent {
        event: EK::CompactedEvent,
        region_split_check_diff: u64,
    },
    #[cfg(any(test, feature = "testexport"))]
    Validate(Box<dyn FnOnce(&Config) + Send>),
}

impl<EK> Task<EK>
where
    EK: KvEngine,
{
    pub fn split_check(
        region: Region,
        auto_split: bool,
        policy: CheckPolicy,
        bucket_ranges: Option<Vec<BucketRange>>,
    ) -> Self {
        Task::SplitCheckTask {
            region,
            start_key: None,
            end_key: None,
            auto_split,
            policy,
            bucket_ranges,
        }
    }

    pub fn split_check_key_range(
        region: Region,
        start_key: Option<Vec<u8>>,
        end_key: Option<Vec<u8>>,
        auto_split: bool,
        policy: CheckPolicy,
        bucket_ranges: Option<Vec<BucketRange>>,
    ) -> Self {
        Task::SplitCheckTask {
            region,
            start_key,
            end_key,
            auto_split,
            policy,
            bucket_ranges,
        }
    }
}

impl<EK> Display for Task<EK>
where

Review comment (suggestion):
Could `SplitReason::Load` be `auto_split`?

Perhaps wrapping a function for the split `reason` generating and integrate the comments together.

Code after:
    ChangeConfig(ConfigChange),
    // Compaction finished event
    CompactedEvent {
        event: EK::CompactedEvent,
        region_split_check_diff: u64,
    },
    #[cfg(any(test, feature = "testexport"))]
    Validate(Box<dyn FnOnce(&Config) + Send>),
}

impl<EK> Task<EK>
where
    EK: KvEngine,
{
    pub fn split_check(
        region: Region,
        auto_split: bool,
        policy: CheckPolicy,
        bucket_ranges: Option<Vec<BucketRange>>,
    ) -> Self {
        Task::SplitCheckTask {
            region,
            start_key: None,
            end_key: None,
            split_reason: if auto_split {
                SplitReason::Size
            } else {
                SplitReason::Admin
            },
            policy,
            bucket_ranges,
        }
    }

    pub fn split_check_key_range(
        region: Region,
        start_key: Option<Vec<u8>>,
        end_key: Option<Vec<u8>>,
        split_reason: SplitReason,
        policy: CheckPolicy,
        bucket_ranges: Option<Vec<BucketRange>>,
    ) -> Self {
        Task::SplitCheckTask {
            region,
            start_key,
            end_key,
            split_reason,
            policy,
            bucket_ranges,
        }
    }
Repository: tikv/tikv
File: tests/integrations/raftstore/test_region_heartbeat.rs (Rust)

Code before:
    // add a new peer to trigger heartbeat pd.
    pd_client.must_add_peer(region_id, new_peer(3, 3));
    test_util::eventually(
        Duration::from_millis(100),
        Duration::from_millis(3000),
        || {
            let size = cluster
                .pd_client
                .get_region_approximate_size(region_id)
                .unwrap_or_default();
            size >= 2000
        },
    );

    // transfer leader back to the old leader and the region approximate size
    // should not change.
    cluster.must_transfer_leader(region_id, new_peer(1, 1));
    sleep_ms(100);
    // the first region heartbeat after leader transfer will report with 0 value.
    // Real pd with ignore this 0 value while the current test pd-client just keep
    // this 0 value.
    let size = cluster
        .pd_client
        .get_region_approximate_size(region_id)
        .unwrap_or_default();
    assert_eq!(size, 0);
}


Review comment (bug):
It's technically possible to let the test pd-client ignore the 0 value as well, right? If we do that, we will be able to assert `size >= 2000` in the end. But we will need to ensure that the new leader has uploaded its heartbeat (possibily using a failpoint).

Code after:
    // add a new peer to trigger heartbeat pd.
    pd_client.must_add_peer(region_id, new_peer(3, 3));
    test_util::eventually(
        Duration::from_millis(100),
        Duration::from_millis(3000),
        || {
            let size = cluster
                .pd_client
                .get_region_approximate_size(region_id)
                .unwrap_or_default();
            size >= 2000
        },
    );
    let approximate_size = cluster
        .pd_client
        .get_region_approximate_size(region_id)
        .unwrap_or_default();

    // transfer leader back to the old leader and the region approximate size
    // should not change.
    cluster.must_transfer_leader(region_id, new_peer(1, 1));
    sleep_ms(100);
    // after `must_transfer_leader` the new leader must have already sent a new
    // region heartbeat.
    let size = cluster
        .pd_client
        .get_region_approximate_size(region_id)
        .unwrap_or_default();
    assert_eq!(size, approximate_size);
}