From cb96ebaea52e51afe47f8947e4b8634fbb6a5589 Mon Sep 17 00:00:00 2001 From: sundaresanr Date: Thu, 24 Sep 2026 20:00:56 -0700 Subject: [PATCH 1/2] feat(vrl): add parse_parquet function Add parse_parquet, which returns the file's rows as an array of objects using the parquet crate already in the tree (snappy, gzip and zstd column compression). Timestamps and dates become timestamps, decimals and u64 values past i64::MAX become strings to stay lossless, and NaN becomes null. The parquet 39 row reader panics on corrupt page data instead of returning an error, so reads are wrapped in catch_unwind and a malformed file becomes a fallible-call error that remap can route to its dropped output rather than crashing the process. --- Cargo.lock | 2 + lib/vector-vrl/functions/Cargo.toml | 2 + lib/vector-vrl/functions/src/lib.rs | 7 +- lib/vector-vrl/functions/src/parse_parquet.rs | 259 ++++++++++++++++++ .../functions/tests/data/types_zstd.parquet | Bin 0 -> 5147 bytes .../tests/data/vpc_flow_gzip.parquet | Bin 0 -> 7519 bytes 6 files changed, 269 insertions(+), 1 deletion(-) create mode 100644 lib/vector-vrl/functions/src/parse_parquet.rs create mode 100644 lib/vector-vrl/functions/tests/data/types_zstd.parquet create mode 100644 lib/vector-vrl/functions/tests/data/vpc_flow_gzip.parquet diff --git a/Cargo.lock b/Cargo.lock index 1ecb861658bbe..efb99844a473f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -13116,9 +13116,11 @@ dependencies = [ name = "vector-vrl-functions" version = "0.1.0" dependencies = [ + "chrono", "dnstap-parser", "enrichment", "indoc", + "parquet", "vector-vrl-category", "vector-vrl-metrics", "vrl", diff --git a/lib/vector-vrl/functions/Cargo.toml b/lib/vector-vrl/functions/Cargo.toml index 874ffbd759ad6..1e7e88bf6443e 100644 --- a/lib/vector-vrl/functions/Cargo.toml +++ b/lib/vector-vrl/functions/Cargo.toml @@ -13,6 +13,8 @@ enrichment = { path = "../enrichment" } dnstap-parser = { path = "../dnstap-parser", optional = true } vector-vrl-metrics = { path = "../metrics", optional = true } vector-vrl-category.workspace = true +chrono.workspace = true +parquet = { version = "39.0.0", default-features = false, features = ["snap", "flate2", "zstd"] } [features] default = ["dnstap", "vrl-metrics"] diff --git a/lib/vector-vrl/functions/src/lib.rs b/lib/vector-vrl/functions/src/lib.rs index d3846912aaba8..7ea4872550b5c 100644 --- a/lib/vector-vrl/functions/src/lib.rs +++ b/lib/vector-vrl/functions/src/lib.rs @@ -5,6 +5,7 @@ //! - Standard VRL library functions (`vrl::stdlib::all`) //! - Vector-specific functions (`vector_vrl::secret_functions`) //! - Enrichment table functions (`enrichment::vrl_functions`) +//! - Parquet parsing (`parse_parquet`) //! - DNS tap parsing functions (optional, with `dnstap` feature) #![deny(warnings)] @@ -12,6 +13,7 @@ use vrl::{compiler::Function, path::OwnedTargetPath}; pub mod get_secret; +pub mod parse_parquet; pub mod remove_secret; pub mod set_secret; pub mod set_semantic_meaning; @@ -51,7 +53,10 @@ pub fn all_without_vrl_stdlib() -> Vec> { fn iter_all_without_vrl_stdlib() -> impl Iterator> { let functions = secret_functions() .into_iter() - .chain(enrichment::vrl_functions()); + .chain(enrichment::vrl_functions()) + .chain(std::iter::once( + Box::new(parse_parquet::ParseParquet) as Box + )); #[cfg(feature = "dnstap")] let functions = functions.chain(dnstap_parser::vrl_functions()); diff --git a/lib/vector-vrl/functions/src/parse_parquet.rs b/lib/vector-vrl/functions/src/parse_parquet.rs new file mode 100644 index 0000000000000..1d98b6872956f --- /dev/null +++ b/lib/vector-vrl/functions/src/parse_parquet.rs @@ -0,0 +1,259 @@ +use std::panic::{self, AssertUnwindSafe}; + +use chrono::{DateTime, NaiveDate, Utc}; +use parquet::{ + file::reader::{FileReader, SerializedFileReader}, + record::{Field as PqField, Row}, +}; +use vrl::prelude::*; + +fn parse_parquet(value: Value) -> Resolved { + let bytes = value.try_bytes()?; + // parquet 39's row reader panics instead of returning errors on corrupt pages. + panic::catch_unwind(AssertUnwindSafe(|| read_rows(bytes))) + .unwrap_or_else(|_| Err("unable to parse parquet: malformed page data".into())) +} + +fn read_rows(bytes: Bytes) -> Resolved { + let reader = SerializedFileReader::new(bytes) + .map_err(|err| format!("unable to parse parquet: {err}"))?; + let rows = reader + .get_row_iter(None) + .map_err(|err| format!("unable to parse parquet: {err}"))?; + Ok(Value::Array(rows.map(|row| row_to_value(&row)).collect())) +} + +fn row_to_value(row: &Row) -> Value { + Value::Object( + row.get_column_iter() + .map(|(name, field)| (name.as_str().into(), field_to_value(field))) + .collect(), + ) +} + +fn field_to_value(field: &PqField) -> Value { + match field { + PqField::Null => Value::Null, + PqField::Bool(v) => Value::Boolean(*v), + PqField::Byte(v) => Value::Integer(i64::from(*v)), + PqField::Short(v) => Value::Integer(i64::from(*v)), + PqField::Int(v) => Value::Integer(i64::from(*v)), + PqField::Long(v) => Value::Integer(*v), + PqField::UByte(v) => Value::Integer(i64::from(*v)), + PqField::UShort(v) => Value::Integer(i64::from(*v)), + PqField::UInt(v) => Value::Integer(i64::from(*v)), + // Values above i64::MAX stay lossless as strings. + PqField::ULong(v) => i64::try_from(*v) + .map(Value::Integer) + .unwrap_or_else(|_| Value::from(v.to_string())), + PqField::Float(v) => float_to_value(f64::from(*v)), + PqField::Double(v) => float_to_value(*v), + // Strings keep full precision; floats would not. + PqField::Decimal(_) => Value::from(field.to_string()), + PqField::Str(v) => Value::from(v.as_str()), + PqField::Bytes(v) => Value::Bytes(Bytes::copy_from_slice(v.data())), + PqField::Date(days) => NaiveDate::from_ymd_opt(1970, 1, 1) + .and_then(|epoch| epoch.checked_add_signed(chrono::Duration::days(i64::from(*days)))) + .and_then(|date| date.and_hms_opt(0, 0, 0)) + .map(|dt| Value::Timestamp(dt.and_utc())) + .unwrap_or(Value::Integer(i64::from(*days))), + PqField::TimestampMillis(v) => DateTime::::from_timestamp_millis(*v) + .map(Value::Timestamp) + .unwrap_or(Value::Integer(*v)), + PqField::TimestampMicros(v) => DateTime::::from_timestamp_micros(*v) + .map(Value::Timestamp) + .unwrap_or(Value::Integer(*v)), + PqField::Group(row) => row_to_value(row), + PqField::ListInternal(list) => { + Value::Array(list.elements().iter().map(field_to_value).collect()) + } + PqField::MapInternal(map) => Value::Object( + map.entries() + .iter() + .map(|(key, value)| (map_key(key).into(), field_to_value(value))) + .collect(), + ), + } +} + +fn float_to_value(v: f64) -> Value { + NotNan::new(v).map(Value::Float).unwrap_or(Value::Null) +} + +fn map_key(key: &PqField) -> String { + match key { + PqField::Str(s) => s.clone(), + other => other.to_string(), + } +} + +#[derive(Clone, Copy, Debug)] +pub struct ParseParquet; + +impl Function for ParseParquet { + fn identifier(&self) -> &'static str { + "parse_parquet" + } + + fn usage(&self) -> &'static str { + indoc! {" + Parses the provided `value` as an Apache Parquet file and returns its rows as an + array of objects. Snappy, gzip and zstd column compression are supported. + Timestamps and dates become timestamps, decimals become strings, and NaN + floats become null. + "} + } + + fn category(&self) -> &'static str { + Category::Parse.as_ref() + } + + fn return_kind(&self) -> u16 { + kind::ARRAY + } + + fn parameters(&self) -> &'static [Parameter] { + const PARAMETERS: &[Parameter] = &[Parameter::required( + "value", + kind::BYTES, + "The Parquet file contents to parse.", + )]; + PARAMETERS + } + + fn examples(&self) -> &'static [Example] { + &[example! { + title: "Invalid Parquet", + source: r#"parse_parquet!("not parquet")"#, + result: Err( + r#"function call error for "parse_parquet" at (0:29): unable to parse parquet: Parquet error: Invalid Parquet file. Corrupt footer"#, + ), + }] + } + + fn compile( + &self, + _state: &TypeState, + _ctx: &mut FunctionCompileContext, + arguments: ArgumentList, + ) -> Compiled { + let value = arguments.required("value"); + Ok(ParseParquetFn { value }.as_expr()) + } +} + +#[derive(Debug, Clone)] +struct ParseParquetFn { + value: Box, +} + +impl FunctionExpression for ParseParquetFn { + fn resolve(&self, ctx: &mut Context) -> Resolved { + let value = self.value.resolve(ctx)?; + parse_parquet(value) + } + + fn type_def(&self, _: &TypeState) -> TypeDef { + TypeDef::array(Collection::from_unknown(Kind::object(Collection::any()))).fallible() + } +} + +#[cfg(test)] +mod tests { + use super::*; + use vrl::value; + + const VPC_FLOW_GZIP: &[u8] = include_bytes!("../tests/data/vpc_flow_gzip.parquet"); + const TYPES_ZSTD: &[u8] = include_bytes!("../tests/data/types_zstd.parquet"); + + fn parse(bytes: &[u8]) -> Resolved { + parse_parquet(Value::Bytes(Bytes::copy_from_slice(bytes))) + } + + fn ts(s: &str) -> Value { + Value::Timestamp(DateTime::parse_from_rfc3339(s).unwrap().with_timezone(&Utc)) + } + + #[test] + fn flat_rows_across_row_groups() { + let rows = parse(VPC_FLOW_GZIP).unwrap(); + let rows = rows.as_array().unwrap(); + assert_eq!(rows.len(), 3); + assert_eq!( + rows[0], + value!({ + "version": 2, + "account_id": "123456789012", + "interface_id": "eni-0a1b2c3d", + "srcaddr": "10.0.0.1", + "dstaddr": "10.0.1.1", + "srcport": 443, + "dstport": 49152, + "protocol": 6, + "packets": 10, + "bytes": 8400, + "start": 1727136000, + "end": 1727136060, + "action": "ACCEPT", + "log_status": "OK", + }) + ); + assert_eq!(rows[1].get("srcport"), Some(&Value::Null)); + assert_eq!(rows[2].get("action"), Some(&value!("REJECT"))); + } + + #[test] + fn type_mapping() { + let rows = parse(TYPES_ZSTD).unwrap(); + let rows = rows.as_array().unwrap(); + assert_eq!(rows.len(), 1); + let row = &rows[0]; + let expected = [ + ("b", value!(true)), + ("i8", value!(-8)), + ("u64", value!("18446744073709551615")), + ("f32", Value::Null), + ("f64", value!(1.5)), + ("dec", value!("-12345.67")), + ("s", value!("hello")), + ("bin", Value::Bytes(Bytes::from_static(b"\x00\xff"))), + ("d", ts("2024-09-24T00:00:00Z")), + ("ts_ms", ts("2024-09-24T01:02:03.456Z")), + ("ts_us", ts("2024-09-24T01:02:03.456789Z")), + ("st", value!({"x": 1, "y": "a"})), + ("lst", value!([1, null, 3])), + ("m", value!({"k1": 1, "k2": 2})), + ("los", value!([{"n": "a"}, {"n": "b"}])), + ]; + assert_eq!(row.as_object().unwrap().len(), expected.len()); + for (name, want) in expected { + assert_eq!(row.get(name), Some(&want), "column {name}"); + } + } + + #[test] + fn truncated_file_errors() { + let err = parse(&VPC_FLOW_GZIP[..VPC_FLOW_GZIP.len() - 1]).unwrap_err(); + assert!( + err.to_string().starts_with("unable to parse parquet"), + "{err}" + ); + } + + #[test] + fn corrupt_page_header_errors_instead_of_panicking() { + // Flipping the first page-header byte panics parquet 39's row reader. + let mut bytes = VPC_FLOW_GZIP.to_vec(); + bytes[4] ^= 0xff; + let err = parse(&bytes).unwrap_err(); + assert_eq!( + err.to_string(), + "unable to parse parquet: malformed page data" + ); + } + + #[test] + fn non_bytes_input_errors() { + assert!(parse_parquet(value!(1)).is_err()); + } +} diff --git a/lib/vector-vrl/functions/tests/data/types_zstd.parquet b/lib/vector-vrl/functions/tests/data/types_zstd.parquet new file mode 100644 index 0000000000000000000000000000000000000000..5c5b0668d9677182915bff65a4a78377f3b22950 GIT binary patch literal 5147 zcmb7IO>7%Q6rNdcj2qlG3E5?htSTTpN@>8(UvZ!aW)s&AbwU!Fq=}HgPFy#Q^Fy38 zKNo~lA%r-fszMwP;!q(D2nhs;0|$fD}4M80-Bw`FGD9BaCSvXq+3m5i#z` zqf3`FtKZAmpSNz^0+24_enjg~XgX>wdIDyh zkhPK8QZJzK$*l>XUp@j%xDDdIRH{@q8cCquZ8Tx2S8Ld)al!{1?pIxAfi(h{8oOn| zw>4U`zL}GZHO6P@s8vrWL7mhZyV-jC{L@$Oy~dm1N8A~!HUgg7*wfw~pYg#5N58!< z9Ly);&RDe(@&&c6`I^-ZQVVkFsI};cwezGV*czw3+H7^AU!k!vr{O>;SiI00s>c|} z!|tygDIG5|hEVtksTYO}W_@t9ILAxGov|XWt*P}k(R#dyiw9XKjJ}~q84qc|YZ(vt z=%WQ?w%U)zs{=rN!btVe`Gn^rB&4klSlVj82Yr0sf!5GR*Az0Xue!wWJi}norV-WB zQS0@*#So6sy? zWQGQBZ6%#Jk$>;52;LEf4w}%;>7hg%R8PRYz>Em0E)EVbqn$>lhw!dxH$b~wT1vD- zExY@k!6#Zr!G>28EFX8JP;q5lQSs$OG+%LP+P16Du2mCV`ByvRXzZ|~dZ(J`Ms**l zhat%7wxK&9dk)FILju#G3e0OpIDZBlhILD?te`@rb4RICS}Uz@Gce?9V5a2}7tD^V zmiC{$QmE{d5ImY-RtPk1>{y&Onnll0Wuw}R0;XI7B&77$9SBp4#bY-+1&bRjmI$*L z6y1oLnt9&5C?aP9cXSNw;Gf3MoD9bVtt|A_V#6%l5yMWA_jPW5%Eb*X*5m3*ZgDZk zaG4_xav{v;%&S~n*Ts(^fMka|x7VDV zfIj2qPh7m+CVtlA-p%Wc%vN{s;XKD&9$38;lXgHMLtdpLH~ z!RA#$eubOwbMa+FT#aB!8}Hlzmf0SwZ@i0!JBaZ1%s09C?TGmB2pwFt>oy3a`+(dP z#Fq(1EugD+if`Is1s!tL25^1r(p^DnBNMV}*_ToAQWRQt8<*_=XWCJvRZKUIiC>Qb zQ}80i#Szs0X9Jx|^}>O}_v(su^SJoxI1~$yVeee`q!z%i=<)2wUF=xXOr)C`+HaPuE7e(n<2yPz(&ajyn=PYO+s zorMH+e+Fs3<{g}PU_vsw`5HGr;^O;majl!yX6;f9;F^(`&@{}*1$K(=16C(5LBGq* zuerEtimwdlfOnTdi|eGO9Y7*Tp~mDiA%~~AuXY?am~Otz&G)$Y?rw4IZt971xdwC4 zDTg_LCT#PUyG%_)%ug&9i(64SGxOB^BkI1j-iRjgGgQt4AiHmn(x|F}P7b44UXFo98errs%61v#v0Z?Z`{?%MF2L9eDCd+6 zuJ8j9C^`zIoSaG~0n#3`NnSt_b>amMnccgxv@2)P{^_ZMWD<4LsaZ72;X|2{ zDH$-A@Si-tGD`4=wCREm_@Xv(pth@pN}*MDq=_?FO1 zQBf6#P*y>pY#%Jca}(noL4Jz2MSkdI55q(5qkZzgwI3zCKt0Hw_$H0P^=yrSwgs_I2ad#%F#Ddlo*@?!AQ7GL)4HS~@zc>`uP)9{;Q0e&)3tL*oln&7j1#Xd_sFYp|I&I~ zyu6T-lSPMbTo&E^I6Xm0)25>$C}!Fs*@c4%D(M39W}T%A1{t|t^H=+^sef*EV3Od- zAM(PbuR3|~U;fCOUB$IV{RMuYaQiE5kO>O<2=fcl-lbyUVxoL-)>iAMalcUj3j{?+ zKwZ5bdyH8toIBIK9qW>AO3uRzm0D9hy8Js Ku_^d(!G8h1N!L>V literal 0 HcmV?d00001 diff --git a/lib/vector-vrl/functions/tests/data/vpc_flow_gzip.parquet b/lib/vector-vrl/functions/tests/data/vpc_flow_gzip.parquet new file mode 100644 index 0000000000000000000000000000000000000000..ed81aaee0d3ecb3f4b0562023a5e0fc2a6e4ac9c GIT binary patch literal 7519 zcmbtZeM}tJ5r4ZrpD!3yBuq>HoG%#-<#jeym>P_5L4rR$t8ItU$-PmWG5jHKI0*HiPc&vks##9yN*8N z0$w80PU+x&m|Q8jC3lVD5@9rdmr0$8tEz$*#bOf4bUI1M@8YBH-iNYWQmquVBHm6W z624HN+20iW%|gf*5=r{kPmNaEpf{7ca(^HgYHB{v((3aElp^knX%Kzu&E=+Os5{Wr zd`uvJm-?PvPlL?Z*?M2np9*Nfv{G*79BJT_!rkDCVz1v$pbwU|sFC|!H$CL}={?_2 z>>c3o`x@|HK;e{s9oPYpoP6maRd>Bl>R1;xxD;Ft-g#`{SrY!^-uFrZ`wd`!gEL^; zD7*RAwM9Y}zWK=q_EGk#fKi5bS*&BvJZER>Ksucwq;B6Xmu_mM)M7PNYgt@Oys-$V zPe$Ju!)Uk0Hi5T_s|ns=Y!^5?REX_j2UkiV8&oZksC1>s1MjaKvB$(?qqy8UW`T2r zS!;)RJ9Un9)K!pw{pW>cyBJnHWq^j4;sFP<3+%NSb*O`lw_m-yNZgNBeqSK}`_9S* zdo7k~^Vo`~J5Qk3J=Rf&S{$_Te&X}_KM>dX*lzNVLpOTu)ewY_1`uEI5S>CbtfQM( z+;|i$p+3NTHVz0(wLybcwHLRnK)ObKTWX-c0#NtG5db$F3z4k{JYA_1co`Ggk z=T;+;_SgwUf3Aa9oY%`S4(%$*hyQhxxi{Y%!uYnq5|1BI;Kvb$Z62|&R^q0rRf^@2 zi)5zti&ZO*4!$g^Pa9V&=NjazRnyVnk71R{ zEgC2ioU64vYg8+ZbiTcPsob==CTL!tf<~nZI)r>|T}W1`++^aVKiw#a>TlqvKsiFS z4XYMFC{RNQ#2-dqt4r+T&mAYO_=fvE;v7zRt(Ca2DZ78l>~D7->2TT*FiMu0jE$0PAdkrc^zKS z@;Z$E^57nO9Ue7w;1ue>X~m!%>9B51@?<~BPTen%pDV*<=9S45vXZiojW0GGpnyZtkR(1O3e!l#PrJo`%V}sY!iU zc6;}GS1GBHL~j#uS9%S*C70}y=LPwWKu=Il>5kF6A3GsQ#BYO2cpTd2_ELhU4<(oD zl1)Bj!ha>mPm6R(q^m*=1V`y2Yk`n`UOSMY9WlWyyXc2o7>v2ov-EvVVfg-Xgu4QL z$4x)piU_=57Pod*$+fly@F?MuKN9Hc9{SI11<7lBw~ZtlJFCYUlbqk#%Vl)=KG@q5 zeVRo&?yZQwJ}8PBJ#v}sOM<*A(DUW=x7$HBS;WF)+l|@EHsXfsZJYxxd_|xiSI`$L zKzO0if|O?BAhnbh?&9IDpDHH%x*&g7q_=j^^E+zpsl9TBl|Qo_>AXl+s_1+bAX%Fw zu*PQ8u~l1HM&L*nMEYqpy<81Q)_w?)tfP*RFyEMPo|fq^B*3vYJcwf*-4KT-4?UYP zs2>ROy8@k8=+#|#Dy*H3WjkcW_NGdx& zn37G2)oJ!Hrc^4*kUHKM=uf^vmueA#r&8Vp_*fTUm0ZcTW32(ERLUenQu^Un>6M2H zlGpYZ8%gFV33bLo4N=4Q_U>N#!d}?h61{{)IxCZ0YeRM+m2%ndQ2K2OvI~8JytZp_ zCfl4!x$rA>^nr48X&Mk-}U z@ELJwKV5nRkgQ+#V2#bFW2=l*%8;Nuyx2gO>jBC74Ub6HQO8JnD&;sY`{=Dkz_EU( zBaU@+LmZww3aM0-^QrV|fZp-rsjz;aTed?kw>@jcsZ^9%DxD9}*MlgxR4R2D^5m5& zS%eHJ{|s_zjZ&`0wQRz8m=%3T*>IRfqA`i8s}U zH<$<-%(XT)wgA!I!6t=gaFmfqsOmWUgeOyh^nezOpGilbm;-rQ<3M9$AQl9B;C0VQ zW=~!;Hw{)HbZ5srwx)W_w)f0TazJ7@F>#av$M8jwL0=bcN^r$+( z=j-dIIj^seB8U+RY`<~+Y$_O^?hCX|ve(_*Uuu{IL5%b5y8qD0=Ka!u4G+U!&7^|h zhvR&{zCO|hLCE&I)cQGX^jLmZ*Ns zWATRsSdBL=*frGG8z0sNBf~glCgBJwbI1&$E8DGpFU5!-t7dJCV#jIhD!PYZDXo$!vCfrdIFXYXd>}Bb(p5 S3N5?C|D+0j2OEPwt^OZTYUiN< literal 0 HcmV?d00001 From f51dc33e8d61da07539fa23d5c9a0e64dca33605 Mon Sep 17 00:00:00 2001 From: sundaresanr Date: Thu, 24 Sep 2026 20:09:13 -0700 Subject: [PATCH 2/2] test(vrl): cover snappy-compressed parquet in parse_parquet Run the VPC flow row assertions against a snappy-compressed copy of the fixture as well as the gzip one, since snappy is the default codec for most Parquet writers. --- lib/vector-vrl/functions/src/parse_parquet.rs | 14 ++++++++++++-- .../functions/tests/data/vpc_flow_snappy.parquet | Bin 0 -> 6580 bytes 2 files changed, 12 insertions(+), 2 deletions(-) create mode 100644 lib/vector-vrl/functions/tests/data/vpc_flow_snappy.parquet diff --git a/lib/vector-vrl/functions/src/parse_parquet.rs b/lib/vector-vrl/functions/src/parse_parquet.rs index 1d98b6872956f..96aa78cb41333 100644 --- a/lib/vector-vrl/functions/src/parse_parquet.rs +++ b/lib/vector-vrl/functions/src/parse_parquet.rs @@ -164,6 +164,7 @@ mod tests { use vrl::value; const VPC_FLOW_GZIP: &[u8] = include_bytes!("../tests/data/vpc_flow_gzip.parquet"); + const VPC_FLOW_SNAPPY: &[u8] = include_bytes!("../tests/data/vpc_flow_snappy.parquet"); const TYPES_ZSTD: &[u8] = include_bytes!("../tests/data/types_zstd.parquet"); fn parse(bytes: &[u8]) -> Resolved { @@ -175,8 +176,17 @@ mod tests { } #[test] - fn flat_rows_across_row_groups() { - let rows = parse(VPC_FLOW_GZIP).unwrap(); + fn flat_rows_across_row_groups_gzip() { + assert_vpc_flow_rows(VPC_FLOW_GZIP); + } + + #[test] + fn flat_rows_across_row_groups_snappy() { + assert_vpc_flow_rows(VPC_FLOW_SNAPPY); + } + + fn assert_vpc_flow_rows(bytes: &[u8]) { + let rows = parse(bytes).unwrap(); let rows = rows.as_array().unwrap(); assert_eq!(rows.len(), 3); assert_eq!( diff --git a/lib/vector-vrl/functions/tests/data/vpc_flow_snappy.parquet b/lib/vector-vrl/functions/tests/data/vpc_flow_snappy.parquet new file mode 100644 index 0000000000000000000000000000000000000000..2b28ab46f8ce89bcd1c31cefaf9f69717c8ba822 GIT binary patch literal 6580 zcmbVRPiP$18J`)CR}*EEtYpn(HrXUfYCjZWOZ!KXqhbf%uKrn8?9~!SmMDbX*_CWd z)~c(ORqIr8=)s2)N-?IG&_l?fhZ0Kh!7atmkV7dUm|{W>CYT&T$jOHsLMZ*dH}lQR z?i=l{J4kQd`@Z*nzxU^RzxjlUyp%C@(=g+Ure-RGilWCgMNv$}95k)t;7XV=GnTS- zw7R9GjD%VP1K@aS7~Q(6>Iw`QHjffo;#3?QnQU%+V)EphQ|U~$9eGk2iHH4mORE?3 z+ScV0>2ju$b#m1x^swKKhg%WoX*0VBOTeA5Z3E|?9#yq;#;c}B)1y&KFrb_8*;LE4 zf*;ZVHEoXj;J@$@Got}|rUN?oJaiZlf!=b6e^a9l5v{h4rKOFCx7!M#9a2>d4>4g5 z(Oiv`a!*kvU^Cjc*VfTv>*&L<2wEXT2n(_L7Q`5gz{rsi(Q51HN5r5NfhYpaJ*>p3 zei(rdEKWQUvOzs;gZcnNgtZ&f7sD>YqWhB9U{DX+pgzbXTe~qS>f<#eQN6dkCh|b6 zdVP>dwsvDW<&8dlAQO3PgZd(qY>i+VGHnus{-b!Zyik}azSF*%No6GFdTj6AiY3G) zZYBsTHiW8k{%iz}cK)mlzY*hGZb?|`v?q}|&ATN%NgW}>`dY;8K&d+rU5Hn0Vr{Z~ zvO=p4qlf*T)Oo!_J3;8SnWCMHANSQLJ(`O^CRNM%`j+b%G1CDNe7>edA-0^B1*hFU zH(N)mTS7dBS`qVWK`Z_<3JtBcjwPZ6t!`)-)$-iX8faP2ia-H$5 zuuZT)zwLUX>2X4jpk-_~>dm@S--I*)C$wF5Hfqfsw{^I3y;-9=;35R}I#k=Lx^s(_ zohHmbJcItt`ucm&xw*UJnXOpzc=CIuq)aV2p~MnNf8C~L>DKR6>$b|$%;=n$CNY79 zCEz_u!0XiOu_RM4@)0@iA&2%*fd5ss{;slbRQ8dYf@AJ1*%F17PDYWmoJfe~kB8XT znxF|Mrc#>PD-E45G2B(zcRIVRBLitTWszx-Masy-kqe+GA!OL1k7Mk?;Q+ZkKkbI> zpR5;a0@*#;yM63;eQ>rN@>K~P&sM_^dP~%Jkt?uotJX&iy23c?d=fk<=)*D2h+TP3jnty=$4*`24^5B(`SB4GuN2y4$= z?wt)W-FSxmD-KL@!ir3?wRa|=2qC72&$17n1tvLRMJCzWGZX4fi03Z}_V_vAk<(S= zk*!^MNIX0>6I36l)-9F&IKVy|z*`~bveMC8cH~_P>7fArmGx_t-8jM?4kB*wy^!$? zAJ6)B(9)SgVOOQn>~IW8AwVgd4;4oDq^gJIF9xOgrt;^KTooo z$pE=MEtcopicFG(bH)!fL=BzoqZioS7vO9=q{$Mx_)OBQee6I=3H!ermrCcCf9|H`_AO%!pS|^I&iNAj;&YbsJ)U3Zvz)Kuoqu;9@SL%=v9WA12kQa%(gm>|7n+s^ zqmX)T_DXeb^I8S!`Ql0FKmCTsS8(}6{}3N8J-=DGy2MWsK5h>@i())fmd-@;p>_UT zdgbDRSl>ML_r_nOcBk=k3lr5jTqEc2P=9ay6>1+e9`|BCe`zJ#+^o*O&DW^EH~t#6 zz47x%;)jAbZybNUk}F+X$xiJE_af~V&x;@^QhUbRKiuT-c{$L`Z@^jYRdTQo=c&Ip zzAz0zD9*dn_u}3ojC%+uVioy1sgK_R8wJQ`@U(u_RGst zyEq=^2~&6QK6}rM*fh6s`m|yXZSZ`k|meK?zg{}*JE-@~8m PJ^m;I2}LQw|7-pS+t;@z literal 0 HcmV?d00001