EXPOSED-121, allowing option for "real" blobs in postgres (#1822)

* EXPOSED-121, allowing "real" blobs in postgres

- Uses inputStream from ResultSet to avoid large memory allocations
- Default parameter for blobs now created by `lo_bytea`
- Uses setBlob() for postgres, other DBs are fine with setBinaryStream()
- Changed code according to comments from @bog-walk

* EXPOSED-121 updates Table#blob documentation

* EXPOSED-121 update from apiDump

- Blob-handling introduced a new boolean parameter

* EXPOSED-121 re-introduced stream-length for setBinaryStream
- fix for sqlite semantic

exposed-core/api/exposed-core.api

@@ -236,6 +236,9 @@ public class org/jetbrains/exposed/sql/BinaryColumnType : org/jetbrains/exposed/
 
 public final class org/jetbrains/exposed/sql/BlobColumnType : org/jetbrains/exposed/sql/ColumnType {
 	public fun <init> ()V
+	public fun <init> (Z)V
+	public synthetic fun <init> (ZILkotlin/jvm/internal/DefaultConstructorMarker;)V
+	public final fun getUseObjectIdentifier ()Z
 	public fun nonNullValueToString (Ljava/lang/Object;)Ljava/lang/String;
 	public fun notNullValueToDB (Ljava/lang/Object;)Ljava/lang/Object;
 	public synthetic fun readObject (Ljava/sql/ResultSet;I)Ljava/lang/Object;
@@ -1534,7 +1537,8 @@ public final class org/jetbrains/exposed/sql/OpKt {
 	public static final fun andNot (Lorg/jetbrains/exposed/sql/Expression;Lkotlin/jvm/functions/Function1;)Lorg/jetbrains/exposed/sql/Op;
 	public static final fun arrayLiteral (Ljava/util/List;Lorg/jetbrains/exposed/sql/ColumnType;)Lorg/jetbrains/exposed/sql/LiteralOp;
 	public static final fun arrayParam (Ljava/util/List;Lorg/jetbrains/exposed/sql/ColumnType;)Lorg/jetbrains/exposed/sql/Expression;
-	public static final fun blobParam (Lorg/jetbrains/exposed/sql/statements/api/ExposedBlob;)Lorg/jetbrains/exposed/sql/Expression;
+	public static final fun blobParam (Lorg/jetbrains/exposed/sql/statements/api/ExposedBlob;Z)Lorg/jetbrains/exposed/sql/Expression;
+	public static synthetic fun blobParam$default (Lorg/jetbrains/exposed/sql/statements/api/ExposedBlob;ZILjava/lang/Object;)Lorg/jetbrains/exposed/sql/Expression;
 	public static final fun booleanLiteral (Z)Lorg/jetbrains/exposed/sql/LiteralOp;
 	public static final fun booleanParam (Z)Lorg/jetbrains/exposed/sql/Expression;
 	public static final fun byteLiteral (B)Lorg/jetbrains/exposed/sql/LiteralOp;
@@ -2213,7 +2217,8 @@ public class org/jetbrains/exposed/sql/Table : org/jetbrains/exposed/sql/ColumnS
 	public static synthetic fun autoinc$default (Lorg/jetbrains/exposed/sql/Table;Lorg/jetbrains/exposed/sql/Column;Ljava/lang/String;ILjava/lang/Object;)Lorg/jetbrains/exposed/sql/Column;
 	public final fun binary (Ljava/lang/String;)Lorg/jetbrains/exposed/sql/Column;
 	public final fun binary (Ljava/lang/String;I)Lorg/jetbrains/exposed/sql/Column;
-	public final fun blob (Ljava/lang/String;)Lorg/jetbrains/exposed/sql/Column;
+	public final fun blob (Ljava/lang/String;Z)Lorg/jetbrains/exposed/sql/Column;
+	public static synthetic fun blob$default (Lorg/jetbrains/exposed/sql/Table;Ljava/lang/String;ZILjava/lang/Object;)Lorg/jetbrains/exposed/sql/Column;
 	public final fun bool (Ljava/lang/String;)Lorg/jetbrains/exposed/sql/Column;
 	public final fun byte (Ljava/lang/String;)Lorg/jetbrains/exposed/sql/Column;
 	public final fun char (Ljava/lang/String;)Lorg/jetbrains/exposed/sql/Column;
@@ -3147,7 +3152,7 @@ public abstract interface class org/jetbrains/exposed/sql/statements/api/Prepare
 	public abstract fun set (ILjava/lang/Object;)V
 	public abstract fun setArray (ILjava/lang/String;[Ljava/lang/Object;)V
 	public abstract fun setFetchSize (Ljava/lang/Integer;)V
-	public abstract fun setInputStream (ILjava/io/InputStream;)V
+	public abstract fun setInputStream (ILjava/io/InputStream;Z)V
 	public abstract fun setNull (ILorg/jetbrains/exposed/sql/IColumnType;)V
 	public abstract fun setTimeout (Ljava/lang/Integer;)V
 }

exposed-core/src/main/kotlin/org/jetbrains/exposed/sql/ColumnType.kt

@@ -809,12 +809,17 @@ open class BinaryColumnType(
 /**
  * Binary column for storing BLOBs.
  */
-class BlobColumnType : ColumnType() {
-    override fun sqlType(): String = currentDialect.dataTypeProvider.blobType()
-
+class BlobColumnType(
+    /** Returns whether an OID column should be used instead of BYTEA. This value only applies to PostgreSQL databases. */
+    val useObjectIdentifier: Boolean = false
+) : ColumnType() {
+    override fun sqlType(): String = when {
+        useObjectIdentifier && currentDialect is PostgreSQLDialect -> "oid"
+        useObjectIdentifier -> error("Storing BLOBs using OID columns is only supported by PostgreSQL")
+        else -> currentDialect.dataTypeProvider.blobType()
+    }
     override fun valueFromDB(value: Any): ExposedBlob = when (value) {
         is ExposedBlob -> value
-        is Blob -> ExposedBlob(value.binaryStream)
         is InputStream -> ExposedBlob(value)
         is ByteArray -> ExposedBlob(value)
         else -> error("Unexpected value of type Blob: $value of ${value::class.qualifiedName}")
@@ -838,12 +843,13 @@ class BlobColumnType : ColumnType() {
 
     override fun readObject(rs: ResultSet, index: Int) = when {
         currentDialect is SQLServerDialect -> rs.getBytes(index)?.let(::ExposedBlob)
+        currentDialect is PostgreSQLDialect && useObjectIdentifier -> rs.getBlob(index)?.binaryStream?.let(::ExposedBlob)
         else -> rs.getBinaryStream(index)?.let(::ExposedBlob)
     }
 
     override fun setParameter(stmt: PreparedStatementApi, index: Int, value: Any?) {
         when (val toSetValue = (value as? ExposedBlob)?.inputStream ?: value) {
-            is InputStream -> stmt.setInputStream(index, toSetValue)
+            is InputStream -> stmt.setInputStream(index, toSetValue, useObjectIdentifier)
             null, is Op.NULL -> stmt.setNull(index, this)
             else -> super.setParameter(stmt, index, toSetValue)
         }

exposed-core/src/main/kotlin/org/jetbrains/exposed/sql/Op.kt

@@ -745,8 +745,14 @@ fun stringParam(value: String): Expression<String> = QueryParameter(value, TextC
 /** Returns the specified [value] as a decimal query parameter. */
 fun decimalParam(value: BigDecimal): Expression<BigDecimal> = QueryParameter(value, DecimalColumnType(value.precision(), value.scale()))
 
-/** Returns the specified [value] as a blob query parameter. */
-fun blobParam(value: ExposedBlob): Expression<ExposedBlob> = QueryParameter(value, BlobColumnType())
+/**
+ * Returns the specified [value] as a blob query parameter.
+ *
+ * Set [useObjectIdentifier] to `true` if the parameter should be processed using an OID column instead of a
+ * BYTEA column. This is only supported by PostgreSQL databases.
+ */
+fun blobParam(value: ExposedBlob, useObjectIdentifier: Boolean = false): Expression<ExposedBlob> =
+    QueryParameter(value, BlobColumnType(useObjectIdentifier))
 
 /** Returns the specified [value] as an array query parameter, with elements parsed by the [delegateType]. */
 fun <T> arrayParam(value: List<T>, delegateType: ColumnType): Expression<List<T>> =

exposed-core/src/main/kotlin/org/jetbrains/exposed/sql/Table.kt

@@ -756,10 +756,13 @@ open class Table(name: String = "") : ColumnSet(), DdlAware {
 
     /**
      * Creates a binary column, with the specified [name], for storing BLOBs.
+     * If [useObjectIdentifier] is `true`, then the column will use the `OID` type on PostgreSQL
+     * for storing large binary objects. The parameter must not be `true` for other databases.
      *
      * @sample org.jetbrains.exposed.sql.tests.shared.DDLTests.testBlob
      */
-    fun blob(name: String): Column<ExposedBlob> = registerColumn(name, BlobColumnType())
+    fun blob(name: String, useObjectIdentifier: Boolean = false): Column<ExposedBlob> =
+        registerColumn(name, BlobColumnType(useObjectIdentifier))
 
     /** Creates a binary column, with the specified [name], for storing UUIDs. */
     fun uuid(name: String): Column<UUID> = registerColumn(name, UUIDColumnType())

exposed-core/src/main/kotlin/org/jetbrains/exposed/sql/statements/api/PreparedStatementApi.kt

@@ -61,8 +61,11 @@ interface PreparedStatementApi {
     /** Sets the statement parameter at the [index] position to SQL NULL, if allowed wih the specified [columnType]. */
     fun setNull(index: Int, columnType: IColumnType)
 
-    /** Sets the statement parameter at the [index] position to the provided [inputStream]. */
-    fun setInputStream(index: Int, inputStream: InputStream)
+    /**
+     * Sets the statement parameter at the [index] position to the provided [inputStream],
+     * either directly as a BLOB if `setAsBlobObject` is `true` or as determined by the driver.
+     */
+    fun setInputStream(index: Int, inputStream: InputStream, setAsBlobObject: Boolean)
 
     /** Sets the statement parameter at the [index] position to the provided [array] of SQL [type]. */
     fun setArray(index: Int, type: String, array: Array<*>)

exposed-core/src/main/kotlin/org/jetbrains/exposed/sql/vendors/PostgreSQL.kt

@@ -28,6 +28,9 @@ internal object PostgreSQLDataTypeProvider : DataTypeProvider() {
             val cast = if (e.columnType.usesBinaryFormat) "::jsonb" else "::json"
             "${super.processForDefaultValue(e)}$cast"
         }
+        e is LiteralOp<*> && e.columnType is BlobColumnType && e.columnType.useObjectIdentifier && (currentDialect as? H2Dialect) == null -> {
+            "lo_from_bytea(0, ${super.processForDefaultValue(e)} :: bytea)"
+        }
         e is LiteralOp<*> && e.columnType is ArrayColumnType -> {
             val processed = super.processForDefaultValue(e)
             processed

exposed-jdbc/api/exposed-jdbc.api

@@ -77,7 +77,7 @@ public final class org/jetbrains/exposed/sql/statements/jdbc/JdbcPreparedStateme
 	public fun set (ILjava/lang/Object;)V
 	public fun setArray (ILjava/lang/String;[Ljava/lang/Object;)V
 	public fun setFetchSize (Ljava/lang/Integer;)V
-	public fun setInputStream (ILjava/io/InputStream;)V
+	public fun setInputStream (ILjava/io/InputStream;Z)V
 	public fun setNull (ILorg/jetbrains/exposed/sql/IColumnType;)V
 	public fun setTimeout (Ljava/lang/Integer;)V
 }

exposed-jdbc/src/main/kotlin/org/jetbrains/exposed/sql/statements/jdbc/JdbcPreparedStatementImpl.kt

@@ -72,15 +72,19 @@ class JdbcPreparedStatementImpl(
     }
 
     override fun setNull(index: Int, columnType: IColumnType) {
-        if (columnType is BinaryColumnType || columnType is BlobColumnType) {
+        if (columnType is BinaryColumnType || (columnType is BlobColumnType && !columnType.useObjectIdentifier)) {
             statement.setNull(index, Types.LONGVARBINARY)
         } else {
             statement.setObject(index, null)
         }
     }
 
-    override fun setInputStream(index: Int, inputStream: InputStream) {
-        statement.setBinaryStream(index, inputStream, inputStream.available())
+    override fun setInputStream(index: Int, inputStream: InputStream, setAsBlobObject: Boolean) {
+        if (setAsBlobObject) {
+            statement.setBlob(index, inputStream)
+        } else {
+            statement.setBinaryStream(index, inputStream, inputStream.available())
+        }
     }
 
     override fun setArray(index: Int, type: String, array: Array<*>) {

exposed-tests/src/test/kotlin/org/jetbrains/exposed/sql/tests/shared/DDLTests.kt

@@ -20,13 +20,15 @@ import org.jetbrains.exposed.sql.transactions.transaction
 import org.jetbrains.exposed.sql.vendors.H2Dialect
 import org.jetbrains.exposed.sql.vendors.MysqlDialect
 import org.jetbrains.exposed.sql.vendors.OracleDialect
+import org.jetbrains.exposed.sql.vendors.PostgreSQLDialect
 import org.jetbrains.exposed.sql.vendors.SQLServerDialect
 import org.jetbrains.exposed.sql.vendors.SQLiteDialect
 import org.junit.Assume
 import org.junit.Test
 import org.postgresql.util.PGobject
 import java.util.*
 import kotlin.random.Random
+import kotlin.test.assertContentEquals
 import kotlin.test.assertNotNull
 import kotlin.test.expect
 
@@ -677,6 +679,46 @@ class DDLTests : DatabaseTestsBase() {
         }
     }
 
+    @Test
+    fun testBlobAsOid() {
+        val defaultBytes = "test".toByteArray()
+        val defaultBlob = ExposedBlob(defaultBytes)
+        val tester = object : Table("blob_tester") {
+            val blobCol = blob("blob_col", useObjectIdentifier = true).default(defaultBlob)
+        }
+
+        withDb {
+            if (currentDialectTest !is PostgreSQLDialect) {
+                expectException<IllegalStateException> {
+                    SchemaUtils.create(tester)
+                }
+            } else {
+                assertEquals("oid", tester.blobCol.descriptionDdl().split(" ")[1])
+                SchemaUtils.create(tester)
+
+                tester.insert {}
+
+                val result1 = tester.selectAll().single()[tester.blobCol]
+                assertContentEquals(defaultBytes, result1.bytes)
+
+                tester.insert {
+                    defaultBlob.inputStream.reset()
+                    it[blobCol] = defaultBlob
+                }
+                tester.insert {
+                    defaultBlob.inputStream.reset()
+                    it[blobCol] = blobParam(defaultBlob, useObjectIdentifier = true)
+                }
+
+                val result2 = tester.selectAll()
+                assertEquals(3, result2.count())
+                assertTrue(result2.all { it[tester.blobCol].bytes.contentEquals(defaultBytes) })
+
+                SchemaUtils.drop(tester)
+            }
+        }
+    }
+
     @Test
     fun testBinaryWithoutLength() {
         val tableWithBinary = object : Table("TableWithBinary") {