Implementation of `DataEngine::scan`.
(
&self,
key_info: &TableKeyInfo,
limit: Option<i64>,
exclusive_start_key: Option<&Item>,
segment: Option<i64>,
total_segments: Option<i64>,
ind
| 183 | |
| 184 | /// Implementation of `DataEngine::scan`. |
| 185 | pub(crate) async fn scan_impl( |
| 186 | &self, |
| 187 | key_info: &TableKeyInfo, |
| 188 | limit: Option<i64>, |
| 189 | exclusive_start_key: Option<&Item>, |
| 190 | segment: Option<i64>, |
| 191 | total_segments: Option<i64>, |
| 192 | index_name: Option<&str>, |
| 193 | ) -> Result<(Vec<Item>, Option<Item>), StorageError> { |
| 194 | use std::fmt::Write; |
| 195 | |
| 196 | let ddb_table = if let Some(idx_name) = index_name { |
| 197 | let idx_info = self |
| 198 | .fetch_index_info_by_table_id(&key_info.table_id, idx_name) |
| 199 | .await?; |
| 200 | index_table_name(&idx_info.index_id) |
| 201 | } else { |
| 202 | data_table_name(&key_info.table_id) |
| 203 | }; |
| 204 | let sk_info_val = sk_info(&key_info.key_schema, &key_info.attribute_definitions); |
| 205 | |
| 206 | let mut sql = format!("SELECT item_data FROM {ddb_table}"); |
| 207 | let mut conditions: Vec<String> = Vec::new(); |
| 208 | let param_idx: u32 = 1; |
| 209 | |
| 210 | // Parallel scan: hash-based segment assignment. |
| 211 | // CB-20 / SP-SCN-002: use bigint bitmask instead of abs() to avoid |
| 212 | // SQL error 22003 on the one-in-4-billion hashtext() == i32::MIN case. |
| 213 | if let (Some(seg), Some(total)) = (segment, total_segments) { |
| 214 | conditions.push(format!( |
| 215 | "(hashtext(pk)::bigint & 2147483647) % {total} = {seg}" |
| 216 | )); |
| 217 | } |
| 218 | |
| 219 | // Pagination via exclusive start key |
| 220 | if let Some(start_key) = exclusive_start_key { |
| 221 | let pk_name = &key_info.key_schema[0].attribute_name; |
| 222 | if !start_key.contains_key(pk_name) { |
| 223 | return Err(StorageError::Validation( |
| 224 | "The provided starting key is invalid: The provided key element does not match the schema".to_owned(), |
| 225 | )); |
| 226 | } |
| 227 | // Actual PK/SK binding happens in execute_scan_sql. |
| 228 | |
| 229 | if let Some((_, sk_type)) = sk_info_val { |
| 230 | let sk_col = sk_column(sk_type); |
| 231 | let collate = if sk_type == ScalarAttributeType::S { |
| 232 | " COLLATE \"C\"" |
| 233 | } else { |
| 234 | "" |
| 235 | }; |
| 236 | conditions.push(format!( |
| 237 | "(pk, {sk_col}{collate}) > (${param_idx}, ${next})", |
| 238 | next = param_idx + 1 |
| 239 | )); |
| 240 | } else { |
| 241 | conditions.push(format!("pk > ${param_idx}")); |
| 242 | } |
no test coverage detected