Line data Source code
1 : //! Batch processing: scan, parse and analyze many scripts against shared
2 : //! descriptor context in one call.
3 :
4 : use dashql_core::catalog::{Catalog, CatalogLayer};
5 : use dashql_core::error::{Error, Result};
6 : use dashql_core::pack::{pack_analyzed_script, pack_catalog_descriptor, pack_parsed_script, pack_scanned_script};
7 : use crate::visualize::pack_visualization;
8 : use dashql_core::script::Script;
9 : use dashql_buffers::analyzer::AnalyzerErrorSeverity;
10 : use dashql_buffers::batch::{BatchDiagnosticT, BatchRequest, BatchResultT, BatchScriptResultT};
11 : pub use dashql_buffers::batch::{DiagnosticSeverity, ProcessingStage};
12 : use dashql_parser::parser::context::resolve_text_span;
13 : use dashql_parser::span::TextSpan;
14 :
15 : /// Pack an object API value into FlatBuffer bytes
16 : macro_rules! pack_bytes {
17 : ($value:expr) => {{
18 : let mut builder = flatbuffers::FlatBufferBuilder::new();
19 : let packed = $value.pack(&mut builder);
20 : builder.finish(packed, None);
21 : builder.finished_data().to_vec()
22 : }};
23 : }
24 :
25 : /// A vector that is left out of the packed buffer when empty
26 74 : fn non_empty<T>(v: Vec<T>) -> Option<Vec<T>> {
27 74 : (!v.is_empty()).then_some(v)
28 74 : }
29 :
30 : /// A diagnostic of a checked script
31 : #[derive(Clone, Debug, PartialEq, Eq)]
32 : pub struct Diagnostic {
33 : /// The processing stage that reported the diagnostic
34 : pub stage: ProcessingStage,
35 : /// Whether the diagnostic is an error or a warning
36 : pub severity: DiagnosticSeverity,
37 : /// The UTF-8 byte span in the script text
38 : pub span: TextSpan,
39 : /// The diagnostic message
40 : pub message: String,
41 : }
42 :
43 : /// A script processed by [`check_script`]
44 : pub struct CheckedScript {
45 : /// The processed script, not published to the catalog
46 : pub script: Script,
47 : /// The diagnostics of all stages that ran, in stage order
48 : pub diagnostics: Vec<Diagnostic>,
49 : /// Why the analysis failed, the diagnostics of the earlier stages are kept
50 : pub failure: Option<String>,
51 : }
52 :
53 : impl CheckedScript {
54 : /// Whether processing failed or found an error
55 17 : pub fn has_errors(&self) -> bool {
56 17 : self.failure.is_some() || self.diagnostics.iter().any(|d| d.severity == DiagnosticSeverity::ERROR)
57 17 : }
58 : }
59 :
60 : /// Scan, parse and analyze a script against a catalog up to `until`, collecting the diagnostics of every
61 : /// stage. The script is not published to the catalog.
62 52 : pub fn check_script(catalog: &Catalog, text: &str, until: ProcessingStage) -> CheckedScript {
63 52 : let mut diagnostics = Vec::new();
64 52 : let mut diagnostic = |stage, severity, span, message: &str| {
65 22 : diagnostics.push(Diagnostic { stage, severity, span, message: message.to_string() });
66 22 : };
67 52 : let mut script = Script::new(catalog);
68 52 : script.replace_text(text);
69 52 : script.scan();
70 52 : for e in &script.scanned.as_ref().unwrap().errors {
71 3 : diagnostic(ProcessingStage::SCAN, DiagnosticSeverity::ERROR, e.location, &e.message);
72 3 : }
73 52 : if until >= ProcessingStage::PARSE {
74 46 : script.parse();
75 46 : let parsed = script.parsed.as_ref().unwrap();
76 46 : for e in &parsed.errors {
77 10 : let span = resolve_text_span(&parsed.scanned.text, &parsed.scanned.symbols, e.location);
78 10 : diagnostic(ProcessingStage::PARSE, DiagnosticSeverity::ERROR, span, &e.message);
79 10 : }
80 6 : }
81 52 : let mut failure = None;
82 52 : if until >= ProcessingStage::ANALYZE {
83 43 : match script.analyze(catalog, false) {
84 : Ok(()) => {
85 43 : let analyzed = script.analyzed.as_ref().unwrap();
86 43 : let scanned = &analyzed.parsed.scanned;
87 43 : for e in &analyzed.errors {
88 9 : let severity = if e.severity == AnalyzerErrorSeverity::WARNING {
89 3 : DiagnosticSeverity::WARNING
90 : } else {
91 6 : DiagnosticSeverity::ERROR
92 : };
93 9 : let span = match (e.text_span, e.symbol_span) {
94 9 : (Some(t), _) => t,
95 0 : (None, Some(s)) => resolve_text_span(&scanned.text, &scanned.symbols, s),
96 0 : (None, None) => TextSpan::default(),
97 : };
98 9 : diagnostic(ProcessingStage::ANALYZE, severity, span, &e.message);
99 : }
100 : }
101 0 : Err(e) => failure = Some(e.message),
102 : }
103 9 : }
104 52 : CheckedScript { script, diagnostics, failure }
105 52 : }
106 :
107 : /// Process a batch request. Inputs share only the descriptor context, they never
108 : /// publish declarations to each other.
109 29 : pub fn process_batch(request: &[u8]) -> Result<Vec<u8>> {
110 29 : let request = if request.is_empty() { None } else { flatbuffers::root::<BatchRequest>(request).ok() };
111 29 : let request = request.ok_or_else(|| Error::invalid_argument("invalid BatchRequest FlatBuffer"))?;
112 26 : let mut catalog = Catalog::new();
113 26 : if let Some(descriptors) = request.catalog_descriptors() {
114 : // Earlier descriptors get older entry ids and win on name conflicts
115 26 : for descriptor in descriptors.iter() {
116 7 : let data = descriptor.data().ok_or_else(|| Error::invalid_argument("missing batch catalog descriptor bytes"))?;
117 7 : let id = catalog.allocate_entry_id();
118 7 : catalog.replace_descriptor(id, CatalogLayer::CONNECTION, data.bytes())?;
119 : }
120 0 : }
121 :
122 24 : let mut result = BatchResultT::default();
123 24 : let mut scripts = Vec::new();
124 26 : for input in request.scripts().into_iter().flatten() {
125 26 : let mut out = BatchScriptResultT::default();
126 26 : out.id = input.id().filter(|id| !id.is_empty()).map(str::to_string);
127 26 : let outputs = input.outputs();
128 26 : if outputs & !15 != 0 {
129 1 : out.failure = Some("invalid batch output mask".into());
130 25 : } else if outputs != 0 {
131 24 : let until = if outputs & 12 != 0 {
132 17 : ProcessingStage::ANALYZE
133 7 : } else if outputs & 2 != 0 {
134 2 : ProcessingStage::PARSE
135 : } else {
136 5 : ProcessingStage::SCAN
137 : };
138 : // The FlatBuffers verifier already rejects the whole request if a text is not valid UTF-8
139 24 : let checked = check_script(&catalog, input.text().unwrap_or(""), until);
140 24 : let script = &checked.script;
141 24 : if outputs & 1 != 0 {
142 5 : out.scanned = script.scanned.as_deref().map(|s| pack_bytes!(pack_scanned_script(s)));
143 19 : }
144 24 : if outputs & 2 != 0 {
145 3 : out.parsed = script.parsed.as_deref().map(|p| pack_bytes!(pack_parsed_script(p)));
146 21 : }
147 24 : out.failure = checked.failure.clone();
148 24 : if let (None, Some(analyzed)) = (&checked.failure, script.analyzed.as_deref()) {
149 17 : if outputs & 4 != 0 {
150 13 : out.analyzed = Some(pack_bytes!(pack_analyzed_script(analyzed, &pack_visualization)));
151 13 : }
152 17 : if outputs & 8 != 0 && !checked.has_errors() {
153 5 : match pack_catalog_descriptor(&analyzed.entry) {
154 5 : Ok(descriptor) => out.catalog_descriptor = Some(descriptor),
155 0 : Err(e) => out.failure = Some(e.message),
156 : }
157 12 : }
158 7 : }
159 24 : let diagnostics = checked.diagnostics.into_iter().map(|d| {
160 8 : let mut out = BatchDiagnosticT::default();
161 8 : out.stage = d.stage;
162 8 : out.severity = d.severity;
163 8 : out.offset = d.span.offset;
164 8 : out.length = d.span.length;
165 8 : out.message = (!d.message.is_empty()).then_some(d.message);
166 8 : out
167 8 : });
168 24 : out.diagnostics = non_empty(diagnostics.collect());
169 1 : }
170 : // The object API leaves out empty vectors
171 26 : out.scanned = out.scanned.and_then(non_empty);
172 26 : out.parsed = out.parsed.and_then(non_empty);
173 26 : out.analyzed = out.analyzed.and_then(non_empty);
174 26 : out.catalog_descriptor = out.catalog_descriptor.and_then(non_empty);
175 26 : scripts.push(out);
176 : }
177 24 : result.scripts = non_empty(scripts);
178 24 : let mut builder = flatbuffers::FlatBufferBuilder::new();
179 24 : let packed = result.pack(&mut builder);
180 24 : builder.finish(packed, None);
181 24 : Ok(builder.finished_data().to_vec())
182 29 : }
|
