Files
9abd4173f2 MM-68076 Chunk bulk INSERTs to respect PostgreSQL parameter limit (#35767) (#35876)
* MM-68076 Chunk bulk INSERTs to respect PostgreSQL 65,535 parameter limit (#35761)

PostgreSQL's wire protocol uses a 16-bit integer for parameter count,
  causing bulk imports to fail when multi-row INSERTs exceed 65,535
  parameters. Add a generic chunkSlice helper that splits rows into
  sub-batches capped at 50,000 parameters, and apply it to
  saveMultipleMembers (channel), SaveMultipleMembers (team), and
  SaveMultipleMemberships (thread). Normal operations (< 3,333 rows)
  remain a single INSERT with negligible overhead.

  Wrap all chunked INSERT loops in transactions so multi-chunk batches
  are atomic — previously channel and team member inserts could leave
  partial data if a later chunk failed. Add threadMembershipSliceColumns
  helper so thread membership chunk size is derived dynamically.

  Includes integration tests for multi-chunk insertion and rollback
  verification for channel members, team members, posts, and groups.

(cherry picked from commit 4f1f6ef847)

Co-authored-by: Doug Lauder <[email protected]>
2026-03-31 01:08:01 -04:00

194 lines
5.4 KiB
Go

// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved.
// See LICENSE.txt for license information.
package sqlstore
import (
"database/sql"
"fmt"
"time"
sq "github.com/mattermost/squirrel"
"github.com/pkg/errors"
"github.com/mattermost/mattermost/server/public/model"
"github.com/mattermost/mattermost/server/v8/channels/store"
)
type SqlStatusStore struct {
*SqlStore
statusSelectQuery sq.SelectBuilder
}
func statusSliceColumns() []string {
return []string{"UserId", "Status", "Manual", "LastActivityAt", "DNDEndTime", "PrevStatus"}
}
func statusToSlice(st *model.Status) []any {
return []any{st.UserId, st.Status, st.Manual, st.LastActivityAt, st.DNDEndTime, st.PrevStatus}
}
func newSqlStatusStore(sqlStore *SqlStore) store.StatusStore {
s := SqlStatusStore{
SqlStore: sqlStore,
}
s.statusSelectQuery = s.getQueryBuilder().
Select(
"COALESCE(UserId, '') AS UserId",
"COALESCE(Status, '') AS Status",
"COALESCE(Manual, FALSE) AS Manual",
"COALESCE(LastActivityAt, 0) AS LastActivityAt",
"COALESCE(DNDEndTime, 0) AS DNDEndTime",
"COALESCE(PrevStatus, '') AS PrevStatus",
).
From("Status")
return &s
}
func (s SqlStatusStore) SaveOrUpdate(st *model.Status) error {
query := s.getQueryBuilder().
Insert("Status").
Columns(statusSliceColumns()...).
Values(statusToSlice(st)...)
query = query.SuffixExpr(sq.Expr("ON CONFLICT (userid) DO UPDATE SET Status = EXCLUDED.Status, Manual = EXCLUDED.Manual, LastActivityAt = EXCLUDED.LastActivityAt, DNDEndTime = EXCLUDED.DNDEndTime, PrevStatus = EXCLUDED.PrevStatus"))
if _, err := s.GetMaster().ExecBuilder(query); err != nil {
return errors.Wrap(err, "failed to upsert Status")
}
return nil
}
func (s SqlStatusStore) SaveOrUpdateMany(statuses map[string]*model.Status) (retErr error) {
if len(statuses) == 0 {
return nil
}
// If there's only one status, use the existing method
if len(statuses) == 1 {
for _, st := range statuses {
return s.SaveOrUpdate(st)
}
}
statusList := make([]*model.Status, 0, len(statuses))
for _, st := range statuses {
statusList = append(statusList, st)
}
transaction, err := s.GetMaster().Beginx()
if err != nil {
return errors.Wrap(err, "begin_transaction")
}
defer finalizeTransactionX(transaction, &retErr)
chunks := chunkSlice(statusList, len(statusSliceColumns()), s.SqlStore.getMaxInsertParams())
for _, chunk := range chunks {
query := s.getQueryBuilder().
Insert("Status").
Columns(statusSliceColumns()...)
for _, st := range chunk {
query = query.Values(statusToSlice(st)...)
}
query = query.SuffixExpr(sq.Expr("ON CONFLICT (userid) DO UPDATE SET Status = EXCLUDED.Status, Manual = EXCLUDED.Manual, LastActivityAt = EXCLUDED.LastActivityAt, DNDEndTime = EXCLUDED.DNDEndTime, PrevStatus = EXCLUDED.PrevStatus"))
if _, err := transaction.ExecBuilder(query); err != nil {
return errors.Wrap(err, "failed to upsert multiple Status records")
}
}
if err := transaction.Commit(); err != nil {
return errors.Wrap(err, "commit_transaction")
}
return nil
}
func (s SqlStatusStore) Get(userId string) (*model.Status, error) {
query := s.statusSelectQuery.Where(sq.Eq{"UserId": userId})
var status model.Status
if err := s.GetReplica().GetBuilder(&status, query); err != nil {
if err == sql.ErrNoRows {
return nil, store.NewErrNotFound("Status", fmt.Sprintf("userId=%s", userId))
}
return nil, errors.Wrapf(err, "failed to get Status with userId=%s", userId)
}
return &status, nil
}
func (s SqlStatusStore) GetByIds(userIds []string) ([]*model.Status, error) {
query := s.statusSelectQuery.Where(sq.Eq{"UserId": userIds})
statuses := []*model.Status{}
err := s.GetReplica().SelectBuilder(&statuses, query)
if err != nil {
return nil, errors.Wrap(err, "failed to find Statuses")
}
return statuses, nil
}
func (s SqlStatusStore) UpdateExpiredDNDStatuses() (_ []*model.Status, err error) {
queryString := s.getQueryBuilder().
Update("Status").
Where(
sq.And{
sq.Eq{"Status": model.StatusDnd},
sq.Gt{"DNDEndTime": 0},
sq.LtOrEq{"DNDEndTime": time.Now().UTC().Unix()},
},
).
Set("Status", sq.Expr("PrevStatus")).
Set("PrevStatus", model.StatusDnd).
Set("DNDEndTime", 0).
Set("Manual", false).
Suffix("RETURNING *")
statuses := []*model.Status{}
err = s.GetMaster().SelectBuilder(&statuses, queryString)
if err != nil {
return nil, errors.Wrap(err, "failed to find Statuses")
}
return statuses, nil
}
func (s SqlStatusStore) ResetAll() error {
if _, err := s.GetMaster().Exec("UPDATE Status SET Status = ? WHERE Manual = false", model.StatusOffline); err != nil {
return errors.Wrap(err, "failed to update Statuses")
}
return nil
}
func (s SqlStatusStore) GetTotalActiveUsersCount() (int64, error) {
time := model.GetMillis() - (1000 * 60 * 60 * 24)
query := s.getQueryBuilder().
Select("COUNT(UserId)").
From("Status").
Where(sq.Gt{"LastActivityAt": time})
var count int64
err := s.GetReplica().GetBuilder(&count, query)
if err != nil {
return count, errors.Wrap(err, "failed to count active users")
}
return count, nil
}
func (s SqlStatusStore) UpdateLastActivityAt(userId string, lastActivityAt int64) error {
builder := s.getQueryBuilder().
Update("Status").
Set("LastActivityAt", lastActivityAt).
Where(sq.Eq{"UserId": userId})
if _, err := s.GetMaster().ExecBuilder(builder); err != nil {
return errors.Wrapf(err, "failed to update last activity for userId=%s", userId)
}
return nil
}