feat(fuzz): add validator for inserted rows

tests-fuzz/Cargo.toml

@@ -23,6 +23,7 @@ common-query = { workspace = true }
 common-runtime = { workspace = true }
 common-telemetry = { workspace = true }
 common-time = { workspace = true }
+chrono = { workspace = true }
 datatypes = { workspace = true }
 derive_builder = { workspace = true }
 dotenv = "0.15"

tests-fuzz/src/generator/insert_expr.rs

@@ -93,11 +93,8 @@ impl<R: Rng + 'static> Generator<InsertIntoExpr, R> for InsertExprGenerator<R> {
 
         Ok(InsertIntoExpr {
             table_name: self.table_ctx.name.to_string(),
-            columns: if self.omit_column_list {
-                vec![]
-            } else {
-                values_columns
-            },
+            omit_column_list: self.omit_column_list,
+            columns: values_columns,
             values_list,
         })
     }

tests-fuzz/src/ir/insert_expr.rs

@@ -12,25 +12,36 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use std::fmt::Display;
+use std::fmt::{Debug, Display};
 
 use datatypes::value::Value;
 
 use crate::ir::Column;
 
 pub struct InsertIntoExpr {
     pub table_name: String,
+    pub omit_column_list: bool,
     pub columns: Vec<Column>,
     pub values_list: Vec<RowValues>,
 }
 
 pub type RowValues = Vec<RowValue>;
 
+#[derive(PartialEq, PartialOrd)]
 pub enum RowValue {
     Value(Value),
     Default,
 }
 
+impl RowValue {
+    pub fn cmp(&self, other: &Self) -> Option<std::cmp::Ordering> {
+        match (self, other) {
+            (RowValue::Value(v1), RowValue::Value(v2)) => v1.partial_cmp(v2),
+            _ => panic!("Invalid comparison: {:?} and {:?}", self, other),
+        }
+    }
+}
+
 impl Display for RowValue {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         match self {
@@ -46,3 +57,38 @@ impl Display for RowValue {
         }
     }
 }
+
+impl Debug for RowValue {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            RowValue::Value(v) => match v {
+                Value::Null => write!(f, "NULL"),
+                v @ (Value::String(_)
+                | Value::Timestamp(_)
+                | Value::DateTime(_)
+                | Value::Date(_)) => write!(f, "'{}'", v),
+                v => write!(f, "{}", v),
+            },
+            RowValue::Default => write!(f, "DEFAULT"),
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use common_time::Timestamp;
+    use datatypes::value::Value;
+
+    use crate::ir::insert_expr::RowValue;
+
+    #[test]
+    fn test_value_cmp() {
+        let time_stampe1 =
+            Value::Timestamp(Timestamp::from_str_utc("-39988-01-31 01:21:12.848697+0000").unwrap());
+        let time_stampe2 =
+            Value::Timestamp(Timestamp::from_str_utc("+12970-09-22 08:40:58.392839+0000").unwrap());
+        let v1 = RowValue::Value(time_stampe1);
+        let v2 = RowValue::Value(time_stampe2);
+        assert_eq!(v1.cmp(&v2), Some(std::cmp::Ordering::Less));
+    }
+}

tests-fuzz/src/translator/mysql/insert_expr.rs

@@ -33,7 +33,7 @@ impl DslTranslator<InsertIntoExpr, String> for InsertIntoExprTranslator {
 
 impl InsertIntoExprTranslator {
     fn format_columns(input: &InsertIntoExpr) -> String {
-        if input.columns.is_empty() {
+        if input.omit_column_list {
             "".to_string()
         } else {
             let list = input

tests-fuzz/src/validator.rs

@@ -13,3 +13,4 @@
 // limitations under the License.
 
 pub mod column;
+pub mod row;

tests-fuzz/src/validator/row.rs

@@ -0,0 +1,144 @@
+// Copyright 2023 Greptime Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use chrono::{DateTime, Utc};
+use common_time::Timestamp;
+use datatypes::value::Value;
+use snafu::{ensure, ResultExt};
+use sqlx::database::HasArguments;
+use sqlx::{
+    Column, ColumnIndex, Database, Decode, Encode, Executor, IntoArguments, Row, Type, TypeInfo,
+    ValueRef,
+};
+
+use crate::error::{self, Result};
+use crate::ir::insert_expr::{RowValue, RowValues};
+
+/// Asserts fetched_rows are equal to rows
+pub fn assert_eq<'a, DB>(
+    fetched_rows: &'a [<DB as Database>::Row],
+    rows: &[RowValues],
+) -> Result<()>
+where
+    DB: Database,
+    usize: ColumnIndex<<DB as Database>::Row>,
+    bool: sqlx::Type<DB> + sqlx::Decode<'a, DB>,
+    i8: sqlx::Type<DB> + sqlx::Decode<'a, DB>,
+    i16: sqlx::Type<DB> + sqlx::Decode<'a, DB>,
+    i32: sqlx::Type<DB> + sqlx::Decode<'a, DB>,
+    i64: sqlx::Type<DB> + sqlx::Decode<'a, DB>,
+    f32: sqlx::Type<DB> + sqlx::Decode<'a, DB>,
+    f64: sqlx::Type<DB> + sqlx::Decode<'a, DB>,
+    String: sqlx::Type<DB> + sqlx::Decode<'a, DB>,
+    Vec<u8>: sqlx::Type<DB> + sqlx::Decode<'a, DB>,
+    DateTime<Utc>: sqlx::Type<DB> + sqlx::Decode<'a, DB>,
+{
+    ensure!(
+        fetched_rows.len() == rows.len(),
+        error::AssertSnafu {
+            reason: format!(
+                "Expected values length: {}, got: {}",
+                rows.len(),
+                fetched_rows.len(),
+            )
+        }
+    );
+
+    for (idx, fetched_row) in fetched_rows.iter().enumerate() {
+        let row = &rows[idx];
+
+        ensure!(
+            fetched_row.len() == row.len(),
+            error::AssertSnafu {
+                reason: format!(
+                    "Expected row length: {}, got: {}",
+                    row.len(),
+                    fetched_row.len(),
+                )
+            }
+        );
+
+        for (idx, value) in row.iter().enumerate() {
+            let fetched_value = if fetched_row.try_get_raw(idx).unwrap().is_null() {
+                RowValue::Value(Value::Null)
+            } else {
+                let value_type = fetched_row.column(idx).type_info().name();
+                match value_type {
+                    "BOOL" | "BOOLEAN" => RowValue::Value(Value::Boolean(
+                        fetched_row.try_get::<bool, usize>(idx).unwrap(),
+                    )),
+                    "TINYINT" => {
+                        RowValue::Value(Value::Int8(fetched_row.try_get::<i8, usize>(idx).unwrap()))
+                    }
+                    "SMALLINT" => RowValue::Value(Value::Int16(
+                        fetched_row.try_get::<i16, usize>(idx).unwrap(),
+                    )),
+                    "INT" => RowValue::Value(Value::Int32(
+                        fetched_row.try_get::<i32, usize>(idx).unwrap(),
+                    )),
+                    "BIGINT" => RowValue::Value(Value::Int64(
+                        fetched_row.try_get::<i64, usize>(idx).unwrap(),
+                    )),
+                    "FLOAT" => RowValue::Value(Value::Float32(datatypes::value::OrderedFloat(
+                        fetched_row.try_get::<f32, usize>(idx).unwrap(),
+                    ))),
+                    "DOUBLE" => RowValue::Value(Value::Float64(datatypes::value::OrderedFloat(
+                        fetched_row.try_get::<f64, usize>(idx).unwrap(),
+                    ))),
+                    "VARCHAR" | "CHAR" | "TEXT" => RowValue::Value(Value::String(
+                        fetched_row.try_get::<String, usize>(idx).unwrap().into(),
+                    )),
+                    "VARBINARY" | "BINARY" | "BLOB" => RowValue::Value(Value::Binary(
+                        fetched_row.try_get::<Vec<u8>, usize>(idx).unwrap().into(),
+                    )),
+                    "TIMESTAMP" => RowValue::Value(Value::Timestamp(
+                        Timestamp::from_chrono_datetime(
+                            fetched_row
+                                .try_get::<DateTime<Utc>, usize>(idx)
+                                .unwrap()
+                                .naive_utc(),
+                        )
+                        .unwrap(),
+                    )),
+                    _ => panic!("Unsupported type: {}", value_type),
+                }
+            };
+
+            // println!("Expected value: {:?}, got: {:?}", value, fetched_value);
+            ensure!(
+                value == &fetched_value,
+                error::AssertSnafu {
+                    reason: format!("Expected value: {:?}, got: {:?}", value, fetched_value)
+                }
+            )
+        }
+    }
+
+    Ok(())
+}
+
+/// Returns all [RowEntry] of the `table_name`.
+pub async fn fetch_values<'a, DB, E>(e: E, sql: &'a str) -> Result<Vec<<DB as Database>::Row>>
+where
+    DB: Database,
+    <DB as HasArguments<'a>>::Arguments: IntoArguments<'a, DB>,
+    for<'c> E: 'a + Executor<'c, Database = DB>,
+    for<'c> String: Decode<'c, DB> + Type<DB>,
+    for<'c> String: Encode<'c, DB> + Type<DB>,
+{
+    sqlx::query(sql)
+        .fetch_all(e)
+        .await
+        .context(error::ExecuteQuerySnafu { sql })
+}

tests-fuzz/targets/fuzz_insert.rs

@@ -37,6 +37,7 @@ use tests_fuzz::translator::mysql::create_expr::CreateTableExprTranslator;
 use tests_fuzz::translator::mysql::insert_expr::InsertIntoExprTranslator;
 use tests_fuzz::translator::DslTranslator;
 use tests_fuzz::utils::{init_greptime_connections_via_env, Connections};
+use tests_fuzz::validator;
 
 struct FuzzContext {
     greptime: Pool<MySql>,
@@ -135,7 +136,37 @@ async fn execute_insert(ctx: FuzzContext, input: FuzzInput) -> Result<()> {
         }
     );
 
-    // TODO: Validate inserted rows
+    // Validate inserted rows
+    let ts_column_idx = create_expr
+        .columns
+        .iter()
+        .position(|c| c.is_time_index())
+        .unwrap();
+    let ts_column_name = create_expr.columns[ts_column_idx].name.clone();
+    let ts_column_idx_in_insert = insert_expr
+        .columns
+        .iter()
+        .position(|c| c.name == ts_column_name)
+        .unwrap();
+    let column_list = insert_expr
+        .columns
+        .iter()
+        .map(|c| c.name.to_string())
+        .collect::<Vec<_>>()
+        .join(", ")
+        .to_string();
+    let select_sql = format!(
+        "SELECT {} FROM {} ORDER BY {}",
+        column_list, create_expr.table_name, ts_column_name
+    );
+    let fetched_rows = validator::row::fetch_values(&ctx.greptime, select_sql.as_str()).await?;
+    let mut expected_rows = insert_expr.values_list;
+    expected_rows.sort_by(|a, b| {
+        a[ts_column_idx_in_insert]
+            .cmp(&b[ts_column_idx_in_insert])
+            .unwrap()
+    });
+    validator::row::assert_eq::<MySql>(&fetched_rows, &expected_rows)?;
 
     // Cleans up
     let sql = format!("DROP TABLE {}", create_expr.table_name);

tests-fuzz/targets/fuzz_insert_logical_table.rs

@@ -39,6 +39,7 @@ use tests_fuzz::translator::mysql::create_expr::CreateTableExprTranslator;
 use tests_fuzz::translator::mysql::insert_expr::InsertIntoExprTranslator;
 use tests_fuzz::translator::DslTranslator;
 use tests_fuzz::utils::{init_greptime_connections_via_env, Connections};
+use tests_fuzz::validator;
 
 struct FuzzContext {
     greptime: Pool<MySql>,
@@ -160,7 +161,39 @@ async fn execute_insert(ctx: FuzzContext, input: FuzzInput) -> Result<()> {
         }
     );
 
-    // TODO: Validate inserted rows
+    // Validate inserted rows
+    let ts_column_idx = create_logical_table_expr
+        .columns
+        .iter()
+        .position(|c| c.is_time_index())
+        .unwrap();
+    let ts_column_name = create_logical_table_expr.columns[ts_column_idx]
+        .name
+        .clone();
+    let ts_column_idx_in_insert = insert_expr
+        .columns
+        .iter()
+        .position(|c| c.name == ts_column_name)
+        .unwrap();
+    let column_list = insert_expr
+        .columns
+        .iter()
+        .map(|c| c.name.to_string())
+        .collect::<Vec<_>>()
+        .join(", ")
+        .to_string();
+    let select_sql = format!(
+        "SELECT {} FROM {} ORDER BY {}",
+        column_list, create_logical_table_expr.table_name, ts_column_name
+    );
+    let fetched_rows = validator::row::fetch_values(&ctx.greptime, select_sql.as_str()).await?;
+    let mut expected_rows = insert_expr.values_list;
+    expected_rows.sort_by(|a, b| {
+        a[ts_column_idx_in_insert]
+            .cmp(&b[ts_column_idx_in_insert])
+            .unwrap()
+    });
+    validator::row::assert_eq::<MySql>(&fetched_rows, &expected_rows)?;
 
     // Clean up logical table
     let sql = format!("DROP TABLE {}", create_logical_table_expr.table_name);