Allow Enumerable for query args instead of Array (#207)

* 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,36 @@
+## 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)
+- Close a transaction when `return`ing from within its block. ([#167](https://github.com/crystal-lang/crystal-db/pull/167), thanks @jgaskins)
+- Fix race conditions in multi-thread mode. ([#178](https://github.com/crystal-lang/crystal-db/pull/178), thanks @bcardiff)
+- Allow the use of `Enum`s when reading a `ResultSet`. ([#168](https://github.com/crystal-lang/crystal-db/pull/168), thanks @jgaskins)
+- Fix specs for Crystal 1.4 and 1.5. ([#163](https://github.com/crystal-lang/crystal-db/pull/163), [#173](https://github.com/crystal-lang/crystal-db/pull/173), thanks @straight-shoota)
+- Update README. ([#172](https://github.com/crystal-lang/crystal-db/pull/172), [#180](https://github.com/crystal-lang/crystal-db/pull/180), thanks @amauryt, @jgaskins)
+
+Note: The breaking-change introduced in this release does not affect consumers of the library, only driver implementors.
+
 ## v0.11.0 (2022-01-27)
 
 * Fix `Connection#transaction` method to return the block value as the result. ([#159](https://github.com/crystal-lang/crystal-db/pull/159), [#160](https://github.com/crystal-lang/crystal-db/pull/160), thanks @bcardiff)

README.md

@@ -9,6 +9,8 @@ Common db api for crystal. You will need to have a specific driver to access a d
 * [PostgreSQL](https://github.com/will/crystal-pg)
 * [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
 
@@ -34,7 +36,7 @@ Note: Multiple drivers can be included in the same application.
 
 ## Documentation
 
-* [Latest API](http://crystal-lang.github.io/crystal-db/api/latest/)
+* [Latest API](https://crystal-lang.github.io/crystal-db/api/latest/)
 * [Crystal book](https://crystal-lang.org/docs/database/)
 
 ## Usage

shard.yml

@@ -1,8 +1,8 @@
 name: db
-version: 0.11.0
+version: 0.13.1
 
 authors:
-  - Brian J. Cardiff <bcardiff@manas.tech>
+  - Brian J. Cardiff <bcardiff@gmail.com>
 
 crystal: ">= 1.0.0, < 2.0.0"
 

spec/custom_drivers_types_spec.cr

@@ -36,6 +36,15 @@ class FooValue
 end
 
 class FooDriver < DB::Driver
+  class FooConnectionBuilder < DB::ConnectionBuilder
+    def initialize(@options : DB::Connection::Options)
+    end
+
+    def build : DB::Connection
+      FooConnection.new(@options)
+    end
+  end
+
   alias Any = DB::Any | FooValue
   @@row = [] of Any
 
@@ -47,16 +56,17 @@ class FooDriver < DB::Driver
     @@row
   end
 
-  def build_connection(context : DB::ConnectionContext) : DB::Connection
-    FooConnection.new(context)
+  def connection_builder(uri : URI) : DB::ConnectionBuilder
+    params = HTTP::Params.parse(uri.query || "")
+    FooConnectionBuilder.new(connection_options(params))
   end
 
   class FooConnection < DB::Connection
-    def build_prepared_statement(command) : DB::Statement
-      FooStatement.new(self, command)
+    def build_prepared_statement(query) : DB::Statement
+      FooStatement.new(self, query)
     end
 
-    def build_unprepared_statement(command) : DB::Statement
+    def build_unprepared_statement(query) : DB::Statement
       raise "not implemented"
     end
   end
@@ -99,6 +109,15 @@ class BarValue
 end
 
 class BarDriver < DB::Driver
+  class BarConnectionBuilder < DB::ConnectionBuilder
+    def initialize(@options : DB::Connection::Options)
+    end
+
+    def build : DB::Connection
+      BarConnection.new(@options)
+    end
+  end
+
   alias Any = DB::Any | BarValue
   @@row = [] of Any
 
@@ -110,8 +129,9 @@ class BarDriver < DB::Driver
     @@row
   end
 
-  def build_connection(context : DB::ConnectionContext) : DB::Connection
-    BarConnection.new(context)
+  def connection_builder(uri : URI) : DB::ConnectionBuilder
+    params = HTTP::Params.parse(uri.query || "")
+    BarConnectionBuilder.new(connection_options(params))
   end
 
   class BarConnection < DB::Connection
@@ -156,8 +176,8 @@ DB.register_driver "bar", BarDriver
 
 describe DB do
   it "should be able to register multiple drivers" do
-    DB.open("foo://host").driver.should be_a(FooDriver)
-    DB.open("bar://host").driver.should be_a(BarDriver)
+    DB.open("foo://host").checkout.should be_a(FooDriver::FooConnection)
+    DB.open("bar://host").checkout.should be_a(BarDriver::BarConnection)
   end
 
   it "Foo and Bar drivers should return fake_row" do

spec/database_spec.cr

@@ -8,7 +8,7 @@ describe DB::Database do
 
       db.setup_connection do |cnn|
         cnn_setup += 1
-        cnn.scalar("1").should eq "1"
+        cnn.scalar("a").should eq "a"
       end
 
       cnn_setup.should eq(2)
@@ -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/db_spec.cr

@@ -9,12 +9,9 @@ describe DB do
     DB.driver_class("dummy").should eq(DummyDriver)
   end
 
-  it "should instantiate driver with connection uri" do
+  it "should create dummy connection" do
     db = DB.open "dummy://localhost:1027"
-    db.driver.should be_a(DummyDriver)
-    db.uri.scheme.should eq("dummy")
-    db.uri.host.should eq("localhost")
-    db.uri.port.should eq(1027)
+    db.checkout.should be_a(DummyDriver::DummyConnection)
   end
 
   it "should create a connection and close it" do

spec/dummy_driver.cr

@@ -1,36 +1,69 @@
-require "spec"
 require "../src/db"
 
 class DummyDriver < DB::Driver
-  def build_connection(context : DB::ConnectionContext) : DB::Connection
-    DummyConnection.new(context)
+  class DummyConnectionBuilder < DB::ConnectionBuilder
+    def initialize(@options : DB::Connection::Options)
+    end
+
+    def build : DB::Connection
+      DummyConnection.new(@options)
+    end
+  end
+
+  def connection_builder(uri : URI) : DB::ConnectionBuilder
+    params = HTTP::Params.parse(uri.query || "")
+    DummyConnectionBuilder.new(connection_options(params))
   end
 
   class DummyConnection < DB::Connection
-    def initialize(context)
-      super(context)
+    @@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
 
@@ -43,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
@@ -94,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"
@@ -141,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
@@ -151,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
 
@@ -187,7 +253,7 @@ class DummyDriver < DB::Driver
         return (@statement.as(DummyStatement)).params[0]
       end
 
-      return n
+      n.to_i64? || n
     end
 
     def next_column_index : Int32

spec/http_client_pool_spec.cr

@@ -22,10 +22,10 @@ describe DB::Pool do
       expected_per_connection = 5
       requests = fixed_pool_size * expected_per_connection
 
-      pool = DB::Pool.new(
+      pool = DB::Pool.new(DB::Pool::Options.new(
         initial_pool_size: fixed_pool_size,
         max_pool_size: fixed_pool_size,
-        max_idle_pool_size: fixed_pool_size) {
+        max_idle_pool_size: fixed_pool_size)) {
         HTTP::Client.new(URI.parse("http://127.0.0.1:#{address.port}/"))
       }
 

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/pool_spec.cr

@@ -57,15 +57,19 @@ class Closable
   end
 end
 
+private def create_pool(**options, &factory : -> T) forall T
+  DB::Pool.new(DB::Pool::Options.new(**options), &factory)
+end
+
 describe DB::Pool do
   it "should use proc to create objects" do
     block_called = 0
-    pool = DB::Pool.new(initial_pool_size: 3) { block_called += 1; Closable.new }
+    pool = create_pool(initial_pool_size: 3) { block_called += 1; Closable.new }
     block_called.should eq(3)
   end
 
   it "should get resource" do
-    pool = DB::Pool.new { Closable.new }
+    pool = create_pool { Closable.new }
     resource = pool.checkout
     resource.should be_a Closable
     resource.before_checkout_called.should be_true
@@ -73,18 +77,18 @@ describe DB::Pool do
 
   it "should be available if not checkedout" do
     resource = uninitialized Closable
-    pool = DB::Pool.new(initial_pool_size: 1) { resource = Closable.new }
+    pool = create_pool(initial_pool_size: 1) { resource = Closable.new }
     pool.is_available?(resource).should be_true
   end
 
   it "should not be available if checkedout" do
-    pool = DB::Pool.new { Closable.new }
+    pool = create_pool { Closable.new }
     resource = pool.checkout
     pool.is_available?(resource).should be_false
   end
 
   it "should be available if returned" do
-    pool = DB::Pool.new { Closable.new }
+    pool = create_pool { Closable.new }
     resource = pool.checkout
     resource.after_release_called.should be_false
     pool.release resource
@@ -93,7 +97,7 @@ describe DB::Pool do
   end
 
   it "should wait for available resource" do
-    pool = DB::Pool.new(max_pool_size: 1, initial_pool_size: 1) { Closable.new }
+    pool = create_pool(max_pool_size: 1, initial_pool_size: 1) { Closable.new }
 
     b_cnn_request = ShouldSleepingOp.new
     wait_a = WaitFor.new
@@ -121,7 +125,7 @@ describe DB::Pool do
 
   it "should create new if max was not reached" do
     block_called = 0
-    pool = DB::Pool.new(max_pool_size: 2, initial_pool_size: 1) { block_called += 1; Closable.new }
+    pool = create_pool(max_pool_size: 2, initial_pool_size: 1) { block_called += 1; Closable.new }
     block_called.should eq 1
     pool.checkout
     block_called.should eq 1
@@ -131,7 +135,7 @@ describe DB::Pool do
 
   it "should reuse returned resources" do
     all = [] of Closable
-    pool = DB::Pool.new(max_pool_size: 2, initial_pool_size: 1) { Closable.new.tap { |c| all << c } }
+    pool = create_pool(max_pool_size: 2, initial_pool_size: 1) { Closable.new.tap { |c| all << c } }
     pool.checkout
     b1 = pool.checkout
     pool.release b1
@@ -143,7 +147,7 @@ describe DB::Pool do
 
   it "should close available and total" do
     all = [] of Closable
-    pool = DB::Pool.new(max_pool_size: 2, initial_pool_size: 1) { Closable.new.tap { |c| all << c } }
+    pool = create_pool(max_pool_size: 2, initial_pool_size: 1) { Closable.new.tap { |c| all << c } }
     a = pool.checkout
     b = pool.checkout
     pool.release b
@@ -157,7 +161,7 @@ describe DB::Pool do
   end
 
   it "should timeout" do
-    pool = DB::Pool.new(max_pool_size: 1, checkout_timeout: 0.1) { Closable.new }
+    pool = create_pool(max_pool_size: 1, checkout_timeout: 0.1) { Closable.new }
     pool.checkout
     expect_raises DB::PoolTimeout do
       pool.checkout
@@ -165,7 +169,7 @@ describe DB::Pool do
   end
 
   it "should be able to release after a timeout" do
-    pool = DB::Pool.new(max_pool_size: 1, checkout_timeout: 0.1) { Closable.new }
+    pool = create_pool(max_pool_size: 1, checkout_timeout: 0.1) { Closable.new }
     a = pool.checkout
     pool.checkout rescue nil
     pool.release a
@@ -173,7 +177,7 @@ describe DB::Pool do
 
   it "should close if max idle amount is reached" do
     all = [] of Closable
-    pool = DB::Pool.new(max_pool_size: 3, max_idle_pool_size: 1) { Closable.new.tap { |c| all << c } }
+    pool = create_pool(max_pool_size: 3, max_idle_pool_size: 1) { Closable.new.tap { |c| all << c } }
     pool.checkout
     pool.checkout
     pool.checkout
@@ -191,7 +195,7 @@ describe DB::Pool do
   end
 
   it "should not return closed resources to the pool" do
-    pool = DB::Pool.new(max_pool_size: 1, max_idle_pool_size: 1) { Closable.new }
+    pool = create_pool(max_pool_size: 1, max_idle_pool_size: 1) { Closable.new }
 
     # pool size 1 should be reusing the one resource
     resource1 = pool.checkout
@@ -209,7 +213,7 @@ describe DB::Pool do
 
   it "should create resource after max_pool was reached if idle forced some close up" do
     all = [] of Closable
-    pool = DB::Pool.new(max_pool_size: 3, max_idle_pool_size: 1) { Closable.new.tap { |c| all << c } }
+    pool = create_pool(max_pool_size: 3, max_idle_pool_size: 1) { Closable.new.tap { |c| all << c } }
     pool.checkout
     pool.checkout
     pool.checkout

spec/serializable_spec.cr

@@ -81,6 +81,29 @@ class ModelWithJSON
   property c1 : String
 end
 
+struct ModelWithEnum
+  include DB::Serializable
+
+  getter c0 : Int32
+  getter c1 : MyEnum
+  # Ensure multiple enum types work together
+  getter c2 : MyOtherEnum
+
+  enum MyEnum
+    Foo  = 0
+    Bar  = 1
+    Baz  = 2
+    Quux = 3
+  end
+
+  enum MyOtherEnum
+    OMG
+    LOL
+    WTF
+    BBQ
+  end
+end
+
 macro from_dummy(query, type)
   with_dummy do |db|
     rs = db.query({{ query }})
@@ -105,7 +128,7 @@ describe "DB::Serializable" do
   end
 
   it "should fail to initialize a simple model if types do not match" do
-    expect_raises DB::MappingException, "Invalid Int32: b\n  deserializing SimpleModel#c0" do
+    expect_raises DB::MappingException, /Invalid Int32: "?b"?\n  deserializing SimpleModel#c0/ do
       from_dummy("b,a", SimpleModel)
     end
   end
@@ -172,6 +195,17 @@ describe "DB::Serializable" do
     expect_model("1,a", ModelWithJSON, {c0: 1, c1: "a"})
   end
 
+  it "should initialize a model with an enum property" do
+    expect_model("1,2,LOL", ModelWithEnum, {
+      c0: 1,
+      c1: ModelWithEnum::MyEnum::Baz,
+      c2: ModelWithEnum::MyOtherEnum::LOL,
+    })
+    expect_raises DB::MappingException, "Unknown enum ModelWithEnum::MyEnum value: adsf" do
+      from_dummy("1,adsf,BBQ", ModelWithEnum)
+    end
+  end
+
   it "should initialize multiple instances from a single resultset" do
     with_dummy do |db|
       db.query("1,a 2,b") do |rs|

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)

spec/transaction_spec.cr

@@ -95,6 +95,20 @@ describe DB::Transaction do
     t.committed.should be_false
   end
 
+  it "transaction with block from connection should be committed if `return` is called" do
+    t = uninitialized DummyDriver::DummyTransaction
+
+    with_witness do |w|
+      with_dummy_connection do |cnn|
+        t = return_from_txn(cnn).as(DummyDriver::DummyTransaction)
+        w.check
+      end
+    end
+
+    t.rolledback.should be_false
+    t.committed.should be_true
+  end
+
   it "transaction can be committed within block" do
     with_dummy_connection do |cnn|
       cnn.transaction do |tx|
@@ -211,3 +225,9 @@ describe DB::Transaction do
     end
   end
 end
+
+private def return_from_txn(cnn)
+  cnn.transaction do |tx|
+    return tx
+  end
+end

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)
 
@@ -152,7 +156,13 @@ module DB
   end
 
   private def self.build_database(uri : URI)
-    Database.new(build_driver(uri), uri)
+    driver = build_driver(uri)
+    params = HTTP::Params.parse(uri.query || "")
+    connection_options = driver.connection_options(params)
+    pool_options = driver.pool_options(params)
+    builder = driver.connection_builder(uri)
+    factory = ->{ builder.build }
+    Database.new(connection_options, pool_options, &factory)
   end
 
   private def self.build_connection(connection_string : String)
@@ -160,7 +170,7 @@ module DB
   end
 
   private def self.build_connection(uri : URI)
-    build_driver(uri).build_connection(SingleConnectionContext.new(uri)).as(Connection)
+    build_driver(uri).connection_builder(uri).build
   end
 
   private def self.build_driver(uri : URI)
@@ -188,6 +198,7 @@ require "./db/enumerable_concat"
 require "./db/query_methods"
 require "./db/session_methods"
 require "./db/disposable"
+require "./db/connection_builder"
 require "./db/driver"
 require "./db/statement"
 require "./db/begin_transaction"

src/db/begin_transaction.cr

@@ -13,25 +13,26 @@ module DB
     # can be called explicitly.
     # Returns the value of the block.
     def transaction(& : Transaction -> T) : T? forall T
+      rollback = false
       # TODO: Cast to workaround crystal-lang/crystal#9483
       # begin_transaction returns a Tx where Tx < Transaction
       tx = begin_transaction.as(Transaction)
       begin
         res = yield tx
       rescue DB::Rollback
-        tx.rollback unless tx.closed?
+        rollback = true
         res
       rescue e
-        unless tx.closed?
-          # Ignore error in rollback.
-          # It would only be a secondary error to the original one, caused by
-          # corrupted connection state.
-          tx.rollback rescue nil
-        end
+        rollback = true
         raise e
-      else
-        tx.commit unless tx.closed?
-        res
+      ensure
+        unless tx.closed?
+          if rollback
+            tx.rollback
+          else
+            tx.commit
+          end
+        end
       end
     end
   end

src/db/connection.cr

@@ -23,21 +23,44 @@ module DB
     include SessionMethods(Connection, Statement)
     include BeginTransaction
 
+    record Options,
+      # Return whether the statements should be prepared by default
+      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_cache: DB.fetch_bool(params, "prepared_statements_cache", default.prepared_statements)
+        )
+      end
+    end
+
     # :nodoc:
-    getter context
+    property context : ConnectionContext = SingleConnectionContext.default
     @statements_cache = StringKeyCache(Statement).new
     @transaction = false
-    getter? prepared_statements : Bool
     # :nodoc:
     property auto_release : Bool = true
 
-    def initialize(@context : ConnectionContext)
-      @prepared_statements = @context.prepared_statements?
+    def initialize(@options : Options)
+    end
+
+    def prepared_statements? : Bool
+      @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:
@@ -59,7 +82,7 @@ module DB
     protected def do_close
       @statements_cache.each_value &.close
       @statements_cache.clear
-      @context.discard self
+      context.discard self
     end
 
     # :nodoc:
@@ -75,7 +98,7 @@ module DB
     # managed by the database. Should be used
     # only if the connection was obtained by `Database#checkout`.
     def release
-      @context.release(self)
+      context.release(self)
     end
 
     # :nodoc:

src/db/connection_builder.cr

@@ -0,0 +1,8 @@
+module DB
+  # A connection factory with a specific configuration.
+  #
+  # See `Driver#connection_builder`.
+  abstract class ConnectionBuilder
+    abstract def build : Connection
+  end
+end

src/db/connection_context.cr

@@ -1,11 +1,5 @@
 module DB
   module ConnectionContext
-    # Returns the uri with the connection settings to the database
-    abstract def uri : URI
-
-    # Return whether the statements should be prepared by default
-    abstract def prepared_statements? : Bool
-
     # Indicates that the *connection* was permanently closed
     # and should not be used in the future.
     abstract def discard(connection : Connection)
@@ -19,13 +13,7 @@ module DB
   class SingleConnectionContext
     include ConnectionContext
 
-    getter uri : URI
-    getter? prepared_statements : Bool
-
-    def initialize(@uri : URI)
-      params = HTTP::Params.parse(uri.query || "")
-      @prepared_statements = DB.fetch_bool(params, "prepared_statements", true)
-    end
+    class_getter default : SingleConnectionContext = SingleConnectionContext.new
 
     def discard(connection : Connection)
     end

src/db/database.cr

@@ -10,8 +10,9 @@ module DB
   #
   # ## Database URI
   #
-  # Connection parameters are configured in a URI. The format is specified by the individual
-  # database drivers. See the [reference book](https://crystal-lang.org/reference/database/) for examples.
+  # Connection parameters are usually in a URI. The format is specified by the individual
+  # database drivers, yet there are some common properties names usually shared.
+  # See the [reference book](https://crystal-lang.org/reference/database/) for examples.
   #
   # The connection pool can be configured from URI parameters:
   #
@@ -31,36 +32,36 @@ module DB
     include SessionMethods(Database, PoolStatement)
     include ConnectionContext
 
-    # :nodoc:
-    getter driver
     # :nodoc:
     getter pool
 
-    # Returns the uri with the connection settings to the database
-    getter uri : URI
-
-    getter? prepared_statements : Bool
-
+    @connection_options : Connection::Options
     @pool : Pool(Connection)
     @setup_connection : Connection -> Nil
-    @statements_cache = StringKeyCache(PoolPreparedStatement).new
-
-    # :nodoc:
-    def initialize(@driver : Driver, @uri : URI)
-      params = HTTP::Params.parse(uri.query || "")
-      @prepared_statements = DB.fetch_bool(params, "prepared_statements", true)
-      pool_options = @driver.connection_pool_options(params)
 
+    # 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.
+    def initialize(connection_options : Connection::Options, pool_options : Pool::Options, &factory : -> Connection)
+      @connection_options = connection_options
       @setup_connection = ->(conn : Connection) {}
       @pool = uninitialized Pool(Connection) # in order to use self in the factory proc
-      @pool = Pool.new(**pool_options) {
-        conn = @driver.build_connection(self).as(Connection)
+      @pool = Pool(Connection).new(pool_options) {
+        conn = factory.call
         conn.auto_release = false
+        conn.context = self
         @setup_connection.call conn
         conn
       }
     end
 
+    def prepared_statements? : Bool
+      @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.
     #
@@ -79,9 +80,6 @@ module DB
 
     # Closes all connection to the database.
     def close
-      @statements_cache.each_value &.close
-      @statements_cache.clear
-
       @pool.close
     end
 
@@ -97,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

@@ -1,21 +1,34 @@
 module DB
   # Database driver implementors must subclass `Driver`,
   # register with a driver_name using `DB#register_driver` and
-  # override the factory method `#build_connection`.
+  # override the factory method `#connection_builder`.
   #
   # ```
   # require "db"
   #
   # class FakeDriver < DB::Driver
-  #   def build_connection(context : DB::ConnectionContext)
-  #     FakeConnection.new context
+  #   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)
+  #     # If needed, parse custom options from uri here
+  #     # so they are parsed only once.
+  #     FakeConnectionBuilder.new(options)
   #   end
   # end
   #
   # DB.register_driver "fake", FakeDriver
   # ```
   #
-  # Access to this fake datbase will be available with
+  # Access to this fake database will be available with
   #
   # ```
   # DB.open "fake://..." do |db|
@@ -25,18 +38,22 @@ module DB
   #
   # Refer to `Connection`, `Statement` and `ResultSet` for further
   # driver implementation instructions.
+  #
+  # Override `#connection_options` and `#pool_options` to provide custom
+  # defaults or parsing of the connection string URI.
   abstract class Driver
-    abstract def build_connection(context : ConnectionContext) : Connection
+    # Returns a new connection factory.
+    #
+    # NOTE: For implementors *uri* should be parsed once. If all the options
+    # are sound a ConnectionBuilder is returned.
+    abstract def connection_builder(uri : URI) : ConnectionBuilder
 
-    def connection_pool_options(params : HTTP::Params)
-      {
-        initial_pool_size:  params.fetch("initial_pool_size", 1).to_i,
-        max_pool_size:      params.fetch("max_pool_size", 0).to_i,
-        max_idle_pool_size: params.fetch("max_idle_pool_size", 1).to_i,
-        checkout_timeout:   params.fetch("checkout_timeout", 5.0).to_f,
-        retry_attempts:     params.fetch("retry_attempts", 1).to_i,
-        retry_delay:        params.fetch("retry_delay", 1.0).to_f,
-      }
+    def connection_options(params : HTTP::Params) : Connection::Options
+      Connection::Options.from_http_params(params)
+    end
+
+    def pool_options(params : HTTP::Params) : Pool::Options
+      Pool::Options.from_http_params(params)
     end
   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

@@ -4,6 +4,31 @@ require "./error"
 
 module DB
   class Pool(T)
+    record Options,
+      # initial number of connections in the pool
+      initial_pool_size : Int32 = 1,
+      # maximum amount of connections in the pool (Idle + InUse). 0 means no maximum.
+      max_pool_size : Int32 = 0,
+      # maximum amount of idle connections in the pool
+      max_idle_pool_size : Int32 = 1,
+      # seconds to wait before timeout while doing a checkout
+      checkout_timeout : Float64 = 5.0,
+      # maximum amount of retry attempts to reconnect to the db. See `Pool#retry`
+      retry_attempts : Int32 = 1,
+      # seconds to wait before a retry attempt
+      retry_delay : Float64 = 0.2 do
+      def self.from_http_params(params : HTTP::Params, default = Options.new)
+        Options.new(
+          initial_pool_size: params.fetch("initial_pool_size", default.initial_pool_size).to_i,
+          max_pool_size: params.fetch("max_pool_size", default.max_pool_size).to_i,
+          max_idle_pool_size: params.fetch("max_idle_pool_size", default.max_idle_pool_size).to_i,
+          checkout_timeout: params.fetch("checkout_timeout", default.checkout_timeout).to_f,
+          retry_attempts: params.fetch("retry_attempts", default.retry_attempts).to_i,
+          retry_delay: params.fetch("retry_delay", default.retry_delay).to_f,
+        )
+      end
+    end
+
     # Pool configuration
 
     # initial number of connections in the pool
@@ -32,15 +57,29 @@ 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
 
-    def initialize(@initial_pool_size = 1, @max_pool_size = 0, @max_idle_pool_size = 1, @checkout_timeout = 5.0,
-                   @retry_attempts = 1, @retry_delay = 0.2, &@factory : -> T)
+    @[Deprecated("Use `#new` with DB::Pool::Options instead")]
+    def initialize(initial_pool_size = 1, max_pool_size = 0, max_idle_pool_size = 1, checkout_timeout = 5.0,
+                   retry_attempts = 1, retry_delay = 0.2, &factory : -> T)
+      initialize(
+        Options.new(
+          initial_pool_size: initial_pool_size, max_pool_size: max_pool_size,
+          max_idle_pool_size: max_idle_pool_size, checkout_timeout: checkout_timeout,
+          retry_attempts: retry_attempts, retry_delay: retry_delay),
+        &factory)
+    end
+
+    def initialize(pool_options : Options = Options.new, &@factory : -> T)
+      @initial_pool_size = pool_options.initial_pool_size
+      @max_pool_size = pool_options.max_pool_size
+      @max_idle_pool_size = pool_options.max_idle_pool_size
+      @checkout_timeout = pool_options.checkout_timeout
+      @retry_attempts = pool_options.retry_attempts
+      @retry_delay = pool_options.retry_delay
+
       @availability_channel = Channel(Nil).new
-      @waiting_resource = 0
       @inflight = 0
       @mutex = Mutex.new
 
@@ -78,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 }
@@ -116,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
 
@@ -155,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
 
@@ -182,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
@@ -216,8 +238,10 @@ module DB
 
     private def build_resource : T
       resource = @factory.call
-      @total << resource
-      @idle << resource
+      sync do
+        @total << resource
+        @idle << resource
+      end
       resource
     end
 
@@ -234,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,53 +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
-
+  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
-      # 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
 
     # builds a statement over a real connection
-    # the connection is registered in `@connections`
     private def build_statement : Statement
-      clean_connections
-      conn, existing = @db.checkout_some(@connections)
+      conn = @db.pool.checkout
       begin
-        stmt = conn.prepared.build(@query)
+        conn.prepared.build(@query)
       rescue ex
         conn.release
         raise ex
       end
-      @connections << WeakRef.new(conn) unless existing
-      stmt
-    end
-
-    private def clean_connections
-      # remove disposed or closed connections
-      @connections.each do |ref|
-        conn = ref.value
-        if !conn || conn.closed?
-          @connections.delete ref
-        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)
@@ -15,7 +15,7 @@ module DB
     end
 
     # See `QueryMethods#exec`
-    def exec(*args_, args : Array? = nil) : ExecResult
+    def exec(*args_, args : Enumerable? = nil) : ExecResult
       statement_with_retry &.exec(*args_, args: args)
     end
 
@@ -25,12 +25,12 @@ module DB
     end
 
     # See `QueryMethods#query`
-    def query(*args_, args : Array? = nil) : ResultSet
+    def query(*args_, args : Enumerable? = nil) : ResultSet
       statement_with_retry &.query(*args_, args: args)
     end
 
     # See `QueryMethods#scalar`
-    def scalar(*args_, args : Array? = nil)
+    def scalar(*args_, args : Enumerable? = nil)
       statement_with_retry &.scalar(*args_, args: args)
     end
 

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/query_methods.cr

@@ -42,7 +42,7 @@ module DB
     # result = db.query "select name from contacts where id = ?", args: [10]
     # ```
     #
-    def query(query, *args_, args : Array? = nil)
+    def query(query, *args_, args : Enumerable? = nil)
       build(query).query(*args_, args: args)
     end
 
@@ -56,7 +56,7 @@ module DB
     #   end
     # end
     # ```
-    def query(query, *args_, args : Array? = nil)
+    def query(query, *args_, args : Enumerable? = nil)
       # CHECK build(query).query(*args, &block)
       rs = query(query, *args_, args: args)
       yield rs ensure rs.close
@@ -73,7 +73,7 @@ module DB
     # ```
     # name = db.query_one "select name from contacts where id = ?", 18, &.read(String)
     # ```
-    def query_one(query, *args_, args : Array? = nil, &block : ResultSet -> U) : U forall U
+    def query_one(query, *args_, args : Enumerable? = nil, &block : ResultSet -> U) : U forall U
       query(query, *args_, args: args) do |rs|
         raise DB::NoResultsError.new("no results") unless rs.move_next
 
@@ -92,7 +92,7 @@ module DB
     # ```
     # db.query_one "select name, age from contacts where id = ?", 1, as: {String, Int32}
     # ```
-    def query_one(query, *args_, args : Array? = nil, as types : Tuple)
+    def query_one(query, *args_, args : Enumerable? = nil, as types : Tuple)
       query_one(query, *args_, args: args) do |rs|
         rs.read(*types)
       end
@@ -108,7 +108,7 @@ module DB
     # ```
     # db.query_one "select name, age from contacts where id = ?", 1, as: {name: String, age: Int32}
     # ```
-    def query_one(query, *args_, args : Array? = nil, as types : NamedTuple)
+    def query_one(query, *args_, args : Enumerable? = nil, as types : NamedTuple)
       query_one(query, *args_, args: args) do |rs|
         rs.read(**types)
       end
@@ -123,7 +123,7 @@ module DB
     # ```
     # db.query_one "select name from contacts where id = ?", 1, as: String
     # ```
-    def query_one(query, *args_, args : Array? = nil, as type : Class)
+    def query_one(query, *args_, args : Enumerable? = nil, as type : Class)
       query_one(query, *args_, args: args) do |rs|
         rs.read(type)
       end
@@ -141,7 +141,7 @@ module DB
     # name = db.query_one? "select name from contacts where id = ?", 18, &.read(String)
     # typeof(name) # => String | Nil
     # ```
-    def query_one?(query, *args_, args : Array? = nil, &block : ResultSet -> U) : U? forall U
+    def query_one?(query, *args_, args : Enumerable? = nil, &block : ResultSet -> U) : U? forall U
       query(query, *args_, args: args) do |rs|
         return nil unless rs.move_next
 
@@ -162,7 +162,7 @@ module DB
     # result = db.query_one? "select name, age from contacts where id = ?", 1, as: {String, Int32}
     # typeof(result) # => Tuple(String, Int32) | Nil
     # ```
-    def query_one?(query, *args_, args : Array? = nil, as types : Tuple)
+    def query_one?(query, *args_, args : Enumerable? = nil, as types : Tuple)
       query_one?(query, *args_, args: args) do |rs|
         rs.read(*types)
       end
@@ -180,7 +180,7 @@ module DB
     # result = db.query_one? "select name, age from contacts where id = ?", 1, as: {age: String, name: Int32}
     # typeof(result) # => NamedTuple(age: String, name: Int32) | Nil
     # ```
-    def query_one?(query, *args_, args : Array? = nil, as types : NamedTuple)
+    def query_one?(query, *args_, args : Enumerable? = nil, as types : NamedTuple)
       query_one?(query, *args_, args: args) do |rs|
         rs.read(**types)
       end
@@ -197,7 +197,7 @@ module DB
     # name = db.query_one? "select name from contacts where id = ?", 1, as: String
     # typeof(name) # => String?
     # ```
-    def query_one?(query, *args_, args : Array? = nil, as type : Class)
+    def query_one?(query, *args_, args : Enumerable? = nil, as type : Class)
       query_one?(query, *args_, args: args) do |rs|
         rs.read(type)
       end
@@ -209,7 +209,7 @@ module DB
     # ```
     # names = db.query_all "select name from contacts", &.read(String)
     # ```
-    def query_all(query, *args_, args : Array? = nil, &block : ResultSet -> U) : Array(U) forall U
+    def query_all(query, *args_, args : Enumerable? = nil, &block : ResultSet -> U) : Array(U) forall U
       ary = [] of U
       query_each(query, *args_, args: args) do |rs|
         ary.push(yield rs)
@@ -223,7 +223,7 @@ module DB
     # ```
     # contacts = db.query_all "select name, age from contacts", as: {String, Int32}
     # ```
-    def query_all(query, *args_, args : Array? = nil, as types : Tuple)
+    def query_all(query, *args_, args : Enumerable? = nil, as types : Tuple)
       query_all(query, *args_, args: args) do |rs|
         rs.read(*types)
       end
@@ -236,7 +236,7 @@ module DB
     # ```
     # contacts = db.query_all "select name, age from contacts", as: {name: String, age: Int32}
     # ```
-    def query_all(query, *args_, args : Array? = nil, as types : NamedTuple)
+    def query_all(query, *args_, args : Enumerable? = nil, as types : NamedTuple)
       query_all(query, *args_, args: args) do |rs|
         rs.read(**types)
       end
@@ -248,7 +248,7 @@ module DB
     # ```
     # names = db.query_all "select name from contacts", as: String
     # ```
-    def query_all(query, *args_, args : Array? = nil, as type : Class)
+    def query_all(query, *args_, args : Enumerable? = nil, as type : Class)
       query_all(query, *args_, args: args) do |rs|
         rs.read(type)
       end
@@ -262,7 +262,7 @@ module DB
     #   puts rs.read(String)
     # end
     # ```
-    def query_each(query, *args_, args : Array? = nil)
+    def query_each(query, *args_, args : Enumerable? = nil)
       query(query, *args_, args: args) do |rs|
         rs.each do
           yield rs
@@ -271,7 +271,7 @@ module DB
     end
 
     # Performs the `query` and returns an `ExecResult`
-    def exec(query, *args_, args : Array? = nil)
+    def exec(query, *args_, args : Enumerable? = nil)
       build(query).exec(*args_, args: args)
     end
 
@@ -280,7 +280,7 @@ module DB
     # ```
     # puts db.scalar("SELECT MAX(name)").as(String) # => (a String)
     # ```
-    def scalar(query, *args_, args : Array? = nil)
+    def scalar(query, *args_, args : Enumerable? = nil)
       build(query).scalar(*args_, args: args)
     end
   end

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
@@ -96,6 +96,23 @@ module DB
       end
     end
 
+    # Read the value based on the given `enum` type, supporting both string and
+    # numeric column types.
+    #
+    # ```
+    # enum Status
+    #   Pending
+    #   Complete
+    # end
+    #
+    # db.query "SELECT 'complete'" do |rs|
+    #   rs.read Status # => Status::Complete
+    # end
+    # ```
+    def read(type : Enum.class)
+      type.new(self)
+    end
+
     # Reads the next columns and returns a tuple of the values.
     def read(*types : Class)
       internal_read(*types)
@@ -135,3 +152,24 @@ module DB
     # end
   end
 end
+
+struct Enum
+  def self.new(rs : DB::ResultSet) : self
+    index = rs.next_column_index
+
+    case value = rs.read
+    when String
+      parse value
+    when Int
+      from_value value
+    else
+      raise DB::ColumnTypeMismatchError.new(
+        context: "#{self}.new(rs : DB::ResultSet)",
+        column_index: index,
+        column_name: rs.column_name(index),
+        column_type: value.class.to_s,
+        expected_type: "String | Int",
+      )
+    end
+  end
+end

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,13 +2,8 @@ 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)
+    def scalar(*args_, args : Enumerable? = nil)
       query(*args_, args: args) do |rs|
         rs.each do
           return rs.read
@@ -19,7 +14,7 @@ module DB
     end
 
     # See `QueryMethods#query`
-    def query(*args_, args : Array? = nil)
+    def query(*args_, args : Enumerable? = nil)
       rs = query(*args_, args: args)
       yield rs ensure rs.close
     end
@@ -27,12 +22,12 @@ module DB
     # See `QueryMethods#exec`
     abstract def exec : ExecResult
     # See `QueryMethods#exec`
-    abstract def exec(*args_, args : Array? = nil) : ExecResult
+    abstract def exec(*args_, args : Enumerable? = nil) : ExecResult
 
     # See `QueryMethods#query`
     abstract def query : ResultSet
     # See `QueryMethods#query`
-    abstract def query(*args_, args : Array? = nil) : ResultSet
+    abstract def query(*args_, args : Enumerable? = nil) : ResultSet
   end
 
   # Represents a query in a `Connection`.
@@ -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
@@ -66,7 +74,7 @@ module DB
     end
 
     # See `QueryMethods#exec`
-    def exec(*args_, args : Array? = nil) : DB::ExecResult
+    def exec(*args_, args : Enumerable? = nil) : DB::ExecResult
       perform_exec_and_release(EnumerableConcat.build(args_, args))
     end
 
@@ -76,7 +84,7 @@ module DB
     end
 
     # See `QueryMethods#query`
-    def query(*args_, args : Array? = nil) : DB::ResultSet
+    def query(*args_, args : Enumerable? = nil) : DB::ResultSet
       perform_query_with_rescue(EnumerableConcat.build(args_, args))
     end
 

src/db/version.cr

@@ -1,3 +1,3 @@
 module DB
-  VERSION = "0.11.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"))
@@ -523,7 +536,7 @@ module DB
 
     def self.run(description = "as a db")
       ctx = self.new
-      with ctx yield
+      with ctx yield ctx
 
       describe description do
         ctx.include_shared_specs
@@ -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"