This is an automated email from the ASF dual-hosted git repository.
etseidl pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/arrow-rs.git
The following commit(s) were added to refs/heads/main by this push:
new a297bdb88c fix(parquet): add early rejection of unsupported encodings
(#11266)
a297bdb88c is described below
commit a297bdb88ca91b4c7b3b2fd17c4dde191a250e7f
Author: Hippolyte Barraud <[email protected]>
AuthorDate: Tue Sep 29 00:31:38 2026 -0400
fix(parquet): add early rejection of unsupported encodings (#11266)
# Which issue does this PR close?
- Closes #11265.
# Rationale for this change
The writer accepts some encoding and physical-type combinations that no
encoder can represent correctly. Invalid configurations can therefore
fail only when values are written, when a dictionary falls back, or when
a reader encounters the output.
# What changes are included in this PR?
- Define the supported value-encoding matrix explicitly.
- Validate encoding and physical-type combinations at encoder
construction, including the Arrow byte-array fallback path.
- Reject invalid dictionary fallback configurations even when the
fallback would never be used.
- Add a construction-time validation matrix covering physical types,
writer versions, and dictionary settings, and update existing error
expectations.
Supported encodings and their wire representations are unchanged.
# Are these changes tested?
Yes. The tests cover the supported encoding matrix at column
construction, with dictionary encoding both enabled and disabled and
with both Parquet writer versions. The Arrow writer regression now
expects the invalid DOUBLE/DELTA_BINARY_PACKED configuration to fail
earlier.
# Are there any user-facing changes?
Yes. Unsupported encoding configurations are rejected when the physical
column encoder is constructed, before values reach that encoder. Invalid
dictionary fallbacks are rejected even if the column would remain
dictionary-encoded.
This deliberately changes failure timing and error messages for invalid
configurations. It does not change supported encodings, valid output, or
public API signatures.
Signed-off-by: Hippolyte Barraud <[email protected]>
---
parquet/src/encodings/encoding/mod.rs | 174 ++++++++++++++++++++++++++++++++--
parquet/tests/arrow_writer/mod.rs | 4 +-
2 files changed, 168 insertions(+), 10 deletions(-)
diff --git a/parquet/src/encodings/encoding/mod.rs
b/parquet/src/encodings/encoding/mod.rs
index 05117a625a..4fb8580c70 100644
--- a/parquet/src/encodings/encoding/mod.rs
+++ b/parquet/src/encodings/encoding/mod.rs
@@ -120,15 +120,32 @@ pub(crate) mod private {
"Cannot initialize this encoding through this function"
));
}
- Encoding::RLE => Box::new(RleValueEncoder::new()),
- Encoding::DELTA_BINARY_PACKED =>
Box::new(DeltaBitPackEncoder::new()),
- Encoding::DELTA_LENGTH_BYTE_ARRAY =>
Box::new(DeltaLengthByteArrayEncoder::new()),
- Encoding::DELTA_BYTE_ARRAY =>
Box::new(DeltaByteArrayEncoder::new()),
+ Encoding::RLE => match T::get_physical_type() {
+ Type::BOOLEAN => Box::new(RleValueEncoder::new()),
+ physical_type => return
Err(unsupported_column_encoding(encoding, physical_type)),
+ },
+ Encoding::DELTA_BINARY_PACKED => match T::get_physical_type() {
+ Type::INT32 | Type::INT64 =>
Box::new(DeltaBitPackEncoder::new()),
+ physical_type => return
Err(unsupported_column_encoding(encoding, physical_type)),
+ },
+ Encoding::DELTA_LENGTH_BYTE_ARRAY => match T::get_physical_type() {
+ Type::BYTE_ARRAY =>
Box::new(DeltaLengthByteArrayEncoder::new()),
+ physical_type => return
Err(unsupported_column_encoding(encoding, physical_type)),
+ },
+ Encoding::DELTA_BYTE_ARRAY => match T::get_physical_type() {
+ Type::BYTE_ARRAY | Type::FIXED_LEN_BYTE_ARRAY => {
+ Box::new(DeltaByteArrayEncoder::new())
+ }
+ physical_type => return
Err(unsupported_column_encoding(encoding, physical_type)),
+ },
Encoding::BYTE_STREAM_SPLIT => match T::get_physical_type() {
Type::FIXED_LEN_BYTE_ARRAY =>
Box::new(VariableWidthByteStreamSplitEncoder::new(
descr.type_length(),
)),
- _ => Box::new(ByteStreamSplitEncoder::new()),
+ Type::INT32 | Type::INT64 | Type::FLOAT | Type::DOUBLE => {
+ Box::new(ByteStreamSplitEncoder::new())
+ }
+ physical_type => return
Err(unsupported_column_encoding(encoding, physical_type)),
},
Encoding::ALP => {
return Err(general_err!(
@@ -175,6 +192,14 @@ pub(crate) mod private {
}
}
+fn unsupported_column_encoding(encoding: Encoding, physical_type: Type) ->
ParquetError {
+ nyi_err!(
+ "Encoding {} is not supported for physical type {:?}",
+ encoding,
+ physical_type
+ )
+}
+
// ----------------------------------------------------------------------
// Plain encoding
@@ -849,8 +874,6 @@ mod tests {
// supported encodings
create_and_check_encoder::<Int32Type>(0, Encoding::PLAIN, None);
create_and_check_encoder::<Int32Type>(0,
Encoding::DELTA_BINARY_PACKED, None);
- create_and_check_encoder::<Int32Type>(0,
Encoding::DELTA_LENGTH_BYTE_ARRAY, None);
- create_and_check_encoder::<Int32Type>(0, Encoding::DELTA_BYTE_ARRAY,
None);
create_and_check_encoder::<BoolType>(0, Encoding::RLE, None);
// error when initializing
@@ -868,6 +891,22 @@ mod tests {
"Cannot initialize this encoding through this function"
)),
);
+ create_and_check_encoder::<Int32Type>(
+ 0,
+ Encoding::DELTA_LENGTH_BYTE_ARRAY,
+ Some(unsupported_column_encoding(
+ Encoding::DELTA_LENGTH_BYTE_ARRAY,
+ Type::INT32,
+ )),
+ );
+ create_and_check_encoder::<Int32Type>(
+ 0,
+ Encoding::DELTA_BYTE_ARRAY,
+ Some(unsupported_column_encoding(
+ Encoding::DELTA_BYTE_ARRAY,
+ Type::INT32,
+ )),
+ );
create_and_check_encoder::<Int32Type>(
0,
Encoding::ALP,
@@ -885,6 +924,127 @@ mod tests {
);
}
+ #[test]
+ fn column_construction_validates_encoding_fallback_matrix() {
+ use crate::file::properties::{WriterProperties, WriterVersion};
+ use crate::file::writer::SerializedFileWriter;
+ use std::panic::{AssertUnwindSafe, catch_unwind};
+ use std::sync::Arc;
+
+ fn check_encoder<T: DataType>(encoding: Encoding, supported: bool) {
+ let error = if supported {
+ None
+ } else if encoding == Encoding::ALP {
+ Some(general_err!(
+ "Encoding ALP only supports FLOAT and DOUBLE, got {}",
+ T::get_physical_type()
+ ))
+ } else {
+ Some(unsupported_column_encoding(
+ encoding,
+ T::get_physical_type(),
+ ))
+ };
+ create_and_check_encoder::<T>(4, encoding, error);
+ }
+
+ let encodings = [
+ Encoding::PLAIN,
+ Encoding::RLE,
+ Encoding::DELTA_BINARY_PACKED,
+ Encoding::DELTA_LENGTH_BYTE_ARRAY,
+ Encoding::DELTA_BYTE_ARRAY,
+ Encoding::BYTE_STREAM_SPLIT,
+ Encoding::ALP,
+ ];
+ for physical in [
+ Type::BOOLEAN,
+ Type::INT32,
+ Type::INT64,
+ Type::INT96,
+ Type::FLOAT,
+ Type::DOUBLE,
+ Type::BYTE_ARRAY,
+ Type::FIXED_LEN_BYTE_ARRAY,
+ ] {
+ let valid: &[Encoding] = match physical {
+ Type::BOOLEAN => &[Encoding::PLAIN, Encoding::RLE],
+ Type::INT32 | Type::INT64 => &[
+ Encoding::PLAIN,
+ Encoding::DELTA_BINARY_PACKED,
+ Encoding::BYTE_STREAM_SPLIT,
+ ],
+ Type::INT96 => &[Encoding::PLAIN],
+ Type::FLOAT | Type::DOUBLE => {
+ &[Encoding::PLAIN, Encoding::BYTE_STREAM_SPLIT,
Encoding::ALP]
+ }
+ Type::BYTE_ARRAY => &[
+ Encoding::PLAIN,
+ Encoding::DELTA_LENGTH_BYTE_ARRAY,
+ Encoding::DELTA_BYTE_ARRAY,
+ ],
+ Type::FIXED_LEN_BYTE_ARRAY => &[
+ Encoding::PLAIN,
+ Encoding::DELTA_BYTE_ARRAY,
+ Encoding::BYTE_STREAM_SPLIT,
+ ],
+ };
+ // Check the factory's returned errors directly, independently of
the
+ // infallible column writer constructor that unwraps those errors.
+ for encoding in encodings {
+ let supported = valid.contains(&encoding);
+ match physical {
+ Type::BOOLEAN => check_encoder::<BoolType>(encoding,
supported),
+ Type::INT32 => check_encoder::<Int32Type>(encoding,
supported),
+ Type::INT64 => check_encoder::<Int64Type>(encoding,
supported),
+ Type::INT96 => check_encoder::<Int96Type>(encoding,
supported),
+ Type::FLOAT => check_encoder::<FloatType>(encoding,
supported),
+ Type::DOUBLE => check_encoder::<DoubleType>(encoding,
supported),
+ Type::BYTE_ARRAY =>
check_encoder::<ByteArrayType>(encoding, supported),
+ Type::FIXED_LEN_BYTE_ARRAY => {
+ check_encoder::<FixedLenByteArrayType>(encoding,
supported)
+ }
+ }
+ }
+ for version in [WriterVersion::PARQUET_1_0,
WriterVersion::PARQUET_2_0] {
+ for dictionary in [false, true] {
+ for encoding in encodings {
+ let field = Arc::new(
+ SchemaType::primitive_type_builder("a", physical)
+ .with_length(4)
+ .build()
+ .unwrap(),
+ );
+ let schema = Arc::new(
+ SchemaType::group_type_builder("schema")
+ .with_fields(vec![field])
+ .build()
+ .unwrap(),
+ );
+ let props = Arc::new(
+ WriterProperties::builder()
+ .set_writer_version(version)
+ .set_dictionary_enabled(dictionary)
+ .set_encoding(encoding)
+ .build(),
+ );
+ let created = catch_unwind(AssertUnwindSafe(|| {
+ let mut writer =
+ SerializedFileWriter::new(Vec::new(), schema,
props).unwrap();
+ let mut group = writer.next_row_group().unwrap();
+ group.next_column().unwrap().unwrap();
+ }));
+ assert_eq!(
+ created.is_ok(),
+ valid.contains(&encoding),
+ "{physical:?} {version:?} dictionary={dictionary}
{encoding:?}"
+ );
+ }
+ }
+ }
+ }
+ }
+
#[test]
fn test_bool() {
BoolType::test(Encoding::PLAIN, TEST_SET_SIZE, -1);
diff --git a/parquet/tests/arrow_writer/mod.rs
b/parquet/tests/arrow_writer/mod.rs
index 195e855c40..b1f6a98ddb 100644
--- a/parquet/tests/arrow_writer/mod.rs
+++ b/parquet/tests/arrow_writer/mod.rs
@@ -45,9 +45,7 @@ use parquet::file::properties::WriterProperties;
use parquet::file::writer::SerializedFileWriter;
#[test]
-#[should_panic(
- expected = "DeltaBitPackDecoder only supports Int32Type, UInt32Type,
Int64Type, and UInt64Type"
-)]
+#[should_panic(expected = "Encoding DELTA_BINARY_PACKED is not supported for
physical type DOUBLE")]
fn test_delta_bit_pack_type() {
let props = WriterProperties::builder()
.set_column_encoding("col".into(), Encoding::DELTA_BINARY_PACKED)
