Added unit test cases

Author: niknegi

crates/fluss/src/metadata/datatype.rs

@@ -497,6 +497,10 @@ impl DecimalType {
         self.scale
     }
 
+    pub fn is_nullable(&self) -> bool {
+        self.nullable
+    }
+
     pub fn as_non_nullable(&self) -> Self {
         Self::with_nullable(false, self.precision, self.scale)
             .expect("Invalid decimal precision or scale")
@@ -594,6 +598,10 @@ impl TimeType {
         self.precision
     }
 
+    pub fn is_nullable(&self) -> bool {
+        self.nullable
+    }
+
     pub fn as_non_nullable(&self) -> Self {
         Self::with_nullable(false, self.precision).expect("Invalid time precision")
     }
@@ -655,6 +663,10 @@ impl TimestampType {
         self.precision
     }
 
+    pub fn is_nullable(&self) -> bool {
+        self.nullable
+    }
+
     pub fn as_non_nullable(&self) -> Self {
         Self::with_nullable(false, self.precision).expect("Invalid timestamp precision")
     }
@@ -717,6 +729,10 @@ impl TimestampLTzType {
         self.precision
     }
 
+    pub fn is_nullable(&self) -> bool {
+        self.nullable
+    }
+
     pub fn as_non_nullable(&self) -> Self {
         Self::with_nullable(false, self.precision)
             .expect("Invalid timestamp with local time zone precision")
@@ -1405,6 +1421,11 @@ fn test_deeply_nested_types() {
     assert_eq!(nested.to_string(), "ARRAY<MAP<STRING, ROW<x INT, y INT>>>");
 }
 
+
+// ============================================================================
+// DecimalType validation tests
+// ============================================================================
+
 #[test]
 fn test_decimal_invalid_precision() {
     // DecimalType::with_nullable should return an error for invalid precision
@@ -1431,6 +1452,65 @@ fn test_decimal_invalid_scale() {
     );
 }
 
+// ============================================================================
+// DecimalType validation tests - edge cases
+// ============================================================================
+
+#[test]
+fn test_decimal_valid_precision_and_scale() {
+    // Valid: precision=10, scale=2
+    let result = DecimalType::with_nullable(true, 10, 2);
+    assert!(result.is_ok());
+    let decimal = result.unwrap();
+    assert_eq!(decimal.precision(), 10);
+    assert_eq!(decimal.scale(), 2);
+    assert!(decimal.is_nullable());
+
+    // Valid: precision=38, scale=0
+    let result = DecimalType::with_nullable(true, 38, 0);
+    assert!(result.is_ok());
+    let decimal = result.unwrap();
+    assert_eq!(decimal.precision(), 38);
+    assert_eq!(decimal.scale(), 0);
+
+    // Valid: precision=1, scale=0
+    let result = DecimalType::with_nullable(false, 1, 0);
+    assert!(result.is_ok());
+    let decimal = result.unwrap();
+    assert_eq!(decimal.precision(), 1);
+    assert_eq!(decimal.scale(), 0);
+    assert!(!decimal.is_nullable());
+}
+
+#[test]
+fn test_decimal_invalid_precision_zero() {
+    // Invalid: precision=0 (edge case not covered by existing tests)
+    let result = DecimalType::with_nullable(true, 0, 0);
+    assert!(result.is_err());
+    assert!(
+        result
+            .unwrap_err()
+            .to_string()
+            .contains("Decimal precision must be between 1 and 38")
+    );
+}
+
+
+// ============================================================================
+// TimeType validation tests 
+// ============================================================================
+
+#[test]
+fn test_time_valid_precision() {
+    // Test all valid precision values 0 through 9
+    for precision in 0..=9 {
+        let result = TimeType::with_nullable(true, precision);
+        assert!(result.is_ok(), "precision {} should be valid", precision);
+        let time = result.unwrap();
+        assert_eq!(time.precision(), precision);
+    }
+}
+
 #[test]
 fn test_time_invalid_precision() {
     // TimeType::with_nullable should return an error for invalid precision
@@ -1444,6 +1524,25 @@ fn test_time_invalid_precision() {
     );
 }
 
+// ============================================================================
+// TimestampType validation tests 
+// ============================================================================
+
+#[test]
+fn test_timestamp_valid_precision() {
+    // Test all valid precision values 0 through 9
+    for precision in 0..=9 {
+        let result = TimestampType::with_nullable(true, precision);
+        assert!(
+            result.is_ok(),
+            "precision {} should be valid",
+            precision
+        );
+        let timestamp_type = result.unwrap();
+        assert_eq!(timestamp_type.precision(), precision);
+    }
+}
+
 #[test]
 fn test_timestamp_invalid_precision() {
     // TimestampType::with_nullable should return an error for invalid precision
@@ -1469,3 +1568,79 @@ fn test_timestamp_ltz_invalid_precision() {
             .contains("Timestamp with local time zone precision must be between 0 and 9")
     );
 }
+
+// ============================================================================
+// RowType projection tests
+// ============================================================================
+
+#[test]
+fn test_row_type_project_valid_indices() {
+    // Create a 3-column row type
+    let row_type = RowType::with_data_types_and_field_names(
+        vec![DataTypes::int(), DataTypes::string(), DataTypes::bigint()],
+        vec!["id", "name", "age"],
+    );
+
+    // Valid projection by indices: [0, 2]
+    let projected = row_type.project(&[0, 2]).unwrap();
+    assert_eq!(projected.fields().len(), 2);
+    assert_eq!(projected.fields()[0].name, "id");
+    assert_eq!(projected.fields()[1].name, "age");
+}
+
+#[test]
+fn test_row_type_project_with_field_names_valid() {
+    // Create a 3-column row type
+    let row_type = RowType::with_data_types_and_field_names(
+        vec![DataTypes::int(), DataTypes::string(), DataTypes::bigint()],
+        vec!["id", "name", "age"],
+    );
+
+    // Valid projection by names: ["id", "name"]
+    let projected = row_type
+        .project_with_field_names(&["id".to_string(), "name".to_string()])
+        .unwrap();
+    assert_eq!(projected.fields().len(), 2);
+    assert_eq!(projected.fields()[0].name, "id");
+    assert_eq!(projected.fields()[1].name, "name");
+}
+
+#[test]
+fn test_row_type_project_index_out_of_bounds() {
+    // Create a 3-column row type
+    let row_type = RowType::with_data_types_and_field_names(
+        vec![DataTypes::int(), DataTypes::string(), DataTypes::bigint()],
+        vec!["id", "name", "age"],
+    );
+
+    // Error: index out of bounds
+    let result = row_type.project(&[0, 5]);
+    assert!(result.is_err());
+    assert!(
+        result.unwrap_err().to_string().contains("invalid field position: 5")
+    );
+}
+
+#[test]
+fn test_row_type_project_with_field_names_nonexistent() {
+    // Create a 3-column row type
+    let row_type = RowType::with_data_types_and_field_names(
+        vec![DataTypes::int(), DataTypes::string(), DataTypes::bigint()],
+        vec!["id", "name", "age"],
+    );
+
+    // Error: non-existent field name
+    // Note: project_with_field_names filters out non-existent field names using filter_map,
+    // so the result will be an empty row if all field names are non-existent
+    let projected = row_type
+        .project_with_field_names(&["nonexistent".to_string()])
+        .unwrap();
+    assert_eq!(projected.fields().len(), 0);
+
+    // Mixed existing and non-existing: only existing fields are projected
+    let projected = row_type
+        .project_with_field_names(&["id".to_string(), "nonexistent".to_string()])
+        .unwrap();
+    assert_eq!(projected.fields().len(), 1);
+    assert_eq!(projected.fields()[0].name, "id");
+}