feat: unprepare statement references after sql execution

Summary

Picodata SQL now manages the Tarantool statement cache using a dedicated SQL fiber that handles preparation and unpreparation of statements based on an LRU eviction policy. Prepared statements can be executed across different sessions by SQL clients.

Previously, when a client executed a prepared statement, it increased the reference count in the statement cache and linked the statement to the client's session. While this approach seemed fine, it caused issues during eviction, as references to these statements remained in client sessions, preventing proper eviction.

This commit addresses the issue by ensuring that if a statement is added to the current session during execution, it is removed and the session state is restored once execution is complete.

Discussion:

https://docs.google.com/document/d/181CwxhDwrUkyMbZu7MOUG0BJw2M1rJeuiQP2-M2TpFk/edit#heading=h.mekvqbvwrhq8

test/sql-luatest/sql_api_test.lua

@@ -119,6 +119,7 @@ g.test_prepared_stmt_execution = function()
         res = box.execute([[SELECT * FROM t WHERE a = 'B']])
         t.assert_equals(res.rows[1][1], 'B')
 
+        s:unprepare()
         ffi.C.obuf_destroy(obuf)
     end)
 end
@@ -153,6 +154,8 @@ g.test_cross_session_stmt_execution = function()
         -- Check the result.
         local res = box.execute([[SELECT * FROM t WHERE a = 'C']])
         t.assert_equals(res.rows[1][1], 'C')
+
+        s:unprepare()
     end)
 end
 
@@ -266,12 +269,12 @@ g.test_execution_references = function()
         local ffi = require('ffi')
         local stmt_id = ffi.new('uint32_t[1]')
         local session_id = ffi.new('uint64_t[1]')
+        local initial_stmt_count = box.info.sql().cache.stmt_count
 
         -- Prepare the statement.
-        t.assert_equals(box.info.sql().cache.stmt_count, 0)
         res = ffi.C.sql_prepare_ext('VALUES (?)', 10, stmt_id, session_id)
         t.assert_equals(res, 0)
-        t.assert_equals(box.info.sql().cache.stmt_count, 1)
+        t.assert_equals(box.info.sql().cache.stmt_count, initial_stmt_count + 1)
         id = tonumber(stmt_id[0])
         sid = tonumber(session_id[0])
 
@@ -280,27 +283,29 @@ g.test_execution_references = function()
         buf = ffi.cast('char *', '\x91\xd9\x01A')
         res = ffi.C.sql_execute_prepared_ext(id, buf, 1024, obuf)
         t.assert_equals(res, 0)
-        t.assert_equals(box.info.sql().cache.stmt_count(), 1)
+        t.assert_equals(
+            box.info.sql().cache.stmt_count(),
+            initial_stmt_count + 1
+        )
 
         -- Unprepare the statement
         res = ffi.C.sql_unprepare_ext(id, sid)
         t.assert_equals(res, 0)
-        t.assert_equals(box.info.sql().cache.stmt_count(), 0)
+        t.assert_equals(box.info.sql().cache.stmt_count(), initial_stmt_count)
     end)
 end
 
 g.test_execution_references_from_other_session = function()
     g.server:exec(function()
-        local fiber = require('fiber')
-
         local res, ok
-
+        local fiber = require('fiber')
         local ffi = require('ffi')
         local stmt_id = ffi.new('uint32_t[1]')
         local session_id = ffi.new('uint64_t[1]')
+        local initial_stmt_count = box.info.sql().cache.stmt_count
 
         -- Prepare the statement.
-        t.assert_equals(box.info.sql().cache.stmt_count, 0)
+        t.assert_equals(box.info.sql().cache.stmt_count, initial_stmt_count)
         res = ffi.C.sql_prepare_ext('VALUES (?)', 10, stmt_id, session_id)
         t.assert_equals(res, 0)
         t.assert_equals(box.info.sql().cache.stmt_count, 1)
@@ -308,8 +313,11 @@ g.test_execution_references_from_other_session = function()
         local new_session = function(stmt_id, session_id)
             local buf
             local ffi = require('ffi')
+            local slab_cache = ffi.C.cord_slab_cache()
+            local mem_chunk = ffi.new("struct obuf_impl[1]")
+            local obuf = ffi.cast('struct obuf *', mem_chunk)
+            ffi.C.obuf_create(obuf, slab_cache, 1024)
             local id = tonumber(stmt_id[0])
-            local sid = tonumber(session_id[0])
 
             -- Check that execution of the statement from the session,
             -- different from the one where it was prepared, does not
@@ -317,7 +325,10 @@ g.test_execution_references_from_other_session = function()
             buf = ffi.cast('char *', '\x91\xd9\x01A')
             res = ffi.C.sql_execute_prepared_ext(id, buf, 1024, obuf)
             t.assert_equals(res, 0)
-            t.assert_equals(box.info.sql().cache.stmt_count(), 1)
+            t.assert_equals(
+                box.info.sql().cache.stmt_count(),
+                initial_stmt_count + 1
+            )
         end
 
         local f = fiber.new(new_session, stmt_id, session_id)
@@ -329,8 +340,9 @@ g.test_execution_references_from_other_session = function()
         end
 
         -- Unprepare the statement
+        local sid = tonumber(session_id[0])
         res = ffi.C.sql_unprepare_ext(id, sid)
         t.assert_equals(res, 0)
-        t.assert_equals(box.info.sql().cache.stmt_count(), 0)
+        t.assert_equals(box.info.sql().cache.stmt_count(), initial_stmt_count)
     end)
 end