services/horizon/internal/db2: Add column and index to improve trade_…

…type filter performance (#4149)

Add trade_type column which is an enum value that represents whether the trade occurs on the orderbook or against a liquidity pool. The trade_type column is populated during ingestion. This commit also adds an index on the trade_type column to make queries for liquidity pool trades fast.

services/horizon/CHANGELOG.md

@@ -5,8 +5,15 @@ file. This project adheres to [Semantic Versioning](http://semver.org/).
 
 ## Unreleased
 
+* Generate Http Status code of 499 for Client Disconnects, should propagate into `horizon_http_requests_duration_seconds_count`
+  metric key with status=499 label. ([4098](horizon_http_requests_duration_seconds_count))
+* Improve performance of `/trades?trade_type=liquidity_pool` requests. ([4149](https://github.com/stellar/go/pull/4149))
 * Added `absBeforeEpoch` to ClaimableBalance API Resources. It will contain the Unix epoch representation of absolute before date. ([4148](https://github.com/stellar/go/pull/4148))  
 
+### DB Schema Migration
+
+* DB migrations add a column and index to the `history_trades` table. This is very large table so migration may take a long time (depending on your DB hardware). Please test the migrations execution time on the copy of your production DB first.
+
 ## v2.12.1
 
 ### Fixes

services/horizon/internal/actions_trade_test.go

@@ -315,10 +315,13 @@ func assertResponseTradeEqualsDBTrade(ht *HTTPT, row history.Trade, record horiz
 	ht.Assert.Equal(row.PriceN.Int64, record.Price.N)
 	ht.Assert.Equal(row.PriceD.Int64, record.Price.D)
 
-	if row.BaseLiquidityPoolID.Valid || row.CounterLiquidityPoolID.Valid {
-		ht.Assert.Equal(history.LiquidityPoolTrades, record.TradeType)
-	} else {
+	switch row.Type {
+	case history.OrderbookTradeType:
 		ht.Assert.Equal(history.OrderbookTrades, record.TradeType)
+	case history.LiquidityPoolTradeType:
+		ht.Assert.Equal(history.LiquidityPoolTrades, record.TradeType)
+	default:
+		ht.Assert.Fail("invalid trade type %v", row.Type)
 	}
 }
 
@@ -833,6 +836,8 @@ func IngestTestTrade(
 		PriceN:          int64(price.N),
 		PriceD:          int64(price.D),
 		LedgerCloseTime: timestamp.ToTime(),
+
+		Type: history.OrderbookTradeType,
 	})
 	err = batch.Exec(ctx)
 	if err != nil {

services/horizon/internal/db2/history/main.go

@@ -704,6 +704,7 @@ type Trade struct {
 	BaseIsSeller           bool        `db:"base_is_seller"`
 	PriceN                 null.Int    `db:"price_n"`
 	PriceD                 null.Int    `db:"price_d"`
+	Type                   TradeType   `db:"trade_type"`
 }
 
 // Transaction is a row of data from the `history_transactions` table

services/horizon/internal/db2/history/trade.go

@@ -171,9 +171,9 @@ func createTradesSQL(page db2.PageQuery, query historyTradesQuery) (string, []in
 
 	switch query.tradeType {
 	case OrderbookTrades:
-		sql = sql.Where(sq.Eq{"htrd.base_liquidity_pool_id": nil, "htrd.counter_liquidity_pool_id": nil})
+		sql = sql.Where(sq.Eq{"htrd.trade_type": OrderbookTradeType})
 	case LiquidityPoolTrades:
-		sql = sql.Where(sq.Eq{"htrd.base_offer_id": nil, "htrd.counter_offer_id": nil})
+		sql = sql.Where(sq.Eq{"htrd.trade_type": LiquidityPoolTradeType})
 	case AllTrades:
 	default:
 		return "", nil, errors.Errorf("Invalid trade type: %v", query.tradeType)
@@ -308,6 +308,7 @@ var selectTradeFields = sq.Select(
 	"htrd.base_is_seller",
 	"htrd.price_n",
 	"htrd.price_d",
+	"htrd.trade_type",
 )
 
 var selectReverseTradeFields = sq.Select(
@@ -332,6 +333,7 @@ var selectReverseTradeFields = sq.Select(
 	"NOT(htrd.base_is_seller) as base_is_seller",
 	"htrd.price_d as price_n",
 	"htrd.price_n as price_d",
+	"htrd.trade_type",
 )
 
 func getCanonicalAssetOrder(

services/horizon/internal/db2/history/trade_batch_insert_builder.go

@@ -10,6 +10,16 @@ import (
 	"github.com/stellar/go/support/errors"
 )
 
+// TradeType is an enum which indicates the type of trade
+type TradeType int16
+
+const (
+	// OrderbookTradeType is a trade which exercises an offer on the orderbook.
+	OrderbookTradeType = TradeType(1)
+	// LiquidityPoolTradeType is a trade which exercises a liquidity pool.
+	LiquidityPoolTradeType = TradeType(2)
+)
+
 // InsertTrade represents the arguments to TradeBatchInsertBuilder.Add() which is used to insert
 // rows into the history_trades table
 type InsertTrade struct {
@@ -33,6 +43,8 @@ type InsertTrade struct {
 
 	BaseIsSeller bool `db:"base_is_seller"`
 
+	Type TradeType `db:"trade_type"`
+
 	PriceN int64 `db:"price_n"`
 	PriceD int64 `db:"price_d"`
 }

services/horizon/internal/db2/history/trade_scenario.go

@@ -30,6 +30,7 @@ func createInsertTrades(
 		CounterAmount:      896,
 		PriceN:             1,
 		PriceD:             3,
+		Type:               OrderbookTradeType,
 	}
 
 	second := first
@@ -51,6 +52,7 @@ func createInsertTrades(
 		CounterAmount:      6,
 		PriceN:             1156,
 		PriceD:             3,
+		Type:               OrderbookTradeType,
 	}
 
 	fourth := InsertTrade{
@@ -67,6 +69,7 @@ func createInsertTrades(
 		BaseIsSeller:        true,
 		PriceN:              675,
 		PriceD:              981,
+		Type:                LiquidityPoolTradeType,
 	}
 
 	fifth := InsertTrade{
@@ -83,6 +86,7 @@ func createInsertTrades(
 		BaseIsSeller:        true,
 		PriceN:              43,
 		PriceD:              56,
+		Type:                LiquidityPoolTradeType,
 	}
 
 	return []InsertTrade{
@@ -293,6 +297,7 @@ func TradeScenario(tt *test.T, q *Q) TradeFixtures {
 			BaseIsSeller:       true,
 			PriceN:             null.IntFrom(inserts[0].PriceN),
 			PriceD:             null.IntFrom(inserts[0].PriceD),
+			Type:               OrderbookTradeType,
 		},
 		{
 			HistoryOperationID: inserts[1].HistoryOperationID,
@@ -313,6 +318,7 @@ func TradeScenario(tt *test.T, q *Q) TradeFixtures {
 			BaseIsSeller:       true,
 			PriceN:             null.IntFrom(inserts[1].PriceN),
 			PriceD:             null.IntFrom(inserts[1].PriceD),
+			Type:               OrderbookTradeType,
 		},
 		{
 			HistoryOperationID: inserts[2].HistoryOperationID,
@@ -333,6 +339,7 @@ func TradeScenario(tt *test.T, q *Q) TradeFixtures {
 			BaseIsSeller:       false,
 			PriceN:             null.IntFrom(inserts[2].PriceN),
 			PriceD:             null.IntFrom(inserts[2].PriceD),
+			Type:               OrderbookTradeType,
 		},
 		{
 			HistoryOperationID:  inserts[3].HistoryOperationID,
@@ -354,6 +361,7 @@ func TradeScenario(tt *test.T, q *Q) TradeFixtures {
 			LiquidityPoolFee:    inserts[3].LiquidityPoolFee,
 			PriceN:              null.IntFrom(inserts[3].PriceN),
 			PriceD:              null.IntFrom(inserts[3].PriceD),
+			Type:                LiquidityPoolTradeType,
 		},
 		{
 			HistoryOperationID:  inserts[4].HistoryOperationID,
@@ -375,6 +383,7 @@ func TradeScenario(tt *test.T, q *Q) TradeFixtures {
 			LiquidityPoolFee:    inserts[4].LiquidityPoolFee,
 			PriceN:              null.IntFrom(inserts[4].PriceN),
 			PriceD:              null.IntFrom(inserts[4].PriceD),
+			Type:                LiquidityPoolTradeType,
 		},
 	}
 

services/horizon/internal/db2/schema/migrations/52_add_trade_type_index.sql

@@ -0,0 +1,10 @@
+-- +migrate Up
+
+ALTER TABLE history_trades ADD trade_type smallint DEFAULT 1 CHECK(trade_type > 0);
+UPDATE history_trades SET trade_type = 2 WHERE base_liquidity_pool_id IS NOT NULL OR counter_liquidity_pool_id IS NOT NULL;
+CREATE INDEX htrd_by_trade_type ON history_trades USING BTREE(trade_type, history_operation_id, "order");
+
+-- +migrate Down
+
+DROP INDEX htrd_by_trade_type;
+ALTER TABLE history_trades DROP trade_type;

services/horizon/internal/ingest/processors/trades_processor.go

@@ -299,9 +299,11 @@ func (p *TradeProcessor) extractTrades(
 					return nil, err
 				}
 				row.LiquidityPoolFee = null.IntFrom(int64(fee))
+				row.Type = history.LiquidityPoolTradeType
 			} else {
 				row.BaseOfferID = null.IntFrom(int64(trade.OfferId()))
 				sellerAccount = trade.SellerId().Address()
+				row.Type = history.OrderbookTradeType
 			}
 
 			if buyOfferExists {

services/horizon/internal/ingest/processors/trades_processor_test.go

@@ -241,6 +241,7 @@ func (s *TradeProcessorTestSuiteLedger) mockReadTradeTransactions(
 			BaseIsSeller:       false,
 			PriceN:             int64(s.sellPrices[0].D),
 			PriceD:             int64(s.sellPrices[0].N),
+			Type:               history.OrderbookTradeType,
 		},
 		{
 			HistoryOperationID: toid.New(int32(ledger.Header.LedgerSeq), 1, 3).ToInt64(),
@@ -257,6 +258,7 @@ func (s *TradeProcessorTestSuiteLedger) mockReadTradeTransactions(
 			BaseOfferID:        null.IntFrom(int64(s.strictSendTrade.OfferId())),
 			PriceN:             int64(s.sellPrices[1].N),
 			PriceD:             int64(s.sellPrices[1].D),
+			Type:               history.OrderbookTradeType,
 		},
 		{
 			HistoryOperationID: toid.New(int32(ledger.Header.LedgerSeq), 1, 4).ToInt64(),
@@ -273,6 +275,7 @@ func (s *TradeProcessorTestSuiteLedger) mockReadTradeTransactions(
 			CounterOfferID:     null.IntFrom(int64(s.buyOfferTrade.OfferId())),
 			PriceN:             int64(s.sellPrices[2].D),
 			PriceD:             int64(s.sellPrices[2].N),
+			Type:               history.OrderbookTradeType,
 		},
 		{
 			HistoryOperationID: toid.New(int32(ledger.Header.LedgerSeq), 1, 5).ToInt64(),
@@ -289,6 +292,7 @@ func (s *TradeProcessorTestSuiteLedger) mockReadTradeTransactions(
 			BaseOfferID:        null.IntFrom(int64(s.sellOfferTrade.OfferId())),
 			PriceN:             int64(s.sellPrices[3].N),
 			PriceD:             int64(s.sellPrices[3].D),
+			Type:               history.OrderbookTradeType,
 		},
 		{
 			HistoryOperationID: toid.New(int32(ledger.Header.LedgerSeq), 1, 6).ToInt64(),
@@ -305,6 +309,7 @@ func (s *TradeProcessorTestSuiteLedger) mockReadTradeTransactions(
 			CounterOfferID:     null.IntFrom(int64(s.passiveSellOfferTrade.OfferId())),
 			PriceN:             int64(s.sellPrices[4].D),
 			PriceD:             int64(s.sellPrices[4].N),
+			Type:               history.OrderbookTradeType,
 		},
 
 		{
@@ -323,6 +328,7 @@ func (s *TradeProcessorTestSuiteLedger) mockReadTradeTransactions(
 			BaseOfferID:      null.IntFrom(int64(s.otherPassiveSellOfferTrade.OfferId())),
 			PriceN:           int64(s.sellPrices[5].N),
 			PriceD:           int64(s.sellPrices[5].D),
+			Type:             history.OrderbookTradeType,
 		},
 		{
 			HistoryOperationID:     toid.New(int32(ledger.Header.LedgerSeq), 1, 8).ToInt64(),
@@ -339,6 +345,7 @@ func (s *TradeProcessorTestSuiteLedger) mockReadTradeTransactions(
 			LiquidityPoolFee:       null.IntFrom(int64(xdr.LiquidityPoolFeeV18)),
 			PriceN:                 int64(s.sellPrices[6].D),
 			PriceD:                 int64(s.sellPrices[6].N),
+			Type:                   history.LiquidityPoolTradeType,
 		},
 		{
 			HistoryOperationID:  toid.New(int32(ledger.Header.LedgerSeq), 1, 9).ToInt64(),
@@ -355,6 +362,7 @@ func (s *TradeProcessorTestSuiteLedger) mockReadTradeTransactions(
 			LiquidityPoolFee:    null.IntFrom(int64(xdr.LiquidityPoolFeeV18)),
 			PriceN:              int64(s.sellPrices[7].N),
 			PriceD:              int64(s.sellPrices[7].D),
+			Type:                history.LiquidityPoolTradeType,
 		},
 	}
 

services/horizon/internal/integration/trade_aggregations_test.go

@@ -76,6 +76,7 @@ func TestTradeAggregations(t *testing.T) {
 					CounterAssetID:     counterAssetId,
 					PriceN:             23456,
 					PriceD:             10000,
+					Type:               history.OrderbookTradeType,
 				},
 			},
 			resolution: 60_000,
@@ -115,6 +116,7 @@ func TestTradeAggregations(t *testing.T) {
 					CounterAssetID:     counterAssetId,
 					PriceN:             23456,
 					PriceD:             10000,
+					Type:               history.OrderbookTradeType,
 				},
 				{
 					HistoryOperationID: 0,
@@ -130,6 +132,7 @@ func TestTradeAggregations(t *testing.T) {
 					CounterAssetID:     counterAssetId,
 					PriceN:             13456,
 					PriceD:             10000,
+					Type:               history.OrderbookTradeType,
 				},
 			},
 			resolution: 60_000,
@@ -169,6 +172,7 @@ func TestTradeAggregations(t *testing.T) {
 					CounterAssetID:     counterAssetId,
 					PriceN:             23456,
 					PriceD:             10000,
+					Type:               history.OrderbookTradeType,
 				},
 				{
 					HistoryOperationID: 0,
@@ -184,6 +188,7 @@ func TestTradeAggregations(t *testing.T) {
 					CounterAssetID:     counterAssetId,
 					PriceN:             13456,
 					PriceD:             10000,
+					Type:               history.OrderbookTradeType,
 				},
 			},
 			resolution: 86_400_000,

services/horizon/internal/resourceadapter/trade.go

@@ -29,10 +29,8 @@ func PopulateTrade(
 	if row.BaseAccount.Valid {
 		dest.BaseAccount = row.BaseAccount.String
 	}
-	var isLiquidityPoolTrade bool
 	if row.BaseLiquidityPoolID.Valid {
 		dest.BaseLiquidityPoolID = row.BaseLiquidityPoolID.String
-		isLiquidityPoolTrade = true
 	}
 	dest.BaseAssetType = row.BaseAssetType
 	dest.BaseAssetCode = row.BaseAssetCode
@@ -47,7 +45,6 @@ func PopulateTrade(
 	}
 	if row.CounterLiquidityPoolID.Valid {
 		dest.CounterLiquidityPoolID = row.CounterLiquidityPoolID.String
-		isLiquidityPoolTrade = true
 	}
 	dest.CounterAssetType = row.CounterAssetType
 	dest.CounterAssetCode = row.CounterAssetCode
@@ -60,10 +57,11 @@ func PopulateTrade(
 		dest.LiquidityPoolFeeBP = uint32(row.LiquidityPoolFee.Int64)
 	}
 
-	if isLiquidityPoolTrade {
-		dest.TradeType = "liquidity_pool"
-	} else {
-		dest.TradeType = "orderbook"
+	switch row.Type {
+	case history.OrderbookTradeType:
+		dest.TradeType = history.OrderbookTrades
+	case history.LiquidityPoolTradeType:
+		dest.TradeType = history.LiquidityPoolTrades
 	}
 
 	if row.HasPrice() {