Repository: stellar/go
Pull request #5111 (2023-11-14 04:07:57)

services/horizon: Use COPY for inserting into offers 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 `OffersChangeProcessor` to use `FastBatchInsertBuilder` which uses COPY to bulk insert into the database table `offers`. Database update and removal operations remain unchanged.

### Why
https://github.com/stellar/go/issues/5098

### Known limitations

[N/A]



Patch:
diff --git a/services/horizon/internal/db2/history/mock_offers_batch_insert_builder.go b/services/horizon/internal/db2/history/mock_offers_batch_insert_builder.go
new file mode 100644
index 00000000..5caa81f0
--- /dev/null
+++ b/services/horizon/internal/db2/history/mock_offers_batch_insert_builder.go
@@ -0,0 +1,21 @@
+package history
+
+import (
+	"context"
+
+	"github.com/stretchr/testify/mock"
+)
+
+type MockOffersBatchInsertBuilder struct {
+	mock.Mock
+}
+
+func (m *MockOffersBatchInsertBuilder) Add(offer Offer) error {
+	a := m.Called(offer)
+	return a.Error(0)
+}
+
+func (m *MockOffersBatchInsertBuilder) Exec(ctx context.Context) error {
+	a := m.Called(ctx)
+	return a.Error(0)
+}
diff --git a/services/horizon/internal/db2/history/mock_q_offers.go b/services/horizon/internal/db2/history/mock_q_offers.go
index 0c4bc5e9..66ae5025 100644
--- a/services/horizon/internal/db2/history/mock_q_offers.go
+++ b/services/horizon/internal/db2/history/mock_q_offers.go
@@ -40,3 +40,8 @@ func (m *MockQOffers) CompactOffers(ctx context.Context, cutOffSequence uint32)
 	a := m.Called(ctx, cutOffSequence)
 	return a.Get(0).(int64), a.Error(1)
 }
+
+func (m *MockQOffers) NewOffersBatchInsertBuilder() OffersBatchInsertBuilder {
+	a := m.Called()
+	return a.Get(0).(OffersBatchInsertBuilder)
+}
diff --git a/services/horizon/internal/db2/history/offers.go b/services/horizon/internal/db2/history/offers.go
index c80d67c8..98a08fef 100644
--- a/services/horizon/internal/db2/history/offers.go
+++ b/services/horizon/internal/db2/history/offers.go
@@ -20,6 +20,7 @@ type QOffers interface {
 	GetUpdatedOffers(ctx context.Context, newerThanSequence uint32) ([]Offer, error)
 	UpsertOffers(ctx context.Context, offers []Offer) error
 	CompactOffers(ctx context.Context, cutOffSequence uint32) (int64, error)
+	NewOffersBatchInsertBuilder() OffersBatchInsertBuilder
 }
 
 func (q *Q) CountOffers(ctx context.Context) (int, error) {
diff --git a/services/horizon/internal/db2/history/offers_batch_insert_builder.go b/services/horizon/internal/db2/history/offers_batch_insert_builder.go
new file mode 100644
index 00000000..8a345d08
--- /dev/null
+++ b/services/horizon/internal/db2/history/offers_batch_insert_builder.go
@@ -0,0 +1,39 @@
+package history
+
+import (
+	"context"
+
+	"github.com/stellar/go/support/db"
+)
+
+// OffersBatchInsertBuilder is used to insert offers into the offers table
+type OffersBatchInsertBuilder interface {
+	Add(offer Offer) error
+	Exec(ctx context.Context) error
+}
+
+// OffersBatchInsertBuilder is a simple wrapper around db.FastBatchInsertBuilder
+type offersBatchInsertBuilder struct {
+	session db.SessionInterface
+	builder db.FastBatchInsertBuilder
+	table   string
+}
+
+// NewOffersBatchInsertBuilder constructs a new OffersBatchInsertBuilder instance
+func (q *Q) NewOffersBatchInsertBuilder() OffersBatchInsertBuilder {
+	return &offersBatchInsertBuilder{
+		session: q,
+		builder: db.FastBatchInsertBuilder{},
+		table:   "offers",
+	}
+}
+
+// Add adds a new offer to the batch
+func (i *offersBatchInsertBuilder) Add(offer Offer) error {
+	return i.builder.RowStruct(offer)
+}
+
+// Exec writes the batch of offers to the database.
+func (i *offersBatchInsertBuilder) Exec(ctx context.Context) error {
+	return i.builder.Exec(ctx, i.session, i.table)
+}
diff --git a/services/horizon/internal/ingest/processors/offers_processor.go b/services/horizon/internal/ingest/processors/offers_processor.go
index af1d8069..13ee130d 100644
--- a/services/horizon/internal/ingest/processors/offers_processor.go
+++ b/services/horizon/internal/ingest/processors/offers_processor.go
@@ -17,7 +17,8 @@ type OffersProcessor struct {
 	offersQ  history.QOffers
 	sequence uint32
 
-	cache *ingest.ChangeCompactor
+	cache              *ingest.ChangeCompactor
+	insertBatchBuilder history.OffersBatchInsertBuilder
 }
 
 func NewOffersProcessor(offersQ history.QOffers, sequence uint32) *OffersProcessor {
@@ -28,6 +29,7 @@ func NewOffersProcessor(offersQ history.QOffers, sequence uint32) *OffersProcess
 
 func (p *OffersProcessor) reset() {
 	p.cache = ingest.NewChangeCompactor()
+	p.insertBatchBuilder = p.offersQ.NewOffersBatchInsertBuilder()
 }
 
 func (p *OffersProcessor) ProcessChange(ctx context.Context, change ingest.Change) error {
@@ -43,7 +45,6 @@ func (p *OffersProcessor) ProcessChange(ctx context.Context, change ingest.Chang
 		if err := p.flushCache(ctx); err != nil {
 			return errors.Wrap(err, "error in Commit")
 		}
-		p.reset()
 	}
 
 	return nil
@@ -67,12 +68,20 @@ func (p *OffersProcessor) ledgerEntryToRow(entry *xdr.LedgerEntry) history.Offer
 }
 
 func (p *OffersProcessor) flushCache(ctx context.Context) error {
+	defer p.reset()
+
 	var batchUpsertOffers []history.Offer
 	changes := p.cache.GetChanges()
 	for _, change := range changes {
 		switch {
-		case change.Post != nil:
-			// Created and updated
+		case change.Pre == nil && change.Post != nil:
+			// Created
+			err := p.insertBatchBuilder.Add(p.ledgerEntryToRow(change.Post))
+			if err != nil {
+				return errors.New("Error adding to OffersBatchInsertBuilder")
+			}
+		case change.Pre != nil && change.Post != nil:
+			// Updated
 			row := p.ledgerEntryToRow(change.Post)
 			batchUpsertOffers = append(batchUpsertOffers, row)
 		case change.Pre != nil && change.Post == nil:
@@ -86,6 +95,11 @@ func (p *OffersProcessor) flushCache(ctx context.Context) error {
 		}
 	}
 
+	err := p.insertBatchBuilder.Exec(ctx)
+	if err != nil {
+		return errors.New("Error executing OffersBatchInsertBuilder")
+	}
+
 	if len(batchUpsertOffers) > 0 {
 		err := p.offersQ.UpsertOffers(ctx, batchUpsertOffers)
 		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 eb1ffa9e..af7c92a2 100644
--- a/services/horizon/internal/ingest/processor_runner_test.go
+++ b/services/horizon/internal/ingest/processor_runner_test.go
@@ -67,6 +67,10 @@ func TestProcessorRunnerRunHistoryArchiveIngestionGenesis(t *testing.T) {
 
 	mockLiquidityPoolBatchInsertBuilder.On("Exec", ctx).Return(nil).Once()
 
+	mockOffersBatchInsertBuilder := &history.MockOffersBatchInsertBuilder{}
+	mockOffersBatchInsertBuilder.On("Exec", ctx).Return(nil).Once()
+	q.MockQOffers.On("NewOffersBatchInsertBuilder").Return(mockOffersBatchInsertBuilder).Twice()
+
 	q.MockQAssetStats.On("InsertAssetStats", ctx, []history.ExpAssetStat{}, 100000).
 		Return(nil)
 
@@ -150,6 +154,10 @@ func TestProcessorRunnerRunHistoryArchiveIngestionHistoryArchive(t *testing.T) {
 
 	mockLiquidityPoolBatchInsertBuilder.On("Exec", ctx).Return(nil).Once()
 
+	mockOffersBatchInsertBuilder := &history.MockOffersBatchInsertBuilder{}
+	mockOffersBatchInsertBuilder.On("Exec", ctx).Return(nil).Once()
+	q.MockQOffers.On("NewOffersBatchInsertBuilder").Return(mockOffersBatchInsertBuilder).Twice()
+
 	q.MockQAssetStats.On("InsertAssetStats", ctx, []history.ExpAssetStat{}, 100000).
 		Return(nil)
 
@@ -203,6 +211,10 @@ func TestProcessorRunnerRunHistoryArchiveIngestionProtocolVersionNotSupported(t
 
 	mockLiquidityPoolBatchInsertBuilder.On("Exec", ctx).Return(nil).Once()
 
+	mockOffersBatchInsertBuilder := &history.MockOffersBatchInsertBuilder{}
+	mockOffersBatchInsertBuilder.On("Exec", ctx).Return(nil).Once()
+	q.MockQOffers.On("NewOffersBatchInsertBuilder").Return(mockOffersBatchInsertBuilder).Once()
+
 	q.MockQAssetStats.On("InsertAssetStats", ctx, []history.ExpAssetStat{}, 100000).
 		Return(nil)
 
@@ -248,6 +260,9 @@ func TestProcessorRunnerBuildChangeProcessor(t *testing.T) {
 
 	mockLiquidityPoolBatchInsertBuilder.On("Exec", ctx).Return(nil).Once()
 
+	mockOfferBatchInsertBuilder := &history.MockOffersBatchInsertBuilder{}
+	q.MockQOffers.On("NewOffersBatchInsertBuilder").Return(mockOfferBatchInsertBuilder).Twice()
+
 	runner := ProcessorRunner{
 		ctx:      ctx,
 		historyQ: q,
@@ -587,6 +602,10 @@ func mockBatchBuilders(q *mockDBQ, mockSession *db.MockSession, ctx context.Cont
 
 	mockLiquidityPoolBatchInsertBuilder.On("Exec", ctx).Return(nil).Once()
 
+	mockOfferBatchInsertBuilder := &history.MockOffersBatchInsertBuilder{}
+	mockOfferBatchInsertBuilder.On("Exec", ctx).Return(nil)
+	q.MockQOffers.On("NewOffersBatchInsertBuilder").Return(mockOfferBatchInsertBuilder)
+
 	q.On("NewTradeBatchInsertBuilder").Return(&history.MockTradeBatchInsertBuilder{})
 
 	return []interface{}{mockAccountSignersBatchInsertBuilder,
diff --git a/services/horizon/internal/ingest/processors/offers_processor_test.go b/services/horizon/internal/ingest/processors/offers_processor_test.go
index 21905dc1..77128a1b 100644
--- a/services/horizon/internal/ingest/processors/offers_processor_test.go
+++ b/services/horizon/internal/ingest/processors/offers_processor_test.go
@@ -21,16 +21,21 @@ func TestOffersProcessorTestSuiteState(t *testing.T) {
 
 type OffersProcessorTestSuiteState struct {
 	suite.Suite
-	ctx       context.Context
-	processor *OffersProcessor
-	mockQ     *history.MockQOffers
-	sequence  uint32
+	ctx                          context.Context
+	processor                    *OffersProcessor
+	mockQ                        *history.MockQOffers
+	sequence                     uint32
+	mockOffersBatchInsertBuilder *history.MockOffersBatchInsertBuilder
 }
 
 func (s *OffersProcessorTestSuiteState) SetupTest() {
 	s.ctx = context.Background()
 	s.mockQ = &history.MockQOffers{}
 
+	s.mockOffersBatchInsertBuilder = &history.MockOffersBatchInsertBuilder{}
+	s.mockQ.On("NewOffersBatchInsertBuilder").Return(s.mockOffersBatchInsertBuilder).Twice()
+	s.mockOffersBatchInsertBuilder.On("Exec", s.ctx).Return(nil).Once()
+
 	s.sequence = 456
 	s.processor = NewOffersProcessor(s.mockQ, s.sequence)
 }
@@ -57,15 +62,13 @@ func (s *OffersProcessorTestSuiteState) TestCreateOffer() {
 		LastModifiedLedgerSeq: lastModifiedLedgerSeq,
 	}
 
-	s.mockQ.On("UpsertOffers", s.ctx, []history.Offer{
-		{
-			SellerID:           "GC3C4AKRBQLHOJ45U4XG35ESVWRDECWO5XLDGYADO6DPR3L7KIDVUMML",
-			OfferID:            1,
-			Pricen:             int32(1),
-			Priced:             int32(2),
-			Price:              float64(0.5),
-			LastModifiedLedger: uint32(lastModifiedLedgerSeq),
-		},
+	s.mockOffersBatchInsertBuilder.On("Add", history.Offer{
+		SellerID:           "GC3C4AKRBQLHOJ45U4XG35ESVWRDECWO5XLDGYADO6DPR3L7KIDVUMML",
+		OfferID:            1,
+		Pricen:             int32(1),
+		Priced:             int32(2),
+		Price:              float64(0.5),
+		LastModifiedLedger: uint32(lastModifiedLedgerSeq),
 	}).Return(nil).Once()
 
 	err := s.processor.ProcessChange(s.ctx, ingest.Change{
@@ -82,16 +85,21 @@ func TestOffersProcessorTestSuiteLedger(t *testing.T) {
 
 type OffersProcessorTestSuiteLedger struct {
 	suite.Suite
-	ctx       context.Context
-	processor *OffersProcessor
-	mockQ     *history.MockQOffers
-	sequence  uint32
+	ctx                          context.Context
+	processor                    *OffersProcessor
+	mockQ                        *history.MockQOffers
+	sequence                     uint32
+	mockOffersBatchInsertBuilder *history.MockOffersBatchInsertBuilder
 }
 
 func (s *OffersProcessorTestSuiteLedger) SetupTest() {
 	s.ctx = context.Background()
 	s.mockQ = &history.MockQOffers{}
 
+	s.mockOffersBatchInsertBuilder = &history.MockOffersBatchInsertBuilder{}
+	s.mockQ.On("NewOffersBatchInsertBuilder").Return(s.mockOffersBatchInsertBuilder).Twice()
+	s.mockOffersBatchInsertBuilder.On("Exec", s.ctx).Return(nil).Once()
+
 	s.sequence = 456
 	s.processor = NewOffersProcessor(s.mockQ, s.sequence)
 }
@@ -166,15 +174,13 @@ func (s *OffersProcessorTestSuiteLedger) setupInsertOffer() {
 	s.Assert().NoError(err)
 
 	// We use LedgerEntryChangesCache so all changes are squashed
-	s.mockQ.On("UpsertOffers", s.ctx, []history.Offer{
-		{
-			SellerID:           "GC3C4AKRBQLHOJ45U4XG35ESVWRDECWO5XLDGYADO6DPR3L7KIDVUMML",
-			OfferID:            2,
-			Pricen:             int32(1),
-			Priced:             int32(6),
-			Price:              float64(1) / float64(6),
-			LastModifiedLedger: uint32(lastModifiedLedgerSeq),
-		},
+	s.mockOffersBatchInsertBuilder.On("Add", history.Offer{
+		SellerID:           "GC3C4AKRBQLHOJ45U4XG35ESVWRDECWO5XLDGYADO6DPR3L7KIDVUMML",
+		OfferID:            2,
+		Pricen:             int32(1),
+		Priced:             int32(6),
+		Price:              float64(1) / float64(6),
+		LastModifiedLedger: uint32(lastModifiedLedgerSeq),
 	}).Return(nil).Once()
 }
 
@@ -223,6 +229,12 @@ func (s *OffersProcessorTestSuiteLedger) TestUpsertManyOffers() {
 		Price:    xdr.Price{2, 3},
 	}
 
+	yetAnotherOffer := xdr.OfferEntry{
+		SellerId: xdr.MustAddress("GBRPYHIL2CI3FNQ4BXLFMNDLFJUNPU2HY3ZMFSHONUCEOASW7QC7OX2H"),
+		OfferId:  xdr.Int64(4),
+		Price:    xdr.Price{2, 6},
+	}
+
 	updatedEntry := xdr.LedgerEntry{
 		LastModifiedLedgerSeq: lastModifiedLedgerSeq,
 		Data: xdr.LedgerEntryData{
@@ -257,6 +269,37 @@ func (s *OffersProcessorTestSuiteLedger) TestUpsertManyOffers() {
 	})
 	s.Assert().NoError(err)
 
+	err = s.processor.ProcessChange(s.ctx, ingest.Change{
+		Type: xdr.LedgerEntryTypeOffer,
+		Pre:  nil,
+		Post: &xdr.LedgerEntry{
+			LastModifiedLedgerSeq: lastModifiedLedgerSeq,
+			Data: xdr.LedgerEntryData{
+				Type:  xdr.LedgerEntryTypeOffer,
+				Offer: &yetAnotherOffer,
+			},
+		},
+	})
+	s.Assert().NoError(err)
+
+	s.mockOffersBatchInsertBuilder.On("Add", history.Offer{
+		SellerID:           "GDMUVYVYPYZYBDXNJWKFT3X2GCZCICTL3GSVP6AWBGB4ZZG7ZRDA746P",
+		OfferID:            3,
+		Pricen:             int32(2),
+		Priced:             int32(3),
+		Price:              float64(2) / float64(3),
+		LastModifiedLedger: uint32(lastModifiedLedgerSeq),
+	}).Return(nil).Once()
+
+	s.mockOffersBatchInsertBuilder.On("Add", history.Offer{
+		SellerID:           "GBRPYHIL2CI3FNQ4BXLFMNDLFJUNPU2HY3ZMFSHONUCEOASW7QC7OX2H",
+		OfferID:            4,
+		Pricen:             int32(2),
+		Priced:             int32(6),
+		Price:              float64(2) / float64(6),
+		LastModifiedLedger: uint32(lastModifiedLedgerSeq),
+	}).Return(nil).Once()
+
 	s.mockQ.On("UpsertOffers", s.ctx, mock.Anything).Run(func(args mock.Arguments) {
 		// To fix order issue due to using ChangeCompactor
 		offers := args.Get(1).([]history.Offer)
@@ -271,14 +314,6 @@ func (s *OffersProcessorTestSuiteLedger) TestUpsertManyOffers() {
 					Price:              float64(1) / float64(6),
 					LastModifiedLedger: uint32(lastModifiedLedgerSeq),
 				},
-				{
-					SellerID:           "GDMUVYVYPYZYBDXNJWKFT3X2GCZCICTL3GSVP6AWBGB4ZZG7ZRDA746P",
-					OfferID:            3,
-					Pricen:             int32(2),
-					Priced:             int32(3),
-					Price:              float64(2) / float64(3),
-					LastModifiedLedger: uint32(lastModifiedLedgerSeq),
-				},
 			},
 		)
 	}).Return(nil).Once()
@@ -367,15 +402,13 @@ func (s *OffersProcessorTestSuiteLedger) TestProcessUpgradeChange() {
 	s.Assert().NoError(err)
 
 	// We use LedgerEntryChangesCache so all changes are squashed
-	s.mockQ.On("UpsertOffers", s.ctx, []history.Offer{
-		{
-			SellerID:           "GC3C4AKRBQLHOJ45U4XG35ESVWRDECWO5XLDGYADO6DPR3L7KIDVUMML",
-			OfferID:            2,
-			Pricen:             1,
-			Priced:             6,
-			Price:              float64(1) / float64(6),
-			LastModifiedLedger: uint32(lastModifiedLedgerSeq),
-		},
+	s.mockOffersBatchInsertBuilder.On("Add", history.Offer{
+		SellerID:           "GC3C4AKRBQLHOJ45U4XG35ESVWRDECWO5XLDGYADO6DPR3L7KIDVUMML",
+		OfferID:            2,
+		Pricen:             1,
+		Priced:             6,
+		Price:              float64(1) / float64(6),
+		LastModifiedLedger: uint32(lastModifiedLedgerSeq),
 	}).Return(nil).Once()
 
 	s.mockQ.On("CompactOffers", s.ctx, s.sequence-100).Return(int64(0), nil).Once()