R3SOL-0 - merge release branch

components/ledger/ledger-persistence/src/integrationTest/kotlin/net/corda/ledger/persistence/utxo/tests/UtxoPersistenceServiceImplTest.kt

@@ -77,6 +77,8 @@ import net.corda.v5.ledger.utxo.observer.UtxoTokenPoolKey
 import net.corda.v5.ledger.utxo.transaction.UtxoLedgerTransaction
 import org.assertj.core.api.Assertions.assertThat
 import org.assertj.core.api.Assertions.assertThatThrownBy
+import org.hibernate.Session
+import org.hibernate.internal.SessionImpl
 import org.junit.jupiter.api.AfterAll
 import org.junit.jupiter.api.Assertions.assertNotNull
 import org.junit.jupiter.api.BeforeAll
@@ -98,12 +100,14 @@ import java.security.KeyPairGenerator
 import java.security.MessageDigest
 import java.security.PublicKey
 import java.security.spec.ECGenParameterSpec
+import java.sql.Connection
 import java.time.Duration
 import java.time.Instant
 import java.time.temporal.ChronoUnit
 import java.util.Random
 import java.util.UUID
 import java.util.concurrent.atomic.AtomicInteger
+import javax.persistence.EntityManager
 import javax.persistence.EntityManagerFactory
 
 @ExtendWith(ServiceExtension::class, BundleContextExtension::class)
@@ -180,7 +184,7 @@ class UtxoPersistenceServiceImplTest {
             filteredTransactionFactory = ctx.getSandboxSingletonService()
 
             persistenceService = UtxoPersistenceServiceImpl(
-                entityManagerFactory,
+                { getConnection(entityManagerFactory.createEntityManager()) },
                 repository,
                 serializationService,
                 digestService,
@@ -293,13 +297,17 @@ class UtxoPersistenceServiceImplTest {
         assertThat(retval).isEqualTo(expectedRetval)
     }
 
+    private fun getConnection(em: EntityManager): Connection {
+        return (em.unwrap(Session::class.java) as SessionImpl).connection()
+    }
+
     @Test
     fun `find unconsumed visible transaction states`() {
         val entityFactory = UtxoEntityFactory(entityManagerFactory)
         val transaction1 = createSignedTransaction()
         val transaction2 = createSignedTransaction()
-        entityManagerFactory.transaction { em ->
-
+        entityManagerFactory.createEntityManager().transaction { em ->
+            val conn = getConnection(em)
             em.createNativeQuery("DELETE FROM {h-schema}utxo_visible_transaction_output").executeUpdate()
 
             createTransactionEntity(entityFactory, transaction1, status = VERIFIED).also { em.persist(it) }
@@ -334,9 +342,9 @@ class UtxoPersistenceServiceImplTest {
                 )
             )
 
-            repository.persistVisibleTransactionOutputs(em, transaction1.id.toString(), Instant.now(), outputs)
-            repository.persistVisibleTransactionOutputs(em, transaction2.id.toString(), Instant.now(), outputs2)
-            repository.markTransactionVisibleStatesConsumed(em, listOf(StateRef(transaction2.id, 1)), Instant.now())
+            repository.persistVisibleTransactionOutputs(conn, transaction1.id.toString(), Instant.now(), outputs)
+            repository.persistVisibleTransactionOutputs(conn, transaction2.id.toString(), Instant.now(), outputs2)
+            repository.markTransactionVisibleStatesConsumed(conn, listOf(StateRef(transaction2.id, 1)), Instant.now())
         }
 
         val stateClass = TestContractState2::class.java
@@ -1302,7 +1310,11 @@ class UtxoPersistenceServiceImplTest {
 
             val visibleOutputIndexes = listOf(0)
             // prove that the output of filtered tx with index 0 is used as an input
-            val indexes = repository.findConsumedTransactionSourcesForTransaction(em, transactionIdString, visibleOutputIndexes)
+            val indexes = repository.findConsumedTransactionSourcesForTransaction(
+                getConnection(em),
+                transactionIdString,
+                visibleOutputIndexes
+            )
             assertThat(indexes).contains(0)
         }
 
@@ -1404,7 +1416,7 @@ class UtxoPersistenceServiceImplTest {
 
             val visibleOutputIndexes = listOf(0, 1)
             // prove that the output of unverified tx is not consumed in other tx
-            val indexes = repository.findConsumedTransactionSourcesForTransaction(em, txAId, visibleOutputIndexes)
+            val indexes = repository.findConsumedTransactionSourcesForTransaction(getConnection(em), txAId, visibleOutputIndexes)
             assertThat(indexes).isEmpty()
         }
 
@@ -1435,7 +1447,11 @@ class UtxoPersistenceServiceImplTest {
 
             val visibleOutputIndexes = listOf(0)
             // prove that the output of filtered tx with index 0 is used as an input
-            val indexes = repository.findConsumedTransactionSourcesForTransaction(em, transactionIdString, visibleOutputIndexes)
+            val indexes = repository.findConsumedTransactionSourcesForTransaction(
+                getConnection(em),
+                transactionIdString,
+                visibleOutputIndexes
+            )
             assertThat(indexes).contains(0)
         }
 

components/ledger/ledger-persistence/src/main/kotlin/net/corda/ledger/persistence/utxo/impl/UtxoRequestHandlerSelectorImpl.kt

@@ -47,6 +47,8 @@ import net.corda.persistence.common.getSerializationService
 import net.corda.sandboxgroupcontext.SandboxGroupContext
 import net.corda.sandboxgroupcontext.getSandboxSingletonService
 import net.corda.utilities.time.UTCClock
+import org.hibernate.Session
+import org.hibernate.internal.SessionImpl
 import org.osgi.service.component.annotations.Activate
 import org.osgi.service.component.annotations.Component
 import org.osgi.service.component.annotations.Reference
@@ -63,7 +65,10 @@ class UtxoRequestHandlerSelectorImpl @Activate constructor(
     @Suppress("LongMethod")
     override fun selectHandler(sandbox: SandboxGroupContext, request: LedgerPersistenceRequest): RequestHandler {
         val persistenceService = UtxoPersistenceServiceImpl(
-            entityManagerFactory = sandbox.getEntityManagerFactory(),
+            connectionFactory = {
+                val emf = sandbox.getEntityManagerFactory()
+                (emf.createEntityManager().unwrap(Session::class.java) as SessionImpl).connection()
+            },
             repository = sandbox.getSandboxSingletonService(),
             serializationService = sandbox.getSerializationService(),
             sandboxDigestService = sandbox.getSandboxSingletonService(),

components/ledger/ledger-persistence/src/test/kotlin/net/corda/ledger/persistence/utxo/UtxoComponentGroupMapperTest.kt

@@ -19,6 +19,7 @@ class UtxoComponentGroupMapperTest {
     private fun mockTuple(values: List<Any>) =
         mock<Tuple>().apply {
             whenever(this.get(anyInt())).thenAnswer { invocation -> values[invocation.arguments[0] as Int] }
+            whenever(this.toArray()).thenAnswer { values.toTypedArray() }
         }
 
     @Test

components/ledger/ledger-persistence/src/test/kotlin/net/corda/ledger/persistence/utxo/impl/UtxoPersistenceServiceImplTest.kt

@@ -28,6 +28,8 @@ import net.corda.v5.ledger.utxo.StateRef
 import net.corda.v5.ledger.utxo.TransactionState
 import net.corda.v5.ledger.utxo.query.json.ContractStateVaultJsonFactory
 import org.assertj.core.api.Assertions.assertThat
+import org.hibernate.Session
+import org.hibernate.internal.SessionImpl
 import org.junit.jupiter.api.Assertions.assertEquals
 import org.junit.jupiter.api.BeforeEach
 import org.junit.jupiter.api.Test
@@ -37,8 +39,8 @@ import org.mockito.kotlin.doAnswer
 import org.mockito.kotlin.doReturn
 import org.mockito.kotlin.mock
 import org.mockito.kotlin.whenever
-import java.lang.IllegalArgumentException
 import java.security.PublicKey
+import java.sql.Connection
 import javax.persistence.EntityManager
 import javax.persistence.EntityManagerFactory
 
@@ -73,16 +75,24 @@ class UtxoPersistenceServiceImplTest {
         registerJsonFactory(InvalidStateJsonFactory() as ContractStateVaultJsonFactory<ContractState>)
     }
 
+    private val connectionMock = mock<Connection> {
+    }
+
+    private val mockSession = mock<SessionImpl> {
+        on { connection() } doReturn mock()
+    }
+
     private val mockEm = mock<EntityManager> {
         on { transaction } doReturn mock()
+        on { unwrap(Session::class.java) } doReturn mockSession
     }
 
     private val mockEmFactory = mock<EntityManagerFactory> {
         on { createEntityManager() }.doReturn(mockEm)
     }
 
     private val persistenceService = UtxoPersistenceServiceImpl(
-        mockEmFactory,
+        { connectionMock },
         mockRepository,
         mock(),
         mockDigestService,
@@ -140,7 +150,7 @@ class UtxoPersistenceServiceImplTest {
         whenever(emptyDefaultContractStateVaultJsonFactory.create(any(), any())).thenReturn("")
 
         val singlePersistenceService = UtxoPersistenceServiceImpl(
-            mockEmFactory,
+            { connectionMock },
             mockRepository,
             mock(),
             mockDigestService,
@@ -236,7 +246,7 @@ class UtxoPersistenceServiceImplTest {
     @Test
     fun `Persisting a transaction while zero JSON factories are registered will result still store the default state json`() {
         val emptyPersistenceService = UtxoPersistenceServiceImpl(
-            mockEmFactory,
+            { connectionMock },
             mockRepository,
             mock(),
             mockDigestService,
@@ -281,7 +291,7 @@ class UtxoPersistenceServiceImplTest {
         }
 
         val persistenceService = UtxoPersistenceServiceImpl(
-            mockEmFactory,
+            { connectionMock },
             mockRepository,
             mock(),
             mockDigestService,

components/ledger/ledger-utxo-flow/src/main/kotlin/net/corda/ledger/utxo/flow/impl/flows/backchain/TransactionBackchainVerifierImpl.kt

@@ -46,7 +46,7 @@ class TransactionBackchainVerifierImpl @Activate constructor(
             val (transaction, status) = utxoLedgerPersistenceService.findSignedLedgerTransactionWithStatus(
                 transactionId,
                 UNVERIFIED
-            ) ?: throw CordaRuntimeException("Transaction does not exist locally")
+            ) ?: throw CordaRuntimeException("Transaction ($UNVERIFIED) with id: $transactionId does not exist locally")
             when (status) {
                 INVALID -> {
                     log.warn(

libs/db/db-core/src/main/kotlin/net/corda/db/core/utils/ConnectionUtils.kt

@@ -1,11 +1,14 @@
 package net.corda.db.core.utils
 
+import org.slf4j.LoggerFactory
 import java.sql.Connection
 
+private val log = LoggerFactory.getLogger("ConnectionUtils")
+
 /**
  * Executes [block] in a transaction using the [Connection].
  *
- * Commits transaction if no exceptions were thrown by [block]. Otherwise rolls back the transaction.
+ * Commits transaction if no exceptions were thrown by [block]. Otherwise, rolls back the transaction.
  *
  * Finally closes the connection after committing or rolling back the changes.
  *
@@ -14,12 +17,21 @@ import java.sql.Connection
  *
  * @return The result of executing [block].
  */
-inline fun <R> Connection.transaction(block: (Connection) -> R): R {
+fun <R> Connection.transaction(block: (Connection) -> R): R {
+    return transactionWithLogging(null, block)
+}
+
+fun <R> Connection.transactionWithLogging(name: String?, block: (Connection) -> R): R {
+    if(null != name && log.isTraceEnabled) log.trace("Start transaction $name")
     autoCommit = false
     return try {
-        block(this).also { commit() }
+        block(this).also {
+            commit()
+            if(null != name && log.isTraceEnabled) log.trace("Transaction $name committed")
+        }
     } catch (e: Exception) {
         rollback()
+        if(null != name && log.isWarnEnabled) log.error("Transaction $name rolled back")
         throw e
     } finally {
         close()

libs/ledger-lib-persistence/build.gradle

@@ -17,7 +17,6 @@ dependencies {
     implementation project(':libs:ledger:ledger-common-data')
     implementation project(':libs:ledger:ledger-utxo-data')
     implementation project(":libs:db:db-core")
-    implementation project(":libs:db:db-orm")
     implementation project(":libs:utilities")
     implementation project(":libs:serialization:json-validator-lib")
 
@@ -26,7 +25,6 @@ dependencies {
     implementation 'net.corda:corda-ledger-common'
     implementation 'net.corda:corda-ledger-utxo'
     implementation libs.slf4j.api
-    implementation "org.hibernate:hibernate-core:$hibernateVersion"
     implementation 'org.jetbrains.kotlin:kotlin-stdlib'
     implementation libs.jackson.core
 }

libs/ledger-lib-persistence/src/main/kotlin/net/corda/ledger/libs/persistence/common/ComponentGroupMapper.kt

@@ -8,4 +8,10 @@ interface ComponentGroupMapper {
     fun map(tuples: List<Tuple>): Map<Int, List<ByteArray>>
 }
 
+interface ComponentGroupArrayMapper {
+    @Throws(SQLException::class)
+    fun mapColumns(tuples: List<Array<Any?>>): Map<Int, List<ByteArray>>
+}
+
 fun List<Tuple>.mapToComponentGroups(mapper: ComponentGroupMapper): Map<Int, List<ByteArray>> = mapper.map(this)
+fun List<Array<Any?>>.mapToComponentGroups(mapper: ComponentGroupArrayMapper): Map<Int, List<ByteArray>> = mapper.mapColumns(this)

libs/ledger-lib-persistence/src/main/kotlin/net/corda/ledger/libs/persistence/util/NamedParamQuery.kt

@@ -0,0 +1,48 @@
+package net.corda.ledger.libs.persistence.util
+
+interface NamedParamQuery {
+    companion object {
+        fun from(sql: String): NamedParamQuery {
+            val plainSql = StringBuilder()
+            val fields = mutableMapOf<String, Int>()
+            var marker: StringBuilder? = null
+            var markerIndex = 0
+            for (i in sql.indices) {
+                val c = sql[i]
+                if (c == ':' && (i < sql.length - 1 && sql[i + 1].isValidTokenChar())) {
+                    marker = StringBuilder()
+                    markerIndex++
+                    plainSql.append('?')
+                    continue
+                }
+
+                if (null != marker) {
+                    if (!c.isValidTokenChar()) {
+                        fields[marker.toString()] = markerIndex
+                        marker = null
+                    } else {
+                        marker.append(c)
+                        continue
+                    }
+                }
+                plainSql.append(c)
+            }
+
+            if (null != marker) {
+                fields[marker.toString()] = markerIndex
+            }
+
+            return NamedParamQueryImpl(plainSql.toString(), fields)
+        }
+
+        private fun Char.isValidTokenChar(): Boolean = this.isLetterOrDigit() || this == '_' || this == '-'
+    }
+
+    val sql: String
+    val fields: Map<String, Int>
+
+    private class NamedParamQueryImpl(
+        override val sql: String,
+        override val fields: Map<String, Int>,
+    ) : NamedParamQuery
+}