fix(incremental sync): fix cache_key handling for select_by_cache_key (…

…#13815)

* fix(incremental sync): fix cache_key handling for select_by_cache_key

* fix some comments

* fix: remove unnecessary comment

kong/db/dao/init.lua

@@ -1555,7 +1555,7 @@ function DAO:cache_key(key, arg2, arg3, arg4, arg5, ws_id)
     error("key must be a string or an entity table", 2)
   end
 
-  if key.ws_id ~= nil and key.ws_id ~= null then
+  if key.ws_id ~= nil and key.ws_id ~= null and schema.workspaceable then
     ws_id = key.ws_id
   end
 

kong/db/declarative/import.lua

@@ -314,13 +314,12 @@ local function _set_entity_for_txn(t, entity_name, item, options, is_delete)
   -- select_by_cache_key
   if schema.cache_key then
     local cache_key = dao:cache_key(item)
-
-    for _, wid in ipairs {ws_id, GLOBAL_WORKSPACE_TAG} do
-      local key = unique_field_key(entity_name, wid, "cache_key", cache_key)
-
-      -- store item_key or nil into lmdb
-      t:set(key, idx_value)
-    end
+    -- The second parameter (ws_id) is a placeholder here, because the cache_key
+    -- is already unique globally.
+    local key = unique_field_key(entity_name, get_default_workspace(),
+                                 "cache_key", cache_key)
+    -- store item_key or nil into lmdb
+    t:set(key, idx_value)
   end
 
   for fname, fdata in schema:each_field() do

kong/db/strategies/off/init.lua

@@ -208,18 +208,25 @@ local function select_by_field(self, field, value, options)
     _, value = next(value)
   end
 
-  local ws_id = workspace_id(schema, options)
-
-  local key
-  local unique_across_ws = schema.fields[field].unique_across_ws
-  -- only accept global query by field if field is unique across workspaces
-  assert(not options or options.workspace ~= null or unique_across_ws)
+  local ws_id
 
-  if unique_across_ws then
+  if field == "cache_key" then
+    -- align with cache_key insertion logic in _set_entity_for_txn
     ws_id = get_default_workspace()
+
+  else
+    ws_id = workspace_id(schema, options)
+
+    local unique_across_ws = schema.fields[field].unique_across_ws
+    -- only accept global query by field if field is unique across workspaces
+    assert(not options or options.workspace ~= null or unique_across_ws)
+
+    if unique_across_ws then
+      ws_id = get_default_workspace()
+    end
   end
 
-  key = unique_field_key(schema.name, ws_id, field, value)
+  local key = unique_field_key(schema.name, ws_id, field, value)
 
   return select_by_key(schema, key, true)
 end