2 files changed, 55 insertions, 40 deletions

diff --git a/log_db/src/engine.rs b/log_db/src/engine.rs
index 4391451..d028ee0 100644
--- a/log_db/src/engine.rs
+++ b/log_db/src/engine.rs
@@ -163,6 +163,9 @@ impl<R: Recordable> Engine<R> {
             for ForwardLogReaderItem { record, index } in
                 ForwardLogReader::new_with_index(metadata_file, data_file, from_index)
             {
+                // Validate that the values in the record are compatible with the schema
+                record.validate(&self.config.fields)?;
+
                 let log_key = LogKey::new(segnum, index);
 
                 if record.tombstone {
diff --git a/log_db/src/record.rs b/log_db/src/record.rs
index 6d7cbbf..ab9f8dc 100644
--- a/log_db/src/record.rs
+++ b/log_db/src/record.rs
@@ -49,57 +49,69 @@ impl Record {
         &self.values[index]
     }
 
-    pub fn validate<Field: Eq>(&self, schema: &Vec<(Field, Type)>) -> DBResult<()> {
-        // Validate the record length
-        if self.values.len() != schema.len() {
+    pub fn validate<Field: Eq>(&self, schema: &[(Field, Type)]) -> DBResult<()> {
+        // If there are more values than schema fields, it's an error.
+        if self.values.len() > schema.len() {
             return Err(DBError::ValidationError(format!(
-                "Record has an incorrect number of fields: {}, expected {}",
+                "Record has more fields ({}) than expected by the schema ({})",
                 self.values.len(),
                 schema.len()
             )));
         }
 
-        // Validate that record fields match schema types
-        for (i, (_, field)) in schema.iter().enumerate() {
-            match (&self.values[i], field) {
-                (
-                    Value::Null,
-                    Type {
-                        nullable: true,
-                        primitive: _,
-                    },
-                ) => {}
-                (
-                    Value::Int(_),
-                    Type {
-                        primitive: PrimitiveType::Int,
-                        ..
-                    },
-                ) => {}
-                (
-                    Value::String(_),
-                    Type {
-                        primitive: PrimitiveType::String,
-                        ..
-                    },
-                ) => {}
-                (
-                    Value::Bytes(_),
-                    Type {
-                        primitive: PrimitiveType::Bytes,
-                        ..
-                    },
-                ) => {}
-                _ => {
-                    return Err(DBError::ValidationError(format!(
-                        "Record field {} has incorrect type: {:?}, expected {:?}",
-                        &i, &self.values[i], &field.primitive
-                    )));
+        for (i, (_, typ)) in schema.iter().enumerate() {
+            match self.values.get(i) {
+                Some(value) => {
+                    if !Self::value_matches_type(value, typ) {
+                        return Err(DBError::ValidationError(format!(
+                            "Record field {} has incorrect type: {:?}, expected {:?}",
+                            i, value, typ.primitive
+                        )));
+                    }
+                }
+                // If the field is missing from the record...
+                None => {
+                    // ...it's allowed only if the schema says the field is nullable.
+                    if !typ.nullable {
+                        return Err(DBError::ValidationError(format!(
+                            "Record is missing field expected by the schema ({:?}) at index {}",
+                            typ, i
+                        )));
+                    }
                 }
             }
         }
+
         Ok(())
     }
+
+    fn value_matches_type(value: &Value, typ: &Type) -> bool {
+        match (value, typ) {
+            (Value::Null, Type { nullable: true, .. }) => true,
+            (
+                Value::Int(_),
+                Type {
+                    primitive: PrimitiveType::Int,
+                    ..
+                },
+            ) => true,
+            (
+                Value::String(_),
+                Type {
+                    primitive: PrimitiveType::String,
+                    ..
+                },
+            ) => true,
+            (
+                Value::Bytes(_),
+                Type {
+                    primitive: PrimitiveType::Bytes,
+                    ..
+                },
+            ) => true,
+            _ => false,
+        }
+    }
 }
 
 /// A trait that describes how to convert a data structure into a database record and vice versa.