ifc_lite_server/routes/parse/
parquet.rs

1// This Source Code Form is subject to the terms of the Mozilla Public
2// License, v. 2.0. If a copy of the MPL was not distributed with this
3// file, You can obtain one at https://mozilla.org/MPL/2.0/.
4
5//! Binary Parquet parse endpoints.
6
7use super::cache_keys::{
8    data_model_cache_key, has_current_data_model, has_cached_symbolic, parquet_geometry_key,
9    parquet_metadata_key, request_cache_key,
10};
11use super::replay_header::mark_header_from_cache;
12use super::{cache_symbolic_data_off_runtime, extract_file, ParseQuery};
13use crate::error::ApiError;
14use crate::services::baked_basis_zup;
15use crate::services::parquet::serialize_combined_for_layout;
16use crate::services::{extract_data_model, serialize_data_model_to_parquet};
17use crate::types::{finish_meshes, ModelMetadata, ProcessingStats};
18use ifc_lite_processing::style::ModelFinishes;
19use crate::AppState;
20use axum::{
21    body::Body,
22    extract::{Multipart, Query, State},
23    http::{header, StatusCode},
24    response::Response,
25};
26use ifc_lite_processing::{
27    extract_symbolic_data_with_provenance_in_frame, process_geometry_filtered_with_quality,
28    MeshCoordinateSpace,
29};
30use serde::{Deserialize, Serialize};
31
32/// Response header containing metadata for Parquet response.
33#[derive(Debug, Clone, Serialize, Deserialize)]
34pub struct ParquetMetadataHeader {
35    pub cache_key: String,
36    pub metadata: ModelMetadata,
37    pub stats: ProcessingStats,
38    #[serde(skip_serializing_if = "Option::is_none")]
39    pub mesh_coordinate_space: Option<MeshCoordinateSpace>,
40    #[serde(skip_serializing_if = "Option::is_none")]
41    pub site_transform: Option<Vec<f64>>,
42    #[serde(skip_serializing_if = "Option::is_none")]
43    pub building_transform: Option<Vec<f64>>,
44    /// Data model statistics (if included).
45    #[serde(skip_serializing_if = "Option::is_none")]
46    pub data_model_stats: Option<DataModelStats>,
47}
48
49/// Data model extraction statistics.
50#[derive(Debug, Clone, Serialize, Deserialize)]
51pub struct DataModelStats {
52    pub entity_count: usize,
53    pub property_set_count: usize,
54    pub relationship_count: usize,
55    pub spatial_node_count: usize,
56}
57
58/// POST /api/v1/parse/parquet - Full parse with Parquet-encoded geometry.
59///
60/// Returns binary Parquet data with ~15x smaller payload than JSON.
61/// Response format:
62/// - Content-Type: application/x-parquet-geometry
63/// - X-IFC-Metadata: JSON-encoded ParquetMetadataHeader
64/// - Body: Binary Parquet data (mesh_parquet + vertex_parquet + index_parquet)
65pub async fn parse_parquet(
66    State(state): State<AppState>,
67    Query(query): Query<ParseQuery>,
68    mut multipart: Multipart,
69) -> Result<Response, ApiError> {
70    // Extract file from multipart
71    // Admission gate (bounded concurrency + byte budget): acquired BEFORE the
72    // upload is buffered, reserving the max upload size since multipart rarely
73    // declares a length up front. Held for the request's whole lifetime so a
74    // disconnected-but-still-running job keeps its memory slot.
75    let admission_guard = state
76        .admission
77        .acquire(state.config.max_file_size_mb as u64 * 1024 * 1024)
78        .await?;
79    let data = extract_file(&mut multipart, state.config.max_file_size_mb).await?;
80
81    // Generate cache key (include opening filter so different modes get different cache entries)
82    let tessellation_quality = query.resolved_tessellation_quality()?;
83    let cache_key = request_cache_key(&data, &query, tessellation_quality);
84
85    // Check cache first (before any processing)
86    let layout = query.parquet_layout;
87    let parquet_cache_key = parquet_geometry_key(&cache_key, layout);
88    let metadata_cache_key = parquet_metadata_key(&cache_key);
89
90    // The cached-geometry short-circuit skips the parse, and the parse is what
91    // writes the data model. A geometry entry that outlived a data-model
92    // version bump must fall through so both get rewritten (issue #3869).
93    // The same rule applies to symbolic schema freshness (#4459).
94    if let (Some(cached_parquet), Some(cached_metadata_json), true, true) = (
95        state.cache.get_bytes(&parquet_cache_key).await?,
96        state.cache.get_bytes(&metadata_cache_key).await?,
97        has_current_data_model(&state.cache, &cache_key, query.data_model_entities).await,
98        has_cached_symbolic(&state.cache, &cache_key).await,
99    ) {
100        tracing::info!(
101            cache_key = %cache_key,
102            parquet_size = cached_parquet.len(),
103            "Parquet cache HIT - returning cached response"
104        );
105
106        // Build response from cached data. The stored header is the one the
107        // live parse wrote, `from_cache: false` included (#5542).
108        let response = Response::builder()
109            .status(StatusCode::OK)
110            .header(header::CONTENT_TYPE, "application/x-parquet-geometry")
111            .header(
112                "X-IFC-Metadata",
113                mark_header_from_cache(
114                    String::from_utf8(cached_metadata_json)
115                        .map_err(|error| ApiError::Internal(error.to_string()))?,
116                ),
117            )
118            .header(header::CONTENT_LENGTH, cached_parquet.len())
119            .body(Body::from(cached_parquet))
120            .map_err(|e| ApiError::Internal(e.to_string()))?;
121
122        return Ok(response);
123    }
124
125    tracing::info!(
126        cache_key = %cache_key,
127        size = data.len(),
128        "Parquet cache MISS - processing file"
129    );
130
131    // Parse content
132    let content = data;
133
134    // Process geometry and data model extraction + serialization ALL in parallel
135    // rayon::join works correctly here because rayon has its own thread pool
136    // that's independent of tokio's blocking thread pool
137    let serialize_start = tokio::time::Instant::now();
138    let opening_filter = query.opening_filter;
139    let data_model_entities = query.data_model_entities;
140    // Guard rides the blocking task (see parse_full): a cancelled handler
141    // future must not release the admission slot while the work runs on.
142    let (
143        (
144            ((geometry_result, mesh_count), combined_parquet),
145            (data_model_stats, data_model_parquet),
146            symbolic_data,
147        ),
148        _admission,
149    ) = tokio::task::spawn_blocking(move || {
150            // First: extract geometry and the data model in parallel.
151            // #5984: the finish index is a third independent read of the bytes.
152            let (mut geometry_result, (mut data_model, mut finishes)) = rayon::join(
153                || process_geometry_filtered_with_quality(&content, opening_filter, tessellation_quality),
154                || rayon::join(|| extract_data_model(&content), || ModelFinishes::from_content(&content)),
155            );
156            let meshes = finish_meshes(&mut finishes, std::mem::take(&mut geometry_result.meshes));
157            drop(finishes);
158
159            // Capture stats before moving data_model
160            let dm_stats = DataModelStats {
161                entity_count: data_model.entities.len(),
162                property_set_count: data_model.property_sets.len(),
163                relationship_count: data_model.relationships.len(),
164                spatial_node_count: data_model.spatial_hierarchy.nodes.len(),
165            };
166            // After the stats (#6034): they describe the MODEL, so the
167            // metadata header both entities variants share stays one value.
168            data_model_entities.apply(&mut data_model);
169
170            // Second: the 2D symbol stream (IfcAnnotation + IfcGrid, endpoint
171            // parity, issue #900) alongside serializing BOTH geometry and data
172            // model, so the data model is ready by the time the client needs it.
173            //
174            // Symbolic extraction used to run in the join ABOVE, beside the
175            // parse. It cannot: it needs the frame the parse selected, or a
176            // site-local model's symbols keep the site translation and rotation
177            // its meshes dropped (#4706). Moved down beside the serialization
178            // instead of made sequential, so it still overlaps other work.
179            let (symbolic_data, (geo_parquet, dm_parquet)) = rayon::join(
180                || extract_symbolic_data_with_provenance_in_frame(&content, geometry_result.frame),
181                || rayon::join(
182                    || {
183                        // The frame `geometry_result`'s vertices were baked in
184                        // (#4118). Without it a site-rotated model's repeated
185                        // shapes fail the residual check and silently keep
186                        // their per-occurrence geometry.
187                        let basis = baked_basis_zup(
188                            Some(geometry_result.mesh_coordinate_space),
189                            geometry_result.site_transform.as_deref(),
190                            geometry_result.metadata.coordinate_info.origin_shift,
191                        );
192                        serialize_combined_for_layout(
193                            &meshes,
194                            layout,
195                            Some(&basis),
196                        )
197                    },
198                    || serialize_data_model_to_parquet(&data_model),
199                ),
200            );
201
202            (
203                (
204                    ((geometry_result, meshes.len()), geo_parquet),
205                    (dm_stats, dm_parquet),
206                    symbolic_data,
207                ),
208                admission_guard,
209            )
210        })
211        .await?;
212
213    // Unwrap serialization results
214    let combined_parquet = combined_parquet?;
215    let data_model_parquet = data_model_parquet?;
216
217    let serialize_time = serialize_start.elapsed();
218    tracing::info!(
219        meshes = mesh_count,
220        geometry_parquet_size = combined_parquet.len(),
221        data_model_parquet_size = data_model_parquet.len(),
222        total_serialize_time_ms = serialize_time.as_millis(),
223        "Geometry and data model serialization complete (parallel)"
224    );
225
226    // Cache data model IMMEDIATELY (not in background) so it's ready when client polls
227    let data_model_cache_key = data_model_cache_key(&cache_key, data_model_entities);
228    if let Err(e) = state
229        .cache
230        .set_bytes(&data_model_cache_key, &data_model_parquet)
231        .await
232    {
233        tracing::error!(error = %e, "Failed to cache data model");
234    } else {
235        tracing::info!(
236            cache_key = %data_model_cache_key,
237            size = data_model_parquet.len(),
238            "Data model cached (ready for client)"
239        );
240    }
241
242    // Cache the symbolic stream immediately so it's ready when the client
243    // fetches `GET /api/v1/parse/symbolic/{cache_key}` (issue #900).
244    cache_symbolic_data_off_runtime(state.cache.clone(), cache_key.clone(), symbolic_data).await;
245
246    // Create metadata header with data model stats (captured before background task)
247    let cache_key_clone = cache_key.clone();
248    let metadata_header = ParquetMetadataHeader {
249        cache_key: cache_key_clone.clone(),
250        metadata: geometry_result.metadata,
251        stats: geometry_result.stats,
252        mesh_coordinate_space: Some(geometry_result.mesh_coordinate_space),
253        site_transform: geometry_result.site_transform,
254        building_transform: geometry_result.building_transform,
255        data_model_stats: Some(data_model_stats),
256    };
257
258    let metadata_json = serde_json::to_string(&metadata_header)?;
259
260    // Cache the results for future requests. `Bytes` makes the cache task's
261    // copy an O(1) refcount bump instead of duplicating the whole payload.
262    let parquet_cache_key = parquet_geometry_key(&cache_key_clone, layout);
263    let metadata_cache_key = parquet_metadata_key(&cache_key_clone);
264    let combined_parquet_clone = combined_parquet.clone();
265    let metadata_json_clone = metadata_json.clone();
266    let cache = state.cache.clone();
267
268    // Cache in background (don't block response). Deliberately NOT the
269    // optimized route's synchronous write (#3889): this payload is the large
270    // one, so blocking the response on it costs the client real time. The
271    // trade is a window where an immediate repeat request re-parses because
272    // the write has not landed yet.
273    tokio::spawn(async move {
274        if let Err(e) = cache
275            .set_bytes(&parquet_cache_key, &combined_parquet_clone)
276            .await
277        {
278            tracing::error!(error = %e, "Failed to cache Parquet bytes");
279        }
280        if let Err(e) = cache
281            .set_bytes(&metadata_cache_key, metadata_json_clone.as_bytes())
282            .await
283        {
284            tracing::error!(error = %e, "Failed to cache metadata");
285        }
286        tracing::info!(
287            cache_key = %cache_key_clone,
288            parquet_size = combined_parquet_clone.len(),
289            "Cached Parquet response"
290        );
291    });
292
293    // Build response with binary body and metadata header
294    let response = Response::builder()
295        .status(StatusCode::OK)
296        .header(header::CONTENT_TYPE, "application/x-parquet-geometry")
297        .header("X-IFC-Metadata", metadata_json)
298        .header(header::CONTENT_LENGTH, combined_parquet.len())
299        .body(Body::from(combined_parquet))
300        .map_err(|e| ApiError::Internal(e.to_string()))?;
301
302    Ok(response)
303}
ifc_lite_server/routes/parse/parquet.rs

ifc_lite_server/routes/parse/
parquet.rs
