diff --git a/activerecord/lib/active_record/connection_adapters/abstract/database_statements.rb b/activerecord/lib/active_record/connection_adapters/abstract/database_statements.rb
index 769f488469d21..32386b6b14c24 100644
--- a/activerecord/lib/active_record/connection_adapters/abstract/database_statements.rb
+++ b/activerecord/lib/active_record/connection_adapters/abstract/database_statements.rb
@@ -51,9 +51,7 @@ def select_one(arel, name = nil, binds = [])
 
       # Returns a single value from a record
       def select_value(arel, name = nil, binds = [])
-        if result = select_rows(arel, name, binds).first
-          result.first
-        end
+        single_value_from_rows(select_rows(arel, name, binds))
       end
 
       # Returns an array of the values of the first column in a select:
@@ -68,6 +66,18 @@ def select_rows(arel, name = nil, binds = [])
         select_all(arel, name, binds).rows
       end
 
+      def query_value(sql, name = nil) # :nodoc:
+        single_value_from_rows(query(sql, name))
+      end
+
+      def query_values(sql, name = nil) # :nodoc:
+        query(sql, name).map(&:first)
+      end
+
+      def query(sql, name = nil) # :nodoc:
+        exec_query(sql, name).rows
+      end
+
       # Executes the SQL statement in the context of this connection and returns
       # the raw result from the connection adapter.
       # Note: depending on your database connector, the result returned by this
@@ -370,7 +380,11 @@ def sql_for_insert(sql, pk, id_value, sequence_name, binds)
         end
 
         def last_inserted_id(result)
-          row = result.rows.first
+          single_value_from_rows(result.rows)
+        end
+
+        def single_value_from_rows(rows)
+          row = rows.first
           row && row.first
         end
 
diff --git a/activerecord/lib/active_record/connection_adapters/abstract/schema_statements.rb b/activerecord/lib/active_record/connection_adapters/abstract/schema_statements.rb
index 32f9d3e5cdfd9..b2bad27d61f3b 100644
--- a/activerecord/lib/active_record/connection_adapters/abstract/schema_statements.rb
+++ b/activerecord/lib/active_record/connection_adapters/abstract/schema_statements.rb
@@ -31,7 +31,7 @@ def table_alias_for(table_name)
       # Returns the relation names useable to back Active Record models.
       # For most adapters this means all #tables and #views.
       def data_sources
-        select_values(data_source_sql, "SCHEMA")
+        query_values(data_source_sql, "SCHEMA")
       rescue NotImplementedError
         tables | views
       end
@@ -41,14 +41,14 @@ def data_sources
       #   data_source_exists?(:ebooks)
       #
       def data_source_exists?(name)
-        select_values(data_source_sql(name), "SCHEMA").any? if name.present?
+        query_values(data_source_sql(name), "SCHEMA").any? if name.present?
       rescue NotImplementedError
         data_sources.include?(name.to_s)
       end
 
       # Returns an array of table names defined in the database.
       def tables
-        select_values(data_source_sql(type: "BASE TABLE"), "SCHEMA")
+        query_values(data_source_sql(type: "BASE TABLE"), "SCHEMA")
       end
 
       # Checks to see if the table +table_name+ exists on the database.
@@ -56,14 +56,14 @@ def tables
       #   table_exists?(:developers)
       #
       def table_exists?(table_name)
-        select_values(data_source_sql(table_name, type: "BASE TABLE"), "SCHEMA").any? if table_name.present?
+        query_values(data_source_sql(table_name, type: "BASE TABLE"), "SCHEMA").any? if table_name.present?
       rescue NotImplementedError
         tables.include?(table_name.to_s)
       end
 
       # Returns an array of view names defined in the database.
       def views
-        select_values(data_source_sql(type: "VIEW"), "SCHEMA")
+        query_values(data_source_sql(type: "VIEW"), "SCHEMA")
       end
 
       # Checks to see if the view +view_name+ exists on the database.
@@ -71,7 +71,7 @@ def views
       #   view_exists?(:ebooks)
       #
       def view_exists?(view_name)
-        select_values(data_source_sql(view_name, type: "VIEW"), "SCHEMA").any? if view_name.present?
+        query_values(data_source_sql(view_name, type: "VIEW"), "SCHEMA").any? if view_name.present?
       rescue NotImplementedError
         views.include?(view_name.to_s)
       end
diff --git a/activerecord/lib/active_record/connection_adapters/abstract_mysql_adapter.rb b/activerecord/lib/active_record/connection_adapters/abstract_mysql_adapter.rb
index 0fc9a7fa056ee..b22bc073b31b4 100644
--- a/activerecord/lib/active_record/connection_adapters/abstract_mysql_adapter.rb
+++ b/activerecord/lib/active_record/connection_adapters/abstract_mysql_adapter.rb
@@ -140,11 +140,11 @@ def supports_advisory_locks?
       end
 
       def get_advisory_lock(lock_name, timeout = 0) # :nodoc:
-        select_value("SELECT GET_LOCK(#{quote(lock_name)}, #{timeout})") == 1
+        query_value("SELECT GET_LOCK(#{quote(lock_name)}, #{timeout})") == 1
       end
 
       def release_advisory_lock(lock_name) # :nodoc:
-        select_value("SELECT RELEASE_LOCK(#{quote(lock_name)})") == 1
+        query_value("SELECT RELEASE_LOCK(#{quote(lock_name)})") == 1
       end
 
       def native_database_types
@@ -176,7 +176,7 @@ def error_number(exception) # :nodoc:
       # REFERENTIAL INTEGRITY ====================================
 
       def disable_referential_integrity #:nodoc:
-        old = select_value("SELECT @@FOREIGN_KEY_CHECKS")
+        old = query_value("SELECT @@FOREIGN_KEY_CHECKS")
 
         begin
           update("SET FOREIGN_KEY_CHECKS = 0")
@@ -291,7 +291,7 @@ def drop_database(name) #:nodoc:
       end
 
       def current_database
-        select_value "SELECT DATABASE() as db"
+        query_value("SELECT database()", "SCHEMA")
       end
 
       # Returns the database character set.
@@ -351,7 +351,7 @@ def new_column_from_field(table_name, field) # :nodoc:
       def table_comment(table_name) # :nodoc:
         scope = quoted_scope(table_name)
 
-        select_value(<<-SQL.strip_heredoc, "SCHEMA")
+        query_value(<<-SQL.strip_heredoc, "SCHEMA")
           SELECT table_comment
           FROM information_schema.tables
           WHERE table_schema = #{scope[:schema]}
@@ -457,7 +457,7 @@ def foreign_keys(table_name)
 
         scope = quoted_scope(table_name)
 
-        fk_info = select_all(<<-SQL.strip_heredoc, "SCHEMA")
+        fk_info = exec_query(<<-SQL.strip_heredoc, "SCHEMA")
           SELECT fk.referenced_table_name AS 'to_table',
                  fk.referenced_column_name AS 'primary_key',
                  fk.column_name AS 'column',
@@ -534,7 +534,7 @@ def type_to_sql(type, limit: nil, precision: nil, scale: nil, unsigned: nil, **)
 
       # SHOW VARIABLES LIKE 'name'
       def show_variable(name)
-        select_value("SELECT @@#{name}", "SCHEMA")
+        query_value("SELECT @@#{name}", "SCHEMA")
       rescue ActiveRecord::StatementInvalid
         nil
       end
@@ -544,7 +544,7 @@ def primary_keys(table_name) # :nodoc:
 
         scope = quoted_scope(table_name)
 
-        select_values(<<-SQL.strip_heredoc, "SCHEMA")
+        query_values(<<-SQL.strip_heredoc, "SCHEMA")
           SELECT column_name
           FROM information_schema.key_column_usage
           WHERE constraint_name = 'PRIMARY'
@@ -745,7 +745,7 @@ def rename_column_sql(table_name, column_name, new_column_name)
             auto_increment: column.auto_increment?
           }
 
-          current_type = select_one("SHOW COLUMNS FROM #{quote_table_name(table_name)} LIKE '#{column_name}'", "SCHEMA")["Type"]
+          current_type = exec_query("SHOW COLUMNS FROM #{quote_table_name(table_name)} LIKE #{quote(column_name)}", "SCHEMA").first["Type"]
           td = create_table_definition(table_name)
           cd = td.new_column_definition(new_column_name, current_type, options)
           schema_creation.accept(ChangeColumnDefinition.new(cd, column.name))
@@ -864,7 +864,7 @@ def extract_foreign_key_action(specifier) # :nodoc:
         end
 
         def create_table_info(table_name) # :nodoc:
-          select_one("SHOW CREATE TABLE #{quote_table_name(table_name)}")["Create Table"]
+          exec_query("SHOW CREATE TABLE #{quote_table_name(table_name)}", "SCHEMA").first["Create Table"]
         end
 
         def create_table_definition(*args) # :nodoc:
diff --git a/activerecord/lib/active_record/connection_adapters/mysql/database_statements.rb b/activerecord/lib/active_record/connection_adapters/mysql/database_statements.rb
index 9f1021456bb52..bda482a00fca1 100644
--- a/activerecord/lib/active_record/connection_adapters/mysql/database_statements.rb
+++ b/activerecord/lib/active_record/connection_adapters/mysql/database_statements.rb
@@ -13,6 +13,10 @@ def select_all(arel, name = nil, binds = [], preparable: nil) # :nodoc:
           result
         end
 
+        def query(sql, name = nil) # :nodoc:
+          execute(sql, name).to_a
+        end
+
         # Executes the SQL statement in the context of this connection.
         def execute(sql, name = nil)
           # make sure we carry over any changes to ActiveRecord::Base.default_timezone that have been
diff --git a/activerecord/lib/active_record/connection_adapters/mysql/schema_dumper.rb b/activerecord/lib/active_record/connection_adapters/mysql/schema_dumper.rb
index e2ba0ba1a0f6f..eff96e329fffc 100644
--- a/activerecord/lib/active_record/connection_adapters/mysql/schema_dumper.rb
+++ b/activerecord/lib/active_record/connection_adapters/mysql/schema_dumper.rb
@@ -47,7 +47,7 @@ def schema_precision(column)
           def schema_collation(column)
             if column.collation && table_name = column.table_name
               @table_collation_cache ||= {}
-              @table_collation_cache[table_name] ||= select_one("SHOW TABLE STATUS LIKE '#{table_name}'")["Collation"]
+              @table_collation_cache[table_name] ||= exec_query("SHOW TABLE STATUS LIKE #{quote(table_name)}", "SCHEMA").first["Collation"]
               column.collation.inspect if column.collation != @table_collation_cache[table_name]
             end
           end
@@ -64,7 +64,7 @@ def extract_expression_for_virtual_column(column)
                     " WHERE table_schema = #{scope[:schema]}" \
                     "   AND table_name = #{scope[:name]}" \
                     "   AND column_name = #{quote(column.name)}"
-              select_value(sql, "SCHEMA").inspect
+              query_value(sql, "SCHEMA").inspect
             end
           end
       end
diff --git a/activerecord/lib/active_record/connection_adapters/postgresql/quoting.rb b/activerecord/lib/active_record/connection_adapters/postgresql/quoting.rb
index 4b18dceb467b9..ee4230c6f2f67 100644
--- a/activerecord/lib/active_record/connection_adapters/postgresql/quoting.rb
+++ b/activerecord/lib/active_record/connection_adapters/postgresql/quoting.rb
@@ -77,6 +77,9 @@ def lookup_cast_type_from_column(column) # :nodoc:
         end
 
         private
+          def lookup_cast_type(sql_type)
+            super(query_value("SELECT #{quote(sql_type)}::regtype::oid", "SCHEMA").to_i)
+          end
 
           def _quote(value)
             case value
diff --git a/activerecord/lib/active_record/connection_adapters/postgresql/schema_statements.rb b/activerecord/lib/active_record/connection_adapters/postgresql/schema_statements.rb
index 596c079388e10..6c1a4d9ded0fb 100644
--- a/activerecord/lib/active_record/connection_adapters/postgresql/schema_statements.rb
+++ b/activerecord/lib/active_record/connection_adapters/postgresql/schema_statements.rb
@@ -60,7 +60,7 @@ def drop_table(table_name, options = {}) # :nodoc:
 
         # Returns true if schema exists.
         def schema_exists?(name)
-          select_value("SELECT COUNT(*) FROM pg_namespace WHERE nspname = #{quote(name)}", "SCHEMA").to_i > 0
+          query_value("SELECT COUNT(*) FROM pg_namespace WHERE nspname = #{quote(name)}", "SCHEMA").to_i > 0
         end
 
         # Verifies existence of an index with a given name.
@@ -73,7 +73,7 @@ def index_name_exists?(table_name, index_name, default = nil)
           table = quoted_scope(table_name)
           index = quoted_scope(index_name)
 
-          select_value(<<-SQL, "SCHEMA").to_i > 0
+          query_value(<<-SQL, "SCHEMA").to_i > 0
             SELECT COUNT(*)
             FROM pg_class t
             INNER JOIN pg_index d ON t.oid = d.indrelid
@@ -173,7 +173,7 @@ def table_options(table_name) # :nodoc:
         def table_comment(table_name) # :nodoc:
           scope = quoted_scope(table_name, type: "BASE TABLE")
           if scope[:name]
-            select_value(<<-SQL.strip_heredoc, "SCHEMA")
+            query_value(<<-SQL.strip_heredoc, "SCHEMA")
               SELECT pg_catalog.obj_description(c.oid, 'pg_class')
               FROM pg_catalog.pg_class c
                 LEFT JOIN pg_namespace n ON n.oid = c.relnamespace
@@ -186,32 +186,32 @@ def table_comment(table_name) # :nodoc:
 
         # Returns the current database name.
         def current_database
-          select_value("SELECT current_database()", "SCHEMA")
+          query_value("SELECT current_database()", "SCHEMA")
         end
 
         # Returns the current schema name.
         def current_schema
-          select_value("SELECT current_schema", "SCHEMA")
+          query_value("SELECT current_schema", "SCHEMA")
         end
 
         # Returns the current database encoding format.
         def encoding
-          select_value("SELECT pg_encoding_to_char(encoding) FROM pg_database WHERE datname LIKE '#{current_database}'", "SCHEMA")
+          query_value("SELECT pg_encoding_to_char(encoding) FROM pg_database WHERE datname = current_database()", "SCHEMA")
         end
 
         # Returns the current database collation.
         def collation
-          select_value("SELECT datcollate FROM pg_database WHERE datname LIKE '#{current_database}'", "SCHEMA")
+          query_value("SELECT datcollate FROM pg_database WHERE datname = current_database()", "SCHEMA")
         end
 
         # Returns the current database ctype.
         def ctype
-          select_value("SELECT datctype FROM pg_database WHERE datname LIKE '#{current_database}'", "SCHEMA")
+          query_value("SELECT datctype FROM pg_database WHERE datname = current_database()", "SCHEMA")
         end
 
         # Returns an array of schema names.
         def schema_names
-          select_values(<<-SQL, "SCHEMA")
+          query_values(<<-SQL, "SCHEMA")
             SELECT nspname
               FROM pg_namespace
              WHERE nspname !~ '^pg_.*'
@@ -244,12 +244,12 @@ def schema_search_path=(schema_csv)
 
         # Returns the active schema search path.
         def schema_search_path
-          @schema_search_path ||= select_value("SHOW search_path", "SCHEMA")
+          @schema_search_path ||= query_value("SHOW search_path", "SCHEMA")
         end
 
         # Returns the current client message level.
         def client_min_messages
-          select_value("SHOW client_min_messages", "SCHEMA")
+          query_value("SHOW client_min_messages", "SCHEMA")
         end
 
         # Set the client message level.
@@ -267,7 +267,7 @@ def default_sequence_name(table_name, pk = "id") #:nodoc:
         end
 
         def serial_sequence(table, column)
-          select_value("SELECT pg_get_serial_sequence('#{table}', '#{column}')", "SCHEMA")
+          query_value("SELECT pg_get_serial_sequence(#{quote(table)}, #{quote(column)})", "SCHEMA")
         end
 
         # Sets the sequence of a table's primary key to the specified value.
@@ -278,7 +278,7 @@ def set_pk_sequence!(table, value) #:nodoc:
             if sequence
               quoted_sequence = quote_table_name(sequence)
 
-              select_value("SELECT setval('#{quoted_sequence}', #{value})", "SCHEMA")
+              query_value("SELECT setval(#{quote(quoted_sequence)}, #{value})", "SCHEMA")
             else
               @logger.warn "#{table} has primary key #{pk} with no default sequence." if @logger
             end
@@ -300,18 +300,16 @@ def reset_pk_sequence!(table, pk = nil, sequence = nil) #:nodoc:
 
           if pk && sequence
             quoted_sequence = quote_table_name(sequence)
-            max_pk = select_value("select MAX(#{quote_column_name pk}) from #{quote_table_name(table)}")
+            max_pk = query_value("SELECT MAX(#{quote_column_name pk}) FROM #{quote_table_name(table)}", "SCHEMA")
             if max_pk.nil?
               if postgresql_version >= 100000
-                minvalue = select_value("SELECT seqmin from pg_sequence where seqrelid = '#{quoted_sequence}'::regclass")
+                minvalue = query_value("SELECT seqmin FROM pg_sequence WHERE seqrelid = #{quote(quoted_sequence)}::regclass", "SCHEMA")
               else
-                minvalue = select_value("SELECT min_value FROM #{quoted_sequence}")
+                minvalue = query_value("SELECT min_value FROM #{quoted_sequence}", "SCHEMA")
               end
             end
 
-            select_value(<<-end_sql, "SCHEMA")
-              SELECT setval('#{quoted_sequence}', #{max_pk ? max_pk : minvalue}, #{max_pk ? true : false})
-            end_sql
+            query_value("SELECT setval(#{quote(quoted_sequence)}, #{max_pk ? max_pk : minvalue}, #{max_pk ? true : false})", "SCHEMA")
           end
         end
 
@@ -370,7 +368,7 @@ def pk_and_sequence_for(table) #:nodoc:
         end
 
         def primary_keys(table_name) # :nodoc:
-          select_values(<<-SQL.strip_heredoc, "SCHEMA")
+          query_values(<<-SQL.strip_heredoc, "SCHEMA")
             SELECT a.attname
               FROM (
                      SELECT indrelid, indkey, generate_subscripts(indkey, 1) idx
@@ -520,7 +518,7 @@ def rename_index(table_name, old_name, new_name)
 
         def foreign_keys(table_name)
           scope = quoted_scope(table_name)
-          fk_info = select_all(<<-SQL.strip_heredoc, "SCHEMA")
+          fk_info = exec_query(<<-SQL.strip_heredoc, "SCHEMA")
             SELECT t2.oid::regclass::text AS to_table, a1.attname AS column, a2.attname AS primary_key, c.conname AS name, c.confupdtype AS on_update, c.confdeltype AS on_delete
             FROM pg_constraint c
             JOIN pg_class t1 ON c.conrelid = t1.oid
diff --git a/activerecord/lib/active_record/connection_adapters/postgresql_adapter.rb b/activerecord/lib/active_record/connection_adapters/postgresql_adapter.rb
index 0d86f36ea3d0b..2040947d7de4f 100644
--- a/activerecord/lib/active_record/connection_adapters/postgresql_adapter.rb
+++ b/activerecord/lib/active_record/connection_adapters/postgresql_adapter.rb
@@ -326,14 +326,14 @@ def get_advisory_lock(lock_id) # :nodoc:
         unless lock_id.is_a?(Integer) && lock_id.bit_length <= 63
           raise(ArgumentError, "Postgres requires advisory lock ids to be a signed 64 bit integer")
         end
-        select_value("SELECT pg_try_advisory_lock(#{lock_id});")
+        query_value("SELECT pg_try_advisory_lock(#{lock_id})")
       end
 
       def release_advisory_lock(lock_id) # :nodoc:
         unless lock_id.is_a?(Integer) && lock_id.bit_length <= 63
           raise(ArgumentError, "Postgres requires advisory lock ids to be a signed 64 bit integer")
         end
-        select_value("SELECT pg_advisory_unlock(#{lock_id})")
+        query_value("SELECT pg_advisory_unlock(#{lock_id})")
       end
 
       def enable_extension(name)
@@ -350,15 +350,14 @@ def disable_extension(name)
 
       def extension_enabled?(name)
         if supports_extensions?
-          res = exec_query "SELECT EXISTS(SELECT * FROM pg_available_extensions WHERE name = '#{name}' AND installed_version IS NOT NULL) as enabled",
-            "SCHEMA"
+          res = exec_query("SELECT EXISTS(SELECT * FROM pg_available_extensions WHERE name = '#{name}' AND installed_version IS NOT NULL) as enabled", "SCHEMA")
           res.cast_values.first
         end
       end
 
       def extensions
         if supports_extensions?
-          exec_query("SELECT extname from pg_extension", "SCHEMA").cast_values
+          exec_query("SELECT extname FROM pg_extension", "SCHEMA").cast_values
         else
           super
         end
@@ -366,14 +365,14 @@ def extensions
 
       # Returns the configured supported identifier length supported by PostgreSQL
       def table_alias_length
-        @max_identifier_length ||= select_value("SHOW max_identifier_length", "SCHEMA").to_i
+        @max_identifier_length ||= query_value("SHOW max_identifier_length", "SCHEMA").to_i
       end
       alias index_name_length table_alias_length
 
       # Set the authorized user for this session
       def session_auth=(user)
         clear_cache!
-        exec_query "SET SESSION AUTHORIZATION #{user}"
+        execute("SET SESSION AUTHORIZATION #{user}")
       end
 
       def use_insert_returning?
@@ -384,11 +383,6 @@ def update_table_definition(table_name, base) #:nodoc:
         PostgreSQL::Table.new(table_name, base)
       end
 
-      def lookup_cast_type(sql_type) # :nodoc:
-        oid = execute("SELECT #{quote(sql_type)}::regtype::oid", "SCHEMA").first["oid"].to_i
-        super(oid)
-      end
-
       def column_name_for_operation(operation, node) # :nodoc:
         OPERATION_ALIASES.fetch(operation) { operation.downcase }
       end
diff --git a/activerecord/lib/active_record/connection_adapters/sqlite3_adapter.rb b/activerecord/lib/active_record/connection_adapters/sqlite3_adapter.rb
index d24bfc0c93672..f70b6b716b79b 100644
--- a/activerecord/lib/active_record/connection_adapters/sqlite3_adapter.rb
+++ b/activerecord/lib/active_record/connection_adapters/sqlite3_adapter.rb
@@ -183,7 +183,7 @@ def supports_explain?
       # REFERENTIAL INTEGRITY ====================================
 
       def disable_referential_integrity # :nodoc:
-        old = select_value("PRAGMA foreign_keys")
+        old = query_value("PRAGMA foreign_keys")
 
         begin
           execute("PRAGMA foreign_keys = OFF")
@@ -398,7 +398,7 @@ def add_reference(table_name, ref_name, **options) # :nodoc:
       alias :add_belongs_to :add_reference
 
       def foreign_keys(table_name)
-        fk_info = select_all("PRAGMA foreign_key_list(#{quote(table_name)})", "SCHEMA")
+        fk_info = exec_query("PRAGMA foreign_key_list(#{quote(table_name)})", "SCHEMA")
         fk_info.map do |row|
           options = {
             column: row["from"],
@@ -498,7 +498,7 @@ def copy_table_contents(from, to, columns, rename = {})
         end
 
         def sqlite_version
-          @sqlite_version ||= SQLite3Adapter::Version.new(select_value("SELECT sqlite_version(*)"))
+          @sqlite_version ||= SQLite3Adapter::Version.new(query_value("SELECT sqlite_version(*)"))
         end
 
         def translate_exception(exception, message)