diff --git a/apps/common/src/main/resources/db/migration/canton-network/postgres/stable/V073__acs_snapshot_table_name.sql b/apps/common/src/main/resources/db/migration/canton-network/postgres/stable/V073__acs_snapshot_table_name.sql deleted file mode 100644 index 5e0aaad57e..0000000000 --- a/apps/common/src/main/resources/db/migration/canton-network/postgres/stable/V073__acs_snapshot_table_name.sql +++ /dev/null @@ -1,14 +0,0 @@ -alter table acs_snapshot - -- the name 'table_name' is not reserved, but it is a PostgreSQL keyword, so we use a different name to avoid confusion - add column data_table_name text default null, - -- these values won't be set anymore - drop constraint acs_snapshot_first_row_id_fkey, - drop constraint acs_snapshot_last_row_id_fkey, - alter column first_row_id drop not null, - alter column last_row_id drop not null, - -- ensure consistency - add constraint legacy_or_per_snapshot check - -- per-snapshot tables - ((first_row_id is null and last_row_id is null and data_table_name is not null) or - -- legacy table - (first_row_id is not null and last_row_id is not null and data_table_name is null)); diff --git a/apps/common/src/main/resources/db/migration/canton-network/postgres/stable/V074__acs_snapshot_table_name.sql b/apps/common/src/main/resources/db/migration/canton-network/postgres/stable/V074__acs_snapshot_table_name.sql new file mode 100644 index 0000000000..200f31d44e --- /dev/null +++ b/apps/common/src/main/resources/db/migration/canton-network/postgres/stable/V074__acs_snapshot_table_name.sql @@ -0,0 +1,91 @@ +alter table acs_snapshot + -- the name 'table_name' is not reserved, but it is a PostgreSQL keyword, so we use a different name to avoid confusion + add column data_table_name text default null, + -- these values won't be set anymore + drop constraint acs_snapshot_first_row_id_fkey, + drop constraint acs_snapshot_last_row_id_fkey, + alter column first_row_id drop not null, + alter column last_row_id drop not null, + -- ensure consistency + add constraint legacy_or_per_snapshot check + -- per-snapshot tables + ((first_row_id is null and last_row_id is null and data_table_name is not null) or + -- legacy table + (first_row_id is not null and last_row_id is not null and data_table_name is null)); + +-- TODO: template ids can be interned already + +-- Same as acs_incremental_snapshot_data_next, +-- but includes ALL update_history_creates data necessary to build a CreatedEvent. +create table acs_incremental_snapshot_data_next_v2 +( + -- In production, we have a separate table for each scan instance so this + -- column will be the same for all rows. However, in tests we can have multiple + -- scan instances writing to the same table, so we need to include it. + snapshot_id bigint not null references acs_incremental_snapshot (snapshot_id), + contract_id text not null, + + -- All the data necessary to reconstruct a created event + create_arguments jsonb not null, + event_id text not null, + record_time bigint not null, + template_id_package_id text not null, + -- template_id = package_name:template_id_module_name:template_id_entity_name + package_name text not null, + template_id_module_name text not null, + template_id_entity_name text not null, + contract_key text null, + created_at bigint not null, + -- stakeholders = array_cat(signatories, observers) + signatories text[] not null, + observers text[] not null, + -- plus the Amulet-specific balance columns currently computed in the working table + unlocked_amulet_balance numeric, + locked_amulet_balance numeric +); + +-- Needed for fast insert/remove by contract id +alter table acs_incremental_snapshot_data_next_v2 + add constraint acs_incremental_snapshot_data_next_v2_pk + primary key (snapshot_id, contract_id); + +-- Needed because ACS snapshots are ordered by creation time +create index acs_incremental_snapshot_data_next_v2_ca_ci + on acs_incremental_snapshot_data_next_v2 (snapshot_id, created_at, contract_id); + +-- Template table for acs_snapshot_creates__. +-- This allows the code to just CREATE TABLE LIKE acs_snapshot_creates_template or acs_snapshot_stakeholders_template. +-- Design decision: we don't have a single table per (contract_id, stakeholder) in order to avoid duplicating the create_arguments. +create table acs_snapshot_creates_template +( + contract_id text primary key, + -- All the data necessary to reconstruct a created event + create_arguments jsonb not null, + event_id text not null, + record_time bigint not null, + template_id_package_id text not null, -- the package_name is already included as part of the template_id + contract_key text null, + created_at bigint not null, + signatories text[] not null, + observers text[] not null, + -- plus the Amulet-specific balance columns currently computed in the working table + unlocked_amulet_balance numeric, + locked_amulet_balance numeric +); + +create table acs_snapshot_stakeholders_template +( + -- Important: insertion order should happen by (created_at, contract_id) for: + -- 1) backwards compatibility + -- 2) deterministic ordering across SVs + -- then we can sort by row_id to preserve that order and support pagination with after: Long + row_id bigint generated by default as identity primary key, + stakeholder text not null, + template_id text not null, + contract_id text not null +); + +-- Necessary indexes: +-- 1) (stakeholder, template_id, row_id) for where stakeholder=? and template_id=? (and row_id > $after) order by row_id +-- 2) (stakeholder, row_id) for where stakeholder=? (and row_id > $after) order by row_id +-- In both cases `include (contract_id)` allows index-only scans to then merge with acs_snapshot_creates_template diff --git a/apps/common/src/main/scala/org/lfdecentralizedtrust/splice/util/QualifiedName.scala b/apps/common/src/main/scala/org/lfdecentralizedtrust/splice/util/QualifiedName.scala index 962bf45ce4..fc7622f247 100644 --- a/apps/common/src/main/scala/org/lfdecentralizedtrust/splice/util/QualifiedName.scala +++ b/apps/common/src/main/scala/org/lfdecentralizedtrust/splice/util/QualifiedName.scala @@ -45,6 +45,14 @@ object PackageQualifiedName { QualifiedName(companion.TEMPLATE_ID.getModuleName, companion.TEMPLATE_ID.getEntityName), ) } + + def assertFromString(s: String): PackageQualifiedName = { + val segments = s.split(":") + if (segments.length != 3) { + throw new IllegalArgumentException(s"Expect qualified name with two identifiers but got $s") + } + PackageQualifiedName(segments(0), QualifiedName(segments(1), segments(2))) + } } object QualifiedName { diff --git a/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/automation/AcsSnapshotTrigger.scala b/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/automation/AcsSnapshotTrigger.scala index 31ed151b0f..124046d3d9 100644 --- a/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/automation/AcsSnapshotTrigger.scala +++ b/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/automation/AcsSnapshotTrigger.scala @@ -35,7 +35,11 @@ class AcsSnapshotTrigger( ) extends AcsSnapshotTriggerBase(store, updateHistory, context) { override val snapshotTable: IncrementalAcsSnapshotTable = - AcsSnapshotStore.IncrementalAcsSnapshotTable.Next + if (storageConfig.perAcsSnapshotTablesEnabled) { + AcsSnapshotStore.IncrementalAcsSnapshotTable.NextV2 + } else { + AcsSnapshotStore.IncrementalAcsSnapshotTable.Next + } override val snapshotMetrics: AcsSnapshotsMetrics = new HistoryMetrics(context.metricsFactory)( MetricsContext.Empty diff --git a/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/config/ScanStorageConfig.scala b/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/config/ScanStorageConfig.scala index 1655bc8eee..c492630bed 100644 --- a/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/config/ScanStorageConfig.scala +++ b/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/config/ScanStorageConfig.scala @@ -23,6 +23,7 @@ import scala.util.matching.Regex case class ScanStorageConfig( dbAcsSnapshotPeriodHours: Int, // Period between two consecutive acs snapshots to be computed and stored in the DB + perAcsSnapshotTablesEnabled: Boolean, // Whether each ACS snapshot should be stored in its own table bulkAcsSnapshotPeriodHours: Int, // Period between two consecutive acs snapshots to be dumped to bulk storage (currently must be <=24 hr, and a multiple of dbAcsSnapshotPeriodHours) bulkDbReadChunkSize: Int, // Chunk size to read from the DB for copying to bulk storage bulkZstdFrameSize: Long, // Size of each zstd frame. In prod, must be >= 5 MB as each frame is written as a part in multi-part upload, which are enforced by most s3 implementations to be >= 5MB each @@ -162,6 +163,7 @@ object ScanStorageConfig { object ScanStorageConfigs { val scanStorageConfigV1 = ScanStorageConfig( dbAcsSnapshotPeriodHours = 3, + perAcsSnapshotTablesEnabled = false, bulkAcsSnapshotPeriodHours = 24, bulkDbReadChunkSize = 1000, bulkZstdFrameSize = 12L * 1024 * 1024, 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 b19d1f64d4..086643e617 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 @@ -4,7 +4,7 @@ package org.lfdecentralizedtrust.splice.scan.store import cats.data.NonEmptyVector -import com.daml.ledger.javaapi.data.CreatedEvent +import com.daml.ledger.javaapi.data.{CreatedEvent, Identifier} import org.lfdecentralizedtrust.splice.codegen.java.splice.amulet.{Amulet, LockedAmulet} import org.lfdecentralizedtrust.splice.scan.store.AcsSnapshotStore.{ AcsSnapshot, @@ -18,12 +18,7 @@ import org.lfdecentralizedtrust.splice.scan.store.AcsSnapshotStore.{ } import org.lfdecentralizedtrust.splice.store.UpdateHistory.SelectFromCreateEvents import org.lfdecentralizedtrust.splice.store.{HardLimit, Limit, LimitHelpers, UpdateHistory} -import org.lfdecentralizedtrust.splice.store.db.{ - AcsJdbcTypes, - AcsQueries, - AdvisoryLocks, - AdvisoryLockIds, -} +import org.lfdecentralizedtrust.splice.store.db.{AdvisoryLocks} import org.lfdecentralizedtrust.splice.util.{Contract, HoldingsSummary, PackageQualifiedName} import com.digitalasset.canton.data.CantonTimestamp import com.digitalasset.canton.lifecycle.{CloseContext, FutureUnlessShutdown} @@ -31,17 +26,22 @@ import com.digitalasset.canton.logging.pretty.{Pretty, PrettyPrinting} import com.digitalasset.canton.logging.{NamedLoggerFactory, NamedLogging} import com.digitalasset.canton.resource.DbStorage import com.digitalasset.canton.resource.DbStorage.Implicits.BuilderChain.toSQLActionBuilderChain +import com.digitalasset.canton.resource.DbStorage.SQLActionBuilderChain import com.digitalasset.canton.topology.PartyId import com.digitalasset.canton.tracing.TraceContext -import org.lfdecentralizedtrust.splice.scan.store.AcsSnapshotStore.QueryParts.* +import com.google.protobuf.ByteString +import org.lfdecentralizedtrust.splice.scan.store.AcsSnapshotStore.* +import org.lfdecentralizedtrust.splice.store.db.{AcsJdbcTypes, AcsQueries, AdvisoryLockIds} import org.lfdecentralizedtrust.splice.store.events.SpliceCreatedEvent -import slick.dbio.{DBIOAction, Effect, NoStream} +import org.lfdecentralizedtrust.splice.util.{EventId, ValueJsonCodecProtobuf as ProtobufCodec} +import slick.dbio.{DBIO, DBIOAction, Effect, NoStream} import slick.jdbc.canton.ActionBasedSQLInterpolation.Implicits.actionBasedSQLInterpolationCanton -import slick.jdbc.canton.SQLActionBuilder import slick.jdbc.{GetResult, JdbcProfile} import java.util.concurrent.Semaphore import scala.concurrent.{ExecutionContext, Future} +import scala.jdk.CollectionConverters.* +import scala.jdk.OptionConverters.* class AcsSnapshotStore( storage: DbStorage, @@ -129,7 +129,9 @@ class AcsSnapshotStore( case Some(LegacyAcsSnapshot(_, _, _, firstRowId, lastRowId, _, _)) => sql"where snapshot.row_id >= $firstRowId and snapshot.row_id <= $lastRowId" case Some(_: PerTableAcsSnapshot) => - throw io.grpc.Status.UNIMPLEMENTED.withDescription("TODO #6264").asRuntimeException() + throw io.grpc.Status.UNIMPLEMENTED + .withDescription("This should not be called if we have PerTableAcsSnapshot enabled.") + .asRuntimeException() case None => sql"where false" } @@ -241,10 +243,22 @@ class AcsSnapshotStore( sqlu"""delete from acs_snapshot where snapshot_record_time = ${snapshot.snapshotRecordTime}""", sqlu"""delete from acs_snapshot_data where row_id between ${snapshot.firstRowId} and ${snapshot.lastRowId}""", ) - case _: PerTableAcsSnapshot => - throw io.grpc.Status.UNIMPLEMENTED.withDescription("TODO #6263").asRuntimeException() + case table: PerTableAcsSnapshot => + DBIOAction.seq( + sqlu"""delete from acs_snapshot where snapshot_record_time = ${snapshot.snapshotRecordTime}""", + AdvisoryLocks.withDdlLock(sqlu"""drop table #${AcsTableDDL.acsSnapshotCreatesTableName( + historyId, + table.snapshotRecordTime, + )};"""), + AdvisoryLocks.withDdlLock( + sqlu"""drop table #${AcsTableDDL.acsSnapshotStakeholdersTableName( + historyId, + table.snapshotRecordTime, + )};""" + ), + ) } - storage.update(statement.transactionally, "deleteSnapshot") + storage.queryAndUpdate(statement.transactionally, "deleteSnapshot") } def queryAcsSnapshot( @@ -256,7 +270,7 @@ class AcsSnapshotStore( templates: Seq[PackageQualifiedName], )(implicit tc: TraceContext): Future[QueryAcsSnapshotResult] = { for { - snapshotTrait <- storage + snapshot <- storage .querySingle( sql"""select snapshot_record_time, migration_id, history_id, first_row_id, last_row_id, unlocked_amulet_balance, locked_amulet_balance, data_table_name from acs_snapshot @@ -275,13 +289,135 @@ class AcsSnapshotStore( .asRuntimeException() ) ) - snapshot <- (snapshotTrait match { - case snapshot: LegacyAcsSnapshot => Future.successful(snapshot) - case _: PerTableAcsSnapshot => - Future.failed( - io.grpc.Status.UNIMPLEMENTED.withDescription("TODO #6264").asRuntimeException() + events <- snapshot match { + case snapshot: LegacyAcsSnapshot => + queryLegacyTable(snapshot, after, limit, partyIds, templates) + case ownTable: PerTableAcsSnapshot => + querySnapshotInOwnTable(ownTable, after, limit, partyIds, templates) + } + } yield { + val eventsInPage = + applyLimitOrFail("queryAcsSnapshot", limit, events.map(_._2)) + val afterToken = if (eventsInPage.size == limit.limit) events.lastOption.map(_._1) else None + QueryAcsSnapshotResult( + migrationId = snapshot.migrationId, + snapshotRecordTime = snapshot.snapshotRecordTime, + createdEventsInPage = eventsInPage, + afterToken = afterToken, + ) + } + } + + private def querySnapshotInOwnTable( + snapshot: PerTableAcsSnapshot, + after: Option[Long], + limit: Limit, + partyIds: Seq[PartyId], + templates: Seq[PackageQualifiedName], + )(implicit tc: TraceContext): Future[Vector[(Long, SpliceCreatedEvent)]] = { + val createsTableName = + AcsTableDDL.acsSnapshotCreatesTableName(historyId, snapshot.snapshotRecordTime) + val stakeholdersTableName = + AcsTableDDL.acsSnapshotStakeholdersTableName(historyId, snapshot.snapshotRecordTime) + val afterFilter = after.fold(sql"")(after => sql" and s.row_id > $after") + storage + .query( + (sql""" + with contracts as ( + select distinct on (contract_id) contract_id, row_id, template_id + from #$stakeholdersTableName + where """ ++ stakeholdersFilter(partyIds) ++ + templatesFilter(templates) ++ + afterFilter ++ sql""" + order by contract_id + limit ${sqlLimit(limit)} + ) + select + s.row_id, + event_id, + record_time, + template_id_package_id, + template_id, + s.contract_id, + create_arguments, + contract_key, + signatories, + observers, + created_at + from contracts s + join #$createsTableName c on s.contract_id = c.contract_id + -- This will only sort over LIMIT rows, which is acceptable + order by s.row_id + """).toActionBuilder.as[ + ( + Long, + String, + CantonTimestamp, + String, + String, + String, + String, + Option[String], + Seq[String], + Seq[String], + CantonTimestamp, + ) + ], + "querySnapshotInOwnTable", + ) + .map(_.map { + case ( + rowId, + eventId, + recordTime, + packageId, + rawTemplateIdPackageQualifiedName, + contractId, + createArguments, + contractKey, + signatories, + observers, + createdAt, + ) => + val templateIdPackageQualifiedName = + PackageQualifiedName.assertFromString(rawTemplateIdPackageQualifiedName) + rowId -> SpliceCreatedEvent( + eventId = eventId, + recordTime = recordTime, + new CreatedEvent( + /*witnessParties = */ java.util.Collections.emptyList(), + /*offset = */ 0, // not populated + /*nodeId = */ EventId.nodeIdFromEventId(eventId), + /*templateId = */ new Identifier( + packageId, + templateIdPackageQualifiedName.qualifiedName.moduleName, + templateIdPackageQualifiedName.qualifiedName.entityName, + ), + /* packageName = */ templateIdPackageQualifiedName.packageName, + /*contractId = */ contractId, + /*arguments = */ ProtobufCodec.deserializeValue(createArguments).asRecord().get(), + /*createdEventBlob = */ ByteString.EMPTY, + /*interfaceViews = */ java.util.Collections.emptyMap(), + /*failedInterfaceViews = */ java.util.Collections.emptyMap(), + /*contractKey = */ contractKey.map(ProtobufCodec.deserializeValue).toJava, + /*signatories = */ signatories.asJava, + /*observers = */ observers.asJava, + /*createdAt = */ createdAt.toInstant, + /*acsDelta = */ false, + /*representativePackageId = */ packageId, + ), ) }) + } + + private def queryLegacyTable( + snapshot: LegacyAcsSnapshot, + after: Option[Long], + limit: Limit, + partyIds: Seq[PartyId], + templates: Seq[PackageQualifiedName], + )(implicit tc: TraceContext): Future[Vector[(Long, SpliceCreatedEvent)]] = { + for { begin <- after match { case Some(value) if value < snapshot.firstRowId || value > snapshot.lastRowId => Future.failed( @@ -295,34 +431,14 @@ class AcsSnapshotStore( case None => Future.successful(snapshot.firstRowId) } end = snapshot.lastRowId - partyIdsFilter = partyIds match { - case Nil => - // 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 - } - templatesFilter = templates match { - case Nil => sql"" - case _ => - (sql" and " ++ inClause( - "template_id", - templates.map(t => - lengthLimited( - s"${t.packageName}:${t.qualifiedName.moduleName}:${t.qualifiedName.entityName}" - ) - ), - )).toActionBuilder - } events <- storage .query( (sql""" with snapshot as ( select create_id, max(row_id) as row_id from acs_snapshot_data - where row_id between $begin and $end - """ ++ partyIdsFilter ++ templatesFilter ++ sql""" + where row_id between $begin and $end and + """ ++ stakeholdersFilter(partyIds) ++ templatesFilter(templates) ++ sql""" group by create_id order by row_id asc -- this CTE already will contain all snapshot rows (filtered by party id and template, if necessary). @@ -353,17 +469,29 @@ class AcsSnapshotStore( .as[(Long, SelectFromCreateEvents)], "queryAcsSnapshot.getCreatedEvents", ) - } yield { - val eventsInPage = - applyLimitOrFail("queryAcsSnapshot", limit, events.map(_._2.toCreatedEvent)) - val afterToken = if (eventsInPage.size == limit.limit) events.lastOption.map(_._1) else None - QueryAcsSnapshotResult( - migrationId = migrationId, - snapshotRecordTime = snapshot.snapshotRecordTime, - createdEventsInPage = eventsInPage, - afterToken = afterToken, - ) - } + } yield events.map { case (rowId, select) => rowId -> select.toCreatedEvent } + } + + private def stakeholdersFilter(partyIds: Seq[PartyId]) = partyIds match { + case Nil => + // 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" stakeholder = $dsoParty" + case partyIds => + inClause("stakeholder", partyIds) + } + + private def templatesFilter(templates: Seq[PackageQualifiedName]) = templates match { + case Nil => sql"" + case _ => + (sql" and " ++ inClause( + "template_id", + templates.map(t => + lengthLimited( + s"${t.packageName}:${t.qualifiedName.moduleName}:${t.qualifiedName.entityName}" + ) + ), + )).toActionBuilder } def getHoldingsState( @@ -510,7 +638,13 @@ class AcsSnapshotStore( val initializeFrom = initializeFromT match { case legacy: LegacyAcsSnapshot => legacy case _: PerTableAcsSnapshot => - throw io.grpc.Status.UNIMPLEMENTED.withDescription("TODO #6263").asRuntimeException() + // If we enable PerTableAcsSnapshots, we will necessarily initialize from a LegacyAcsSnapshot, + // never from a PerTableAcsSnapshot. Then initializeIncrementalSnapshot will never be called again. + throw io.grpc.Status.FAILED_PRECONDITION + .withDescription( + "BUG: This shouldn't be called: we shouldn't be initializing from a PerTableAcsSnapshot." + ) + .asRuntimeException() } assert(targetRecordTime.isAfter(initializeFrom.snapshotRecordTime)) val statement = for { @@ -533,11 +667,11 @@ class AcsSnapshotStore( """.as[Long].head insertedRows <- (sql""" insert into #${table.tableName} ( - """ ++ copyFromUpdateHistoryTargetColumns ++ sql""", + """ ++ table.copyFromUpdateHistoryTargetColumns ++ sql""", snapshot_id ) select - """ ++ copyFromUpdateHistorySourceColumns ++ sql""", + """ ++ table.copyFromUpdateHistorySourceColumns ++ sql""", $snapshotId from acs_snapshot_data d join update_history_creates c on d.create_id=c.row_id @@ -601,11 +735,11 @@ class AcsSnapshotStore( insertedRows <- (sql""" insert into #${table.tableName} ( - """ ++ copyFromUpdateHistoryTargetColumns ++ sql""", + """ ++ table.copyFromUpdateHistoryTargetColumns ++ sql""", snapshot_id ) select - """ ++ copyFromUpdateHistorySourceColumns ++ sql""", + """ ++ table.copyFromUpdateHistorySourceColumns ++ sql""", $snapshotId from update_history_creates c where history_id = $historyId @@ -645,7 +779,111 @@ class AcsSnapshotStore( assert(snapshot.tableName == table.tableName) assert(snapshot.historyId == historyId) assert(snapshot.recordTime == snapshot.targetRecordTime) - val statement = for { + val statement: DBIO[Int] = table match { + case IncrementalAcsSnapshotTable.NextV2 => + saveV2IncrementalSnapshot(table, snapshot, nextSnapshotTargetRecordTime)(tc) + case IncrementalAcsSnapshotTable.Next | IncrementalAcsSnapshotTable.Backfill => + saveLegacyIncrementalSnapshotStatement(table, snapshot, nextSnapshotTargetRecordTime) + } + storage.queryAndUpdate( + withExclusiveSnapshotDataLock( + withIncrementalSnapshotIdempotencyCheck( + table, + statement, + Some(snapshot), + ) + ), + "saveIncrementalSnapshot", + ) + } + + private def saveV2IncrementalSnapshot( + table: IncrementalAcsSnapshotTable, + snapshot: IncrementalAcsSnapshot, + nextSnapshotTargetRecordTime: CantonTimestamp, + )(implicit tc: TraceContext) = { + val createsTableName = + AcsTableDDL.acsSnapshotCreatesTableName(historyId, snapshot.targetRecordTime) + val stakeholdersTableName = + AcsTableDDL.acsSnapshotStakeholdersTableName(historyId, snapshot.targetRecordTime) + + for { + _ <- sqlu"create table #$createsTableName (like acs_snapshot_creates_template including all)" + _ <- + sqlu"create table #$stakeholdersTableName (like acs_snapshot_stakeholders_template including all)" + copiedCreateRows <- (sql""" + insert into #$createsTableName (contract_id, create_arguments, event_id, record_time, template_id_package_id, contract_key, created_at, signatories, observers, unlocked_amulet_balance, locked_amulet_balance) + select s.contract_id, s.create_arguments, s.event_id, s.record_time, s.template_id_package_id, s.contract_key, s.created_at, s.signatories, s.observers, """ ++ IncrementalAcsSnapshotTable.QueryParts + .unlockedAmuletBalance() ++ sql", " ++ IncrementalAcsSnapshotTable.QueryParts + .lockedAmuletBalance() ++ sql""" + from #${table.tableName} s + where s.snapshot_id = ${snapshot.snapshotId} + -- ensure consistent ordering across SVs + order by created_at, contract_id + """).toActionBuilder.asUpdate + copiedStakeholderRows <- sqlu""" + insert into #${stakeholdersTableName} (stakeholder, template_id, contract_id) + select stakeholder, concat(s.package_name, ':', s.template_id_module_name, ':', s.template_id_entity_name) as template_id, contract_id + from #${table.tableName} s + cross join unnest(array_cat(s.observers, s.signatories)) as stakeholder + where s.snapshot_id = ${snapshot.snapshotId} + order by created_at, contract_id + """ + // TODO: we should create the necessary indexes + + (unlocked_amulet_balance, locked_amulet_balance) <- sql""" + select + sum(s.unlocked_amulet_balance) AS unlocked_amulet_balance, + sum(s.locked_amulet_balance) AS locked_amulet_balance + from #${table.tableName} s + where snapshot_id = ${snapshot.snapshotId} + """.as[(BigDecimal, BigDecimal)].head + + _ <- sqlu""" + insert into acs_snapshot ( + snapshot_record_time, + migration_id, + history_id, + first_row_id, + last_row_id, + unlocked_amulet_balance, + locked_amulet_balance, + data_table_name + ) + values ( + ${snapshot.recordTime}, + ${snapshot.migrationId}, + ${snapshot.historyId}, + null, + null, + ${unlocked_amulet_balance}, + ${locked_amulet_balance}, + ${createsTableName} + ) + """ + + _ <- sqlu""" + update acs_incremental_snapshot + set + target_record_time = ${nextSnapshotTargetRecordTime} + where snapshot_id = ${snapshot.snapshotId} + """ + } yield { + logger.debug( + s"Saved incremental snapshot ${snapshot.snapshotId} at ${snapshot.recordTime} with $copiedCreateRows create rows and $copiedStakeholderRows stakeholder rows." + + s" Next snapshot target record time: $nextSnapshotTargetRecordTime" + ) + // This doesn't make much sense anymore + copiedCreateRows + } + } + + private def saveLegacyIncrementalSnapshotStatement( + table: IncrementalAcsSnapshotTable, + snapshot: IncrementalAcsSnapshot, + nextSnapshotTargetRecordTime: CantonTimestamp, + )(implicit tc: TraceContext): DBIOAction[Int, NoStream, Effect.Read & Effect.Write] = { + for { // Note: Only one client can write to acs_snapshot_data at a time, enforced via advisory locks. // We therefore don't need to worry about concurrent writes between getting max_row_id_before and using it. max_row_id_before <- sql""" @@ -713,16 +951,6 @@ class AcsSnapshotStore( ) copied_rows } - storage.queryAndUpdate( - withExclusiveSnapshotDataLock( - withIncrementalSnapshotIdempotencyCheck( - table, - statement, - Some(snapshot), - ) - ), - "saveIncrementalSnapshot", - ) } /** Updates an incremental snapshot to a new record time. @@ -749,11 +977,11 @@ class AcsSnapshotStore( insertedRows <- (sql""" insert into #${table.tableName} ( - """ ++ copyFromUpdateHistoryTargetColumns ++ sql""", + """ ++ table.copyFromUpdateHistoryTargetColumns ++ sql""", snapshot_id ) select - """ ++ copyFromUpdateHistorySourceColumns ++ sql""", + """ ++ table.copyFromUpdateHistorySourceColumns ++ sql""", ${snapshot.snapshotId} from update_history_creates c where history_id = $historyId @@ -809,13 +1037,129 @@ class AcsSnapshotStore( object AcsSnapshotStore { - sealed trait IncrementalAcsSnapshotTable { def tableName: String } + sealed trait IncrementalAcsSnapshotTable { + def tableName: String + def copyFromUpdateHistoryTargetColumns: SQLActionBuilderChain + def copyFromUpdateHistorySourceColumns: SQLActionBuilderChain + } object IncrementalAcsSnapshotTable { + case object NextV2 extends IncrementalAcsSnapshotTable { + val tableName: String = "acs_incremental_snapshot_data_next_v2" + + override def copyFromUpdateHistoryTargetColumns: SQLActionBuilderChain = + QueryParts.v2CopyFromUpdateHistoryTargetColumns + + override def copyFromUpdateHistorySourceColumns: SQLActionBuilderChain = + QueryParts.v2CopyFromUpdateHistorySourceColumns + } case object Next extends IncrementalAcsSnapshotTable { val tableName: String = "acs_incremental_snapshot_data_next" + + override def copyFromUpdateHistoryTargetColumns: SQLActionBuilderChain = + QueryParts.legacyCopyFromUpdateHistoryTargetColumns + + override def copyFromUpdateHistorySourceColumns: SQLActionBuilderChain = + QueryParts.legacyCopyFromUpdateHistorySourceColumns } case object Backfill extends IncrementalAcsSnapshotTable { val tableName: String = "acs_incremental_snapshot_data_backfill" + + override def copyFromUpdateHistoryTargetColumns: SQLActionBuilderChain = + QueryParts.legacyCopyFromUpdateHistoryTargetColumns + + override def copyFromUpdateHistorySourceColumns: SQLActionBuilderChain = + QueryParts.legacyCopyFromUpdateHistorySourceColumns + } + + object QueryParts { + + private[IncrementalAcsSnapshotTable] val v2CopyFromUpdateHistoryTargetColumns + : SQLActionBuilderChain = { + sql""" + create_arguments, + event_id, + record_time, + template_id_package_id, + package_name, + template_id_module_name, + template_id_entity_name, + contract_key, + signatories, + observers, + contract_id, + created_at, + unlocked_amulet_balance, + locked_amulet_balance + """ + } + + private[IncrementalAcsSnapshotTable] val v2CopyFromUpdateHistorySourceColumns + : SQLActionBuilderChain = { + sql""" + c.create_arguments, + c.event_id, + c.record_time, + c.template_id_package_id, + c.package_name, + c.template_id_module_name, + c.template_id_entity_name, + c.contract_key, + c.signatories, + c.observers, + c.contract_id, + c.created_at,""" ++ + unlockedAmuletBalance() ++ sql"," ++ + lockedAmuletBalance() + } + + private[IncrementalAcsSnapshotTable] val legacyCopyFromUpdateHistoryTargetColumns + : SQLActionBuilderChain = { + sql""" + create_id, + template_id, + stakeholders, + contract_id, + created_at, + unlocked_amulet_balance, + locked_amulet_balance""" + } + + private[IncrementalAcsSnapshotTable] val legacyCopyFromUpdateHistorySourceColumns + : SQLActionBuilderChain = { + sql""" + c.row_id, + concat(c.package_name, ':', c.template_id_module_name, ':', c.template_id_entity_name) as template_id, + array_cat(c.signatories, c.observers) as stakeholder, + c.contract_id, + c.created_at, + """ ++ unlockedAmuletBalance() ++ sql"," ++ + lockedAmuletBalance() + } + + def unlockedAmuletBalance() = { + sql""" + case + when package_name = ${Amulet.COMPANION.PACKAGE_NAME} + and template_id_module_name = ${Amulet.COMPANION.TEMPLATE_ID.getModuleName} + and template_id_entity_name = ${Amulet.COMPANION.TEMPLATE_ID.getEntityName} + then (create_arguments->'record'->'fields'->2->'value'->'record'->'fields'->0->'value'->>'numeric')::numeric + else 0 + end + """ + } + + def lockedAmuletBalance() = { + sql""" + case + when package_name = ${LockedAmulet.COMPANION.PACKAGE_NAME} + and template_id_module_name = ${LockedAmulet.COMPANION.TEMPLATE_ID.getModuleName} + and template_id_entity_name = ${LockedAmulet.COMPANION.TEMPLATE_ID.getEntityName} + then (create_arguments->'record'->'fields'->0->'value'->'record'->'fields'->2->'value'->'record'->'fields'->0->'value'->>'numeric')::numeric + else 0 + end + """ + } + } } @@ -856,42 +1200,6 @@ object AcsSnapshotStore { ) } - object QueryParts { - - val copyFromUpdateHistoryTargetColumns: SQLActionBuilder = - sql""" - create_id, - contract_id, - created_at, - unlocked_amulet_balance, - locked_amulet_balance, - template_id, - stakeholders - """ - val copyFromUpdateHistorySourceColumns: SQLActionBuilder = - sql""" - c.row_id, - c.contract_id, - c.created_at, - case - when package_name = ${Amulet.COMPANION.PACKAGE_NAME} - and template_id_module_name = ${Amulet.COMPANION.TEMPLATE_ID.getModuleName} - and template_id_entity_name = ${Amulet.COMPANION.TEMPLATE_ID.getEntityName} - then (c.create_arguments->'record'->'fields'->2->'value'->'record'->'fields'->0->'value'->>'numeric')::numeric - else 0 - end, - case - when package_name = ${LockedAmulet.COMPANION.PACKAGE_NAME} - and template_id_module_name = ${LockedAmulet.COMPANION.TEMPLATE_ID.getModuleName} - and template_id_entity_name = ${LockedAmulet.COMPANION.TEMPLATE_ID.getEntityName} - then (c.create_arguments->'record'->'fields'->0->'value'->'record'->'fields'->2->'value'->'record'->'fields'->0->'value'->>'numeric')::numeric - else 0 - end, - concat(c.package_name, ':', c.template_id_module_name, ':', c.template_id_entity_name) as template_id, - array_cat(c.signatories, c.observers) - """ - } - sealed trait AcsSnapshot extends PrettyPrinting { val snapshotRecordTime: CantonTimestamp val migrationId: Long @@ -1039,6 +1347,14 @@ object AcsSnapshotStore { }) } + object AcsTableDDL { + def acsSnapshotCreatesTableName(historyId: Long, snapshotRecordTime: CantonTimestamp) = + s"acs_snapshot_creates_${historyId}_${snapshotRecordTime.toEpochMilli}" + + def acsSnapshotStakeholdersTableName(historyId: Long, snapshotRecordTime: CantonTimestamp) = + s"acs_snapshot_stakeholders_${historyId}_${snapshotRecordTime.toEpochMilli}" + } + def apply( storage: DbStorage, updateHistory: UpdateHistory, diff --git a/apps/scan/src/test/scala/org/lfdecentralizedtrust/splice/scan/automation/AcsSnapshotTriggerTest.scala b/apps/scan/src/test/scala/org/lfdecentralizedtrust/splice/scan/automation/AcsSnapshotTriggerTest.scala index e256b7a9b9..d27668ac7c 100644 --- a/apps/scan/src/test/scala/org/lfdecentralizedtrust/splice/scan/automation/AcsSnapshotTriggerTest.scala +++ b/apps/scan/src/test/scala/org/lfdecentralizedtrust/splice/scan/automation/AcsSnapshotTriggerTest.scala @@ -14,12 +14,14 @@ import org.scalatest.wordspec.AnyWordSpec import scala.concurrent.Future -class AcsSnapshotTriggerTest +trait AcsSnapshotTriggerTest extends AnyWordSpec with BaseTest with HasExecutionContext with HasActorSystem { + protected def storageConfig: ScanStorageConfig + "AcsSnapshotTrigger" should { "initialize from an existing snapshot" in { @@ -690,15 +692,6 @@ class AcsSnapshotTriggerTest } } - private def storageConfig = ScanStorageConfig( - dbAcsSnapshotPeriodHours = 1, - bulkAcsSnapshotPeriodHours = 1, // ignored in this test - bulkDbReadChunkSize = 1, // ignored in this test - bulkZstdFrameSize = 0L, // ignored in this test - bulkMaxFileSize = 0L, // ignored in this test - zstdCompressionLevel = 0, // ignored in this test - ) - private def unused0[T]: () => Future[T] = () => fail("This argument should not be used") private def unused1[T]: Long => Future[T] = _ => fail("This argument should not be used") @@ -748,3 +741,27 @@ class AcsSnapshotTriggerTest private def migration3 = 3L private def migration4 = 4L } + +class LegacyAcsSnapshotTriggerTest extends AcsSnapshotTriggerTest { + protected def storageConfig = ScanStorageConfig( + dbAcsSnapshotPeriodHours = 1, + perAcsSnapshotTablesEnabled = false, + bulkAcsSnapshotPeriodHours = 1, // ignored in this test + bulkDbReadChunkSize = 1, // ignored in this test + bulkZstdFrameSize = 0L, // ignored in this test + bulkMaxFileSize = 0L, // ignored in this test + zstdCompressionLevel = 0, // ignored in this test + ) +} + +class V2AcsSnapshotTriggerTest extends AcsSnapshotTriggerTest { + protected def storageConfig = ScanStorageConfig( + dbAcsSnapshotPeriodHours = 1, + perAcsSnapshotTablesEnabled = true, + bulkAcsSnapshotPeriodHours = 1, // ignored in this test + bulkDbReadChunkSize = 1, // ignored in this test + bulkZstdFrameSize = 0L, // ignored in this test + bulkMaxFileSize = 0L, // ignored in this test + zstdCompressionLevel = 0, // ignored in this test + ) +} diff --git a/apps/scan/src/test/scala/org/lfdecentralizedtrust/splice/scan/config/ScanStorageConfigTest.scala b/apps/scan/src/test/scala/org/lfdecentralizedtrust/splice/scan/config/ScanStorageConfigTest.scala index 408c014bd3..0e2ffb1fd7 100644 --- a/apps/scan/src/test/scala/org/lfdecentralizedtrust/splice/scan/config/ScanStorageConfigTest.scala +++ b/apps/scan/src/test/scala/org/lfdecentralizedtrust/splice/scan/config/ScanStorageConfigTest.scala @@ -12,41 +12,30 @@ class ScanStorageConfigTest "ScanStorageConfig" should { "computeSnapshotTimeAfter" should { + def mkConfig(periodHours: Int) = ScanStorageConfig( + dbAcsSnapshotPeriodHours = periodHours, + perAcsSnapshotTablesEnabled = false, + bulkAcsSnapshotPeriodHours = 4, + bulkDbReadChunkSize = 1, + bulkZstdFrameSize = 0L, + bulkMaxFileSize = 0L, + zstdCompressionLevel = 0, + ) + "return correct time if the previous one is not a valid snapshot time" in { - val config = ScanStorageConfig( - dbAcsSnapshotPeriodHours = 2, - bulkAcsSnapshotPeriodHours = 4, - bulkDbReadChunkSize = 1, - bulkZstdFrameSize = 0L, - bulkMaxFileSize = 0L, - zstdCompressionLevel = 0, - ) + val config = mkConfig(periodHours = 2) val prev = cantonTimestamp("2007-12-03T11:30:00.00Z") val next = cantonTimestamp("2007-12-03T12:00:00.00Z") config.computeDbSnapshotTimeAfter(prev) shouldBe next } "return correct time if the previous one is a valid snapshot time" in { - val config = ScanStorageConfig( - dbAcsSnapshotPeriodHours = 2, - bulkAcsSnapshotPeriodHours = 4, - bulkDbReadChunkSize = 1, - bulkZstdFrameSize = 0L, - bulkMaxFileSize = 0L, - zstdCompressionLevel = 0, - ) + val config = mkConfig(periodHours = 2) val prev = cantonTimestamp("2007-12-03T12:00:00.00Z") val next = cantonTimestamp("2007-12-03T14:00:00.00Z") config.computeDbSnapshotTimeAfter(prev) shouldBe next } "return correct time if the next one is on the day after" in { - val config = ScanStorageConfig( - dbAcsSnapshotPeriodHours = 4, - bulkAcsSnapshotPeriodHours = 8, - bulkDbReadChunkSize = 1, - bulkZstdFrameSize = 0L, - bulkMaxFileSize = 0L, - zstdCompressionLevel = 0, - ) + val config = mkConfig(periodHours = 4) val prev = cantonTimestamp("2007-12-03T21:00:00.00Z") val next = cantonTimestamp("2007-12-04T00:00:00.00Z") config.computeDbSnapshotTimeAfter(prev) shouldBe next diff --git a/apps/scan/src/test/scala/org/lfdecentralizedtrust/splice/scan/store/bulk/AcsSnapshotBulkStorageCommitFromStagingTest.scala b/apps/scan/src/test/scala/org/lfdecentralizedtrust/splice/scan/store/bulk/AcsSnapshotBulkStorageCommitFromStagingTest.scala index 5159c40f1b..b3036b0d84 100644 --- a/apps/scan/src/test/scala/org/lfdecentralizedtrust/splice/scan/store/bulk/AcsSnapshotBulkStorageCommitFromStagingTest.scala +++ b/apps/scan/src/test/scala/org/lfdecentralizedtrust/splice/scan/store/bulk/AcsSnapshotBulkStorageCommitFromStagingTest.scala @@ -45,6 +45,7 @@ class AcsSnapshotBulkStorageCommitFromStagingTest val bulkStorageTestConfig = ScanStorageConfig( dbAcsSnapshotPeriodHours = 3, + perAcsSnapshotTablesEnabled = false, bulkAcsSnapshotPeriodHours = 24, bulkDbReadChunkSize = 1000, bulkZstdFrameSize = 10000L, diff --git a/apps/scan/src/test/scala/org/lfdecentralizedtrust/splice/scan/store/bulk/AcsSnapshotBulkStorageWriterFromDbTest.scala b/apps/scan/src/test/scala/org/lfdecentralizedtrust/splice/scan/store/bulk/AcsSnapshotBulkStorageWriterFromDbTest.scala index c148a4a7c5..a38280d877 100644 --- a/apps/scan/src/test/scala/org/lfdecentralizedtrust/splice/scan/store/bulk/AcsSnapshotBulkStorageWriterFromDbTest.scala +++ b/apps/scan/src/test/scala/org/lfdecentralizedtrust/splice/scan/store/bulk/AcsSnapshotBulkStorageWriterFromDbTest.scala @@ -70,6 +70,7 @@ class AcsSnapshotBulkStorageWriterFromDbTest val acsSnapshotSize = 48500 val bulkStorageTestConfig = ScanStorageConfig( dbAcsSnapshotPeriodHours = 3, + perAcsSnapshotTablesEnabled = false, bulkAcsSnapshotPeriodHours = 24, bulkDbReadChunkSize = 1000, bulkZstdFrameSize = 10000L, diff --git a/apps/scan/src/test/scala/org/lfdecentralizedtrust/splice/scan/store/bulk/UpdateHistoryBulkStorageTest.scala b/apps/scan/src/test/scala/org/lfdecentralizedtrust/splice/scan/store/bulk/UpdateHistoryBulkStorageTest.scala index eedad265c6..f0672f06ac 100644 --- a/apps/scan/src/test/scala/org/lfdecentralizedtrust/splice/scan/store/bulk/UpdateHistoryBulkStorageTest.scala +++ b/apps/scan/src/test/scala/org/lfdecentralizedtrust/splice/scan/store/bulk/UpdateHistoryBulkStorageTest.scala @@ -54,6 +54,7 @@ class UpdateHistoryBulkStorageTest val maxFileSize = 25000L val bulkStorageTestConfig = ScanStorageConfig( dbAcsSnapshotPeriodHours = 1, + perAcsSnapshotTablesEnabled = false, bulkAcsSnapshotPeriodHours = 2, bulkDbReadChunkSize = 500, bulkZstdFrameSize = 10000L, diff --git a/apps/scan/src/test/scala/org/lfdecentralizedtrust/splice/store/db/AcsSnapshotStoreTest.scala b/apps/scan/src/test/scala/org/lfdecentralizedtrust/splice/store/db/AcsSnapshotStoreTest.scala index 4c27a6c133..6af6df3e9d 100644 --- a/apps/scan/src/test/scala/org/lfdecentralizedtrust/splice/store/db/AcsSnapshotStoreTest.scala +++ b/apps/scan/src/test/scala/org/lfdecentralizedtrust/splice/store/db/AcsSnapshotStoreTest.scala @@ -1,7 +1,7 @@ package org.lfdecentralizedtrust.splice.store.db import cats.data.NonEmptyVector -import com.daml.ledger.javaapi.data.{Unit as damlUnit} +import com.daml.ledger.javaapi.data.Unit as damlUnit import com.daml.ledger.javaapi.data.codegen.ContractId import com.daml.metrics.api.noop.NoOpMetricsFactory import org.lfdecentralizedtrust.splice.environment.DarResources @@ -32,8 +32,9 @@ import java.time.Instant import scala.concurrent.Future import scala.util.{Failure, Success} import StoreTestBase.* +import org.lfdecentralizedtrust.splice.scan.store.AcsSnapshotStore.IncrementalAcsSnapshotTable -class AcsSnapshotStoreTest +trait AcsSnapshotStoreTest extends StoreTestBase with HasExecutionContext with StoreErrors @@ -42,6 +43,8 @@ class AcsSnapshotStoreTest with AcsJdbcTypes with AcsTables { + val nextTable: IncrementalAcsSnapshotTable + private val DefaultMigrationId = 0L private val timestamp1 = CantonTimestamp.Epoch.plusSeconds(3600) private val timestamp2 = CantonTimestamp.Epoch.plusSeconds(3600 * 2) @@ -1077,9 +1080,7 @@ class AcsSnapshotStoreTest for { updateHistory <- mkUpdateHistory() store = mkStore(updateHistory) - incrementalSnapshotN <- store.getIncrementalSnapshot( - AcsSnapshotStore.IncrementalAcsSnapshotTable.Next - ) + incrementalSnapshotN <- store.getIncrementalSnapshot(nextTable) incrementalSnapshotB <- store.getIncrementalSnapshot( AcsSnapshotStore.IncrementalAcsSnapshotTable.Backfill ) @@ -1122,13 +1123,13 @@ class AcsSnapshotStoreTest } yield snapshot1) _ <- store.initializeIncrementalSnapshot( - AcsSnapshotStore.IncrementalAcsSnapshotTable.Next, + nextTable, snapshot1.value, timestamp2, ) incrementalSnapshotN <- store.getIncrementalSnapshot( - AcsSnapshotStore.IncrementalAcsSnapshotTable.Next + nextTable ) incrementalSnapshotB <- store.getIncrementalSnapshot( AcsSnapshotStore.IncrementalAcsSnapshotTable.Backfill @@ -1136,10 +1137,10 @@ class AcsSnapshotStoreTest _ <- clueF(s"Update snapshot")(for { snapshot <- store.getIncrementalSnapshot( - AcsSnapshotStore.IncrementalAcsSnapshotTable.Next + nextTable ) _ <- store.updateIncrementalSnapshot( - AcsSnapshotStore.IncrementalAcsSnapshotTable.Next, + nextTable, snapshot.value, timestamp2, ) @@ -1147,10 +1148,10 @@ class AcsSnapshotStoreTest _ <- clueF(s"Save snapshot")(for { snapshot <- store.getIncrementalSnapshot( - AcsSnapshotStore.IncrementalAcsSnapshotTable.Next + nextTable ) _ <- store.saveIncrementalSnapshot( - AcsSnapshotStore.IncrementalAcsSnapshotTable.Next, + nextTable, snapshot.value, nextSnapshotTargetRecordTime = timestamp3, ) @@ -1193,13 +1194,13 @@ class AcsSnapshotStoreTest snapshotRecordTime = timestamp1.minusSeconds(1L) _ <- store.initializeIncrementalSnapshotFromImportUpdates( - AcsSnapshotStore.IncrementalAcsSnapshotTable.Next, + nextTable, snapshotRecordTime, timestamp2, DefaultMigrationId, ) incrementalSnapshotN <- store.getIncrementalSnapshot( - AcsSnapshotStore.IncrementalAcsSnapshotTable.Next + nextTable ) incrementalSnapshotB <- store.getIncrementalSnapshot( AcsSnapshotStore.IncrementalAcsSnapshotTable.Backfill @@ -1253,7 +1254,7 @@ class AcsSnapshotStoreTest _ <- clueF(s"Snapshot A: start from import updates at T0")( storeM1.initializeIncrementalSnapshotFromImportUpdates( - AcsSnapshotStore.IncrementalAcsSnapshotTable.Next, + nextTable, recordTime = timestamp1, targetRecordTime = timestamp1.plusSeconds(9L), 1L, @@ -1266,10 +1267,10 @@ class AcsSnapshotStoreTest for { (_, cid1) <- ingestHistory(updateHistoryM1, 0L) snapshot <- storeM1.getIncrementalSnapshot( - AcsSnapshotStore.IncrementalAcsSnapshotTable.Next + nextTable ) _ <- storeM1.updateIncrementalSnapshot( - AcsSnapshotStore.IncrementalAcsSnapshotTable.Next, + nextTable, snapshot.value, // Note: there is no event at exactly T9 timestamp1.plusSeconds(9L), @@ -1279,11 +1280,11 @@ class AcsSnapshotStoreTest _ <- clueF(s"Snapshot A: finalize snapshot at T9")(for { snapshot <- storeM1.getIncrementalSnapshot( - AcsSnapshotStore.IncrementalAcsSnapshotTable.Next + nextTable ) _ = snapshot.value.recordTime shouldBe timestamp1.plusSeconds(9L) _ <- storeM1.saveIncrementalSnapshot( - AcsSnapshotStore.IncrementalAcsSnapshotTable.Next, + nextTable, snapshot.value, nextSnapshotTargetRecordTime = timestamp1.plusSeconds(20L), ) @@ -1311,10 +1312,10 @@ class AcsSnapshotStoreTest )(for { (_, cid2) <- ingestHistory(updateHistoryM1, 12L) snapshot <- storeM1.getIncrementalSnapshot( - AcsSnapshotStore.IncrementalAcsSnapshotTable.Next + nextTable ) _ <- storeM1.updateIncrementalSnapshot( - AcsSnapshotStore.IncrementalAcsSnapshotTable.Next, + nextTable, snapshot.value, // Note: there is an event at exactly T20 timestamp1.plusSeconds(20L), @@ -1323,11 +1324,11 @@ class AcsSnapshotStoreTest _ <- clueF(s"Snapshot B: finalize snapshot at T20")(for { snapshot <- storeM1.getIncrementalSnapshot( - AcsSnapshotStore.IncrementalAcsSnapshotTable.Next + nextTable ) _ = snapshot.value.recordTime shouldBe timestamp1.plusSeconds(20L) _ <- storeM1.saveIncrementalSnapshot( - AcsSnapshotStore.IncrementalAcsSnapshotTable.Next, + nextTable, snapshot.value, nextSnapshotTargetRecordTime = timestamp1.plusSeconds(30L), ) @@ -1364,23 +1365,23 @@ class AcsSnapshotStoreTest _ <- ingestCreate(updateHistory, amuletRules(), CantonTimestamp.MinValue) _ <- store.initializeIncrementalSnapshotFromImportUpdates( - AcsSnapshotStore.IncrementalAcsSnapshotTable.Next, + nextTable, timestamp2, timestamp3, DefaultMigrationId, ) incrementalSnapshotBefore <- store.getIncrementalSnapshot( - AcsSnapshotStore.IncrementalAcsSnapshotTable.Next + nextTable ) _ <- store.deleteIncrementalSnapshot( - AcsSnapshotStore.IncrementalAcsSnapshotTable.Next, + nextTable, incrementalSnapshotBefore.value, ) incrementalSnapshotAfter <- store.getIncrementalSnapshot( - AcsSnapshotStore.IncrementalAcsSnapshotTable.Next + nextTable ) } yield { incrementalSnapshotAfter shouldBe None @@ -1531,3 +1532,11 @@ class AcsSnapshotStoreTest } yield () } + +class LegacyAcsSnapshotStoreTest extends AcsSnapshotStoreTest { + override val nextTable: IncrementalAcsSnapshotTable = IncrementalAcsSnapshotTable.Next +} + +class TablePerAcsSnapshotStoreTest extends AcsSnapshotStoreTest { + override val nextTable: IncrementalAcsSnapshotTable = IncrementalAcsSnapshotTable.NextV2 +}