Encode document rows as a Parquet file and upload to cold storage. Returns the object path where the Parquet file was stored.
(
&self,
collection: &str,
tenant_id: u64,
rows: &[(String, serde_json::Value)],
min_lsn: u64,
max_lsn: u64,
)
| 181 | /// |
| 182 | /// Returns the object path where the Parquet file was stored. |
| 183 | pub async fn encode_and_upload( |
| 184 | &self, |
| 185 | collection: &str, |
| 186 | tenant_id: u64, |
| 187 | rows: &[(String, serde_json::Value)], |
| 188 | min_lsn: u64, |
| 189 | max_lsn: u64, |
| 190 | ) -> crate::Result<String> { |
| 191 | if rows.is_empty() { |
| 192 | return Err(crate::Error::BadRequest { |
| 193 | detail: "no rows to encode".into(), |
| 194 | }); |
| 195 | } |
| 196 | |
| 197 | // Build Arrow schema from first row. |
| 198 | let first_obj = rows[0] |
| 199 | .1 |
| 200 | .as_object() |
| 201 | .ok_or_else(|| crate::Error::ColdStorage { |
| 202 | detail: "first row is not an object".into(), |
| 203 | })?; |
| 204 | |
| 205 | let mut fields = vec![Field::new("_id", DataType::Utf8, false)]; |
| 206 | for (key, value) in first_obj { |
| 207 | let dt = match value { |
| 208 | serde_json::Value::Number(n) if n.is_i64() => DataType::Int64, |
| 209 | serde_json::Value::Number(_) => DataType::Float64, |
| 210 | _ => DataType::Utf8, |
| 211 | }; |
| 212 | fields.push(Field::new(key, dt, true)); |
| 213 | } |
| 214 | let schema = Arc::new(Schema::new(fields)); |
| 215 | |
| 216 | // Build column arrays. |
| 217 | let field_names: Vec<String> = first_obj.keys().cloned().collect(); |
| 218 | let mut ids: Vec<String> = Vec::with_capacity(rows.len()); |
| 219 | let mut columns: Vec<Vec<serde_json::Value>> = |
| 220 | vec![Vec::with_capacity(rows.len()); field_names.len()]; |
| 221 | |
| 222 | for (doc_id, data) in rows { |
| 223 | ids.push(doc_id.clone()); |
| 224 | let obj = data.as_object(); |
| 225 | for (i, name) in field_names.iter().enumerate() { |
| 226 | let val = obj |
| 227 | .and_then(|o| o.get(name)) |
| 228 | .cloned() |
| 229 | .unwrap_or(serde_json::Value::Null); |
| 230 | columns[i].push(val); |
| 231 | } |
| 232 | } |
| 233 | |
| 234 | let mut arrays: Vec<ArrayRef> = vec![Arc::new(StringArray::from(ids))]; |
| 235 | for (i, field) in schema.fields().iter().skip(1).enumerate() { |
| 236 | let arr: ArrayRef = match field.data_type() { |
| 237 | DataType::Int64 => { |
| 238 | let vals: Vec<Option<i64>> = columns[i].iter().map(|v| v.as_i64()).collect(); |
| 239 | Arc::new(Int64Array::from(vals)) |
| 240 | } |