Optimize queries based on Timescale hypertable configuration

internal/data/accounts.go

@@ -5,7 +5,6 @@ package data
 import (
 	"context"
 	"fmt"
-	"strings"
 	"time"
 
 	"github.com/lib/pq"
@@ -77,33 +76,3 @@ func (m *AccountModel) BatchGetByOperationIDs(ctx context.Context, operationIDs
 	m.MetricsService.IncDBQuery("BatchGetByOperationIDs", "operations_accounts")
 	return accounts, nil
 }
-
-// BatchGetByStateChangeIDs gets the accounts that are associated with the given state change IDs.
-func (m *AccountModel) BatchGetByStateChangeIDs(ctx context.Context, scToIDs []int64, scOpIDs []int64, scOrders []int64, columns string) ([]*types.AccountWithStateChangeID, error) {
-	// Build tuples for the IN clause. Since (to_id, operation_id, state_change_order) is the primary key of state_changes,
-	// it will be faster to search on this tuple.
-	tuples := make([]string, len(scOrders))
-	for i := range scOrders {
-		tuples[i] = fmt.Sprintf("(%d, %d, %d)", scToIDs[i], scOpIDs[i], scOrders[i])
-	}
-
-	query := fmt.Sprintf(`
-		SELECT account_id AS stellar_address, CONCAT(to_id, '-', operation_id, '-', state_change_order) AS state_change_id
-		FROM state_changes
-		WHERE (to_id, operation_id, state_change_order) IN (%s)
-		ORDER BY ledger_created_at DESC
-	`, strings.Join(tuples, ", "))
-
-	var accountsWithStateChanges []*types.AccountWithStateChangeID
-	start := time.Now()
-	err := m.DB.SelectContext(ctx, &accountsWithStateChanges, query)
-	duration := time.Since(start).Seconds()
-	m.MetricsService.ObserveDBQueryDuration("BatchGetByStateChangeIDs", "state_changes", duration)
-	m.MetricsService.ObserveDBBatchSize("BatchGetByStateChangeIDs", "state_changes", len(scOrders))
-	if err != nil {
-		m.MetricsService.IncDBQueryError("BatchGetByStateChangeIDs", "state_changes", utils.GetDBErrorType(err))
-		return nil, fmt.Errorf("getting accounts by state change IDs: %w", err)
-	}
-	m.MetricsService.IncDBQuery("BatchGetByStateChangeIDs", "state_changes")
-	return accountsWithStateChanges, nil
-}

internal/data/accounts_test.go

@@ -151,69 +151,3 @@ func TestAccountModel_IsAccountFeeBumpEligible(t *testing.T) {
 	require.NoError(t, err)
 	assert.True(t, isFeeBumpEligible)
 }
-
-func TestAccountModelBatchGetByStateChangeIDs(t *testing.T) {
-	dbt := dbtest.Open(t)
-	defer dbt.Close()
-	dbConnectionPool, err := db.OpenDBConnectionPool(dbt.DSN)
-	require.NoError(t, err)
-	defer dbConnectionPool.Close()
-
-	mockMetricsService := metrics.NewMockMetricsService()
-	mockMetricsService.On("ObserveDBQueryDuration", "BatchGetByStateChangeIDs", "state_changes", mock.Anything).Return()
-	mockMetricsService.On("IncDBQuery", "BatchGetByStateChangeIDs", "state_changes").Return()
-	mockMetricsService.On("ObserveDBBatchSize", "BatchGetByStateChangeIDs", "state_changes", mock.Anything).Return().Maybe()
-	defer mockMetricsService.AssertExpectations(t)
-
-	m := &AccountModel{
-		DB:             dbConnectionPool,
-		MetricsService: mockMetricsService,
-	}
-
-	ctx := context.Background()
-	address1 := keypair.MustRandom().Address()
-	address2 := keypair.MustRandom().Address()
-	toID1 := int64(4096)
-	toID2 := int64(8192)
-	stateChangeOrder1 := int64(1)
-	stateChangeOrder2 := int64(1)
-
-	// Insert test transactions first (hash is BYTEA, using valid 64-char hex strings)
-	testHash1 := types.HashBytea("0000000000000000000000000000000000000000000000000000000000000001")
-	testHash2 := types.HashBytea("0000000000000000000000000000000000000000000000000000000000000002")
-	_, err = m.DB.ExecContext(ctx, "INSERT INTO transactions (hash, to_id, envelope_xdr, fee_charged, result_code, meta_xdr, ledger_number, ledger_created_at) VALUES ($1, 4096, 'env1', 100, 'TransactionResultCodeTxSuccess', 'meta1', 1, NOW()), ($2, 8192, 'env2', 200, 'TransactionResultCodeTxSuccess', 'meta2', 2, NOW())", testHash1, testHash2)
-	require.NoError(t, err)
-
-	// Insert test operations (IDs must be in TOID range for each transaction)
-	xdr1 := types.XDRBytea([]byte("xdr1"))
-	xdr2 := types.XDRBytea([]byte("xdr2"))
-	_, err = m.DB.ExecContext(ctx, "INSERT INTO operations (id, operation_type, operation_xdr, result_code, successful, ledger_number, ledger_created_at) VALUES (4097, 'PAYMENT', $1, 'op_success', true, 1, NOW()), (8193, 'PAYMENT', $2, 'op_success', true, 2, NOW())", xdr1, xdr2)
-	require.NoError(t, err)
-
-	// Insert test state changes that reference the accounts (state_changes.account_id is TEXT)
-	_, err = m.DB.ExecContext(ctx, `
-		INSERT INTO state_changes (
-			to_id, state_change_order, state_change_category, ledger_created_at,
-			ledger_number, account_id, operation_id
-		) VALUES
-		($1, $2, 'BALANCE', NOW(), 1, $3, 4097),
-		($4, $5, 'BALANCE', NOW(), 2, $6, 8193)
-	`, toID1, stateChangeOrder1, types.AddressBytea(address1), toID2, stateChangeOrder2, types.AddressBytea(address2))
-	require.NoError(t, err)
-
-	// Test BatchGetByStateChangeIDs function
-	scToIDs := []int64{toID1, toID2}
-	scOpIDs := []int64{4097, 8193}
-	scOrders := []int64{stateChangeOrder1, stateChangeOrder2}
-	accounts, err := m.BatchGetByStateChangeIDs(ctx, scToIDs, scOpIDs, scOrders, "")
-	require.NoError(t, err)
-	assert.Len(t, accounts, 2)
-
-	// Verify accounts are returned with correct state_change_id (format: to_id-operation_id-state_change_order)
-	addressSet := make(map[string]string)
-	for _, acc := range accounts {
-		addressSet[string(acc.StellarAddress)] = acc.StateChangeID
-	}
-	assert.Equal(t, "4096-4097-1", addressSet[address1])
-	assert.Equal(t, "8192-8193-1", addressSet[address2])
-}

internal/data/operations.go

@@ -38,36 +38,44 @@ func (m *OperationModel) GetByID(ctx context.Context, id int64, columns string)
 	return &operation, nil
 }
 
-func (m *OperationModel) GetAll(ctx context.Context, columns string, limit *int32, cursor *int64, sortOrder SortOrder) ([]*types.OperationWithCursor, error) {
+func (m *OperationModel) GetAll(ctx context.Context, columns string, limit *int32, cursor *types.CompositeCursor, sortOrder SortOrder) ([]*types.OperationWithCursor, error) {
 	columns = prepareColumnsWithID(columns, types.Operation{}, "", "id")
 	queryBuilder := strings.Builder{}
-	queryBuilder.WriteString(fmt.Sprintf(`SELECT %s, id as cursor FROM operations`, columns))
+	var args []interface{}
+	argIndex := 1
 
+	queryBuilder.WriteString(fmt.Sprintf(`SELECT %s, ledger_created_at as "cursor.cursor_ledger_created_at", id as "cursor.cursor_id" FROM operations`, columns))
+
+	// Decomposed cursor pagination: expands ROW() tuple comparison into OR clauses so
+	// TimescaleDB ColumnarScan can push filters into vectorized batch processing.
 	if cursor != nil {
-		if sortOrder == DESC {
-			queryBuilder.WriteString(fmt.Sprintf(" WHERE id < %d", *cursor))
-		} else {
-			queryBuilder.WriteString(fmt.Sprintf(" WHERE id > %d", *cursor))
-		}
+		clause, cursorArgs, nextIdx := buildDecomposedCursorCondition([]CursorColumn{
+			{Name: "ledger_created_at", Value: cursor.LedgerCreatedAt},
+			{Name: "id", Value: cursor.ID},
+		}, sortOrder, argIndex)
+		queryBuilder.WriteString(" WHERE " + clause)
+		args = append(args, cursorArgs...)
+		argIndex = nextIdx
 	}
 
 	if sortOrder == DESC {
-		queryBuilder.WriteString(" ORDER BY id DESC")
+		queryBuilder.WriteString(" ORDER BY ledger_created_at DESC, id DESC")
 	} else {
-		queryBuilder.WriteString(" ORDER BY id ASC")
+		queryBuilder.WriteString(" ORDER BY ledger_created_at ASC, id ASC")
 	}
 
 	if limit != nil {
-		queryBuilder.WriteString(fmt.Sprintf(" LIMIT %d", *limit))
+		queryBuilder.WriteString(fmt.Sprintf(" LIMIT $%d", argIndex))
+		args = append(args, *limit)
 	}
 	query := queryBuilder.String()
 	if sortOrder == DESC {
-		query = fmt.Sprintf(`SELECT * FROM (%s) AS operations ORDER BY cursor ASC`, query)
+		query = fmt.Sprintf(`SELECT * FROM (%s) AS operations ORDER BY operations."cursor.cursor_ledger_created_at" ASC, operations."cursor.cursor_id" ASC`, query)
 	}
 
 	var operations []*types.OperationWithCursor
 	start := time.Now()
-	err := m.DB.SelectContext(ctx, &operations, query)
+	err := m.DB.SelectContext(ctx, &operations, query, args...)
 	duration := time.Since(start).Seconds()
 	m.MetricsService.ObserveDBQueryDuration("GetAll", "operations", duration)
 	if err != nil {
@@ -131,15 +139,15 @@ func (m *OperationModel) BatchGetByToIDs(ctx context.Context, toIDs []int64, col
 				JOIN
 					inputs i ON o.id > i.to_id AND o.id < i.to_id + 4096
 			)
-		SELECT %s, id as cursor FROM ranked_operations_per_to_id
+		SELECT %s, ledger_created_at as "cursor.cursor_ledger_created_at", id as "cursor.cursor_id" FROM ranked_operations_per_to_id
 	`
 	queryBuilder.WriteString(fmt.Sprintf(query, sortOrder, columns))
 	if limit != nil {
 		queryBuilder.WriteString(fmt.Sprintf(" WHERE rn <= %d", *limit))
 	}
 	query = queryBuilder.String()
 	if sortOrder == DESC {
-		query = fmt.Sprintf(`SELECT * FROM (%s) AS operations ORDER BY cursor ASC`, query)
+		query = fmt.Sprintf(`SELECT * FROM (%s) AS operations ORDER BY operations."cursor.cursor_ledger_created_at" ASC, operations."cursor.cursor_id" ASC`, query)
 	}
 
 	var operations []*types.OperationWithCursor
@@ -162,7 +170,7 @@ func (m *OperationModel) BatchGetByToID(ctx context.Context, toID int64, columns
 	columns = prepareColumnsWithID(columns, types.Operation{}, "", "id")
 	queryBuilder := strings.Builder{}
 	// Operations for a tx_to_id are in range (tx_to_id, tx_to_id + 4096) based on TOID encoding.
-	queryBuilder.WriteString(fmt.Sprintf(`SELECT %s, id as cursor FROM operations WHERE id > $1 AND id < $1 + 4096`, columns))
+	queryBuilder.WriteString(fmt.Sprintf(`SELECT %s, ledger_created_at as "cursor.cursor_ledger_created_at", id as "cursor.cursor_id" FROM operations WHERE id > $1 AND id < $1 + 4096`, columns))
 
 	args := []interface{}{toID}
 	argIndex := 2
@@ -190,7 +198,7 @@ func (m *OperationModel) BatchGetByToID(ctx context.Context, toID int64, columns
 
 	query := queryBuilder.String()
 	if sortOrder == DESC {
-		query = fmt.Sprintf(`SELECT * FROM (%s) AS operations ORDER BY cursor ASC`, query)
+		query = fmt.Sprintf(`SELECT * FROM (%s) AS operations ORDER BY operations."cursor.cursor_ledger_created_at" ASC, operations."cursor.cursor_id" ASC`, query)
 	}
 
 	var operations []*types.OperationWithCursor
@@ -207,21 +215,72 @@ func (m *OperationModel) BatchGetByToID(ctx context.Context, toID int64, columns
 }
 
 // BatchGetByAccountAddress gets the operations that are associated with a single account address.
-func (m *OperationModel) BatchGetByAccountAddress(ctx context.Context, accountAddress string, columns string, limit *int32, cursor *int64, orderBy SortOrder) ([]*types.OperationWithCursor, error) {
-	columns = prepareColumnsWithID(columns, types.Operation{}, "operations", "id")
+// Uses a MATERIALIZED CTE + LATERAL join pattern to allow TimescaleDB ChunkAppend optimization
+// on the operations_accounts hypertable by ordering on ledger_created_at first.
+func (m *OperationModel) BatchGetByAccountAddress(ctx context.Context, accountAddress string, columns string, limit *int32, cursor *types.CompositeCursor, orderBy SortOrder, timeRange *TimeRange) ([]*types.OperationWithCursor, error) {
+	columns = prepareColumnsWithID(columns, types.Operation{}, "o", "id")
+
+	var queryBuilder strings.Builder
+	args := []interface{}{types.AddressBytea(accountAddress)}
+	argIndex := 2
+
+	// MATERIALIZED CTE scans operations_accounts with ledger_created_at leading the ORDER BY,
+	// enabling TimescaleDB ChunkAppend on the hypertable.
+	queryBuilder.WriteString(`
+		WITH account_ops AS MATERIALIZED (
+			SELECT operation_id, ledger_created_at
+			FROM operations_accounts
+			WHERE account_id = $1`)
+
+	// Time range filter: enables TimescaleDB chunk pruning at the earliest query stage
+	args, argIndex = appendTimeRangeConditions(&queryBuilder, "ledger_created_at", timeRange, args, argIndex)
+
+	// Decomposed cursor pagination: expands ROW() tuple comparison into OR clauses so
+	// TimescaleDB ColumnarScan can push filters into vectorized batch processing.
+	if cursor != nil {
+		clause, cursorArgs, nextIdx := buildDecomposedCursorCondition([]CursorColumn{
+			{Name: "ledger_created_at", Value: cursor.LedgerCreatedAt},
+			{Name: "operation_id", Value: cursor.ID},
+		}, orderBy, argIndex)
+		queryBuilder.WriteString("\n\t\t\tAND " + clause)
+		args = append(args, cursorArgs...)
+		argIndex = nextIdx
+	}
+
+	if orderBy == DESC {
+		queryBuilder.WriteString(`
+			ORDER BY ledger_created_at DESC, operation_id DESC`)
+	} else {
+		queryBuilder.WriteString(`
+			ORDER BY ledger_created_at ASC, operation_id ASC`)
+	}
+
+	if limit != nil {
+		queryBuilder.WriteString(fmt.Sprintf(` LIMIT $%d`, argIndex))
+		args = append(args, *limit)
+	}
 
-	// Build paginated query using shared utility
-	query, args := buildGetByAccountAddressQuery(paginatedQueryConfig{
-		TableName:      "operations",
-		CursorColumn:   "id",
-		JoinTable:      "operations_accounts",
-		JoinCondition:  "operations_accounts.operation_id = operations.id",
-		Columns:        columns,
-		AccountAddress: accountAddress,
-		Limit:          limit,
-		Cursor:         cursor,
-		OrderBy:        orderBy,
-	})
+	// Close CTE and LATERAL join to fetch full operation rows
+	queryBuilder.WriteString(fmt.Sprintf(`
+		)
+		SELECT %s, o.ledger_created_at as "cursor.cursor_ledger_created_at", o.id as "cursor.cursor_id"
+		FROM account_ops ao,
+		LATERAL (SELECT * FROM operations o WHERE o.id = ao.operation_id AND o.ledger_created_at = ao.ledger_created_at LIMIT 1) o`, columns))
+
+	if orderBy == DESC {
+		queryBuilder.WriteString(`
+		ORDER BY o.ledger_created_at DESC, o.id DESC`)
+	} else {
+		queryBuilder.WriteString(`
+		ORDER BY o.ledger_created_at ASC, o.id ASC`)
+	}
+
+	query := queryBuilder.String()
+
+	// For backward pagination, wrap query to reverse the final order
+	if orderBy == DESC {
+		query = fmt.Sprintf(`SELECT * FROM (%s) AS operations ORDER BY operations."cursor.cursor_ledger_created_at" ASC, operations."cursor.cursor_id" ASC`, query)
+	}
 
 	var operations []*types.OperationWithCursor
 	start := time.Now()

internal/data/operations_test.go

@@ -263,17 +263,17 @@ func TestOperationModel_GetAll(t *testing.T) {
 	operations, err := m.GetAll(ctx, "", nil, nil, ASC)
 	require.NoError(t, err)
 	assert.Len(t, operations, 3)
-	assert.Equal(t, int64(2), operations[0].Cursor)
-	assert.Equal(t, int64(4098), operations[1].Cursor)
-	assert.Equal(t, int64(8194), operations[2].Cursor)
+	assert.Equal(t, int64(2), operations[0].Cursor.ID)
+	assert.Equal(t, int64(4098), operations[1].Cursor.ID)
+	assert.Equal(t, int64(8194), operations[2].Cursor.ID)
 
 	// Test GetAll with smaller limit
 	limit := int32(2)
 	operations, err = m.GetAll(ctx, "", &limit, nil, ASC)
 	require.NoError(t, err)
 	assert.Len(t, operations, 2)
-	assert.Equal(t, int64(2), operations[0].Cursor)
-	assert.Equal(t, int64(4098), operations[1].Cursor)
+	assert.Equal(t, int64(2), operations[0].Cursor.ID)
+	assert.Equal(t, int64(4098), operations[1].Cursor.ID)
 }
 
 func TestOperationModel_BatchGetByToIDs(t *testing.T) {
@@ -582,7 +582,7 @@ func TestOperationModel_BatchGetByAccountAddresses(t *testing.T) {
 	require.NoError(t, err)
 
 	// Test BatchGetByAccount
-	operations, err := m.BatchGetByAccountAddress(ctx, address1, "", nil, nil, "ASC")
+	operations, err := m.BatchGetByAccountAddress(ctx, address1, "", nil, nil, ASC, nil)
 	require.NoError(t, err)
 	assert.Len(t, operations, 2)
 	assert.Equal(t, int64(4097), operations[0].Operation.ID)