Split composite DB index on htlc_infos table

We were previously using a composite index on both the `channel_id` and
the `commitment_number` in the `htlc_infos` table. This table is used to
store historical HTLC information to be able to publish penalty txs when
a malicious peer publishes a revoked commitment. This table is usually
the largest DB table on nodes that relay a lot of payments, because it
grows with the number of HTLCs received and sent (and only shrinks when
channels are closed or spliced).

Using a composite index makes sense, but leads to increased memory usage
compared to separate indices, thus reducing performance because this is
a table on which we write a lot, but almost never read (only when we
detect a revoked force-close). The read performance doesn't seem to be
negatively impacted anyway when splitting the indices, and the memory
usage is greatly improved.

The migration may take some time depending on the size of the table,
but we cannot get around this.

Thanks to @DerEwige for the performance investigation that lead to
this change (see #2932 for more details).

eclair-core/src/main/scala/fr/acinq/eclair/db/pg/PgChannelsDb.scala

@@ -36,7 +36,7 @@ import javax.sql.DataSource
 
 object PgChannelsDb {
   val DB_NAME = "channels"
-  val CURRENT_VERSION = 9
+  val CURRENT_VERSION = 10
 }
 
 class PgChannelsDb(implicit ds: DataSource, lock: PgLock) extends ChannelsDb with Logging {
@@ -118,6 +118,13 @@ class PgChannelsDb(implicit ds: DataSource, lock: PgLock) extends ChannelsDb wit
       def migration89(statement: Statement): Unit = {
         statement.executeUpdate("CREATE TABLE local.htlc_infos_to_remove (channel_id TEXT NOT NULL PRIMARY KEY, before_commitment_number BIGINT NOT NULL)")
       }
+
+      def migration910(statement: Statement): Unit = {
+        // We're changing our composite index to two distinct indices to improve performance.
+        statement.executeUpdate("CREATE INDEX htlc_infos_channel_id_idx ON local.htlc_infos(channel_id)")
+        statement.executeUpdate("CREATE INDEX htlc_infos_commitment_number_idx ON local.htlc_infos(commitment_number)")
+        statement.executeUpdate("DROP INDEX IF EXISTS htlc_infos_idx")
+      }
 
       getVersion(statement, DB_NAME) match {
         case None =>
@@ -129,8 +136,11 @@ class PgChannelsDb(implicit ds: DataSource, lock: PgLock) extends ChannelsDb wit
 
           statement.executeUpdate("CREATE INDEX local_channels_type_idx ON local.channels ((json->>'type'))")
           statement.executeUpdate("CREATE INDEX local_channels_remote_node_id_idx ON local.channels(remote_node_id)")
-          statement.executeUpdate("CREATE INDEX htlc_infos_idx ON local.htlc_infos(channel_id, commitment_number)")
-        case Some(v@(2 | 3 | 4 | 5 | 6 | 7 | 8)) =>
+          // Note that we use two distinct indices instead of a composite index on (channel_id, commitment_number).
+          // This is more efficient because we're writing a lot to this table but only reading when a channel is force-closed.
+          statement.executeUpdate("CREATE INDEX htlc_infos_channel_id_idx ON local.htlc_infos(channel_id)")
+          statement.executeUpdate("CREATE INDEX htlc_infos_commitment_number_idx ON local.htlc_infos(commitment_number)")
+        case Some(v@(2 | 3 | 4 | 5 | 6 | 7 | 8 | 9)) =>
           logger.warn(s"migrating db $DB_NAME, found version=$v current=$CURRENT_VERSION")
           if (v < 3) {
             migration23(statement)
@@ -153,6 +163,9 @@ class PgChannelsDb(implicit ds: DataSource, lock: PgLock) extends ChannelsDb wit
           if (v < 9) {
             migration89(statement)
           }
+          if (v < 10) {
+            migration910(statement)
+          }
         case Some(CURRENT_VERSION) => () // table is up-to-date, nothing to do
         case Some(unknownVersion) => throw new RuntimeException(s"Unknown version of DB $DB_NAME found, version=$unknownVersion")
       }

eclair-core/src/main/scala/fr/acinq/eclair/db/sqlite/SqliteChannelsDb.scala

@@ -32,7 +32,7 @@ import java.sql.{Connection, Statement}
 
 object SqliteChannelsDb {
   val DB_NAME = "channels"
-  val CURRENT_VERSION = 5
+  val CURRENT_VERSION = 6
 }
 
 class SqliteChannelsDb(val sqlite: Connection) extends ChannelsDb with Logging {
@@ -83,13 +83,23 @@ class SqliteChannelsDb(val sqlite: Connection) extends ChannelsDb with Logging {
       statement.executeUpdate("CREATE TABLE htlc_infos_to_remove (channel_id BLOB NOT NULL PRIMARY KEY, before_commitment_number INTEGER NOT NULL)")
     }
 
+    def migration56(): Unit = {
+      // We're changing our composite index to two distinct indices to improve performance.
+      statement.executeUpdate("CREATE INDEX htlc_infos_channel_id_idx ON htlc_infos(channel_id)")
+      statement.executeUpdate("CREATE INDEX htlc_infos_commitment_number_idx ON htlc_infos(commitment_number)")
+      statement.executeUpdate("DROP INDEX IF EXISTS htlc_infos_idx")
+    }
+
     getVersion(statement, DB_NAME) match {
       case None =>
         statement.executeUpdate("CREATE TABLE local_channels (channel_id BLOB NOT NULL PRIMARY KEY, data BLOB NOT NULL, is_closed BOOLEAN NOT NULL DEFAULT 0, created_timestamp INTEGER, last_payment_sent_timestamp INTEGER, last_payment_received_timestamp INTEGER, last_connected_timestamp INTEGER, closed_timestamp INTEGER)")
         statement.executeUpdate("CREATE TABLE htlc_infos (channel_id BLOB NOT NULL, commitment_number INTEGER NOT NULL, payment_hash BLOB NOT NULL, cltv_expiry INTEGER NOT NULL, FOREIGN KEY(channel_id) REFERENCES local_channels(channel_id))")
         statement.executeUpdate("CREATE TABLE htlc_infos_to_remove (channel_id BLOB NOT NULL PRIMARY KEY, before_commitment_number INTEGER NOT NULL)")
-        statement.executeUpdate("CREATE INDEX htlc_infos_idx ON htlc_infos(channel_id, commitment_number)")
-      case Some(v@(1 | 2 | 3 | 4)) =>
+        // Note that we use two distinct indices instead of a composite index on (channel_id, commitment_number).
+        // This is more efficient because we're writing a lot to this table but only reading when a channel is force-closed.
+        statement.executeUpdate("CREATE INDEX htlc_infos_channel_id_idx ON htlc_infos(channel_id)")
+        statement.executeUpdate("CREATE INDEX htlc_infos_commitment_number_idx ON htlc_infos(commitment_number)")
+      case Some(v@(1 | 2 | 3 | 4 | 5)) =>
         logger.warn(s"migrating db $DB_NAME, found version=$v current=$CURRENT_VERSION")
         if (v < 2) {
           migration12(statement)
@@ -103,6 +113,9 @@ class SqliteChannelsDb(val sqlite: Connection) extends ChannelsDb with Logging {
         if (v < 5) {
           migration45()
         }
+        if (v < 6) {
+          migration56()
+        }
       case Some(CURRENT_VERSION) => () // table is up-to-date, nothing to do
       case Some(unknownVersion) => throw new RuntimeException(s"Unknown version of DB $DB_NAME found, version=$unknownVersion")
     }