DynamoDB高级查询 #

DynamoDB查询能力：
├── 键条件查询
├── 过滤表达式
├── 投影表达式
├── 排序和分页
├── 索引查询
└── 批量查询

查询限制：
├── 不支持JOIN
├── 不支持GROUP BY
├── 不支持HAVING
├── 排序键排序有限制
└── 过滤不减少RCU

// 日期范围
KeyConditionExpression: 'PK = :pk AND CreatedAt BETWEEN :start AND :end'

// 数值范围
KeyConditionExpression: 'PK = :pk AND Score BETWEEN :min AND :max'

// 字符串范围
KeyConditionExpression: 'PK = :pk AND Name BETWEEN :a AND :z'

// 查询特定前缀的订单
const response = await docClient.send(new QueryCommand({
  TableName: 'Orders',
  KeyConditionExpression: 'UserId = :uid AND begins_with(OrderId, :prefix)',
  ExpressionAttributeValues: {
    ':uid': 'user123',
    ':prefix': 'ORDER#2024-01'
  }
}));

// 排序键设计：STATUS#PRIORITY#TIMESTAMP
// 示例：OPEN#HIGH#2024-01-01T10:00:00

// 查询高优先级的开放任务
const response = await docClient.send(new QueryCommand({
  TableName: 'Tasks',
  KeyConditionExpression: 'ProjectId = :pid AND begins_with(SK, :prefix)',
  ExpressionAttributeValues: {
    ':pid': 'project001',
    ':prefix': 'OPEN#HIGH'
  }
}));

// 检查属性存在
FilterExpression: 'attribute_exists(Email)'

// 检查属性不存在
FilterExpression: 'attribute_not_exists(DeletedAt)'

// 检查属性类型
FilterExpression: 'attribute_type(Age, :type)',
ExpressionAttributeValues: { ':type': 'N' }

// 前缀匹配
FilterExpression: 'begins_with(Name, :prefix)'

// 包含子串
FilterExpression: 'contains(Description, :keyword)'

// 组合字符串条件
FilterExpression: 'begins_with(Email, :prefix) AND contains(Email, :domain)'

// 检查集合包含元素
FilterExpression: 'contains(Tags, :tag)'

// 检查多个标签（OR）
FilterExpression: 'contains(Tags, :tag1) OR contains(Tags, :tag2)'

// 检查属性是否在集合中
FilterExpression: 'Status IN (:s1, :s2, :s3)'

// 字符串长度
FilterExpression: 'size(Description) < :maxSize'

// 列表长度
FilterExpression: 'size(Items) > :minItems'

// 检查非空字符串
FilterExpression: 'size(Name) > :zero',
ExpressionAttributeValues: { ':zero': 0 }

// 过滤嵌套Map属性
FilterExpression: 'Address.City = :city'

// 过滤嵌套List属性
FilterExpression: 'contains(Items[0].Category, :category)'

// 深层嵌套
FilterExpression: 'Profile.Contacts.Email = :email'

const response = await docClient.send(new QueryCommand({
  TableName: 'Users',
  KeyConditionExpression: '#pk = :pk',
  FilterExpression: '#n = :name AND #s = :status',
  ExpressionAttributeNames: {
    '#pk': 'UserId',
    '#n': 'Name',      // Name是保留字
    '#s': 'Status'     // Status是保留字
  },
  ExpressionAttributeValues: {
    ':pk': 'user123',
    ':name': 'John Doe',
    ':status': 'active'
  }
}));

DynamoDB保留字：
├── Name, Names
├── Status, State
├── Date, Year, Month, Day
├── User, Order, Key, Value
├── Data, Timestamp
├── Index, Table, Item
├── Number, String, Boolean
└── 更多见AWS文档

function buildQueryWithDynamicField(fieldName, value) {
  return {
    TableName: 'Users',
    KeyConditionExpression: 'UserId = :pk',
    FilterExpression: '#field = :value',
    ExpressionAttributeNames: {
      '#field': fieldName
    },
    ExpressionAttributeValues: {
      ':pk': 'user123',
      ':value': value
    }
  };
}

// 表设计：
// PK: ENTITY#ID
// SK: TYPE#SUBTYPE

// 查询用户所有数据
async function getUserData(userId) {
  const response = await docClient.send(new QueryCommand({
    TableName: 'AppTable',
    KeyConditionExpression: 'PK = :pk',
    ExpressionAttributeValues: {
      ':pk': `USER#${userId}`
    }
  }));
  
  // 按类型分组
  const grouped = {
    profile: null,
    orders: [],
    addresses: []
  };
  
  for (const item of response.Items) {
    if (item.SK === 'PROFILE') {
      grouped.profile = item;
    } else if (item.SK.startsWith('ORDER#')) {
      grouped.orders.push(item);
    } else if (item.SK.startsWith('ADDRESS#')) {
      grouped.addresses.push(item);
    }
  }
  
  return grouped;
}

// 设计：PK = DEVICE#ID, SK = TIMESTAMP

// 查询最近N条数据
async function getRecentData(deviceId, limit = 100) {
  const response = await docClient.send(new QueryCommand({
    TableName: 'TimeSeries',
    KeyConditionExpression: 'PK = :pk',
    ExpressionAttributeValues: {
      ':pk': `DEVICE#${deviceId}`
    },
    ScanIndexForward: false,  // 降序
    Limit: limit
  }));
  
  return response.Items;
}

// 查询时间范围数据
async function getDataInRange(deviceId, startTime, endTime) {
  const response = await docClient.send(new QueryCommand({
    TableName: 'TimeSeries',
    KeyConditionExpression: 'PK = :pk AND SK BETWEEN :start AND :end',
    ExpressionAttributeValues: {
      ':pk': `DEVICE#${deviceId}`,
      ':start': startTime,
      ':end': endTime
    }
  }));
  
  return response.Items;
}

// 设计：
// PK = PARENT#ID
// SK = CHILD#TYPE#ID

// 查询所有子项
async function getChildren(parentId, childType = null) {
  const params = {
    TableName: 'Hierarchy',
    KeyConditionExpression: 'PK = :pk',
    ExpressionAttributeValues: {
      ':pk': `PARENT#${parentId}`
    }
  };
  
  if (childType) {
    params.KeyConditionExpression += ' AND begins_with(SK, :type)';
    params.ExpressionAttributeValues[':type'] = `CHILD#${childType}`;
  }
  
  return await docClient.send(new QueryCommand(params));
}

// 通过Email查询用户
async function getUserByEmail(email) {
  const response = await docClient.send(new QueryCommand({
    TableName: 'Users',
    IndexName: 'EmailIndex',
    KeyConditionExpression: 'Email = :email',
    ExpressionAttributeValues: {
      ':email': email
    }
  }));
  
  return response.Items[0];
}

// 一个GSI支持多种查询
// GSI1PK = QUERY_TYPE#VALUE
// GSI1SK = TIMESTAMP

// 按状态查询订单
async function getOrdersByStatus(status, limit = 100) {
  return await docClient.send(new QueryCommand({
    TableName: 'Orders',
    IndexName: 'GSI1',
    KeyConditionExpression: 'GSI1PK = :pk',
    ExpressionAttributeValues: {
      ':pk': `STATUS#${status}`
    },
    ScanIndexForward: false,
    Limit: limit
  }));
}

// 按日期查询订单
async function getOrdersByDate(date, limit = 100) {
  return await docClient.send(new QueryCommand({
    TableName: 'Orders',
    IndexName: 'GSI1',
    KeyConditionExpression: 'GSI1PK = :pk',
    ExpressionAttributeValues: {
      ':pk': `DATE#${date}`
    },
    ScanIndexForward: false,
    Limit: limit
  }));
}

// 只有有Phone属性的用户会被索引
async function getUsersWithPhone(phoneNumber) {
  const response = await docClient.send(new QueryCommand({
    TableName: 'Users',
    IndexName: 'PhoneIndex',
    KeyConditionExpression: 'Phone = :phone',
    ExpressionAttributeValues: {
      ':phone': phoneNumber
    }
  }));
  
  return response.Items;
}

// 不好的做法：大量过滤
const response = await docClient.send(new QueryCommand({
  TableName: 'Orders',
  KeyConditionExpression: 'UserId = :uid',
  FilterExpression: 'Status = :status AND TotalAmount > :min',
  ExpressionAttributeValues: {
    ':uid': 'user123',
    ':status': 'COMPLETED',
    ':min': 100
  }
}));

// 好的做法：使用索引
// 创建GSI：PK=Status, SK=TotalAmount
const response = await docClient.send(new QueryCommand({
  TableName: 'Orders',
  IndexName: 'StatusAmountIndex',
  KeyConditionExpression: 'Status = :status AND TotalAmount > :min',
  ExpressionAttributeValues: {
    ':status': 'COMPLETED',
    ':min': 100
  }
}));

// 只获取需要的属性
const response = await docClient.send(new QueryCommand({
  TableName: 'Users',
  KeyConditionExpression: 'UserId = :uid',
  ProjectionExpression: 'UserId, Name, Email',
  ExpressionAttributeValues: {
    ':uid': 'user123'
  }
}));

// 并行查询多个分区
async function batchQuery(partitionKeys) {
  const promises = partitionKeys.map(pk =>
    docClient.send(new QueryCommand({
      TableName: 'Orders',
      KeyConditionExpression: 'PK = :pk',
      ExpressionAttributeValues: { ':pk': pk }
    }))
  );
  
  const results = await Promise.all(promises);
  return results.flatMap(r => r.Items);
}

async function paginatedQuery(params) {
  const { pageSize = 20, lastKey } = params;
  
  const response = await docClient.send(new QueryCommand({
    TableName: params.tableName,
    KeyConditionExpression: params.keyCondition,
    FilterExpression: params.filterExpression,
    ExpressionAttributeValues: params.values,
    Limit: pageSize,
    ExclusiveStartKey: lastKey
  }));
  
  return {
    items: response.Items,
    lastKey: response.LastEvaluatedKey,
    hasMore: !!response.LastEvaluatedKey
  };
}

async function searchItems(tableName, searchTerm, options = {}) {
  const { fields = ['Name', 'Description'], limit = 20 } = options;
  
  // 构建过滤表达式
  const filterParts = fields.map((field, i) => `contains(#f${i}, :term)`);
  const expressionAttributeNames = {};
  fields.forEach((field, i) => {
    expressionAttributeNames[`#f${i}`] = field;
  });
  
  const response = await docClient.send(new ScanCommand({
    TableName: tableName,
    FilterExpression: filterParts.join(' OR '),
    ExpressionAttributeNames,
    ExpressionAttributeValues: {
      ':term': searchTerm
    },
    Limit: limit
  }));
  
  return response.Items;
}

async function aggregateByField(tableName, fieldName, keyCondition, values) {
  const counts = {};
  let lastKey = null;
  
  do {
    const response = await docClient.send(new QueryCommand({
      TableName: tableName,
      KeyConditionExpression: keyCondition,
      ExpressionAttributeValues: values,
      ExclusiveStartKey: lastKey
    }));
    
    for (const item of response.Items) {
      const fieldValue = item[fieldName];
      counts[fieldValue] = (counts[fieldValue] || 0) + 1;
    }
    
    lastKey = response.LastEvaluatedKey;
  } while (lastKey);
  
  return counts;
}

// 使用示例
const statusCounts = await aggregateByField(
  'Orders',
  'Status',
  'UserId = :uid',
  { ':uid': 'user123' }
);
console.log(statusCounts);
// { PENDING: 5, COMPLETED: 10, CANCELLED: 2 }

async function safeQuery(params) {
  try {
    return await docClient.send(new QueryCommand(params));
  } catch (error) {
    switch (error.name) {
      case 'ProvisionedThroughputExceededException':
        console.error('超过预置吞吐量');
        break;
      case 'ResourceNotFoundException':
        console.error('表或索引不存在');
        break;
      case 'ValidationException':
        console.error('参数验证失败:', error.message);
        break;
      default:
        console.error('查询错误:', error);
    }
    throw error;
  }
}

async function queryWithRetry(params, maxRetries = 3) {
  for (let i = 0; i < maxRetries; i++) {
    try {
      return await docClient.send(new QueryCommand(params));
    } catch (error) {
      if (error.name === 'ProvisionedThroughputExceededException' && i < maxRetries - 1) {
        await new Promise(resolve => setTimeout(resolve, Math.pow(2, i) * 100));
        continue;
      }
      throw error;
    }
  }
}

设计原则：
├── 以访问模式为中心
├── 使用索引支持查询
├── 避免过度使用Scan
├── 合理设计排序键
└── 使用投影减少数据传输

性能建议：
├── 使用Query而非Scan
├── 使用索引覆盖查询
├── 合理设置Limit
├── 使用并行查询
└── 监控查询性能