Repository: risingwavelabs/risingwave
File: src/common/src/gap_fill.rs (Rust)

Code before:
        }
        (ScalarRefImpl::Float64(v1), ScalarRefImpl::Float64(v2)) => {
            Some(ScalarImpl::Float64((v2 - v1) / steps as f64))
        }
        (ScalarRefImpl::Decimal(v1), ScalarRefImpl::Decimal(v2)) => {
            Some(ScalarImpl::Decimal((v2 - v1) / Decimal::from(steps)))
        }
        _ => None,
    }
}

/// Mutates the `current` datum by adding the value of `step` to it.
///
/// This function is used during the interpolation process in gap filling,
/// where it incrementally updates the datum to generate intermediate values
/// between known data points.
pub fn apply_interpolation_step(current: &mut Datum, step: &ScalarImpl) {
    if let Some(curr) = current.as_mut() {
        match (curr, step) {
            (ScalarImpl::Int16(v1), ScalarImpl::Int16(v2)) => *v1 += v2,
            (ScalarImpl::Int32(v1), ScalarImpl::Int32(v2)) => *v1 += v2,
            (ScalarImpl::Int64(v1), ScalarImpl::Int64(v2)) => *v1 += v2,
            (ScalarImpl::Float32(v1), ScalarImpl::Float32(v2)) => *v1 += v2,
            (ScalarImpl::Float64(v1), ScalarImpl::Float64(v2)) => *v1 += v2,
            (ScalarImpl::Decimal(v1), ScalarImpl::Decimal(v2)) => *v1 = *v1 + *v2,
            _ => (),
        }
    }
}


Review comment (bug):
Another issue with this approach for floating points. Consider if our gap has the following parameters:
```
start: 0.1
steps: 10
end: 0.2
```

We can write the following example:
```
fn main() {
  let start = 0.1;
  let end = 0.2;
  let steps: f64 = 10.0;
  let step_size = (end - start) / steps;
  {
      let mut end = start;
      for _ in 0..10 {
          end += step_size;
      }
      println!("end: {end}");     
  }

}
```

Because of binary precision issues, we will get the following output for addition:
```
end: 0.20000000000000007
```

You can see it will overflow the end bounds. For the last step let's make sure to clamp it to the end bound. You can do this in a separate PR.

Code after:
        }
        (ScalarRefImpl::Float64(v1), ScalarRefImpl::Float64(v2)) => {
            Some(ScalarImpl::Float64((v2 - v1) / steps as f64))
        }
        (ScalarRefImpl::Decimal(v1), ScalarRefImpl::Decimal(v2)) => {
            Some(ScalarImpl::Decimal((v2 - v1) / Decimal::from(steps)))
        }
        _ => None,
    }
}

/// Mutates the `current` datum by adding the value of `step` to it.
///
/// This function is used during the interpolation process in gap filling,
/// where it incrementally updates the datum to generate intermediate values
/// between known data points.
pub fn apply_interpolation_step(current: &mut Datum, step: &ScalarImpl) {
    if let Some(curr) = current.as_mut() {
        match (curr, step) {
            (ScalarImpl::Int16(v1), &ScalarImpl::Int16(v2)) => *v1 += v2,
            (ScalarImpl::Int32(v1), &ScalarImpl::Int32(v2)) => *v1 += v2,
            (ScalarImpl::Int64(v1), &ScalarImpl::Int64(v2)) => *v1 += v2,
            (ScalarImpl::Float32(v1), &ScalarImpl::Float32(v2)) => *v1 += v2,
            (ScalarImpl::Float64(v1), &ScalarImpl::Float64(v2)) => *v1 += v2,
            (ScalarImpl::Decimal(v1), &ScalarImpl::Decimal(v2)) => *v1 = *v1 + v2,
            _ => (),
        }
    }
}