diff --git a/exposed-core/src/main/kotlin/org/jetbrains/exposed/sql/Expression.kt b/exposed-core/src/main/kotlin/org/jetbrains/exposed/sql/Expression.kt
index 51d344538c..7ef98dc2ad 100644
--- a/exposed-core/src/main/kotlin/org/jetbrains/exposed/sql/Expression.kt
+++ b/exposed-core/src/main/kotlin/org/jetbrains/exposed/sql/Expression.kt
@@ -1,6 +1,7 @@
 package org.jetbrains.exposed.sql
 
 import org.jetbrains.exposed.sql.statements.DefaultValueMarker
+import org.jetbrains.exposed.sql.statements.api.ExposedBlob
 import org.jetbrains.exposed.sql.transactions.TransactionManager
 
 /**
@@ -12,6 +13,7 @@ class QueryBuilder(
 ) {
     private val internalBuilder = StringBuilder()
     private val _args = mutableListOf<Pair<IColumnType, Any?>>()
+
     /** Returns the list of arguments used in this query. */
     val args: List<Pair<IColumnType, Any?>> get() = _args
 
@@ -79,19 +81,20 @@ class QueryBuilder(
 
     /** Adds the specified sequence of [arguments] as values of the specified [sqlType]. */
     fun <T> registerArguments(sqlType: IColumnType, arguments: Iterable<T>) {
-        fun toString(value: T) = when {
-            prepared && value is String -> value
-            else -> sqlType.valueToString(value)
-        }
-
-        arguments.map { it to toString(it) }
-            .sortedBy { it.second }
+        arguments
             .appendTo {
                 if (prepared) {
-                    _args.add(sqlType to it.first)
+                    if (sqlType is BlobColumnType && it is ExposedBlob) {
+                        // if it.inputStream isn't used, the lazy initialization of ExposedBlob.bytes
+                        // will replace the original stream with a ByteArrayInputStream containing
+                        // the fully read stream.
+                        _args.add(sqlType to it.inputStream)
+                    } else {
+                        _args.add(sqlType to it)
+                    }
                     append("?")
                 } else {
-                    append(it.second)
+                    append(sqlType.valueToString(it))
                 }
             }
     }
diff --git a/exposed-jdbc/src/main/kotlin/org/jetbrains/exposed/sql/statements/jdbc/JdbcPreparedStatementImpl.kt b/exposed-jdbc/src/main/kotlin/org/jetbrains/exposed/sql/statements/jdbc/JdbcPreparedStatementImpl.kt
index e6edcac443..908d18e02e 100644
--- a/exposed-jdbc/src/main/kotlin/org/jetbrains/exposed/sql/statements/jdbc/JdbcPreparedStatementImpl.kt
+++ b/exposed-jdbc/src/main/kotlin/org/jetbrains/exposed/sql/statements/jdbc/JdbcPreparedStatementImpl.kt
@@ -4,9 +4,12 @@ import org.jetbrains.exposed.sql.BinaryColumnType
 import org.jetbrains.exposed.sql.BlobColumnType
 import org.jetbrains.exposed.sql.IColumnType
 import org.jetbrains.exposed.sql.statements.api.PreparedStatementApi
+import java.io.ByteArrayInputStream
+import java.io.FileInputStream
 import java.io.InputStream
 import java.sql.PreparedStatement
 import java.sql.ResultSet
+import java.sql.SQLFeatureNotSupportedException
 import java.sql.Types
 
 class JdbcPreparedStatementImpl(val statement: PreparedStatement, val wasGeneratedKeysRequested: Boolean) : PreparedStatementApi {
@@ -15,7 +18,9 @@ class JdbcPreparedStatementImpl(val statement: PreparedStatement, val wasGenerat
 
     override var fetchSize: Int?
         get() = statement.fetchSize
-        set(value) { value?.let { statement.fetchSize = value } }
+        set(value) {
+            value?.let { statement.fetchSize = value }
+        }
 
     override fun addBatch() {
         statement.addBatch()
@@ -38,7 +43,25 @@ class JdbcPreparedStatementImpl(val statement: PreparedStatement, val wasGenerat
     }
 
     override fun setInputStream(index: Int, inputStream: InputStream) {
-        statement.setBinaryStream(index, inputStream, inputStream.available())
+        try {
+            when {
+                // streams with known length where available matches the actual length
+                inputStream is ByteArrayInputStream ->
+                    statement.setBinaryStream(index, inputStream, inputStream.available())
+
+                // FileInputStream.available() returns returns Int.MAX_VALUE
+                // if the underlying file is larger than 2GB
+                inputStream is FileInputStream && inputStream.available() < Int.MAX_VALUE ->
+                    statement.setBinaryStream(index, inputStream, inputStream.available())
+
+                // default handling for unknown length
+                else -> statement.setBinaryStream(index, inputStream)
+
+            }
+        } catch (e: SQLFeatureNotSupportedException) {
+            // fallback to bytes
+            statement.setBytes(index, inputStream.readAllBytes())
+        }
     }
 
     override fun closeIfPossible() {
diff --git a/exposed-tests/src/test/kotlin/org/jetbrains/exposed/sql/tests/shared/DDLTests.kt b/exposed-tests/src/test/kotlin/org/jetbrains/exposed/sql/tests/shared/DDLTests.kt
index 603d536575..c49b78ba3a 100644
--- a/exposed-tests/src/test/kotlin/org/jetbrains/exposed/sql/tests/shared/DDLTests.kt
+++ b/exposed-tests/src/test/kotlin/org/jetbrains/exposed/sql/tests/shared/DDLTests.kt
@@ -21,6 +21,8 @@ import org.jetbrains.exposed.sql.vendors.SQLServerDialect
 import org.jetbrains.exposed.sql.vendors.SQLiteDialect
 import org.junit.Test
 import org.postgresql.util.PGobject
+import java.io.ByteArrayInputStream
+import java.io.SequenceInputStream
 import java.util.*
 import kotlin.random.Random
 import kotlin.test.assertNotNull
@@ -407,7 +409,12 @@ class DDLTests : DatabaseTestsBase() {
             val shortBytes = "Hello there!".toByteArray()
             val longBytes = Random.nextBytes(1024)
             val shortBlob = ExposedBlob(shortBytes)
-            val longBlob = ExposedBlob(longBytes)
+            val longBlob = ExposedBlob(
+                inputStream = SequenceInputStream(
+                    ByteArrayInputStream(longBytes, 0, 512),
+                    ByteArrayInputStream(longBytes, 512, 512)
+                )
+            )
 //            if (currentDialectTest.dataTypeProvider.blobAsStream) {
 //                    SerialBlob(bytes)
 //                } else connection.createBlob().apply {