CORE-17430 Persist reference states into utxo_transaction_sources tab…

…le and align with column changes (#4849)

The utxo_transaction_sources table has been modified. See CORE-17430 Modify the utxo_transaction_sources table by removing unused columns and renaming corda-api#1288. This PR contains the changes to align with those DB changes.
Reference states were not persisted in utxo_transaction_sources until now. Added that persisting logic.

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

@@ -207,7 +207,7 @@ class UtxoLedgerMessageProcessorTests {
                 listOf("4".toByteArray()),
                 listOf("5".toByteArray()),
                 emptyList(),
-                listOf("7".toByteArray()),
+                emptyList(),
                 listOf(outputState),
                 listOf("9".toByteArray())
             ),

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

@@ -443,6 +443,46 @@ class UtxoPersistenceServiceImplTest {
             assertThat(dbMetadata.field<String>("groupParametersHash")).isNotNull
             assertThat(dbMetadata.field<String>("cpiFileChecksum")).isNotNull
 
+            val dbTransactionSources = em.createNamedQuery(
+                "UtxoTransactionSourceEntity.findByTransactionId",
+                entityFactory.utxoTransactionSource
+            )
+                .setParameter("transactionId", signedTransaction.id.toString())
+                .resultList
+
+            assertThat(dbTransactionSources).allMatch {
+                it.field<Int>("groupIndex") == UtxoComponentGroup.INPUTS.ordinal
+                        || it.field<Int>("groupIndex") == UtxoComponentGroup.REFERENCES.ordinal
+            }
+
+            val (dbTransactionInputs, dbTransactionReferences) = dbTransactionSources.partition {
+                it.field<Int>("groupIndex") == UtxoComponentGroup.INPUTS.ordinal
+            }
+
+            assertThat(dbTransactionInputs).isNotNull
+                .hasSameSizeAs(defaultInputStateRefs)
+
+            assertThat(dbTransactionReferences).isNotNull
+                .hasSameSizeAs(defaultReferenceStateRefs)
+
+            dbTransactionInputs
+                .sortedWith(compareBy<Any> { it.field<Int>("groupIndex") }.thenBy { it.field<Int>("leafIndex") })
+                .zip(defaultInputStateRefs)
+                .forEachIndexed { leafIndex, (dbInput, transactionInput) ->
+                    assertThat(dbInput.field<Int>("leafIndex")).isEqualTo(leafIndex)
+                    assertThat(dbInput.field<String>("refTransactionId")).isEqualTo(transactionInput.transactionId.toString())
+                    assertThat(dbInput.field<Int>("refLeafIndex")).isEqualTo(transactionInput.index)
+                }
+
+            dbTransactionReferences
+                .sortedWith(compareBy<Any> { it.field<Int>("groupIndex") }.thenBy { it.field<Int>("leafIndex") })
+                .zip(defaultReferenceStateRefs)
+                .forEachIndexed { leafIndex, (dbInput, transactionInput) ->
+                    assertThat(dbInput.field<Int>("leafIndex")).isEqualTo(leafIndex)
+                    assertThat(dbInput.field<String>("refTransactionId")).isEqualTo(transactionInput.transactionId.toString())
+                    assertThat(dbInput.field<Int>("refLeafIndex")).isEqualTo(transactionInput.index)
+                }
+
             val dbTransactionOutputs = em.createNamedQuery(
                 "UtxoVisibleTransactionOutputEntity.findByTransactionId",
                 entityFactory.utxoVisibleTransactionOutput
@@ -697,6 +737,10 @@ class UtxoPersistenceServiceImplTest {
             TODO("Not yet implemented")
         }
 
+        override fun getReferenceStateRefs(): List<StateRef> {
+            return listOf(StateRef(SecureHashImpl("SHA-256", ByteArray(34)), 2))
+        }
+
         override fun getConsumedStateRefs(): List<StateRef> {
             return listOf(StateRef(SecureHashImpl("SHA-256", ByteArray(12)), 1))
         }

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

@@ -12,6 +12,7 @@ class UtxoEntityFactory(private val entityManagerFactory: EntityManagerFactory)
     val utxoTransactionComponent: Class<*> get() = classFor("UtxoTransactionComponentEntity")
     val utxoVisibleTransactionOutput: Class<*> get() = classFor("UtxoVisibleTransactionOutputEntity")
     val utxoTransactionSignature: Class<*> get() = classFor("UtxoTransactionSignatureEntity")
+    val utxoTransactionSource: Class<*> get() = classFor("UtxoTransactionSourceEntity")
 
     fun createUtxoTransactionEntity(
         transactionId: String,

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

@@ -89,6 +89,17 @@ interface UtxoRepository {
         cpiFileChecksum: String
     )
 
+    /** Persists transaction source (operation is idempotent) */
+    @Suppress("LongParameterList")
+    fun persistTransactionSource(
+        entityManager: EntityManager,
+        transactionId: String,
+        groupIndex: Int,
+        leafIndex: Int,
+        sourceStateTransactionId: String,
+        sourceStateIndex: Int
+    )
+
     /** Persists transaction component leaf [data] (operation is idempotent) */
     @Suppress("LongParameterList")
     fun persistTransactionComponentLeaf(

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

@@ -35,4 +35,6 @@ interface UtxoTransactionReader {
     fun getConsumedStates(persistenceService: UtxoPersistenceService): List<StateAndRef<ContractState>>
 
     fun getConsumedStateRefs(): List<StateRef>
+
+    fun getReferenceStateRefs(): List<StateRef>
 }

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

@@ -33,10 +33,19 @@ class PostgresUtxoQueryProvider @Activate constructor(
             ON CONFLICT DO NOTHING"""
             .trimIndent()
 
+    override val persistTransactionSource: String
+        get() = """
+            INSERT INTO {h-schema}utxo_transaction_sources(
+                transaction_id, group_idx, leaf_idx, source_state_transaction_id, source_state_idx)
+            VALUES(
+                :transactionId, :groupIndex, :leafIndex, :sourceStateTransactionId, :sourceStateIndex)
+            ON CONFLICT DO NOTHING"""
+            .trimIndent()
+
     override val persistTransactionComponentLeaf: String
         get() = """
             INSERT INTO {h-schema}utxo_transaction_component(transaction_id, group_idx, leaf_idx, data, hash)
-            VALUES(:transactionId, :groupIndex, :leafIndex, :data, :hash)
+                VALUES(:transactionId, :groupIndex, :leafIndex, :data, :hash)
             ON CONFLICT DO NOTHING"""
             .trimIndent()
 

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

@@ -182,6 +182,30 @@ class UtxoPersistenceServiceImpl(
             }
         }
 
+        // Insert inputs data
+        transaction.getConsumedStateRefs().forEachIndexed { index, input ->
+            repository.persistTransactionSource(
+                em,
+                transactionIdString,
+                UtxoComponentGroup.INPUTS.ordinal,
+                index,
+                input.transactionId.toString(),
+                input.index
+            )
+        }
+
+        // Insert reference data
+        transaction.getReferenceStateRefs().forEachIndexed { index, reference ->
+            repository.persistTransactionSource(
+                em,
+                transactionIdString,
+                UtxoComponentGroup.REFERENCES.ordinal,
+                index,
+                reference.transactionId.toString(),
+                reference.index
+            )
+        }
+
         // Insert outputs data
         transaction.getVisibleStates().entries.forEach { (stateIndex, stateAndRef) ->
             val utxoToken = utxoTokenMap[stateAndRef.ref]

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

@@ -66,6 +66,11 @@ interface UtxoQueryProvider {
      */
     val persistTransactionMetadata: String
 
+    /**
+     * @property persistTransactionSource SQL text for [UtxoRepositoryImpl.persistTransactionSource].
+     */
+    val persistTransactionSource: String
+
     /**
      * @property persistTransactionComponentLeaf SQL text for [UtxoRepositoryImpl.persistTransactionComponentLeaf].
      */

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

@@ -214,6 +214,24 @@ class UtxoRepositoryImpl @Activate constructor(
             .logResult("transaction metadata [$hash]")
     }
 
+    override fun persistTransactionSource(
+        entityManager: EntityManager,
+        transactionId: String,
+        groupIndex: Int,
+        leafIndex: Int,
+        sourceStateTransactionId: String,
+        sourceStateIndex: Int
+    ) {
+        entityManager.createNativeQuery(queryProvider.persistTransactionSource)
+            .setParameter("transactionId", transactionId)
+            .setParameter("groupIndex", groupIndex)
+            .setParameter("leafIndex", leafIndex)
+            .setParameter("sourceStateTransactionId", sourceStateTransactionId)
+            .setParameter("sourceStateIndex", sourceStateIndex)
+            .executeUpdate()
+            .logResult("transaction source [$transactionId, $groupIndex, $leafIndex]")
+    }
+
     override fun persistTransactionComponentLeaf(
         entityManager: EntityManager,
         transactionId: String,

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

@@ -121,4 +121,6 @@ class UtxoTransactionReaderImpl(
     }
 
     override fun getConsumedStateRefs(): List<StateRef> = wrappedWireTransaction.inputStateRefs
+
+    override fun getReferenceStateRefs(): List<StateRef> = wrappedWireTransaction.referenceStateRefs
 }

components/ledger/ledger-persistence/testing-datamodel/src/main/kotlin/com/example/ledger/testing/datamodel/utxo/UtxoTransactionSourceEntity.kt

@@ -0,0 +1,49 @@
+package com.example.ledger.testing.datamodel.utxo
+
+import net.corda.v5.base.annotations.CordaSerializable
+import java.io.Serializable
+import javax.persistence.Column
+import javax.persistence.Embeddable
+import javax.persistence.Entity
+import javax.persistence.Id
+import javax.persistence.IdClass
+import javax.persistence.JoinColumn
+import javax.persistence.ManyToOne
+import javax.persistence.NamedQuery
+import javax.persistence.Table
+
+@CordaSerializable
+@NamedQuery(
+    name = "UtxoTransactionSourceEntity.findByTransactionId",
+    query = "from UtxoTransactionSourceEntity where transaction.id = :transactionId"
+)
+@Entity
+@Table(name = "utxo_transaction_sources")
+@IdClass(UtxoTransactionSourceEntityId::class)
+data class UtxoTransactionSourceEntity(
+    @get:Id
+    @get:ManyToOne
+    @get:JoinColumn(name = "transaction_id", nullable = false, updatable = false)
+    var transaction: UtxoTransactionEntity,
+
+    @get:Id
+    @get:Column(name = "group_idx", nullable = false)
+    var groupIndex: Int,
+
+    @get:Id
+    @get:Column(name = "leaf_idx", nullable = false)
+    var leafIndex: Int,
+
+    @get:Column(name = "source_state_transaction_id", nullable = false)
+    var refTransactionId: String,
+
+    @get:Column(name = "source_state_idx", nullable = false)
+    var refLeafIndex: Int
+)
+
+@Embeddable
+data class UtxoTransactionSourceEntityId(
+    var transaction: UtxoTransactionEntity,
+    var groupIndex: Int,
+    var leafIndex: Int
+) : Serializable

gradle.properties

@@ -46,7 +46,7 @@ commonsTextVersion = 1.10.0
 bouncycastleVersion=1.76
 # Corda API libs revision (change in 4th digit indicates a breaking change)
 # Change to 5.1.0.xx-SNAPSHOT to pick up maven local published copy
-cordaApiVersion=5.1.0.33-beta+
+cordaApiVersion=5.1.0.34-beta+
 
 disruptorVersion=3.4.4
 felixConfigAdminVersion=1.9.26

testing/ledger/ledger-hsqldb/src/integrationTest/kotlin/net/corda/testing/ledger/tests/HsqldbVaultNamedQueryTest.kt

@@ -8,6 +8,8 @@ import com.example.ledger.testing.datamodel.utxo.UtxoVisibleTransactionOutputEnt
 import com.example.ledger.testing.datamodel.utxo.UtxoVisibleTransactionOutputEntityId
 import com.example.ledger.testing.datamodel.utxo.UtxoTransactionSignatureEntity
 import com.example.ledger.testing.datamodel.utxo.UtxoTransactionSignatureEntityId
+import com.example.ledger.testing.datamodel.utxo.UtxoTransactionSourceEntity
+import com.example.ledger.testing.datamodel.utxo.UtxoTransactionSourceEntityId
 import java.time.Instant
 import java.util.UUID
 import javax.persistence.EntityManagerFactory
@@ -85,7 +87,9 @@ class HsqldbVaultNamedQueryTest {
             UtxoVisibleTransactionOutputEntity::class.java,
             UtxoVisibleTransactionOutputEntityId::class.java,
             UtxoTransactionSignatureEntity::class.java,
-            UtxoTransactionSignatureEntityId::class.java
+            UtxoTransactionSignatureEntityId::class.java,
+            UtxoTransactionSourceEntity::class.java,
+            UtxoTransactionSourceEntityId::class.java
         ))
         entityManagerFactory = dbConnectionManager.createEntityManagerFactory(dbConnectionId, entities)
     }

testing/ledger/ledger-hsqldb/src/main/kotlin/net/corda/testing/ledger/utxo/HsqldbUtxoQueryProvider.kt

@@ -41,6 +41,18 @@ class HsqldbUtxoQueryProvider @Activate constructor(
                 VALUES (x.hash, x.canonical_data, x.group_parameters_hash, x.cpi_file_checksum)"""
             .trimIndent()
 
+    override val persistTransactionSource: String
+        get() = """
+            MERGE INTO {h-schema}utxo_transaction_sources AS uts
+            USING (VALUES :transactionId, CAST(:groupIndex AS INT), CAST(:leafIndex AS INT),
+                          :sourceStateTransactionId, CAST(:sourceStateIndex AS INT))
+                AS x(transaction_id, group_idx, leaf_idx, source_state_transaction_id, source_state_idx)
+            ON uts.transaction_id = x.transaction_id AND uts.group_idx = x.group_idx AND uts.leaf_idx = x.leaf_idx
+            WHEN NOT MATCHED THEN
+                INSERT (transaction_id, group_idx, leaf_idx, source_state_transaction_id, source_state_idx)
+                VALUES (x.transaction_id, x.group_idx, x.leaf_idx, x.source_state_transaction_id, x.source_state_idx)"""
+            .trimIndent()
+
     override val persistTransactionComponentLeaf: String
         get() = """
             MERGE INTO {h-schema}utxo_transaction_component AS utc