Repository: hyperledger/firefly
Pull request #967 (2022-08-16 04:15:52)

Performance improvements for batch lookups
Ensure batch ID queries match the current indexes (to avoid sequential scan), and improve the caching in aggregator when processing a group of pins.

codecov-commenter: # [Codecov](https://codecov.io/gh/hyperledger/firefly/pull/967?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=hyperledger) Report
> Merging [#967](https://codecov.io/gh/hyperledger/firefly/pull/967?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=hyperledger) (852af08) into [main](https://codecov.io/gh/hyperledger/firefly/commit/daa036a2c37394d6e3641a611076ba178d2d106d?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=hyperledger) (daa036a) will **not change** coverage.
> The diff coverage is `100.00%`.

```diff
@@           Coverage Diff           @@
##             main     #967   +/-   ##
=======================================
  Coverage   99.99%   99.99%           
=======================================
  Files         299      299           
  Lines       19492    19497    +5     
=======================================
+ Hits        19491    19496    +5     
  Misses          1        1           
```


| [Impacted Files](https://codecov.io/gh/hyperledger/firefly/pull/967?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=hyperledger) | Coverage Δ | |
|---|---|---|
| [cmd/firefly.go](https://codecov.io/gh/hyperledger/firefly/pull/967/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=hyperledger#diff-Y21kL2ZpcmVmbHkuZ28=) | `100.00% <100.00%> (ø)` | |
| [internal/database/sqlcommon/message\_sql.go](https://codecov.io/gh/hyperledger/firefly/pull/967/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=hyperledger#diff-aW50ZXJuYWwvZGF0YWJhc2Uvc3FsY29tbW9uL21lc3NhZ2Vfc3FsLmdv) | `100.00% <100.00%> (ø)` | |
| [internal/events/aggregator.go](https://codecov.io/gh/hyperledger/firefly/pull/967/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=hyperledger#diff-aW50ZXJuYWwvZXZlbnRzL2FnZ3JlZ2F0b3IuZ28=) | `100.00% <100.00%> (ø)` | |
| [internal/events/aggregator\_rewind.go](https://codecov.io/gh/hyperledger/firefly/pull/967/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=hyperledger#diff-aW50ZXJuYWwvZXZlbnRzL2FnZ3JlZ2F0b3JfcmV3aW5kLmdv) | `100.00% <100.00%> (ø)` | |

Help us with your feedback. Take ten seconds to tell us [how you rate us](https://about.codecov.io/nps?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=hyperledger). Have a feature suggestion? [Share it here.](https://app.codecov.io/gh/feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=hyperledger)


Patch:
diff --git a/cmd/firefly.go b/cmd/firefly.go
index 54f111b15..a4df2b212 100644
--- a/cmd/firefly.go
+++ b/cmd/firefly.go
@@ -58,6 +58,7 @@ var showConfigCommand = &cobra.Command{
 	Short:   "List out the configuration options",
 	Run: func(cmd *cobra.Command, args []string) {
 		// Initialize config of all plugins
+		resetConfig()
 		getRootManager()
 		_ = config.ReadConfig(configSuffix, cfgFile)
 
@@ -79,6 +80,11 @@ func init() {
 	rootCmd.AddCommand(showConfigCommand)
 }
 
+func resetConfig() {
+	coreconfig.Reset()
+	apiserver.InitConfig()
+}
+
 func getRootManager() namespace.Manager {
 	if _utManager != nil {
 		return _utManager
@@ -94,8 +100,7 @@ func Execute() error {
 func run() error {
 
 	// Read the configuration
-	coreconfig.Reset()
-	apiserver.InitConfig()
+	resetConfig()
 	err := config.ReadConfig(configSuffix, cfgFile)
 
 	// Setup logging after reading config (even if failed), to output header correctly
@@ -132,8 +137,7 @@ func run() error {
 			log.L(ctx).Infof("Restarting due to configuration change")
 			mgr.WaitStop()
 			// Re-read the configuration
-			coreconfig.Reset()
-			apiserver.InitConfig()
+			resetConfig()
 			if err := config.ReadConfig(configSuffix, cfgFile); err != nil {
 				cancelCtx()
 				return err
diff --git a/db/migrations/postgres/000096_add_namespace_to_message_indexes.up.sql b/db/migrations/postgres/000096_add_namespace_to_message_indexes.up.sql
index 4a240b442..f4ca7c67e 100644
--- a/db/migrations/postgres/000096_add_namespace_to_message_indexes.up.sql
+++ b/db/migrations/postgres/000096_add_namespace_to_message_indexes.up.sql
@@ -44,6 +44,7 @@ DROP TABLE messages_data_old;
 UPDATE messages_data SET namespace = msg.namespace
   FROM (SELECT namespace, id FROM messages) AS msg
   WHERE messages_data.message_id = msg.id;
+ALTER TABLE messages_data ALTER COLUMN namespace SET NOT NULL;
 
 CREATE INDEX messages_data_message ON messages_data(namespace, message_id);
 CREATE INDEX messages_data_data ON messages_data(namespace, data_id);
diff --git a/internal/database/sqlcommon/message_sql.go b/internal/database/sqlcommon/message_sql.go
index 5fa7a4021..0e78ef148 100644
--- a/internal/database/sqlcommon/message_sql.go
+++ b/internal/database/sqlcommon/message_sql.go
@@ -208,6 +208,7 @@ func (s *SQLCommon) InsertMessages(ctx context.Context, messages []*core.Message
 	if s.features.MultiRowInsert {
 		msgQuery := sq.Insert(messagesTable).Columns(msgColumns...)
 		dataRefQuery := sq.Insert(messagesDataJoinTable).Columns(
+			"namespace",
 			"message_id",
 			"data_id",
 			"data_hash",
@@ -217,7 +218,7 @@ func (s *SQLCommon) InsertMessages(ctx context.Context, messages []*core.Message
 		for _, message := range messages {
 			msgQuery = s.setMessageInsertValues(msgQuery, message)
 			for idx, dataRef := range message.Data {
-				dataRefQuery = dataRefQuery.Values(message.Header.ID, dataRef.ID, dataRef.Hash, idx)
+				dataRefQuery = dataRefQuery.Values(message.LocalNamespace, message.Header.ID, dataRef.ID, dataRef.Hash, idx)
 				dataRefCount++
 			}
 		}
@@ -517,13 +518,15 @@ func (s *SQLCommon) GetMessageIDs(ctx context.Context, namespace string, filter
 	return ids, nil
 }
 
-func (s *SQLCommon) GetBatchIDsForDataAttachments(ctx context.Context, dataIDs []*fftypes.UUID) (batchIDs []*fftypes.UUID, err error) {
-	query := sq.Select("m.batch_id").From("messages_data AS md").LeftJoin("messages AS m ON m.id = md.message_id").Where(sq.Eq{"md.data_id": dataIDs})
+func (s *SQLCommon) GetBatchIDsForDataAttachments(ctx context.Context, namespace string, dataIDs []*fftypes.UUID) (batchIDs []*fftypes.UUID, err error) {
+	query := sq.Select("m.batch_id").From("messages_data AS md").LeftJoin("messages AS m ON m.id = md.message_id").
+		Where(sq.Eq{"md.data_id": dataIDs, "md.namespace": namespace})
 	return s.queryBatchIDs(ctx, query)
 }
 
-func (s *SQLCommon) GetBatchIDsForMessages(ctx context.Context, msgIDs []*fftypes.UUID) (batchIDs []*fftypes.UUID, err error) {
-	return s.queryBatchIDs(ctx, sq.Select("batch_id").From(messagesTable).Where(sq.Eq{"id": msgIDs}))
+func (s *SQLCommon) GetBatchIDsForMessages(ctx context.Context, namespace string, msgIDs []*fftypes.UUID) (batchIDs []*fftypes.UUID, err error) {
+	return s.queryBatchIDs(ctx, sq.Select("batch_id").From(messagesTable).
+		Where(sq.Eq{"id": msgIDs, "namespace_local": namespace}))
 }
 
 func (s *SQLCommon) queryBatchIDs(ctx context.Context, query sq.SelectBuilder) (batchIDs []*fftypes.UUID, err error) {
@@ -571,7 +574,7 @@ func (s *SQLCommon) GetMessagesForData(ctx context.Context, namespace string, da
 	query, fop, fi, err := s.filterSelect(
 		ctx, "m", sq.Select(cols...).From("messages_data AS md"),
 		filter, msgFilterFieldMap, []interface{}{"sequence"},
-		sq.Eq{"md.data_id": dataID, "namespace_local": namespace})
+		sq.Eq{"md.data_id": dataID, "md.namespace": namespace})
 	if err != nil {
 		return nil, nil, err
 	}
diff --git a/internal/events/aggregator.go b/internal/events/aggregator.go
index 1735803b4..98e294749 100644
--- a/internal/events/aggregator.go
+++ b/internal/events/aggregator.go
@@ -329,22 +329,22 @@ func (ag *aggregator) processPins(ctx context.Context, pins []*core.Pin, state *
 	l := log.L(ctx)
 
 	// Keep a batch cache for this list of pins
-	var batch *core.BatchPersisted
-	var manifest *core.BatchManifest
+	manifests := make(map[fftypes.UUID]*core.BatchManifest)
 	// As messages can have multiple topics, we need to avoid processing the message twice in the same poll loop.
 	// We must check all the contexts in the message, and mark them dispatched together.
 	dupMsgCheck := make(map[fftypes.UUID]bool)
 	for _, pin := range pins {
-
-		if batch == nil || !batch.ID.Equals(pin.Batch) {
-			batch, manifest, err = ag.GetBatchForPin(ctx, pin)
+		manifest, ok := manifests[*pin.Batch]
+		if !ok {
+			_, manifest, err = ag.GetBatchForPin(ctx, pin)
 			if err != nil {
 				return err
 			}
-			if batch == nil {
-				l.Debugf("Pin %.10d batch unavailable: batch=%s pinIndex=%d hash=%s masked=%t", pin.Sequence, pin.Batch, pin.Index, pin.Hash, pin.Masked)
-				continue
-			}
+			manifests[*pin.Batch] = manifest
+		}
+		if manifest == nil {
+			l.Debugf("Pin %.10d batch unavailable: batch=%s pinIndex=%d hash=%s masked=%t", pin.Sequence, pin.Batch, pin.Index, pin.Hash, pin.Masked)
+			continue
 		}
 
 		// Extract the message from the batch - where the index is of a topic within a message
diff --git a/internal/events/aggregator_rewind.go b/internal/events/aggregator_rewind.go
index 3c4eb181f..6dc4a8d9f 100644
--- a/internal/events/aggregator_rewind.go
+++ b/internal/events/aggregator_rewind.go
@@ -253,7 +253,7 @@ func (rw *rewinder) getRewindsForMessages(ctx context.Context, msgRewinds []*fft
 
 	if len(cacheMisses) > 0 {
 		log.L(ctx).Debugf("Cache misses for message rewinds: %v", cacheMisses)
-		msgBatchIDs, err := rw.database.GetBatchIDsForMessages(ctx, cacheMisses)
+		msgBatchIDs, err := rw.database.GetBatchIDsForMessages(ctx, rw.aggregator.namespace, cacheMisses)
 		if err != nil {
 			return err
 		}
@@ -283,7 +283,7 @@ func (rw *rewinder) getRewindsForBlobs(ctx context.Context, newHashes []driver.V
 		for i, d := range data {
 			dataIDs[i] = d.ID
 		}
-		msgBatchIDs, err := rw.database.GetBatchIDsForDataAttachments(ctx, dataIDs)
+		msgBatchIDs, err := rw.database.GetBatchIDsForDataAttachments(ctx, rw.aggregator.namespace, dataIDs)
 		if err != nil {
 			return err
 		}
diff --git a/mocks/databasemocks/plugin.go b/mocks/databasemocks/plugin.go
index 4c4630f3b..8b33436ac 100644
--- a/mocks/databasemocks/plugin.go
+++ b/mocks/databasemocks/plugin.go
@@ -130,13 +130,13 @@ func (_m *Plugin) GetBatchByID(ctx context.Context, namespace string, id *fftype
 	return r0, r1
 }
 
-// GetBatchIDsForDataAttachments provides a mock function with given fields: ctx, dataIDs
-func (_m *Plugin) GetBatchIDsForDataAttachments(ctx context.Context, dataIDs []*fftypes.UUID) ([]*fftypes.UUID, error) {
-	ret := _m.Called(ctx, dataIDs)
+// GetBatchIDsForDataAttachments provides a mock function with given fields: ctx, namespace, dataIDs
+func (_m *Plugin) GetBatchIDsForDataAttachments(ctx context.Context, namespace string, dataIDs []*fftypes.UUID) ([]*fftypes.UUID, error) {
+	ret := _m.Called(ctx, namespace, dataIDs)
 
 	var r0 []*fftypes.UUID
-	if rf, ok := ret.Get(0).(func(context.Context, []*fftypes.UUID) []*fftypes.UUID); ok {
-		r0 = rf(ctx, dataIDs)
+	if rf, ok := ret.Get(0).(func(context.Context, string, []*fftypes.UUID) []*fftypes.UUID); ok {
+		r0 = rf(ctx, namespace, dataIDs)
 	} else {
 		if ret.Get(0) != nil {
 			r0 = ret.Get(0).([]*fftypes.UUID)
@@ -144,8 +144,8 @@ func (_m *Plugin) GetBatchIDsForDataAttachments(ctx context.Context, dataIDs []*
 	}
 
 	var r1 error
-	if rf, ok := ret.Get(1).(func(context.Context, []*fftypes.UUID) error); ok {
-		r1 = rf(ctx, dataIDs)
+	if rf, ok := ret.Get(1).(func(context.Context, string, []*fftypes.UUID) error); ok {
+		r1 = rf(ctx, namespace, dataIDs)
 	} else {
 		r1 = ret.Error(1)
 	}
@@ -153,13 +153,13 @@ func (_m *Plugin) GetBatchIDsForDataAttachments(ctx context.Context, dataIDs []*
 	return r0, r1
 }
 
-// GetBatchIDsForMessages provides a mock function with given fields: ctx, msgIDs
-func (_m *Plugin) GetBatchIDsForMessages(ctx context.Context, msgIDs []*fftypes.UUID) ([]*fftypes.UUID, error) {
-	ret := _m.Called(ctx, msgIDs)
+// GetBatchIDsForMessages provides a mock function with given fields: ctx, namespace, msgIDs
+func (_m *Plugin) GetBatchIDsForMessages(ctx context.Context, namespace string, msgIDs []*fftypes.UUID) ([]*fftypes.UUID, error) {
+	ret := _m.Called(ctx, namespace, msgIDs)
 
 	var r0 []*fftypes.UUID
-	if rf, ok := ret.Get(0).(func(context.Context, []*fftypes.UUID) []*fftypes.UUID); ok {
-		r0 = rf(ctx, msgIDs)
+	if rf, ok := ret.Get(0).(func(context.Context, string, []*fftypes.UUID) []*fftypes.UUID); ok {
+		r0 = rf(ctx, namespace, msgIDs)
 	} else {
 		if ret.Get(0) != nil {
 			r0 = ret.Get(0).([]*fftypes.UUID)
@@ -167,8 +167,8 @@ func (_m *Plugin) GetBatchIDsForMessages(ctx context.Context, msgIDs []*fftypes.
 	}
 
 	var r1 error
-	if rf, ok := ret.Get(1).(func(context.Context, []*fftypes.UUID) error); ok {
-		r1 = rf(ctx, msgIDs)
+	if rf, ok := ret.Get(1).(func(context.Context, string, []*fftypes.UUID) error); ok {
+		r1 = rf(ctx, namespace, msgIDs)
 	} else {
 		r1 = ret.Error(1)
 	}
diff --git a/pkg/database/plugin.go b/pkg/database/plugin.go
index cd15af04a..ce7c073ae 100644
--- a/pkg/database/plugin.go
+++ b/pkg/database/plugin.go
@@ -106,10 +106,10 @@ type iMessageCollection interface {
 	GetMessagesForData(ctx context.Context, namespace string, dataID *fftypes.UUID, filter Filter) (message []*core.Message, res *FilterResult, err error)
 
 	// GetBatchIDsForMessages - an optimized query to retrieve any non-null batch IDs for a list of message IDs
-	GetBatchIDsForMessages(ctx context.Context, msgIDs []*fftypes.UUID) (batchIDs []*fftypes.UUID, err error)
+	GetBatchIDsForMessages(ctx context.Context, namespace string, msgIDs []*fftypes.UUID) (batchIDs []*fftypes.UUID, err error)
 
 	// GetBatchIDsForDataAttachments - an optimized query to retrieve any non-null batch IDs for a list of data IDs that might be attached to messages in batches
-	GetBatchIDsForDataAttachments(ctx context.Context, dataIDs []*fftypes.UUID) (batchIDs []*fftypes.UUID, err error)
+	GetBatchIDsForDataAttachments(ctx context.Context, namespace string, dataIDs []*fftypes.UUID) (batchIDs []*fftypes.UUID, err error)
 }
 
 type iDataCollection interface {


Test patch:
diff --git a/internal/database/sqlcommon/message_sql_test.go b/internal/database/sqlcommon/message_sql_test.go
index 2246225a7..ea12d305d 100644
--- a/internal/database/sqlcommon/message_sql_test.go
+++ b/internal/database/sqlcommon/message_sql_test.go
@@ -171,12 +171,12 @@ func TestUpsertE2EWithDB(t *testing.T) {
 	assert.Equal(t, msg.Header.ID, &msgIDs[0].ID)
 	assert.Equal(t, msg.Sequence, msgIDs[0].Sequence)
 
-	batchIDs, err := s.GetBatchIDsForMessages(ctx, []*fftypes.UUID{msg.Header.ID})
+	batchIDs, err := s.GetBatchIDsForMessages(ctx, "ns12345", []*fftypes.UUID{msg.Header.ID})
 	assert.NoError(t, err)
 	assert.Equal(t, 1, len(batchIDs))
 	assert.Equal(t, *msgUpdated.BatchID, *batchIDs[0])
 
-	batchIDs, err = s.GetBatchIDsForDataAttachments(ctx, []*fftypes.UUID{dataID2})
+	batchIDs, err = s.GetBatchIDsForDataAttachments(ctx, "ns12345", []*fftypes.UUID{dataID2})
 	assert.NoError(t, err)
 	assert.Equal(t, 1, len(batchIDs))
 	assert.Equal(t, *msgUpdated.BatchID, *batchIDs[0])
@@ -668,7 +668,7 @@ func TestGetBatchIDsForMessagesSelectFail(t *testing.T) {
 	s, mock := newMockProvider().init()
 	msgID := fftypes.NewUUID()
 	mock.ExpectQuery("SELECT .*").WillReturnError(fmt.Errorf("pop"))
-	_, err := s.GetBatchIDsForMessages(context.Background(), []*fftypes.UUID{msgID})
+	_, err := s.GetBatchIDsForMessages(context.Background(), "ns1", []*fftypes.UUID{msgID})
 	assert.Regexp(t, "FF10115", err)
 	assert.NoError(t, mock.ExpectationsWereMet())
 }
@@ -677,7 +677,7 @@ func TestGetBatchIDsForMessagesScanFail(t *testing.T) {
 	s, mock := newMockProvider().init()
 	msgID := fftypes.NewUUID()
 	mock.ExpectQuery("SELECT .*").WillReturnRows(sqlmock.NewRows([]string{"batch_id"}).AddRow("not a UUID"))
-	_, err := s.GetBatchIDsForMessages(context.Background(), []*fftypes.UUID{msgID})
+	_, err := s.GetBatchIDsForMessages(context.Background(), "ns1", []*fftypes.UUID{msgID})
 	assert.Regexp(t, "FF10121", err)
 	assert.NoError(t, mock.ExpectationsWereMet())
 }
diff --git a/internal/events/aggregator_rewind_test.go b/internal/events/aggregator_rewind_test.go
index d18932d24..4a53c871f 100644
--- a/internal/events/aggregator_rewind_test.go
+++ b/internal/events/aggregator_rewind_test.go
@@ -46,14 +46,14 @@ func TestRewinderE2E(t *testing.T) {
 	mockRunAsGroupPassthrough(mdi)
 	mdi.On("GetDataRefs", mock.Anything, "ns1", mock.Anything).
 		Return(core.DataRefs{{ID: dataID}}, nil, nil)
-	mdi.On("GetBatchIDsForDataAttachments", mock.Anything, []*fftypes.UUID{dataID}).
+	mdi.On("GetBatchIDsForDataAttachments", mock.Anything, "ns1", []*fftypes.UUID{dataID}).
 		Return([]*fftypes.UUID{batchID2}, nil)
 	mdm.On("PeekMessageCache", mock.Anything, mock.Anything, data.CRORequireBatchID).Return(nil, nil)
-	mdi.On("GetBatchIDsForMessages", mock.Anything, mock.Anything).
+	mdi.On("GetBatchIDsForMessages", mock.Anything, "ns1", mock.Anything).
 		Return([]*fftypes.UUID{batchID3}, nil).Once()
 	mdi.On("GetMessageIDs", mock.Anything, "ns1", mock.Anything).
 		Return([]*core.IDAndSequence{{ID: *fftypes.NewUUID()}}, nil).Once()
-	mdi.On("GetBatchIDsForMessages", mock.Anything, mock.Anything).
+	mdi.On("GetBatchIDsForMessages", mock.Anything, "ns1", mock.Anything).
 		Return([]*fftypes.UUID{batchID4}, nil).Once()
 
 	ag.rewinder.start()
@@ -96,7 +96,7 @@ func TestProcessStagedRewindsErrorMessages(t *testing.T) {
 
 	mockRunAsGroupPassthrough(mdi)
 	mdm.On("PeekMessageCache", mock.Anything, mock.Anything, data.CRORequireBatchID).Return(nil, nil)
-	mdi.On("GetBatchIDsForMessages", mock.Anything, mock.Anything).Return(nil, fmt.Errorf("pop"))
+	mdi.On("GetBatchIDsForMessages", mock.Anything, "ns1", mock.Anything).Return(nil, fmt.Errorf("pop"))
 
 	ag.rewinder.stagedRewinds = []*rewind{
 		{rewindType: rewindMessage},
@@ -143,7 +143,7 @@ func TestProcessStagedRewindsErrorBlobBatchIDs(t *testing.T) {
 	mockRunAsGroupPassthrough(mdi)
 	mdi.On("GetDataRefs", mock.Anything, "ns1", mock.Anything).
 		Return(core.DataRefs{{ID: dataID}}, nil, nil)
-	mdi.On("GetBatchIDsForDataAttachments", mock.Anything, []*fftypes.UUID{dataID}).
+	mdi.On("GetBatchIDsForDataAttachments", mock.Anything, "ns1", []*fftypes.UUID{dataID}).
 		Return(nil, fmt.Errorf("pop"))
 
 	ag.rewinder.stagedRewinds = []*rewind{