allow subtables of composite key tables

lib/mosql/schema.rb

@@ -54,6 +54,10 @@ def check_columns!(ns, spec)
       end
     end
 
+    def parent_scope_column(parent, colname)
+      (parent.to_s.singularize + "_" + colname.to_s).to_sym
+    end
+
     def parse_spec(ns, spec, source=[])
       out = spec.dup
       out[:columns] = to_array(spec.delete(:columns))
@@ -62,7 +66,7 @@ def parse_spec(ns, spec, source=[])
         newsource = source + [name]
         subspec = parse_spec(ns , subspec,  newsource)
         subspec[:meta][:source] = newsource
-        subspec[:meta][:parent_fkey] = (meta[:table].to_s.singularize + "_id").to_sym
+        subspec[:meta][:parent_fkeys] = primary_sql_keys_for_schema(out).map { |k| parent_scope_column(meta[:table], k) }
         subspec
       end
       check_columns!(ns, out)
@@ -103,12 +107,11 @@ def qualified_table_name(meta)
        end
     end
 
-    def create_table(db, spec, clobber, parent_table=nil, parent_pk_type = nil)
+    def create_table(db, spec, clobber, parent_table=nil, parent_pks={})
         meta = spec[:meta]
         table_name = qualified_table_name(meta)
         composite_key = meta[:composite_key]
-        keys = []
-        keytypes = []
+        primary_keys = {}
         log.info("Creating table #{db.literal(table_name)}...")
         db.drop_table?(table_name, :cascade => true) if clobber
         db.create_table(table_name) do
@@ -129,11 +132,9 @@ def create_table(db, spec, clobber, parent_table=nil, parent_pk_type = nil)
             column col[:name], col[:type], opts
 
             if composite_key and composite_key.include?(col[:name])
-              keys << col[:name].to_sym
-              keytypes << col[:type]
+              primary_keys[col[:name].to_sym] = col[:type]
             elsif not composite_key and col[:source].to_sym == :_id
-              keys << col[:name].to_sym
-              keytypes << col[:type]
+              primary_keys[col[:name].to_sym] = col[:type]
             end
           end
 
@@ -151,20 +152,20 @@ def create_table(db, spec, clobber, parent_table=nil, parent_pk_type = nil)
           end
 
           if !parent_table.nil?
-            foreign_key meta[:parent_fkey], parent_table, {
-                :type => parent_pk_type,
+            parent_pks.each do |k, type|
+              column k, type
+            end
+            foreign_key parent_pks.keys, parent_table, {
                 :on_delete => :cascade,
                 :on_update => :cascade
             }
-            keys << meta[:parent_fkey]
-            keytypes << parent_pk_type
           end
-          primary_key keys
+          primary_key primary_keys.keys + parent_pks.keys
         end
 
+        parent_pks = Hash[primary_keys.map { |k, t| [parent_scope_column(meta[:table], k), t] }].merge(parent_pks)
         spec[:subtables].each do |subspec|
-          raise "Too many keys for sub table in #{table_name}: #{keys}" unless keys.length == 1
-          create_table(db, subspec, clobber, table_name, keytypes.first)
+          create_table(db, subspec, clobber, table_name, parent_pks)
         end
     end
 
@@ -316,13 +317,13 @@ def transform_one(schema, obj)
           raise "Invalid null #{source.inspect} for #{pks.inspect}"
         elsif v.is_a? Sequel::SQL::Blob and type != "bytea"
           raise "Failed to convert binary #{source.inspect} to #{type.inspect} for #{pks.inspect}"
-        elsif col[:array_type]
+        elsif col[:array_type] and not v.nil?
           v.each_with_index do |e, i|
             if not sanity_check_type(e, col[:array_type])
               raise "Failed to convert array element #{i} of #{source.inspect} to #{type.inspect}: got #{e.inspect} for #{pks.inspect}"
             end
          end
-        elsif not sanity_check_type(v, type)
+        elsif not v.nil? and not sanity_check_type(v, type)
           raise "Failed to convert #{source.inspect} to #{type.inspect}: got #{v.inspect} for #{pks.inspect}"
         end
         row[name] = v
@@ -417,15 +418,18 @@ def all_transforms_for_obj(schema, obj, parent_pks={}, &block)
 
       yield table_ident, primary_keys, transformed
 
+      parent_pks = Hash[primary_keys.map { |k| [
+            parent_scope_column(schema[:meta][:table], k),
+            transformed[k]
+      ] } ]
+      parent_pks = parent_pks.to_h
       schema[:subtables].each do |subspec|
         source = subspec[:meta][:source]
         subobjs = bson_dig(obj, *source)
         break if subobjs.nil?
 
-        raise "Too many primary keys" if primary_keys.length > 1
-        pks = {subspec[:meta][:parent_fkey] => transformed[primary_keys[0]]}
         subobjs.each do |subobj|
-          all_transforms_for_obj(subspec, subobj, pks, &block)
+          all_transforms_for_obj(subspec, subobj, parent_pks, &block)
         end
       end
     end
@@ -452,8 +456,8 @@ def primary_sql_keys_for_schema(schema)
       else
         keys << schema[:columns].find {|c| c[:source] == '_id'}[:name]
       end
-      if schema[:meta][:parent_fkey]
-        keys << schema[:meta][:parent_fkey]
+      if schema[:meta][:parent_fkeys]
+        keys += schema[:meta][:parent_fkeys]
       end
 
       return keys