parquet/src/record/triplet.rs - arrow-experimental-rs-parquet2 - Git at Google

 // Licensed to the Apache Software Foundation (ASF) under one
 // or more contributor license agreements.  See the NOTICE file
 // distributed with this work for additional information
 // regarding copyright ownership.  The ASF licenses this file
 // to you under the Apache License, Version 2.0 (the
 // "License"); you may not use this file except in compliance
 // with the License.  You may obtain a copy of the License at
 //
 //   http://www.apache.org/licenses/LICENSE-2.0
 //
 // Unless required by applicable law or agreed to in writing,
 // software distributed under the License is distributed on an
 // "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 // KIND, either express or implied.  See the License for the
 // specific language governing permissions and limitations
 // under the License.

 use crate::basic::Type as PhysicalType;
 use crate::column::reader::{get_typed_column_reader, ColumnReader, ColumnReaderImpl};
 use crate::data_type::*;
 use crate::errors::{ParquetError, Result};
 use crate::record::api::Field;
 use crate::schema::types::ColumnDescPtr;

 /// Macro to generate simple functions that cover all types of triplet iterator.
 /// $func is a function of a typed triplet iterator and $token is a either {`ref`} or
 /// {`ref`, `mut`}
 macro_rules! triplet_enum_func {
   ($self:ident, $func:ident, $( $token:tt ),*) => ({
     match *$self {
       TripletIter::BoolTripletIter($($token)* typed) => typed.$func(),
       TripletIter::Int32TripletIter($($token)* typed) => typed.$func(),
       TripletIter::Int64TripletIter($($token)* typed) => typed.$func(),
       TripletIter::Int96TripletIter($($token)* typed) => typed.$func(),
       TripletIter::FloatTripletIter($($token)* typed) => typed.$func(),
       TripletIter::DoubleTripletIter($($token)* typed) => typed.$func(),
       TripletIter::ByteArrayTripletIter($($token)* typed) => typed.$func(),
       TripletIter::FixedLenByteArrayTripletIter($($token)* typed) => typed.$func()
     }
   });
 }

 /// High level API wrapper on column reader.
 /// Provides per-element access for each primitive column.
 pub enum TripletIter {
     BoolTripletIter(TypedTripletIter<BoolType>),
     Int32TripletIter(TypedTripletIter<Int32Type>),
     Int64TripletIter(TypedTripletIter<Int64Type>),
     Int96TripletIter(TypedTripletIter<Int96Type>),
     FloatTripletIter(TypedTripletIter<FloatType>),
     DoubleTripletIter(TypedTripletIter<DoubleType>),
     ByteArrayTripletIter(TypedTripletIter<ByteArrayType>),
     FixedLenByteArrayTripletIter(TypedTripletIter<FixedLenByteArrayType>),
 }

 impl TripletIter {
     /// Creates new triplet for column reader
     pub fn new(descr: ColumnDescPtr, reader: ColumnReader, batch_size: usize) -> Self {
         match descr.physical_type() {
             PhysicalType::BOOLEAN => TripletIter::BoolTripletIter(TypedTripletIter::new(
                 descr, batch_size, reader,
             )),
             PhysicalType::INT32 => TripletIter::Int32TripletIter(TypedTripletIter::new(
                 descr, batch_size, reader,
             )),
             PhysicalType::INT64 => TripletIter::Int64TripletIter(TypedTripletIter::new(
                 descr, batch_size, reader,
             )),
             PhysicalType::INT96 => TripletIter::Int96TripletIter(TypedTripletIter::new(
                 descr, batch_size, reader,
             )),
             PhysicalType::FLOAT => TripletIter::FloatTripletIter(TypedTripletIter::new(
                 descr, batch_size, reader,
             )),
             PhysicalType::DOUBLE => TripletIter::DoubleTripletIter(
                 TypedTripletIter::new(descr, batch_size, reader),
             ),
             PhysicalType::BYTE_ARRAY => TripletIter::ByteArrayTripletIter(
                 TypedTripletIter::new(descr, batch_size, reader),
             ),
             PhysicalType::FIXED_LEN_BYTE_ARRAY => {
                 TripletIter::FixedLenByteArrayTripletIter(TypedTripletIter::new(
                     descr, batch_size, reader,
                 ))
             }
         }
     }

     /// Invokes underlying typed triplet iterator to buffer current value.
     /// Should be called once - either before `is_null` or `current_value`.
     #[inline]
     pub fn read_next(&mut self) -> Result<bool> {
         triplet_enum_func!(self, read_next, ref, mut)
     }

     /// Provides check on values/levels left without invoking the underlying typed triplet
     /// iterator.
     /// Returns true if more values/levels exist, false otherwise.
     /// It is always in sync with `read_next` method.
     #[inline]
     pub fn has_next(&self) -> bool {
         triplet_enum_func!(self, has_next, ref)
     }

     /// Returns current definition level for a leaf triplet iterator
     #[inline]
     pub fn current_def_level(&self) -> i16 {
         triplet_enum_func!(self, current_def_level, ref)
     }

     /// Returns max definition level for a leaf triplet iterator
     #[inline]
     pub fn max_def_level(&self) -> i16 {
         triplet_enum_func!(self, max_def_level, ref)
     }

     /// Returns current repetition level for a leaf triplet iterator
     #[inline]
     pub fn current_rep_level(&self) -> i16 {
         triplet_enum_func!(self, current_rep_level, ref)
     }

     /// Returns max repetition level for a leaf triplet iterator
     #[inline]
     pub fn max_rep_level(&self) -> i16 {
         triplet_enum_func!(self, max_rep_level, ref)
     }

     /// Returns true, if current value is null.
     /// Based on the fact that for non-null value current definition level
     /// equals to max definition level.
     #[inline]
     pub fn is_null(&self) -> bool {
         self.current_def_level() < self.max_def_level()
     }

     /// Updates non-null value for current row.
     pub fn current_value(&self) -> Field {
         assert!(!self.is_null(), "Value is null");
         match *self {
             TripletIter::BoolTripletIter(ref typed) => {
                 Field::convert_bool(typed.column_descr(), *typed.current_value())
             }
             TripletIter::Int32TripletIter(ref typed) => {
                 Field::convert_int32(typed.column_descr(), *typed.current_value())
             }
             TripletIter::Int64TripletIter(ref typed) => {
                 Field::convert_int64(typed.column_descr(), *typed.current_value())
             }
             TripletIter::Int96TripletIter(ref typed) => {
                 Field::convert_int96(typed.column_descr(), typed.current_value().clone())
             }
             TripletIter::FloatTripletIter(ref typed) => {
                 Field::convert_float(typed.column_descr(), *typed.current_value())
             }
             TripletIter::DoubleTripletIter(ref typed) => {
                 Field::convert_double(typed.column_descr(), *typed.current_value())
             }
             TripletIter::ByteArrayTripletIter(ref typed) => Field::convert_byte_array(
                 typed.column_descr(),
                 typed.current_value().clone(),
             ),
             TripletIter::FixedLenByteArrayTripletIter(ref typed) => {
                 Field::convert_byte_array(
                     typed.column_descr(),
                     typed.current_value().clone().into(),
                 )
             }
         }
     }
 }

 /// Internal typed triplet iterator as a wrapper for column reader
 /// (primitive leaf column), provides per-element access.
 pub struct TypedTripletIter<T: DataType> {
     reader: ColumnReaderImpl<T>,
     column_descr: ColumnDescPtr,
     batch_size: usize,
     // type properties
     max_def_level: i16,
     max_rep_level: i16,
     // values and levels
     values: Vec<T::T>,
     def_levels: Option<Vec<i16>>,
     rep_levels: Option<Vec<i16>>,
     // current index for the triplet (value, def, rep)
     curr_triplet_index: usize,
     // how many triplets are left before we need to buffer
     triplets_left: usize,
     // helper flag to quickly check if we have more values/levels to read
     has_next: bool,
 }

 impl<T: DataType> TypedTripletIter<T> {
     /// Creates new typed triplet iterator based on provided column reader.
     /// Use batch size to specify the amount of values to buffer from column reader.
     fn new(descr: ColumnDescPtr, batch_size: usize, column_reader: ColumnReader) -> Self {
         assert!(
             batch_size > 0,
             "Expected positive batch size, found: {}",
             batch_size
         );

         let max_def_level = descr.max_def_level();
         let max_rep_level = descr.max_rep_level();

         let def_levels = if max_def_level == 0 {
             None
         } else {
             Some(vec![0; batch_size])
         };
         let rep_levels = if max_rep_level == 0 {
             None
         } else {
             Some(vec![0; batch_size])
         };

         Self {
             reader: get_typed_column_reader(column_reader),
             column_descr: descr,
             batch_size,
             max_def_level,
             max_rep_level,
             values: vec![T::T::default(); batch_size],
             def_levels,
             rep_levels,
             curr_triplet_index: 0,
             triplets_left: 0,
             has_next: false,
         }
     }

     /// Returns column descriptor reference for the current typed triplet iterator.
     #[inline]
     pub fn column_descr(&self) -> &ColumnDescPtr {
         &self.column_descr
     }

     /// Returns maximum definition level for the triplet iterator (leaf column).
     #[inline]
     fn max_def_level(&self) -> i16 {
         self.max_def_level
     }

     /// Returns maximum repetition level for the triplet iterator (leaf column).
     #[inline]
     fn max_rep_level(&self) -> i16 {
         self.max_rep_level
     }

     /// Returns current value.
     /// Method does not advance the iterator, therefore can be called multiple times.
     #[inline]
     fn current_value(&self) -> &T::T {
         assert!(
             self.current_def_level() == self.max_def_level(),
             "Cannot extract value, max definition level: {}, current level: {}",
             self.max_def_level(),
             self.current_def_level()
         );
         &self.values[self.curr_triplet_index]
     }

     /// Returns current definition level.
     /// If field is required, then maximum definition level is returned.
     #[inline]
     fn current_def_level(&self) -> i16 {
         match self.def_levels {
             Some(ref vec) => vec[self.curr_triplet_index],
             None => self.max_def_level,
         }
     }

     /// Returns current repetition level.
     /// If field is required, then maximum repetition level is returned.
     #[inline]
     fn current_rep_level(&self) -> i16 {
         match self.rep_levels {
             Some(ref vec) => vec[self.curr_triplet_index],
             None => self.max_rep_level,
         }
     }

     /// Quick check if iterator has more values/levels to read.
     /// It is updated as a result of `read_next` method, so they are synchronized.
     #[inline]
     fn has_next(&self) -> bool {
         self.has_next
     }

     /// Advances to the next triplet.
     /// Returns true, if there are more records to read, false there are no records left.
     fn read_next(&mut self) -> Result<bool> {
         self.curr_triplet_index += 1;

         if self.curr_triplet_index >= self.triplets_left {
             let (values_read, levels_read) = {
                 // Get slice of definition levels, if available
                 let def_levels = self.def_levels.as_mut().map(|vec| &mut vec[..]);

                 // Get slice of repetition levels, if available
                 let rep_levels = self.rep_levels.as_mut().map(|vec| &mut vec[..]);

                 // Buffer triplets
                 self.reader.read_batch(
                     self.batch_size,
                     def_levels,
                     rep_levels,
                     &mut self.values,
                 )?
             };

             // No more values or levels to read
             if values_read == 0 && levels_read == 0 {
                 self.has_next = false;
                 return Ok(false);
             }

             // We never read values more than levels
             if levels_read == 0 || values_read == levels_read {
                 // There are no definition levels to read, column is required
                 // or definition levels match values, so it does not require spacing
                 self.curr_triplet_index = 0;
                 self.triplets_left = values_read;
             } else if values_read < levels_read {
                 // Add spacing for triplets.
                 // The idea is setting values for positions in def_levels when current
                 // definition level equals to maximum definition level.
                 // Values and levels are guaranteed to line up, because of
                 // the column reader method.

                 // Note: if values_read == 0, then spacing will not be triggered
                 let mut idx = values_read;
                 let def_levels = self.def_levels.as_ref().unwrap();
                 for i in 0..levels_read {
                     if def_levels[levels_read - i - 1] == self.max_def_level {
                         idx -= 1; // This is done to avoid usize becoming a negative value
                         self.values.swap(levels_read - i - 1, idx);
                     }
                 }
                 self.curr_triplet_index = 0;
                 self.triplets_left = levels_read;
             } else {
                 return Err(general_err!(
                     "Spacing of values/levels is wrong, values_read: {}, levels_read: {}",
                     values_read,
                     levels_read
                 ));
             }
         }

         self.has_next = true;
         Ok(true)
     }
 }

 #[cfg(test)]
 mod tests {
     use super::*;

     use crate::file::reader::{FileReader, SerializedFileReader};
     use crate::schema::types::ColumnPath;
     use crate::util::test_common::get_test_file;

     #[test]
     #[should_panic(expected = "Expected positive batch size, found: 0")]
     fn test_triplet_zero_batch_size() {
         let column_path =
             ColumnPath::from(vec!["b_struct".to_string(), "b_c_int".to_string()]);
         test_column_in_file("nulls.snappy.parquet", 0, &column_path, &[], &[], &[]);
     }

     #[test]
     fn test_triplet_null_column() {
         let path = vec!["b_struct", "b_c_int"];
         let values = vec![];
         let def_levels = vec![1, 1, 1, 1, 1, 1, 1, 1];
         let rep_levels = vec![0, 0, 0, 0, 0, 0, 0, 0];
         test_triplet_iter(
             "nulls.snappy.parquet",
             path,
             &values,
             &def_levels,
             &rep_levels,
         );
     }

     #[test]
     fn test_triplet_required_column() {
         let path = vec!["ID"];
         let values = vec![Field::Long(8)];
         let def_levels = vec![0];
         let rep_levels = vec![0];
         test_triplet_iter(
             "nonnullable.impala.parquet",
             path,
             &values,
             &def_levels,
             &rep_levels,
         );
     }

     #[test]
     fn test_triplet_optional_column() {
         let path = vec!["nested_struct", "A"];
         let values = vec![Field::Int(1), Field::Int(7)];
         let def_levels = vec![2, 1, 1, 1, 1, 0, 2];
         let rep_levels = vec![0, 0, 0, 0, 0, 0, 0];
         test_triplet_iter(
             "nullable.impala.parquet",
             path,
             &values,
             &def_levels,
             &rep_levels,
         );
     }

     #[test]
     fn test_triplet_optional_list_column() {
         let path = vec!["a", "list", "element", "list", "element", "list", "element"];
         let values = vec![
             Field::Str("a".to_string()),
             Field::Str("b".to_string()),
             Field::Str("c".to_string()),
             Field::Str("d".to_string()),
             Field::Str("a".to_string()),
             Field::Str("b".to_string()),
             Field::Str("c".to_string()),
             Field::Str("d".to_string()),
             Field::Str("e".to_string()),
             Field::Str("a".to_string()),
             Field::Str("b".to_string()),
             Field::Str("c".to_string()),
             Field::Str("d".to_string()),
             Field::Str("e".to_string()),
             Field::Str("f".to_string()),
         ];
         let def_levels = vec![7, 7, 7, 4, 7, 7, 7, 7, 7, 4, 7, 7, 7, 7, 7, 7, 4, 7];
         let rep_levels = vec![0, 3, 2, 1, 2, 0, 3, 2, 3, 1, 2, 0, 3, 2, 3, 2, 1, 2];
         test_triplet_iter(
             "nested_lists.snappy.parquet",
             path,
             &values,
             &def_levels,
             &rep_levels,
         );
     }

     #[test]
     fn test_triplet_optional_map_column() {
         let path = vec!["a", "key_value", "value", "key_value", "key"];
         let values = vec![
             Field::Int(1),
             Field::Int(2),
             Field::Int(1),
             Field::Int(1),
             Field::Int(3),
             Field::Int(4),
             Field::Int(5),
         ];
         let def_levels = vec![4, 4, 4, 2, 3, 4, 4, 4, 4];
         let rep_levels = vec![0, 2, 0, 0, 0, 0, 0, 2, 2];
         test_triplet_iter(
             "nested_maps.snappy.parquet",
             path,
             &values,
             &def_levels,
             &rep_levels,
         );
     }

     // Check triplet iterator across different batch sizes
     fn test_triplet_iter(
         file_name: &str,
         column_path: Vec<&str>,
         expected_values: &[Field],
         expected_def_levels: &[i16],
         expected_rep_levels: &[i16],
     ) {
         // Convert path into column path
         let path: Vec<String> = column_path.iter().map(|x| x.to_string()).collect();
         let column_path = ColumnPath::from(path);

         let batch_sizes = vec![1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 128, 256];
         for batch_size in batch_sizes {
             test_column_in_file(
                 file_name,
                 batch_size,
                 &column_path,
                 expected_values,
                 expected_def_levels,
                 expected_rep_levels,
             );
         }
     }

     // Check values of a selectd column in a file
     fn test_column_in_file(
         file_name: &str,
         batch_size: usize,
         column_path: &ColumnPath,
         expected_values: &[Field],
         expected_def_levels: &[i16],
         expected_rep_levels: &[i16],
     ) {
         let file = get_test_file(file_name);
         let file_reader = SerializedFileReader::new(file).unwrap();
         let metadata = file_reader.metadata();
         // Get schema descriptor
         let file_metadata = metadata.file_metadata();
         let schema = file_metadata.schema_descr();
         // Get first row group
         let row_group_reader = file_reader.get_row_group(0).unwrap();

         for i in 0..schema.num_columns() {
             let descr = schema.column(i);
             if descr.path() == column_path {
                 let reader = row_group_reader.get_column_reader(i).unwrap();
                 test_triplet_column(
                     descr,
                     reader,
                     batch_size,
                     expected_values,
                     expected_def_levels,
                     expected_rep_levels,
                 );
             }
         }
     }

     // Check values for individual triplet iterator
     fn test_triplet_column(
         descr: ColumnDescPtr,
         reader: ColumnReader,
         batch_size: usize,
         expected_values: &[Field],
         expected_def_levels: &[i16],
         expected_rep_levels: &[i16],
     ) {
         let mut iter = TripletIter::new(descr.clone(), reader, batch_size);
         let mut values: Vec<Field> = Vec::new();
         let mut def_levels: Vec<i16> = Vec::new();
         let mut rep_levels: Vec<i16> = Vec::new();

         assert_eq!(iter.max_def_level(), descr.max_def_level());
         assert_eq!(iter.max_rep_level(), descr.max_rep_level());

         while let Ok(true) = iter.read_next() {
             assert!(iter.has_next());
             if !iter.is_null() {
                 values.push(iter.current_value());
             }
             def_levels.push(iter.current_def_level());
             rep_levels.push(iter.current_rep_level());
         }

         assert_eq!(values, expected_values);
         assert_eq!(def_levels, expected_def_levels);
         assert_eq!(rep_levels, expected_rep_levels);
     }
 }
	// Licensed to the Apache Software Foundation (ASF) under one
	// or more contributor license agreements. See the NOTICE file
	// distributed with this work for additional information
	// regarding copyright ownership. The ASF licenses this file
	// to you under the Apache License, Version 2.0 (the
	// "License"); you may not use this file except in compliance
	// with the License. You may obtain a copy of the License at
	//
	// http://www.apache.org/licenses/LICENSE-2.0
	//
	// Unless required by applicable law or agreed to in writing,
	// software distributed under the License is distributed on an
	// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
	// KIND, either express or implied. See the License for the
	// specific language governing permissions and limitations
	// under the License.

	use crate::basic::Type as PhysicalType;
	use crate::column::reader::{get_typed_column_reader, ColumnReader, ColumnReaderImpl};
	use crate::data_type::*;
	use crate::errors::{ParquetError, Result};
	use crate::record::api::Field;
	use crate::schema::types::ColumnDescPtr;

	/// Macro to generate simple functions that cover all types of triplet iterator.
	/// $func is a function of a typed triplet iterator and $token is a either {`ref`} or
	/// {`ref`, `mut`}
	macro_rules! triplet_enum_func {
	($self:ident, $func:ident, $( $token:tt ),*) => ({
	match *$self {
	TripletIter::BoolTripletIter($($token)* typed) => typed.$func(),
	TripletIter::Int32TripletIter($($token)* typed) => typed.$func(),
	TripletIter::Int64TripletIter($($token)* typed) => typed.$func(),
	TripletIter::Int96TripletIter($($token)* typed) => typed.$func(),
	TripletIter::FloatTripletIter($($token)* typed) => typed.$func(),
	TripletIter::DoubleTripletIter($($token)* typed) => typed.$func(),
	TripletIter::ByteArrayTripletIter($($token)* typed) => typed.$func(),
	TripletIter::FixedLenByteArrayTripletIter($($token)* typed) => typed.$func()
	}
	});
	}

	/// High level API wrapper on column reader.
	/// Provides per-element access for each primitive column.
	pub enum TripletIter {
	BoolTripletIter(TypedTripletIter<BoolType>),
	Int32TripletIter(TypedTripletIter<Int32Type>),
	Int64TripletIter(TypedTripletIter<Int64Type>),
	Int96TripletIter(TypedTripletIter<Int96Type>),
	FloatTripletIter(TypedTripletIter<FloatType>),
	DoubleTripletIter(TypedTripletIter<DoubleType>),
	ByteArrayTripletIter(TypedTripletIter<ByteArrayType>),
	FixedLenByteArrayTripletIter(TypedTripletIter<FixedLenByteArrayType>),
	}

	impl TripletIter {
	/// Creates new triplet for column reader
	pub fn new(descr: ColumnDescPtr, reader: ColumnReader, batch_size: usize) -> Self {
	match descr.physical_type() {
	PhysicalType::BOOLEAN => TripletIter::BoolTripletIter(TypedTripletIter::new(
	descr, batch_size, reader,
	)),
	PhysicalType::INT32 => TripletIter::Int32TripletIter(TypedTripletIter::new(
	descr, batch_size, reader,
	)),
	PhysicalType::INT64 => TripletIter::Int64TripletIter(TypedTripletIter::new(
	descr, batch_size, reader,
	)),
	PhysicalType::INT96 => TripletIter::Int96TripletIter(TypedTripletIter::new(
	descr, batch_size, reader,
	)),
	PhysicalType::FLOAT => TripletIter::FloatTripletIter(TypedTripletIter::new(
	descr, batch_size, reader,
	)),
	PhysicalType::DOUBLE => TripletIter::DoubleTripletIter(
	TypedTripletIter::new(descr, batch_size, reader),
	),
	PhysicalType::BYTE_ARRAY => TripletIter::ByteArrayTripletIter(
	TypedTripletIter::new(descr, batch_size, reader),
	),
	PhysicalType::FIXED_LEN_BYTE_ARRAY => {
	TripletIter::FixedLenByteArrayTripletIter(TypedTripletIter::new(
	descr, batch_size, reader,
	))
	}
	}
	}

	/// Invokes underlying typed triplet iterator to buffer current value.
	/// Should be called once - either before `is_null` or `current_value`.
	#[inline]
	pub fn read_next(&mut self) -> Result<bool> {
	triplet_enum_func!(self, read_next, ref, mut)
	}

	/// Provides check on values/levels left without invoking the underlying typed triplet
	/// iterator.
	/// Returns true if more values/levels exist, false otherwise.
	/// It is always in sync with `read_next` method.
	#[inline]
	pub fn has_next(&self) -> bool {
	triplet_enum_func!(self, has_next, ref)
	}

	/// Returns current definition level for a leaf triplet iterator
	#[inline]
	pub fn current_def_level(&self) -> i16 {
	triplet_enum_func!(self, current_def_level, ref)
	}

	/// Returns max definition level for a leaf triplet iterator
	#[inline]
	pub fn max_def_level(&self) -> i16 {
	triplet_enum_func!(self, max_def_level, ref)
	}

	/// Returns current repetition level for a leaf triplet iterator
	#[inline]
	pub fn current_rep_level(&self) -> i16 {
	triplet_enum_func!(self, current_rep_level, ref)
	}

	/// Returns max repetition level for a leaf triplet iterator
	#[inline]
	pub fn max_rep_level(&self) -> i16 {
	triplet_enum_func!(self, max_rep_level, ref)
	}

	/// Returns true, if current value is null.
	/// Based on the fact that for non-null value current definition level
	/// equals to max definition level.
	#[inline]
	pub fn is_null(&self) -> bool {
	self.current_def_level() < self.max_def_level()
	}

	/// Updates non-null value for current row.
	pub fn current_value(&self) -> Field {
	assert!(!self.is_null(), "Value is null");
	match *self {
	TripletIter::BoolTripletIter(ref typed) => {
	Field::convert_bool(typed.column_descr(), *typed.current_value())
	}
	TripletIter::Int32TripletIter(ref typed) => {
	Field::convert_int32(typed.column_descr(), *typed.current_value())
	}
	TripletIter::Int64TripletIter(ref typed) => {
	Field::convert_int64(typed.column_descr(), *typed.current_value())
	}
	TripletIter::Int96TripletIter(ref typed) => {
	Field::convert_int96(typed.column_descr(), typed.current_value().clone())
	}
	TripletIter::FloatTripletIter(ref typed) => {
	Field::convert_float(typed.column_descr(), *typed.current_value())
	}
	TripletIter::DoubleTripletIter(ref typed) => {
	Field::convert_double(typed.column_descr(), *typed.current_value())
	}
	TripletIter::ByteArrayTripletIter(ref typed) => Field::convert_byte_array(
	typed.column_descr(),
	typed.current_value().clone(),
	),
	TripletIter::FixedLenByteArrayTripletIter(ref typed) => {
	Field::convert_byte_array(
	typed.column_descr(),
	typed.current_value().clone().into(),
	)
	}
	}
	}
	}

	/// Internal typed triplet iterator as a wrapper for column reader
	/// (primitive leaf column), provides per-element access.
	pub struct TypedTripletIter<T: DataType> {
	reader: ColumnReaderImpl<T>,
	column_descr: ColumnDescPtr,
	batch_size: usize,
	// type properties
	max_def_level: i16,
	max_rep_level: i16,
	// values and levels
	values: Vec<T::T>,
	def_levels: Option<Vec<i16>>,
	rep_levels: Option<Vec<i16>>,
	// current index for the triplet (value, def, rep)
	curr_triplet_index: usize,
	// how many triplets are left before we need to buffer
	triplets_left: usize,
	// helper flag to quickly check if we have more values/levels to read
	has_next: bool,
	}

	impl<T: DataType> TypedTripletIter<T> {
	/// Creates new typed triplet iterator based on provided column reader.
	/// Use batch size to specify the amount of values to buffer from column reader.
	fn new(descr: ColumnDescPtr, batch_size: usize, column_reader: ColumnReader) -> Self {
	assert!(
	batch_size > 0,
	"Expected positive batch size, found: {}",
	batch_size
	);

	let max_def_level = descr.max_def_level();
	let max_rep_level = descr.max_rep_level();

	let def_levels = if max_def_level == 0 {
	None
	} else {
	Some(vec![0; batch_size])
	};
	let rep_levels = if max_rep_level == 0 {
	None
	} else {
	Some(vec![0; batch_size])
	};

	Self {
	reader: get_typed_column_reader(column_reader),
	column_descr: descr,
	batch_size,
	max_def_level,
	max_rep_level,
	values: vec![T::T::default(); batch_size],
	def_levels,
	rep_levels,
	curr_triplet_index: 0,
	triplets_left: 0,
	has_next: false,
	}
	}

	/// Returns column descriptor reference for the current typed triplet iterator.
	#[inline]
	pub fn column_descr(&self) -> &ColumnDescPtr {
	&self.column_descr
	}

	/// Returns maximum definition level for the triplet iterator (leaf column).
	#[inline]
	fn max_def_level(&self) -> i16 {
	self.max_def_level
	}

	/// Returns maximum repetition level for the triplet iterator (leaf column).
	#[inline]
	fn max_rep_level(&self) -> i16 {
	self.max_rep_level
	}

	/// Returns current value.
	/// Method does not advance the iterator, therefore can be called multiple times.
	#[inline]
	fn current_value(&self) -> &T::T {
	assert!(
	self.current_def_level() == self.max_def_level(),
	"Cannot extract value, max definition level: {}, current level: {}",
	self.max_def_level(),
	self.current_def_level()
	);
	&self.values[self.curr_triplet_index]
	}

	/// Returns current definition level.
	/// If field is required, then maximum definition level is returned.
	#[inline]
	fn current_def_level(&self) -> i16 {
	match self.def_levels {
	Some(ref vec) => vec[self.curr_triplet_index],
	None => self.max_def_level,
	}
	}

	/// Returns current repetition level.
	/// If field is required, then maximum repetition level is returned.
	#[inline]
	fn current_rep_level(&self) -> i16 {
	match self.rep_levels {
	Some(ref vec) => vec[self.curr_triplet_index],
	None => self.max_rep_level,
	}
	}

	/// Quick check if iterator has more values/levels to read.
	/// It is updated as a result of `read_next` method, so they are synchronized.
	#[inline]
	fn has_next(&self) -> bool {
	self.has_next
	}

	/// Advances to the next triplet.
	/// Returns true, if there are more records to read, false there are no records left.
	fn read_next(&mut self) -> Result<bool> {
	self.curr_triplet_index += 1;

	if self.curr_triplet_index >= self.triplets_left {
	let (values_read, levels_read) = {
	// Get slice of definition levels, if available
	let def_levels = self.def_levels.as_mut().map(\|vec\| &mut vec[..]);

	// Get slice of repetition levels, if available
	let rep_levels = self.rep_levels.as_mut().map(\|vec\| &mut vec[..]);

	// Buffer triplets
	self.reader.read_batch(
	self.batch_size,
	def_levels,
	rep_levels,
	&mut self.values,
	)?
	};

	// No more values or levels to read
	if values_read == 0 && levels_read == 0 {
	self.has_next = false;
	return Ok(false);
	}

	// We never read values more than levels
	if levels_read == 0 \|\| values_read == levels_read {
	// There are no definition levels to read, column is required
	// or definition levels match values, so it does not require spacing
	self.curr_triplet_index = 0;
	self.triplets_left = values_read;
	} else if values_read < levels_read {
	// Add spacing for triplets.
	// The idea is setting values for positions in def_levels when current
	// definition level equals to maximum definition level.
	// Values and levels are guaranteed to line up, because of
	// the column reader method.

	// Note: if values_read == 0, then spacing will not be triggered
	let mut idx = values_read;
	let def_levels = self.def_levels.as_ref().unwrap();
	for i in 0..levels_read {
	if def_levels[levels_read - i - 1] == self.max_def_level {
	idx -= 1; // This is done to avoid usize becoming a negative value
	self.values.swap(levels_read - i - 1, idx);
	}
	}
	self.curr_triplet_index = 0;
	self.triplets_left = levels_read;
	} else {
	return Err(general_err!(
	"Spacing of values/levels is wrong, values_read: {}, levels_read: {}",
	values_read,
	levels_read
	));
	}
	}

	self.has_next = true;
	Ok(true)
	}
	}

	#[cfg(test)]
	mod tests {
	use super::*;

	use crate::file::reader::{FileReader, SerializedFileReader};
	use crate::schema::types::ColumnPath;
	use crate::util::test_common::get_test_file;

	#[test]
	#[should_panic(expected = "Expected positive batch size, found: 0")]
	fn test_triplet_zero_batch_size() {
	let column_path =
	ColumnPath::from(vec!["b_struct".to_string(), "b_c_int".to_string()]);
	test_column_in_file("nulls.snappy.parquet", 0, &column_path, &[], &[], &[]);
	}

	#[test]
	fn test_triplet_null_column() {
	let path = vec!["b_struct", "b_c_int"];
	let values = vec![];
	let def_levels = vec![1, 1, 1, 1, 1, 1, 1, 1];
	let rep_levels = vec![0, 0, 0, 0, 0, 0, 0, 0];
	test_triplet_iter(
	"nulls.snappy.parquet",
	path,
	&values,
	&def_levels,
	&rep_levels,
	);
	}

	#[test]
	fn test_triplet_required_column() {
	let path = vec!["ID"];
	let values = vec![Field::Long(8)];
	let def_levels = vec![0];
	let rep_levels = vec![0];
	test_triplet_iter(
	"nonnullable.impala.parquet",
	path,
	&values,
	&def_levels,
	&rep_levels,
	);
	}

	#[test]
	fn test_triplet_optional_column() {
	let path = vec!["nested_struct", "A"];
	let values = vec![Field::Int(1), Field::Int(7)];
	let def_levels = vec![2, 1, 1, 1, 1, 0, 2];
	let rep_levels = vec![0, 0, 0, 0, 0, 0, 0];
	test_triplet_iter(
	"nullable.impala.parquet",
	path,
	&values,
	&def_levels,
	&rep_levels,
	);
	}

	#[test]
	fn test_triplet_optional_list_column() {
	let path = vec!["a", "list", "element", "list", "element", "list", "element"];
	let values = vec![
	Field::Str("a".to_string()),
	Field::Str("b".to_string()),
	Field::Str("c".to_string()),
	Field::Str("d".to_string()),
	Field::Str("a".to_string()),
	Field::Str("b".to_string()),
	Field::Str("c".to_string()),
	Field::Str("d".to_string()),
	Field::Str("e".to_string()),
	Field::Str("a".to_string()),
	Field::Str("b".to_string()),
	Field::Str("c".to_string()),
	Field::Str("d".to_string()),
	Field::Str("e".to_string()),
	Field::Str("f".to_string()),
	];
	let def_levels = vec![7, 7, 7, 4, 7, 7, 7, 7, 7, 4, 7, 7, 7, 7, 7, 7, 4, 7];
	let rep_levels = vec![0, 3, 2, 1, 2, 0, 3, 2, 3, 1, 2, 0, 3, 2, 3, 2, 1, 2];
	test_triplet_iter(
	"nested_lists.snappy.parquet",
	path,
	&values,
	&def_levels,
	&rep_levels,
	);
	}

	#[test]
	fn test_triplet_optional_map_column() {
	let path = vec!["a", "key_value", "value", "key_value", "key"];
	let values = vec![
	Field::Int(1),
	Field::Int(2),
	Field::Int(1),
	Field::Int(1),
	Field::Int(3),
	Field::Int(4),
	Field::Int(5),
	];
	let def_levels = vec![4, 4, 4, 2, 3, 4, 4, 4, 4];
	let rep_levels = vec![0, 2, 0, 0, 0, 0, 0, 2, 2];
	test_triplet_iter(
	"nested_maps.snappy.parquet",
	path,
	&values,
	&def_levels,
	&rep_levels,
	);
	}

	// Check triplet iterator across different batch sizes
	fn test_triplet_iter(
	file_name: &str,
	column_path: Vec<&str>,
	expected_values: &[Field],
	expected_def_levels: &[i16],
	expected_rep_levels: &[i16],
	) {
	// Convert path into column path
	let path: Vec<String> = column_path.iter().map(\|x\| x.to_string()).collect();
	let column_path = ColumnPath::from(path);

	let batch_sizes = vec![1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 128, 256];
	for batch_size in batch_sizes {
	test_column_in_file(
	file_name,
	batch_size,
	&column_path,
	expected_values,
	expected_def_levels,
	expected_rep_levels,
	);
	}
	}

	// Check values of a selectd column in a file
	fn test_column_in_file(
	file_name: &str,
	batch_size: usize,
	column_path: &ColumnPath,
	expected_values: &[Field],
	expected_def_levels: &[i16],
	expected_rep_levels: &[i16],
	) {
	let file = get_test_file(file_name);
	let file_reader = SerializedFileReader::new(file).unwrap();
	let metadata = file_reader.metadata();
	// Get schema descriptor
	let file_metadata = metadata.file_metadata();
	let schema = file_metadata.schema_descr();
	// Get first row group
	let row_group_reader = file_reader.get_row_group(0).unwrap();

	for i in 0..schema.num_columns() {
	let descr = schema.column(i);
	if descr.path() == column_path {
	let reader = row_group_reader.get_column_reader(i).unwrap();
	test_triplet_column(
	descr,
	reader,
	batch_size,
	expected_values,
	expected_def_levels,
	expected_rep_levels,
	);
	}
	}
	}

	// Check values for individual triplet iterator
	fn test_triplet_column(
	descr: ColumnDescPtr,
	reader: ColumnReader,
	batch_size: usize,
	expected_values: &[Field],
	expected_def_levels: &[i16],
	expected_rep_levels: &[i16],
	) {
	let mut iter = TripletIter::new(descr.clone(), reader, batch_size);
	let mut values: Vec<Field> = Vec::new();
	let mut def_levels: Vec<i16> = Vec::new();
	let mut rep_levels: Vec<i16> = Vec::new();

	assert_eq!(iter.max_def_level(), descr.max_def_level());
	assert_eq!(iter.max_rep_level(), descr.max_rep_level());

	while let Ok(true) = iter.read_next() {
	assert!(iter.has_next());
	if !iter.is_null() {
	values.push(iter.current_value());
	}
	def_levels.push(iter.current_def_level());
	rep_levels.push(iter.current_rep_level());
	}

	assert_eq!(values, expected_values);
	assert_eq!(def_levels, expected_def_levels);
	assert_eq!(rep_levels, expected_rep_levels);
	}
	}