get block range from staging instead

Author: nischitpra

internal/orchestrator/committer.go

@@ -380,13 +380,12 @@ func (c *Committer) getSequentialBlockDataToPublish(ctx context.Context) ([]comm
 	}
 
 	endBlock := new(big.Int).Add(startBlock, big.NewInt(int64(c.blocksPerCommit-1)))
-	blockCount := new(big.Int).Sub(endBlock, startBlock).Int64() + 1
-	blockNumbers := make([]*big.Int, blockCount)
-	for i := int64(0); i < blockCount; i++ {
-		blockNumbers[i] = new(big.Int).Add(startBlock, big.NewInt(i))
-	}
 
-	blocksData, err := c.storage.StagingStorage.GetStagingData(storage.QueryFilter{ChainId: chainID, BlockNumbers: blockNumbers})
+	blocksData, err := c.storage.StagingStorage.GetStagingData(storage.QueryFilter{
+		ChainId:    chainID,
+		StartBlock: startBlock,
+		EndBlock:   endBlock,
+	})
 	if err != nil {
 		return nil, fmt.Errorf("error fetching blocks to publish: %v", err)
 	}

internal/storage/clickhouse.go

@@ -1014,8 +1014,17 @@ func (c *ClickHouseConnector) InsertStagingData(data []common.BlockData) error {
 }
 
 func (c *ClickHouseConnector) GetStagingData(qf QueryFilter) ([]common.BlockData, error) {
-	query := fmt.Sprintf("SELECT data FROM %s.block_data FINAL WHERE block_number IN (%s) AND is_deleted = 0",
-		c.cfg.Database, getBlockNumbersStringArray(qf.BlockNumbers))
+	var query string
+
+	if len(qf.BlockNumbers) > 0 {
+		query = fmt.Sprintf("SELECT data FROM %s.block_data FINAL WHERE block_number IN (%s) AND is_deleted = 0",
+			c.cfg.Database, getBlockNumbersStringArray(qf.BlockNumbers))
+	} else if qf.StartBlock != nil && qf.EndBlock != nil {
+		query = fmt.Sprintf("SELECT data FROM %s.block_data FINAL WHERE block_number >= %s AND block_number <= %s AND is_deleted = 0",
+			c.cfg.Database, qf.StartBlock.String(), qf.EndBlock.String())
+	} else {
+		return nil, fmt.Errorf("either BlockNumbers or StartBlock/EndBlock must be provided")
+	}
 
 	if qf.ChainId.Sign() != 0 {
 		query += fmt.Sprintf(" AND chain_id = %s", qf.ChainId.String())

internal/storage/connector.go

@@ -11,6 +11,8 @@ import (
 type QueryFilter struct {
 	ChainId             *big.Int
 	BlockNumbers        []*big.Int
+	StartBlock          *big.Int
+	EndBlock            *big.Int
 	FilterParams        map[string]string
 	GroupBy             []string
 	SortBy              string

internal/storage/postgres.go

@@ -284,6 +284,13 @@ func (p *PostgresConnector) GetStagingData(qf QueryFilter) ([]common.BlockData,
 			args = append(args, bn.String())
 		}
 		query += fmt.Sprintf(" AND block_number IN (%s)", strings.Join(placeholders, ","))
+	} else if qf.StartBlock != nil && qf.EndBlock != nil {
+		argCount++
+		query += fmt.Sprintf(" AND block_number >= $%d", argCount)
+		args = append(args, qf.StartBlock.String())
+		argCount++
+		query += fmt.Sprintf(" AND block_number <= $%d", argCount)
+		args = append(args, qf.EndBlock.String())
 	}
 
 	query += " ORDER BY block_number ASC"

internal/storage/postgres_connector_test.go

@@ -167,6 +167,17 @@ func TestPostgresConnector_StagingData(t *testing.T) {
 	assert.NoError(t, err)
 	assert.Len(t, retrievedData, 2)
 
+	// Test GetStagingData with StartBlock and EndBlock
+	rangeQf := QueryFilter{
+		ChainId:    big.NewInt(1),
+		StartBlock: big.NewInt(100),
+		EndBlock:   big.NewInt(101),
+	}
+
+	retrievedDataRange, err := conn.GetStagingData(rangeQf)
+	assert.NoError(t, err)
+	assert.Len(t, retrievedDataRange, 2)
+
 	// Test GetLastStagedBlockNumber
 	lastBlock, err := conn.GetLastStagedBlockNumber(big.NewInt(1), big.NewInt(90), big.NewInt(110))
 	assert.NoError(t, err)