diff --git a/lib/codecs/src/encoding/format/parquet.rs b/lib/codecs/src/encoding/format/parquet.rs index bcba326b60aba..3f00e476967b9 100644 --- a/lib/codecs/src/encoding/format/parquet.rs +++ b/lib/codecs/src/encoding/format/parquet.rs @@ -7,7 +7,10 @@ use parquet::{ column::writer::{ColumnWriter::*, ColumnWriterImpl}, data_type::DataType, errors::ParquetError, - file::{properties::WriterProperties, writer::SerializedFileWriter}, + file::{ + properties::{EnabledStatistics, WriterProperties}, + writer::SerializedFileWriter, + }, schema::{ parser::parse_message_type, types::{BasicTypeInfo, ColumnDescriptor, Type, TypePtr}, @@ -302,6 +305,11 @@ impl Encoder> for ParquetSerializer { let props = Arc::new( WriterProperties::builder() .set_compression(self.compression) + // Page-level statistics duplicate field values in the page + // header, which can exceed pyarrow's 16 MB header limit and + // leave the file unreadable. Chunk-level statistics are + // truncated and stay in the footer. + .set_statistics_enabled(EnabledStatistics::Chunk) .build(), ); let mut parquet_writer = @@ -1260,6 +1268,39 @@ mod tests { ); } + #[test] + fn large_value_does_not_inflate_page_header() { + // A page header carries min and max verbatim, so with page statistics + // a one-row file lands at ~3x the value (value + min + max). pyarrow + // rejects headers over 16 MB; uncompressed_size, which counts headers, + // is the cheapest place to catch that. + let message_type = r#" + message test { + required binary blob (STRING); + } + "#; + let schema = Arc::new(parse_message_type(message_type).unwrap()); + let mut encoder = ParquetSerializer::new(schema, Compression::UNCOMPRESSED); + + let blob = "x".repeat(1_000_000); + let blob_len = blob.len() as i64; + let mut buffer = BytesMut::new(); + let events: Vec = vec![LogEvent::from(btreemap! { "blob" => blob }).into()]; + encoder.encode(events, &mut buffer).unwrap(); + + let reader = SerializedFileReader::new(buffer.freeze()).unwrap(); + let column = reader.metadata().row_group(0).column(0); + + // Allow generous slack for thrift framing, but nowhere near a second + // copy of the value. + assert!( + column.uncompressed_size() < blob_len + blob_len / 2, + "page header carries a copy of the value: uncompressed_size {} vs value {}", + column.uncompressed_size(), + blob_len + ); + } + #[test] fn illegal_list_scheme() { let config = ParquetSerializerConfig {