test: Introduce integration test framework.

Cargo.toml

@@ -22,6 +22,7 @@ members = [
   "crates/examples",
   "crates/iceberg",
   "crates/integrations/*",
+  "crates/sqllogictests",
   "crates/test_utils",
 ]
 exclude = ["bindings/python"]
@@ -39,6 +40,7 @@ rust-version = "1.77.1"
 anyhow = "1.0.72"
 apache-avro = "0.17"
 array-init = "2"
+arrow = { version = "52" }
 arrow-arith = { version = "52" }
 arrow-array = { version = "52" }
 arrow-ord = { version = "52" }
@@ -56,6 +58,8 @@ bytes = "1.5"
 chrono = "0.4.34"
 ctor = "0.2.8"
 derive_builder = "0.20"
+datafusion = { version = "41.0.0" }
+datafusion-common = { version = "41.0.0" }
 either = "1"
 env_logger = "0.11.0"
 fnv = "1"
@@ -84,6 +88,7 @@ serde_derive = "1"
 serde_json = "1"
 serde_repr = "0.1.16"
 serde_with = "3.4"
+sqlparser = { version = "0.50.0", features = ["visitor"] }
 tempfile = "3.8"
 tokio = { version = "1", default-features = false }
 typed-builder = "0.19"

crates/iceberg/src/writer/file_writer/track_writer.rs

@@ -42,10 +42,9 @@ impl TrackWriter {
 impl FileWrite for TrackWriter {
     async fn write(&mut self, bs: Bytes) -> Result<()> {
         let size = bs.len();
-        self.inner.write(bs).await.map(|v| {
+        self.inner.write(bs).await.inspect(|_| {
             self.written_size
                 .fetch_add(size as i64, std::sync::atomic::Ordering::Relaxed);
-            v
         })
     }
 

crates/integrations/datafusion/Cargo.toml

@@ -31,7 +31,7 @@ keywords = ["iceberg", "integrations", "datafusion"]
 [dependencies]
 anyhow = { workspace = true }
 async-trait = { workspace = true }
-datafusion = { version = "41.0.0" }
+datafusion = { workspace = true }
 futures = { workspace = true }
 iceberg = { workspace = true }
 tokio = { workspace = true }

crates/sqllogictests/Cargo.toml

@@ -0,0 +1,38 @@
+[package]
+name = "sqllogictests"
+version = { workspace = true }
+edition = { workspace = true }
+homepage = { workspace = true }
+repository = { workspace = true }
+license = { workspace = true }
+rust-version = { workspace = true }
+
+[dependencies]
+anyhow = { workspace = true }
+arrow = { workspace = true }
+# For spark-connect-rs
+arrow_51 = { version = "51", package = "arrow" }
+async-trait = { workspace = true }
+bigdecimal = "0.4.1"
+datafusion = { workspace = true, default-features = true }
+datafusion-common = { workspace = true, default-features = true }
+env_logger = { workspace = true }
+half = "2.4.1"
+iceberg-catalog-rest = { path = "../catalog/rest" }
+iceberg-datafusion = { path = "../integrations/datafusion" }
+itertools = "0.13.0"
+log = "0.4.22"
+rust_decimal = { version = "1.27.0" }
+spark-connect-rs = "0.0.1-beta.5"
+sqllogictest = "0.22"
+tokio = "1.38.0"
+toml = "0.8.19"
+
+[dev-dependencies]
+iceberg_test_utils = { path = "../test_utils", features = ["tests"] }
+libtest-mimic = "0.7.3"
+
+[[test]]
+harness = false
+name = "sqllogictests"
+path = "tests/sqllogictests.rs"

crates/sqllogictests/src/display/conversion.rs

@@ -0,0 +1,82 @@
+use arrow::array::types::{Decimal128Type, Decimal256Type, DecimalType};
+use arrow::datatypes::i256;
+use bigdecimal::BigDecimal;
+use half::f16;
+use rust_decimal::prelude::*;
+
+/// Represents a constant for NULL string in your database.
+pub const NULL_STR: &str = "NULL";
+
+pub(crate) fn bool_to_str(value: bool) -> String {
+    if value {
+        "true".to_string()
+    } else {
+        "false".to_string()
+    }
+}
+
+pub(crate) fn varchar_to_str(value: &str) -> String {
+    if value.is_empty() {
+        "(empty)".to_string()
+    } else {
+        value.trim_end_matches('\n').to_string()
+    }
+}
+
+pub(crate) fn f16_to_str(value: f16) -> String {
+    if value.is_nan() {
+        // The sign of NaN can be different depending on platform.
+        // So the string representation of NaN ignores the sign.
+        "NaN".to_string()
+    } else if value == f16::INFINITY {
+        "Infinity".to_string()
+    } else if value == f16::NEG_INFINITY {
+        "-Infinity".to_string()
+    } else {
+        big_decimal_to_str(BigDecimal::from_str(&value.to_string()).unwrap())
+    }
+}
+
+pub(crate) fn f32_to_str(value: f32) -> String {
+    if value.is_nan() {
+        // The sign of NaN can be different depending on platform.
+        // So the string representation of NaN ignores the sign.
+        "NaN".to_string()
+    } else if value == f32::INFINITY {
+        "Infinity".to_string()
+    } else if value == f32::NEG_INFINITY {
+        "-Infinity".to_string()
+    } else {
+        big_decimal_to_str(BigDecimal::from_str(&value.to_string()).unwrap())
+    }
+}
+
+pub(crate) fn f64_to_str(value: f64) -> String {
+    if value.is_nan() {
+        // The sign of NaN can be different depending on platform.
+        // So the string representation of NaN ignores the sign.
+        "NaN".to_string()
+    } else if value == f64::INFINITY {
+        "Infinity".to_string()
+    } else if value == f64::NEG_INFINITY {
+        "-Infinity".to_string()
+    } else {
+        big_decimal_to_str(BigDecimal::from_str(&value.to_string()).unwrap())
+    }
+}
+
+pub(crate) fn i128_to_str(value: i128, precision: &u8, scale: &i8) -> String {
+    big_decimal_to_str(
+        BigDecimal::from_str(&Decimal128Type::format_decimal(value, *precision, *scale)).unwrap(),
+    )
+}
+
+pub(crate) fn i256_to_str(value: i256, precision: &u8, scale: &i8) -> String {
+    big_decimal_to_str(
+        BigDecimal::from_str(&Decimal256Type::format_decimal(value, *precision, *scale)).unwrap(),
+    )
+}
+
+pub(crate) fn big_decimal_to_str(value: BigDecimal) -> String {
+    value.round(12).normalized().to_string()
+}

crates/sqllogictests/src/display/conversion_51.rs

@@ -0,0 +1,82 @@
+use arrow_51::array::types::{Decimal128Type, Decimal256Type, DecimalType};
+use arrow_51::datatypes::i256;
+use bigdecimal::BigDecimal;
+use half::f16;
+use rust_decimal::prelude::*;
+
+/// Represents a constant for NULL string in your database.
+pub const NULL_STR: &str = "NULL";
+
+pub(crate) fn bool_to_str(value: bool) -> String {
+    if value {
+        "true".to_string()
+    } else {
+        "false".to_string()
+    }
+}
+
+pub(crate) fn varchar_to_str(value: &str) -> String {
+    if value.is_empty() {
+        "(empty)".to_string()
+    } else {
+        value.trim_end_matches('\n').to_string()
+    }
+}
+
+pub(crate) fn f16_to_str(value: f16) -> String {
+    if value.is_nan() {
+        // The sign of NaN can be different depending on platform.
+        // So the string representation of NaN ignores the sign.
+        "NaN".to_string()
+    } else if value == f16::INFINITY {
+        "Infinity".to_string()
+    } else if value == f16::NEG_INFINITY {
+        "-Infinity".to_string()
+    } else {
+        big_decimal_to_str(BigDecimal::from_str(&value.to_string()).unwrap())
+    }
+}
+
+pub(crate) fn f32_to_str(value: f32) -> String {
+    if value.is_nan() {
+        // The sign of NaN can be different depending on platform.
+        // So the string representation of NaN ignores the sign.
+        "NaN".to_string()
+    } else if value == f32::INFINITY {
+        "Infinity".to_string()
+    } else if value == f32::NEG_INFINITY {
+        "-Infinity".to_string()
+    } else {
+        big_decimal_to_str(BigDecimal::from_str(&value.to_string()).unwrap())
+    }
+}
+
+pub(crate) fn f64_to_str(value: f64) -> String {
+    if value.is_nan() {
+        // The sign of NaN can be different depending on platform.
+        // So the string representation of NaN ignores the sign.
+        "NaN".to_string()
+    } else if value == f64::INFINITY {
+        "Infinity".to_string()
+    } else if value == f64::NEG_INFINITY {
+        "-Infinity".to_string()
+    } else {
+        big_decimal_to_str(BigDecimal::from_str(&value.to_string()).unwrap())
+    }
+}
+
+pub(crate) fn i128_to_str(value: i128, precision: &u8, scale: &i8) -> String {
+    big_decimal_to_str(
+        BigDecimal::from_str(&Decimal128Type::format_decimal(value, *precision, *scale)).unwrap(),
+    )
+}
+
+pub(crate) fn i256_to_str(value: i256, precision: &u8, scale: &i8) -> String {
+    big_decimal_to_str(
+        BigDecimal::from_str(&Decimal256Type::format_decimal(value, *precision, *scale)).unwrap(),
+    )
+}
+
+pub(crate) fn big_decimal_to_str(value: BigDecimal) -> String {
+    value.round(12).normalized().to_string()
+}

crates/sqllogictests/src/display/mod.rs

@@ -0,0 +1,4 @@
+pub mod conversion;
+pub mod conversion_51;
+pub mod normalize;
+pub mod normalize_51;