Integrate fixes for Timestamp[MICROS] and infinite loop hang when reading parquet

parquet/src/arrow/array_reader.rs

@@ -15,6 +15,7 @@
 // specific language governing permissions and limitations
 // under the License.
 
+use parquet_format::TimeUnit;
 use std::any::Any;
 use std::cmp::{max, min};
 use std::marker::PhantomData;
@@ -33,16 +34,16 @@ use arrow::compute::take;
 use arrow::datatypes::{
     ArrowPrimitiveType, BooleanType as ArrowBooleanType, DataType as ArrowType,
     Float32Type as ArrowFloat32Type, Float64Type as ArrowFloat64Type,
-    Int32Type as ArrowInt32Type, Int64Type as ArrowInt64Type, ToByteSlice,
-    UInt32Type as ArrowUInt32Type, UInt64Type as ArrowUInt64Type,
+    Int32Type as ArrowInt32Type, Int64Type as ArrowInt64Type, TimeUnit as ArrowTimeUnit,
+    ToByteSlice, UInt32Type as ArrowUInt32Type, UInt64Type as ArrowUInt64Type,
 };
 use arrow::util::bit_util;
 
 use crate::arrow::converter::Converter;
 use crate::arrow::record_reader::buffer::{ScalarValue, ValuesBuffer};
 use crate::arrow::record_reader::{GenericRecordReader, RecordReader};
 use crate::arrow::schema::parquet_to_arrow_field;
-use crate::basic::Type as PhysicalType;
+use crate::basic::{LogicalType, Type as PhysicalType};
 use crate::column::page::PageIterator;
 use crate::column::reader::decoder::ColumnValueDecoder;
 use crate::column::reader::ColumnReaderImpl;
@@ -430,6 +431,49 @@ where
 
                 Arc::new(array) as ArrayRef
             }
+            ArrowType::Timestamp(ArrowTimeUnit::Nanosecond, ref tz) => {
+                if let Some(LogicalType::TIMESTAMP(t)) = self.column_desc.logical_type() {
+                    match t.unit {
+                        TimeUnit::MICROS(_) => {
+                            let a = arrow::compute::cast(
+                                &array,
+                                &ArrowType::Timestamp(
+                                    ArrowTimeUnit::Microsecond,
+                                    tz.clone(),
+                                ),
+                            )?;
+                            arrow::compute::cast(
+                                &a,
+                                &ArrowType::Timestamp(
+                                    ArrowTimeUnit::Nanosecond,
+                                    tz.clone(),
+                                ),
+                            )?
+                        }
+                        TimeUnit::MILLIS(_) => {
+                            let a = arrow::compute::cast(
+                                &array,
+                                &ArrowType::Timestamp(
+                                    ArrowTimeUnit::Millisecond,
+                                    tz.clone(),
+                                ),
+                            )?;
+                            arrow::compute::cast(
+                                &a,
+                                &ArrowType::Timestamp(
+                                    ArrowTimeUnit::Nanosecond,
+                                    tz.clone(),
+                                ),
+                            )?
+                        }
+                        TimeUnit::NANOS(_) => {
+                            arrow::compute::cast(&array, &target_type.clone())?
+                        }
+                    }
+                } else {
+                    arrow::compute::cast(&array, &target_type.clone())?
+                }
+            }
             _ => arrow::compute::cast(&array, &target_type)?,
         };
 

parquet/src/encodings/rle.rs

@@ -419,6 +419,13 @@ impl RleDecoder {
                     &mut buffer[values_read..values_read + num_values],
                     self.bit_width as usize,
                 );
+                if num_values == 0 {
+                    // some writers are putting less values that could fit in the bit packed run
+                    // in such a case bit_packed_left is unreliable as it represents max possible values left
+                    // to avoid infinite loop we need to return so the caller could check if all expected values were read already.
+                    self.bit_packed_left = 0;
+                    continue;
+                }
                 self.bit_packed_left -= num_values as u32;
                 values_read += num_values;
             } else if !self.reload() {