1use std::{
17 fmt::Write as _,
18 fs::{self, File, OpenOptions},
19 io::BufWriter,
20 path::{Path, PathBuf},
21 sync::Arc,
22 time::Duration,
23};
24
25use arrow::{
26 array::{
27 ArrayRef, ListBuilder, RecordBatch, StringBuilder, TimestampMillisecondBuilder,
28 UInt32Builder,
29 },
30 datatypes::{DataType, Field, Schema},
31};
32use parquet::{
33 arrow::ArrowWriter,
34 basic::{Compression, ZstdLevel},
35 file::properties::WriterProperties,
36};
37use serde::{Deserialize, Serialize};
38use tracing::{info_span, instrument};
39use typed_builder::TypedBuilder;
40
41use super::{
42 ExportError, PARSER_VERSION, SCHEMA_MAJOR, SCHEMA_MINOR,
43 json::render_attribute_map,
44 manifest::{Manifest, ManifestFile, write_manifest},
45 schema::resources_schema,
46};
47use crate::ir::{
48 AttributeMap, Component, Expression, Local, ModuleCall, Output, ProviderBlock, ProviderRef,
49 Resource, ResourceKind, Span, Value, Variable, Workspace,
50};
51
52#[derive(Clone, Debug, PartialEq, Eq, TypedBuilder)]
54#[non_exhaustive]
55#[builder(field_defaults(setter(into)))]
56pub struct ExportOptions {
57 pub out_dir: Arc<Path>,
60
61 #[builder(default = 131_072)]
63 pub row_group_rows: usize,
64
65 #[builder(default = 64 * 1024 * 1024)]
67 pub row_group_bytes: usize,
68
69 #[builder(default = CompressionOpt::Zstd(3))]
71 pub compression: CompressionOpt,
72
73 #[builder(default = false)]
75 pub overwrite: bool,
76
77 #[builder(default)]
82 pub parsed_at_ms: Option<i64>,
83
84 #[builder(default = Arc::from(""))]
87 pub command_line: Arc<str>,
88
89 #[builder(default)]
95 pub tables: Vec<SecondaryTable>,
96}
97
98#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
102#[serde(rename_all = "kebab-case")]
103#[non_exhaustive]
104pub enum CompressionOpt {
105 Uncompressed,
107 Zstd(i32),
109 Snappy,
111}
112
113#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, Serialize, Deserialize)]
116#[serde(rename_all = "kebab-case")]
117#[non_exhaustive]
118pub enum SecondaryTable {
119 Dependencies,
121 Components,
123 Modules,
125}
126
127impl SecondaryTable {
128 #[must_use]
130 pub const fn file_stem(self) -> &'static str {
131 match self {
132 Self::Dependencies => "dependencies",
133 Self::Components => "components",
134 Self::Modules => "modules",
135 }
136 }
137}
138
139impl CompressionOpt {
140 pub fn zstd(level: i32) -> Result<Self, crate::ValidationError> {
152 if !(1..=22).contains(&level) {
153 return Err(crate::ValidationError::Range {
154 field: "CompressionOpt::Zstd.level",
155 min: 1,
156 max: 22,
157 got: i64::from(level),
158 });
159 }
160 Ok(Self::Zstd(level))
161 }
162
163 fn to_parquet(self) -> Compression {
164 match self {
165 Self::Uncompressed => Compression::UNCOMPRESSED,
166 Self::Zstd(level) => Compression::ZSTD(ZstdLevel::try_new(level).unwrap_or_default()),
171 Self::Snappy => Compression::SNAPPY,
172 }
173 }
174}
175
176#[derive(Clone, Debug, PartialEq, Eq)]
178#[non_exhaustive]
179pub struct ExportedFile {
180 pub path: Arc<Path>,
182 pub rows: u64,
184 pub bytes: u64,
186 pub sha256: String,
188}
189
190#[derive(Clone, Debug, PartialEq, Eq)]
192#[non_exhaustive]
193pub struct ExportReport {
194 pub files: Vec<ExportedFile>,
196 pub total_rows: u64,
198 pub bytes_written: u64,
200 pub elapsed: Duration,
202}
203
204pub trait Exporter: Send + Sync {
207 fn export(&self, ws: &Workspace, opts: &ExportOptions) -> Result<ExportReport, ExportError>;
215}
216
217#[derive(Clone, Copy, Debug, Default)]
219pub struct ParquetExporter;
220
221impl ParquetExporter {
222 #[must_use]
224 pub const fn new() -> Self {
225 Self
226 }
227}
228
229impl Exporter for ParquetExporter {
230 #[instrument(level = "info", skip_all, fields(out = %opts.out_dir.display()))]
231 #[allow(clippy::too_many_lines)] fn export(&self, ws: &Workspace, opts: &ExportOptions) -> Result<ExportReport, ExportError> {
233 let started = std::time::Instant::now();
234
235 validate_out_dir(&opts.out_dir)?;
236
237 let final_path: Arc<Path> = Arc::from(opts.out_dir.join("resources.parquet"));
238 let manifest_path: Arc<Path> = Arc::from(opts.out_dir.join("workspace.manifest.json"));
239
240 let mut secondary_paths: Vec<(SecondaryTable, Arc<Path>)> =
243 Vec::with_capacity(opts.tables.len());
244 for table in &opts.tables {
245 let path: Arc<Path> =
246 Arc::from(opts.out_dir.join(format!("{}.parquet", table.file_stem())));
247 secondary_paths.push((*table, path));
248 }
249 if !opts.overwrite {
250 for p in [&final_path, &manifest_path] {
251 if p.exists() {
252 return Err(ExportError::OutputExists(Arc::clone(p)));
253 }
254 }
255 for (_, p) in &secondary_paths {
256 if p.exists() {
257 return Err(ExportError::OutputExists(Arc::clone(p)));
258 }
259 }
260 }
261
262 let parsed_at_ms = opts
263 .parsed_at_ms
264 .unwrap_or_else(|| jiff::Timestamp::now().as_millisecond());
265
266 let projected_rows = projected_row_count(ws);
267
268 let (rows, bytes) = {
269 let span = info_span!("write_resources", path = %final_path.display());
270 let _entered = span.enter();
271 write_resources_parquet(ws, opts, &final_path, projected_rows, parsed_at_ms)?
272 };
273
274 let mut secondary_outputs: Vec<(SecondaryTable, Arc<Path>, u64, u64, String)> =
276 Vec::with_capacity(secondary_paths.len());
277 for (table, path) in &secondary_paths {
278 let (table_rows, table_bytes) = match table {
279 SecondaryTable::Dependencies => {
280 super::secondary::write_dependencies_parquet(ws, path, opts.compression)?
281 }
282 SecondaryTable::Components => {
283 super::secondary::write_components_parquet(ws, path, opts.compression)?
284 }
285 SecondaryTable::Modules => {
286 super::secondary::write_modules_parquet(ws, path, opts.compression)?
287 }
288 };
289 let sha = sha256_hex_of_file(path)?;
290 secondary_outputs.push((*table, Arc::clone(path), table_rows, table_bytes, sha));
291 }
292
293 let resources_sha = sha256_hex_of_file(&final_path)?;
295 let mut manifest_files: Vec<ManifestFile> = Vec::with_capacity(1 + secondary_outputs.len());
296 manifest_files.push(ManifestFile {
297 name: "resources.parquet".to_string(),
298 rows,
299 bytes,
300 sha256: resources_sha.clone(),
301 });
302 for (table, _, r, b, sha) in &secondary_outputs {
303 manifest_files.push(ManifestFile {
304 name: format!("{}.parquet", table.file_stem()),
305 rows: *r,
306 bytes: *b,
307 sha256: sha.clone(),
308 });
309 }
310 let manifest = Manifest {
311 tfparser_version: PARSER_VERSION.to_string(),
312 schema_major: SCHEMA_MAJOR,
313 schema_minor: SCHEMA_MINOR,
314 generated_at_ms: parsed_at_ms,
315 workspace_root: ws.root.display().to_string(),
316 command_line: opts.command_line.to_string(),
317 files: manifest_files,
318 };
319 let manifest_bytes_written = write_manifest(&manifest, &manifest_path, opts.overwrite)?;
320 let manifest_sha = sha256_hex_of_file(&manifest_path)?;
321
322 let mut files: Vec<ExportedFile> = Vec::with_capacity(2 + secondary_outputs.len());
323 files.push(ExportedFile {
324 path: Arc::clone(&final_path),
325 rows,
326 bytes,
327 sha256: resources_sha,
328 });
329 for (_, path, table_rows, table_bytes, sha) in secondary_outputs.iter().cloned() {
330 files.push(ExportedFile {
331 path,
332 rows: table_rows,
333 bytes: table_bytes,
334 sha256: sha,
335 });
336 }
337 files.push(ExportedFile {
338 path: Arc::clone(&manifest_path),
339 rows: 0,
340 bytes: manifest_bytes_written,
341 sha256: manifest_sha,
342 });
343
344 let total_rows = rows
345 + secondary_outputs
346 .iter()
347 .map(|(_, _, r, _, _)| *r)
348 .sum::<u64>();
349 let bytes_written = bytes
350 + secondary_outputs
351 .iter()
352 .map(|(_, _, _, b, _)| *b)
353 .sum::<u64>()
354 + manifest_bytes_written;
355 Ok(ExportReport {
356 files,
357 total_rows,
358 bytes_written,
359 elapsed: started.elapsed(),
360 })
361 }
362}
363
364fn validate_out_dir(out: &Path) -> Result<(), ExportError> {
365 if !out.exists() {
366 return Err(ExportError::OutDirMissing(Arc::from(out)));
367 }
368 if !out.is_dir() {
369 return Err(ExportError::OutDirNotDir(Arc::from(out)));
370 }
371 Ok(())
372}
373
374struct RowBuilders {
379 workspace_root: StringBuilder,
380 component_path: StringBuilder,
381 module_path: StringBuilder,
382 address: StringBuilder,
383 kind: StringBuilder,
384 resource_type: StringBuilder,
385 resource_name: StringBuilder,
386 provider_local: StringBuilder,
387 provider_source: StringBuilder,
388 account_id: StringBuilder,
389 account_name: StringBuilder,
390 region: StringBuilder,
391 environment: StringBuilder,
392 count_expr: StringBuilder,
393 for_each_expr: StringBuilder,
394 depends_on: ListBuilder<StringBuilder>,
395 attributes_json: StringBuilder,
396 state_account_id: StringBuilder,
397 state_region: StringBuilder,
398 file: StringBuilder,
399 line: UInt32Builder,
400 column: UInt32Builder,
401 parser_version: StringBuilder,
402 parsed_at: TimestampMillisecondBuilder,
403 schema: Arc<Schema>,
404 row_count: usize,
405 approx_bytes: usize,
406}
407
408impl RowBuilders {
409 fn with_capacity(rows: usize, schema: Arc<Schema>) -> Self {
410 Self {
411 workspace_root: StringBuilder::with_capacity(rows, rows * 64),
412 component_path: StringBuilder::with_capacity(rows, rows * 32),
413 module_path: StringBuilder::with_capacity(rows, rows * 16),
414 address: StringBuilder::with_capacity(rows, rows * 48),
415 kind: StringBuilder::with_capacity(rows, rows * 8),
416 resource_type: StringBuilder::with_capacity(rows, rows * 24),
417 resource_name: StringBuilder::with_capacity(rows, rows * 24),
418 provider_local: StringBuilder::with_capacity(rows, rows * 12),
419 provider_source: StringBuilder::with_capacity(rows, rows * 32),
420 account_id: StringBuilder::with_capacity(rows, rows * 12),
421 account_name: StringBuilder::with_capacity(rows, rows * 16),
422 region: StringBuilder::with_capacity(rows, rows * 12),
423 environment: StringBuilder::with_capacity(rows, rows * 12),
424 count_expr: StringBuilder::with_capacity(rows, rows * 16),
425 for_each_expr: StringBuilder::with_capacity(rows, rows * 16),
426 depends_on: ListBuilder::with_capacity(StringBuilder::new(), rows)
427 .with_field(Arc::new(Field::new("item", DataType::Utf8, false))),
428 attributes_json: StringBuilder::with_capacity(rows, rows * 256),
429 state_account_id: StringBuilder::with_capacity(rows, rows * 12),
430 state_region: StringBuilder::with_capacity(rows, rows * 12),
431 file: StringBuilder::with_capacity(rows, rows * 48),
432 line: UInt32Builder::with_capacity(rows),
433 column: UInt32Builder::with_capacity(rows),
434 parser_version: StringBuilder::with_capacity(rows, rows * 8),
435 parsed_at: TimestampMillisecondBuilder::with_capacity(rows)
436 .with_timezone(Arc::<str>::from("UTC")),
437 schema,
438 row_count: 0,
439 approx_bytes: 0,
440 }
441 }
442
443 fn append_row(&mut self, row: &Row<'_>, parsed_at_ms: i64) {
444 self.workspace_root.append_value(row.workspace_root);
445 self.component_path.append_value(row.component_path);
446 self.module_path.append_value(row.module_path);
447 self.address.append_value(row.address);
448 self.kind.append_value(row.kind);
449 self.resource_type.append_value(row.resource_type);
450 self.resource_name.append_value(row.resource_name);
451 self.provider_local.append_value(row.provider_local);
452 self.provider_source.append_value(row.provider_source);
453 self.account_id.append_value(row.account_id);
454 self.account_name.append_value(row.account_name);
455 self.region.append_value(row.region);
456 self.environment.append_value(row.environment);
457 self.count_expr.append_value(row.count_expr);
458 self.for_each_expr.append_value(row.for_each_expr);
459 let inner = self.depends_on.values();
460 for dep in row.depends_on {
461 inner.append_value(dep);
462 }
463 self.depends_on.append(true);
464 self.attributes_json.append_value(row.attributes_json);
465 self.state_account_id.append_value(row.state_account_id);
466 self.state_region.append_value(row.state_region);
467 self.file.append_value(row.file);
468 self.line.append_value(row.line);
469 self.column.append_value(row.column);
470 self.parser_version.append_value(PARSER_VERSION);
471 self.parsed_at.append_value(parsed_at_ms);
472 self.row_count += 1;
473 self.approx_bytes += approx_row_bytes(row);
474 }
475
476 fn batch(&mut self) -> Result<RecordBatch, arrow::error::ArrowError> {
477 let arrays: Vec<ArrayRef> = vec![
478 Arc::new(self.workspace_root.finish()),
479 Arc::new(self.component_path.finish()),
480 Arc::new(self.module_path.finish()),
481 Arc::new(self.address.finish()),
482 Arc::new(self.kind.finish()),
483 Arc::new(self.resource_type.finish()),
484 Arc::new(self.resource_name.finish()),
485 Arc::new(self.provider_local.finish()),
486 Arc::new(self.provider_source.finish()),
487 Arc::new(self.account_id.finish()),
488 Arc::new(self.account_name.finish()),
489 Arc::new(self.region.finish()),
490 Arc::new(self.environment.finish()),
491 Arc::new(self.count_expr.finish()),
492 Arc::new(self.for_each_expr.finish()),
493 Arc::new(self.depends_on.finish()),
494 Arc::new(self.attributes_json.finish()),
495 Arc::new(self.state_account_id.finish()),
496 Arc::new(self.state_region.finish()),
497 Arc::new(self.file.finish()),
498 Arc::new(self.line.finish()),
499 Arc::new(self.column.finish()),
500 Arc::new(self.parser_version.finish()),
501 Arc::new(self.parsed_at.finish()),
502 ];
503 let batch = RecordBatch::try_new(Arc::clone(&self.schema), arrays)?;
504 self.row_count = 0;
505 self.approx_bytes = 0;
506 Ok(batch)
507 }
508}
509
510struct Row<'a> {
512 workspace_root: &'a str,
513 component_path: &'a str,
514 module_path: &'a str,
515 address: &'a str,
516 kind: &'a str,
517 resource_type: &'a str,
518 resource_name: &'a str,
519 provider_local: &'a str,
520 provider_source: &'a str,
521 account_id: &'a str,
522 account_name: &'a str,
523 region: &'a str,
524 environment: &'a str,
525 count_expr: &'a str,
526 for_each_expr: &'a str,
527 depends_on: &'a [String],
528 attributes_json: &'a str,
529 state_account_id: &'a str,
530 state_region: &'a str,
531 file: &'a str,
532 line: u32,
533 column: u32,
534}
535
536fn approx_row_bytes(row: &Row<'_>) -> usize {
537 row.workspace_root.len()
538 + row.component_path.len()
539 + row.module_path.len()
540 + row.address.len()
541 + row.kind.len()
542 + row.resource_type.len()
543 + row.resource_name.len()
544 + row.provider_local.len()
545 + row.provider_source.len()
546 + row.account_id.len()
547 + row.account_name.len()
548 + row.region.len()
549 + row.environment.len()
550 + row.count_expr.len()
551 + row.for_each_expr.len()
552 + row.depends_on.iter().map(String::len).sum::<usize>()
553 + row.attributes_json.len()
554 + row.state_account_id.len()
555 + row.state_region.len()
556 + row.file.len()
557 + 8
558}
559
560const MAX_PREALLOC_ROWS: usize = 1_000_000;
566
567fn projected_row_count(ws: &Workspace) -> usize {
571 ws.components
572 .iter()
573 .map(|c| {
574 c.resources.len()
575 + c.providers.len()
576 + c.modules.len()
577 + c.outputs.len()
578 + c.variables.len()
579 + c.locals.len()
580 })
581 .sum::<usize>()
582 .min(MAX_PREALLOC_ROWS)
583}
584
585fn write_resources_parquet(
586 ws: &Workspace,
587 opts: &ExportOptions,
588 final_path: &Path,
589 projected_rows: usize,
590 parsed_at_ms: i64,
591) -> Result<(u64, u64), ExportError> {
592 let partial: PathBuf = partial_path(final_path);
593 if partial.exists() {
594 fs::remove_file(&partial).map_err(|source| ExportError::Io {
595 path: Arc::from(partial.as_path()),
596 source,
597 })?;
598 }
599
600 let schema = Arc::new(resources_schema());
601 let mut builders = RowBuilders::with_capacity(projected_rows.max(64), Arc::clone(&schema));
602 let workspace_root_str = ws.root.display().to_string();
603
604 let file = OpenOptions::new()
605 .write(true)
606 .create_new(true)
607 .open(&partial)
608 .map_err(|source| ExportError::Io {
609 path: Arc::from(partial.as_path()),
610 source,
611 })?;
612 let buf = BufWriter::with_capacity(256 * 1024, file);
613 let writer_props = WriterProperties::builder()
614 .set_compression(opts.compression.to_parquet())
615 .set_key_value_metadata(Some(vec![
616 parquet::file::metadata::KeyValue::new(
617 "tfparser.schema.major".to_string(),
618 Some(SCHEMA_MAJOR.to_string()),
619 ),
620 parquet::file::metadata::KeyValue::new(
621 "tfparser.schema.minor".to_string(),
622 Some(SCHEMA_MINOR.to_string()),
623 ),
624 parquet::file::metadata::KeyValue::new(
625 "tfparser.parser.version".to_string(),
626 Some(PARSER_VERSION.to_string()),
627 ),
628 ]))
629 .build();
630 let mut arrow_writer = ArrowWriter::try_new(buf, Arc::clone(&schema), Some(writer_props))
631 .map_err(|source| ExportError::Parquet {
632 path: Arc::from(partial.as_path()),
633 source,
634 })?;
635
636 let mut total_rows: u64 = 0;
637 let mut sorted_components: Vec<&Component> = ws.components.iter().collect();
638 sorted_components.sort_by(|a, b| a.path.as_os_str().cmp(b.path.as_os_str()));
639
640 for component in sorted_components {
641 let component_path_str = render_path(&component.path);
642 emit_component_rows(
643 component,
644 &workspace_root_str,
645 &component_path_str,
646 parsed_at_ms,
647 &mut builders,
648 &mut arrow_writer,
649 &Arc::from(partial.as_path()),
650 opts,
651 &mut total_rows,
652 )?;
653 }
654
655 if builders.row_count > 0 {
656 flush_batch(&mut builders, &mut arrow_writer, &partial)?;
657 }
658
659 let buf = arrow_writer
660 .into_inner()
661 .map_err(|source| ExportError::Parquet {
662 path: Arc::from(partial.as_path()),
663 source,
664 })?;
665 let file = buf.into_inner().map_err(|err| ExportError::Io {
666 path: Arc::from(partial.as_path()),
667 source: err.into_error(),
668 })?;
669 file.sync_all().map_err(|source| ExportError::Io {
670 path: Arc::from(partial.as_path()),
671 source,
672 })?;
673 drop(file);
674
675 fs::rename(&partial, final_path).map_err(|source| ExportError::Io {
676 path: Arc::from(partial.as_path()),
677 source,
678 })?;
679
680 let bytes = fs::metadata(final_path)
681 .map(|m| m.len())
682 .map_err(|source| ExportError::Io {
683 path: Arc::from(final_path),
684 source,
685 })?;
686
687 Ok((total_rows, bytes))
688}
689
690fn partial_path(final_path: &Path) -> PathBuf {
691 let mut s: std::ffi::OsString = final_path.as_os_str().to_os_string();
692 s.push(".partial");
693 PathBuf::from(s)
694}
695
696#[allow(clippy::too_many_arguments)]
697fn emit_component_rows(
698 component: &Component,
699 workspace_root_str: &str,
700 component_path_str: &str,
701 parsed_at_ms: i64,
702 builders: &mut RowBuilders,
703 arrow_writer: &mut ArrowWriter<BufWriter<File>>,
704 partial_path: &Arc<Path>,
705 opts: &ExportOptions,
706 total_rows: &mut u64,
707) -> Result<(), ExportError> {
708 let mut rows: Vec<EmittedRow> = Vec::new();
711
712 let state_account_id = component
713 .state_backend
714 .as_ref()
715 .and_then(|b| b.state_account_id.as_ref())
716 .map(|a| a.as_str().to_string())
717 .unwrap_or_default();
718 let state_region = component
719 .state_backend
720 .as_ref()
721 .and_then(|b| b.state_region.as_ref())
722 .map(|r| r.as_str().to_string())
723 .unwrap_or_default();
724
725 for r in &component.resources {
726 rows.push(resource_row(r, &state_account_id, &state_region));
727 }
728 for p in &component.providers {
729 rows.push(provider_row(p));
730 }
731 for m in &component.modules {
732 rows.push(module_call_row(m));
733 }
734 for v in &component.variables {
735 rows.push(variable_row(v));
736 }
737 for l in &component.locals {
738 rows.push(local_row(l));
739 }
740 for o in &component.outputs {
741 rows.push(output_row(o));
742 }
743
744 rows.sort_by(|a, b| {
745 (a.module_path.as_str(), a.address.as_str())
746 .cmp(&(b.module_path.as_str(), b.address.as_str()))
747 });
748
749 let mut json_scratch = String::with_capacity(4096);
750 for emitted in &rows {
751 json_scratch.clear();
752 render_attribute_map(&emitted.attributes, &mut json_scratch);
753 let row = Row {
754 workspace_root: workspace_root_str,
755 component_path: component_path_str,
756 module_path: emitted.module_path.as_str(),
757 address: emitted.address.as_str(),
758 kind: emitted.kind,
759 resource_type: emitted.resource_type.as_str(),
760 resource_name: emitted.resource_name.as_str(),
761 provider_local: emitted.provider_local.as_str(),
762 provider_source: emitted.provider_source.as_str(),
763 account_id: emitted.account_id.as_str(),
764 account_name: emitted.account_name.as_str(),
765 region: emitted.region.as_str(),
766 environment: emitted.environment.as_str(),
767 count_expr: emitted.count_expr.as_str(),
768 for_each_expr: emitted.for_each_expr.as_str(),
769 depends_on: emitted.depends_on.as_slice(),
770 attributes_json: json_scratch.as_str(),
771 state_account_id: emitted.state_account_id.as_str(),
772 state_region: emitted.state_region.as_str(),
773 file: emitted.file.as_str(),
774 line: emitted.line,
775 column: emitted.column,
776 };
777 builders.append_row(&row, parsed_at_ms);
778 *total_rows = total_rows.saturating_add(1);
779
780 if builders.row_count >= opts.row_group_rows
781 || builders.approx_bytes >= opts.row_group_bytes
782 {
783 flush_batch(builders, arrow_writer, partial_path)?;
784 }
785 }
786 Ok(())
787}
788
789fn flush_batch(
790 builders: &mut RowBuilders,
791 arrow_writer: &mut ArrowWriter<BufWriter<File>>,
792 partial_path: &Path,
793) -> Result<(), ExportError> {
794 let batch = builders.batch().map_err(|source| ExportError::Arrow {
795 path: Arc::from(partial_path),
796 source,
797 })?;
798 arrow_writer
799 .write(&batch)
800 .map_err(|source| ExportError::Parquet {
801 path: Arc::from(partial_path),
802 source,
803 })?;
804 Ok(())
805}
806
807struct EmittedRow {
809 module_path: String,
810 address: String,
811 kind: &'static str,
812 resource_type: String,
813 resource_name: String,
814 provider_local: String,
815 provider_source: String,
816 account_id: String,
817 account_name: String,
818 region: String,
819 environment: String,
820 count_expr: String,
821 for_each_expr: String,
822 depends_on: Vec<String>,
823 attributes: AttributeMap,
824 state_account_id: String,
825 state_region: String,
826 file: String,
827 line: u32,
828 column: u32,
829}
830
831fn resource_row(r: &Resource, state_account_id: &str, state_region: &str) -> EmittedRow {
832 let provider_local = r
833 .provider_ref
834 .as_ref()
835 .map(provider_ref_string)
836 .unwrap_or_default();
837 let account_id = r
838 .account_id
839 .as_ref()
840 .map(|a| a.as_str().to_string())
841 .unwrap_or_default();
842 let account_name = r
843 .account_name
844 .as_deref()
845 .map(str::to_string)
846 .unwrap_or_default();
847 let region = r
848 .region
849 .as_ref()
850 .map(|reg| reg.as_str().to_string())
851 .unwrap_or_default();
852 EmittedRow {
853 module_path: r.address.module_path(),
854 address: r.address.as_str().to_string(),
855 kind: match r.kind {
856 ResourceKind::Managed => "resource",
857 ResourceKind::Data => "data",
858 },
859 resource_type: r.type_.to_string(),
860 resource_name: r.name.to_string(),
861 provider_local,
862 provider_source: String::new(),
863 account_id,
864 account_name,
865 region,
866 environment: String::new(),
867 count_expr: r
868 .count_expr
869 .as_ref()
870 .map(render_expression_source)
871 .unwrap_or_default(),
872 for_each_expr: r
873 .for_each_expr
874 .as_ref()
875 .map(render_expression_source)
876 .unwrap_or_default(),
877 depends_on: r
878 .depends_on
879 .iter()
880 .map(|a| a.as_str().to_string())
881 .collect(),
882 attributes: r.attributes.clone(),
883 state_account_id: state_account_id.to_string(),
884 state_region: state_region.to_string(),
885 file: span_relative_file(&r.span),
886 line: r.span.line,
887 column: r.span.column,
888 }
889}
890
891fn provider_row(p: &ProviderBlock) -> EmittedRow {
892 let local = p.local_name.to_string();
893 let provider_local = match p.alias.as_deref() {
894 Some(a) if !a.is_empty() => format!("{local}.{a}"),
895 _ => local.clone(),
896 };
897 let address = format!("provider.{provider_local}");
898 EmittedRow {
899 module_path: String::new(),
900 address,
901 kind: "provider",
902 resource_type: String::new(),
903 resource_name: provider_local.clone(),
904 provider_local,
905 provider_source: p
906 .source_addr
907 .as_deref()
908 .map(str::to_string)
909 .unwrap_or_default(),
910 account_id: String::new(),
911 account_name: String::new(),
912 region: String::new(),
913 environment: String::new(),
914 count_expr: String::new(),
915 for_each_expr: String::new(),
916 depends_on: Vec::new(),
917 attributes: p.raw.clone(),
918 state_account_id: String::new(),
919 state_region: String::new(),
920 file: span_relative_file(&p.span),
921 line: p.span.line,
922 column: p.span.column,
923 }
924}
925
926fn module_call_row(m: &ModuleCall) -> EmittedRow {
927 let attrs: AttributeMap = m.inputs.clone();
928 let provider_local = m
929 .providers
930 .first()
931 .map(|(_, r)| provider_ref_string(r))
932 .unwrap_or_default();
933 EmittedRow {
934 module_path: m.address.module_path(),
935 address: m.address.as_str().to_string(),
936 kind: "module",
937 resource_type: String::new(),
938 resource_name: m
939 .address
940 .as_str()
941 .strip_prefix("module.")
942 .map_or_else(|| m.address.as_str().to_string(), str::to_string),
943 provider_local,
944 provider_source: m.source_raw.to_string(),
945 account_id: String::new(),
946 account_name: String::new(),
947 region: String::new(),
948 environment: String::new(),
949 count_expr: m
950 .count_expr
951 .as_ref()
952 .map(render_expression_source)
953 .unwrap_or_default(),
954 for_each_expr: m
955 .for_each_expr
956 .as_ref()
957 .map(render_expression_source)
958 .unwrap_or_default(),
959 depends_on: Vec::new(),
960 attributes: attrs,
961 state_account_id: String::new(),
962 state_region: String::new(),
963 file: span_relative_file(&m.span),
964 line: m.span.line,
965 column: m.span.column,
966 }
967}
968
969fn variable_row(v: &Variable) -> EmittedRow {
970 let mut attrs: AttributeMap = Vec::new();
971 if let Some(t) = &v.type_expr {
972 attrs.push((Arc::from("type"), t.clone()));
973 }
974 if let Some(d) = &v.default {
975 attrs.push((Arc::from("default"), d.clone()));
976 }
977 if let Some(d) = &v.description {
978 attrs.push((
979 Arc::from("description"),
980 Expression::Literal(Value::Str(Arc::clone(d))),
981 ));
982 }
983 attrs.push((
984 Arc::from("sensitive"),
985 Expression::Literal(Value::Bool(v.sensitive)),
986 ));
987 EmittedRow {
988 module_path: String::new(),
989 address: format!("var.{}", v.name),
990 kind: "variable",
991 resource_type: String::new(),
992 resource_name: v.name.to_string(),
993 provider_local: String::new(),
994 provider_source: String::new(),
995 account_id: String::new(),
996 account_name: String::new(),
997 region: String::new(),
998 environment: String::new(),
999 count_expr: String::new(),
1000 for_each_expr: String::new(),
1001 depends_on: Vec::new(),
1002 attributes: attrs,
1003 state_account_id: String::new(),
1004 state_region: String::new(),
1005 file: span_relative_file(&v.span),
1006 line: v.span.line,
1007 column: v.span.column,
1008 }
1009}
1010
1011fn local_row(l: &Local) -> EmittedRow {
1012 let attrs: AttributeMap = vec![(Arc::from("value"), l.value.clone())];
1013 EmittedRow {
1014 module_path: String::new(),
1015 address: format!("local.{}", l.name),
1016 kind: "local",
1017 resource_type: String::new(),
1018 resource_name: l.name.to_string(),
1019 provider_local: String::new(),
1020 provider_source: String::new(),
1021 account_id: String::new(),
1022 account_name: String::new(),
1023 region: String::new(),
1024 environment: String::new(),
1025 count_expr: String::new(),
1026 for_each_expr: String::new(),
1027 depends_on: Vec::new(),
1028 attributes: attrs,
1029 state_account_id: String::new(),
1030 state_region: String::new(),
1031 file: span_relative_file(&l.span),
1032 line: l.span.line,
1033 column: l.span.column,
1034 }
1035}
1036
1037fn output_row(o: &Output) -> EmittedRow {
1038 let mut attrs: AttributeMap = Vec::new();
1039 attrs.push((Arc::from("value"), o.value.clone()));
1040 if let Some(d) = &o.description {
1041 attrs.push((
1042 Arc::from("description"),
1043 Expression::Literal(Value::Str(Arc::clone(d))),
1044 ));
1045 }
1046 attrs.push((
1047 Arc::from("sensitive"),
1048 Expression::Literal(Value::Bool(o.sensitive)),
1049 ));
1050 EmittedRow {
1051 module_path: String::new(),
1052 address: format!("output.{}", o.name),
1053 kind: "output",
1054 resource_type: String::new(),
1055 resource_name: o.name.to_string(),
1056 provider_local: String::new(),
1057 provider_source: String::new(),
1058 account_id: String::new(),
1059 account_name: String::new(),
1060 region: String::new(),
1061 environment: String::new(),
1062 count_expr: String::new(),
1063 for_each_expr: String::new(),
1064 depends_on: Vec::new(),
1065 attributes: attrs,
1066 state_account_id: String::new(),
1067 state_region: String::new(),
1068 file: span_relative_file(&o.span),
1069 line: o.span.line,
1070 column: o.span.column,
1071 }
1072}
1073
1074fn provider_ref_string(r: &ProviderRef) -> String {
1075 match r.alias.as_deref() {
1076 Some(a) if !a.is_empty() => format!("{}.{a}", r.local_name),
1077 _ => r.local_name.to_string(),
1078 }
1079}
1080
1081fn render_path(p: &Path) -> String {
1087 let mut out = String::with_capacity(p.as_os_str().len());
1088 for (idx, comp) in p.components().enumerate() {
1089 if idx > 0 {
1090 out.push('/');
1091 }
1092 match comp {
1093 std::path::Component::Normal(s) => {
1094 out.push_str(&s.to_string_lossy());
1095 }
1096 std::path::Component::ParentDir => out.push_str(".."),
1097 std::path::Component::CurDir => out.push('.'),
1098 std::path::Component::Prefix(_) | std::path::Component::RootDir => {
1099 out.push_str(&comp.as_os_str().to_string_lossy());
1102 }
1103 }
1104 }
1105 out
1106}
1107
1108fn span_relative_file(span: &Span) -> String {
1109 render_path(&span.file)
1110}
1111
1112fn render_expression_source(expr: &Expression) -> String {
1116 match expr {
1117 Expression::Literal(Value::Int(n)) => n.to_string(),
1118 Expression::Literal(Value::Bool(b)) => b.to_string(),
1119 Expression::Literal(Value::Str(s)) => s.to_string(),
1120 Expression::Literal(Value::Number(f)) if f.is_finite() => {
1121 let mut buf = ryu::Buffer::new();
1122 buf.format(*f).to_string()
1123 }
1124 Expression::Unresolved(s) => s.source.to_string(),
1125 _ => {
1126 let mut s = String::new();
1127 let map: AttributeMap = vec![(Arc::from(""), expr.clone())];
1128 render_attribute_map(&map, &mut s);
1129 s
1130 }
1131 }
1132}
1133
1134fn sha256_hex_of_file(path: &Path) -> Result<String, ExportError> {
1136 use sha2::{Digest, Sha256};
1137 let bytes = fs::read(path).map_err(|source| ExportError::Io {
1138 path: Arc::from(path),
1139 source,
1140 })?;
1141 let mut h = Sha256::new();
1142 h.update(&bytes);
1143 let digest = h.finalize();
1144 let mut out = String::with_capacity(64);
1145 for b in digest {
1146 let _ = write!(out, "{b:02x}");
1147 }
1148 Ok(out)
1149}
1150
1151#[cfg(test)]
1152#[allow(
1153 clippy::unwrap_used,
1154 clippy::expect_used,
1155 clippy::panic,
1156 clippy::indexing_slicing
1157)]
1158mod tests {
1159 use std::path::PathBuf;
1160
1161 use super::*;
1162 use crate::ir::{
1163 Address, ComponentId, ComponentKind, ResourceKind, Span, SymbolKind, Symbolic,
1164 };
1165
1166 fn arc_path<P: AsRef<Path>>(p: P) -> Arc<Path> {
1167 Arc::from(p.as_ref())
1168 }
1169
1170 fn minimal_resource() -> Resource {
1171 Resource::builder()
1172 .address(Address::new("aws_iam_role.r").unwrap())
1173 .kind(ResourceKind::Managed)
1174 .type_(Arc::<str>::from("aws_iam_role"))
1175 .name(Arc::<str>::from("r"))
1176 .span(Span::synthetic())
1177 .build()
1178 }
1179
1180 fn minimal_component() -> Component {
1181 Component::builder()
1182 .id(ComponentId::from_index(0))
1183 .path(arc_path(PathBuf::from("svc")))
1184 .kind(ComponentKind::Component)
1185 .resources(vec![minimal_resource()])
1186 .build()
1187 }
1188
1189 #[test]
1190 fn test_should_write_resources_parquet_with_one_row() {
1191 let tmp = tempfile::tempdir().unwrap();
1192 let ws = Workspace::builder()
1193 .root(arc_path(tmp.path()))
1194 .components(vec![minimal_component()])
1195 .build();
1196 let opts = ExportOptions::builder()
1197 .out_dir(arc_path(tmp.path()))
1198 .parsed_at_ms(Some(1_700_000_000_000_i64))
1199 .build();
1200 let report = ParquetExporter::new().export(&ws, &opts).unwrap();
1201 assert_eq!(report.total_rows, 1);
1202 assert!(
1203 report
1204 .files
1205 .iter()
1206 .any(|f| f.path.file_name().and_then(|n| n.to_str()) == Some("resources.parquet"))
1207 );
1208 }
1209
1210 #[test]
1211 fn test_should_refuse_overwrite_without_flag() {
1212 let tmp = tempfile::tempdir().unwrap();
1213 let final_path = tmp.path().join("resources.parquet");
1214 fs::write(&final_path, b"sentinel").unwrap();
1215 let ws = Workspace::builder()
1216 .root(arc_path(tmp.path()))
1217 .components(vec![minimal_component()])
1218 .build();
1219 let opts = ExportOptions::builder()
1220 .out_dir(arc_path(tmp.path()))
1221 .parsed_at_ms(Some(1))
1222 .build();
1223 let err = ParquetExporter::new().export(&ws, &opts).unwrap_err();
1224 assert!(matches!(err, ExportError::OutputExists(_)));
1225 }
1226
1227 #[test]
1228 fn test_should_overwrite_when_flag_set() {
1229 let tmp = tempfile::tempdir().unwrap();
1230 let final_path = tmp.path().join("resources.parquet");
1231 fs::write(&final_path, b"sentinel").unwrap();
1232 let ws = Workspace::builder()
1233 .root(arc_path(tmp.path()))
1234 .components(vec![minimal_component()])
1235 .build();
1236 let opts = ExportOptions::builder()
1237 .out_dir(arc_path(tmp.path()))
1238 .parsed_at_ms(Some(1))
1239 .overwrite(true)
1240 .build();
1241 let report = ParquetExporter::new().export(&ws, &opts).unwrap();
1242 assert_eq!(report.total_rows, 1);
1243 let bytes = fs::read(&final_path).unwrap();
1244 assert_ne!(bytes, b"sentinel".to_vec());
1245 }
1246
1247 #[test]
1248 fn test_should_be_byte_deterministic_with_pinned_parsed_at() {
1249 let tmp_a = tempfile::tempdir().unwrap();
1250 let tmp_b = tempfile::tempdir().unwrap();
1251 let ws_a = Workspace::builder()
1252 .root(arc_path(tmp_a.path()))
1253 .components(vec![minimal_component()])
1254 .build();
1255 let ws_b = Workspace::builder()
1256 .root(arc_path(tmp_b.path()))
1257 .components(vec![minimal_component()])
1258 .build();
1259 let opts_a = ExportOptions::builder()
1260 .out_dir(arc_path(tmp_a.path()))
1261 .parsed_at_ms(Some(1_700_000_000_000_i64))
1262 .build();
1263 let opts_b = ExportOptions::builder()
1264 .out_dir(arc_path(tmp_b.path()))
1265 .parsed_at_ms(Some(1_700_000_000_000_i64))
1266 .build();
1267 let r_a = ParquetExporter::new().export(&ws_a, &opts_a).unwrap();
1268 let r_b = ParquetExporter::new().export(&ws_b, &opts_b).unwrap();
1269 let parquet_a = r_a
1270 .files
1271 .iter()
1272 .find(|f| f.path.file_name().and_then(|n| n.to_str()) == Some("resources.parquet"))
1273 .unwrap();
1274 let parquet_b = r_b
1275 .files
1276 .iter()
1277 .find(|f| f.path.file_name().and_then(|n| n.to_str()) == Some("resources.parquet"))
1278 .unwrap();
1279 assert!(parquet_a.bytes > 0);
1283 assert!(parquet_b.bytes > 0);
1284 }
1285
1286 #[test]
1287 fn test_partial_path_appends_suffix() {
1288 let p = partial_path(Path::new("/tmp/resources.parquet"));
1289 assert_eq!(p, PathBuf::from("/tmp/resources.parquet.partial"));
1290 }
1291
1292 #[test]
1293 fn test_render_path_normalises_separators() {
1294 use std::path::PathBuf;
1295 assert_eq!(render_path(&PathBuf::from("a/b/c.tf")), "a/b/c.tf");
1297 assert_eq!(render_path(&PathBuf::from("main.tf")), "main.tf");
1299 assert_eq!(render_path(&PathBuf::from("")), "");
1301 assert_eq!(
1303 render_path(&PathBuf::from("../foo/main.tf")),
1304 "../foo/main.tf"
1305 );
1306 }
1307
1308 #[test]
1309 fn test_render_expression_source_int_and_unresolved() {
1310 assert_eq!(
1311 render_expression_source(&Expression::Literal(Value::Int(3))),
1312 "3"
1313 );
1314 let expr = Expression::Unresolved(
1315 Symbolic::builder()
1316 .kind(SymbolKind::Var)
1317 .source(Arc::<str>::from("var.x"))
1318 .span(Span::synthetic())
1319 .build(),
1320 );
1321 assert_eq!(render_expression_source(&expr), "var.x");
1322 }
1323
1324 #[test]
1325 fn test_should_refuse_when_out_dir_missing() {
1326 let ws = Workspace::builder()
1327 .root(arc_path(PathBuf::from("/tmp/x")))
1328 .build();
1329 let opts = ExportOptions::builder()
1330 .out_dir(arc_path(PathBuf::from("/this/does/not/exist/zzz")))
1331 .parsed_at_ms(Some(0))
1332 .build();
1333 let err = ParquetExporter::new().export(&ws, &opts).unwrap_err();
1334 assert!(matches!(err, ExportError::OutDirMissing(_)));
1335 }
1336
1337 #[test]
1338 fn test_should_refuse_when_out_dir_is_file() {
1339 let tmp = tempfile::tempdir().unwrap();
1340 let f = tmp.path().join("not-a-dir");
1341 fs::write(&f, b"x").unwrap();
1342 let ws = Workspace::builder().root(arc_path(tmp.path())).build();
1343 let opts = ExportOptions::builder()
1344 .out_dir(arc_path(f))
1345 .parsed_at_ms(Some(0))
1346 .build();
1347 let err = ParquetExporter::new().export(&ws, &opts).unwrap_err();
1348 assert!(matches!(err, ExportError::OutDirNotDir(_)));
1349 }
1350}