Repository: stellar/go
Pull request #5118 (2023-11-16 07:04:08)

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



Patch:
diff --git a/services/horizon/internal/db2/history/account_data_batch_insert_builder.go b/services/horizon/internal/db2/history/account_data_batch_insert_builder.go
new file mode 100644
index 00000000..75c8ff61
--- /dev/null
+++ b/services/horizon/internal/db2/history/account_data_batch_insert_builder.go
@@ -0,0 +1,50 @@
+package history
+
+import (
+	"context"
+
+	"github.com/stellar/go/support/db"
+)
+
+type AccountDataBatchInsertBuilder interface {
+	Add(data Data) error
+	Exec(ctx context.Context) error
+}
+
+type accountDataBatchInsertBuilder struct {
+	session db.SessionInterface
+	builder db.FastBatchInsertBuilder
+	table   string
+}
+
+func (q *Q) NewAccountDataBatchInsertBuilder() AccountDataBatchInsertBuilder {
+	return &accountDataBatchInsertBuilder{
+		session: q,
+		builder: db.FastBatchInsertBuilder{},
+		table:   "accounts_data",
+	}
+}
+
+// Add adds a new account data to the batch
+func (i *accountDataBatchInsertBuilder) Add(data Data) error {
+	ledgerKey, err := accountDataKeyToString(AccountDataKey{
+		AccountID: data.AccountID,
+		DataName:  data.Name,
+	})
+	if err != nil {
+		return err
+	}
+	return i.builder.Row(map[string]interface{}{
+		"ledger_key":           ledgerKey,
+		"account_id":           data.AccountID,
+		"name":                 data.Name,
+		"value":                data.Value,
+		"last_modified_ledger": data.LastModifiedLedger,
+		"sponsor":              data.Sponsor,
+	})
+}
+
+// Exec writes the batch of account data to the database.
+func (i *accountDataBatchInsertBuilder) Exec(ctx context.Context) error {
+	return i.builder.Exec(ctx, i.session, i.table)
+}
diff --git a/services/horizon/internal/db2/history/account_data_value.go b/services/horizon/internal/db2/history/account_data_value.go
index efcd8d31..693420e0 100644
--- a/services/horizon/internal/db2/history/account_data_value.go
+++ b/services/horizon/internal/db2/history/account_data_value.go
@@ -22,7 +22,9 @@ func (t *AccountDataValue) Scan(src interface{}) error {
 
 // Value implements driver.Valuer
 func (value AccountDataValue) Value() (driver.Value, error) {
-	return driver.Value([]uint8(base64.StdEncoding.EncodeToString(value))), nil
+	// Return string to bypass buggy encoding in pq driver for []byte.
+	// More info https://github.com/stellar/go/issues/5086#issuecomment-1773215436)
+	return driver.Value(base64.StdEncoding.EncodeToString(value)), nil
 }
 
 func (value AccountDataValue) Base64() string {
diff --git a/services/horizon/internal/db2/history/main.go b/services/horizon/internal/db2/history/main.go
index ab2f69aa..0acae7cb 100644
--- a/services/horizon/internal/db2/history/main.go
+++ b/services/horizon/internal/db2/history/main.go
@@ -357,6 +357,7 @@ type QData interface {
 	GetAccountDataByKeys(ctx context.Context, keys []AccountDataKey) ([]Data, error)
 	UpsertAccountData(ctx context.Context, data []Data) error
 	RemoveAccountData(ctx context.Context, keys []AccountDataKey) (int64, error)
+	NewAccountDataBatchInsertBuilder() AccountDataBatchInsertBuilder
 }
 
 // Asset is a row of data from the `history_assets` table
diff --git a/services/horizon/internal/db2/history/mock_account_data_batch_insert_builder.go b/services/horizon/internal/db2/history/mock_account_data_batch_insert_builder.go
new file mode 100644
index 00000000..aa5af107
--- /dev/null
+++ b/services/horizon/internal/db2/history/mock_account_data_batch_insert_builder.go
@@ -0,0 +1,21 @@
+package history
+
+import (
+	"context"
+
+	"github.com/stretchr/testify/mock"
+)
+
+type MockAccountDataBatchInsertBuilder struct {
+	mock.Mock
+}
+
+func (m *MockAccountDataBatchInsertBuilder) Add(data Data) error {
+	a := m.Called(data)
+	return a.Error(0)
+}
+
+func (m *MockAccountDataBatchInsertBuilder) Exec(ctx context.Context) error {
+	a := m.Called(ctx)
+	return a.Error(0)
+}
diff --git a/services/horizon/internal/db2/history/mock_q_data.go b/services/horizon/internal/db2/history/mock_q_data.go
index 3316aaa5..8d418e89 100644
--- a/services/horizon/internal/db2/history/mock_q_data.go
+++ b/services/horizon/internal/db2/history/mock_q_data.go
@@ -30,3 +30,8 @@ func (m *MockQData) RemoveAccountData(ctx context.Context, keys []AccountDataKey
 	a := m.Called(ctx, keys)
 	return a.Get(0).(int64), a.Error(1)
 }
+
+func (m *MockQData) NewAccountDataBatchInsertBuilder() AccountDataBatchInsertBuilder {
+	a := m.Called()
+	return a.Get(0).(AccountDataBatchInsertBuilder)
+}
diff --git a/services/horizon/internal/ingest/processors/account_data_processor.go b/services/horizon/internal/ingest/processors/account_data_processor.go
index dfdbecb4..f774700f 100644
--- a/services/horizon/internal/ingest/processors/account_data_processor.go
+++ b/services/horizon/internal/ingest/processors/account_data_processor.go
@@ -12,7 +12,8 @@ import (
 type AccountDataProcessor struct {
 	dataQ history.QData
 
-	cache *ingest.ChangeCompactor
+	cache              *ingest.ChangeCompactor
+	batchInsertBuilder history.AccountDataBatchInsertBuilder
 }
 
 func NewAccountDataProcessor(dataQ history.QData) *AccountDataProcessor {
@@ -23,6 +24,7 @@ func NewAccountDataProcessor(dataQ history.QData) *AccountDataProcessor {
 
 func (p *AccountDataProcessor) reset() {
 	p.cache = ingest.NewChangeCompactor()
+	p.batchInsertBuilder = p.dataQ.NewAccountDataBatchInsertBuilder()
 }
 
 func (p *AccountDataProcessor) ProcessChange(ctx context.Context, change ingest.Change) error {
@@ -41,13 +43,13 @@ func (p *AccountDataProcessor) ProcessChange(ctx context.Context, change ingest.
 		if err != nil {
 			return errors.Wrap(err, "error in Commit")
 		}
-		p.reset()
 	}
 
 	return nil
 }
 
 func (p *AccountDataProcessor) Commit(ctx context.Context) error {
+	defer p.reset()
 	var (
 		datasToUpsert []history.Data
 		datasToDelete []history.AccountDataKey
@@ -57,7 +59,10 @@ func (p *AccountDataProcessor) Commit(ctx context.Context) error {
 		switch {
 		case change.Pre == nil && change.Post != nil:
 			// Created
-			datasToUpsert = append(datasToUpsert, p.ledgerEntryToRow(change.Post))
+			err := p.batchInsertBuilder.Add(p.ledgerEntryToRow(change.Post))
+			if err != nil {
+				return errors.Wrap(err, "Error adding to AccountDataBatchInsertBuilder")
+			}
 		case change.Pre != nil && change.Post == nil:
 			// Removed
 			data := change.Pre.Data.MustData()
@@ -72,6 +77,11 @@ func (p *AccountDataProcessor) Commit(ctx context.Context) error {
 		}
 	}
 
+	err := p.batchInsertBuilder.Exec(ctx)
+	if err != nil {
+		return errors.Wrap(err, "Error executing AccountDataBatchInsertBuilder")
+	}
+
 	if len(datasToUpsert) > 0 {
 		if err := p.dataQ.UpsertAccountData(ctx, datasToUpsert); err != nil {
 			return errors.Wrap(err, "error executing upsert")


Test patch:
diff --git a/services/horizon/internal/ingest/processor_runner_test.go b/services/horizon/internal/ingest/processor_runner_test.go
index 7e7838cf..fc82ed3c 100644
--- a/services/horizon/internal/ingest/processor_runner_test.go
+++ b/services/horizon/internal/ingest/processor_runner_test.go
@@ -528,6 +528,13 @@ func mockChangeProcessorBatchBuilders(q *mockDBQ, ctx context.Context, mockExec
 	q.MockQOffers.On("NewOffersBatchInsertBuilder").
 		Return(mockOfferBatchInsertBuilder).Twice()
 
+	mockAccountDataBatchInsertBuilder := &history.MockAccountDataBatchInsertBuilder{}
+	if mockExec {
+		mockAccountDataBatchInsertBuilder.On("Exec", ctx).Return(nil).Once()
+	}
+	q.MockQData.On("NewAccountDataBatchInsertBuilder").
+		Return(mockAccountDataBatchInsertBuilder).Twice()
+
 	mockTrustLinesBatchInsertBuilder := &history.MockTrustLinesBatchInsertBuilder{}
 	if mockExec {
 		mockTrustLinesBatchInsertBuilder.On("Exec", ctx).Return(nil).Once()
@@ -541,6 +548,7 @@ func mockChangeProcessorBatchBuilders(q *mockDBQ, ctx context.Context, mockExec
 		mockClaimableBalanceClaimantBatchInsertBuilder,
 		mockLiquidityPoolBatchInsertBuilder,
 		mockOfferBatchInsertBuilder,
+		mockAccountDataBatchInsertBuilder,
 		mockTrustLinesBatchInsertBuilder,
 	}
 }
diff --git a/services/horizon/internal/ingest/processors/accounts_data_processor_test.go b/services/horizon/internal/ingest/processors/accounts_data_processor_test.go
index 86ff1e13..273cfcca 100644
--- a/services/horizon/internal/ingest/processors/accounts_data_processor_test.go
+++ b/services/horizon/internal/ingest/processors/accounts_data_processor_test.go
@@ -18,15 +18,21 @@ func TestAccountsDataProcessorTestSuiteState(t *testing.T) {
 
 type AccountsDataProcessorTestSuiteState struct {
 	suite.Suite
-	ctx       context.Context
-	processor *AccountDataProcessor
-	mockQ     *history.MockQData
+	ctx                               context.Context
+	processor                         *AccountDataProcessor
+	mockQ                             *history.MockQData
+	mockAccountDataBatchInsertBuilder *history.MockAccountDataBatchInsertBuilder
 }
 
 func (s *AccountsDataProcessorTestSuiteState) SetupTest() {
 	s.ctx = context.Background()
 	s.mockQ = &history.MockQData{}
 
+	s.mockAccountDataBatchInsertBuilder = &history.MockAccountDataBatchInsertBuilder{}
+	s.mockQ.On("NewAccountDataBatchInsertBuilder").
+		Return(s.mockAccountDataBatchInsertBuilder)
+	s.mockAccountDataBatchInsertBuilder.On("Exec", s.ctx).Return(nil)
+
 	s.processor = NewAccountDataProcessor(s.mockQ)
 }
 
@@ -60,7 +66,7 @@ func (s *AccountsDataProcessorTestSuiteState) TestCreatesAccounts() {
 		Value:              history.AccountDataValue(data.DataValue),
 		LastModifiedLedger: uint32(entry.LastModifiedLedgerSeq),
 	}
-	s.mockQ.On("UpsertAccountData", s.ctx, []history.Data{historyData}).Return(nil).Once()
+	s.mockAccountDataBatchInsertBuilder.On("Add", historyData).Return(nil).Once()
 
 	err := s.processor.ProcessChange(s.ctx, ingest.Change{
 		Type: xdr.LedgerEntryTypeData,
@@ -76,15 +82,21 @@ func TestAccountsDataProcessorTestSuiteLedger(t *testing.T) {
 
 type AccountsDataProcessorTestSuiteLedger struct {
 	suite.Suite
-	ctx       context.Context
-	processor *AccountDataProcessor
-	mockQ     *history.MockQData
+	ctx                               context.Context
+	processor                         *AccountDataProcessor
+	mockQ                             *history.MockQData
+	mockAccountDataBatchInsertBuilder *history.MockAccountDataBatchInsertBuilder
 }
 
 func (s *AccountsDataProcessorTestSuiteLedger) SetupTest() {
 	s.ctx = context.Background()
 	s.mockQ = &history.MockQData{}
 
+	s.mockAccountDataBatchInsertBuilder = &history.MockAccountDataBatchInsertBuilder{}
+	s.mockQ.On("NewAccountDataBatchInsertBuilder").
+		Return(s.mockAccountDataBatchInsertBuilder)
+	s.mockAccountDataBatchInsertBuilder.On("Exec", s.ctx).Return(nil)
+
 	s.processor = NewAccountDataProcessor(s.mockQ)
 }
 
@@ -152,7 +164,7 @@ func (s *AccountsDataProcessorTestSuiteLedger) TestNewAccountData() {
 		Value:              history.AccountDataValue(updatedData.DataValue),
 		LastModifiedLedger: uint32(updatedEntry.LastModifiedLedgerSeq),
 	}
-	s.mockQ.On("UpsertAccountData", s.ctx, []history.Data{historyData}).Return(nil).Once()
+	s.mockAccountDataBatchInsertBuilder.On("Add", historyData).Return(nil).Once()
 }
 
 func (s *AccountsDataProcessorTestSuiteLedger) TestUpdateAccountData() {
@@ -196,6 +208,7 @@ func (s *AccountsDataProcessorTestSuiteLedger) TestUpdateAccountData() {
 		Value:              history.AccountDataValue(updatedData.DataValue),
 		LastModifiedLedger: uint32(updatedEntry.LastModifiedLedgerSeq),
 	}
+	s.mockAccountDataBatchInsertBuilder.On("Add", historyData).Return(nil).Once()
 	s.mockQ.On("UpsertAccountData", s.ctx, []history.Data{historyData}).Return(nil).Once()
 }