diff --git a/Makefile b/Makefile
index 2ecdd2ba6ff..8b0fbc059b4 100644
--- a/Makefile
+++ b/Makefile
@@ -60,8 +60,7 @@ test:
 	@echo
 	@echo @ HA binpack test
 	@./ocaml/xapi/binpack
-#	The following test no longer runs:
-#	./ocaml/database/database_test
+	./ocaml/database/database_test
 #	The following test no longer compiles:
 #	./ocaml/xenops/device_number_test
 #	The following test must be run in dom0:
diff --git a/ocaml/database/OMakefile b/ocaml/database/OMakefile
index 0a46c00a49f..0e2d2a05cca 100644
--- a/ocaml/database/OMakefile
+++ b/ocaml/database/OMakefile
@@ -1,10 +1,11 @@
 
 OCAMLINCLUDES = ../idl/ocaml_backend ../xapi ../idl ../util ../autogen ..
-OCAMLPACKS    = xml-light2 stdext stunnel http-svr xcp sexpr rpclib uuid gzip xcp-inventory
+OCAMLPACKS    = xml-light2 stdext stunnel http-svr xcp sexpr rpclib uuid gzip xcp-inventory oUnit
 #OCAMLPPFLAGS  = -pp "camlp4o" 
 #OCAMLDEPFLAGS = -pp "camlp4o"
 
 UseCamlp4(rpclib.syntax, db_rpc_common_v2 db_cache_types db_filter_types)
+UseCamlp4(sexplib.syntax, schema)
 
 OCamlGeneratedFiles(db_filter_parse.ml db_filter_parse.mli db_filter_lex.ml)
 
@@ -19,8 +20,8 @@ BLOCK_DEVICE_IO_FILES = \
 OCamlProgram(block_device_io, $(BLOCK_DEVICE_IO_FILES))
 OCamlDocProgram(block_device_io, $(BLOCK_DEVICE_IO_FILES))
 
-DATABASE_SERVER_FILES = database_server_main test_schemas database_test ../autogen/db_actions 
-DATABASE_TEST_FILES = database_test database_test_main test_schemas
+DATABASE_SERVER_FILES = database_server_main test_schemas ../autogen/db_actions 
+DATABASE_TEST_FILES = db_cache_test test_schemas
 section:
 	#XXX there are lots of interdependencies which we should be aim to remove
 	OCAML_LIBS += ../util/version ../idl/ocaml_backend/common ../idl/ocaml_backend/client ../util/stats ../idl/ocaml_backend/server
diff --git a/ocaml/database/database_test.ml b/ocaml/database/database_test.ml
index f6afc22e91c..9e9e58e67bf 100644
--- a/ocaml/database/database_test.ml
+++ b/ocaml/database/database_test.ml
@@ -175,61 +175,61 @@ module Tests = functor(Client: Db_interface.DB_ACCESS) -> struct
 		let dump db g = 
 			let tables = Db_cache_types.Database.tableset db in
 			Db_cache_types.TableSet.fold_over_recent g 
-				(fun c u d name table acc ->
+				(fun name _ table acc ->
 					 Db_cache_types.Table.fold_over_recent g 
-						 (fun c u d r acc -> 
+                                                 (fun r { Db_cache_types.Stat.created; modified; deleted } _ acc -> 
 							  let s = 
 								  try 
 									  let row = Db_cache_types.Table.find r table in
 									  let s = Db_cache_types.Row.fold_over_recent g 
-										  (fun c u d k v acc ->
+										  (fun k _ v acc ->
 											   Printf.sprintf "%s %s=%s" acc k v) row "" in
 									  s 
 								  with _ -> "(deleted)"
 							  in
-							  Printf.printf "%s(%s): (%Ld %Ld %Ld) %s\n" name r c u d s;
-							  ())
-						 (fun () -> ()) table ()) tables ()
+							  Printf.printf "%s(%s): (%Ld %Ld %Ld) %s\n" name r created modified deleted s;
+							  ()
+						 ) table ())  tables ()
 		in
 
 		let get_created db g =
 			let tables = Db_cache_types.Database.tableset db in
 			Db_cache_types.TableSet.fold_over_recent g
-				(fun c u d name table acc ->
+				(fun name _ table acc ->
 					 Db_cache_types.Table.fold_over_recent g
-						 (fun c u d r acc ->
-							  if c>=g then (name,r)::acc else acc) ignore table acc
+                                                 (fun r { Db_cache_types.Stat.created } _ acc ->
+							  if created>=g then (name,r)::acc else acc) table acc
 				) tables []
 		in
 
 		let get_updated db g =
 			let tables = Db_cache_types.Database.tableset db in
 			Db_cache_types.TableSet.fold_over_recent g
-				(fun c u d name table acc ->
+				(fun name _ table acc ->
 					 Db_cache_types.Table.fold_over_recent g
-						 (fun c u d r acc ->
+						 (fun r _ _ acc ->
 							  let row = Db_cache_types.Table.find r table in
 							  Db_cache_types.Row.fold_over_recent g 
-								  (fun c u d k v acc ->
+								  (fun k _ v acc ->
 									   (r,(k,v))::acc) row acc)
-						 ignore table acc) tables []
+						 table acc) tables []
 		in
 
 		let get_deleted db g =
 			let tables = Db_cache_types.Database.tableset db in
 			Db_cache_types.TableSet.fold_over_recent g
-				(fun c u d name table acc ->
-					 Db_cache_types.Table.fold_over_recent g
-						 (fun c u d r acc ->
-							  if d > g then r::acc else acc)
-						 ignore table acc) tables []
+				(fun name _ table acc ->
+					 Db_cache_types.Table.fold_over_deleted g
+                                                 (fun r { Db_cache_types.Stat.deleted } acc ->
+							  if deleted > g then r::acc else acc)
+						 table acc) tables []
 		in
 
 		let get_max db =
 			let tables = Db_cache_types.Database.tableset db in
 			Db_cache_types.TableSet.fold_over_recent (-1L)
-				(fun c u d _ _ largest ->
-					 max c (max u (max d largest))) tables (-1L)
+                        (fun _ { Db_cache_types.Stat.created; modified; deleted } _ largest ->
+					 max created (max modified (max deleted largest))) tables (-1L)
 		in
 
 		let db = Db_ref.get_database t in
diff --git a/ocaml/database/db_action_helper.ml b/ocaml/database/db_action_helper.ml
index dc1783b08b9..96d911b3eb2 100644
--- a/ocaml/database/db_action_helper.ml
+++ b/ocaml/database/db_action_helper.ml
@@ -22,15 +22,11 @@ let events_notify ?(snapshot) ty op ref =
   match !__callback with
     | None -> ()
     | Some f -> f ?snapshot ty op ref
-	  
+	 (* 
 exception Db_set_or_map_parse_fail of string
   
 let parse_sexpr s : SExpr.t list =
   match SExpr_TS.of_string s with
     | SExpr.Node xs -> xs
     | _ -> raise (Db_set_or_map_parse_fail s)
-	
-let add_key_to_set key set =
-  if List.mem (SExpr.String key) set 
-  then set
-  else SExpr.String key :: set
+*)
diff --git a/ocaml/database/db_backend.ml b/ocaml/database/db_backend.ml
index 364fc8d227a..1e9bbaef811 100644
--- a/ocaml/database/db_backend.ml
+++ b/ocaml/database/db_backend.ml
@@ -42,27 +42,28 @@ let blow_away_non_persistent_fields (schema: Schema.t) db =
 	(* Generate a new row given a table schema *)
 	let row schema row : Row.t * int64 =
 		Row.fold 
-			(fun name created updated v (acc,max_upd) ->
+                        (fun name { Stat.created; modified } v (acc,max_upd) ->
 				try
 					let col = Schema.Table.find name schema in
-					let v',updated' = if col.Schema.Column.persistent then v,updated else col.Schema.Column.empty,g in
-					(Row.update updated' name "" (fun _ -> v') (Row.add created name v' acc),max max_upd updated')
+                                        let empty = col.Schema.Column.empty in
+					let v',modified' = if col.Schema.Column.persistent then v,modified else empty,g in
+					(Row.update modified' name empty (fun _ -> v') (Row.add created name v' acc),max max_upd modified')
 				with Not_found ->
 					Printf.printf "Skipping unknown column: %s\n%!" name;
-					(acc,max max_upd updated)) row (Row.empty,0L) in
+					(acc,max max_upd modified)) row (Row.empty,0L) in
 	(* Generate a new table *)
 	let table tblname tbl : Table.t =
 		let schema = Schema.Database.find tblname schema.Schema.database in
 		Table.fold
-			(fun objref created updated r acc ->
+                        (fun objref { Stat.created; modified } r acc ->
 				let (r,updated) = row schema r in
-				Table.update updated objref Row.empty (fun _ -> r) (Table.add created objref r acc)) tbl Table.empty in
+				Table.update modified objref Row.empty (fun _ -> r) (Table.add created objref r acc)) tbl Table.empty in
 	Database.update
 		(fun ts -> 
 			TableSet.fold 
-				(fun tblname created updated tbl acc ->
+                                (fun tblname { Stat.created; modified } tbl acc ->
 					let tbl' = table tblname tbl in
-					TableSet.add updated tblname tbl' acc) ts TableSet.empty)
+					TableSet.add modified tblname tbl' acc) ts TableSet.empty)
 		db
 
 let db_registration_mutex = Mutex.create ()
diff --git a/ocaml/database/db_cache_impl.ml b/ocaml/database/db_cache_impl.ml
index 63618d871c8..365eb814994 100644
--- a/ocaml/database/db_cache_impl.ml
+++ b/ocaml/database/db_cache_impl.ml
@@ -50,14 +50,14 @@ let is_valid_ref t objref =
 		| None -> false
 			
 let read_field_internal t tblname fldname objref db = 
-	Row.find fldname (Table.find_exn tblname objref (TableSet.find tblname (Database.tableset db)))
+        try
+	        Row.find fldname (Table.find objref (TableSet.find tblname (Database.tableset db)))
+        with Not_found ->
+                raise (DBCache_NotFound ("missing row", tblname, objref))
 
 (* Read field from cache *)
 let read_field t tblname fldname objref =
-	read_field_internal t tblname fldname objref (get_database t)
-
-
-
+        Schema.Value.marshal (read_field_internal t tblname fldname objref (get_database t))
 
 (** Finds the longest XML-compatible UTF-8 prefix of the given *)
 (** string, by truncating the string at the first incompatible *)
@@ -78,11 +78,15 @@ let write_field_locked t tblname objref fldname newval =
 	Database.notify (WriteField(tblname, objref, fldname, current_val, newval)) (get_database t)
 			
 let write_field t tblname objref fldname newval =
+        let db = get_database t in
+	let schema = Schema.table tblname (Database.schema db) in
+        let column = Schema.Table.find fldname schema in
+        let newval = Schema.Value.unmarshal column.Schema.Column.ty newval in
 	with_lock (fun () -> 
 		write_field_locked t tblname objref fldname newval)
 
-let refresh_row t tblname objref =
-	update_database t (update_generation tblname objref);
+let touch_row t tblname objref =
+	update_database t (touch tblname objref);
 	Database.notify (RefreshRow(tblname, objref)) (get_database t)
 
 (* This function *should* only be used by db_actions code looking up Set(Ref _) fields:
@@ -104,16 +108,16 @@ let read_set_ref t rcd =
 		let _, many_tbl, many_fld = List.find (fun (a, _, _) -> a = one_fld) rels in
 		let objref = rcd.where_value in
 		
-		let str = read_field_internal t many_tbl many_fld objref db in
-		String_unmarshall_helper.set (fun x -> x) str		
+		Schema.Value.Unsafe_cast.set (read_field_internal t many_tbl many_fld objref db)
 	end else begin
 		error "Illegal read_set_ref query { table = %s; where_field = %s; where_value = %s; return = %s }; falling back to linear scan" rcd.table rcd.where_field rcd.where_value rcd.return;
 		Printf.printf "Illegal read_set_ref query { table = %s; where_field = %s; where_value = %s; return = %s }; falling back to linear scan\n%!" rcd.table rcd.where_field rcd.where_value rcd.return;
 		let tbl = TableSet.find rcd.table (Database.tableset db) in
 		Table.fold
-			(fun rf _ _ row acc ->
-				if Row.find rcd.where_field row = rcd.where_value 
-				then Row.find rcd.return row :: acc else acc)
+			(fun rf _ row acc ->
+                                let v = Schema.Value.Unsafe_cast.string (Row.find rcd.where_field row) in
+				if v = rcd.where_value 
+				then v :: acc else acc)
 			tbl []
 	end
 			
@@ -125,50 +129,61 @@ let read_set_ref t rcd =
    name of the Set Ref field in tbl; and ref list is the list of foreign keys from related
    table with remote-fieldname=objref] *)
 let read_record_internal db tblname objref =
-	let tbl = TableSet.find tblname (Database.tableset db) in
-	let row = Table.find_exn tblname objref tbl in
-	let fvlist = Row.fold (fun k _ _ d env -> (k,d)::env) row [] in
-	(* Unfortunately the interface distinguishes between Set(Ref _) types and 
-	   ordinary fields *)
-	let schema = Schema.table tblname (Database.schema db) in
-	let set_ref = List.filter (fun (k, _) ->
-		try
-			let column = Schema.Table.find k schema in
-			column.Schema.Column.issetref
-		with Not_found as e ->
-			Printf.printf "Failed to find table %s in schema\n%!" k;
-			raise e
-	) fvlist in
-	(* the set_ref fields must be converted back into lists *)
-	let set_ref = List.map (fun (k, v) -> 
-		k, String_unmarshall_helper.set (fun x -> x) v) set_ref in
-	(fvlist, set_ref)
+        try
+	        let tbl = TableSet.find tblname (Database.tableset db) in
+	        let row = Table.find objref tbl in
+	        let fvlist = Row.fold (fun k _ d env -> (k,d)::env) row [] in
+	        (* Unfortunately the interface distinguishes between Set(Ref _) types and 
+	           ordinary fields *)
+	        let schema = Schema.table tblname (Database.schema db) in
+	        let set_ref = List.filter (fun (k, _) ->
+                        let column = Schema.Table.find k schema in
+                        column.Schema.Column.issetref
+                ) fvlist in
+                let fvlist = List.map (fun (k, v) ->
+                        k, Schema.Value.marshal v
+                ) fvlist in
+	        (* the set_ref fields must be converted back into lists *)
+	        let set_ref = List.map (fun (k, v) -> 
+                        k, Schema.Value.Unsafe_cast.set v) set_ref in
+	        (fvlist, set_ref)
+        with Not_found ->
+                raise (DBCache_NotFound ("missing row", tblname, objref))
+
 let read_record t = read_record_internal (get_database t)
 
 (* Delete row from tbl *)
 let delete_row_locked t tblname objref =
-	W.debug "delete_row %s (%s)" tblname objref;
+        try
+                W.debug "delete_row %s (%s)" tblname objref;
 	
-	let tbl = TableSet.find tblname (Database.tableset (get_database t)) in
-	let row = Table.find_exn tblname objref tbl in
+	        let tbl = TableSet.find tblname (Database.tableset (get_database t)) in
+	        let row = Table.find objref tbl in
 	
-	let db = get_database t in
-	Database.notify (PreDelete(tblname, objref)) db;
-	update_database t (remove_row tblname objref);
-	Database.notify (Delete(tblname, objref, Row.fold (fun k _ _ v acc -> (k, v) :: acc) row [])) (get_database t)
-		
+	        let db = get_database t in
+	        Database.notify (PreDelete(tblname, objref)) db;
+	        update_database t (remove_row tblname objref);
+	        Database.notify (Delete(tblname, objref, Row.fold (fun k _ v acc -> (k, v) :: acc) row [])) (get_database t)
+        with Not_found ->
+                raise (DBCache_NotFound ("missing row", tblname, objref))
+
 let delete_row t tblname objref = 
 	with_lock (fun () -> delete_row_locked t tblname objref)
 
 (* Create new row in tbl containing specified k-v pairs *)
 let create_row_locked t tblname kvs' new_objref =
-	
-    (* Ensure values are valid for UTF-8-encoded XML. *)
-    let kvs' = List.map (fun (key, value) -> (key, ensure_utf8_xml value)) kvs' in
+        let db = get_database t in
+	let schema = Schema.table tblname (Database.schema db) in
+
+        let kvs' = List.map (fun (key, value) ->
+                let value = ensure_utf8_xml value in
+                let column = Schema.Table.find key schema in
+                key, Schema.Value.unmarshal column.Schema.Column.ty value
+        ) kvs' in
 
     (* we add the reference to the row itself so callers can use read_field_where to
 	   return the reference: awkward if it is just the key *)
-    let kvs' = (Db_names.ref, new_objref) :: kvs' in
+    let kvs' = (Db_names.ref, Schema.Value.String new_objref) :: kvs' in
 	let g = Manifest.generation (Database.manifest (get_database t)) in
 	let row = List.fold_left (fun row (k, v) -> Row.add g k v row) Row.empty kvs' in
 	let schema = Schema.table tblname (Database.schema (get_database t)) in
@@ -176,7 +191,7 @@ let create_row_locked t tblname kvs' new_objref =
 	let row = Row.add_defaults g schema row in
 	W.debug "create_row %s (%s) [%s]" tblname new_objref (String.concat "," (List.map (fun (k,v)->"("^k^","^"v"^")") kvs'));
 	update_database t (add_row tblname new_objref row);
-	Database.notify (Create(tblname, new_objref, Row.fold (fun k _ _ v acc -> (k, v) :: acc) row [])) (get_database t)
+	Database.notify (Create(tblname, new_objref, Row.fold (fun k _ v acc -> (k, v) :: acc) row [])) (get_database t)
 		
 let create_row t tblname kvs' new_objref =
 	with_lock (fun () -> create_row_locked t tblname kvs' new_objref)
@@ -186,9 +201,9 @@ let read_field_where t rcd =
 	let db = get_database t in
 	let tbl = TableSet.find rcd.table (Database.tableset db) in
 	Table.fold
-		(fun r _ _ row acc ->
-			let field = Row.find rcd.where_field row in
-			if field = rcd.where_value then Row.find rcd.return row :: acc else acc
+		(fun r _ row acc ->
+			let field = Schema.Value.marshal (Row.find rcd.where_field row) in
+			if field = rcd.where_value then Schema.Value.marshal (Row.find rcd.return row) :: acc else acc
 		) tbl []
 		
 let db_get_by_uuid t tbl uuid_val =
@@ -209,18 +224,18 @@ let db_get_by_name_label t tbl label =
 (* Read references from tbl *)
 let read_refs t tblname =
 	let tbl = TableSet.find tblname (Database.tableset (get_database t)) in
-	Table.fold (fun r _ _ _ acc -> r :: acc) tbl []
+	Table.fold (fun r _ _ acc -> r :: acc) tbl []
 		
 (* Return a list of all the refs for which the expression returns true. *)
 let find_refs_with_filter_internal db (tblname: string) (expr: Db_filter_types.expr) =
 	let tbl = TableSet.find tblname (Database.tableset db) in
 	let eval_val row = function
 		| Db_filter_types.Literal x -> x
-		| Db_filter_types.Field x -> Row.find x row in
+		| Db_filter_types.Field x -> Schema.Value.marshal (Row.find x row) in
 	Table.fold
-		(fun r _ _ row acc ->
+		(fun r _ row acc ->
 			if Db_filter.eval_expr (eval_val row) expr
-			then Row.find Db_names.ref row :: acc else acc
+			then Schema.Value.Unsafe_cast.string (Row.find Db_names.ref row) :: acc else acc
 		) tbl []
 let find_refs_with_filter t = find_refs_with_filter_internal (get_database t)
 		
@@ -231,15 +246,14 @@ let read_records_where t tbl expr =
 	List.map (fun ref->ref, read_record_internal db tbl ref) reqd_refs
 	
 let process_structured_field_locked t (key,value) tblname fld objref proc_fn_selector =
-	
-    (* Ensure that both keys and values are valid for UTF-8-encoded XML. *)
-    let key = ensure_utf8_xml key in
-    let value = ensure_utf8_xml value in
-	
-	let tbl = TableSet.find tblname (Database.tableset (get_database t)) in
-	let row = Table.find_exn tblname objref tbl in
-	let existing_str = Row.find fld row in
-	let new_str = match proc_fn_selector with
+        (* Ensure that both keys and values are valid for UTF-8-encoded XML. *)
+        let key = ensure_utf8_xml key in
+        let value = ensure_utf8_xml value in
+	try
+	        let tbl = TableSet.find tblname (Database.tableset (get_database t)) in
+	        let row = Table.find objref tbl in
+	        let existing_str = Row.find fld row in
+	        let newval = match proc_fn_selector with
 		| AddSet -> add_to_set key existing_str
 		| RemoveSet -> remove_from_set key existing_str
 		| AddMap -> 
@@ -251,7 +265,9 @@ let process_structured_field_locked t (key,value) tblname fld objref proc_fn_sel
 					raise (Duplicate_key (tblname,fld,objref,key));
 			end
 		| RemoveMap -> remove_from_map key existing_str in
-	write_field t tblname objref fld new_str
+		write_field_locked t tblname objref fld newval
+        with Not_found ->
+                raise (DBCache_NotFound ("missing row", tblname, objref))
 	
 let process_structured_field t (key,value) tblname fld objref proc_fn_selector =
 	with_lock (fun () -> 
@@ -377,8 +393,8 @@ let make t connections default_schema =
 
 (** Return an association list of table name * record count *)
 let stats t = 
-	TableSet.fold (fun name _ _ tbl acc ->
-		let size = Table.fold (fun _ _ _ _ acc -> acc + 1) tbl 0 in
+	TableSet.fold (fun name _ tbl acc ->
+		let size = Table.fold (fun _ _ _ acc -> acc + 1) tbl 0 in
 		(name, size) :: acc)
 		(Database.tableset (get_database t))
 		[]
diff --git a/ocaml/database/db_cache_impl.mli b/ocaml/database/db_cache_impl.mli
index 42e2dc820ce..4304586f1ab 100644
--- a/ocaml/database/db_cache_impl.mli
+++ b/ocaml/database/db_cache_impl.mli
@@ -13,8 +13,9 @@ val sync : Parse_db_conf.db_connection list -> Db_cache_types.Database.t -> unit
 (** [stats t] returns some stats data for logging *)
 val stats : Db_ref.t -> (string * int) list
 
-(** [refresh_row context tbl ref] generates a RefreshRow event *)
-val refresh_row : Db_ref.t -> string -> string -> unit
+(** [touch_row context tbl ref] bumps the generation count on [tbl], [ref] and
+    generates a RefreshRow event *)
+val touch_row : Db_ref.t -> string -> string -> unit
 
 (** Used for Test_db_lowlevel *)
 val fist_delay_read_records_where : bool ref
diff --git a/ocaml/database/db_cache_test.ml b/ocaml/database/db_cache_test.ml
new file mode 100644
index 00000000000..96c9237e923
--- /dev/null
+++ b/ocaml/database/db_cache_test.ml
@@ -0,0 +1,75 @@
+(*
+ * Copyright (C) 2010-2014 Citrix Systems Inc.
+ *
+ * This program is free software; you can redistribute it and/or modify
+ * it under the terms of the GNU Lesser General Public License as published
+ * by the Free Software Foundation; version 2.1 only. with the special
+ * exception on linking described in file LICENSE.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Lesser General Public License for more details.
+ *)
+
+open Pervasiveext
+open Db_cache_types
+
+let create_test_db () =
+  let schema = Test_schemas.many_to_many in
+  let db = 
+    ((fun x -> x)
+     ++ (Db_backend.blow_away_non_persistent_fields schema)
+     ++ (Db_upgrade.generic_database_upgrade))
+      (Database.make schema) in
+
+  db
+
+let check_many_to_many () = 
+  let db = create_test_db () in
+  (* make a foo with bars = [] *)
+  (* make a bar with foos = [] *)
+  (* add 'bar' to foo.bars *)
+  let db = 
+    ((fun x -> x)
+     ++ (set_field "foo" "foo:1" "bars" (add_to_set "bar:1" (Schema.Value.Set [])))
+     ++ (add_row "foo" "foo:1" (Row.add 0L Db_names.ref (Schema.Value.String "foo:1") (Row.add 0L "bars" (Schema.Value.Set []) Row.empty)))
+     ++ (add_row "bar" "bar:1" (Row.add 0L Db_names.ref (Schema.Value.String "bar:1") (Row.add 0L "foos" (Schema.Value.Set []) Row.empty)))) db in
+  (* check that 'bar.foos' includes 'foo' *)
+  let bar_1 = Table.find "bar:1" (TableSet.find "bar" (Database.tableset db)) in
+  let bar_foos = Row.find "foos" bar_1 in
+  if bar_foos <> (Schema.Value.Set [ "foo:1" ])
+  then failwith (Printf.sprintf "check_many_to_many: bar(bar:1).foos expected ('foo:1') got %s" (Sexplib.Sexp.to_string (Schema.Value.sexp_of_t bar_foos)));
+
+  (* set foo.bars to [] *)
+  (*		let foo_1 = Table.find "foo:1" (TableSet.find "foo" (Database.tableset db)) in*)
+  let db = set_field "foo" "foo:1" "bars" (Schema.Value.Set []) db in
+  (* check that 'bar.foos' is empty *)
+  let bar_1 = Table.find "bar:1" (TableSet.find "bar" (Database.tableset db)) in
+  let bar_foos = Row.find "foos" bar_1 in
+  if bar_foos <> (Schema.Value.Set [])
+  then failwith (Printf.sprintf "check_many_to_many: bar(bar:1).foos expected () got %s" (Sexplib.Sexp.to_string (Schema.Value.sexp_of_t bar_foos)));
+  (* add 'bar' to foo.bars *)
+  let db = set_field "foo" "foo:1" "bars" (Schema.Value.Set [ "bar:1" ]) db in
+  (* check that 'bar.foos' includes 'foo' *)
+  let bar_1 = Table.find "bar:1" (TableSet.find "bar" (Database.tableset db)) in
+  let bar_foos = Row.find "foos" bar_1 in
+  if bar_foos <> (Schema.Value.Set [ "foo:1" ])
+  then failwith (Printf.sprintf "check_many_to_many: bar(bar:1).foos expected ('foo:1') got %s - 2" (Sexplib.Sexp.to_string (Schema.Value.sexp_of_t bar_foos)));
+  (* delete 'bar' *)
+  let db = remove_row "bar" "bar:1" db in
+  (* check that 'foo.bars' is empty *)
+  let foo_1 = Table.find "foo:1" (TableSet.find "foo" (Database.tableset db)) in		
+  let foo_bars = Row.find "bars" foo_1 in
+  if foo_bars <> (Schema.Value.Set [])
+  then failwith (Printf.sprintf "check_many_to_many: foo(foo:1).foos expected () got %s" (Sexplib.Sexp.to_string (Schema.Value.sexp_of_t foo_bars)));
+  ()
+
+open OUnit
+
+let _ =
+  let suite = "db_cache" >:::
+              [
+                "many to many" >:: check_many_to_many;
+              ] in
+  OUnit2.run_test_tt_main (OUnit.ounit2_of_ounit1 suite)
diff --git a/ocaml/database/db_cache_types.ml b/ocaml/database/db_cache_types.ml
index 3ecb52796db..79fd0926994 100644
--- a/ocaml/database/db_cache_types.ml
+++ b/ocaml/database/db_cache_types.ml
@@ -1,5 +1,32 @@
+(*
+ * Copyright (C) 2006-2014 Citrix Systems Inc.
+ *
+ * This program is free software; you can redistribute it and/or modify
+ * it under the terms of the GNU Lesser General Public License as published
+ * by the Free Software Foundation; version 2.1 only. with the special
+ * exception on linking described in file LICENSE.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Lesser General Public License for more details.
+ *)
+
 open Db_exn
 
+module Time = struct
+        type t = Generation.t
+end
+
+module Stat = struct
+        type t = {
+                created: Time.t;
+                modified: Time.t;
+                deleted: Time.t;
+        }
+        let make x = { created = x; modified = x; deleted = 0L }
+end
+
 (** Database tables, columns and rows are all indexed by string, each
 	using a specialised StringMap *)
 module StringMap = struct
@@ -13,27 +40,46 @@ module StringMap = struct
 end
 
 module type VAL = sig
-	type v
+	type t
+end
+
+module type MAP = sig
+	type t
+        type value
+	val empty : t
+	val add: Time.t -> string -> value -> t -> t
+	val remove : Time.t -> string -> t -> t
+	val fold : (string -> Stat.t -> value -> 'b -> 'b) -> t -> 'b -> 'b
+	val fold_over_recent : Time.t -> (string -> Stat.t -> value -> 'b -> 'b) -> t -> 'b -> 'b
+	val find : string -> t -> value
+	val mem : string -> t -> bool
+	val iter : (string -> value -> unit) -> t -> unit
+	val update : int64 -> string -> value -> (value -> value) -> t -> t
+        val touch : int64 -> string -> value -> t -> t
 end
 
-(** A specialised StringMap whose range type is V.v, and which keeps a record of when records are created/updated *)
-module Map2 = functor(V: VAL) -> struct
+(** A specialised StringMap whose range type is V.t, and which keeps a record of when records are created/updated *)
+module Make = functor(V: VAL) -> struct
 	type x = {
-		created : int64;
-		updated : int64;
-		v : V.v }
+                stat: Stat.t;
+		v : V.t
+        }
 	type map_t = x StringMap.t
 	let empty = StringMap.empty
-	let fold f = StringMap.fold (fun key x -> f key x.created x.updated x.v)
-	let add generation key value = StringMap.add key {created=generation; updated=generation; v=value}
+	let fold f = StringMap.fold (fun key x -> f key x.stat x.v)
+	let add generation key v =
+                let stat = Stat.make generation in
+                StringMap.add key { stat; v }
 	let find key map = (StringMap.find key map).v 
 	let mem = StringMap.mem
 	let iter f = StringMap.iter (fun key x -> f key x.v)
-	let remove = StringMap.remove
-	let update_generation generation key default f row =
-		StringMap.update key {created=generation; updated=generation; v=default} (fun x -> {x with updated=generation; v=f x.v}) row
+	let remove _ = StringMap.remove
+	let touch generation key default row =
+                let default = { stat = Stat.make generation; v = default } in
+                StringMap.update key default (fun x -> { x with stat = { x.stat with Stat.modified=generation } }) row
 	let update generation key default f row = 
-		let updatefn () = StringMap.update key {created=generation; updated=generation; v=default} (fun x -> {x with updated=generation; v=f x.v}) row in
+                let default = { stat = Stat.make generation; v = default } in
+                let updatefn () = StringMap.update key default (fun x -> { stat = { x.stat with Stat.modified=generation }; v=f x.v}) row in
 		if mem key row 
 		then
 			let old = find key row in
@@ -43,28 +89,14 @@ module Map2 = functor(V: VAL) -> struct
 			else updatefn ()				
 		else
 			updatefn ()
-	let fold_over_recent since f = StringMap.fold (fun x y z -> if y.updated > since then f y.created y.updated 0L x y.v z else z)
+	let fold_over_recent since f t initial = StringMap.fold (fun x y z -> if y.stat.Stat.modified > since then f x y.stat y.v z else z) t initial
 end
 
-module StringStringMap = Map2(struct type v = string end)
+module Row = struct
+        include Make(Schema.Value)
 
-module type ROW = sig
-	type t
-	val add: int64 -> string -> string -> t -> t
-	val add_defaults: int64 -> Schema.Table.t -> t -> t
-	val empty : t
-	val fold : (string -> int64 -> int64 -> string -> 'b -> 'b) -> t -> 'b -> 'b
-	val find : string -> t -> string
-	val mem : string -> t -> bool
-	val iter : (string -> string -> unit) -> t -> unit
-	val remove : string -> t -> t
-	val update : int64 -> string -> string -> (string -> string) -> t -> t
-	val fold_over_recent : int64 -> (int64 -> int64 -> int64 -> string -> string -> 'b -> 'b) -> t -> 'b -> 'b
-end
-
-module Row : ROW = struct
-	include StringStringMap
 	type t=map_t
+        type value = Schema.Value.t
 	let find key t =
 		try find key t
 		with Not_found -> raise (DBCache_NotFound ("missing field", key, ""))
@@ -77,89 +109,55 @@ module Row : ROW = struct
 			else t) t schema.Schema.Table.columns
 end
 
-module StringRowMap = Map2(struct type v = Row.t end)
-
-module type TABLE = sig
-	type t
-	val add: int64 -> string -> Row.t -> t -> t
-	val empty : t
-	val fold : (string -> int64 -> int64 -> Row.t -> 'b -> 'b) -> t -> 'b -> 'b
-	val find_exn : string -> string -> t -> Row.t
-	val find : string -> t -> Row.t
-	val mem : string -> t -> bool
-	val iter : (string -> Row.t -> unit) -> t -> unit
-	val remove : int64 -> string -> t -> t
-	val update_generation : int64 -> string -> Row.t -> (Row.t -> Row.t) -> t -> t
-	val update : int64 -> string -> Row.t -> (Row.t -> Row.t) -> t -> t
-	val fold_over_recent : int64 -> (int64 -> int64 -> int64 -> string -> 'b -> 'b) -> (unit -> unit) -> t -> 'b -> 'b
-	val rows : t -> Row.t list
-end
+module Table = struct
+        module StringRowMap = Make(Row)
 
-module Table : TABLE = struct
 	type t = { rows : StringRowMap.map_t;
 			   deleted_len : int;
-			   deleted : (int64 * int64 * string) list }
+			   deleted : (Time.t * Time.t * string) list }
+        type value = Row.t
 	let add g key value t = {t with rows=StringRowMap.add g key value t.rows}
 	let empty = {rows=StringRowMap.empty; deleted_len = 1; deleted=[(0L,0L,"")] }
 	let fold f t acc = StringRowMap.fold f t.rows acc
-	let find_exn tbl key t =
-		try StringRowMap.find key t.rows
-		with Not_found -> raise (DBCache_NotFound ("missing row", tbl, key))
 	let find key t = StringRowMap.find key t.rows
 	let mem key t = StringRowMap.mem key t.rows
 	let iter f t = StringRowMap.iter f t.rows
 	let remove g key t =
 		let upper_length_deleted_queue = 512 in
 		let lower_length_deleted_queue = 256 in
-		let created = (StringMap.find key t.rows).StringRowMap.created in
+		let created = (StringMap.find key t.rows).StringRowMap.stat.Stat.created in
 		let new_element = (created,g,key) in
 		let new_len,new_deleted =
 			if t.deleted_len + 1 < upper_length_deleted_queue
 			then t.deleted_len + 1, (new_element::t.deleted)
 			else lower_length_deleted_queue + 1, (new_element::(Listext.List.take lower_length_deleted_queue t.deleted))
 		in
-		{rows = StringRowMap.remove key t.rows;
+		{rows = StringRowMap.remove g key t.rows;
 		 deleted_len = new_len;
 		 deleted = new_deleted}
-	let update_generation g key default f t = {t with rows = StringRowMap.update_generation g key default f t.rows }
+	let touch g key default t = {t with rows = StringRowMap.touch g key default t.rows }
 	let update g key default f t = {t with rows = StringRowMap.update g key default f t.rows}
-	let fold_over_recent since f errf t acc =
-		let acc = StringRowMap.fold_over_recent since (fun c u d x _ z -> f c u d x z) t.rows acc in
-		let rec fold_over_deleted deleted acc =
-			match deleted with
-				| (created,destroyed,r)::xs ->
-					let new_acc =
-						if (destroyed > since) && (created <= since)
-						then (f created 0L destroyed r acc)
-						else acc
-					in
-					if destroyed <= since then new_acc else fold_over_deleted xs new_acc
-				| [] ->
-					errf ();
-					acc
-		in fold_over_deleted t.deleted acc
-	let rows t =
-		fold (fun _ _ _ r rs -> r :: rs) t []
+	let fold_over_recent since f t acc = StringRowMap.fold_over_recent since f t.rows acc
+
+	let fold_over_deleted since f t acc =
+                let rec loop xs acc = match xs with
+			| (created,deleted,r)::xs ->
+				let new_acc =
+					if (deleted > since) && (created <= since)
+                                        then (f r { Stat.created; modified = deleted; deleted } acc)
+					else acc
+				in
+				if deleted <= since then new_acc else loop xs new_acc
+			| [] ->
+				acc in
+                loop t.deleted acc
 end
 
-module StringTableMap = Map2(struct type v = Table.t end)
+module TableSet = struct
+	include Make(Table)
 
-module type TABLESET = sig
-	type t
-	val add: int64 -> string -> Table.t -> t -> t
-	val empty : t
-	val fold : (string -> int64 -> int64 -> Table.t -> 'b -> 'b) -> t -> 'b -> 'b
-	val find : string -> t -> Table.t
-	val mem : string -> t -> bool
-	val iter : (string -> Table.t -> unit) -> t -> unit
-	val remove : string -> t -> t
-	val update : int64 -> string -> Table.t -> (Table.t -> Table.t) -> t -> t
-	val fold_over_recent : int64 -> (int64 -> int64 -> int64 -> string -> Table.t -> 'b -> 'b) -> t -> 'b -> 'b
-end
-
-module TableSet : TABLESET = struct
-	include StringTableMap
 	type t=map_t
+        type value = Table.t
 	let find key t =
 		try find key t
 		with Not_found -> raise (DBCache_NotFound ("missing table", key, ""))
@@ -201,11 +199,11 @@ module Manifest = struct
 
 	let generation x = x.generation_count
 
-	let update_generation f x = {
+	let touch f x = {
 		x with generation_count = f x.generation_count
 	}
 
-	let next = update_generation (Int64.add 1L)
+	let next = touch (Int64.add 1L)
 
 	let schema x = match x.schema with
 		| None -> 0, 0
@@ -219,10 +217,10 @@ end
 (** The core database updates (RefreshRow and PreDelete is more of an 'event') *)
 type update =
 	| RefreshRow of string (* tblname *) * string (* objref *)
-	| WriteField of string (* tblname *) * string (* objref *) * string (* fldname *) * string (* oldval *) * string (* newval *)
+	| WriteField of string (* tblname *) * string (* objref *) * string (* fldname *) * Schema.Value.t (* oldval *) * Schema.Value.t (* newval *)
 	| PreDelete of string (* tblname *) * string (* objref *)
-	| Delete of string (* tblname *) * string (* objref *) * (string * string) list (* values *)
-	| Create of string (* tblname *) * string (* objref *) * (string * string) list (* values *)
+	| Delete of string (* tblname *) * string (* objref *) * (string * Schema.Value.t) list (* values *)
+	| Create of string (* tblname *) * string (* objref *) * (string * Schema.Value.t) list (* values *)
 
 module Database = struct
 	type t = {
@@ -247,7 +245,7 @@ module Database = struct
 		{ x with tables = f x.tables }
 
 	let set_generation g =
-		update_manifest (Manifest.update_generation (fun _ -> g))
+		update_manifest (Manifest.touch (fun _ -> g))
 
 	let update_tableset f x =
 		{ x with tables = f x.tables }
@@ -275,19 +273,19 @@ module Database = struct
 		(* Recompute the keymap *)
 		let keymap =
 			TableSet.fold
-				(fun tblname _ _ tbl acc ->
+				(fun tblname _ tbl acc ->
 					Table.fold
-						(fun rf _ _ row acc ->
+						(fun rf _ row acc ->
 							let acc = KeyMap.add_unique tblname Db_names.ref (Ref rf) (tblname, rf) acc in
 							if Row.mem Db_names.uuid row
-							then KeyMap.add_unique tblname Db_names.uuid (Uuid (Row.find Db_names.uuid row)) (tblname, rf) acc
-							else acc
+                                                        then KeyMap.add_unique tblname Db_names.uuid (Uuid (Schema.Value.Unsafe_cast.string (Row.find Db_names.uuid row))) (tblname, rf) acc
+                                                        else acc
 						)
 						tbl acc)
 				x.tables KeyMap.empty in
 		(* For each of the one-to-many relationships, recompute the many end *)
 		let tables =
-			Schema.StringMap.fold
+			Schema.ForeignMap.fold
 				(fun one_tblname rels tables ->
 					List.fold_left (fun tables (one_fldname, many_tblname, many_fldname) ->
 						(* VBD.VM : Ref(VM) -> VM.VBDs : Set(Ref(VBD)) *)
@@ -297,27 +295,26 @@ module Database = struct
 						   VBDs may be missing a VBDs field altogether on
 						   upgrade) *)
 						let many_tbl' = Table.fold
-							(fun vm _ _ row acc ->
-								let row' = Row.add g many_fldname (SExpr.string_of (SExpr.Node [])) row in
+							(fun vm _ row acc ->
+                                                                let row' = Row.add g many_fldname (Schema.Value.Set []) row in
 								Table.add g vm row' acc)
 							many_tbl Table.empty in
 
 						(* Build up a table of VM -> VBDs *)
 
 						let vm_to_vbds = Table.fold
-							(fun vbd _ _ row acc ->
-								let vm = Row.find one_fldname row in
-								let existing = if Schema.StringMap.mem vm acc then Schema.StringMap.find vm acc else [] in
-								Schema.StringMap.add vm (vbd :: existing) acc)
-							one_tbl Schema.StringMap.empty in
-						let many_tbl'' = Schema.StringMap.fold
+							(fun vbd _ row acc ->
+								let vm = Schema.Value.Unsafe_cast.string (Row.find one_fldname row) in
+								let existing = if Schema.ForeignMap.mem vm acc then Schema.ForeignMap.find vm acc else [] in
+								Schema.ForeignMap.add vm (vbd :: existing) acc)
+							one_tbl Schema.ForeignMap.empty in
+						let many_tbl'' = Schema.ForeignMap.fold
 							(fun vm vbds acc ->
 								if not(Table.mem vm acc)
 								then acc
 								else
 									let row = Table.find vm acc in
-									let vbds' = SExpr.string_of (SExpr.Node (List.map (fun x -> SExpr.String x) vbds)) in
-									let row' = Row.add g many_fldname vbds' row in
+									let row' = Row.add g many_fldname (Schema.Value.Set vbds) row in
 									Table.add g vm row' acc)
 							vm_to_vbds many_tbl' in
 						TableSet.add g many_tblname many_tbl'' tables)
@@ -345,38 +342,22 @@ end
 
 (* Helper functions to deal with Sets and Maps *)
 let add_to_set key t =
-	let existing = Db_action_helper.parse_sexpr t in
-	let processed = Db_action_helper.add_key_to_set key existing in
-	SExpr.string_of (SExpr.Node processed)
+        let t = Schema.Value.Unsafe_cast.set t in
+        Schema.Value.Set (if List.mem key t then t else key :: t)
 
 let remove_from_set key t =
-	let existing = Db_action_helper.parse_sexpr t in
-	let processed = List.filter (function SExpr.String x -> x <> key | _ -> true) existing in
-	SExpr.string_of (SExpr.Node processed)
-
-let set_of_string t =
-	List.map
-		(function SExpr.String x -> x
-			| x -> failwith (Printf.sprintf "Unexpected sexpr: %s" t))
-		(Db_action_helper.parse_sexpr t)
-let string_of_set t = SExpr.string_of (SExpr.Node (List.map (fun x -> SExpr.String x) t))
+        let t = Schema.Value.Unsafe_cast.set t in
+        Schema.Value.Set (List.filter (fun x -> x <> key) t)
 
 exception Duplicate
 let add_to_map key value t =
-	let existing = Db_action_helper.parse_sexpr t in
-	let kv = SExpr.Node [ SExpr.String key; SExpr.String value ] in
-	let duplicate = List.fold_left (||) false
-			(List.map (function SExpr.Node (SExpr.String k :: _) when k = key -> true
-				| _ -> false) existing) in
-	if duplicate then raise Duplicate;
-	let processed = kv::existing in
-	SExpr.string_of (SExpr.Node processed)
+        let t = Schema.Value.Unsafe_cast.pairs t in
+        if List.mem key (List.map fst t) then raise Duplicate;
+        Schema.Value.Pairs ((key, value) :: t)
 
 let remove_from_map key t =
-	let existing = Db_action_helper.parse_sexpr t in
-	let processed = List.filter (function SExpr.Node [ SExpr.String x; _ ] -> x <> key
-		| _ -> true) existing in
-	SExpr.string_of (SExpr.Node processed)
+        let t = Schema.Value.Unsafe_cast.pairs t in
+        Schema.Value.Pairs (List.filter (fun (k, _) -> k <> key) t)
 
 
 let (++) f g x = f (g x)
@@ -388,20 +369,23 @@ let is_valid tblname objref db =
 
 
 let get_field tblname objref fldname db =
-	Row.find fldname (Table.find_exn tblname objref (TableSet.find tblname (Database.tableset db)))
+        try
+	  Row.find fldname (Table.find objref (TableSet.find tblname (Database.tableset db)))
+        with Not_found ->
+          raise (DBCache_NotFound ("missing row", tblname, objref))
 
 let unsafe_set_field g tblname objref fldname newval =
 	(Database.update 
 	++ (TableSet.update g tblname Table.empty)
 	++ (Table.update g objref Row.empty)
-	++ (Row.update g fldname ""))
+	++ (Row.update g fldname (Schema.Value.String "")))
 		(fun _ -> newval) 
 
 let update_one_to_many g tblname objref f db =
 	if not (is_valid tblname objref db) then db else
 	List.fold_left (fun db (one_fld, many_tbl, many_fld) ->
 		(* the value one_fld_val is the Ref _ *)
-		let one_fld_val = get_field tblname objref one_fld db in
+		let one_fld_val = Schema.Value.Unsafe_cast.string (get_field tblname objref one_fld db) in
 		let valid = try ignore(Database.table_of_ref one_fld_val db); true with _ -> false in
 		if valid
 		then unsafe_set_field g many_tbl one_fld_val many_fld (f objref (get_field many_tbl one_fld_val many_fld db)) db
@@ -411,8 +395,7 @@ let update_one_to_many g tblname objref f db =
 let update_many_to_many g tblname objref f db =
 	if not (is_valid tblname objref db) then db else
 	List.fold_left (fun db (this_fld, other_tbl, other_fld) ->
-		let this_fld_val = get_field tblname objref this_fld db in
-		let this_fld_refs = set_of_string this_fld_val in
+                let this_fld_refs = Schema.Value.Unsafe_cast.set (get_field tblname objref this_fld db) in
 		(* for each of this_fld_refs, apply f *)
 		List.fold_left (fun db other_ref ->
 			let valid = try ignore(Database.table_of_ref other_ref db); true with _ -> false in
@@ -445,7 +428,7 @@ let set_field tblname objref fldname newval db =
 		 ++ ((Database.update 
 			  ++ (TableSet.update g tblname Table.empty)
 			  ++ (Table.update g objref Row.empty)
-			  ++ (Row.update g fldname "")) (fun _ -> newval))
+			  ++ (Row.update g fldname (Schema.Value.String ""))) (fun _ -> newval))
 		 ++ (update_one_to_many g tblname objref remove_from_set)
 		 ++ (update_many_to_many g tblname objref remove_from_set)) db
 	end else begin
@@ -454,11 +437,11 @@ let set_field tblname objref fldname newval db =
 		 ++ ((Database.update 
 			 ++ (TableSet.update g tblname Table.empty)
 			 ++ (Table.update g objref Row.empty)
-			 ++ (Row.update g fldname ""))
+			 ++ (Row.update g fldname (Schema.Value.String "")))
 				(fun _ -> newval))) db
 	end
 
-let update_generation tblname objref db =
+let touch tblname objref db =
 	let g = Manifest.generation (Database.manifest db) in
 	(* We update the generation twice so that we can return the lower count
 	   for the "event.inject" API to guarantee that the token from a later
@@ -467,7 +450,7 @@ let update_generation tblname objref db =
 	(Database.increment ++ Database.increment
 		++ ((Database.update
 			++ (TableSet.update g tblname Table.empty)
-			++ (Table.update_generation g objref Row.empty)) id
+			++ (Table.touch g objref)) Row.empty
 		)) db
 
 let add_row tblname objref newval db =
@@ -482,13 +465,13 @@ let add_row tblname objref newval db =
 	 ++ (Database.update_keymap (KeyMap.add_unique tblname Db_names.ref (Ref objref) (tblname, objref)))
 	 ++ (Database.update_keymap (fun m ->
 									 if Row.mem Db_names.uuid newval
-									 then KeyMap.add_unique tblname Db_names.uuid (Uuid (Row.find Db_names.uuid newval)) (tblname, objref) m
+									 then KeyMap.add_unique tblname Db_names.uuid (Uuid (Schema.Value.Unsafe_cast.string (Row.find Db_names.uuid newval))) (tblname, objref) m
 									 else m))) db
 
 let remove_row tblname objref db =
 	let uuid =
 		try
-			Some (Row.find Db_names.uuid (Table.find objref (TableSet.find tblname (Database.tableset db))))
+			Some (Schema.Value.Unsafe_cast.string (Row.find Db_names.uuid (Table.find objref (TableSet.find tblname (Database.tableset db)))))
 		with _ -> None in
 	let g = db.Database.manifest.Manifest.generation_count in
 	(Database.increment
diff --git a/ocaml/database/db_cache_types.mli b/ocaml/database/db_cache_types.mli
index b15d2079061..d1a9354eec1 100644
--- a/ocaml/database/db_cache_types.mli
+++ b/ocaml/database/db_cache_types.mli
@@ -1,45 +1,105 @@
-module Row :
-  sig
-    type t
-    val add : int64 -> string -> string -> t -> t
-	val add_defaults : int64 -> Schema.Table.t -> t -> t
-    val empty : t
-    val fold : (string -> int64 -> int64 -> string -> 'a -> 'a) -> t -> 'a -> 'a
-    val find : string -> t -> string
-    val iter : (string -> string -> unit) -> t -> unit
-    val remove : string -> t -> t
-    val update : int64 -> string -> string -> (string -> string) -> t -> t
-	val fold_over_recent : int64 -> (int64 -> int64 -> int64 -> string -> string -> 'b -> 'b) -> t -> 'b -> 'b
-  end
+(*
+ * Copyright (C) 2006-2014 Citrix Systems Inc.
+ *
+ * This program is free software; you can redistribute it and/or modify
+ * it under the terms of the GNU Lesser General Public License as published
+ * by the Free Software Foundation; version 2.1 only. with the special
+ * exception on linking described in file LICENSE.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Lesser General Public License for more details.
+ *)
 
-module Table :
-  sig
-    type t
-    val add : int64 -> string -> Row.t -> t -> t
-    val empty : t
-    val fold : (string -> int64 -> int64 -> Row.t -> 'a -> 'a) -> t -> 'a -> 'a
-	val find_exn : string -> string -> t -> Row.t
-    val find : string -> t -> Row.t
-	val mem : string -> t -> bool
-    val iter : (string -> Row.t -> unit) -> t -> unit
-    val remove : int64 -> string -> t -> t
-    val update : int64 -> string -> Row.t -> (Row.t -> Row.t) -> t -> t
-    val fold_over_recent : int64 -> (int64 -> int64 -> int64 -> string -> 'b -> 'b) -> (unit -> unit) -> t -> 'b -> 'b
-	val rows : t -> Row.t list
-  end
+module Time : sig
+        type t = Generation.t
+        (** A monotonically increasing counter associated with this database *)
+end
 
-module TableSet :
-  sig
-    type t
-    val add : int64 -> string -> Table.t -> t -> t
-    val empty : t
-    val fold : (string -> int64 -> int64 -> Table.t -> 'a -> 'a) -> t -> 'a -> 'a
-    val find : string -> t -> Table.t
-    val iter : (string -> Table.t -> unit) -> t -> unit
-    val remove : string -> t -> t
-    val update : int64 -> string -> Table.t -> (Table.t -> Table.t) -> t -> t
-    val fold_over_recent : int64 -> (int64 -> int64 -> int64 -> string -> Table.t -> 'b -> 'b) -> t -> 'b -> 'b
-  end
+module Stat : sig
+        type t = {
+                created: Time.t;  (** Time this value was created *)
+                modified: Time.t; (** Time this value was last modified *)
+                deleted: Time.t;  (** Time this value was deleted (or 0L meaning it is still alive) *)
+        }
+        (** Metadata associated with a database value *)
+end
+
+module type MAP = sig
+        type t
+        (** A map from string to some value *)
+
+        type value
+        (** The type of the values in the map *)
+
+        val empty : t
+        (** The empty map *)
+
+        val add: Time.t -> string -> value -> t -> t
+        (** [add now key value map] returns a new map with [key] associated with [value],
+            with creation time [now] *)
+
+        val remove : Time.t -> string -> t -> t
+        (** [remove now key t] removes the binding of [key] from [t]. *)
+
+        val fold : (string -> Stat.t -> value -> 'b -> 'b) -> t -> 'b -> 'b
+        (** [fold f t initial] folds [f key stats value acc] over the items in [t] *)
+
+        val fold_over_recent : Time.t -> (string -> Stat.t -> value -> 'b -> 'b) -> t -> 'b -> 'b
+        (** [fold_over_recent since f t initial] folds [f key stats value acc] over all the
+            items with a modified time larger than [since] *)
+
+        val find : string -> t -> value
+        (** [find key t] returns the value associated with [key] in [t] or raises
+            [DBCache_NotFound] *)
+
+        val mem : string -> t -> bool
+        (** [mem key t] returns true if [value] is associated with [key] in [t] or false
+            otherwise *)
+
+        val iter : (string -> value -> unit) -> t -> unit
+        (** [iter f t] applies [f key value] to each binding in [t] *)
+
+        val update : Time.t -> string -> value -> (value -> value) -> t -> t
+        (** [update now key default f t] returns a new map which is the same as [t] except:
+            if there is a value associated with [key] it is replaced with [f key]
+            or if there is no value associated with [key] then [default] is associated with [key].
+            This function touches the modification time of [key] *unless* [f key] is physically
+            equal with the current value: in this case the modification time isn't bumped as
+            an optimisation.
+          *)
+
+        val touch : Time.t -> string -> value -> t -> t
+        (** [touch now key default t] returns a new map which is the same as [t] except:
+            if there is a value associated with [t] then its modification time is set to [now];
+            if there is no value asssociated with [t] then one is created with value [default].
+            On exit there will be a binding of [key] whose modification time is [now] *)
+end
+
+module Row : sig
+        include MAP
+          with type value = Schema.Value.t
+
+        val add_defaults: Time.t -> Schema.Table.t -> t -> t
+        (** [add_defaults now schema t]: returns a row which is [t] extended to contain
+            all the columns specified in the schema, with default values set if not already
+            in [t]. If the schema is missing a default value then raises [DBCache_NotFound]:
+            this would happen if a client failed to provide a necessary field. *)
+
+end
+
+module Table : sig
+        include MAP
+          with type value = Row.t
+
+        val fold_over_deleted : Time.t -> (string -> Stat.t -> 'b -> 'b) -> t -> 'b -> 'b
+        (** [fold_over_deleted now f t initial] folds [f key stat acc] over the keys
+            which have been recently deleted. Note this is not guaranteed to remember
+            all events, so the list may be short. *)
+end
+
+module TableSet : MAP with type value = Table.t
 
 module Manifest :
   sig
@@ -47,7 +107,7 @@ module Manifest :
     val empty : t
     val make : int -> int -> Generation.t -> t
     val generation : t -> Generation.t
-    val update_generation : (Generation.t -> Generation.t) -> t -> t
+    val touch : (Generation.t -> Generation.t) -> t -> t
     val next : t -> t
 	val schema : t -> int * int
 	val update_schema : ((int * int) option -> (int * int) option) -> t -> t
@@ -56,10 +116,10 @@ module Manifest :
 (** The core database updates (RefreshRow and PreDelete is more of an 'event') *)
 type update = 
 	| RefreshRow of string (* tblname *) * string (* objref *)
-	| WriteField of string (* tblname *) * string (* objref *) * string (* fldname *) * string  (* oldval *) * string (* newval *)
+	| WriteField of string (* tblname *) * string (* objref *) * string (* fldname *) * Schema.Value.t (* oldval *) * Schema.Value.t (* newval *)
 	| PreDelete of string (* tblname *) * string (* objref *)
-	| Delete of string (* tblname *) * string (* objref *) * (string * string) list (* values *)
-	| Create of string (* tblname *) * string (* objref *) * (string * string) list (* values *)
+	| Delete of string (* tblname *) * string (* objref *) * (string * Schema.Value.t) list (* values *)
+	| Create of string (* tblname *) * string (* objref *) * (string * Schema.Value.t) list (* values *)
 
 module Database :
   sig
@@ -84,17 +144,16 @@ module Database :
   end
 
 exception Duplicate
-val add_to_set : string -> string -> string
-val remove_from_set : string -> string -> string
-val add_to_map : string -> string -> string -> string
-val remove_from_map : string -> string -> string
+val add_to_set : string -> Schema.Value.t -> Schema.Value.t
+val remove_from_set : string -> Schema.Value.t -> Schema.Value.t
+val add_to_map : string -> string -> Schema.Value.t -> Schema.Value.t
+val remove_from_map : string -> Schema.Value.t -> Schema.Value.t
 
-val set_field : string -> string -> string -> string -> Database.t -> Database.t
-val get_field : string -> string -> string -> Database.t -> string
+val set_field : string -> string -> string -> Schema.Value.t -> Database.t -> Database.t
+val get_field : string -> string -> string -> Database.t -> Schema.Value.t
 val remove_row : string -> string -> Database.t -> Database.t
 val add_row : string -> string -> Row.t -> Database.t -> Database.t
-
-val update_generation : string -> string -> Database.t -> Database.t
+val touch : string -> string -> Database.t -> Database.t
 
 type where_record = {
 	table: string;       (** table from which ... *)
@@ -112,4 +171,3 @@ type structured_op_t =
 	| RemoveMap
 val structured_op_t_of_rpc: Rpc.t -> structured_op_t
 val rpc_of_structured_op_t: structured_op_t -> Rpc.t
-
diff --git a/ocaml/database/db_upgrade.ml b/ocaml/database/db_upgrade.ml
index 08bcfd84346..2584e62430d 100644
--- a/ocaml/database/db_upgrade.ml
+++ b/ocaml/database/db_upgrade.ml
@@ -21,7 +21,7 @@ open Pervasiveext
 
 (** Automatically insert blank tables and new columns with default values *)
 let generic_database_upgrade db =
-  let existing_table_names = TableSet.fold (fun name _ _ _ acc -> name :: acc) (Database.tableset db) [] in
+  let existing_table_names = TableSet.fold (fun name _ _ acc -> name :: acc) (Database.tableset db) [] in
   let schema_table_names = Schema.table_names (Database.schema db) in
   let created_table_names = Listext.List.set_difference schema_table_names existing_table_names in
   let g = Manifest.generation (Database.manifest db) in
@@ -36,7 +36,7 @@ let generic_database_upgrade db =
       (fun db tblname ->
 		  let tbl = TableSet.find tblname (Database.tableset db) in
 		  let schema = Schema.table tblname (Database.schema db) in
-		  let add_fields_to_row objref _ _ r tbl : Table.t =
+		  let add_fields_to_row objref _ r tbl : Table.t =
 			  let row = Row.add_defaults g schema r in
 			  Table.add g objref row tbl in
 			let tbl = Table.fold add_fields_to_row tbl Table.empty in
diff --git a/ocaml/database/db_xml.ml b/ocaml/database/db_xml.ml
index 33baa98c657..31740c0477f 100644
--- a/ocaml/database/db_xml.ml
+++ b/ocaml/database/db_xml.ml
@@ -43,13 +43,13 @@ module To = struct
 
   (* Marshal a whole database table to an Xmlm output abstraction *)
   let table schema (output: Xmlm.output) name (tbl: Table.t) = 
-    let record rf ctime mtime (row: Row.t) _ = 
+          let record rf { Stat.created; modified } (row: Row.t) _ = 
 		let preamble = 
 			if persist_generation_counts 
-			then [("__mtime",Int64.to_string mtime); ("__ctime",Int64.to_string ctime); ("ref",rf)] 
+			then [("__mtime",Generation.to_string modified); ("__ctime",Generation.to_string created); ("ref",rf)] 
 			else [("ref",rf)] 
 		in
-	  let (tag: Xmlm.tag) = make_tag "row" (List.rev (Row.fold (fun k _ _ v acc -> (k, Xml_spaces.protect v) :: acc) row preamble)) in
+	  let (tag: Xmlm.tag) = make_tag "row" (List.rev (Row.fold (fun k _ v acc -> (k, Xml_spaces.protect (Schema.Value.marshal v)) :: acc) row preamble)) in
       Xmlm.output output (`El_start tag);
       Xmlm.output output `El_end in
     let tag = make_tag "table" [ "name", name ] in
@@ -102,14 +102,14 @@ module From = struct
 					raise (Unmarshall_error "Unexpected end of file")
 			end else
 				f accu in
-		let rec f ((tableset, table, manifest) as acc) = match Xmlm.input input with
+		let rec f ((tableset, table, tblname, manifest) as acc) = match Xmlm.input input with
 				(* On reading a start tag... *)
 			| `El_start (tag: Xmlm.tag) ->
 				Stack.push tag tags;
 				begin match tag with
 					| (_, ("database" | "manifest")), _ -> f acc
-					| (_, "table"), [ (_, "name"), _ ] ->
-						f (tableset, Table.empty, manifest)
+					| (_, "table"), [ (_, "name"), tblname ] ->
+						f (tableset, Table.empty, tblname, manifest)
 					| (_, "row"), ((_, "ref"), rf) :: rest ->
 						(* Remove any other duplicate "ref"s which might have sneaked in there *)
 						let rest = List.filter (fun ((_,k), _) -> k <> "ref") rest in
@@ -118,11 +118,15 @@ module From = struct
 						let ctime = match ctime_l with | [(_,ctime_s)] -> Int64.of_string ctime_s | _ -> 0L in
 						let mtime = match mtime_l with | [(_,mtime_s)] -> Int64.of_string mtime_s | _ -> 0L in
 						let row = List.fold_left (fun row ((_, k), v) -> 
-							Row.update mtime k "" (fun _ -> (Xml_spaces.unprotect v)) (Row.add ctime k (Xml_spaces.unprotect v) row)
+                                                        let table_schema = Schema.Database.find tblname schema.Schema.database in
+                                                        let column_schema = Schema.Table.find k table_schema in
+                                                        let value = Schema.Value.unmarshal column_schema.Schema.Column.ty (Xml_spaces.unprotect v) in
+                                                        let empty = column_schema.Schema.Column.empty in
+							Row.update mtime k empty (fun _ -> value) (Row.add ctime k value row)
 						) Row.empty rest in
-						f (tableset, (Table.update mtime rf Row.empty (fun _ -> row) (Table.add ctime rf row table)), manifest)
+						f (tableset, (Table.update mtime rf Row.empty (fun _ -> row) (Table.add ctime rf row table)), tblname, manifest)
 					| (_, "pair"), [ (_, "key"), k; (_, "value"), v ] ->
-						f (tableset, table, (k, v) :: manifest)
+						f (tableset, table, tblname, (k, v) :: manifest)
 					| (_, name), _ -> 
 						raise (Unmarshall_error (Printf.sprintf "Unexpected tag: %s" name))
 				end
@@ -132,13 +136,13 @@ module From = struct
 				begin match tag with
 					| (_, ("database" | "manifest" | "row" | "pair")), _ -> maybe_return f acc
 					| (_, "table"), [ (_, "name"), name ] ->
-						maybe_return f (TableSet.add 0L name table tableset, Table.empty, manifest)
+						maybe_return f (TableSet.add 0L name table tableset, Table.empty, "", manifest)
 					| (_, name), _ -> 
 						raise (Unmarshall_error (Printf.sprintf "Unexpected tag: %s" name))
 				end
 			| _ -> f acc
 		in
-		let (ts, _, manifest) = f (TableSet.empty, Table.empty, []) in
+		let (ts, _, _, manifest) = f (TableSet.empty, Table.empty, "", []) in
 		let g = Int64.of_string (List.assoc _generation_count manifest) in
 		let major_vsn = int_of_string (List.assoc _schema_major_vsn manifest) in
 		let minor_vsn = int_of_string (List.assoc _schema_minor_vsn manifest) in
diff --git a/ocaml/database/eventgen.ml b/ocaml/database/eventgen.ml
index 558a64ebed2..428251b0b3a 100644
--- a/ocaml/database/eventgen.ml
+++ b/ocaml/database/eventgen.ml
@@ -79,11 +79,15 @@ let database_callback event db =
 	let other_tbl_refs_for_this_field tblname fldname =
 		List.filter (fun (_,fld) -> fld=fldname) (other_tbl_refs tblname) in	
 
-	let is_valid_ref r = 
-		try
-			ignore(Database.table_of_ref r db);
-			true
-		with _ -> false in
+	let is_valid_ref = function
+                | Schema.Value.String r ->
+                        begin
+	        	        try
+	        	        	ignore(Database.table_of_ref r db);
+	        	        	true
+	        	        with _ -> false
+                        end
+                | _ -> false in
 
 	match event with
 		| RefreshRow (tblname, objref) ->
@@ -100,12 +104,14 @@ let database_callback event db =
 		| WriteField (tblname, objref, fldname, oldval, newval) ->
 			let events_old_val = 
 				if is_valid_ref oldval then 
+                                        let oldval = Schema.Value.Unsafe_cast.string oldval in
 					events_of_other_tbl_refs
 						(List.map (fun (tbl,fld) ->
 							(tbl, oldval, find_get_record tbl ~__context:context ~self:oldval)) (other_tbl_refs_for_this_field tblname fldname)) 
 				else [] in
 			let events_new_val =
 				if is_valid_ref newval then
+                                        let newval = Schema.Value.Unsafe_cast.string newval in
 					events_of_other_tbl_refs
 						(List.map (fun (tbl,fld) ->
 							(tbl, newval, find_get_record tbl ~__context:context ~self:newval)) (other_tbl_refs_for_this_field tblname fldname)) 
@@ -148,9 +154,10 @@ let database_callback event db =
 			let other_tbl_refs =
 				List.fold_left (fun accu (remote_tbl,fld) ->
 					let fld_value = List.assoc fld kv in
-					if is_valid_ref fld_value 
-					then (remote_tbl, fld_value, find_get_record remote_tbl ~__context:context ~self:fld_value) :: accu 
-					else accu) 
+					if is_valid_ref fld_value then begin
+                                                let fld_value = Schema.Value.Unsafe_cast.string fld_value in
+					        (remote_tbl, fld_value, find_get_record remote_tbl ~__context:context ~self:fld_value) :: accu 
+                                        end else accu) 
 					[] other_tbl_refs in
 			let other_tbl_ref_events = events_of_other_tbl_refs other_tbl_refs in
 			List.iter (function
@@ -167,9 +174,10 @@ let database_callback event db =
 			let other_tbl_refs =
 				List.fold_left (fun accu (tbl,fld) ->
 					let fld_value = List.assoc fld kv in
-					if is_valid_ref fld_value 
-					then (tbl, fld_value, find_get_record tbl ~__context:context ~self:fld_value) :: accu
-					else accu) 
+                                        if is_valid_ref fld_value then begin
+                                                let fld_value = Schema.Value.Unsafe_cast.string fld_value in
+					         (tbl, fld_value, find_get_record tbl ~__context:context ~self:fld_value) :: accu
+                                        end else accu) 
 					[] other_tbl_refs in
 			let other_tbl_events = events_of_other_tbl_refs other_tbl_refs in
 			begin match snapshot() with
diff --git a/ocaml/database/redo_log.ml b/ocaml/database/redo_log.ml
index 047228cb4c0..19aebbeab75 100644
--- a/ocaml/database/redo_log.ml
+++ b/ocaml/database/redo_log.ml
@@ -759,9 +759,9 @@ let database_callback event db =
 			| Db_cache_types.RefreshRow (tblname, objref) ->
 				None
 			| Db_cache_types.WriteField (tblname, objref, fldname, oldval, newval) ->
-				R.debug "WriteField(%s, %s, %s, %s, %s)" tblname objref fldname oldval newval;
+				R.debug "WriteField(%s, %s, %s, %s, %s)" tblname objref fldname (Schema.Value.marshal oldval) (Schema.Value.marshal newval);
 				if Schema.is_field_persistent (Db_cache_types.Database.schema db) tblname fldname 
-				then Some (WriteField(tblname, objref, fldname, newval))
+				then Some (WriteField(tblname, objref, fldname, Schema.Value.marshal newval))
 				else None
 			| Db_cache_types.PreDelete (tblname, objref) ->
 				None
@@ -771,7 +771,7 @@ let database_callback event db =
 				else None
 			| Db_cache_types.Create (tblname, objref, kvs) ->
 				if Schema.is_table_persistent (Db_cache_types.Database.schema db) tblname
-				then Some (CreateRow(tblname, objref, kvs))
+				then Some (CreateRow(tblname, objref, (List.map (fun (k, v) -> k, Schema.Value.marshal v) kvs)))
 				else None 
 	in
 
diff --git a/ocaml/database/ref_index.ml b/ocaml/database/ref_index.ml
index c0625036a5d..676e26f8806 100644
--- a/ocaml/database/ref_index.ml
+++ b/ocaml/database/ref_index.ml
@@ -30,9 +30,9 @@ let lookup key =
 	let db = Db_ref.get_database t in
 	let r (tblname, objref) = 
 		let row = Table.find objref (TableSet.find tblname (Database.tableset db)) in {
-			name_label = (try Some (Row.find Db_names.name_label row) with _ -> None);
-			uuid = Row.find Db_names.uuid row;
-			_ref = Row.find Db_names.ref row;
+			name_label = (try Some (Schema.Value.Unsafe_cast.string (Row.find Db_names.name_label row)) with _ -> None);
+			uuid = Schema.Value.Unsafe_cast.string (Row.find Db_names.uuid row);
+			_ref = Schema.Value.Unsafe_cast.string (Row.find Db_names.ref row);
 		} in
 	Opt.map r (Database.lookup_key key db)
 
diff --git a/ocaml/database/schema.ml b/ocaml/database/schema.ml
index 96e710c8c9e..d0b32b6aa0c 100644
--- a/ocaml/database/schema.ml
+++ b/ocaml/database/schema.ml
@@ -11,16 +11,69 @@
  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  * GNU Lesser General Public License for more details.
  *)
+open Sexplib.Std
+
+module Type = struct
+	type t =
+	| String
+	| Set (* of strings *)
+	| Pairs (* of string * string *)
+	with sexp
+
+        exception Error of t * t
+        let _ = Printexc.register_printer (function
+                | Error (expected, actual) ->
+                        Some (Printf.sprintf "Schema.Type.Error: expected %s; received %s"
+                                (Sexplib.Sexp.to_string_hum (sexp_of_t expected))
+                                (Sexplib.Sexp.to_string_hum (sexp_of_t actual)))
+                | _ -> None
+        )
+end
+
+module Value = struct
+        type t =
+        | String of string
+        | Set of string list
+        | Pairs of (string * string) list
+        with sexp
+
+        let marshal = function
+        | String x -> x
+        | Set xs -> String_marshall_helper.set (fun x -> x) xs
+        | Pairs xs -> String_marshall_helper.map (fun x -> x) (fun x -> x) xs
+
+        let unmarshal ty x = match ty with
+        | Type.String -> String x
+        | Type.Set -> Set (String_unmarshall_helper.set (fun x -> x) x)
+        | Type.Pairs -> Pairs (String_unmarshall_helper.map (fun x -> x) (fun x -> x) x)
+
+        module Unsafe_cast = struct
+                let string = function
+                | String x -> x
+                | Set _ -> raise (Type.Error(Type.String, Type.Set))
+                | Pairs _ -> raise (Type.Error(Type.String, Type.Pairs))
+
+                let set = function
+                | Set xs -> xs
+                | String _ -> raise (Type.Error(Type.Set, Type.String))
+                | Pairs _ -> raise (Type.Error(Type.Set, Type.Pairs))
+
+                let pairs = function
+                | Pairs x -> x
+                | String _ -> raise (Type.Error(Type.Pairs, Type.String))
+                | Set _ -> raise (Type.Error(Type.Pairs, Type.Set))
+        end
+end
 
 module Column = struct
 	type t = {
 		name: string;
 		persistent: bool;         (** see is_field_persistent *)
-		empty: string;            (** fresh value used when loading non-persistent fields *)
-		default: string option;   (** if column is missing, this is default value is used *)
-
+		empty: Value.t;           (** fresh value used when loading non-persistent fields *)
+		default: Value.t option;  (** if column is missing, this is default value is used *)
+		ty: Type.t;               (** the type of the value in the column *)
 		issetref: bool;           (** only so we can special case set refs in the interface *)
-	}
+	} with sexp
 end
 
 module Table = struct
@@ -28,33 +81,53 @@ module Table = struct
 		name: string;
 		columns: Column.t list;
 		persistent: bool;
-	}
+	} with sexp
 	let find name t = List.find (fun col -> col.Column.name = name) t.columns
 end
 
 type relationship = 
 	| OneToMany of string * string * string * string
+	with sexp
 
 module Database = struct
 	type t = {
 		tables: Table.t list;
-	}
+	} with sexp
+
 	let find name t = List.find (fun tbl -> tbl.Table.name = name) t.tables
 end
 
-module StringMap = Map.Make(struct
-	type t = string
-	let compare = Pervasives.compare
-end)
+(** indexed by table name, a list of (this field, foreign table, foreign field) *)
+type foreign = (string * string * string) list
+with sexp
+
+module ForeignMap = struct
+	include Map.Make(struct
+		type t = string
+		let compare = Pervasives.compare
+	end)
+
+	type t' = (string * foreign) list
+	with sexp
+
+	type m = foreign t
+	let sexp_of_m t : Sexplib.Sexp.t =
+		let t' = fold (fun key foreign acc -> (key, foreign) :: acc) t [] in
+		sexp_of_t' t'
+
+	let m_of_sexp sexp : m =
+		let t' = t'_of_sexp sexp in
+		List.fold_left (fun acc (key, foreign) -> add key foreign acc) empty t'
+end
 
 type t = {
 	major_vsn: int;
 	minor_vsn: int;
 	database: Database.t;
 	(** indexed by table name, a list of (this field, foreign table, foreign field) *)
-	one_to_many: ((string * string * string) list) StringMap.t;
-	many_to_many: ((string * string * string) list) StringMap.t;
-}
+	one_to_many: ForeignMap.m;
+	many_to_many: ForeignMap.m;
+} with sexp
 
 let database x = x.database
 
@@ -69,8 +142,8 @@ let empty = {
 	major_vsn = 0;
 	minor_vsn = 0;
 	database = { Database.tables = [] };
-	one_to_many = StringMap.empty;
-	many_to_many = StringMap.empty;
+	one_to_many = ForeignMap.empty;
+	many_to_many = ForeignMap.empty;
 }
 
 let is_table_persistent schema tblname = 
@@ -89,12 +162,12 @@ open D
 let one_to_many tblname schema = 
 	(* If there is no entry in the map it means that the table has no one-to-many relationships *)
 	try
-		StringMap.find tblname schema.one_to_many
+		ForeignMap.find tblname schema.one_to_many
 	with Not_found -> []
 
 let many_to_many tblname schema = 
 	(* If there is no entry in the map it means that the table has no many-to-many relationships *)
 	try
-		StringMap.find tblname schema.many_to_many
+		ForeignMap.find tblname schema.many_to_many
 	with Not_found -> []
 
diff --git a/ocaml/database/test_schemas.ml b/ocaml/database/test_schemas.ml
index 2831a6fe799..90452e57a0c 100644
--- a/ocaml/database/test_schemas.ml
+++ b/ocaml/database/test_schemas.ml
@@ -2,64 +2,73 @@ let schema =
 	let _ref = {
 		Schema.Column.name = Db_names.ref;
 		persistent = true;
-		empty = "";
+		empty = Schema.Value.String "";
 		default = None;
+		ty = Schema.Type.String;
 		issetref = false;
 	} in
 	let uuid = {
 		Schema.Column.name = Db_names.uuid;
 		persistent = true;
-		empty = "";
+		empty = Schema.Value.String "";
 		default = None;
+		ty = Schema.Type.String;
 		issetref = false;
 	} in
 	let name_label = {
 		Schema.Column.name = Db_names.name_label;
 		persistent = true;
-		empty = "";
+		empty = Schema.Value.String "";
 		default = None;
+		ty = Schema.Type.String;
 		issetref = false;
 	} in
 	let name_description = {
 		Schema.Column.name = "name__description";
 		persistent = true;
-		empty = "";
+		empty = Schema.Value.String "";
 		default = None;
+		ty = Schema.Type.String;
 		issetref = false;
 	} in
 	let vbds = {
 		Schema.Column.name = "VBDs";
 		persistent = false;
-		empty = "()";
-		default = Some("()");
+                empty = Schema.Value.Set [];
+                default = Some(Schema.Value.Set []);
+		ty = Schema.Type.Set;
 		issetref = true;
 	} in
 	let other_config = {
 		Schema.Column.name = "other_config";
 		persistent = false;
-		empty = "()";
-		default = Some("()");
+                empty = Schema.Value.Pairs [];
+                default = Some(Schema.Value.Pairs []);
+		ty = Schema.Type.Pairs;
 		issetref = false;
 	} in
 	let pp = {
 		Schema.Column.name = "protection_policy";
 		persistent = true;
-		empty = "";
-		default = Some("OpaqueRef:NULL");
+                empty = Schema.Value.String "";
+		default = Some(Schema.Value.String "OpaqueRef:NULL");
+		ty = Schema.Type.String;
 		issetref = false;
 	} in
 	let tags = {
 		Schema.Column.name = "tags";
 		persistent = true;
-		empty = "";
-		default = Some("()");
+                empty = Schema.Value.Set [];
+                default = Some(Schema.Value.Set []);
+		ty = Schema.Type.Set;
 		issetref = false;
 	} in
 	let vm = {
 		Schema.Column.name = "VM";
 		persistent = true;
-		empty = "";
+                empty = Schema.Value.String "";
 		default = None;
+		ty = Schema.Type.String;
 		issetref = false;
 	} in
 
@@ -76,7 +85,7 @@ let schema =
 	let database = { 
 		Schema.Database.tables = [ vm_table; vbd_table ];
 	} in
-	let one_to_many = Schema.StringMap.add "VBD" [ "VM", "VM", "VBDs" ] (Schema.StringMap.empty) in
+	let one_to_many = Schema.ForeignMap.add "VBD" [ "VM", "VM", "VBDs" ] (Schema.ForeignMap.empty) in
 	{
 		
 		Schema.major_vsn = 1;
@@ -84,15 +93,16 @@ let schema =
 		database = database;
 		(** indexed by table name, a list of (this field, foreign table, foreign field) *)
 		one_to_many = one_to_many;
-		many_to_many = Schema.StringMap.empty;
+		many_to_many = Schema.ForeignMap.empty;
 	}
 
 
 let many_to_many =
 	let bar_column = { Schema.Column.name = "bars";
 					   persistent = false;
-					   empty = "()";
+                                           empty = Schema.Value.Pairs [];
 					   default = None;
+					   ty = Schema.Type.Pairs;
 					   issetref = false;
 					 } in
 	let foo_column = { bar_column with Schema.Column.name = "foos" } in
@@ -101,9 +111,9 @@ let many_to_many =
 	
 	let database = { Schema.Database.tables = [ foo_table; bar_table ] } in
 	let many_to_many = 
-		Schema.StringMap.add "foo" [ "bars", "bar", "foos" ]
-			(Schema.StringMap.add "bar" [ "foos", "foo", "bars" ]
-				 Schema.StringMap.empty) in
+		Schema.ForeignMap.add "foo" [ "bars", "bar", "foos" ]
+			(Schema.ForeignMap.add "bar" [ "foos", "foo", "bars" ]
+				 Schema.ForeignMap.empty) in
 	let schema = { Schema.empty with
 					   Schema.database = database;
 					   many_to_many = many_to_many 
diff --git a/ocaml/db_process/xapi_db_process.ml b/ocaml/db_process/xapi_db_process.ml
index ba3cc0baddb..bc24da824fa 100644
--- a/ocaml/db_process/xapi_db_process.ml
+++ b/ocaml/db_process/xapi_db_process.ml
@@ -99,7 +99,7 @@ let find_my_host_row() =
   let localhost_uuid = Xapi_inventory.lookup Xapi_inventory._installation_uuid in
   let db = Db_ref.get_database (Db_backend.make ()) in
   let tbl = TableSet.find Db_names.host (Database.tableset db) in
-  Table.fold (fun r _ _ row acc -> if Row.find Db_names.uuid row = localhost_uuid then (Some (r, row)) else acc) tbl None
+  Table.fold (fun r _ row acc -> if Schema.Value.Unsafe_cast.string (Row.find Db_names.uuid row) = localhost_uuid then (Some (r, row)) else acc) tbl None
 
 let _iscsi_iqn = "iscsi_iqn"
 let _other_config = "other_config"
@@ -109,8 +109,7 @@ let do_read_hostiqn() =
   match find_my_host_row() with
 	  | None -> failwith "No row for localhost"
 	  | Some (_, row) ->
-		  let other_config_sexpr = Row.find Db_names.other_config row in
-		  let other_config = String_unmarshall_helper.map (fun x->x) (fun x->x) other_config_sexpr in
+		  let other_config = Schema.Value.Unsafe_cast.pairs (Row.find Db_names.other_config row) in
 		  Printf.printf "%s" (List.assoc _iscsi_iqn other_config)
 
 let do_write_hostiqn() =
@@ -122,8 +121,7 @@ let do_write_hostiqn() =
 	  | None -> failwith "No row for localhost"
 	  | Some (r, row) ->
 		  (* read other_config from my row, replace host_iqn if already there, add it if its not there and write back *)
-		  let other_config_sexpr = Row.find Db_names.other_config row in
-		  let other_config = String_unmarshall_helper.map (fun x->x) (fun x->x) other_config_sexpr in
+		  let other_config = Schema.Value.Unsafe_cast.pairs (Row.find Db_names.other_config row) in
 		  let other_config =
 			  if List.mem_assoc _iscsi_iqn other_config then
 				  (* replace if key already exists *)
@@ -131,7 +129,7 @@ let do_write_hostiqn() =
 			  else
 				  (* ... otherwise add new key/value pair *)
 				  (_iscsi_iqn,new_iqn)::other_config in
-		  let other_config = String_marshall_helper.map (fun x->x) (fun x->x) other_config in
+		  let other_config = Schema.Value.Pairs other_config in
 		  Db_ref.update_database (Db_backend.make ()) (set_field Db_names.host r Db_names.other_config other_config);
 		  write_out_databases()
 
diff --git a/ocaml/idl/datamodel_schema.ml b/ocaml/idl/datamodel_schema.ml
index 17119bced14..05969a4d99d 100644
--- a/ocaml/idl/datamodel_schema.ml
+++ b/ocaml/idl/datamodel_schema.ml
@@ -30,6 +30,10 @@ let of_datamodel () =
 			let this = obj.Datamodel_types.name, f.Datamodel_types.field_name in
 			Datamodel_utils.Relations.is_in_relation api this && 
 				(Datamodel_utils.Relations.classify api (this,(Datamodel_utils.Relations.other_end_of api this)) = (`Many, `Many)) in
+		let ty = match f.Datamodel_types.ty with
+			| Datamodel_types.Set _ -> Type.Set
+			| Datamodel_types.Map(_,_) -> Type.Pairs
+			| _ -> Type.String in
 		{
 			Column.name = Escaping.escape_id f.Datamodel_types.full_name;
 			(* NB we always regenerate one-to-many Set(Ref _) fields *)
@@ -38,8 +42,9 @@ let of_datamodel () =
 			(* NB Set(Ref _) fields aren't allowed to have a default value specified so we hardcode one here *)
 			default = 
 				if issetref 
-				then Some (SExpr.string_of (SExpr.Node []))
-				else Opt.map Datamodel_values.to_db_string f.Datamodel_types.default_value ;
+                                then Some (Value.Set [])
+				else Opt.map Datamodel_values.to_db f.Datamodel_types.default_value ;
+			ty = ty;
 			issetref = issetref;
 		} in
 
@@ -47,8 +52,9 @@ let of_datamodel () =
 	let _ref = {
 		Column.name = Db_names.ref;
 		persistent = true;
-		empty = "";
+		empty = Value.String "";
 		default = None;
+		ty = Type.String;
 		issetref = false;
 	} in
 
@@ -66,9 +72,9 @@ let of_datamodel () =
 			| `Many, `Many -> true
 			| _ -> false in
 	let add_relation p t (((one_tbl, one_fld), (many_tbl, many_fld)) as r) =
-		let l = if StringMap.mem one_tbl t then StringMap.find one_tbl t else [] in
+		let l = if ForeignMap.mem one_tbl t then ForeignMap.find one_tbl t else [] in
 		if p r 
-		then StringMap.add one_tbl ((one_fld, many_tbl, many_fld) :: l) t
+		then ForeignMap.add one_tbl ((one_fld, many_tbl, many_fld) :: l) t
 		else t in
 
 	let database api = {
@@ -78,6 +84,16 @@ let of_datamodel () =
 		major_vsn = Datamodel.schema_major_vsn;
 		minor_vsn = Datamodel.schema_minor_vsn;
 		database = database Datamodel.all_api;
-		one_to_many = List.fold_left (add_relation is_one_to_many) StringMap.empty (Dm_api.relations_of_api Datamodel.all_api);
-		many_to_many = List.fold_left (add_relation is_many_to_many) StringMap.empty (Dm_api.relations_of_api Datamodel.all_api);
+		one_to_many = List.fold_left (add_relation is_one_to_many) ForeignMap.empty (Dm_api.relations_of_api Datamodel.all_api);
+		many_to_many = List.fold_left (add_relation is_many_to_many) ForeignMap.empty (Dm_api.relations_of_api Datamodel.all_api);
 	}
+
+(* For now this is a convenience debugging function. Eventually we should
+   separate the datamodel from the database and load the schema from disk. *)
+let write_schema_to_file filename =
+	let t = of_datamodel () in
+	let sexp = Schema.sexp_of_t t in
+	let oc = open_out filename in
+	let txt = Sexplib.Sexp.to_string_hum sexp in
+	output_string oc txt;
+	close_out oc
diff --git a/ocaml/idl/datamodel_values.ml b/ocaml/idl/datamodel_values.ml
index dc518ce8729..c6a1429efb5 100644
--- a/ocaml/idl/datamodel_values.ml
+++ b/ocaml/idl/datamodel_values.ml
@@ -62,30 +62,34 @@ let to_ocaml_string v =
 		| Rpc.DateTime t -> sprintf "Rpc.DateTime %s" t in
 	aux (to_rpc v)
       
-let rec to_db_string v =
+let rec to_db v =
+  let open Schema.Value in
   match v with
-    VString s -> s
-  | VInt i -> Int64.to_string i
-  | VFloat f -> string_of_float f
-  | VBool true -> "true"
-  | VBool false -> "false"
-  | VDateTime d -> Date.to_string d
-  | VEnum e -> e
-  | VMap vvl -> String_marshall_helper.map to_db_string to_db_string vvl
-  | VSet vl -> String_marshall_helper.set to_db_string vl
-  | VRef r -> r
+    VString s -> String s
+  | VInt i -> String (Int64.to_string i)
+  | VFloat f -> String (string_of_float f)
+  | VBool true -> String "true"
+  | VBool false -> String "false"
+  | VDateTime d -> String (Date.to_string d)
+  | VEnum e -> String e
+  | VMap vvl ->
+    Pairs(List.map (fun (k, v) -> to_string k, to_string v) vvl)
+  | VSet vl ->
+    Set(List.map to_string vl)
+  | VRef r -> String r
       
 (* Generate suitable "empty" database value of specified type *)
 let gen_empty_db_val t =
+  let open Schema in
   match t with
-  | String -> ""
-  | Int -> "0"
-  | Float -> string_of_float 0.0
-  | Bool -> "false"
-  | DateTime -> Date.to_string Date.never
-  | Enum (_,(enum_value,_)::_) -> enum_value
+  | String -> Value.String ""
+  | Int -> Value.String "0"
+  | Float -> Value.String (string_of_float 0.0)
+  | Bool -> Value.String "false"
+  | DateTime -> Value.String (Date.to_string Date.never)
+  | Enum (_,(enum_value,_)::_) -> Value.String enum_value
   | Enum (_, []) -> assert false
-  | Set _ -> String_marshall_helper.map to_db_string to_db_string []
-  | Map _ -> String_marshall_helper.set to_db_string []
-  | Ref _ -> Ref.string_of Ref.null
-  | Record _ -> ""
+  | Set _ -> Value.Set []
+  | Map _ -> Value.Pairs []
+  | Ref _ -> Value.String (Ref.string_of Ref.null)
+  | Record _ -> Value.String ""
diff --git a/ocaml/idl/ocaml_backend/exnHelper.ml b/ocaml/idl/ocaml_backend/exnHelper.ml
index 59d00a36613..3272d0bb53d 100644
--- a/ocaml/idl/ocaml_backend/exnHelper.ml
+++ b/ocaml/idl/ocaml_backend/exnHelper.ml
@@ -33,8 +33,6 @@ let error_of_exn e =
 		| Db_exn.Too_many_values(tbl, objref, uuid) ->
 			(* Very bad: database has duplicate references or UUIDs *)
 			internal_error, [ sprintf "duplicate objects in database: tbl='%s'; object_ref='%s'; uuid='%s'" tbl objref uuid ]
-		| Db_action_helper.Db_set_or_map_parse_fail s ->
-			internal_error, [ sprintf "db set/map failure: %s" s ]
 		| Db_exn.DBCache_NotFound (reason,p1,p2) ->
 			begin
 				match reason with
diff --git a/ocaml/xapi/xapi_event.ml b/ocaml/xapi/xapi_event.ml
index 5d13669e80d..4342f8001a5 100644
--- a/ocaml/xapi/xapi_event.ml
+++ b/ocaml/xapi/xapi_event.ml
@@ -434,24 +434,31 @@ let from_inner __context session subs from from_t deadline =
 			if Subscription.table_matches subs "message" then (!Message.get_since_for_events) ~__context !last_msg_gen else (0L, []) in
 		(msg_gen, messages, tableset, List.fold_left
 			(fun acc table ->
-		 		Db_cache_types.Table.fold_over_recent !last_generation
-					 (fun ctime mtime dtime objref (creates,mods,deletes,last) ->
+                                (* Fold over the live objects *)
+                                let acc = Db_cache_types.Table.fold_over_recent !last_generation
+                                         (fun objref { Db_cache_types.Stat.created; modified; deleted } _ (creates,mods,deletes,last) ->
 						if Subscription.object_matches subs (String.lowercase table) objref then begin
-							let last = max last (max mtime dtime) in (* mtime guaranteed to always be larger than ctime *)
-							if dtime > 0L then begin
-								if ctime > !last_generation then
-									(creates,mods,deletes,last) (* It was created and destroyed since the last update *)
-								else
-									(creates,mods,(table, objref, dtime)::deletes,last) (* It might have been modified, but we can't tell now *)
-							end else begin
-								((if ctime > !last_generation then (table, objref, ctime)::creates else creates),
-								(if mtime > !last_generation then (table, objref, mtime)::mods else mods),
-								deletes, last)
-							end 
+							let last = max last (max modified deleted) in (* mtime guaranteed to always be larger than ctime *)
+							((if created > !last_generation then (table, objref, created)::creates else creates),
+							(if modified > !last_generation then (table, objref, modified)::mods else mods),
+							deletes, last)
 						end else begin
 							(creates,mods,deletes,last)
 						end
-					) (fun () -> ()) (Db_cache_types.TableSet.find table tableset) acc
+					) (Db_cache_types.TableSet.find table tableset) acc in
+                                (* Fold over the deleted objects *)
+		 		Db_cache_types.Table.fold_over_deleted !last_generation
+                                         (fun objref { Db_cache_types.Stat.created; modified; deleted } (creates,mods,deletes,last) ->
+						if Subscription.object_matches subs (String.lowercase table) objref then begin
+							let last = max last (max modified deleted) in (* mtime guaranteed to always be larger than ctime *)
+							if created > !last_generation then
+								(creates,mods,deletes,last) (* It was created and destroyed since the last update *)
+							else
+								(creates,mods,(table, objref, deleted)::deletes,last) (* It might have been modified, but we can't tell now *)
+						end else begin
+							(creates,mods,deletes,last)
+						end
+					) (Db_cache_types.TableSet.find table tableset) acc
 			) ([],[],[],!last_generation) tables) in
 
 	(* Each event.from should have an independent subscription record *)
@@ -505,10 +512,10 @@ let from_inner __context session subs from from_t deadline =
 
 	let valid_ref_counts =
 		Db_cache_types.TableSet.fold
-			(fun tablename _ _ table acc ->
+			(fun tablename _ table acc ->
 				(String.lowercase tablename,
 					(Db_cache_types.Table.fold
-						(fun r _ _ _ acc -> Int32.add 1l acc) table 0l))::acc)
+						(fun r _ _ acc -> Int32.add 1l acc) table 0l))::acc)
 		tableset [] in
 
 	{
@@ -551,7 +558,7 @@ let inject ~__context ~_class ~_ref =
 		(fun () ->
 			let db_ref = Db_backend.make () in
 			let g = Manifest.generation (Database.manifest (Db_ref.get_database db_ref)) in
-			Db_cache_impl.refresh_row db_ref _class _ref; (* consumes this generation *)
+			Db_cache_impl.touch_row db_ref _class _ref; (* consumes this generation *)
 			g
 		) in
 	let token = Int64.sub generation 1L, 0L in
diff --git a/ocaml/xenops/xn_cfg_parser.mly b/ocaml/xenops/xn_cfg_parser.mly
new file mode 100644
index 00000000000..3c549fb2614
--- /dev/null
+++ b/ocaml/xenops/xn_cfg_parser.mly
@@ -0,0 +1,17 @@
+(*
+ * Copyright (C) 2006-2009 Citrix Systems Inc.
+ *
+ * This program is free software; you can redistribute it and/or modify
+ * it under the terms of the GNU Lesser General Public License as published
+ * by the Free Software Foundation; version 2.1 only. with the special
+ * exception on linking described in file LICENSE.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Lesser General Public License for more details.
+ *)
+
+(** Takes a function which is supplied with an fd representing the input to the
+    sha1sum and returns the checksum as a string *)
+val sha1sum: (Unix.file_descr -> unit) -> string
diff --git a/opam b/opam
index 1c886287aae..9a7e9551fa7 100644
--- a/opam
+++ b/opam
@@ -10,7 +10,6 @@ depends: [
   "ocamlfind"
   "xapi-idl"
   "xapi-libs-transitional"
-  "xen-api-client"
   "xapi-netdev"
   "omake"
   "cdrom"