| 143 | /// Returns `CommitMessage`s for the caller to commit via [`TableCommit`](super::TableCommit). |
| 144 | #[must_use = "commit messages must be passed to TableCommit"] |
| 145 | pub async fn prepare_commit(self) -> Result<Vec<CommitMessage>> { |
| 146 | let total_matched: usize = self.matched_batches.iter().map(|b| b.num_rows()).sum(); |
| 147 | if total_matched == 0 { |
| 148 | return Ok(Vec::new()); |
| 149 | } |
| 150 | |
| 151 | // 1. Scan file metadata and build row_id -> file group index. |
| 152 | // In data-evolution tables, multiple files can share the same first_row_id |
| 153 | // (base file + partial-column files). We must group them so the reader |
| 154 | // can merge columns correctly. |
| 155 | let scan = self.table.new_read_builder().new_scan(); |
| 156 | let plan = scan.plan().await?; |
| 157 | |
| 158 | let mut file_index: Vec<FileRowRange> = Vec::new(); |
| 159 | for split in plan.splits() { |
| 160 | let partition_bytes = split.partition().to_serialized_bytes(); |
| 161 | let bucket = split.bucket(); |
| 162 | let bucket_path = split.bucket_path().to_string(); |
| 163 | let snapshot_id = split.snapshot_id(); |
| 164 | let total_buckets = split.total_buckets(); |
| 165 | |
| 166 | let all_files: Vec<DataFileMeta> = split |
| 167 | .data_files() |
| 168 | .iter() |
| 169 | .filter(|f| f.first_row_id.is_some()) |
| 170 | .cloned() |
| 171 | .collect(); |
| 172 | |
| 173 | let groups = group_by_overlapping_row_id(all_files); |
| 174 | for group in groups { |
| 175 | // Compute the overall row_id range for this group. |
| 176 | // The base file has the widest range; partial-column files share it. |
| 177 | let first_row_id = group.iter().filter_map(|f| f.first_row_id).min().unwrap(); |
| 178 | let last_row_id = group |
| 179 | .iter() |
| 180 | .filter_map(|f| f.row_id_range().map(|(_, end)| end)) |
| 181 | .max() |
| 182 | .unwrap(); |
| 183 | // The actual row count is the max among the group (base file's count). |
| 184 | let row_count = group.iter().map(|f| f.row_count).max().unwrap(); |
| 185 | |
| 186 | file_index.push(FileRowRange { |
| 187 | first_row_id, |
| 188 | last_row_id, |
| 189 | row_count, |
| 190 | partition: partition_bytes.clone(), |
| 191 | bucket, |
| 192 | bucket_path: bucket_path.clone(), |
| 193 | snapshot_id, |
| 194 | total_buckets, |
| 195 | files: group, |
| 196 | }); |
| 197 | } |
| 198 | } |
| 199 | file_index.sort_by_key(|f| f.first_row_id); |
| 200 | |
| 201 | if file_index.is_empty() { |
| 202 | return Err(crate::Error::DataInvalid { |