diff --git a/kong/db/schema/init.lua b/kong/db/schema/init.lua
index a0a4b49be6d5..abea07533033 100644
--- a/kong/db/schema/init.lua
+++ b/kong/db/schema/init.lua
@@ -78,7 +78,8 @@ local validation_errors = {
   NOT_PK                    = "not a primary key",
   MISSING_PK                = "missing primary key",
   -- subschemas
-  SUBSCHEMA_NOT_FOUND       = "module not found for entity: %s",
+  SUBSCHEMA_UNKNOWN         = "unknown type: %s",
+  SUBSCHEMA_BAD_PARENT      = "entities of type '%s' cannot have subschemas",
   SUBSCHEMA_UNDEFINED_FIELD = "error in schema definition: abstract field was not specialized",
   SUBSCHEMA_BAD_TYPE        = "error in schema definition: cannot change type in a specialized field",
 }
@@ -1058,28 +1059,6 @@ function Schema:process_auto_fields(input, context)
 end
 
 
--- @param self the Schema object.
-local function load_subschema(self, input)
-  local patt = self.subschema_module
-  local key = input[self.subschema_key]
-  local mod_name = patt:gsub("%?", key)
-  local pok, definition = pcall(require, mod_name)
-  if not pok then
-    return nil, validation_errors.SUBSCHEMA_NOT_FOUND:format(mod_name)
-  end
-
-  local subschema, err = Schema.new(definition)
-  if not subschema then
-    return nil, err
-  end
-
-  if not self.subschemas then
-    self.subschemas = {}
-  end
-  self.subschemas[key] = subschema
-end
-
-
 --- Validate a table against the schema, ensuring that the entity is complete.
 -- It validates fields for their attributes,
 -- and runs the global entity checks against the entire table.
@@ -1110,13 +1089,20 @@ function Schema:validate(input, full_check)
     full_check = true
   end
 
-  local ok, subschema_error, _
+  local ok, subschema_error
   if self.subschema_key then
-    _, subschema_error = load_subschema(self, input)
+    local key = input[self.subschema_key]
+    if not (self.subschemas and self.subschemas[key]) then
+      subschema_error = validation_errors.SUBSCHEMA_UNKNOWN:format(key)
+    end
   end
 
   local _, field_errors = validate_fields(self, input)
 
+  if subschema_error then
+    field_errors[self.subschema_key] = subschema_error
+  end
+
   for name, field in self:each_field() do
     if field.required
        and (input[name] == null
@@ -1134,11 +1120,6 @@ function Schema:validate(input, full_check)
     merge_into_table(field_errors, f_errs)
   end
 
-  if subschema_error then
-    field_errors["@entity"] = field_errors["@entity"] or {}
-    table.insert(field_errors["@entity"], subschema_error)
-  end
-
   if next(field_errors) then
     return nil, field_errors
   end
@@ -1338,6 +1319,28 @@ function Schema.new(definition)
 end
 
 
+function Schema.new_subschema(self, key, definition)
+  assert(type(key) == "string", "key must be a string")
+  assert(type(definition) == "table", "definition must be a table")
+
+  if not self.subschema_key then
+    return nil, validation_errors.SUBSCHEMA_BAD_PARENT:format(self.name)
+  end
+
+  local subschema, err = Schema.new(definition)
+  if not subschema then
+    return nil, err
+  end
+
+  if not self.subschemas then
+    self.subschemas = {}
+  end
+  self.subschemas[key] = subschema
+
+  return true
+end
+
+
 function Schema.define(tbl)
   return setmetatable(tbl, {
     __call = function(t, arg)
diff --git a/kong/db/schema/metaschema.lua b/kong/db/schema/metaschema.lua
index b6adec029cc2..b76f02d71dba 100644
--- a/kong/db/schema/metaschema.lua
+++ b/kong/db/schema/metaschema.lua
@@ -296,13 +296,6 @@ local MetaSchema = Schema.new({
         nilable = true,
       },
     },
-    {
-      subschema_module = {
-        type = "string",
-        nilable = true,
-        match = "^[^%?]*%?[^%?]*$",
-      },
-    },
     {
       fields = fields_array,
     },
diff --git a/spec/01-unit/000-new-dao/01-schema/01-schema_spec.lua b/spec/01-unit/000-new-dao/01-schema/01-schema_spec.lua
index dd1090b42a29..edf3b750e463 100644
--- a/spec/01-unit/000-new-dao/01-schema/01-schema_spec.lua
+++ b/spec/01-unit/000-new-dao/01-schema/01-schema_spec.lua
@@ -882,7 +882,15 @@ describe("schema", function()
 
     describe("subschemas", function()
       it("validates loading a subschema", function()
-        package.loaded["test.subschema.my_subschema"] = {
+        local Test = Schema.new({
+          name = "test",
+          subschema_key = "name",
+          fields = {
+            { name = { type = "string", required = true, } },
+            { config = { type = "record", abstract = true, } },
+          }
+        })
+        Test:new_subschema("my_subschema", {
           fields = {
             { config = {
                 type = "record",
@@ -892,15 +900,6 @@ describe("schema", function()
                 }
             } }
           }
-        }
-        local Test = Schema.new({
-          name = "test",
-          subschema_key = "name",
-          subschema_module = "test.subschema.?",
-          fields = {
-            { name = { type = "string", required = true, } },
-            { config = { type = "record", abstract = true, } },
-          }
         })
         assert.truthy(Test:validate({
           name = "my_subschema",
@@ -915,7 +914,6 @@ describe("schema", function()
         local Test = Schema.new({
           name = "test",
           subschema_key = "name",
-          subschema_module = "test.subschema.?",
           fields = {
             { name = { type = "string", required = true, } },
           }
@@ -925,14 +923,21 @@ describe("schema", function()
         })
         assert.falsy(ok)
         assert.same({
-          ["@entity"] = {
-            "module not found for entity: test.subschema.my_invalid_subschema",
-          }
+          ["name"] = "unknown type: my_invalid_subschema",
         }, errors)
       end)
 
       it("ignores missing non-required abstract fields", function()
-        package.loaded["test.subschema.my_subschema"] = {
+        local Test = Schema.new({
+          name = "test",
+          subschema_key = "name",
+          fields = {
+            { name = { type = "string", required = true, } },
+            { config = { type = "record", abstract = true, } },
+            { bla = { type = "integer", abstract = true, } },
+          }
+        })
+        Test:new_subschema("my_subschema", {
           fields = {
             { config = {
                 type = "record",
@@ -942,16 +947,6 @@ describe("schema", function()
                 }
             } }
           }
-        }
-        local Test = Schema.new({
-          name = "test",
-          subschema_key = "name",
-          subschema_module = "test.subschema.?",
-          fields = {
-            { name = { type = "string", required = true, } },
-            { config = { type = "record", abstract = true, } },
-            { bla = { type = "integer", abstract = true, } },
-          }
         })
         assert.truthy(Test:validate({
           name = "my_subschema",
@@ -963,32 +958,32 @@ describe("schema", function()
       end)
 
       it("cannot introduce new top-level fields", function()
-        package.loaded["test.subschema.my_subschema"] = {
+        local Test = Schema.new({
+          name = "test",
+          subschema_key = "name",
           fields = {
+            { name = { type = "string", required = true, } },
             { config = {
                 type = "record",
                 fields = {
                   { foo = { type = "integer" } },
-                  { bar = { type = "integer" } },
                 }
             } },
-            { new_field = { type = "string", required = true, } },
           }
-        }
-        local Test = Schema.new({
-          name = "test",
-          subschema_key = "name",
-          subschema_module = "test.subschema.?",
+        })
+        Test:new_subschema("my_subschema", {
           fields = {
-            { name = { type = "string", required = true, } },
             { config = {
                 type = "record",
                 fields = {
                   { foo = { type = "integer" } },
+                  { bar = { type = "integer" } },
                 }
             } },
+            { new_field = { type = "string", required = true, } },
           }
         })
+
         local ok, errors = Test:validate({
           name = "my_subschema",
           config = {
@@ -1003,7 +998,16 @@ describe("schema", function()
       end)
 
       it("fails when trying to use an abstract field (incomplete subschema)", function()
-        package.loaded["test.subschema.my_subschema"] = {
+        local Test = Schema.new({
+          name = "test",
+          subschema_key = "name",
+          fields = {
+            { name = { type = "string", required = true, } },
+            { config = { type = "record", abstract = true, } },
+            { bla = { type = "integer", abstract = true, } },
+          }
+        })
+        Test:new_subschema("my_subschema", {
           fields = {
             { config = {
                 type = "record",
@@ -1013,16 +1017,6 @@ describe("schema", function()
                 }
             } }
           }
-        }
-        local Test = Schema.new({
-          name = "test",
-          subschema_key = "name",
-          subschema_module = "test.subschema.?",
-          fields = {
-            { name = { type = "string", required = true, } },
-            { config = { type = "record", abstract = true, } },
-            { bla = { type = "integer", abstract = true, } },
-          }
         })
         local ok, errors = Test:validate({
           name = "my_subschema",
@@ -1039,7 +1033,16 @@ describe("schema", function()
       end)
 
       it("validates using both schema and subschema", function()
-        package.loaded["test.subschema.my_subschema"] = {
+        local Test = Schema.new({
+          name = "test",
+          subschema_key = "name",
+          fields = {
+            { name = { type = "string", required = true, } },
+            { bla = { type = "integer", } },
+            { config = { type = "record", abstract = true, } },
+          }
+        })
+        Test:new_subschema("my_subschema", {
           fields = {
             { config = {
                 type = "record",
@@ -1049,16 +1052,6 @@ describe("schema", function()
                 }
             } }
           }
-        }
-        local Test = Schema.new({
-          name = "test",
-          subschema_key = "name",
-          subschema_module = "test.subschema.?",
-          fields = {
-            { name = { type = "string", required = true, } },
-            { bla = { type = "integer", } },
-            { config = { type = "record", abstract = true, } },
-          }
         })
         local ok, errors = Test:validate({
           name = "my_subschema",
@@ -1078,26 +1071,25 @@ describe("schema", function()
       end)
 
       it("can specialize a field of the parent schema", function()
-        package.loaded["test.subschema.length_5"] = {
-          fields = {
-            { consumer = {
-                type = "string",
-                len_eq = 5,
-            } }
-          }
-        }
-        package.loaded["test.subschema.no_restrictions"] = {
-          fields = {}
-        }
         local Test = Schema.new({
           name = "test",
           subschema_key = "name",
-          subschema_module = "test.subschema.?",
           fields = {
             { name = { type = "string", required = true, } },
             { consumer = { type = "string", } },
           }
         })
+        Test:new_subschema("length_5", {
+          fields = {
+            { consumer = {
+                type = "string",
+                len_eq = 5,
+            } }
+          }
+        })
+        Test:new_subschema("no_restrictions", {
+          fields = {}
+        })
 
         local ok, errors = Test:validate({
           name = "length_5",
@@ -1123,25 +1115,24 @@ describe("schema", function()
       end)
 
       it("cannot change type when specializing a field", function()
-        package.loaded["test.subschema.length_5"] = {
-          fields = {
-            { consumer = {
-                type = "integer",
-            } }
-          }
-        }
-        package.loaded["test.subschema.no_restrictions"] = {
-          fields = {}
-        }
         local Test = Schema.new({
           name = "test",
           subschema_key = "name",
-          subschema_module = "test.subschema.?",
           fields = {
             { name = { type = "string", required = true, } },
             { consumer = { type = "string", } },
           }
         })
+        Test:new_subschema("length_5", {
+          fields = {
+            { consumer = {
+                type = "integer",
+            } }
+          }
+        })
+        Test:new_subschema("no_restrictions", {
+          fields = {}
+        })
 
         local ok, errors = Test:validate({
           name = "length_5",
@@ -1161,11 +1152,6 @@ describe("schema", function()
       end)
 
       it("a specialized field can force a value using 'eq'", function()
-        package.loaded["test.subschema.no_consumer"] = {
-          fields = {
-            { consumer = { type = "foreign", reference = "mock_consumers", eq = ngx.null } }
-          }
-        }
         package.loaded["kong.db.schema.entities.mock_consumers"] = {
           name = "mock_consumer",
           primary_key = { "id" },
@@ -1173,18 +1159,22 @@ describe("schema", function()
             { id = { type = "string" }, },
           }
         }
-        package.loaded["test.subschema.no_restrictions"] = {
-          fields = {}
-        }
         local Test = Schema.new({
           name = "test",
           subschema_key = "name",
-          subschema_module = "test.subschema.?",
           fields = {
             { name = { type = "string", required = true, } },
             { consumer = { type = "foreign", reference = "mock_consumers" } },
           }
         })
+        Test:new_subschema("no_consumer", {
+          fields = {
+            { consumer = { type = "foreign", reference = "mock_consumers", eq = ngx.null } }
+          }
+        })
+        Test:new_subschema("no_restrictions", {
+          fields = {}
+        })
 
         local ok, errors = Test:validate({
           name = "no_consumer",
diff --git a/spec/01-unit/000-new-dao/01-schema/02-metaschema_spec.lua b/spec/01-unit/000-new-dao/01-schema/02-metaschema_spec.lua
index 9294aac7433f..922e149ad089 100644
--- a/spec/01-unit/000-new-dao/01-schema/02-metaschema_spec.lua
+++ b/spec/01-unit/000-new-dao/01-schema/02-metaschema_spec.lua
@@ -292,7 +292,6 @@ describe("metaschema", function()
       local s = {
         name = "test",
         subschema_key = "str",
-        subschema_module = "foo.?",
         fields = {
           { str = { type = "string", unique = true } },
         },
@@ -301,49 +300,10 @@ describe("metaschema", function()
       assert.truthy(MetaSchema:validate(s))
     end)
 
-    it("subschema_module must have exactly one '?'", function()
-      local s = {
-        name = "test",
-        subschema_key = "str",
-        subschema_module = "foo",
-        fields = {
-          { str = { type = "string", unique = true } },
-        },
-        primary_key = { "str" },
-      }
-
-      -- no '?'
-      local ok, err = MetaSchema:validate(s)
-      assert.falsy(ok)
-      assert.match("invalid value", err.subschema_module)
-
-      -- one '?'
-      s.subschema_module = "?foofoo"
-      ok = MetaSchema:validate(s)
-      assert.truthy(ok)
-      s.subschema_module = "foo?foo"
-      ok = MetaSchema:validate(s)
-      assert.truthy(ok)
-      s.subschema_module = "foofoo?"
-      ok = MetaSchema:validate(s)
-      assert.truthy(ok)
-
-      -- two '?'
-      s.subschema_module = "?foo?"
-      ok, err = MetaSchema:validate(s)
-      assert.falsy(ok)
-      assert.match("invalid value", err.subschema_module)
-      s.subschema_module = "foo??"
-      ok, err = MetaSchema:validate(s)
-      assert.falsy(ok)
-      assert.match("invalid value", err.subschema_module)
-    end)
-
     it("subschema_key must be an existing field name", function()
       local s = {
         name = "test",
         subschema_key = "str",
-        subschema_module = "foo.?",
         fields = {
           { str = { type = "string", unique = true } },
         },
@@ -364,7 +324,6 @@ describe("metaschema", function()
       local s = {
         name = "test",
         subschema_key = "num",
-        subschema_module = "foo.?",
         fields = {
           { str = { type = "string", unique = true } },
           { num = { type = "number", unique = true } },
@@ -380,7 +339,6 @@ describe("metaschema", function()
       local s = {
         name = "test",
         subschema_key = "str",
-        subschema_module = "foo.?",
         fields = {
           { str = { type = "string", unique = true } },
           { num = { type = "number", abstract = true } },
@@ -396,7 +354,6 @@ describe("metaschema", function()
       local s = {
         name = "test",
         subschema_key = "str",
-        subschema_module = "foo.?",
         fields = {
           { str = { type = "string", unique = true } },
           -- abstract arrays, sets and maps need their types
@@ -424,7 +381,6 @@ describe("metaschema", function()
       s = {
         name = "test",
         subschema_key = "str",
-        subschema_module = "foo.?",
         fields = {
           { str = { type = "string", unique = true } },
           { arr = { type = "array", elements = { type = "string" }, abstract = true } },