Interval Arithmetic Updates (#8276)

* Interval lib can be accessible from logical plan

* committed to merge precision PR

* minor fix after merge, adding ts-interval handling

* bound openness removed

* Remove all interval bound related code

* test fix

* fix docstrings

* Fix after merge

* Minor changes

* Simplifications

* Resolve linter errors

* Addressing reviews

* Fix win tests

* Update interval_arithmetic.rs

* Code simplifications

* Review Part 1

* Review Part 2

* Addressing Ozan's feedback

* Resolving conflicts

* Review Part 3

* Better cardinality calculation

* fix clippy

* Review Part 4

* type check, test polish, bug fix

* Constructs filter graph with datatypes

* Update Cargo.lock

* Update Cargo.toml

* Review

* Other expectations of AND

* OR operator implementation

* Certainly false asserting comparison operators

* Update interval_arithmetic.rs

* Tests added, is_superset renamed

* Final review

* Resolving conflicts

* Address review feedback

---------

Co-authored-by: Mustafa Akur <[email protected]>
Co-authored-by: Mehmet Ozan Kabak <[email protected]>

datafusion/common/Cargo.toml

@@ -38,14 +38,22 @@ backtrace = []
 pyarrow = ["pyo3", "arrow/pyarrow", "parquet"]
 
 [dependencies]
-ahash = { version = "0.8", default-features = false, features = ["runtime-rng"] }
-apache-avro = { version = "0.16", default-features = false, features = ["bzip", "snappy", "xz", "zstandard"], optional = true }
+ahash = { version = "0.8", default-features = false, features = [
+    "runtime-rng",
+] }
+apache-avro = { version = "0.16", default-features = false, features = [
+    "bzip",
+    "snappy",
+    "xz",
+    "zstandard",
+], optional = true }
 arrow = { workspace = true }
 arrow-array = { workspace = true }
 arrow-buffer = { workspace = true }
 arrow-schema = { workspace = true }
 chrono = { workspace = true }
 half = { version = "2.1", default-features = false }
+libc = "0.2.140"
 num_cpus = { workspace = true }
 object_store = { workspace = true, optional = true }
 parquet = { workspace = true, optional = true, default-features = true }

datafusion/common/src/lib.rs

@@ -34,6 +34,7 @@ pub mod file_options;
 pub mod format;
 pub mod hash_utils;
 pub mod parsers;
+pub mod rounding;
 pub mod scalar;
 pub mod stats;
 pub mod test_util;

datafusion/physical-expr/src/intervals/rounding.rs → datafusion/common/src/rounding.rs

@@ -22,8 +22,8 @@
 
 use std::ops::{Add, BitAnd, Sub};
 
-use datafusion_common::Result;
-use datafusion_common::ScalarValue;
+use crate::Result;
+use crate::ScalarValue;
 
 // Define constants for ARM
 #[cfg(all(target_arch = "aarch64", not(target_os = "windows")))]
@@ -162,7 +162,7 @@ impl FloatBits for f64 {
 /// # Examples
 ///
 /// ```
-/// use datafusion_physical_expr::intervals::rounding::next_up;
+/// use datafusion_common::rounding::next_up;
 ///
 /// let f: f32 = 1.0;
 /// let next_f = next_up(f);
@@ -195,7 +195,7 @@ pub fn next_up<F: FloatBits + Copy>(float: F) -> F {
 /// # Examples
 ///
 /// ```
-/// use datafusion_physical_expr::intervals::rounding::next_down;
+/// use datafusion_common::rounding::next_down;
 ///
 /// let f: f32 = 1.0;
 /// let next_f = next_down(f);

datafusion/common/src/scalar.rs

@@ -988,6 +988,48 @@ impl ScalarValue {
         Self::try_from_array(r.as_ref(), 0)
     }
 
+    /// Wrapping multiplication of `ScalarValue`
+    ///
+    /// NB: operating on `ScalarValue` directly is not efficient, performance sensitive code
+    /// should operate on Arrays directly, using vectorized array kernels.
+    pub fn mul<T: Borrow<ScalarValue>>(&self, other: T) -> Result<ScalarValue> {
+        let r = mul_wrapping(&self.to_scalar()?, &other.borrow().to_scalar()?)?;
+        Self::try_from_array(r.as_ref(), 0)
+    }
+
+    /// Checked multiplication of `ScalarValue`
+    ///
+    /// NB: operating on `ScalarValue` directly is not efficient, performance sensitive code
+    /// should operate on Arrays directly, using vectorized array kernels.
+    pub fn mul_checked<T: Borrow<ScalarValue>>(&self, other: T) -> Result<ScalarValue> {
+        let r = mul(&self.to_scalar()?, &other.borrow().to_scalar()?)?;
+        Self::try_from_array(r.as_ref(), 0)
+    }
+
+    /// Performs `lhs / rhs`
+    ///
+    /// Overflow or division by zero will result in an error, with exception to
+    /// floating point numbers, which instead follow the IEEE 754 rules.
+    ///
+    /// NB: operating on `ScalarValue` directly is not efficient, performance sensitive code
+    /// should operate on Arrays directly, using vectorized array kernels.
+    pub fn div<T: Borrow<ScalarValue>>(&self, other: T) -> Result<ScalarValue> {
+        let r = div(&self.to_scalar()?, &other.borrow().to_scalar()?)?;
+        Self::try_from_array(r.as_ref(), 0)
+    }
+
+    /// Performs `lhs % rhs`
+    ///
+    /// Overflow or division by zero will result in an error, with exception to
+    /// floating point numbers, which instead follow the IEEE 754 rules.
+    ///
+    /// NB: operating on `ScalarValue` directly is not efficient, performance sensitive code
+    /// should operate on Arrays directly, using vectorized array kernels.
+    pub fn rem<T: Borrow<ScalarValue>>(&self, other: T) -> Result<ScalarValue> {
+        let r = rem(&self.to_scalar()?, &other.borrow().to_scalar()?)?;
+        Self::try_from_array(r.as_ref(), 0)
+    }
+
     pub fn is_unsigned(&self) -> bool {
         matches!(
             self,

datafusion/expr/Cargo.toml

@@ -35,10 +35,13 @@ path = "src/lib.rs"
 [features]
 
 [dependencies]
-ahash = { version = "0.8", default-features = false, features = ["runtime-rng"] }
+ahash = { version = "0.8", default-features = false, features = [
+    "runtime-rng",
+] }
 arrow = { workspace = true }
 arrow-array = { workspace = true }
 datafusion-common = { workspace = true }
+paste = "^1.0"
 sqlparser = { workspace = true }
 strum = { version = "0.25.0", features = ["derive"] }
 strum_macros = "0.25.0"