services/horizon: [v2.6.0] Add migration to rebuild trade aggregation…

…s when upgrading (#3760)

* Add migration to rebuild trade aggregations when upgrading

* Update changelog

* Fix open/close issue in migration

services/horizon/CHANGELOG.md

@@ -6,9 +6,9 @@ file. This project adheres to [Semantic Versioning](http://semver.org/).
 
 ## v2.6.0
 
-**Upgrading to this version from <= v2.1.1 will trigger a state rebuild. During this process (which can take up to 20 minutes), Horizon will not ingest new ledgers.**
+**Upgrading to this version from <= v2.5.2 will trigger a state rebuild. During this process (which will take at least 10 minutes), Horizon will not ingest new ledgers.**
 
-* Precompute trade aggregations during ingestion to improve performance. Requires a reingestion ([3641](https://github.com/stellar/go/pull/3641)).
+* Precompute trade aggregations during ingestion to improve performance. Will rebuild the aggregations as part of the database migrations. ([3641](https://github.com/stellar/go/pull/3641) & [3760](https://github.com/stellar/go/pull/3760)).
 * Require `COUNT` param when running `horizon db migrate down COUNT` to prevent accidentally running all downwards migrations. Add `horizon db migrate status` command. ([#3737](https://github.com/stellar/go/pull/3737))
 * Fix a bug in `fee_account_muxed` and `fee_account_muxed_id` fields (the fields were incorrectly populated with the source account details). ([3735](https://github.com/stellar/go/pull/3735))
 * Validate ledger range when calling `horizon db reingest range` so that we respond with an error when attempting to ingest ledgers which are not available in the history archives. ([3738](https://github.com/stellar/go/pull/3738))

services/horizon/internal/db2/schema/migrations/48_rebuild_trade_aggregations.sql

@@ -0,0 +1,44 @@
+-- +migrate Up
+
+-- Backfill the table with existing data. This takes about 9 minutes.
+WITH trades AS (
+  SELECT
+    to_millis(ledger_closed_at, 60000) as timestamp,
+    history_operation_id,
+    "order",
+    base_asset_id,
+    base_amount,
+    counter_asset_id,
+    counter_amount,
+    ARRAY[price_n, price_d] as price
+  FROM history_trades
+  ORDER BY base_asset_id, counter_asset_id, history_operation_id, "order"
+), rebuilt as (
+  SELECT
+    timestamp,
+    base_asset_id,
+    counter_asset_id,
+    count(*) as count,
+    sum(base_amount) as base_volume,
+    sum(counter_amount) as counter_volume,
+    sum(counter_amount::numeric)/sum(base_amount::numeric) as avg,
+    (max_price(price))[1] as high_n,
+    (max_price(price))[2] as high_d,
+    (min_price(price))[1] as low_n,
+    (min_price(price))[2] as low_d,
+    first(history_operation_id) as open_ledger_toid,
+    (first(price))[1] as open_n,
+    (first(price))[2] as open_d,
+    last(history_operation_id) as close_ledger_toid,
+    (last(price))[1] as close_n,
+    (last(price))[2] as close_d
+  FROM trades
+  GROUP by base_asset_id, counter_asset_id, timestamp
+)
+  INSERT INTO history_trades_60000 (
+    SELECT * from rebuilt
+  );
+
+-- +migrate Down
+
+TRUNCATE TABLE history_trades_60000;