From 919b75401990bdd5815fc2da61c9641724dbb13f Mon Sep 17 00:00:00 2001 From: Matt Dziuban Date: Tue, 25 Aug 2026 14:41:36 +0000 Subject: [PATCH] Use `DbStorage.toInClause` and remove `inClause`. Fixes #3900 Removes `inClause` in favor of `DbStorage.toInClause`. Signed-off-by: Matt Dziuban --- .../store/DbVotesStoreQueryBuilder.scala | 8 +- .../splice/store/UpdateHistory.scala | 113 ++++----- .../store/db/DbMultiDomainAcsStore.scala | 195 ++++++++-------- .../splice/store/db/Queries.scala | 11 - .../splice/scan/store/AcsSnapshotStore.scala | 17 +- .../store/db/DbAppActivityRecordStore.scala | 35 +-- .../scan/store/db/DbScanAppRewardsStore.scala | 20 +- .../splice/scan/store/db/DbScanStore.scala | 77 ++++--- .../scan/store/db/DbScanVerdictStore.scala | 75 +++--- .../splice/sv/store/db/DbSvDsoStore.scala | 217 ++++++++++-------- 10 files changed, 400 insertions(+), 368 deletions(-) diff --git a/apps/common/src/main/scala/org/lfdecentralizedtrust/splice/store/DbVotesStoreQueryBuilder.scala b/apps/common/src/main/scala/org/lfdecentralizedtrust/splice/store/DbVotesStoreQueryBuilder.scala index cf5227001a..e9d5c79e5e 100644 --- a/apps/common/src/main/scala/org/lfdecentralizedtrust/splice/store/DbVotesStoreQueryBuilder.scala +++ b/apps/common/src/main/scala/org/lfdecentralizedtrust/splice/store/DbVotesStoreQueryBuilder.scala @@ -5,8 +5,10 @@ package org.lfdecentralizedtrust.splice.store import cats.data.NonEmptyList import com.daml.ledger.javaapi.data.codegen.ContractId +import com.daml.nonempty.NonEmpty import com.digitalasset.canton.config.CantonRequireTypes.String3 import com.digitalasset.canton.logging.NamedLogging +import com.digitalasset.canton.resource.DbStorage import com.digitalasset.canton.resource.DbStorage.Implicits.BuilderChain.toSQLActionBuilderChain import org.lfdecentralizedtrust.splice.codegen.java.splice.dsorules.VoteRequest import org.lfdecentralizedtrust.splice.store.db.AcsQueries.AcsStoreId @@ -145,14 +147,14 @@ trait DbVotesAcsStoreQueryBuilder extends AcsQueries with LimitHelpers with Name acsStoreId: AcsStoreId, domainMigrationId: Long, trackingCidColumnName: String, - trackingCids: Seq[VoteRequest.ContractId], + trackingCids: NonEmpty[Seq[VoteRequest.ContractId]], limit: Limit, ): SqlStreamingAction[Vector[ AcsQueries.SelectFromAcsTableResult ], AcsQueries.SelectFromAcsTableResult, Effect.Read] = { - val cids: Seq[ContractId[?]] = trackingCids + val cids: NonEmpty[Seq[ContractId[?]]] = trackingCids val voteRequestTrackingCidsSql = - inClause(trackingCidColumnName, cids) + DbStorage.toInClause(trackingCidColumnName, cids) selectFromAcsTable( acsTableName, acsStoreId, diff --git a/apps/common/src/main/scala/org/lfdecentralizedtrust/splice/store/UpdateHistory.scala b/apps/common/src/main/scala/org/lfdecentralizedtrust/splice/store/UpdateHistory.scala index 3554789a01..57e4376f2e 100644 --- a/apps/common/src/main/scala/org/lfdecentralizedtrust/splice/store/UpdateHistory.scala +++ b/apps/common/src/main/scala/org/lfdecentralizedtrust/splice/store/UpdateHistory.scala @@ -9,6 +9,7 @@ import com.daml.ledger.api.v2.TraceContextOuterClass import com.daml.ledger.javaapi.data.codegen.{ContractId, DamlRecord} import com.daml.ledger.javaapi.data.{CreatedEvent, Event, ExercisedEvent, Identifier, Transaction} import com.daml.metrics.api.MetricsContext +import com.daml.nonempty.NonEmpty import com.google.protobuf.ByteString import com.digitalasset.canton.util.HexString import org.lfdecentralizedtrust.splice.environment.ledger.api.ReassignmentEvent.{Assign, Unassign} @@ -1339,33 +1340,33 @@ class UpdateHistory( private def queryCreateEvents( transactionRowIds: Seq[Long] )(implicit tc: TraceContext): Future[Map[Long, Seq[SelectFromCreateEvents]]] = { - if (transactionRowIds.isEmpty) { - Future.successful(Map.empty) - } else { - storage - .query( - (sql""" - select - update_row_id, - event_id, - contract_id, - created_at, - template_id_package_id, - template_id_module_name, - template_id_entity_name, - package_name, - create_arguments, - signatories, - observers, - contract_key, - record_time - - from update_history_creates - where """ ++ inClause("update_row_id", transactionRowIds)).toActionBuilder - .as[SelectFromCreateEvents], - "queryCreateEvents", - ) - .map(_.groupBy(_.updateRowId)) + NonEmpty.from(transactionRowIds) match { + case None => Future.successful(Map.empty) + case Some(transactionRowIds) => + storage + .query( + (sql""" + select + update_row_id, + event_id, + contract_id, + created_at, + template_id_package_id, + template_id_module_name, + template_id_entity_name, + package_name, + create_arguments, + signatories, + observers, + contract_key, + record_time + + from update_history_creates + where """ ++ DbStorage.toInClause("update_row_id", transactionRowIds)).toActionBuilder + .as[SelectFromCreateEvents], + "queryCreateEvents", + ) + .map(_.groupBy(_.updateRowId)) } } @@ -1410,35 +1411,35 @@ class UpdateHistory( private def queryExerciseEvents( transactionRowIds: Seq[Long] )(implicit tc: TraceContext): Future[Map[Long, Seq[SelectFromExerciseEvents]]] = { - if (transactionRowIds.isEmpty) { - Future.successful(Map.empty) - } else { - storage - .query( - (sql""" - select - update_row_id, - event_id, - child_event_ids, - choice, - template_id_package_id, - template_id_module_name, - template_id_entity_name, - contract_id, - consuming, - package_name, - argument, - result, - acting_parties, - interface_id_package_id, - interface_id_module_name, - interface_id_entity_name - from update_history_exercises - where """ ++ inClause("update_row_id", transactionRowIds)).toActionBuilder - .as[SelectFromExerciseEvents], - "queryExerciseEvents", - ) - .map(_.groupBy(_.updateRowId)) + NonEmpty.from(transactionRowIds) match { + case None => Future.successful(Map.empty) + case Some(transactionRowIds) => + storage + .query( + (sql""" + select + update_row_id, + event_id, + child_event_ids, + choice, + template_id_package_id, + template_id_module_name, + template_id_entity_name, + contract_id, + consuming, + package_name, + argument, + result, + acting_parties, + interface_id_package_id, + interface_id_module_name, + interface_id_entity_name + from update_history_exercises + where """ ++ DbStorage.toInClause("update_row_id", transactionRowIds)).toActionBuilder + .as[SelectFromExerciseEvents], + "queryExerciseEvents", + ) + .map(_.groupBy(_.updateRowId)) } } diff --git a/apps/common/src/main/scala/org/lfdecentralizedtrust/splice/store/db/DbMultiDomainAcsStore.scala b/apps/common/src/main/scala/org/lfdecentralizedtrust/splice/store/db/DbMultiDomainAcsStore.scala index 470d5e445b..2f3106530e 100644 --- a/apps/common/src/main/scala/org/lfdecentralizedtrust/splice/store/db/DbMultiDomainAcsStore.scala +++ b/apps/common/src/main/scala/org/lfdecentralizedtrust/splice/store/db/DbMultiDomainAcsStore.scala @@ -55,7 +55,7 @@ import org.lfdecentralizedtrust.splice.store.db.AcsQueries.{ } import org.lfdecentralizedtrust.splice.store.db.AcsTables.ContractStateRowData import AsUpdateReturning.* -import com.daml.nonempty.NonEmpty +import com.daml.nonempty.{NonEmpty, NonEmptyUtil} import com.digitalasset.canton.data.CantonTimestamp import com.daml.metrics.api.MetricHandle.LabeledMetricsFactory import com.digitalasset.canton.resource.DbStorage.SQLActionBuilderChain @@ -218,22 +218,24 @@ final class DbMultiDomainAcsStore[TXE]( companionClass: ContractCompanion[C, TCid, T], traceContext: TraceContext, ): Future[Seq[ContractWithState[TCid, T]]] = { - if (ids.isEmpty) Future.successful(Seq.empty) - else { - waitUntilAcsIngested { - storage - .query( // index: acs_store_template_sid_mid_cid - selectFromAcsTableWithState( - acsTableName, - acsStoreId, - domainMigrationId, - companion, - additionalWhere = (sql"and " ++ inClause("acs.contract_id", ids)).toActionBuilder, - ), - "lookupContractsById", - ) - .map(result => result.map(contractWithStateFromRow(companion)(_))) - } + NonEmpty.from(ids) match { + case None => Future.successful(Seq.empty) + case Some(ids) => + waitUntilAcsIngested { + storage + .query( // index: acs_store_template_sid_mid_cid + selectFromAcsTableWithState( + acsTableName, + acsStoreId, + domainMigrationId, + companion, + additionalWhere = + (sql"and " ++ DbStorage.toInClause("acs.contract_id", ids)).toActionBuilder, + ), + "lookupContractsById", + ) + .map(result => result.map(contractWithStateFromRow(companion)(_))) + } } } @@ -287,25 +289,26 @@ final class DbMultiDomainAcsStore[TXE]( def containsArchived(ids: Seq[ContractId[?]])(implicit traceContext: TraceContext ): Future[Boolean] = waitUntilAcsIngested { - if (ids.isEmpty) Future.successful(false) - else { - val expectedCount = ids.size - storage - .query( - (sql""" - select count(1) - from #$acsTableName acs - where acs.store_id = $acsStoreId - and acs.migration_id = $domainMigrationId - and """ ++ inClause("acs.contract_id", ids) ++ sql""" - """).toActionBuilder - .as[Int] - .head, - "containsArchived", - ) - .map { count => - count != expectedCount - } + NonEmpty.from(ids) match { + case None => Future.successful(false) + case Some(ids) => + val expectedCount = ids.size + storage + .query( + (sql""" + select count(1) + from #$acsTableName acs + where acs.store_id = $acsStoreId + and acs.migration_id = $domainMigrationId + and """ ++ DbStorage.toInClause("acs.contract_id", ids) ++ sql""" + """).toActionBuilder + .as[Int] + .head, + "containsArchived", + ) + .map { count => + count != expectedCount + } } } @@ -1725,18 +1728,21 @@ final class DbMultiDomainAcsStore[TXE]( private def checkIncompleteReassignments( contractIds: Seq[String] ): DBIOAction[Set[String], NoStream, Effect.Read] = { - if (contractIds.isEmpty) DBIO.successful(Set.empty) - else { - DBIO - .sequence(contractIds.grouped(ingestionConfig.maxLookupsPerStatement).map { contractIds => - (sql""" - select distinct contract_id from incomplete_reassignments - where store_id = $acsStoreId and migration_id = $domainMigrationId and """ ++ inClause( - "contract_id", - contractIds.map(lengthLimited), - )).toActionBuilder.as[String].map(_.toSet) - }) - .map(_.foldLeft(Set.empty[String])(_ ++ _)) + NonEmpty.from(contractIds) match { + case None => DBIO.successful(Set.empty) + case Some(contractIds) => + DBIO + .sequence(contractIds.grouped(ingestionConfig.maxLookupsPerStatement).map { + contractIds => + (sql""" + select distinct contract_id from incomplete_reassignments + where store_id = $acsStoreId and migration_id = $domainMigrationId and """ ++ DbStorage + .toInClause( + "contract_id", + NonEmptyUtil.fromUnsafe(contractIds.map(lengthLimited)), + )).toActionBuilder.as[String].map(_.toSet) + }) + .map(_.foldLeft(Set.empty[String])(_ ++ _)) } } @@ -1929,53 +1935,54 @@ final class DbMultiDomainAcsStore[TXE]( } private def doDeleteContracts(deletes: Seq[Delete], summary: MutableIngestionSummary) = { - if (deletes.isEmpty) DBIO.successful(()) - else { - DBIO.sequence(deletes.grouped(ingestionConfig.maxDeletesPerStatement).map { deletes => - val performDeleteSql = acsArchiveConfigOpt match { - case Some(AcsArchiveConfig(archiveTableName, baseColumns)) => - val valuesPairs = deletes.map { d => - val cid = lengthLimited(d.evt.getContractId) - val archivedAt = CantonTimestamp.assertFromInstant(d.recordTime).toMicros - sql"($cid, $archivedAt)" - } - val valuesClause = sqlCommaSeparated(valuesPairs) - (sql""" - WITH deleted AS ( - DELETE FROM #$acsTableName - USING (VALUES """ ++ valuesClause ++ sql""") AS at(cid, archived_at) - WHERE store_id = $acsStoreId - AND migration_id = $domainMigrationId - AND #$acsTableName.contract_id = at.cid - RETURNING #$baseColumns, at.archived_at - ) - INSERT INTO #$archiveTableName (#$baseColumns, archived_at) - SELECT * FROM deleted - RETURNING contract_id - """).toActionBuilder.as[String] - case None => - val contractIds = deletes.map(d => lengthLimited(d.evt.getContractId)) - (sql"""DELETE FROM #$acsTableName - WHERE store_id = $acsStoreId - AND migration_id = $domainMigrationId - AND """ ++ inClause( - "contract_id", - contractIds, - ) ++ sql" RETURNING contract_id").toActionBuilder - .as[String] - } + NonEmpty.from(deletes) match { + case None => DBIO.successful(()) + case Some(deletes) => + DBIO.sequence(deletes.grouped(ingestionConfig.maxDeletesPerStatement).map { deletes => + val performDeleteSql = acsArchiveConfigOpt match { + case Some(AcsArchiveConfig(archiveTableName, baseColumns)) => + val valuesPairs = deletes.map { d => + val cid = lengthLimited(d.evt.getContractId) + val archivedAt = CantonTimestamp.assertFromInstant(d.recordTime).toMicros + sql"($cid, $archivedAt)" + } + val valuesClause = sqlCommaSeparated(valuesPairs) + (sql""" + WITH deleted AS ( + DELETE FROM #$acsTableName + USING (VALUES """ ++ valuesClause ++ sql""") AS at(cid, archived_at) + WHERE store_id = $acsStoreId + AND migration_id = $domainMigrationId + AND #$acsTableName.contract_id = at.cid + RETURNING #$baseColumns, at.archived_at + ) + INSERT INTO #$archiveTableName (#$baseColumns, archived_at) + SELECT * FROM deleted + RETURNING contract_id + """).toActionBuilder.as[String] + case None => + val contractIds = deletes.map(d => lengthLimited(d.evt.getContractId)) + (sql"""DELETE FROM #$acsTableName + WHERE store_id = $acsStoreId + AND migration_id = $domainMigrationId + AND """ ++ DbStorage.toInClause( + "contract_id", + NonEmptyUtil.fromUnsafe(contractIds), + ) ++ sql" RETURNING contract_id").toActionBuilder + .as[String] + } - performDeleteSql.map { deletedCids => - val deletedCidSet = deletedCids.toSet - val ingestedArchivedEvents = - deletes.filter(d => deletedCidSet.contains(d.evt.getContractId)).map(_.evt) - summary.ingestedArchivedEvents.addAll(ingestedArchivedEvents) - // there were no contracts with some id. This can happen because: - // `contractFilter.mightContain` in `getIngestionWork` can return true for a template, - // but that might still satisfy some other filter, so the contract was never inserted - summary.numFilteredArchivedEvents += (deletes.length - deletedCids.size) - } - }) + performDeleteSql.map { deletedCids => + val deletedCidSet = deletedCids.toSet + val ingestedArchivedEvents = + deletes.filter(d => deletedCidSet.contains(d.evt.getContractId)).map(_.evt) + summary.ingestedArchivedEvents.addAll(ingestedArchivedEvents) + // there were no contracts with some id. This can happen because: + // `contractFilter.mightContain` in `getIngestionWork` can return true for a template, + // but that might still satisfy some other filter, so the contract was never inserted + summary.numFilteredArchivedEvents += (deletes.length - deletedCids.size) + } + }) } } diff --git a/apps/common/src/main/scala/org/lfdecentralizedtrust/splice/store/db/Queries.scala b/apps/common/src/main/scala/org/lfdecentralizedtrust/splice/store/db/Queries.scala index 1a55b834e8..5718e6b14d 100644 --- a/apps/common/src/main/scala/org/lfdecentralizedtrust/splice/store/db/Queries.scala +++ b/apps/common/src/main/scala/org/lfdecentralizedtrust/splice/store/db/Queries.scala @@ -26,17 +26,6 @@ trait Queries extends JdbcTypes { .getOrElse(SQLActionBuilderChain(sql"")) } - /* - * TODO(#3900) move to use toInClause when canton fork has it: https://github.com/canton-network/splice/issues/3900 - */ - protected def inClause[V: ClassTag]( - field: String, - seq: Iterable[V], - )(implicit - arraySetParameter: SetParameter[Array[V]] - ): SQLActionBuilder = - sql" #$field = ANY(${seq.toArray[V]})" - protected def notInClause[V: ClassTag]( field: String, seq: Iterable[V], diff --git a/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/AcsSnapshotStore.scala b/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/AcsSnapshotStore.scala index c51128cd09..46fa472de2 100644 --- a/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/AcsSnapshotStore.scala +++ b/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/AcsSnapshotStore.scala @@ -5,6 +5,7 @@ package org.lfdecentralizedtrust.splice.scan.store import cats.data.NonEmptyVector import com.daml.ledger.javaapi.data.CreatedEvent +import com.daml.nonempty.NonEmpty import org.lfdecentralizedtrust.splice.codegen.java.splice.amulet.{Amulet, LockedAmulet} import org.lfdecentralizedtrust.splice.scan.store.AcsSnapshotStore.{ AcsSnapshot, @@ -279,18 +280,18 @@ class AcsSnapshotStore( case None => Future.successful(snapshot.firstRowId) } end = snapshot.lastRowId - partyIdsFilter = partyIds match { - case Nil => + partyIdsFilter = NonEmpty.from(partyIds) match { + case None => // This expression is always true (scan only processes data where the DSO is stakeholder). // It is included to make sure the query plan uses the right index (acs_snapshot_data_all_filters) sql"and stakeholder = ${dsoParty}" - case partyIds => - (sql" and " ++ inClause("stakeholder", partyIds)).toActionBuilder + case Some(partyIds) => + (sql" and " ++ DbStorage.toInClause("stakeholder", partyIds)).toActionBuilder } - templatesFilter = templates match { - case Nil => sql"" - case _ => - (sql" and " ++ inClause( + templatesFilter = NonEmpty.from(templates) match { + case None => sql"" + case Some(templates) => + (sql" and " ++ DbStorage.toInClause( "template_id", templates.map(t => lengthLimited( diff --git a/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/db/DbAppActivityRecordStore.scala b/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/db/DbAppActivityRecordStore.scala index 60a3664e00..8e3582df51 100644 --- a/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/db/DbAppActivityRecordStore.scala +++ b/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/db/DbAppActivityRecordStore.scala @@ -3,6 +3,7 @@ package org.lfdecentralizedtrust.splice.scan.store.db +import com.daml.nonempty.NonEmpty import org.lfdecentralizedtrust.splice.scan.store.AppActivityStore import org.lfdecentralizedtrust.splice.scan.store.AppActivityStore.RoundIngestionStatus import org.lfdecentralizedtrust.splice.store.UpdateHistory @@ -286,22 +287,24 @@ class DbAppActivityRecordStore( def getRecordsByVerdictRowIds( verdictRowIds: Seq[Long] )(implicit tc: TraceContext): Future[Map[Long, AppActivityRecordT]] = { - if (verdictRowIds.isEmpty) Future.successful(Map.empty) - else { - startedIngestingAt.flatMap { - case None => Future.successful(Map.empty) - case Some(_) => - storage - .query( - (sql""" - select verdict_row_id, round_number, app_provider_parties, app_activity_weights - from #${Tables.appActivityRecords} - where history_id = $historyId and """ ++ inClause("verdict_row_id", verdictRowIds)) - .as[AppActivityRecordT], - "appActivity.getRecordsByVerdictRowIds", - ) - .map(rows => rows.map(r => r.verdictRowId -> r).toMap) - } + NonEmpty.from(verdictRowIds) match { + case None => Future.successful(Map.empty) + case Some(verdictRowIds) => + startedIngestingAt.flatMap { + case None => Future.successful(Map.empty) + case Some(_) => + storage + .query( + (sql""" + select verdict_row_id, round_number, app_provider_parties, app_activity_weights + from #${Tables.appActivityRecords} + where history_id = $historyId and """ ++ DbStorage + .toInClause("verdict_row_id", verdictRowIds)) + .as[AppActivityRecordT], + "appActivity.getRecordsByVerdictRowIds", + ) + .map(rows => rows.map(r => r.verdictRowId -> r).toMap) + } } } diff --git a/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/db/DbScanAppRewardsStore.scala b/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/db/DbScanAppRewardsStore.scala index 30f4db856e..66b682a2ca 100644 --- a/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/db/DbScanAppRewardsStore.scala +++ b/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/db/DbScanAppRewardsStore.scala @@ -3,6 +3,7 @@ package org.lfdecentralizedtrust.splice.scan.store.db +import com.daml.nonempty.NonEmpty import org.lfdecentralizedtrust.splice.scan.rewards.{RewardComputationInputs, RewardIssuanceParams} import org.lfdecentralizedtrust.splice.scan.store.ScanAppRewardsStore import org.lfdecentralizedtrust.splice.store.UpdateHistory @@ -615,15 +616,16 @@ class DbScanAppRewardsStore( override def roundsWithComputedRewards(rounds: Seq[Long])(implicit tc: TraceContext ): Future[Set[Long]] = { - if (rounds.isEmpty) Future.successful(Set.empty) - else { - runQuery( - (sql"""select round_number from #${Tables.appRewardRootHashes} - where history_id = $historyId - and """ ++ inClause("round_number", rounds)).toActionBuilder - .as[Long], - "appRewards.roundsWithComputedRewards", - ).map(_.toSet) + NonEmpty.from(rounds) match { + case None => Future.successful(Set.empty) + case Some(rounds) => + runQuery( + (sql"""select round_number from #${Tables.appRewardRootHashes} + where history_id = $historyId + and """ ++ DbStorage.toInClause("round_number", rounds)).toActionBuilder + .as[Long], + "appRewards.roundsWithComputedRewards", + ).map(_.toSet) } } diff --git a/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/db/DbScanStore.scala b/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/db/DbScanStore.scala index 871c2633b6..07bba5a6b9 100644 --- a/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/db/DbScanStore.scala +++ b/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/db/DbScanStore.scala @@ -4,6 +4,7 @@ package org.lfdecentralizedtrust.splice.scan.store.db import com.daml.ledger.javaapi.data.codegen.ContractId +import com.daml.nonempty.NonEmpty import com.digitalasset.canton.data.CantonTimestamp import com.digitalasset.canton.lifecycle.{ AsyncOrSyncCloseable, @@ -496,24 +497,28 @@ class DbScanStore( override def getValidatorLicenseByValidator(validators: Vector[PartyId])(implicit tc: TraceContext ): Future[Seq[Contract[ValidatorLicense.ContractId, ValidatorLicense]]] = waitUntilAcsIngested { - val validatorPartyIds = inClause("validator", validators) - for { - rows <- storage - .query( - selectFromAcsTable( - ScanTables.acsTableName, - acsStoreId, - domainMigrationId, - ValidatorLicense.COMPANION, - where = validatorPartyIds, - ), - "getValidatorLicenseByValidator", - ) - } yield { - rows - .map( - contractFromRow(ValidatorLicense.COMPANION)(_) - ) + NonEmpty.from(validators) match { + case None => Future.successful(Seq.empty) + case Some(validators) => + val validatorPartyIds = DbStorage.toInClause("validator", validators) + for { + rows <- storage + .query( + selectFromAcsTable( + ScanTables.acsTableName, + acsStoreId, + domainMigrationId, + ValidatorLicense.COMPANION, + where = validatorPartyIds, + ), + "getValidatorLicenseByValidator", + ) + } yield { + rows + .map( + contractFromRow(ValidatorLicense.COMPANION)(_) + ) + } } } @@ -665,22 +670,26 @@ class DbScanStore( trackingCids: Seq[VoteRequest.ContractId], limit: Limit, )(implicit tc: TraceContext): Future[Seq[Contract[VoteRequest.ContractId, VoteRequest]]] = { - for { - result <- storage - .query( - listVoteRequestsByTrackingCidQuery( - acsTableName = ScanTables.acsTableName, - acsStoreId = acsStoreId, - domainMigrationId = domainMigrationId, - trackingCidColumnName = "vote_request_tracking_cid", - trackingCids = trackingCids, - limit = limit, - ), - "listVoteRequestsByTrackingCid", - ) - records = applyLimit("listVoteRequestsByTrackingCid", limit, result) - } yield records - .map(contractFromRow(VoteRequest.COMPANION)(_)) + NonEmpty.from(trackingCids) match { + case None => Future.successful(Seq.empty) + case Some(trackingCids) => + for { + result <- storage + .query( + listVoteRequestsByTrackingCidQuery( + acsTableName = ScanTables.acsTableName, + acsStoreId = acsStoreId, + domainMigrationId = domainMigrationId, + trackingCidColumnName = "vote_request_tracking_cid", + trackingCids = trackingCids, + limit = limit, + ), + "listVoteRequestsByTrackingCid", + ) + records = applyLimit("listVoteRequestsByTrackingCid", limit, result) + } yield records + .map(contractFromRow(VoteRequest.COMPANION)(_)) + } } override def lookupVoteRequest(voteRequestCid: VoteRequest.ContractId)(implicit diff --git a/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/db/DbScanVerdictStore.scala b/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/db/DbScanVerdictStore.scala index bd565c986e..4cf0bdd6b9 100644 --- a/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/db/DbScanVerdictStore.scala +++ b/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/db/DbScanVerdictStore.scala @@ -3,6 +3,7 @@ package org.lfdecentralizedtrust.splice.scan.store.db +import com.daml.nonempty.NonEmpty import org.lfdecentralizedtrust.splice.util.FutureUnlessShutdownUtil.futureUnlessShutdownToFuture import com.digitalasset.canton.sequencer.admin.{v30 as seqv30} import com.digitalasset.canton.data.CantonTimestamp @@ -410,41 +411,45 @@ class DbScanVerdictStore( def insertVerdictAndTransactionViewsDBIO( items: Seq[(VerdictT, Long => Seq[TransactionViewT])] )(implicit tc: TraceContext): DBIO[Map[CantonTimestamp, Long]] = { - if (items.isEmpty) DBIO.successful(Map.empty) - else { - val checkExist = (sql""" - select update_id - from #${Tables.verdicts} - where history_id = $historyId - and """ ++ inClause("update_id", items.map(t => lengthLimited(t._1.updateId)))) - .as[String] - - for { - alreadyExisting <- checkExist.map(_.toSet) - nonExisting = items.filter(item => !alreadyExisting.contains(item._1.updateId)) - _ = logger.info( - s"Already ingested verdicts: $alreadyExisting. Non-existing: ${nonExisting.map(_._1.updateId)}." - ) - rowIdMap <- - if (nonExisting.nonEmpty) { - DBIO - .sequence(nonExisting.map { case (verdict, mkViews) => - for { - idOpt <- sqlInsertVerdictReturningId(verdict) - rowId <- idOpt match { - case Some(id) => DBIO.successful(id) - case None => - DBIO.failed(new RuntimeException("insertVerdict did not return row_id")) - } - views = mkViews(rowId) - _ <- DBIO.sequence(views.map(sqlInsertView)).map(_ => ()) - } yield verdict.recordTime -> rowId - }) - .map(_.toMap) - } else { - DBIO.successful(Map.empty[CantonTimestamp, Long]) - } - } yield rowIdMap + NonEmpty.from(items) match { + case None => DBIO.successful(Map.empty) + case Some(items) => + val checkExist = (sql""" + select update_id + from #${Tables.verdicts} + where history_id = $historyId + and """ ++ DbStorage.toInClause( + "update_id", + items.map(t => lengthLimited(t._1.updateId)), + )) + .as[String] + + for { + alreadyExisting <- checkExist.map(_.toSet) + nonExisting = items.filter(item => !alreadyExisting.contains(item._1.updateId)) + _ = logger.info( + s"Already ingested verdicts: $alreadyExisting. Non-existing: ${nonExisting.map(_._1.updateId)}." + ) + rowIdMap <- + if (nonExisting.nonEmpty) { + DBIO + .sequence(nonExisting.map { case (verdict, mkViews) => + for { + idOpt <- sqlInsertVerdictReturningId(verdict) + rowId <- idOpt match { + case Some(id) => DBIO.successful(id) + case None => + DBIO.failed(new RuntimeException("insertVerdict did not return row_id")) + } + views = mkViews(rowId) + _ <- DBIO.sequence(views.map(sqlInsertView)).map(_ => ()) + } yield verdict.recordTime -> rowId + }) + .map(_.toMap) + } else { + DBIO.successful(Map.empty[CantonTimestamp, Long]) + } + } yield rowIdMap } } diff --git a/apps/sv/src/main/scala/org/lfdecentralizedtrust/splice/sv/store/db/DbSvDsoStore.scala b/apps/sv/src/main/scala/org/lfdecentralizedtrust/splice/sv/store/db/DbSvDsoStore.scala index 301d1d387f..0c1d155aaa 100644 --- a/apps/sv/src/main/scala/org/lfdecentralizedtrust/splice/sv/store/db/DbSvDsoStore.scala +++ b/apps/sv/src/main/scala/org/lfdecentralizedtrust/splice/sv/store/db/DbSvDsoStore.scala @@ -7,6 +7,7 @@ import cats.data.OptionT import cats.implicits.* import com.daml.ledger.javaapi.data as javab import com.daml.ledger.javaapi.data.codegen.ContractId +import com.daml.nonempty.NonEmpty import org.lfdecentralizedtrust.splice.automation.MultiDomainExpiredContractTrigger.ListExpiredContracts import org.lfdecentralizedtrust.splice.codegen.java.splice import org.lfdecentralizedtrust.splice.codegen.java.splice.amulet.* @@ -923,48 +924,52 @@ class DbSvDsoStore( ]], ) ] = { - if (rounds.isEmpty) - Future.successful((Seq.empty, Seq.empty)) - else { - val roundsClause = inClause("mining_round", rounds) - val calculateRewardsF = storage - .query( - selectFromAcsTableWithState( - DsoTables.acsTableName, - acsStoreId, - domainMigrationId, - splice.amulet.rewardaccountingv2.CalculateRewardsV2.COMPANION, - additionalWhere = (sql" and " ++ roundsClause).toActionBuilder, - ), - "listDryRunCalculateRewardsV2ByRounds", - ) - .map( - _.map( - assignedContractFromRow(splice.amulet.rewardaccountingv2.CalculateRewardsV2.COMPANION)( - _ - ) - ).filter(_.payload.dryRun) - ) - val processRewardsF = storage - .query( - selectFromAcsTableWithState( - DsoTables.acsTableName, - acsStoreId, - domainMigrationId, - splice.amulet.rewardaccountingv2.ProcessRewardsV2.COMPANION, - additionalWhere = (sql" and " ++ roundsClause).toActionBuilder, - ), - "listDryRunProcessRewardsV2ByRounds", - ) - .map( - _.map( - assignedContractFromRow(splice.amulet.rewardaccountingv2.ProcessRewardsV2.COMPANION)(_) - ).filter(_.payload.dryRun) - ) - for { - calculateRewards <- calculateRewardsF - processRewards <- processRewardsF - } yield (calculateRewards, processRewards) + NonEmpty.from(rounds) match { + case None => Future.successful((Seq.empty, Seq.empty)) + case Some(rounds) => + val roundsClause = DbStorage.toInClause("mining_round", rounds) + val calculateRewardsF = storage + .query( + selectFromAcsTableWithState( + DsoTables.acsTableName, + acsStoreId, + domainMigrationId, + splice.amulet.rewardaccountingv2.CalculateRewardsV2.COMPANION, + additionalWhere = (sql" and " ++ roundsClause).toActionBuilder, + ), + "listDryRunCalculateRewardsV2ByRounds", + ) + .map( + _.map( + assignedContractFromRow( + splice.amulet.rewardaccountingv2.CalculateRewardsV2.COMPANION + )( + _ + ) + ).filter(_.payload.dryRun) + ) + val processRewardsF = storage + .query( + selectFromAcsTableWithState( + DsoTables.acsTableName, + acsStoreId, + domainMigrationId, + splice.amulet.rewardaccountingv2.ProcessRewardsV2.COMPANION, + additionalWhere = (sql" and " ++ roundsClause).toActionBuilder, + ), + "listDryRunProcessRewardsV2ByRounds", + ) + .map( + _.map( + assignedContractFromRow(splice.amulet.rewardaccountingv2.ProcessRewardsV2.COMPANION)( + _ + ) + ).filter(_.payload.dryRun) + ) + for { + calculateRewards <- calculateRewardsF + processRewards <- processRewardsF + } yield (calculateRewards, processRewards) } } @@ -1563,29 +1568,33 @@ class DbSvDsoStore( tc: TraceContext ): Future[Seq[Contract[AmuletPriceVote.ContractId, AmuletPriceVote]]] = waitUntilAcsIngested { import scala.jdk.CollectionConverters.* - for { - dsoRules <- getDsoRules() - voterParties = inClause( - "voter", - dsoRules.payload.svs.asScala - .map { case (party, _) => - lengthLimited(party) - }, - ) - result <- storage - .query( - selectFromAcsTable( - DsoTables.acsTableName, - acsStoreId, - domainMigrationId, - AmuletPriceVote.COMPANION, - where = voterParties, - orderLimit = sql"""limit ${sqlLimit(limit)}""", - ), - "listSvAmuletPriceVotes", - ) - limited = applyLimit("listSvAmuletPriceVotes", limit, result) - } yield limited.map(contractFromRow(AmuletPriceVote.COMPANION)(_)).distinctBy(_.payload.sv) + getDsoRules().flatMap(dsoRules => + NonEmpty.from( + dsoRules.payload.svs.asScala.toMap.map { case (party, _) => lengthLimited(party) } + ) match { + case None => Future.successful(Seq.empty) + case Some(voters) => + for { + dsoRules <- getDsoRules() + voterParties = DbStorage.toInClause("voter", voters) + result <- storage + .query( + selectFromAcsTable( + DsoTables.acsTableName, + acsStoreId, + domainMigrationId, + AmuletPriceVote.COMPANION, + where = voterParties, + orderLimit = sql"""limit ${sqlLimit(limit)}""", + ), + "listSvAmuletPriceVotes", + ) + limited = applyLimit("listSvAmuletPriceVotes", limit, result) + } yield limited + .map(contractFromRow(AmuletPriceVote.COMPANION)(_)) + .distinctBy(_.payload.sv) + } + ) } override protected def lookupSvOnboardingRequestByCandidatePartyWithOffset( @@ -1705,22 +1714,26 @@ class DbSvDsoStore( )(implicit tc: TraceContext ): Future[Seq[Contract[VoteRequest.ContractId, VoteRequest]]] = waitUntilAcsIngested { - for { - result <- storage - .query( - listVoteRequestsByTrackingCidQuery( - acsTableName = DsoTables.acsTableName, - acsStoreId = acsStoreId, - domainMigrationId = domainMigrationId, - trackingCidColumnName = "vote_request_tracking_cid", - trackingCids = trackingCids, - limit = limit, - ), - "listVoteRequestsByTrackingCid", - ) - records = applyLimit("listVoteRequestsByTrackingCid", limit, result) - } yield records - .map(contractFromRow(VoteRequest.COMPANION)(_)) + NonEmpty.from(trackingCids) match { + case None => Future.successful(Seq.empty) + case Some(trackingCids) => + for { + result <- storage + .query( + listVoteRequestsByTrackingCidQuery( + acsTableName = DsoTables.acsTableName, + acsStoreId = acsStoreId, + domainMigrationId = domainMigrationId, + trackingCidColumnName = "vote_request_tracking_cid", + trackingCids = trackingCids, + limit = limit, + ), + "listVoteRequestsByTrackingCid", + ) + records = applyLimit("listVoteRequestsByTrackingCid", limit, result) + } yield records + .map(contractFromRow(VoteRequest.COMPANION)(_)) + } } override def lookupVoteByThisSvAndVoteRequestWithOffset(voteRequestCid: VoteRequest.ContractId)( @@ -1959,27 +1972,27 @@ class DbSvDsoStore( )(implicit tc: TraceContext): Future[ Seq[Contract[splice.round.ClosedMiningRound.ContractId, splice.round.ClosedMiningRound]] ] = { - if (roundNumbers.isEmpty) - Future.successful(Seq.empty) - else { - val roundNumbersClause = inClause("mining_round", roundNumbers) - waitUntilAcsIngested { - for { - result <- storage - .query( - selectFromAcsTable( - DsoTables.acsTableName, - acsStoreId, - domainMigrationId, - ClosedMiningRound.COMPANION, - where = - (sql"""assigned_domain = $synchronizerId AND """ ++ roundNumbersClause).toActionBuilder, - orderLimit = sql"""limit ${sqlLimit(limit)}""", - ), - "listClosedRounds", - ) - } yield result.map(contractFromRow(ClosedMiningRound.COMPANION)(_)) - } + NonEmpty.from(roundNumbers) match { + case None => Future.successful(Seq.empty) + case Some(roundNumbers) => + val roundNumbersClause = DbStorage.toInClause("mining_round", roundNumbers) + waitUntilAcsIngested { + for { + result <- storage + .query( + selectFromAcsTable( + DsoTables.acsTableName, + acsStoreId, + domainMigrationId, + ClosedMiningRound.COMPANION, + where = + (sql"""assigned_domain = $synchronizerId AND """ ++ roundNumbersClause).toActionBuilder, + orderLimit = sql"""limit ${sqlLimit(limit)}""", + ), + "listClosedRounds", + ) + } yield result.map(contractFromRow(ClosedMiningRound.COMPANION)(_)) + } } }