(&mut self, partition_bytes: Vec<u8>, bucket: i32)
| 562 | } |
| 563 | |
| 564 | async fn create_writer(&mut self, partition_bytes: Vec<u8>, bucket: i32) -> Result<()> { |
| 565 | let partition_path = self.resolve_partition_path(&partition_bytes)?; |
| 566 | |
| 567 | let writer = if self.primary_key_indices.is_empty() { |
| 568 | self.create_append_writer(partition_path, bucket)? |
| 569 | } else if bucket == POSTPONE_BUCKET { |
| 570 | self.create_postpone_writer(partition_path, bucket) |
| 571 | } else { |
| 572 | self.create_kv_writer(partition_path, bucket, &partition_bytes) |
| 573 | .await? |
| 574 | }; |
| 575 | |
| 576 | self.partition_writers |
| 577 | .insert((partition_bytes, bucket), writer); |
| 578 | Ok(()) |
| 579 | } |
| 580 | |
| 581 | fn resolve_partition_path(&self, partition_bytes: &[u8]) -> Result<String> { |
| 582 | if self.partition_keys.is_empty() { |
no test coverage detected