Release 0.13.1 (#203)

* Add pool_concurrency_test manual spec

Add MT connection count without Mutex

* Drop checkout_some, simpler pool_prepared statement

* Make pool statement a struct

* Drop StringKeyCache mutex

The StringKeyCache is now only used inside a connection. It's assumed that connections are not used concurrently with multiple queries.

* Drop do_close in pool statements

* Add specs and update comment

* Fix typo

CHANGELOG.md

@@ -1,3 +1,25 @@
+## v0.13.1 (2023-12-21)
+
+* Gracefully allow spec helper to fail on older crystal. ([#202](https://github.com/crystal-lang/crystal-db/pull/202), thanks @bcardiff)
+
+## v0.13.0 (2023-12-11)
+
+* **(breaking-change)** Deprecate `DB.mapping`. ([#196](https://github.com/crystal-lang/crystal-db/pull/196), thanks @straight-shoota)
+* **(breaking-change)** Drop `Pool#checkout_some`, make `PoolStatement` a struct. ([#200](https://github.com/crystal-lang/crystal-db/pull/200), thanks @bcardiff)
+* Simplifications and performance improvements on pool statements. ([#200](https://github.com/crystal-lang/crystal-db/pull/200), thanks @bcardiff)
+* Allow `prepared_statements_cache=false`` option to disable prepared statements cache. ([#194](https://github.com/crystal-lang/crystal-db/pull/194), [#198](https://github.com/crystal-lang/crystal-db/pull/198), thanks @bcardiff)
+* Add exception cause support to `PoolResourceLost` and `ConnectionLost` constructors. ([#199](https://github.com/crystal-lang/crystal-db/pull/199), thanks @lachlan)
+* Fix inflight counter on `ConnectionRefused`. ([#184](https://github.com/crystal-lang/crystal-db/pull/184), thanks @jgaskins)
+* Fix max_idle_pool_size race condition. ([#186](https://github.com/crystal-lang/crystal-db/pull/186), thanks @bcardiff)
+* Fix `DB::DriverSpecs#with_db` `connection_string` query param support. ([#192](https://github.com/crystal-lang/crystal-db/pull/192), thanks @lachlan)
+* Update docs regarding `ConnectionBuilder`. ([#188](https://github.com/crystal-lang/crystal-db/pull/188), thanks @bcardiff)
+* Add reference to `DB::Serializable` in docs. ([#197](https://github.com/crystal-lang/crystal-db/pull/197), thanks @straight-shoota)
+* Add link to crystal-tds. ([#193](https://github.com/crystal-lang/crystal-db/pull/193), thanks @wonderix)
+
+### Notes for driver implementors
+
+* Use new constructors to preserve the underlying reason of a `PoolResourceLost` or `ConnectionLost` constructors (See [#199](https://github.com/crystal-lang/crystal-db/pull/199))
+
 ## v0.12.0 (2023-06-23)
 
 - **(breaking-change)** Refactor how drivers create connections. Allow creating `Database` without `URI`s. ([#181](https://github.com/crystal-lang/crystal-db/pull/181), thanks @bcardiff)

README.md

@@ -10,6 +10,7 @@ Common db api for crystal. You will need to have a specific driver to access a d
 * [ODBC](https://github.com/naqvis/crystal-odbc)
 * [Cassandra](https://github.com/kaukas/crystal-cassandra)
 * [DuckDB](https://github.com/amauryt/crystal-duckdb)
+* [Microsoft SQL Server](https://github.com/wonderix/crystal-tds)
 
 ## Installation
 

shard.yml

@@ -1,5 +1,5 @@
 name: db
-version: 0.12.0
+version: 0.13.1
 
 authors:
   - Brian J. Cardiff <bcardiff@gmail.com>

spec/database_spec.cr

@@ -57,14 +57,6 @@ describe DB::Database do
     end
   end
 
-  it "should close pool statements when closing db" do
-    stmt = uninitialized DB::PoolStatement
-    with_dummy do |db|
-      stmt = db.build("query1")
-    end
-    stmt.closed?.should be_true
-  end
-
   it "should not reconnect if connection is lost and retry_attempts=0" do
     DummyDriver::DummyConnection.clear_connections
     DB.open "dummy://localhost:1027?initial_pool_size=1&max_pool_size=1&retry_attempts=0" do |db|
@@ -187,6 +179,25 @@ describe DB::Database do
     end
   end
 
+  it "should not checkout multiple connections if there is a statement error" do
+    with_dummy "dummy://localhost:1027?initial_pool_size=1&max_pool_size=10&retry_attempts=10" do |db|
+      expect_raises DB::Error do
+        db.exec("syntax error")
+      end
+      DummyDriver::DummyConnection.connections.size.should eq(1)
+    end
+  end
+
+  it "should attempt all retries if connection is lost" do
+    with_dummy "dummy://localhost:1027?initial_pool_size=1&max_pool_size=1&retry_attempts=10" do |db|
+      expect_raises DB::PoolRetryAttemptsExceeded do
+        db.exec("raise ConnectionLost")
+      end
+      # 1 initial + 10 retries
+      DummyDriver::DummyConnection.connections.size.should eq(11)
+    end
+  end
+
   describe "prepared_statements connection option" do
     it "defaults to true" do
       with_dummy "dummy://localhost:1027" do |db|

spec/dummy_driver.cr

@@ -1,4 +1,3 @@
-require "spec"
 require "../src/db"
 
 class DummyDriver < DB::Driver
@@ -17,30 +16,54 @@ class DummyDriver < DB::Driver
   end
 
   class DummyConnection < DB::Connection
+    @@connections = [] of DummyConnection
+    @@connections_count = Atomic(Int32).new(0)
+
     def initialize(options : DB::Connection::Options)
       super(options)
+      Fiber.yield
+      @@connections_count.add(1)
       @connected = true
-      @@connections ||= [] of DummyConnection
-      @@connections.not_nil! << self
+      {% unless flag?(:preview_mt) %}
+        # @@connections is only used in single-threaded mode in specs
+        # for benchmarks we want to avoid the overhead of synchronizing this array
+        @@connections << self
+      {% end %}
+    end
+
+    def self.connections_count
+      @@connections_count.get
     end
 
     def self.connections
-      @@connections.not_nil!
+      {% if flag?(:preview_mt) %}
+        raise "DummyConnection.connections is only available in single-threaded mode"
+      {% end %}
+      @@connections
     end
 
     def self.clear_connections
-      @@connections.try &.clear
+      {% if flag?(:preview_mt) %}
+        raise "DummyConnection.clear_connections is only available in single-threaded mode"
+      {% end %}
+      @@connections.clear
     end
 
     def build_prepared_statement(query) : DB::Statement
+      assert_not_closed!
+
       DummyStatement.new(self, query, true)
     end
 
     def build_unprepared_statement(query) : DB::Statement
+      assert_not_closed!
+
       DummyStatement.new(self, query, false)
     end
 
     def last_insert_id : Int64
+      assert_not_closed!
+
       0
     end
 
@@ -53,12 +76,18 @@ class DummyDriver < DB::Driver
     end
 
     def create_transaction
+      assert_not_closed!
+
       DummyTransaction.new(self)
     end
 
     protected def do_close
       super
     end
+
+    private def assert_not_closed!
+      raise "Statement is closed" if closed?
+    end
   end
 
   class DummyTransaction < DB::TopLevelTransaction
@@ -104,21 +133,42 @@ class DummyDriver < DB::Driver
   end
 
   class DummyStatement < DB::Statement
+    @@statements_count = Atomic(Int32).new(0)
+    @@statements_exec_count = Atomic(Int32).new(0)
     property params
 
     def initialize(connection, command : String, @prepared : Bool)
       @params = Hash(Int32 | String, DB::Any | Array(DB::Any)).new
       super(connection, command)
+      @@statements_count.add(1)
       raise DB::Error.new(command) if command == "syntax error"
+      raise DB::ConnectionLost.new(connection) if command == "raise ConnectionLost"
+    end
+
+    def self.statements_count
+      @@statements_count.get
+    end
+
+    def self.statements_exec_count
+      @@statements_exec_count.get
     end
 
     protected def perform_query(args : Enumerable) : DB::ResultSet
+      assert_not_closed!
+
+      @@statements_exec_count.add(1)
+
+      Fiber.yield
       @connection.as(DummyConnection).check
       set_params args
       DummyResultSet.new self, command
     end
 
     protected def perform_exec(args : Enumerable) : DB::ExecResult
+      assert_not_closed!
+
+      @@statements_exec_count.add(1)
+
       @connection.as(DummyConnection).check
       set_params args
       raise DB::Error.new("forced exception due to query") if command == "raise"
@@ -151,6 +201,10 @@ class DummyDriver < DB::Driver
     protected def do_close
       super
     end
+
+    private def assert_not_closed!
+      raise "Statement is closed" if closed?
+    end
   end
 
   class DummyResultSet < DB::ResultSet
@@ -161,6 +215,8 @@ class DummyDriver < DB::Driver
 
     def initialize(statement, command)
       super(statement)
+      Fiber.yield
+
       @top_values = command.split.map { |r| r.split(',') }.to_a
       @column_count = @top_values.size > 0 ? @top_values[0].size : 2
 

spec/manual/load_test.cr

@@ -0,0 +1,54 @@
+# This file is to be executed as:
+#
+# % crystal ./spec/manual/load_test.cr
+#
+# It generates a number of producers and consumers. If the process hangs
+# it means that the connection pool is not working properly. Likely a race condition.
+#
+
+require "../dummy_driver"
+require "../../src/db"
+require "json"
+
+CONNECTION = "dummy://host?initial_pool_size=5&max_pool_size=5&max_idle_pool_size=5"
+
+alias TChannel = Channel(Int32)
+alias TDone = Channel(Bool)
+
+COUNT = 200
+
+def start_consumer(channel : TChannel, done : TDone)
+  spawn do
+    indeces = Set(Int32).new
+    loop do
+      indeces << channel.receive
+      puts "Received size=#{indeces.size}"
+      break if indeces.size == COUNT
+    end
+    done.send true
+  end
+end
+
+def start_producers(channel : TChannel)
+  db = DB.open CONNECTION do |db|
+    sql = "1,title,description,2023 " * 100_000
+
+    COUNT.times do |index|
+      spawn(name: "prod #{index}") do
+        puts "Sending #{index}"
+        _films = db.query_all(sql, as: {Int32, String, String, Int32})
+      rescue ex
+        puts "Error: #{ex.message}"
+      ensure
+        channel.send index
+      end
+    end
+  end
+end
+
+channel = TChannel.new
+done = TDone.new
+start_consumer(channel, done)
+start_producers(channel)
+
+done.receive

spec/manual/pool_concurrency_test.cr

@@ -0,0 +1,68 @@
+# This file is to be executed as:
+#
+# % crystal run --release [-Dpreview_mt] ./spec/manual/pool_concurrency_test.cr -- --options="max_pool_size=5" --duration=30 --concurrency=4
+#
+#
+
+require "option_parser"
+require "../dummy_driver"
+require "../../src/db"
+
+options = ""
+duration = 3
+concurrency = 4
+
+OptionParser.parse do |parser|
+  parser.banner = "Usage: pool_concurrency_test [arguments]"
+  parser.on("-o", "--options=VALUE", "Connection string options") { |v| options = v }
+  parser.on("-d", "--duration=SECONDS", "Specifies the duration in seconds") { |v| duration = v.to_i }
+  parser.on("-c", "--concurrency=VALUE", "Specifies the concurrent requests to perform") { |v| concurrency = v.to_i }
+  parser.on("-h", "--help", "Show this help") do
+    puts parser
+    exit
+  end
+  parser.invalid_option do |flag|
+    STDERR.puts "ERROR: #{flag} is not a valid option."
+    STDERR.puts parser
+    exit(1)
+  end
+end
+
+multi_threaded = {% if flag?(:preview_mt) %} ENV["CRYSTAL_WORKERS"]?.try(&.to_i?) || 4 {% else %} false {% end %}
+release = {% if flag?(:release) %} true {% else %} false {% end %}
+
+if !release
+  puts "WARNING: This should be run in release mode."
+end
+
+db = DB.open "dummy://host?#{options}"
+
+start_time = Time.monotonic
+
+puts "Starting test for #{duration} seconds..."
+
+concurrency.times do
+  spawn do
+    loop do
+      db.scalar "1"
+      Fiber.yield
+    end
+  end
+end
+
+sleep duration.seconds
+
+end_time = Time.monotonic
+
+puts "          Options : #{options}"
+puts "   Duration (sec) : #{duration} (actual #{end_time - start_time})"
+puts "      Concurrency : #{concurrency}"
+puts "   Multi Threaded : #{multi_threaded ? "Yes (#{multi_threaded})" : "No"}"
+puts "Total Connections : #{DummyDriver::DummyConnection.connections_count}"
+puts " Total Statements : #{DummyDriver::DummyStatement.statements_count}"
+puts "    Total Queries : #{DummyDriver::DummyStatement.statements_exec_count}"
+puts " Throughput (q/s) : #{DummyDriver::DummyStatement.statements_exec_count / duration}"
+
+if !release
+  puts "WARNING: This should be run in release mode."
+end

spec/statement_spec.cr

@@ -34,6 +34,60 @@ describe DB::Statement do
     end
   end
 
+  describe "prepared_statements_cache flag" do
+    it "should reuse prepared statements if true" do
+      with_dummy_connection("prepared_statements=true&prepared_statements_cache=true") do |cnn|
+        stmt1 = cnn.query("the query").statement
+        stmt2 = cnn.query("the query").statement
+        stmt1.object_id.should eq(stmt2.object_id)
+      end
+    end
+
+    it "should leave statements open to be reused if true" do
+      with_dummy_connection("prepared_statements=true&prepared_statements_cache=true") do |cnn|
+        rs = cnn.query("the query")
+        # do not close while iterating
+        rs.statement.closed?.should be_false
+        rs.close
+        # do not close to be reused
+        rs.statement.closed?.should be_false
+      end
+    end
+
+    it "should not reuse prepared statements if false" do
+      with_dummy_connection("prepared_statements=true&prepared_statements_cache=false") do |cnn|
+        stmt1 = cnn.query("the query").statement
+        stmt2 = cnn.query("the query").statement
+        stmt1.object_id.should_not eq(stmt2.object_id)
+      end
+    end
+
+    it "should close statements if false" do
+      with_dummy_connection("prepared_statements=true&prepared_statements_cache=false") do |cnn|
+        rs = cnn.query("the query")
+        # do not close while iterating
+        rs.statement.closed?.should be_false
+        rs.close
+        # do close after iterating
+        rs.statement.closed?.should be_true
+      end
+    end
+
+    it "should not close statements if false and created explicitly" do
+      with_dummy_connection("prepared_statements=true&prepared_statements_cache=false") do |cnn|
+        stmt = cnn.prepared("the query")
+
+        rs = stmt.query
+        # do not close while iterating
+        stmt.closed?.should be_false
+        rs.close
+
+        # do not close after iterating
+        stmt.closed?.should be_false
+      end
+    end
+  end
+
   it "should initialize positional params in query" do
     with_dummy_connection do |cnn|
       stmt = cnn.prepared("the query").as(DummyDriver::DummyStatement)

src/db.cr

@@ -75,6 +75,10 @@ require "log"
 # end
 # ```
 #
+# ### Object mapping
+#
+# The `DB::Serializable` module implements a declarative mapping from DB result
+# sets to Crystal types.
 module DB
   Log = ::Log.for(self)
 

src/db/connection.cr

@@ -25,10 +25,13 @@ module DB
 
     record Options,
       # Return whether the statements should be prepared by default
-      prepared_statements : Bool = true do
+      prepared_statements : Bool = true,
+      # Return whether the prepared statements should be cached or not
+      prepared_statements_cache : Bool = true do
       def self.from_http_params(params : HTTP::Params, default = Options.new)
         Options.new(
-          prepared_statements: DB.fetch_bool(params, "prepared_statements", default.prepared_statements)
+          prepared_statements: DB.fetch_bool(params, "prepared_statements", default.prepared_statements),
+          prepared_statements_cache: DB.fetch_bool(params, "prepared_statements_cache", default.prepared_statements)
         )
       end
     end
@@ -47,9 +50,17 @@ module DB
       @options.prepared_statements
     end
 
+    def prepared_statements_cache? : Bool
+      @options.prepared_statements_cache
+    end
+
     # :nodoc:
     def fetch_or_build_prepared_statement(query) : Statement
-      @statements_cache.fetch(query) { build_prepared_statement(query) }
+      if @options.prepared_statements_cache
+        @statements_cache.fetch(query) { build_prepared_statement(query) }
+      else
+        build_prepared_statement(query)
+      end
     end
 
     # :nodoc:

src/db/database.cr

@@ -38,7 +38,6 @@ module DB
     @connection_options : Connection::Options
     @pool : Pool(Connection)
     @setup_connection : Connection -> Nil
-    @statements_cache = StringKeyCache(PoolPreparedStatement).new
 
     # Initialize a database with the specified options and connection factory.
     # This covers more advanced use cases that might not be supported by an URI connection string such as tunneling connection.
@@ -59,6 +58,10 @@ module DB
       @connection_options.prepared_statements
     end
 
+    def prepared_statements_cache? : Bool
+      @connection_options.prepared_statements_cache
+    end
+
     # Run the specified block every time a new connection is established, yielding the new connection
     # to the block.
     #
@@ -77,9 +80,6 @@ module DB
 
     # Closes all connection to the database.
     def close
-      @statements_cache.each_value &.close
-      @statements_cache.clear
-
       @pool.close
     end
 
@@ -95,11 +95,6 @@ module DB
 
     # :nodoc:
     def fetch_or_build_prepared_statement(query) : PoolStatement
-      @statements_cache.fetch(query) { build_prepared_statement(query) }
-    end
-
-    # :nodoc:
-    def build_prepared_statement(query) : PoolStatement
       PoolPreparedStatement.new(self, query)
     end
 

src/db/driver.cr

@@ -7,10 +7,21 @@ module DB
   # require "db"
   #
   # class FakeDriver < DB::Driver
-  #   def connection_builder(uri : URI) : Proc(DB::Connection)
+  #   class FakeConnectionBuilder < DB::ConnectionBuilder
+  #     def initialize(@options : DB::Connection::Options)
+  #     end
+  #
+  #     def build : DB::Connection
+  #       FakeConnection.new(@options)
+  #     end
+  #   end
+  #
+  #   def connection_builder(uri : URI) : ConnectionBuilder
   #     params = HTTP::Params.parse(uri.query || "")
   #     options = connection_options(params)
-  #     ->{ FakeConnection.new(options).as(DB::Connection) }
+  #     # If needed, parse custom options from uri here
+  #     # so they are parsed only once.
+  #     FakeConnectionBuilder.new(options)
   #   end
   # end
   #

src/db/error.cr

@@ -30,7 +30,8 @@ module DB
   class PoolResourceLost(T) < Error
     getter resource : T
 
-    def initialize(@resource : T)
+    def initialize(@resource : T, cause : Exception? = nil)
+      super(cause: cause)
       @resource.close
     end
   end

src/db/mapping.cr

@@ -1,5 +1,6 @@
 module DB
   # Empty module used for marking a class as supporting DB:Mapping
+  @[Deprecated("Use `DB::Serializable` instead")]
   module Mappable; end
 
   # The `DB.mapping` macro defines how an object is built from a `ResultSet`.
@@ -57,6 +58,7 @@ module DB
   # it and initializes this type's instance variables.
   #
   # This macro also declares instance variables of the types given in the mapping.
+  @[Deprecated("Use `DB::Serializable` instead")]
   macro mapping(properties, strict = true)
     include ::DB::Mappable
 
@@ -148,6 +150,7 @@ module DB
     end
   end
 
+  @[Deprecated("Use `DB::Serializable` instead")]
   macro mapping(**properties)
     ::DB.mapping({{properties}})
   end

src/db/pool.cr

@@ -57,8 +57,6 @@ module DB
 
     # communicate that a connection is available for checkout
     @availability_channel : Channel(Nil)
-    # signal how many existing connections are waited for
-    @waiting_resource : Int32
     # global pool mutex
     @mutex : Mutex
 
@@ -82,7 +80,6 @@ module DB
       @retry_delay = pool_options.retry_delay
 
       @availability_channel = Channel(Nil).new
-      @waiting_resource = 0
       @inflight = 0
       @mutex = Mutex.new
 
@@ -120,8 +117,11 @@ module DB
           resource = if @idle.empty?
                        if can_increase_pool?
                          @inflight += 1
-                         r = unsync { build_resource }
-                         @inflight -= 1
+                         begin
+                           r = unsync { build_resource }
+                         ensure
+                           @inflight -= 1
+                         end
                          r
                        else
                          unsync { wait_for_available }
@@ -158,27 +158,6 @@ module DB
       end
     end
 
-    # ```
-    # selected, is_candidate = pool.checkout_some(candidates)
-    # ```
-    # `selected` be a resource from the `candidates` list and `is_candidate` == `true`
-    # or `selected` will be a new resource and `is_candidate` == `false`
-    def checkout_some(candidates : Enumerable(WeakRef(T))) : {T, Bool}
-      sync do
-        candidates.each do |ref|
-          resource = ref.value
-          if resource && is_available?(resource)
-            @idle.delete resource
-            resource.before_checkout
-            return {resource, true}
-          end
-        end
-      end
-
-      resource = checkout
-      {resource, candidates.any? { |ref| ref.value == resource }}
-    end
-
     def release(resource : T) : Nil
       idle_pushed = false
 
@@ -197,8 +176,11 @@ module DB
         end
       end
 
-      if idle_pushed && are_waiting_for_resource?
-        @availability_channel.send nil
+      if idle_pushed
+        select
+        when @availability_channel.send(nil)
+        else
+        end
       end
     end
 
@@ -224,8 +206,6 @@ module DB
           # if the connection is lost it will be closed by
           # the exception to release resources
           # we still need to remove it from the known pool.
-          # Closed connection will be evicted from statement cache
-          # in PoolPreparedStatement#clean_connections
           sync { delete(e.resource) }
         rescue e : PoolResourceRefused
           # a ConnectionRefused means a new connection
@@ -278,29 +258,13 @@ module DB
     end
 
     private def wait_for_available
-      sync_inc_waiting_resource
-
       select
       when @availability_channel.receive
-        sync_dec_waiting_resource
       when timeout(@checkout_timeout.seconds)
-        sync_dec_waiting_resource
         raise DB::PoolTimeout.new("Could not check out a connection in #{@checkout_timeout} seconds")
       end
     end
 
-    private def sync_inc_waiting_resource
-      sync { @waiting_resource += 1 }
-    end
-
-    private def sync_dec_waiting_resource
-      sync { @waiting_resource -= 1 }
-    end
-
-    private def are_waiting_for_resource?
-      @waiting_resource > 0
-    end
-
     private def sync
       @mutex.lock
       begin

src/db/pool_prepared_statement.cr

@@ -4,66 +4,20 @@ module DB
   # The execution of the statement is retried according to the pool configuration.
   #
   # See `PoolStatement`
-  class PoolPreparedStatement < PoolStatement
-    # connections where the statement was prepared
-    @connections = Set(WeakRef(Connection)).new
-    @mutex = Mutex.new
-
+  struct PoolPreparedStatement < PoolStatement
     def initialize(db : Database, query : String)
       super
-      # Prepares a statement on some connection
-      # otherwise the preparation is delayed until the first execution.
-      # After the first initialization the connection must be released
-      # it will be checked out when executing it.
-      statement_with_retry &.release_connection
-      # TODO use a round-robin selection in the pool so multiple sequentially
-      #      initialized statements are assigned to different connections.
-    end
-
-    protected def do_close
-      @mutex.synchronize do
-        # TODO close all statements on all connections.
-        # currently statements are closed when the connection is closed.
-
-        # WHAT-IF the connection is busy? Should each statement be able to
-        # deallocate itself when the connection is free.
-        @connections.clear
-      end
     end
 
     # builds a statement over a real connection
-    # the connection is registered in `@connections`
     private def build_statement : Statement
-      clean_connections
-
-      conn, existing = @mutex.synchronize do
-        @db.checkout_some(@connections)
-      end
-
+      conn = @db.pool.checkout
       begin
-        stmt = conn.prepared.build(@query)
+        conn.prepared.build(@query)
       rescue ex
         conn.release
         raise ex
       end
-      unless existing
-        @mutex.synchronize do
-          @connections << WeakRef.new(conn)
-        end
-      end
-      stmt
-    end
-
-    private def clean_connections
-      @mutex.synchronize do
-        # remove disposed or closed connections
-        @connections.each do |ref|
-          conn = ref.value
-          if !conn || conn.closed?
-            @connections.delete ref
-          end
-        end
-      end
     end
   end
 end

src/db/pool_statement.cr

@@ -3,7 +3,7 @@ module DB
   # a statement from the DB needs to be able to represent a statement in any
   # of the connections of the pool. Otherwise the user will need to deal with
   # actual connections in some point.
-  abstract class PoolStatement
+  abstract struct PoolStatement
     include StatementMethods
 
     def initialize(@db : Database, @query : String)

src/db/pool_unprepared_statement.cr

@@ -4,15 +4,11 @@ module DB
   # The execution of the statement is retried according to the pool configuration.
   #
   # See `PoolStatement`
-  class PoolUnpreparedStatement < PoolStatement
+  struct PoolUnpreparedStatement < PoolStatement
     def initialize(db : Database, query : String)
       super
     end
 
-    protected def do_close
-      # unprepared statements do not need to be release in each connection
-    end
-
     # builds a statement over a real connection
     private def build_statement : Statement
       conn = @db.pool.checkout

src/db/result_set.cr

@@ -29,7 +29,7 @@ module DB
     end
 
     protected def do_close
-      statement.release_connection
+      statement.release_from_result_set
     end
 
     # TODO add_next_result_set : Bool

src/db/session_methods.cr

@@ -14,13 +14,27 @@ module DB
     # be prepared or not.
     abstract def prepared_statements? : Bool
 
+    abstract def prepared_statements_cache? : Bool
+
     abstract def fetch_or_build_prepared_statement(query) : Stmt
 
     abstract def build_unprepared_statement(query) : Stmt
 
     def build(query) : Stmt
       if prepared_statements?
-        fetch_or_build_prepared_statement(query)
+        stmt = fetch_or_build_prepared_statement(query)
+
+        # #build is a :nodoc: method used on QueryMethods where
+        # the statements are not exposed. As such if the cache
+        # is disabled we should auto_close the statement.
+        # When the statements are build explicitly the #prepared
+        # and #unprepared methods are used. In that case the
+        # statement is closed by the user explicitly also.
+        if !prepared_statements_cache?
+          stmt.auto_close = true if stmt.responds_to?(:auto_close=)
+        end
+
+        stmt
       else
         build_unprepared_statement(query)
       end

src/db/statement.cr

@@ -2,11 +2,6 @@ module DB
   # Common interface for connection based statements
   # and for connection pool statements.
   module StatementMethods
-    include Disposable
-
-    protected def do_close
-    end
-
     # See `QueryMethods#scalar`
     def scalar(*args_, args : Array? = nil)
       query(*args_, args: args) do |rs|
@@ -47,6 +42,10 @@ module DB
   # 6. `#do_close` is called to release the statement resources.
   abstract class Statement
     include StatementMethods
+    include Disposable
+
+    protected def do_close
+    end
 
     # :nodoc:
     getter connection
@@ -56,6 +55,15 @@ module DB
     def initialize(@connection : Connection, @command : String)
     end
 
+    # :nodoc:
+    property auto_close : Bool = false
+
+    # :nodoc:
+    def release_from_result_set
+      self.close if @auto_close
+      self.release_connection
+    end
+
     def release_connection
       @connection.release_from_statement
     end

src/db/string_key_cache.cr

@@ -1,28 +1,21 @@
 module DB
   class StringKeyCache(T)
     @cache = {} of String => T
-    @mutex = Mutex.new
 
     def fetch(key : String) : T
-      @mutex.synchronize do
-        value = @cache.fetch(key, nil)
-        value = @cache[key] = yield unless value
-        value
-      end
+      value = @cache.fetch(key, nil)
+      value = @cache[key] = yield unless value
+      value
     end
 
     def each_value
-      @mutex.synchronize do
-        @cache.each do |_, value|
-          yield value
-        end
+      @cache.each do |_, value|
+        yield value
       end
     end
 
     def clear
-      @mutex.synchronize do
-        @cache.clear
-      end
+      @cache.clear
     end
   end
 end

src/db/version.cr

@@ -1,3 +1,3 @@
 module DB
-  VERSION = "0.12.0"
+  VERSION = "0.13.1"
 end

src/spec.cr

@@ -396,7 +396,20 @@ module DB
     # :nodoc:
     def with_db(options = nil)
       @before.call
-      DB.open("#{connection_string}#{"?#{options}" if options}") do |db|
+
+      if options
+        {% if compare_versions(Crystal::VERSION, "1.9.0") >= 0 %}
+          uri = URI.parse connection_string
+          uri.query_params.merge! URI::Params.parse(options)
+          connection_string_with_options = uri.to_s
+        {% else %}
+          raise "Crystal 1.9.0 or greater is required to run with_db with options"
+        {% end %}
+      else
+        connection_string_with_options = connection_string
+      end
+
+      DB.open(connection_string_with_options) do |db|
         db.exec(sql_drop_table("table1"))
         db.exec(sql_drop_table("table2"))
         db.exec(sql_drop_table("person"))
@@ -552,14 +565,16 @@ module DB
                 end
               end
             else
-              values.each do |prepared_statements|
-                it("#{db_it.description} (prepared_statements=#{prepared_statements})", db_it.file, db_it.line, db_it.end_line) do
-                  ctx.with_db "prepared_statements=#{prepared_statements}" do |db|
-                    db_it.block.call db
-                    nil
+              {% if compare_versions(Crystal::VERSION, "1.9.0") >= 0 %}
+                values.each do |prepared_statements|
+                  it("#{db_it.description} (prepared_statements=#{prepared_statements})", db_it.file, db_it.line, db_it.end_line) do
+                    ctx.with_db "prepared_statements=#{prepared_statements}" do |db|
+                      db_it.block.call db
+                      nil
+                    end
                   end
                 end
-              end
+              {% end %}
             end
           else
             raise "Invalid prepared value. Allowed values are :both and :default"