Repository: stellar/go
Pull request #5113 (2023-11-14 23:37:55)

services/horizon: Use COPY for inserting into trust_lines table
<!-- 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

Optimize the `TrustLinesProcessor` to use `FastBatchInsertBuilder` which uses COPY to bulk insert into the database table `trust_lines`. Database update and removal operations remain unchanged.


### Why

#5098 

### Known limitations
N/A


Patch:
diff --git a/services/horizon/internal/db2/history/main.go b/services/horizon/internal/db2/history/main.go
index 850b8bef..7c05f2df 100644
--- a/services/horizon/internal/db2/history/main.go
+++ b/services/horizon/internal/db2/history/main.go
@@ -846,6 +846,7 @@ type QTrustLines interface {
 	GetTrustLinesByKeys(ctx context.Context, ledgerKeys []string) ([]TrustLine, error)
 	UpsertTrustLines(ctx context.Context, trustlines []TrustLine) error
 	RemoveTrustLines(ctx context.Context, ledgerKeys []string) (int64, error)
+	NewTrustLinesBatchInsertBuilder() TrustLinesBatchInsertBuilder
 }
 
 func (q *Q) NewAccountSignersBatchInsertBuilder() AccountSignersBatchInsertBuilder {
diff --git a/services/horizon/internal/db2/history/mock_q_trust_lines.go b/services/horizon/internal/db2/history/mock_q_trust_lines.go
index f6b5b530..a074b0e7 100644
--- a/services/horizon/internal/db2/history/mock_q_trust_lines.go
+++ b/services/horizon/internal/db2/history/mock_q_trust_lines.go
@@ -25,3 +25,8 @@ func (m *MockQTrustLines) RemoveTrustLines(ctx context.Context, ledgerKeys []str
 	a := m.Called(ctx, ledgerKeys)
 	return a.Get(0).(int64), a.Error(1)
 }
+
+func (m *MockQTrustLines) NewTrustLinesBatchInsertBuilder() TrustLinesBatchInsertBuilder {
+	a := m.Called()
+	return a.Get(0).(TrustLinesBatchInsertBuilder)
+}
diff --git a/services/horizon/internal/db2/history/mock_trust_lines_batch_insert_builder.go b/services/horizon/internal/db2/history/mock_trust_lines_batch_insert_builder.go
new file mode 100644
index 00000000..38e1b41d
--- /dev/null
+++ b/services/horizon/internal/db2/history/mock_trust_lines_batch_insert_builder.go
@@ -0,0 +1,21 @@
+package history
+
+import (
+	"context"
+
+	"github.com/stretchr/testify/mock"
+)
+
+type MockTrustLinesBatchInsertBuilder struct {
+	mock.Mock
+}
+
+func (m *MockTrustLinesBatchInsertBuilder) Add(line TrustLine) error {
+	a := m.Called(line)
+	return a.Error(0)
+}
+
+func (m *MockTrustLinesBatchInsertBuilder) Exec(ctx context.Context) error {
+	a := m.Called(ctx)
+	return a.Error(0)
+}
diff --git a/services/horizon/internal/db2/history/trust_lines_batch_insert_builder.go b/services/horizon/internal/db2/history/trust_lines_batch_insert_builder.go
new file mode 100644
index 00000000..2c774697
--- /dev/null
+++ b/services/horizon/internal/db2/history/trust_lines_batch_insert_builder.go
@@ -0,0 +1,39 @@
+package history
+
+import (
+	"context"
+
+	"github.com/stellar/go/support/db"
+)
+
+// TrustLinesBatchInsertBuilder is used to insert trustlines into the trust_lines table
+type TrustLinesBatchInsertBuilder interface {
+	Add(line TrustLine) error
+	Exec(ctx context.Context) error
+}
+
+// trustLinesBatchInsertBuilder is a simple wrapper around db.FastBatchInsertBuilder
+type trustLinesBatchInsertBuilder struct {
+	session db.SessionInterface
+	builder db.FastBatchInsertBuilder
+	table   string
+}
+
+// NewTrustLinesBatchInsertBuilder constructs a new TrustLinesBatchInsertBuilder instance
+func (q *Q) NewTrustLinesBatchInsertBuilder() TrustLinesBatchInsertBuilder {
+	return &trustLinesBatchInsertBuilder{
+		session: q,
+		builder: db.FastBatchInsertBuilder{},
+		table:   "trust_lines",
+	}
+}
+
+// Add adds a new trustline to the batch
+func (i *trustLinesBatchInsertBuilder) Add(line TrustLine) error {
+	return i.builder.RowStruct(line)
+}
+
+// Exec writes the batch of trust lines to the database.
+func (i *trustLinesBatchInsertBuilder) Exec(ctx context.Context) error {
+	return i.builder.Exec(ctx, i.session, i.table)
+}
diff --git a/services/horizon/internal/ingest/processors/trust_lines_processor.go b/services/horizon/internal/ingest/processors/trust_lines_processor.go
index f8c9ca92..3152ef0d 100644
--- a/services/horizon/internal/ingest/processors/trust_lines_processor.go
+++ b/services/horizon/internal/ingest/processors/trust_lines_processor.go
@@ -12,7 +12,8 @@ import (
 type TrustLinesProcessor struct {
 	trustLinesQ history.QTrustLines
 
-	cache *ingest.ChangeCompactor
+	cache              *ingest.ChangeCompactor
+	batchInsertBuilder history.TrustLinesBatchInsertBuilder
 }
 
 func NewTrustLinesProcessor(trustLinesQ history.QTrustLines) *TrustLinesProcessor {
@@ -23,6 +24,7 @@ func NewTrustLinesProcessor(trustLinesQ history.QTrustLines) *TrustLinesProcesso
 
 func (p *TrustLinesProcessor) reset() {
 	p.cache = ingest.NewChangeCompactor()
+	p.batchInsertBuilder = p.trustLinesQ.NewTrustLinesBatchInsertBuilder()
 }
 
 func (p *TrustLinesProcessor) ProcessChange(ctx context.Context, change ingest.Change) error {
@@ -40,7 +42,6 @@ func (p *TrustLinesProcessor) ProcessChange(ctx context.Context, change ingest.C
 		if err != nil {
 			return errors.Wrap(err, "error in Commit")
 		}
-		p.reset()
 	}
 
 	return nil
@@ -97,18 +98,31 @@ func xdrToTrustline(ledgerEntry xdr.LedgerEntry) (history.TrustLine, error) {
 }
 
 func (p *TrustLinesProcessor) Commit(ctx context.Context) error {
+	defer p.reset()
+
 	var batchUpsertTrustLines []history.TrustLine
 	var batchRemoveTrustLineKeys []string
 
 	changes := p.cache.GetChanges()
 	for _, change := range changes {
 		switch {
-		case change.Post != nil:
-			tl, err := xdrToTrustline(*change.Post)
+		case change.Pre == nil && change.Post != nil:
+			// Created
+			line, err := xdrToTrustline(*change.Post)
 			if err != nil {
 				return errors.Wrap(err, "Error extracting trustline")
 			}
 
+			err = p.batchInsertBuilder.Add(line)
+			if err != nil {
+				return errors.Wrap(err, "Error adding to TrustLinesBatchInsertBuilder")
+			}
+		case change.Pre != nil && change.Post != nil:
+			// Updated
+			tl, err := xdrToTrustline(*change.Post)
+			if err != nil {
+				return errors.Wrap(err, "Error extracting trustline")
+			}
 			batchUpsertTrustLines = append(batchUpsertTrustLines, tl)
 		case change.Pre != nil && change.Post == nil:
 			// Removed
@@ -124,6 +138,11 @@ func (p *TrustLinesProcessor) Commit(ctx context.Context) error {
 		}
 	}
 
+	err := p.batchInsertBuilder.Exec(ctx)
+	if err != nil {
+		return errors.Wrap(err, "Error executing TrustLinesBatchInsertBuilder")
+	}
+
 	if len(batchUpsertTrustLines) > 0 {
 		err := p.trustLinesQ.UpsertTrustLines(ctx, batchUpsertTrustLines)
 		if err != nil {


Test patch:
diff --git a/services/horizon/internal/ingest/processor_runner_test.go b/services/horizon/internal/ingest/processor_runner_test.go
index b932a051..7dfea071 100644
--- a/services/horizon/internal/ingest/processor_runner_test.go
+++ b/services/horizon/internal/ingest/processor_runner_test.go
@@ -523,10 +523,18 @@ func mockChangeProcessorBatchBuilders(q *mockDBQ, ctx context.Context, mockExec
 	q.MockQOffers.On("NewOffersBatchInsertBuilder").
 		Return(mockOfferBatchInsertBuilder).Twice()
 
+	mockTrustLinesBatchInsertBuilder := &history.MockTrustLinesBatchInsertBuilder{}
+	if mockExec {
+		mockTrustLinesBatchInsertBuilder.On("Exec", ctx).Return(nil).Once()
+	}
+	q.MockQTrustLines.On("NewTrustLinesBatchInsertBuilder").
+		Return(mockTrustLinesBatchInsertBuilder)
+
 	return []interface{}{mockAccountSignersBatchInsertBuilder,
 		mockClaimableBalanceBatchInsertBuilder,
 		mockClaimableBalanceClaimantBatchInsertBuilder,
 		mockLiquidityPoolBatchInsertBuilder,
 		mockOfferBatchInsertBuilder,
+		mockTrustLinesBatchInsertBuilder,
 	}
 }
diff --git a/services/horizon/internal/ingest/processors/trust_lines_processor_test.go b/services/horizon/internal/ingest/processors/trust_lines_processor_test.go
index 2e3d092f..07990a06 100644
--- a/services/horizon/internal/ingest/processors/trust_lines_processor_test.go
+++ b/services/horizon/internal/ingest/processors/trust_lines_processor_test.go
@@ -24,14 +24,20 @@ func TestTrustLinesProcessorTestSuiteState(t *testing.T) {
 
 type TrustLinesProcessorTestSuiteState struct {
 	suite.Suite
-	ctx       context.Context
-	processor *TrustLinesProcessor
-	mockQ     *history.MockQTrustLines
+	ctx                              context.Context
+	processor                        *TrustLinesProcessor
+	mockQ                            *history.MockQTrustLines
+	mockTrustLinesBatchInsertBuilder *history.MockTrustLinesBatchInsertBuilder
 }
 
 func (s *TrustLinesProcessorTestSuiteState) SetupTest() {
 	s.ctx = context.Background()
 	s.mockQ = &history.MockQTrustLines{}
+
+	s.mockTrustLinesBatchInsertBuilder = &history.MockTrustLinesBatchInsertBuilder{}
+	s.mockQ.On("NewTrustLinesBatchInsertBuilder").Return(s.mockTrustLinesBatchInsertBuilder).Twice()
+	s.mockTrustLinesBatchInsertBuilder.On("Exec", s.ctx).Return(nil).Once()
+
 	s.processor = NewTrustLinesProcessor(s.mockQ)
 }
 
@@ -85,42 +91,33 @@ func (s *TrustLinesProcessorTestSuiteState) TestCreateTrustLine() {
 	})
 	s.Assert().NoError(err)
 
-	s.mockQ.On(
-		"UpsertTrustLines",
-		s.ctx,
-		mock.AnythingOfType("[]history.TrustLine"),
-	).Run(func(args mock.Arguments) {
-		arg := args.Get(1).([]history.TrustLine)
-		s.Assert().ElementsMatch(
-			[]history.TrustLine{
-				{
-					AccountID:          trustLine.AccountId.Address(),
-					AssetType:          xdr.AssetTypeAssetTypeCreditAlphanum4,
-					AssetIssuer:        trustLineIssuer.Address(),
-					AssetCode:          "EUR",
-					Balance:            int64(trustLine.Balance),
-					LedgerKey:          "AAAAAQAAAAAdBJqAD9qPq+j2nRDdjdp5KVoUh8riPkNO9ato7BNs8wAAAAFFVVIAAAAAAGL8HQvQkbK2HA3WVjRrKmjX00fG8sLI7m0ERwJW/AX3",
-					Limit:              int64(trustLine.Limit),
-					LiquidityPoolID:    "",
-					BuyingLiabilities:  int64(trustLine.Liabilities().Buying),
-					SellingLiabilities: int64(trustLine.Liabilities().Selling),
-					Flags:              uint32(trustLine.Flags),
-					LastModifiedLedger: uint32(lastModifiedLedgerSeq),
-					Sponsor:            null.String{},
-				},
-				{
-					AccountID:          poolShareTrustLine.AccountId.Address(),
-					AssetType:          xdr.AssetTypeAssetTypePoolShare,
-					Balance:            int64(poolShareTrustLine.Balance),
-					LedgerKey:          "AAAAAQAAAAC2LgFRDBZ3J52nLm30kq2iMgrO7dYzYAN3hvjtf1IHWgAAAAMBAgMEAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA==",
-					Limit:              int64(poolShareTrustLine.Limit),
-					LiquidityPoolID:    "0102030400000000000000000000000000000000000000000000000000000000",
-					Flags:              uint32(poolShareTrustLine.Flags),
-					LastModifiedLedger: uint32(lastModifiedLedgerSeq),
-				},
-			},
-			arg)
+	s.mockTrustLinesBatchInsertBuilder.On("Add", history.TrustLine{
+		AccountID:          trustLine.AccountId.Address(),
+		AssetType:          xdr.AssetTypeAssetTypeCreditAlphanum4,
+		AssetIssuer:        trustLineIssuer.Address(),
+		AssetCode:          "EUR",
+		Balance:            int64(trustLine.Balance),
+		LedgerKey:          "AAAAAQAAAAAdBJqAD9qPq+j2nRDdjdp5KVoUh8riPkNO9ato7BNs8wAAAAFFVVIAAAAAAGL8HQvQkbK2HA3WVjRrKmjX00fG8sLI7m0ERwJW/AX3",
+		Limit:              int64(trustLine.Limit),
+		LiquidityPoolID:    "",
+		BuyingLiabilities:  int64(trustLine.Liabilities().Buying),
+		SellingLiabilities: int64(trustLine.Liabilities().Selling),
+		Flags:              uint32(trustLine.Flags),
+		LastModifiedLedger: uint32(lastModifiedLedgerSeq),
+		Sponsor:            null.String{},
 	}).Return(nil).Once()
+
+	s.mockTrustLinesBatchInsertBuilder.On("Add", history.TrustLine{
+		AccountID:          poolShareTrustLine.AccountId.Address(),
+		AssetType:          xdr.AssetTypeAssetTypePoolShare,
+		Balance:            int64(poolShareTrustLine.Balance),
+		LedgerKey:          "AAAAAQAAAAC2LgFRDBZ3J52nLm30kq2iMgrO7dYzYAN3hvjtf1IHWgAAAAMBAgMEAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA==",
+		Limit:              int64(poolShareTrustLine.Limit),
+		LiquidityPoolID:    "0102030400000000000000000000000000000000000000000000000000000000",
+		Flags:              uint32(poolShareTrustLine.Flags),
+		LastModifiedLedger: uint32(lastModifiedLedgerSeq),
+	}).Return(nil).Once()
+
 }
 
 func (s *TrustLinesProcessorTestSuiteState) TestCreateTrustLineUnauthorized() {
@@ -142,27 +139,21 @@ func (s *TrustLinesProcessorTestSuiteState) TestCreateTrustLineUnauthorized() {
 	})
 	s.Assert().NoError(err)
 
-	s.mockQ.On(
-		"UpsertTrustLines",
-		s.ctx,
-		[]history.TrustLine{
-			{
-				AccountID:          trustLine.AccountId.Address(),
-				AssetType:          xdr.AssetTypeAssetTypeCreditAlphanum4,
-				AssetIssuer:        trustLineIssuer.Address(),
-				AssetCode:          "EUR",
-				Balance:            int64(trustLine.Balance),
-				LedgerKey:          "AAAAAQAAAAAdBJqAD9qPq+j2nRDdjdp5KVoUh8riPkNO9ato7BNs8wAAAAFFVVIAAAAAAGL8HQvQkbK2HA3WVjRrKmjX00fG8sLI7m0ERwJW/AX3",
-				Limit:              int64(trustLine.Limit),
-				LiquidityPoolID:    "",
-				BuyingLiabilities:  int64(trustLine.Liabilities().Buying),
-				SellingLiabilities: int64(trustLine.Liabilities().Selling),
-				Flags:              uint32(trustLine.Flags),
-				LastModifiedLedger: uint32(lastModifiedLedgerSeq),
-				Sponsor:            null.String{},
-			},
-		},
-	).Return(nil).Once()
+	s.mockTrustLinesBatchInsertBuilder.On("Add", history.TrustLine{
+		AccountID:          trustLine.AccountId.Address(),
+		AssetType:          xdr.AssetTypeAssetTypeCreditAlphanum4,
+		AssetIssuer:        trustLineIssuer.Address(),
+		AssetCode:          "EUR",
+		Balance:            int64(trustLine.Balance),
+		LedgerKey:          "AAAAAQAAAAAdBJqAD9qPq+j2nRDdjdp5KVoUh8riPkNO9ato7BNs8wAAAAFFVVIAAAAAAGL8HQvQkbK2HA3WVjRrKmjX00fG8sLI7m0ERwJW/AX3",
+		Limit:              int64(trustLine.Limit),
+		LiquidityPoolID:    "",
+		BuyingLiabilities:  int64(trustLine.Liabilities().Buying),
+		SellingLiabilities: int64(trustLine.Liabilities().Selling),
+		Flags:              uint32(trustLine.Flags),
+		LastModifiedLedger: uint32(lastModifiedLedgerSeq),
+		Sponsor:            null.String{},
+	}).Return(nil).Once()
 }
 
 func TestTrustLinesProcessorTestSuiteLedger(t *testing.T) {
@@ -171,14 +162,20 @@ func TestTrustLinesProcessorTestSuiteLedger(t *testing.T) {
 
 type TrustLinesProcessorTestSuiteLedger struct {
 	suite.Suite
-	ctx       context.Context
-	processor *TrustLinesProcessor
-	mockQ     *history.MockQTrustLines
+	ctx                              context.Context
+	processor                        *TrustLinesProcessor
+	mockQ                            *history.MockQTrustLines
+	mockTrustLinesBatchInsertBuilder *history.MockTrustLinesBatchInsertBuilder
 }
 
 func (s *TrustLinesProcessorTestSuiteLedger) SetupTest() {
 	s.ctx = context.Background()
 	s.mockQ = &history.MockQTrustLines{}
+
+	s.mockTrustLinesBatchInsertBuilder = &history.MockTrustLinesBatchInsertBuilder{}
+	s.mockQ.On("NewTrustLinesBatchInsertBuilder").Return(s.mockTrustLinesBatchInsertBuilder).Twice()
+	s.mockTrustLinesBatchInsertBuilder.On("Exec", s.ctx).Return(nil).Once()
+
 	s.processor = NewTrustLinesProcessor(s.mockQ)
 }
 
@@ -298,47 +295,36 @@ func (s *TrustLinesProcessorTestSuiteLedger) TestInsertTrustLine() {
 	})
 	s.Assert().NoError(err)
 
-	s.mockQ.On(
-		"UpsertTrustLines",
-		s.ctx,
-		mock.AnythingOfType("[]history.TrustLine"),
-	).Run(func(args mock.Arguments) {
-		arg := args.Get(1).([]history.TrustLine)
-		s.Assert().ElementsMatch(
-			[]history.TrustLine{
-				{
-					AccountID:          updatedTrustLine.AccountId.Address(),
-					AssetType:          updatedTrustLine.Asset.Type,
-					AssetIssuer:        trustLineIssuer.Address(),
-					AssetCode:          "EUR",
-					Balance:            int64(updatedTrustLine.Balance),
-					LedgerKey:          "AAAAAQAAAAAdBJqAD9qPq+j2nRDdjdp5KVoUh8riPkNO9ato7BNs8wAAAAFFVVIAAAAAAGL8HQvQkbK2HA3WVjRrKmjX00fG8sLI7m0ERwJW/AX3",
-					Limit:              int64(updatedTrustLine.Limit),
-					LiquidityPoolID:    "",
-					BuyingLiabilities:  int64(updatedTrustLine.Liabilities().Buying),
-					SellingLiabilities: int64(updatedTrustLine.Liabilities().Selling),
-					Flags:              uint32(updatedTrustLine.Flags),
-					LastModifiedLedger: uint32(lastModifiedLedgerSeq),
-					Sponsor:            null.String{},
-				},
-				{
-					AccountID:          updatedUnauthorizedTrustline.AccountId.Address(),
-					AssetType:          updatedUnauthorizedTrustline.Asset.Type,
-					AssetIssuer:        trustLineIssuer.Address(),
-					AssetCode:          "USD",
-					Balance:            int64(updatedUnauthorizedTrustline.Balance),
-					LedgerKey:          "AAAAAQAAAAC2LgFRDBZ3J52nLm30kq2iMgrO7dYzYAN3hvjtf1IHWgAAAAFVU0QAAAAAAGL8HQvQkbK2HA3WVjRrKmjX00fG8sLI7m0ERwJW/AX3",
-					Limit:              int64(updatedUnauthorizedTrustline.Limit),
-					LiquidityPoolID:    "",
-					BuyingLiabilities:  int64(updatedUnauthorizedTrustline.Liabilities().Buying),
-					SellingLiabilities: int64(updatedUnauthorizedTrustline.Liabilities().Selling),
-					Flags:              uint32(updatedUnauthorizedTrustline.Flags),
-					LastModifiedLedger: uint32(lastModifiedLedgerSeq),
-					Sponsor:            null.String{},
-				},
-			},
-			arg,
-		)
+	s.mockTrustLinesBatchInsertBuilder.On("Add", history.TrustLine{
+		AccountID:          updatedTrustLine.AccountId.Address(),
+		AssetType:          updatedTrustLine.Asset.Type,
+		AssetIssuer:        trustLineIssuer.Address(),
+		AssetCode:          "EUR",
+		Balance:            int64(updatedTrustLine.Balance),
+		LedgerKey:          "AAAAAQAAAAAdBJqAD9qPq+j2nRDdjdp5KVoUh8riPkNO9ato7BNs8wAAAAFFVVIAAAAAAGL8HQvQkbK2HA3WVjRrKmjX00fG8sLI7m0ERwJW/AX3",
+		Limit:              int64(updatedTrustLine.Limit),
+		LiquidityPoolID:    "",
+		BuyingLiabilities:  int64(updatedTrustLine.Liabilities().Buying),
+		SellingLiabilities: int64(updatedTrustLine.Liabilities().Selling),
+		Flags:              uint32(updatedTrustLine.Flags),
+		LastModifiedLedger: uint32(lastModifiedLedgerSeq),
+		Sponsor:            null.String{},
+	}).Return(nil).Once()
+
+	s.mockTrustLinesBatchInsertBuilder.On("Add", history.TrustLine{
+		AccountID:          updatedUnauthorizedTrustline.AccountId.Address(),
+		AssetType:          updatedUnauthorizedTrustline.Asset.Type,
+		AssetIssuer:        trustLineIssuer.Address(),
+		AssetCode:          "USD",
+		Balance:            int64(updatedUnauthorizedTrustline.Balance),
+		LedgerKey:          "AAAAAQAAAAC2LgFRDBZ3J52nLm30kq2iMgrO7dYzYAN3hvjtf1IHWgAAAAFVU0QAAAAAAGL8HQvQkbK2HA3WVjRrKmjX00fG8sLI7m0ERwJW/AX3",
+		Limit:              int64(updatedUnauthorizedTrustline.Limit),
+		LiquidityPoolID:    "",
+		BuyingLiabilities:  int64(updatedUnauthorizedTrustline.Liabilities().Buying),
+		SellingLiabilities: int64(updatedUnauthorizedTrustline.Liabilities().Selling),
+		Flags:              uint32(updatedUnauthorizedTrustline.Flags),
+		LastModifiedLedger: uint32(lastModifiedLedgerSeq),
+		Sponsor:            null.String{},
 	}).Return(nil).Once()
 	s.Assert().NoError(s.processor.Commit(s.ctx))
 }
@@ -626,26 +612,22 @@ func (s *TrustLinesProcessorTestSuiteLedger) TestProcessUpgradeChange() {
 	})
 	s.Assert().NoError(err)
 
-	s.mockQ.On(
-		"UpsertTrustLines", s.ctx,
-		[]history.TrustLine{
-			{
-				AccountID:          updatedTrustLine.AccountId.Address(),
-				AssetType:          updatedTrustLine.Asset.Type,
-				AssetIssuer:        trustLineIssuer.Address(),
-				AssetCode:          "EUR",
-				Balance:            int64(updatedTrustLine.Balance),
-				LedgerKey:          "AAAAAQAAAAAdBJqAD9qPq+j2nRDdjdp5KVoUh8riPkNO9ato7BNs8wAAAAFFVVIAAAAAAGL8HQvQkbK2HA3WVjRrKmjX00fG8sLI7m0ERwJW/AX3",
-				Limit:              int64(updatedTrustLine.Limit),
-				LiquidityPoolID:    "",
-				BuyingLiabilities:  int64(updatedTrustLine.Liabilities().Buying),
-				SellingLiabilities: int64(updatedTrustLine.Liabilities().Selling),
-				Flags:              uint32(updatedTrustLine.Flags),
-				LastModifiedLedger: uint32(lastModifiedLedgerSeq),
-				Sponsor:            null.String{},
-			},
-		},
-	).Return(nil).Once()
+	s.mockTrustLinesBatchInsertBuilder.On("Add", history.TrustLine{
+		AccountID:          updatedTrustLine.AccountId.Address(),
+		AssetType:          updatedTrustLine.Asset.Type,
+		AssetIssuer:        trustLineIssuer.Address(),
+		AssetCode:          "EUR",
+		Balance:            int64(updatedTrustLine.Balance),
+		LedgerKey:          "AAAAAQAAAAAdBJqAD9qPq+j2nRDdjdp5KVoUh8riPkNO9ato7BNs8wAAAAFFVVIAAAAAAGL8HQvQkbK2HA3WVjRrKmjX00fG8sLI7m0ERwJW/AX3",
+		Limit:              int64(updatedTrustLine.Limit),
+		LiquidityPoolID:    "",
+		BuyingLiabilities:  int64(updatedTrustLine.Liabilities().Buying),
+		SellingLiabilities: int64(updatedTrustLine.Liabilities().Selling),
+		Flags:              uint32(updatedTrustLine.Flags),
+		LastModifiedLedger: uint32(lastModifiedLedgerSeq),
+		Sponsor:            null.String{},
+	}).Return(nil).Once()
+
 	s.Assert().NoError(s.processor.Commit(s.ctx))
 }