Spanner数据更新 #

Spanner数据更新方式
├── DML语句 (UPDATE)
│   ├── 标准SQL语法
│   ├── 支持条件更新
│   ├── 实时执行
│   └── 返回影响行数
│
└── Mutation API
    ├── 编程接口
    ├── 批量操作
    ├── 需要先读取
    └── 适合大量数据

UPDATE table_name
SET column1 = value1, column2 = value2, ...
WHERE condition;

CREATE TABLE users (
    user_id INT64 NOT NULL,
    name STRING(100) NOT NULL,
    email STRING(255),
    age INT64,
    status STRING(20) DEFAULT 'active',
    updated_at TIMESTAMP
) PRIMARY KEY (user_id);

-- 更新单个字段
UPDATE users SET name = 'John Smith' WHERE user_id = 1;

-- 更新多个字段
UPDATE users 
SET name = 'John Smith', 
    email = 'john.smith@example.com',
    age = 31
WHERE user_id = 1;

-- 使用表达式更新
UPDATE users 
SET age = age + 1 
WHERE user_id = 1;

-- 使用函数更新
UPDATE users 
SET updated_at = CURRENT_TIMESTAMP() 
WHERE user_id = 1;

-- 使用比较条件
UPDATE users SET status = 'inactive' WHERE age > 60;

-- 使用IN条件
UPDATE users SET status = 'premium' 
WHERE user_id IN (1, 2, 3, 4, 5);

-- 使用子查询
UPDATE users SET status = 'vip'
WHERE user_id IN (
    SELECT user_id FROM orders 
    GROUP BY user_id 
    HAVING SUM(total_amount) > 10000
);

-- 使用LIKE条件
UPDATE users SET status = 'verified'
WHERE email LIKE '%@company.com';

-- 更新所有行(谨慎使用)
UPDATE users SET status = 'active' WHERE TRUE;

-- 或
UPDATE users SET status = 'active';

// Java Update Mutation
import com.google.cloud.spanner.Mutation;
import com.google.cloud.spanner.DatabaseClient;

// 创建Update Mutation(需要指定主键和所有非NULL列)
Mutation mutation = Mutation.newUpdateBuilder("users")
    .set("user_id").to(1L)
    .set("name").to("John Smith")
    .set("email").to("john.smith@example.com")
    .set("age").to(31)
    .set("status").to("active")
    .build();

// 执行更新
DatabaseClient client = spanner.getDatabaseClient(databaseId);
client.write(Collections.singletonList(mutation));

// InsertOrUpdate: 存在则更新，不存在则插入
Mutation mutation = Mutation.newInsertOrUpdateBuilder("users")
    .set("user_id").to(1L)
    .set("name").to("John Smith")
    .set("email").to("john.smith@example.com")
    .build();

# Python更新
def update_user(database, user_id, name, email):
    with database.batch() as batch:
        batch.update(
            table="users",
            columns=("user_id", "name", "email"),
            values=[
                (user_id, name, email)
            ]
        )

// Go更新
func updateUser(ctx context.Context, client *spanner.Client, userID int64, name, email string) error {
    m := spanner.Update("users",
        []string{"user_id", "name", "email"},
        []interface{}{userID, name, email},
    )
    
    _, err := client.Apply(ctx, []*spanner.Mutation{m})
    return err
}

-- 使用IN条件批量更新
UPDATE users SET status = 'inactive'
WHERE user_id IN (1, 2, 3, 4, 5);

-- 使用子查询批量更新
UPDATE users SET status = 'premium'
WHERE user_id IN (
    SELECT user_id FROM premium_users
);

// Java批量更新
List<Mutation> mutations = new ArrayList<>();

for (long userId : userIds) {
    Mutation mutation = Mutation.newUpdateBuilder("users")
        .set("user_id").to(userId)
        .set("status").to("inactive")
        .build();
    mutations.add(mutation);
}

DatabaseClient client = spanner.getDatabaseClient(databaseId);
client.write(mutations);

# Python批量更新
def batch_update_status(database, user_ids, status):
    values = [(uid, status) for uid in user_ids]
    with database.batch() as batch:
        batch.update(
            table="users",
            columns=("user_id", "status"),
            values=values
        )

批量更新限制:
├── 单次Mutation数量: 无限制
├── 单次数据大小: 100MB
├── 单行大小: 10GB
└── 建议: 每批1000-10000行

// Java读写事务
DatabaseClient client = spanner.getDatabaseClient(databaseId);

TransactionRunner runner = client.readWriteTransaction();
runner.run(transaction -> {
    // 先读取当前值
    Struct row = transaction.readRow("users", 
        Key.of(1L), Arrays.asList("age"));
    
    long currentAge = row.getLong(0);
    
    // 更新
    transaction.buffer(Mutation.newUpdateBuilder("users")
        .set("user_id").to(1L)
        .set("age").to(currentAge + 1)
        .build());
    
    return null;
});

// 条件更新: 只有满足条件才更新
TransactionRunner runner = client.readWriteTransaction();
runner.run(transaction -> {
    // 读取并检查条件
    Struct row = transaction.readRow("users", 
        Key.of(1L), Arrays.asList("status", "age"));
    
    String status = row.getString(0);
    long age = row.getLong(1);
    
    // 只有active用户才能更新年龄
    if ("active".equals(status)) {
        transaction.buffer(Mutation.newUpdateBuilder("users")
            .set("user_id").to(1L)
            .set("age").to(age + 1)
            .build());
    }
    
    return null;
});

# Python事务更新
def update_user_age(database, user_id):
    def transaction_func(transaction):
        # 读取当前年龄
        row = list(transaction.read(
            "users", ["age"], 
            spanner.KeySet([spanner.Key([user_id])])
        ))[0]
        
        current_age = row[0]
        
        # 更新年龄
        transaction.update(
            table="users",
            columns=("user_id", "age"),
            values=[(user_id, current_age + 1)]
        )
    
    database.run_in_transaction(transaction_func)

-- 只更新指定列
UPDATE users SET email = 'new@example.com' WHERE user_id = 1;

-- 其他列保持不变

// Mutation更新时，未指定的列会被设置为NULL
// 如果需要保持原值，必须先读取

// 方式1: 先读取再更新
Struct row = client.singleUse().readRow("users", 
    Key.of(1L), Arrays.asList("name", "email", "age"));

Mutation mutation = Mutation.newUpdateBuilder("users")
    .set("user_id").to(1L)
    .set("name").to(row.getString(0))      // 保持原值
    .set("email").to("new@example.com")    // 更新
    .set("age").to(row.getLong(2))         // 保持原值
    .build();

批量更新建议:
├── 使用Mutation批量更新
├── 合理设置批量大小
├── 避免单行更新循环
└── 使用并发更新

-- 创建索引加速更新条件的查询
CREATE INDEX idx_users_status ON users(status);

-- 更新时利用索引
UPDATE users SET age = age + 1 
WHERE status = 'active';  -- 使用索引

-- 不推荐: 全表扫描
UPDATE users SET status = 'inactive' WHERE TRUE;

-- 推荐: 使用条件限制范围
UPDATE users SET status = 'inactive' 
WHERE last_login < TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 1 YEAR);

-- 使用版本号实现乐观锁
CREATE TABLE products (
    product_id INT64 NOT NULL,
    name STRING(100),
    stock INT64,
    version INT64 DEFAULT 0
) PRIMARY KEY (product_id);

-- 更新时检查版本
UPDATE products 
SET stock = stock - 1, version = version + 1
WHERE product_id = 1 AND version = 5;

-- 如果影响行数为0，说明版本不匹配

// Java重试策略
int maxRetries = 3;
for (int i = 0; i < maxRetries; i++) {
    try {
        runner.run(transaction -> {
            // 读取
            Struct row = transaction.readRow("products", 
                Key.of(1L), Arrays.asList("stock", "version"));
            
            long stock = row.getLong(0);
            long version = row.getLong(1);
            
            // 更新
            transaction.buffer(Mutation.newUpdateBuilder("products")
                .set("product_id").to(1L)
                .set("stock").to(stock - 1)
                .set("version").to(version + 1)
                .build());
            
            return null;
        });
        break;  // 成功则退出
    } catch (AbortedException e) {
        if (i == maxRetries - 1) {
            throw e;
        }
        // 等待后重试
        Thread.sleep(100 * (i + 1));
    }
}

选择建议:
├── 条件更新: DML
├── 批量更新: Mutation
├── 需要返回值: DML
└── 高吞吐量: Mutation

性能优化建议:
├── 使用批量更新
├── 减少事务大小
├── 合理设计索引
├── 避免热点主键
└── 使用并发更新

数据一致性建议:
├── 使用事务保证原子性
├── 使用乐观锁处理并发
├── 处理更新冲突
└── 设置合理的重试策略

1. 选择合适的更新方式
   └── 根据场景选择DML或Mutation

2. 使用批量更新
   └── 减少网络往返次数

3. 处理并发冲突
   └── 使用乐观锁和重试

4. 监控更新性能
   └── 及时发现瓶颈

5. 保证数据一致性
   └── 使用事务