Repository: stellar/go
Pull request #4690 (2022-11-10 20:31:11)

services/horizon: Improve performance of claimable balances queries
<!-- If you're making a doc PR or something tiny where the below is irrelevant, delete this
template and use a short description, but in your description aim to include both what the
change is, and why it is being made, with enough context for anyone to understand. -->

<details>
  <summary>PR Checklist</summary>
  
### PR Structure

* [ ] This PR has reasonably narrow scope (if not, break it down into smaller PRs).
* [ ] This PR avoids mixing refactoring changes with feature changes (split into two PRs
  otherwise).
* [ ] This PR's title starts with name of package that is most changed in the PR, ex.
  `services/friendbot`, or `all` or `doc` if the changes are broad or impact many
  packages.

### Thoroughness

* [ ] This PR adds tests for the most critical parts of the new functionality or fixes.
* [ ] I've updated any docs ([developer docs](https://developers.stellar.org/api/), `.md`
  files, etc... affected by this change). Take a look in the `docs` folder for a given service,
  like [this one](https://github.com/stellar/go/tree/master/services/horizon/internal/docs).

### Release planning

* [ ] I've updated the relevant CHANGELOG ([here](services/horizon/CHANGELOG.md) for Horizon) if
  needed with deprecations, added features, breaking changes, and DB schema changes.
* [ ] I've decided if this PR requires a new major/minor version according to
  [semver](https://semver.org/), or if it's mainly a patch change. The PR is targeted at the next
  release branch if it's not a patch change.
</details>

### What

Add a new table `claimable_balance_claimants` which holds all claimants' destinations for corresponding claimable balances IDs. Also, improve other filters (by `sponsor` and `asset`) but adding better indexes for such queries.

### Why

We noticed that "claimable balances for claimants" query (`/claimable_balances?claimant=...`) is very slow in Postgres 12. The reason, apart from possible changes to gin index on `claimants` field, is that the `claimable_balances` table size significantly increased in the last couple months.

### Known limitations

This change requires a state rebuild to fill the new table. The new query has not been tested yet in stg because this environment is currently being rebuilt. I will test the query performance before merging the PR.


bartekn: I was finally able to test the code in this PR in stg and the results aren't great.

In one of Horizon meetings I mentioned that we should avoid a mistake we made in `history_transaction_participants` and `history_operation_participants` and include ordering fields in the _subtable_ used for outer queries (https://github.com/stellar/go/issues/1808). The code in this PR does not contain ordering fields because I was hoping that with limited number of balances per destination we could still make it fast with easier to maintain queries.

Unfortunately, while replaying production `/claimable_balances?claimaint=...` requests to staging I realized that there are destinations that are claimants on hundreds of thousands claimable balances. For example `GCJOXRRZWS7KQU6QWG6KHSJR3RVRYPAQ3YZNO52EXNJCMJ67ERUISO2O` is a destination on 610382 balances! This means that query engine must first get all the 610k rows, and the find them in the `claimable_balances` query and order by `last_modified_ledger`.

Another observation was that, contrary to history tables like participants mentioned above, current state tables are not immutable. This means that if a claimable balance entry changes (ex. by adding/changing a sponsor) all of it's destinations in a new table `claimable_balance_claimants` with ordering field will have to be updated as well. While this means up to 10 rows will be updated per operation, this can increase to 10,000 (technically slightly less because of sponsor sandwich ops) if all ops in a ledger change the sponsor.

This makes me think we have two options:
* Add ordering field to `claimable_balance_claimants` risking potential performance security issues. (Please note that in the end the update can be fast, it's hard to tell without testing the actual ingestion).
* Remove `order by cb.last_modified_ledger [asc|desc]` from the "Claimable Balances by Claimant" query. Technially this changes the contract because the order of results is different.

The second option is easier and, to be honest, I don't understand why we sort results by `last_modified_ledger` here but also in other state endpoints. Chronological order makes more sense because while it's still impossible to ensure we don't skip some entries (like balances created with IDs before the current page) we can at least walk through (almost entire) list. 

Both solutions have pros and cons so decided to open a discussion here.
bartekn: I decided to add order by field to `claimable_balance_claimants` in https://github.com/stellar/go/pull/4690/commits/73327c4cbd7b4274f085b1d3fb4e72fb9809b17c because it won't change the contract and also we still have plenty of time to test performance using verify-range. We can even write a small stress test. Once test pass, I'll deploy to stg.

I also noticed that "Claimable Balances for Asset" and "Claimable Balances for Sponsor" also have performance issues if an asset or sponsor is popular. I added new indexes to support such queries and fixed the query builder.
bartekn: OK, initial tests in stg look very good. PTAL! I'll share detailed comparison report tomorrow.
bartekn: I started mirroring yesterday. Results look good, `/claimable_balances` endpoint is much faster (in chart: 90, 95 and 99 percentiles) and the number of 503s dropped to 0.

<img width="1318" alt="Screenshot 2022-11-18 at 13 10 11" src="https://user-images.githubusercontent.com/464938/202704607-1bd1cb5c-6ca7-4564-a620-962aefadebbb.png">


Patch:
diff --git a/services/horizon/internal/db2/history/claimable_balance_claimant_batch_insert_builder.go b/services/horizon/internal/db2/history/claimable_balance_claimant_batch_insert_builder.go
new file mode 100644
index 00000000..a1a13d84
--- /dev/null
+++ b/services/horizon/internal/db2/history/claimable_balance_claimant_batch_insert_builder.go
@@ -0,0 +1,42 @@
+package history
+
+import (
+	"context"
+
+	"github.com/stellar/go/support/db"
+	"github.com/stellar/go/xdr"
+)
+
+// ClaimableBalanceClaimantBatchInsertBuilder is used to insert transactions into the
+// history_transactions table
+type ClaimableBalanceClaimantBatchInsertBuilder interface {
+	Add(ctx context.Context, claimableBalanceClaimant ClaimableBalanceClaimant) error
+	Exec(ctx context.Context) error
+}
+
+// ClaimableBalanceClaimantBatchInsertBuilder is a simple wrapper around db.BatchInsertBuilder
+type claimableBalanceClaimantBatchInsertBuilder struct {
+	encodingBuffer *xdr.EncodingBuffer
+	builder        db.BatchInsertBuilder
+}
+
+// NewClaimableBalanceClaimantBatchInsertBuilder constructs a new ClaimableBalanceClaimantBatchInsertBuilder instance
+func (q *Q) NewClaimableBalanceClaimantBatchInsertBuilder(maxBatchSize int) ClaimableBalanceClaimantBatchInsertBuilder {
+	return &claimableBalanceClaimantBatchInsertBuilder{
+		encodingBuffer: xdr.NewEncodingBuffer(),
+		builder: db.BatchInsertBuilder{
+			Table:        q.GetTable("claimable_balance_claimants"),
+			MaxBatchSize: maxBatchSize,
+			Suffix:       "ON CONFLICT (id, destination) DO UPDATE SET last_modified_ledger=EXCLUDED.last_modified_ledger",
+		},
+	}
+}
+
+// Add adds a new transaction to the batch
+func (i *claimableBalanceClaimantBatchInsertBuilder) Add(ctx context.Context, claimableBalanceClaimant ClaimableBalanceClaimant) error {
+	return i.builder.RowStruct(ctx, claimableBalanceClaimant)
+}
+
+func (i *claimableBalanceClaimantBatchInsertBuilder) Exec(ctx context.Context) error {
+	return i.builder.Exec(ctx)
+}
diff --git a/services/horizon/internal/db2/history/claimable_balances.go b/services/horizon/internal/db2/history/claimable_balances.go
index 31401ff2..3fe4029d 100644
--- a/services/horizon/internal/db2/history/claimable_balances.go
+++ b/services/horizon/internal/db2/history/claimable_balances.go
@@ -4,6 +4,7 @@ import (
 	"context"
 	"database/sql/driver"
 	"encoding/json"
+	"fmt"
 	"strconv"
 	"strings"
 
@@ -56,39 +57,42 @@ func (cbq ClaimableBalancesQuery) Cursor() (int64, string, error) {
 // ApplyCursor applies cursor to the given sql. For performance reason the limit
 // is not applied here. This allows us to hint the planner later to use the right
 // indexes.
-func (cbq ClaimableBalancesQuery) ApplyCursor(sql sq.SelectBuilder) (sq.SelectBuilder, error) {
-	p := cbq.PageQuery
+func applyClaimableBalancesQueriesCursor(sql sq.SelectBuilder, lCursor int64, rCursor string, order string) (sq.SelectBuilder, error) {
 	hasPagedLimit := false
-	l, r, err := cbq.Cursor()
-	if err != nil {
-		return sql, err
-	}
-	if l > 0 && r != "" {
+	if lCursor > 0 && rCursor != "" {
 		hasPagedLimit = true
 	}
 
-	switch p.Order {
+	switch order {
 	case db2.OrderAscending:
 		if hasPagedLimit {
 			sql = sql.
-				Where(sq.Expr("(cb.last_modified_ledger, cb.id) > (?, ?)", l, r))
+				Where(sq.Expr("(last_modified_ledger, id) > (?, ?)", lCursor, rCursor))
 
 		}
-		sql = sql.OrderBy("cb.last_modified_ledger asc, cb.id asc")
+		sql = sql.OrderBy("last_modified_ledger asc, id asc")
 	case db2.OrderDescending:
 		if hasPagedLimit {
 			sql = sql.
-				Where(sq.Expr("(cb.last_modified_ledger, cb.id) < (?, ?)", l, r))
+				Where(sq.Expr("(last_modified_ledger, id) < (?, ?)", lCursor, rCursor))
 		}
 
-		sql = sql.OrderBy("cb.last_modified_ledger desc, cb.id desc")
+		sql = sql.OrderBy("last_modified_ledger desc, id desc")
 	default:
-		return sql, errors.Errorf("invalid order: %s", p.Order)
+		return sql, errors.Errorf("invalid order: %s", order)
 	}
 
 	return sql, nil
 }
 
+// ClaimableBalanceClaimant is a row of data from the `claimable_balances_claimants` table.
+// This table exists to allow faster querying for claimable balances for a specific claimant.
+type ClaimableBalanceClaimant struct {
+	BalanceID          string `db:"id"`
+	Destination        string `db:"destination"`
+	LastModifiedLedger uint32 `db:"last_modified_ledger"`
+}
+
 // ClaimableBalance is a row of data from the `claimable_balances` table.
 type ClaimableBalance struct {
 	BalanceID          string      `db:"id"`
@@ -124,8 +128,11 @@ type Claimant struct {
 type QClaimableBalances interface {
 	UpsertClaimableBalances(ctx context.Context, cb []ClaimableBalance) error
 	RemoveClaimableBalances(ctx context.Context, ids []string) (int64, error)
+	RemoveClaimableBalanceClaimants(ctx context.Context, ids []string) (int64, error)
 	GetClaimableBalancesByID(ctx context.Context, ids []string) ([]ClaimableBalance, error)
 	CountClaimableBalances(ctx context.Context) (int, error)
+	NewClaimableBalanceClaimantBatchInsertBuilder(maxBatchSize int) ClaimableBalanceClaimantBatchInsertBuilder
+	GetClaimantsByClaimableBalances(ctx context.Context, ids []string) (map[string][]ClaimableBalanceClaimant, error)
 }
 
 // CountClaimableBalances returns the total number of claimable balances in the DB
@@ -148,6 +155,22 @@ func (q *Q) GetClaimableBalancesByID(ctx context.Context, ids []string) ([]Claim
 	return cBalances, err
 }
 
+// GetClaimantsByClaimableBalances finds all claimants for ClaimableBalanceIds.
+// The returned list is sorted by ids and then destination ids for each balance id.
+func (q *Q) GetClaimantsByClaimableBalances(ctx context.Context, ids []string) (map[string][]ClaimableBalanceClaimant, error) {
+	var claimants []ClaimableBalanceClaimant
+	sql := sq.Select("*").From("claimable_balance_claimants cbc").
+		Where(map[string]interface{}{"cbc.id": ids}).
+		OrderBy("id asc, destination asc")
+	err := q.Select(ctx, &claimants, sql)
+
+	claimantsMap := make(map[string][]ClaimableBalanceClaimant)
+	for _, claimant := range claimants {
+		claimantsMap[claimant.BalanceID] = append(claimantsMap[claimant.BalanceID], claimant)
+	}
+	return claimantsMap, err
+}
+
 // UpsertClaimableBalances upserts a batch of claimable balances in the claimable_balances table.
 // There's currently no limit of the number of offers this method can
 // accept other than 2GB limit of the query string length what should be enough
@@ -191,6 +214,19 @@ func (q *Q) RemoveClaimableBalances(ctx context.Context, ids []string) (int64, e
 	return result.RowsAffected()
 }
 
+// RemoveClaimableBalanceClaimants deletes claimable balance claimants.
+// Returns number of rows affected and error.
+func (q *Q) RemoveClaimableBalanceClaimants(ctx context.Context, ids []string) (int64, error) {
+	sql := sq.Delete("claimable_balance_claimants").
+		Where(sq.Eq{"id": ids})
+	result, err := q.Exec(ctx, sql)
+	if err != nil {
+		return 0, err
+	}
+
+	return result.RowsAffected()
+}
+
 // FindClaimableBalanceByID returns a claimable balance.
 func (q *Q) FindClaimableBalanceByID(ctx context.Context, balanceID string) (ClaimableBalance, error) {
 	var claimableBalance ClaimableBalance
@@ -201,12 +237,12 @@ func (q *Q) FindClaimableBalanceByID(ctx context.Context, balanceID string) (Cla
 
 // GetClaimableBalances finds all claimable balances where accountID is one of the claimants
 func (q *Q) GetClaimableBalances(ctx context.Context, query ClaimableBalancesQuery) ([]ClaimableBalance, error) {
-	sql, err := query.ApplyCursor(selectClaimableBalances)
-	// we need to use WITH syntax and correct LIMIT placement to force the query planner to use the right
-	// indexes, otherwise when the limit is small, it will use an index scan
-	// which will be very slow once we have millions of records
-	limitClausePlacement := "LIMIT ?) select " + claimableBalancesSelectStatement + " from cb"
+	l, r, err := query.Cursor()
+	if err != nil {
+		return nil, errors.Wrap(err, "error getting cursor")
+	}
 
+	sql, err := applyClaimableBalancesQueriesCursor(selectClaimableBalances, l, r, query.PageQuery.Order)
 	if err != nil {
 		return nil, errors.Wrap(err, "could not apply query to page")
 	}
@@ -221,19 +257,26 @@ func (q *Q) GetClaimableBalances(ctx context.Context, query ClaimableBalancesQue
 	}
 
 	if query.Claimant != nil {
+		var selectClaimableBalanceClaimants = sq.Select("id").From("claimable_balance_claimants").
+			Where("destination = ?", query.Claimant.Address()).
+			// Given that each destination can be a claimant for each balance maximum once
+			// we can LIMIT the subquery.
+			Limit(query.PageQuery.Limit)
+		subSql, err := applyClaimableBalancesQueriesCursor(selectClaimableBalanceClaimants, l, r, query.PageQuery.Order)
+		if err != nil {
+			return nil, errors.Wrap(err, "could not apply subquery to page")
+		}
+
+		subSqlString, subSqlArgs, err := subSql.ToSql()
+		if err != nil {
+			return nil, errors.Wrap(err, "could not build subquery")
+		}
+
 		sql = sql.
-			Where(`cb.claimants @> '[{"destination": "` + query.Claimant.Address() + `"}]'`)
-		// when search by claimant, profiling has shown the LIMIT should be on the outer query to
-		// hint appropriate indexes for best performance
-		limitClausePlacement = ") select " + claimableBalancesSelectStatement + " from cb LIMIT ?"
-	}
-
-	sql = sql.
-		Prefix("WITH cb AS (").
-		Suffix(
-			limitClausePlacement,
-			query.PageQuery.Limit,
-		)
+			Where(fmt.Sprintf("cb.id IN (%s)", subSqlString), subSqlArgs...)
+	}
+
+	sql = sql.Limit(query.PageQuery.Limit)
 
 	var results []ClaimableBalance
 	if err := q.Select(ctx, &results, sql); err != nil {
diff --git a/services/horizon/internal/db2/history/ingestion.go b/services/horizon/internal/db2/history/ingestion.go
index 518dfd86..d82188d8 100644
--- a/services/horizon/internal/db2/history/ingestion.go
+++ b/services/horizon/internal/db2/history/ingestion.go
@@ -15,6 +15,7 @@ func (q *Q) TruncateIngestStateTables(ctx context.Context) error {
 		"accounts_data",
 		"accounts_signers",
 		"claimable_balances",
+		"claimable_balance_claimants",
 		"exp_asset_stats",
 		"liquidity_pools",
 		"offers",
diff --git a/services/horizon/internal/db2/history/mock_claimable_balance_claimant_batch_insert_builder.go b/services/horizon/internal/db2/history/mock_claimable_balance_claimant_batch_insert_builder.go
new file mode 100644
index 00000000..63e2788e
--- /dev/null
+++ b/services/horizon/internal/db2/history/mock_claimable_balance_claimant_batch_insert_builder.go
@@ -0,0 +1,21 @@
+package history
+
+import (
+	"context"
+
+	"github.com/stretchr/testify/mock"
+)
+
+type MockClaimableBalanceClaimantBatchInsertBuilder struct {
+	mock.Mock
+}
+
+func (m *MockClaimableBalanceClaimantBatchInsertBuilder) Add(ctx context.Context, claimableBalanceClaimant ClaimableBalanceClaimant) error {
+	a := m.Called(ctx, claimableBalanceClaimant)
+	return a.Error(0)
+}
+
+func (m *MockClaimableBalanceClaimantBatchInsertBuilder) Exec(ctx context.Context) error {
+	a := m.Called(ctx)
+	return a.Error(0)
+}
diff --git a/services/horizon/internal/db2/history/mock_q_claimable_balances.go b/services/horizon/internal/db2/history/mock_q_claimable_balances.go
index 9d1cdcdc..2493d9eb 100644
--- a/services/horizon/internal/db2/history/mock_q_claimable_balances.go
+++ b/services/horizon/internal/db2/history/mock_q_claimable_balances.go
@@ -30,3 +30,18 @@ func (m *MockQClaimableBalances) RemoveClaimableBalances(ctx context.Context, id
 	a := m.Called(ctx, ids)
 	return a.Get(0).(int64), a.Error(1)
 }
+
+func (m *MockQClaimableBalances) RemoveClaimableBalanceClaimants(ctx context.Context, ids []string) (int64, error) {
+	a := m.Called(ctx, ids)
+	return a.Get(0).(int64), a.Error(1)
+}
+
+func (m *MockQClaimableBalances) NewClaimableBalanceClaimantBatchInsertBuilder(maxBatchSize int) ClaimableBalanceClaimantBatchInsertBuilder {
+	a := m.Called(maxBatchSize)
+	return a.Get(0).(ClaimableBalanceClaimantBatchInsertBuilder)
+}
+
+func (m *MockQClaimableBalances) GetClaimantsByClaimableBalances(ctx context.Context, ids []string) (map[string][]ClaimableBalanceClaimant, error) {
+	a := m.Called(ctx, ids)
+	return a.Get(0).(map[string][]ClaimableBalanceClaimant), a.Error(1)
+}
diff --git a/services/horizon/internal/db2/schema/bindata.go b/services/horizon/internal/db2/schema/bindata.go
index 36b6a2ca..e7131a62 100644
--- a/services/horizon/internal/db2/schema/bindata.go
+++ b/services/horizon/internal/db2/schema/bindata.go
@@ -58,6 +58,7 @@
 // migrations/5_create_trades_table.sql (1.1kB)
 // migrations/60_add_asset_id_indexes.sql (289B)
 // migrations/61_trust_lines_by_account_type_code_issuer.sql (383B)
+// migrations/62_claimable_balance_claimants.sql (1.428kB)
 // migrations/6_create_assets_table.sql (366B)
 // migrations/7_modify_trades_table.sql (2.303kB)
 // migrations/8_add_aggregators.sql (907B)
@@ -1291,6 +1292,26 @@ func migrations61_trust_lines_by_account_type_code_issuerSql() (*asset, error) {
 	return a, nil
 }
 
+var _migrations62_claimable_balance_claimantsSql = []byte("\x1f\x8b\x08\x00\x00\x00\x00\x00\x00\xff\xbc\x94\xc1\x6f\xd3\x30\x18\xc5\xef\xf9\x2b\x9e\x72\x6a\x43\xab\x5e\x60\x97\x0a\xa1\x6e\x8d\x50\x45\x49\xa7\x2c\x95\xb6\x53\xf4\xc5\xfe\xd6\x59\x4a\xec\xc8\x36\x1b\xfd\xef\x51\xd2\x0e\x02\x6c\x4d\x0a\x88\x53\x24\xcb\xef\x7d\xef\xfd\xec\x78\x3a\xc5\x9b\x4a\xed\x2c\x79\xc6\xb6\x0e\x82\xab\x34\x5e\x64\x31\xb2\xc5\xe5\x3a\x86\x28\x49\x55\x54\x94\x9c\x17\x54\x92\x16\x9c\x1f\x56\xb4\x77\x18\x05\x00\xa0\x24\xb2\xf8\x36\x43\xb2\xc9\x90\x6c\xd7\xeb\x09\xa6\x53\x5c\x3d\xcb\x2e\x0f\xaa\xd5\x12\x4a\xa3\x20\xc7\x17\x6f\x5b\x95\x64\xe7\x95\x26\xaf\x8c\x86\x78\x20\x4b\xc2\xb3\xc5\x23\xd9\xbd\xd2\xbb\xd1\xbb\x8b\xf1\x0f\xbf\x76\x7f\x49\xce\xe7\x95\x91\xea\x5e\xb1\xcc\x4b\x96\x3b\xb6\x50\xda\x73\xf3\xfd\x79\xeb\x75\xba\xfa\xbc\x48\xef\xf0\x29\xbe\xc3\x48\xc9\x49\x77\xd6\x38\x18\xcf\xbf\x37\x5c\x25\xcb\xf8\x16\xe1\x89\x8a\x79\xb1\xcf\x3b\xea\xfc\xa5\x14\xb9\x92\x21\x36\xc9\x49\x52\xdb\x9b\x55\xf2\x11\x85\xb7\xcc\x18\x75\x0c\x27\x2f\xf6\x9a\x40\xc9\x26\xe6\x2c\xc2\xcd\x97\xba\x36\xd6\x3b\x84\x8e\x4b\x16\x1e\x11\xee\xad\xa9\x7e\x1f\xe6\xf0\xf4\xc0\x96\x41\xce\xb1\xc7\x7b\x7c\x80\xb1\x92\x2d\x8a\xfd\xab\x23\x42\x44\xb3\x3e\x14\x2d\x81\xd6\xf3\xbc\xee\xbf\x54\x6e\x1d\x4e\x96\x9d\x45\x48\xb9\x32\x8f\x0c\xa5\x25\x7f\x85\xe5\xba\x24\xc1\xb2\x29\x40\x45\xb3\x1e\xcd\x82\x65\xba\xb9\x1e\x10\x35\xfc\x43\x76\xae\x36\xda\x19\xfb\x8f\xe9\x1d\x5d\xff\x8a\xdf\xd1\xe3\x7f\x10\x3c\x8e\x6a\x18\x76\x1f\x86\xa5\x79\xd2\xc1\x41\xde\xff\x30\x08\x72\x82\x24\xf7\xff\x69\x9d\x33\x1b\x7c\x8d\xc6\xf3\xa1\xf7\xe0\x55\xe4\x03\x83\x3d\xa3\x38\xe3\x84\x06\x85\xeb\xbb\x11\xf3\x20\xf8\x16\x00\x00\xff\xff\x84\x8f\x50\xb2\x94\x05\x00\x00")
+
+func migrations62_claimable_balance_claimantsSqlBytes() ([]byte, error) {
+	return bindataRead(
+		_migrations62_claimable_balance_claimantsSql,
+		"migrations/62_claimable_balance_claimants.sql",
+	)
+}
+
+func migrations62_claimable_balance_claimantsSql() (*asset, error) {
+	bytes, err := migrations62_claimable_balance_claimantsSqlBytes()
+	if err != nil {
+		return nil, err
+	}
+
+	info := bindataFileInfo{name: "migrations/62_claimable_balance_claimants.sql", size: 0, mode: os.FileMode(0), modTime: time.Unix(0, 0)}
+	a := &asset{bytes: bytes, info: info, digest: [32]uint8{0x88, 0x4b, 0xc2, 0xa2, 0x4c, 0x82, 0x68, 0xe8, 0xa8, 0x71, 0x70, 0x49, 0x44, 0x59, 0x8f, 0xd7, 0xa1, 0xc8, 0x31, 0x86, 0xf1, 0x5b, 0x30, 0x67, 0xdd, 0x15, 0x2c, 0x54, 0xfe, 0xe4, 0xe7, 0x3e}}
+	return a, nil
+}
+
 var _migrations6_create_assets_tableSql = []byte("\x1f\x8b\x08\x00\x00\x00\x00\x00\x00\xff\x6c\x90\x3d\x4f\xc3\x30\x18\x84\x77\xff\x8a\x1b\x1d\x91\x0e\x20\xe8\x92\xc9\x34\x16\x58\x18\xa7\xb8\x31\xa2\x53\xe5\x26\x16\x78\x80\x54\xb6\x11\xca\xbf\x47\xaa\x28\xf9\x50\xe6\x7b\xf4\xbc\xef\xdd\x6a\x85\xab\x4f\xff\x1e\x6c\x72\x30\x27\xb2\xd1\x9c\xd5\x1c\x35\xbb\x97\x1c\x1f\x3e\xa6\x2e\xf4\x07\x1b\xa3\x4b\x11\x94\x00\x80\x6f\xb1\xe3\x5a\x30\x89\xad\x16\xcf\x4c\xef\xf1\xc4\xf7\xc8\xcf\xd9\x19\x3c\xa4\xfe\xe4\xf0\xca\xf4\xe6\x91\x69\xba\xbe\xcd\xa0\xaa\x1a\xca\x48\x39\x86\x9a\xae\x1d\xa0\xeb\x9b\x65\xc8\xc7\xf8\xed\xc2\x3f\x76\xb7\x9e\x63\x46\x89\x17\xc3\xe9\xa0\xcc\x47\x3f\xe4\x13\x4b\x46\xb2\x82\x5c\xfa\x09\x55\xf2\xb7\xbf\xf8\xd8\x5f\xee\x54\x6a\x5e\xd9\xec\x84\x7a\xc0\x31\x05\xe7\x40\x27\xb6\x82\x90\xf1\x74\x65\xf7\xf3\x45\x4a\x5d\x6d\x97\xa7\x6b\x6c\x6c\x6c\xeb\x8a\xdf\x00\x00\x00\xff\xff\xfb\x53\x3e\x81\x6e\x01\x00\x00")
 
 func migrations6_create_assets_tableSqlBytes() ([]byte, error) {
@@ -1540,6 +1561,7 @@ var _bindata = map[string]func() (*asset, error){
 	"migrations/5_create_trades_table.sql":                               migrations5_create_trades_tableSql,
 	"migrations/60_add_asset_id_indexes.sql":                             migrations60_add_asset_id_indexesSql,
 	"migrations/61_trust_lines_by_account_type_code_issuer.sql":          migrations61_trust_lines_by_account_type_code_issuerSql,
+	"migrations/62_claimable_balance_claimants.sql":                      migrations62_claimable_balance_claimantsSql,
 	"migrations/6_create_assets_table.sql":                               migrations6_create_assets_tableSql,
 	"migrations/7_modify_trades_table.sql":                               migrations7_modify_trades_tableSql,
 	"migrations/8_add_aggregators.sql":                                   migrations8_add_aggregatorsSql,
@@ -1649,6 +1671,7 @@ var _bintree = &bintree{nil, map[string]*bintree{
 		"5_create_trades_table.sql":                               &bintree{migrations5_create_trades_tableSql, map[string]*bintree{}},
 		"60_add_asset_id_indexes.sql":                             &bintree{migrations60_add_asset_id_indexesSql, map[string]*bintree{}},
 		"61_trust_lines_by_account_type_code_issuer.sql":          &bintree{migrations61_trust_lines_by_account_type_code_issuerSql, map[string]*bintree{}},
+		"62_claimable_balance_claimants.sql":                      &bintree{migrations62_claimable_balance_claimantsSql, map[string]*bintree{}},
 		"6_create_assets_table.sql":                               &bintree{migrations6_create_assets_tableSql, map[string]*bintree{}},
 		"7_modify_trades_table.sql":                               &bintree{migrations7_modify_trades_tableSql, map[string]*bintree{}},
 		"8_add_aggregators.sql":                                   &bintree{migrations8_add_aggregatorsSql, map[string]*bintree{}},
diff --git a/services/horizon/internal/db2/schema/migrations/62_claimable_balance_claimants.sql b/services/horizon/internal/db2/schema/migrations/62_claimable_balance_claimants.sql
new file mode 100644
index 00000000..140a7215
--- /dev/null
+++ b/services/horizon/internal/db2/schema/migrations/62_claimable_balance_claimants.sql
@@ -0,0 +1,31 @@
+-- +migrate Up
+
+CREATE TABLE claimable_balance_claimants (
+    id TEXT NOT NULL, -- ClaimableBalanceID in base64
+    destination character varying(56) NOT NULL,
+    last_modified_ledger integer NOT NULL,
+    PRIMARY KEY (id, destination)
+);
+
+CREATE INDEX "claimable_balance_claimants_by_destination_last_modified_ledger_id" ON claimable_balance_claimants USING btree (destination, last_modified_ledger, id);
+
+/* Supports "select * from claimable_balances where asset = ? order by last_modified_ledger, id" */
+CREATE INDEX "claimable_balances_by_asset_last_modified_ledger_id" ON claimable_balances USING btree (asset, last_modified_ledger, id);
+/* Remove index replaced by above */
+DROP INDEX "claimable_balances_by_asset";
+
+/* Supports "select * from claimable_balances where sponsor = ? order by last_modified_ledger, id" */
+CREATE INDEX "claimable_balances_by_sponsor_last_modified_ledger_id" ON claimable_balances USING btree (sponsor, last_modified_ledger, id);
+/* Remove index replaced by above */
+DROP INDEX "claimable_balances_by_sponsor";
+
+-- +migrate Down
+
+DROP TABLE claimable_balance_claimants cascade;
+
+CREATE INDEX "claimable_balances_by_asset" ON claimable_balances USING btree (asset);
+DROP INDEX "claimable_balances_by_asset_last_modified_ledger_id";
+
+CREATE INDEX "claimable_balances_by_sponsor" ON claimable_balances USING btree (sponsor);
+DROP INDEX "claimable_balances_by_sponsor_last_modified_ledger_id";
+
diff --git a/services/horizon/internal/ingest/main.go b/services/horizon/internal/ingest/main.go
index 465965de..2603a3d4 100644
--- a/services/horizon/internal/ingest/main.go
+++ b/services/horizon/internal/ingest/main.go
@@ -55,7 +55,9 @@ const (
 	// - 13: Trigger state rebuild to include more than just authorized assets.
 	// - 14: Trigger state rebuild to include claimable balances in the asset stats processor.
 	// - 15: Fixed bug in asset stat ingestion where clawback is enabled (#3846).
-	CurrentVersion = 15
+	// - 16: Extract claimants to a separate table for better performance of
+	//       claimable balances for claimant queries.
+	CurrentVersion = 16
 
 	// MaxDBConnections is the size of the postgres connection pool dedicated to Horizon ingestion:
 	//  * Ledger ingestion,
diff --git a/services/horizon/internal/ingest/processors/claimable_balances_change_processor.go b/services/horizon/internal/ingest/processors/claimable_balances_change_processor.go
index fae72f2c..a9299273 100644
--- a/services/horizon/internal/ingest/processors/claimable_balances_change_processor.go
+++ b/services/horizon/internal/ingest/processors/claimable_balances_change_processor.go
@@ -10,9 +10,10 @@ import (
 )
 
 type ClaimableBalancesChangeProcessor struct {
-	encodingBuffer     *xdr.EncodingBuffer
-	qClaimableBalances history.QClaimableBalances
-	cache              *ingest.ChangeCompactor
+	encodingBuffer         *xdr.EncodingBuffer
+	qClaimableBalances     history.QClaimableBalances
+	cache                  *ingest.ChangeCompactor
+	claimantsInsertBuilder history.ClaimableBalanceClaimantBatchInsertBuilder
 }
 
 func NewClaimableBalancesChangeProcessor(Q history.QClaimableBalances) *ClaimableBalancesChangeProcessor {
@@ -26,6 +27,7 @@ func NewClaimableBalancesChangeProcessor(Q history.QClaimableBalances) *Claimabl
 
 func (p *ClaimableBalancesChangeProcessor) reset() {
 	p.cache = ingest.NewChangeCompactor()
+	p.claimantsInsertBuilder = p.qClaimableBalances.NewClaimableBalanceClaimantBatchInsertBuilder(maxBatchSize)
 }
 
 func (p *ClaimableBalancesChangeProcessor) ProcessChange(ctx context.Context, change ingest.Change) error {
@@ -86,6 +88,25 @@ func (p *ClaimableBalancesChangeProcessor) Commit(ctx context.Context) error {
 		if err := p.qClaimableBalances.UpsertClaimableBalances(ctx, cbsToUpsert); err != nil {
 			return errors.Wrap(err, "error executing upsert")
 		}
+
+		// Add ClaimableBalanceClaimants
+		for _, cb := range cbsToUpsert {
+			for _, claimant := range cb.Claimants {
+				claimant := history.ClaimableBalanceClaimant{
+					BalanceID:          cb.BalanceID,
+					Destination:        claimant.Destination,
+					LastModifiedLedger: cb.LastModifiedLedger,
+				}
+				if err := p.claimantsInsertBuilder.Add(ctx, claimant); err != nil {
+					return errors.Wrap(err, "error adding to claimantsInsertBuilder")
+				}
+			}
+		}
+
+		err := p.claimantsInsertBuilder.Exec(ctx)
+		if err != nil {
+			return errors.Wrap(err, "error executing claimantsInsertBuilder")
+		}
 	}
 
 	if len(cbIDsToDelete) > 0 {
@@ -100,6 +121,12 @@ func (p *ClaimableBalancesChangeProcessor) Commit(ctx context.Context) error {
 				len(cbIDsToDelete),
 			))
 		}
+
+		// Remove ClaimableBalanceClaimants
+		_, err = p.qClaimableBalances.RemoveClaimableBalanceClaimants(ctx, cbIDsToDelete)
+		if err != nil {
+			return errors.Wrap(err, "error executing removal of claimants")
+		}
 	}
 
 	return nil
diff --git a/services/horizon/internal/ingest/verify.go b/services/horizon/internal/ingest/verify.go
index 18af7b33..43a80da0 100644
--- a/services/horizon/internal/ingest/verify.go
+++ b/services/horizon/internal/ingest/verify.go
@@ -29,7 +29,7 @@ const assetStatsBatchSize = 500
 // check them.
 // There is a test that checks it, to fix it: update the actual `verifyState`
 // method instead of just updating this value!
-const stateVerifierExpectedIngestionVersion = 15
+const stateVerifierExpectedIngestionVersion = 16
 
 // verifyState is called as a go routine from pipeline post hook every 64
 // ledgers. It checks if the state is correct. If another go routine is already
@@ -672,6 +672,11 @@ func addClaimableBalanceToStateVerifier(
 		return errors.Wrap(err, "Error running history.Q.GetClaimableBalancesByID")
 	}
 
+	cBalancesClaimants, err := q.GetClaimantsByClaimableBalances(ctx, idStrings)
+	if err != nil {
+		return errors.Wrap(err, "Error running history.Q.GetClaimantsByClaimableBalances")
+	}
+
 	for _, row := range cBalances {
 		claimants := []xdr.Claimant{}
 		for _, claimant := range row.Claimants {
@@ -684,6 +689,31 @@ func addClaimableBalanceToStateVerifier(
 			})
 		}
 		claimants = xdr.SortClaimantsByDestination(claimants)
+
+		// Check if balances in claimable_balance_claimants table match.
+		if len(claimants) != len(cBalancesClaimants[row.BalanceID]) {
+			return ingest.NewStateError(
+				fmt.Errorf(
+					"claimable_balance_claimants length (%d) for claimants doesn't match claimable_balance table (%d)",
+					len(cBalancesClaimants[row.BalanceID]), len(claimants),
+				),
+			)
+		}
+
+		for i, claimant := range claimants {
+			if claimant.MustV0().Destination.Address() != cBalancesClaimants[row.BalanceID][i].Destination ||
+				row.LastModifiedLedger != cBalancesClaimants[row.BalanceID][i].LastModifiedLedger {
+				return fmt.Errorf(
+					"claimable_balance_claimants table for balance %s does not match. expectedDestination=%s actualDestination=%s, expectedLastModifiedLedger=%d actualLastModifiedLedger=%d",
+					row.BalanceID,
+					claimant.MustV0().Destination.Address(),
+					cBalancesClaimants[row.BalanceID][i].Destination,
+					row.LastModifiedLedger,
+					cBalancesClaimants[row.BalanceID][i].LastModifiedLedger,
+				)
+			}
+		}
+
 		var balanceID xdr.ClaimableBalanceId
 		if err := xdr.SafeUnmarshalHex(row.BalanceID, &balanceID); err != nil {
 			return err


Test patch:
diff --git a/services/horizon/internal/actions/claimable_balance_test.go b/services/horizon/internal/actions/claimable_balance_test.go
index ad6bcd57..ff59b961 100644
--- a/services/horizon/internal/actions/claimable_balance_test.go
+++ b/services/horizon/internal/actions/claimable_balance_test.go
@@ -46,6 +46,17 @@ func TestGetClaimableBalanceByID(t *testing.T) {
 	err = q.UpsertClaimableBalances(tt.Ctx, []history.ClaimableBalance{cBalance})
 	tt.Assert.NoError(err)
 
+	claimantsInsertBuilder := q.NewClaimableBalanceClaimantBatchInsertBuilder(10)
+	for _, claimant := range cBalance.Claimants {
+		claimant := history.ClaimableBalanceClaimant{
+			BalanceID:          cBalance.BalanceID,
+			Destination:        claimant.Destination,
+			LastModifiedLedger: cBalance.LastModifiedLedger,
+		}
+		err = claimantsInsertBuilder.Add(tt.Ctx, claimant)
+		tt.Assert.NoError(err)
+	}
+
 	handler := GetClaimableBalanceByIDHandler{}
 	response, err := handler.GetResource(httptest.NewRecorder(), makeRequest(
 		t,
@@ -179,6 +190,23 @@ func TestGetClaimableBalances(t *testing.T) {
 	err := q.UpsertClaimableBalances(tt.Ctx, hCBs)
 	tt.Assert.NoError(err)
 
+	claimantsInsertBuilder := q.NewClaimableBalanceClaimantBatchInsertBuilder(10)
+
+	for _, cBalance := range hCBs {
+		for _, claimant := range cBalance.Claimants {
+			claimant := history.ClaimableBalanceClaimant{
+				BalanceID:          cBalance.BalanceID,
+				Destination:        claimant.Destination,
+				LastModifiedLedger: cBalance.LastModifiedLedger,
+			}
+			err = claimantsInsertBuilder.Add(tt.Ctx, claimant)
+			tt.Assert.NoError(err)
+		}
+	}
+
+	err = claimantsInsertBuilder.Exec(tt.Ctx)
+	tt.Assert.NoError(err)
+
 	handler := GetClaimableBalancesHandler{}
 	response, err := handler.GetResourcePage(httptest.NewRecorder(), makeRequest(
 		t,
@@ -291,6 +319,21 @@ func TestGetClaimableBalances(t *testing.T) {
 	err = q.UpsertClaimableBalances(tt.Ctx, hCBs)
 	tt.Assert.NoError(err)
 
+	for _, cBalance := range hCBs {
+		for _, claimant := range cBalance.Claimants {
+			claimant := history.ClaimableBalanceClaimant{
+				BalanceID:          cBalance.BalanceID,
+				Destination:        claimant.Destination,
+				LastModifiedLedger: cBalance.LastModifiedLedger,
+			}
+			err = claimantsInsertBuilder.Add(tt.Ctx, claimant)
+			tt.Assert.NoError(err)
+		}
+	}
+
+	err = claimantsInsertBuilder.Exec(tt.Ctx)
+	tt.Assert.NoError(err)
+
 	response, err = handler.GetResourcePage(httptest.NewRecorder(), makeRequest(
 		t,
 		map[string]string{},
diff --git a/services/horizon/internal/db2/history/claimable_balances_test.go b/services/horizon/internal/db2/history/claimable_balances_test.go
index 9321e635..49cc722f 100644
--- a/services/horizon/internal/db2/history/claimable_balances_test.go
+++ b/services/horizon/internal/db2/history/claimable_balances_test.go
@@ -58,6 +58,55 @@ func TestRemoveClaimableBalance(t *testing.T) {
 	}
 }
 
+func TestRemoveClaimableBalanceClaimants(t *testing.T) {
+	tt := test.Start(t)
+	defer tt.Finish()
+	test.ResetHorizonDB(t, tt.HorizonDB)
+	q := &Q{tt.HorizonSession()}
+
+	accountID := "GC3C4AKRBQLHOJ45U4XG35ESVWRDECWO5XLDGYADO6DPR3L7KIDVUMML"
+	asset := xdr.MustNewCreditAsset("USD", accountID)
+	balanceID := xdr.ClaimableBalanceId{
+		Type: xdr.ClaimableBalanceIdTypeClaimableBalanceIdTypeV0,
+		V0:   &xdr.Hash{1, 2, 3},
+	}
+	id, err := xdr.MarshalHex(balanceID)
+	tt.Assert.NoError(err)
+	cBalance := ClaimableBalance{
+		BalanceID: id,
+		Claimants: []Claimant{
+			{
+				Destination: accountID,
+				Predicate: xdr.ClaimPredicate{
+					Type: xdr.ClaimPredicateTypeClaimPredicateUnconditional,
+				},
+			},
+		},
+		Asset:              asset,
+		LastModifiedLedger: 123,
+		Amount:             10,
+	}
+
+	claimantsInsertBuilder := q.NewClaimableBalanceClaimantBatchInsertBuilder(10)
+
+	for _, claimant := range cBalance.Claimants {
+		claimant := ClaimableBalanceClaimant{
+			BalanceID:          cBalance.BalanceID,
+			Destination:        claimant.Destination,
+			LastModifiedLedger: cBalance.LastModifiedLedger,
+		}
+		err = claimantsInsertBuilder.Add(tt.Ctx, claimant)
+		tt.Assert.NoError(err)
+	}
+
+	err = claimantsInsertBuilder.Exec(tt.Ctx)
+	tt.Assert.NoError(err)
+
+	removed, err := q.RemoveClaimableBalanceClaimants(tt.Ctx, []string{id})
+	tt.Assert.NoError(err)
+	tt.Assert.Equal(int64(1), removed)
+}
+
 func TestFindClaimableBalancesByDestination(t *testing.T) {
 	tt := test.Start(t)
 	defer tt.Finish()
@@ -92,6 +141,17 @@ func TestFindClaimableBalancesByDestination(t *testing.T) {
 	err = q.UpsertClaimableBalances(tt.Ctx, []ClaimableBalance{cBalance})
 	tt.Assert.NoError(err)
 
+	claimantsInsertBuilder := q.NewClaimableBalanceClaimantBatchInsertBuilder(10)
+	for _, claimant := range cBalance.Claimants {
+		claimant := ClaimableBalanceClaimant{
+			BalanceID:          cBalance.BalanceID,
+			Destination:        claimant.Destination,
+			LastModifiedLedger: cBalance.LastModifiedLedger,
+		}
+		err = claimantsInsertBuilder.Add(tt.Ctx, claimant)
+		tt.Assert.NoError(err)
+	}
+
 	balanceID = xdr.ClaimableBalanceId{
 		Type: xdr.ClaimableBalanceIdTypeClaimableBalanceIdTypeV0,
 		V0:   &xdr.Hash{3, 2, 1},
@@ -122,6 +182,19 @@ func TestFindClaimableBalancesByDestination(t *testing.T) {
 	err = q.UpsertClaimableBalances(tt.Ctx, []ClaimableBalance{cBalance})
 	tt.Assert.NoError(err)
 
+	for _, claimant := range cBalance.Claimants {
+		claimant := ClaimableBalanceClaimant{
+			BalanceID:          cBalance.BalanceID,
+			Destination:        claimant.Destination,
+			LastModifiedLedger: cBalance.LastModifiedLedger,
+		}
+		err = claimantsInsertBuilder.Add(tt.Ctx, claimant)
+		tt.Assert.NoError(err)
+	}
+
+	err = claimantsInsertBuilder.Exec(tt.Ctx)
+	tt.Assert.NoError(err)
+
 	query := ClaimableBalancesQuery{
 		PageQuery: db2.MustPageQuery("", false, "", 10),
 		Claimant:  xdr.MustAddressPtr(dest1),
diff --git a/services/horizon/internal/ingest/processor_runner_test.go b/services/horizon/internal/ingest/processor_runner_test.go
index ea2a99ef..f1fa7747 100644
--- a/services/horizon/internal/ingest/processor_runner_test.go
+++ b/services/horizon/internal/ingest/processor_runner_test.go
@@ -48,6 +48,9 @@ func TestProcessorRunnerRunHistoryArchiveIngestionGenesis(t *testing.T) {
 	q.MockQSigners.On("NewAccountSignersBatchInsertBuilder", maxBatchSize).
 		Return(mockAccountSignersBatchInsertBuilder).Once()
 
+	q.MockQClaimableBalances.On("NewClaimableBalanceClaimantBatchInsertBuilder", maxBatchSize).
+		Return(&history.MockClaimableBalanceClaimantBatchInsertBuilder{}).Once()
+
 	q.MockQAssetStats.On("InsertAssetStats", ctx, []history.ExpAssetStat{}, 100000).
 		Return(nil)
 
@@ -112,6 +115,8 @@ func TestProcessorRunnerRunHistoryArchiveIngestionHistoryArchive(t *testing.T) {
 	mockAccountSignersBatchInsertBuilder.On("Exec", ctx).Return(nil).Once()
 	q.MockQSigners.On("NewAccountSignersBatchInsertBuilder", maxBatchSize).
 		Return(mockAccountSignersBatchInsertBuilder).Once()
+	q.MockQClaimableBalances.On("NewClaimableBalanceClaimantBatchInsertBuilder", maxBatchSize).
+		Return(&history.MockClaimableBalanceClaimantBatchInsertBuilder{}).Once()
 
 	q.MockQAssetStats.On("InsertAssetStats", ctx, []history.ExpAssetStat{}, 100000).
 		Return(nil)
@@ -147,6 +152,8 @@ func TestProcessorRunnerRunHistoryArchiveIngestionProtocolVersionNotSupported(t
 	defer mock.AssertExpectationsForObjects(t, mockAccountSignersBatchInsertBuilder)
 	q.MockQSigners.On("NewAccountSignersBatchInsertBuilder", maxBatchSize).
 		Return(mockAccountSignersBatchInsertBuilder).Once()
+	q.MockQClaimableBalances.On("NewClaimableBalanceClaimantBatchInsertBuilder", maxBatchSize).
+		Return(&history.MockClaimableBalanceClaimantBatchInsertBuilder{}).Once()
 
 	q.MockQAssetStats.On("InsertAssetStats", ctx, []history.ExpAssetStat{}, 100000).
 		Return(nil)
@@ -178,6 +185,8 @@ func TestProcessorRunnerBuildChangeProcessor(t *testing.T) {
 	// Twice = checking ledgerSource and historyArchiveSource
 	q.MockQSigners.On("NewAccountSignersBatchInsertBuilder", maxBatchSize).
 		Return(&history.MockAccountSignersBatchInsertBuilder{}).Twice()
+	q.MockQClaimableBalances.On("NewClaimableBalanceClaimantBatchInsertBuilder", maxBatchSize).
+		Return(&history.MockClaimableBalanceClaimantBatchInsertBuilder{}).Twice()
 	runner := ProcessorRunner{
 		ctx:      ctx,
 		historyQ: q,
@@ -233,6 +242,8 @@ func TestProcessorRunnerBuildTransactionProcessor(t *testing.T) {
 		Return(&history.MockOperationsBatchInsertBuilder{}).Twice() // Twice = with/without failed
 	q.MockQTransactions.On("NewTransactionBatchInsertBuilder", maxBatchSize).
 		Return(&history.MockTransactionsBatchInsertBuilder{}).Twice()
+	q.MockQClaimableBalances.On("NewClaimableBalanceClaimantBatchInsertBuilder", maxBatchSize).
+		Return(&history.MockClaimableBalanceClaimantBatchInsertBuilder{}).Twice()
 
 	runner := ProcessorRunner{
 		ctx:      ctx,
@@ -299,6 +310,9 @@ func TestProcessorRunnerWithFilterEnabled(t *testing.T) {
 	q.MockQTransactions.On("NewTransactionFilteredTmpBatchInsertBuilder", maxBatchSize).
 		Return(mockTransactionsBatchInsertBuilder)
 
+	q.MockQClaimableBalances.On("NewClaimableBalanceClaimantBatchInsertBuilder", maxBatchSize).
+		Return(&history.MockClaimableBalanceClaimantBatchInsertBuilder{}).Once()
+
 	q.On("DeleteTransactionsFilteredTmpOlderThan", ctx, mock.AnythingOfType("uint64")).
 		Return(int64(0), nil)
 
@@ -355,6 +369,9 @@ func TestProcessorRunnerRunAllProcessorsOnLedger(t *testing.T) {
 	q.MockQTransactions.On("NewTransactionBatchInsertBuilder", maxBatchSize).
 		Return(mockTransactionsBatchInsertBuilder).Twice()
 
+	q.MockQClaimableBalances.On("NewClaimableBalanceClaimantBatchInsertBuilder", maxBatchSize).
+		Return(&history.MockClaimableBalanceClaimantBatchInsertBuilder{}).Once()
+
 	q.MockQLedgers.On("InsertLedger", ctx, ledger.V0.LedgerHeader, 0, 0, 0, 0, CurrentVersion).
 		Return(int64(1), nil).Once()
 
diff --git a/services/horizon/internal/ingest/processors/claimable_balances_change_processor_test.go b/services/horizon/internal/ingest/processors/claimable_balances_change_processor_test.go
index 0ece2661..70cd43bb 100644
--- a/services/horizon/internal/ingest/processors/claimable_balances_change_processor_test.go
+++ b/services/horizon/internal/ingest/processors/claimable_balances_change_processor_test.go
@@ -20,14 +20,19 @@ func TestClaimableBalancesChangeProcessorTestSuiteState(t *testing.T) {
 
 type ClaimableBalancesChangeProcessorTestSuiteState struct {
 	suite.Suite
-	ctx       context.Context
-	processor *ClaimableBalancesChangeProcessor
-	mockQ     *history.MockQClaimableBalances
+	ctx                    context.Context
+	processor              *ClaimableBalancesChangeProcessor
+	mockQ                  *history.MockQClaimableBalances
+	mockBatchInsertBuilder *history.MockClaimableBalanceClaimantBatchInsertBuilder
 }
 
 func (s *ClaimableBalancesChangeProcessorTestSuiteState) SetupTest() {
 	s.ctx = context.Background()
+	s.mockBatchInsertBuilder = &history.MockClaimableBalanceClaimantBatchInsertBuilder{}
 	s.mockQ = &history.MockQClaimableBalances{}
+	s.mockQ.
+		On("NewClaimableBalanceClaimantBatchInsertBuilder", maxBatchSize).
+		Return(s.mockBatchInsertBuilder).Once()
 
 	s.processor = NewClaimableBalancesChangeProcessor(s.mockQ)
 }
@@ -50,22 +55,40 @@ func (s *ClaimableBalancesChangeProcessorTestSuiteState) TestCreatesClaimableBal
 
 	cBalance := xdr.ClaimableBalanceEntry{
 		BalanceId: balanceID,
-		Claimants: []xdr.Claimant{},
-		Asset:     xdr.MustNewCreditAsset("USD", "GC3C4AKRBQLHOJ45U4XG35ESVWRDECWO5XLDGYADO6DPR3L7KIDVUMML"),
-		Amount:    10,
+		Claimants: []xdr.Claimant{
+			{
+				V0: &xdr.ClaimantV0{
+					Destination: xdr.MustAddress("GC3C4AKRBQLHOJ45U4XG35ESVWRDECWO5XLDGYADO6DPR3L7KIDVUMML"),
+				},
+			},
+		},
+		Asset:  xdr.MustNewCreditAsset("USD", "GC3C4AKRBQLHOJ45U4XG35ESVWRDECWO5XLDGYADO6DPR3L7KIDVUMML"),
+		Amount: 10,
 	}
 	id, err := xdr.MarshalHex(balanceID)
 	s.Assert().NoError(err)
 	s.mockQ.On("UpsertClaimableBalances", s.ctx, []history.ClaimableBalance{
 		{
-			BalanceID:          id,
-			Claimants:          []history.Claimant{},
+			BalanceID: id,
+			Claimants: []history.Claimant{
+				{
+					Destination: "GC3C4AKRBQLHOJ45U4XG35ESVWRDECWO5XLDGYADO6DPR3L7KIDVUMML",
+				},
+			},
 			Asset:              cBalance.Asset,
 			Amount:             cBalance.Amount,
 			LastModifiedLedger: uint32(lastModifiedLedgerSeq),
 		},
 	}).Return(nil).Once()
 
+	s.mockBatchInsertBuilder.On("Add", s.ctx, history.ClaimableBalanceClaimant{
+		BalanceID:          id,
+		Destination:        "GC3C4AKRBQLHOJ45U4XG35ESVWRDECWO5XLDGYADO6DPR3L7KIDVUMML",
+		LastModifiedLedger: uint32(lastModifiedLedgerSeq),
+	}).Return(nil).Once()
+
+	s.mockBatchInsertBuilder.On("Exec", s.ctx).Return(nil).Once()
+
 	err = s.processor.ProcessChange(s.ctx, ingest.Change{
 		Type: xdr.LedgerEntryTypeClaimableBalance,
 		Pre:  nil,
@@ -86,14 +109,19 @@ func TestClaimableBalancesChangeProcessorTestSuiteLedger(t *testing.T) {
 
 type ClaimableBalancesChangeProcessorTestSuiteLedger struct {
 	suite.Suite
-	ctx       context.Context
-	processor *ClaimableBalancesChangeProcessor
-	mockQ     *history.MockQClaimableBalances
+	ctx                    context.Context
+	processor              *ClaimableBalancesChangeProcessor
+	mockQ                  *history.MockQClaimableBalances
+	mockBatchInsertBuilder *history.MockClaimableBalanceClaimantBatchInsertBuilder
 }
 
 func (s *ClaimableBalancesChangeProcessorTestSuiteLedger) SetupTest() {
 	s.ctx = context.Background()
+	s.mockBatchInsertBuilder = &history.MockClaimableBalanceClaimantBatchInsertBuilder{}
 	s.mockQ = &history.MockQClaimableBalances{}
+	s.mockQ.
+		On("NewClaimableBalanceClaimantBatchInsertBuilder", maxBatchSize).
+		Return(s.mockBatchInsertBuilder).Once()
 
 	s.processor = NewClaimableBalancesChangeProcessor(s.mockQ)
 }
@@ -132,6 +160,8 @@ func (s *ClaimableBalancesChangeProcessorTestSuiteLedger) TestNewClaimableBalanc
 			},
 		},
 	}
+	s.mockBatchInsertBuilder.On("Exec", s.ctx).Return(nil).Once()
+
 	err := s.processor.ProcessChange(s.ctx, ingest.Change{
 		Type: xdr.LedgerEntryTypeClaimableBalance,
 		Pre:  nil,
@@ -222,6 +252,7 @@ func (s *ClaimableBalancesChangeProcessorTestSuiteLedger) TestUpdateClaimableBal
 			},
 		},
 	}
+	s.mockBatchInsertBuilder.On("Exec", s.ctx).Return(nil).Once()
 
 	err := s.processor.ProcessChange(s.ctx, ingest.Change{
 		Type: xdr.LedgerEntryTypeClaimableBalance,
@@ -287,4 +318,10 @@ func (s *ClaimableBalancesChangeProcessorTestSuiteLedger) TestRemoveClaimableBal
 		s.ctx,
 		[]string{id},
 	).Return(int64(1), nil).Once()
+
+	s.mockQ.On(
+		"RemoveClaimableBalanceClaimants",
+		s.ctx,
+		[]string{id},
+	).Return(int64(1), nil).Once()
 }
diff --git a/services/horizon/internal/ingest/verify_range_state_test.go b/services/horizon/internal/ingest/verify_range_state_test.go
index 75298db9..703b7447 100644
--- a/services/horizon/internal/ingest/verify_range_state_test.go
+++ b/services/horizon/internal/ingest/verify_range_state_test.go
@@ -573,6 +573,18 @@ func (s *VerifyRangeStateTestSuite) TestSuccessWithVerify() {
 		On("GetClaimableBalancesByID", s.ctx, []string{balanceIDStr}).
 		Return([]history.ClaimableBalance{claimableBalance}, nil).Once()
 
+	clonedQ.MockQClaimableBalances.
+		On("GetClaimantsByClaimableBalances", s.ctx, []string{balanceIDStr}).
+		Return(map[string][]history.ClaimableBalanceClaimant{
+			claimableBalance.BalanceID: {
+				{
+					BalanceID:          claimableBalance.BalanceID,
+					Destination:        claimableBalance.Claimants[0].Destination,
+					LastModifiedLedger: claimableBalance.LastModifiedLedger,
+				},
+			},
+		}, nil).Once()
+
 	clonedQ.MockQLiquidityPools.On("CountLiquidityPools", s.ctx).Return(1, nil).Once()
 	clonedQ.MockQLiquidityPools.
 		On("GetLiquidityPoolsByID", s.ctx, []string{liquidityPool.PoolID}).