Write per snapshot table

apps/common/src/main/resources/db/migration/canton-network/postgres/stable/V073__acs_snapshot_table_name.sql

@@ -12,3 +12,80 @@ alter table acs_snapshot
     ((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_<record_time_epoch>.
+-- 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

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 {

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

apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/config/ScanStorageConfig.scala

@@ -20,6 +20,7 @@ import java.time.temporal.{ChronoField, ChronoUnit}
 
 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
@@ -139,6 +140,7 @@ case class ScanStorageConfig(
 object ScanStorageConfigs {
   val scanStorageConfigV1 = ScanStorageConfig(
     dbAcsSnapshotPeriodHours = 3,
+    perAcsSnapshotTablesEnabled = false,
     bulkAcsSnapshotPeriodHours = 24,
     bulkDbReadChunkSize = 1000,
     bulkZstdFrameSize = 12L * 1024 * 1024,