MongoDB Upsert时能否仅在记录其他数据变更时设置时间戳字段？

核心思路 #

无需将所有缓存记录加载到内存做对比，直接通过MongoDB的**更新管道（Update Pipeline）**配合Upsert实现状态标记逻辑，所有判断在数据库端执行，既高效又避免内存占用过高。

具体实现步骤 #

1. 定义C#员工数据模型 #

给实体添加同步相关的状态和时间字段：

public class Employee
{
    public ObjectId Id { get; set; }
    // 源系统的员工唯一标识，用于Upsert匹配
    public string SourceEmployeeId { get; set; }
    // 业务字段示例
    public string Name { get; set; }
    public string Department { get; set; }
    // 同步状态：New/Updated/Unchanged
    public string SyncStatus { get; set; }
    // 本次同步的时间戳
    public DateTime LastSyncTime { get; set; }
    // 记录数据最后一次实际变更的时间
    public DateTime? LastModifiedTime { get; set; }
}

2. 使用更新管道实现Upsert逻辑 #

通过MongoDB驱动的聚合管道，在Upsert时自动判断记录状态：

var mongoClient = new MongoClient("mongodb://localhost:27017");
var db = mongoClient.GetDatabase("EmployeeCacheDB");
var empCollection = db.GetCollection<Employee>("Employees");

// 从源API拉取全量员工数据
var sourceEmployees = FetchAllEmployeesFromApi();

// 批量构建Upsert操作（单条或批量都适用）
var bulkOperations = new List<WriteModel<Employee>>();
foreach (var sourceEmp in sourceEmployees)
{
    // 构建更新管道
    var updatePipeline = new PipelineDefinition<Employee, Employee>(new[]
    {
        // 第一步：判断是新增还是已有记录，再判断数据是否变更
        new BsonDocument("$set", new BsonDocument
        {
            ["LastSyncTime"] = DateTime.UtcNow,
            ["SyncStatus"] = new BsonDocument("$cond", new BsonArray
            {
                // 新增判断：_id为空说明是插入操作
                new BsonDocument("$eq", new BsonArray { "$_id", BsonNull.Value }),
                "New",
                // 已有记录则对比业务字段
                new BsonDocument("$cond", new BsonArray
                {
                    new BsonDocument("$or", new BsonArray
                    {
                        new BsonDocument("$ne", new BsonArray { "$Name", sourceEmp.Name }),
                        new BsonDocument("$ne", new BsonArray { "$Department", sourceEmp.Department })
                        // 继续添加其他需要对比的业务字段，排除时间戳类字段
                    }),
                    "Updated",
                    "Unchanged"
                })
            })
        }),
        // 第二步：仅当标记为Updated时更新LastModifiedTime
        new BsonDocument("$set", new BsonDocument
        {
            ["LastModifiedTime"] = new BsonDocument("$cond", new BsonArray
            {
                new BsonDocument("$eq", new BsonArray { "$SyncStatus", "Updated" }),
                DateTime.UtcNow,
                "$LastModifiedTime"
            })
        }),
        // 第三步：覆盖最新的业务字段值
        new BsonDocument("$set", new BsonDocument
        {
            ["Name"] = sourceEmp.Name,
            ["Department"] = sourceEmp.Department
            // 其他业务字段
        })
    });

    // 构建Upsert操作模型
    var upsertModel = new UpdateOneModel<Employee>(
        Builders<Employee>.Filter.Eq(e => e.SourceEmployeeId, sourceEmp.SourceEmployeeId),
        updatePipeline
    ) { IsUpsert = true };
    bulkOperations.Add(upsertModel);
}

// 执行批量Upsert
await empCollection.BulkWriteAsync(bulkOperations);

3. 关键逻辑说明 #

• 新增判断：利用Upsert插入时_id为BsonNull.Value的特性，直接识别新增记录。
• 变更判断：在数据库端对比现有文档和源数据的业务字段，无需拉取到应用内存。
• 性能优化：所有逻辑在MongoDB服务器执行，减少应用层内存占用和网络传输开销。

附加优化 #

• 索引优化：给SourceEmployeeId添加唯一索引，确保Upsert匹配效率：

  await empCollection.Indexes.CreateOneAsync(new CreateIndexModel<Employee>(
      Builders<Employee>.IndexKeys.Ascending(e => e.SourceEmployeeId),
      new CreateIndexOptions { Unique = true }
  ));
  

• 低版本兼容方案：如果MongoDB版本低于4.2（不支持更新管道），只能退化为内存对比：
  1. 从MongoDB拉取所有记录的SourceEmployeeId和业务字段哈希值（提前存储）。
  2. 对源数据每条记录计算哈希，对比现有记录标记状态。
  3. 批量执行Upsert。此方案仅适合数据量较小的场景。

内容的提问来源于stack exchange，提问作者Brian Kessler