You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@arrow.apache.org by we...@apache.org on 2019/06/25 17:46:06 UTC
[arrow] branch master updated: ARROW-4139: [Python][Parquet] Wrap
new parquet::LogicalType, cast min/max statistics based on LogicalType
This is an automated email from the ASF dual-hosted git repository.
wesm pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/arrow.git
The following commit(s) were added to refs/heads/master by this push:
new 74841f5 ARROW-4139: [Python][Parquet] Wrap new parquet::LogicalType, cast min/max statistics based on LogicalType
74841f5 is described below
commit 74841f5bfcfdeab7d90fda9e469a92aef02a23f3
Author: Wes McKinney <we...@apache.org>
AuthorDate: Tue Jun 25 12:45:54 2019 -0500
ARROW-4139: [Python][Parquet] Wrap new parquet::LogicalType, cast min/max statistics based on LogicalType
This changes the behavior of `Statistics.min` and `Statistics.max` to return the decoded logical value, such as a UTF-8 string, uint64 value, timestamp, etc. I added `min_raw` and `max_raw` attributes to return the raw physical type value
Also fixes ARROW-5166 (uint64 as int64 overflow)
I found the `isAdjustedToUTC` parameter for `TIME` logical type to be weird, so I'm changing the Arrow conversion layer to always set this so we can get the TIME_MILLIS / TIME_MICROS converted type to be set in those cases.
Author: Wes McKinney <we...@apache.org>
Closes #4680 from wesm/ARROW-4139 and squashes the following commits:
11ced9e6c <Wes McKinney> Add utf-8 coding to test_parquet.py
54f974a53 <Wes McKinney> Follow changes in parquet::arrow unit tests
283b69aba <Wes McKinney> Write version 2.0 files
a67746e44 <Wes McKinney> Wrap new parquet::LogicalType, cast min/max statistics based on LogicalType in Python API
---
cpp/src/parquet/arrow/arrow-schema-test.cc | 6 +-
cpp/src/parquet/arrow/schema.cc | 9 +-
cpp/src/parquet/statistics.cc | 2 +
cpp/src/parquet/statistics.h | 3 +
python/pyarrow/_parquet.pxd | 91 +++++++++-
python/pyarrow/_parquet.pyx | 282 ++++++++++++++++++++++-------
python/pyarrow/parquet.py | 2 +-
python/pyarrow/scalar.pxi | 19 +-
python/pyarrow/tests/test_parquet.py | 45 ++++-
9 files changed, 375 insertions(+), 84 deletions(-)
diff --git a/cpp/src/parquet/arrow/arrow-schema-test.cc b/cpp/src/parquet/arrow/arrow-schema-test.cc
index 1089fbb..9a43c7d 100644
--- a/cpp/src/parquet/arrow/arrow-schema-test.cc
+++ b/cpp/src/parquet/arrow/arrow-schema-test.cc
@@ -849,11 +849,11 @@ TEST_F(TestConvertArrowSchema, ArrowFields) {
{"decimal(32, 8)", ::arrow::decimal(32, 8), LogicalType::Decimal(32, 8),
ParquetType::FIXED_LEN_BYTE_ARRAY, 14},
{"time32", ::arrow::time32(::arrow::TimeUnit::MILLI),
- LogicalType::Time(false, LogicalType::TimeUnit::MILLIS), ParquetType::INT32, -1},
+ LogicalType::Time(true, LogicalType::TimeUnit::MILLIS), ParquetType::INT32, -1},
{"time64(microsecond)", ::arrow::time64(::arrow::TimeUnit::MICRO),
- LogicalType::Time(false, LogicalType::TimeUnit::MICROS), ParquetType::INT64, -1},
+ LogicalType::Time(true, LogicalType::TimeUnit::MICROS), ParquetType::INT64, -1},
{"time64(nanosecond)", ::arrow::time64(::arrow::TimeUnit::NANO),
- LogicalType::Time(false, LogicalType::TimeUnit::NANOS), ParquetType::INT64, -1},
+ LogicalType::Time(true, LogicalType::TimeUnit::NANOS), ParquetType::INT64, -1},
{"timestamp(millisecond)", ::arrow::timestamp(::arrow::TimeUnit::MILLI),
LogicalType::Timestamp(false, LogicalType::TimeUnit::MILLIS), ParquetType::INT64,
-1},
diff --git a/cpp/src/parquet/arrow/schema.cc b/cpp/src/parquet/arrow/schema.cc
index 04587a9..9284c8c 100644
--- a/cpp/src/parquet/arrow/schema.cc
+++ b/cpp/src/parquet/arrow/schema.cc
@@ -703,15 +703,18 @@ Status FieldToNode(const std::shared_ptr<Field>& field,
break;
case ArrowTypeId::TIME32:
type = ParquetType::INT32;
- logical_type = LogicalType::Time(false, LogicalType::TimeUnit::MILLIS);
+ logical_type =
+ LogicalType::Time(/*is_adjusted_to_utc=*/true, LogicalType::TimeUnit::MILLIS);
break;
case ArrowTypeId::TIME64: {
type = ParquetType::INT64;
auto time_type = static_cast<::arrow::Time64Type*>(field->type().get());
if (time_type->unit() == ::arrow::TimeUnit::NANO) {
- logical_type = LogicalType::Time(false, LogicalType::TimeUnit::NANOS);
+ logical_type =
+ LogicalType::Time(/*is_adjusted_to_utc=*/true, LogicalType::TimeUnit::NANOS);
} else {
- logical_type = LogicalType::Time(false, LogicalType::TimeUnit::MICROS);
+ logical_type =
+ LogicalType::Time(/*is_adjusted_to_utc=*/true, LogicalType::TimeUnit::MICROS);
}
} break;
case ArrowTypeId::STRUCT: {
diff --git a/cpp/src/parquet/statistics.cc b/cpp/src/parquet/statistics.cc
index e8b872b..54cad54 100644
--- a/cpp/src/parquet/statistics.cc
+++ b/cpp/src/parquet/statistics.cc
@@ -310,6 +310,8 @@ class TypedStatisticsImpl : public TypedStatistics<DType> {
Type::type physical_type() const override { return descr_->physical_type(); }
+ const ColumnDescriptor* descr() const override { return descr_; }
+
std::string EncodeMin() override {
std::string s;
if (HasMinMax()) this->PlainEncode(min_, &s);
diff --git a/cpp/src/parquet/statistics.h b/cpp/src/parquet/statistics.h
index 2dc78da..7698224 100644
--- a/cpp/src/parquet/statistics.h
+++ b/cpp/src/parquet/statistics.h
@@ -225,6 +225,9 @@ class PARQUET_EXPORT Statistics {
/// \brief The physical type of the column schema
virtual Type::type physical_type() const = 0;
+ /// \brief The full type descriptor from the column schema
+ virtual const ColumnDescriptor* descr() const = 0;
+
protected:
static std::shared_ptr<Statistics> Make(Type::type physical_type, const void* min,
const void* max, int64_t num_values,
diff --git a/python/pyarrow/_parquet.pxd b/python/pyarrow/_parquet.pxd
index c34c89e..6a11fc8 100644
--- a/python/pyarrow/_parquet.pxd
+++ b/python/pyarrow/_parquet.pxd
@@ -54,6 +54,28 @@ cdef extern from "parquet/api/schema.h" namespace "parquet" nogil:
ParquetType_BYTE_ARRAY" parquet::Type::BYTE_ARRAY"
ParquetType_FIXED_LEN_BYTE_ARRAY" parquet::Type::FIXED_LEN_BYTE_ARRAY"
+ enum ParquetLogicalTypeId" parquet::LogicalType::Type::type":
+ ParquetLogicalType_UNKNOWN" parquet::LogicalType::Type::UNKNOWN"
+ ParquetLogicalType_STRING" parquet::LogicalType::Type::STRING"
+ ParquetLogicalType_MAP" parquet::LogicalType::Type::MAP"
+ ParquetLogicalType_LIST" parquet::LogicalType::Type::LIST"
+ ParquetLogicalType_ENUM" parquet::LogicalType::Type::ENUM"
+ ParquetLogicalType_DECIMAL" parquet::LogicalType::Type::DECIMAL"
+ ParquetLogicalType_DATE" parquet::LogicalType::Type::DATE"
+ ParquetLogicalType_TIME" parquet::LogicalType::Type::TIME"
+ ParquetLogicalType_TIMESTAMP" parquet::LogicalType::Type::TIMESTAMP"
+ ParquetLogicalType_INT" parquet::LogicalType::Type::INT"
+ ParquetLogicalType_JSON" parquet::LogicalType::Type::JSON"
+ ParquetLogicalType_BSON" parquet::LogicalType::Type::BSON"
+ ParquetLogicalType_UUID" parquet::LogicalType::Type::UUID"
+ ParquetLogicalType_NONE" parquet::LogicalType::Type::NONE"
+
+ enum ParquetTimeUnit" parquet::LogicalType::TimeUnit::unit":
+ ParquetTimeUnit_UNKNOWN" parquet::LogicalType::TimeUnit::UNKNOWN"
+ ParquetTimeUnit_MILLIS" parquet::LogicalType::TimeUnit::MILLIS"
+ ParquetTimeUnit_MICROS" parquet::LogicalType::TimeUnit::MICROS"
+ ParquetTimeUnit_NANOS" parquet::LogicalType::TimeUnit::NANOS"
+
enum ParquetConvertedType" parquet::ConvertedType::type":
ParquetConvertedType_NONE" parquet::ConvertedType::NONE"
ParquetConvertedType_UTF8" parquet::ConvertedType::UTF8"
@@ -117,7 +139,32 @@ cdef extern from "parquet/api/schema.h" namespace "parquet" nogil:
ParquetSortOrder_UNSIGNED" parquet::SortOrder::UNSIGNED"
ParquetSortOrder_UNKNOWN" parquet::SortOrder::UNKNOWN"
- cdef cppclass ColumnDescriptor:
+ cdef cppclass CParquetLogicalType" parquet::LogicalType":
+ c_string ToString() const
+ c_string ToJSON() const
+ ParquetLogicalTypeId type() const
+
+ cdef cppclass CParquetDecimalType \
+ " parquet::DecimalLogicalType"(CParquetLogicalType):
+ int32_t precision() const
+ int32_t scale() const
+
+ cdef cppclass CParquetIntType \
+ " parquet::IntLogicalType"(CParquetLogicalType):
+ int bit_width() const
+ c_bool is_signed() const
+
+ cdef cppclass CParquetTimeType \
+ " parquet::TimeLogicalType"(CParquetLogicalType):
+ c_bool is_adjusted_to_utc() const
+ ParquetTimeUnit time_unit() const
+
+ cdef cppclass CParquetTimestampType \
+ " parquet::TimestampLogicalType"(CParquetLogicalType):
+ c_bool is_adjusted_to_utc() const
+ ParquetTimeUnit time_unit() const
+
+ cdef cppclass ColumnDescriptor" parquet::ColumnDescriptor":
c_bool Equals(const ColumnDescriptor& other)
shared_ptr[ColumnPath] path()
@@ -125,6 +172,7 @@ cdef extern from "parquet/api/schema.h" namespace "parquet" nogil:
int16_t max_repetition_level()
ParquetType physical_type()
+ const shared_ptr[const CParquetLogicalType]& logical_type()
ParquetConvertedType converted_type()
const c_string& name()
int type_length()
@@ -179,7 +227,14 @@ cdef extern from "parquet/api/reader.h" namespace "parquet" nogil:
bint has_null_count
bint has_distinct_count
- cdef cppclass CRowGroupStatistics" parquet::RowGroupStatistics":
+ cdef cppclass ParquetByteArray" parquet::ByteArray":
+ uint32_t len
+ const uint8_t* ptr
+
+ cdef cppclass ParquetFLBA" parquet::FLBA":
+ const uint8_t* ptr
+
+ cdef cppclass CStatistics" parquet::Statistics":
int64_t null_count() const
int64_t distinct_count() const
int64_t num_values() const
@@ -190,6 +245,36 @@ cdef extern from "parquet/api/reader.h" namespace "parquet" nogil:
CEncodedStatistics Encode()
void SetComparator()
ParquetType physical_type() const
+ const ColumnDescriptor* descr() const
+
+ cdef cppclass CBoolStatistics" parquet::BoolStatistics"(CStatistics):
+ c_bool min()
+ c_bool max()
+
+ cdef cppclass CInt32Statistics" parquet::Int32Statistics"(CStatistics):
+ int32_t min()
+ int32_t max()
+
+ cdef cppclass CInt64Statistics" parquet::Int64Statistics"(CStatistics):
+ int64_t min()
+ int64_t max()
+
+ cdef cppclass CFloatStatistics" parquet::FloatStatistics"(CStatistics):
+ float min()
+ float max()
+
+ cdef cppclass CDoubleStatistics" parquet::DoubleStatistics"(CStatistics):
+ double min()
+ double max()
+
+ cdef cppclass CByteArrayStatistics \
+ " parquet::ByteArrayStatistics"(CStatistics):
+ ParquetByteArray min()
+ ParquetByteArray max()
+
+ cdef cppclass CFLBAStatistics" parquet::FLBAStatistics"(CStatistics):
+ ParquetFLBA min()
+ ParquetFLBA max()
cdef cppclass CColumnChunkMetaData" parquet::ColumnChunkMetaData":
int64_t file_offset() const
@@ -199,7 +284,7 @@ cdef extern from "parquet/api/reader.h" namespace "parquet" nogil:
int64_t num_values() const
shared_ptr[ColumnPath] path_in_schema() const
bint is_stats_set() const
- shared_ptr[CRowGroupStatistics] statistics() const
+ shared_ptr[CStatistics] statistics() const
ParquetCompression compression() const
const vector[ParquetEncoding]& encodings() const
diff --git a/python/pyarrow/_parquet.pyx b/python/pyarrow/_parquet.pyx
index 6cfc104..d9861f2 100644
--- a/python/pyarrow/_parquet.pyx
+++ b/python/pyarrow/_parquet.pyx
@@ -25,6 +25,8 @@ import io
import six
import warnings
+import numpy as np
+
from cython.operator cimport dereference as deref
from pyarrow.includes.common cimport *
from pyarrow.includes.libarrow cimport *
@@ -40,36 +42,45 @@ from pyarrow.lib cimport (Buffer, Array, Schema,
from pyarrow.compat import tobytes, frombytes
from pyarrow.lib import (ArrowException, NativeFile, _stringify_path,
- BufferOutputStream)
+ BufferOutputStream,
+ _datetime_conversion_functions,
+ _box_time_milli,
+ _box_time_micro)
from pyarrow.util import indent
+cimport cpython as cp
+
-cdef class RowGroupStatistics:
+cdef class Statistics:
cdef:
- shared_ptr[CRowGroupStatistics] statistics
+ shared_ptr[CStatistics] statistics
def __cinit__(self):
pass
- cdef init(self, const shared_ptr[CRowGroupStatistics]& statistics):
+ cdef init(self, const shared_ptr[CStatistics]& statistics):
self.statistics = statistics
def __repr__(self):
- return """{0}
- has_min_max: {1}
- min: {2}
- max: {3}
- null_count: {4}
- distinct_count: {5}
- num_values: {6}
- physical_type: {7}""".format(object.__repr__(self),
- self.has_min_max,
- self.min,
- self.max,
- self.null_count,
- self.distinct_count,
- self.num_values,
- self.physical_type)
+ return """{}
+ has_min_max: {}
+ min: {}
+ max: {}
+ null_count: {}
+ distinct_count: {}
+ num_values: {}
+ physical_type: {}
+ logical_type: {}
+ converted_type (legacy): {}""".format(object.__repr__(self),
+ self.has_min_max,
+ self.min,
+ self.max,
+ self.null_count,
+ self.distinct_count,
+ self.num_values,
+ self.physical_type,
+ str(self.logical_type),
+ self.converted_type)
def to_dict(self):
d = dict(
@@ -89,8 +100,8 @@ cdef class RowGroupStatistics:
except TypeError:
return NotImplemented
- def equals(self, RowGroupStatistics other):
- # TODO(kszucs): implement native Equals method for RowGroupStatistics
+ def equals(self, Statistics other):
+ # TODO(kszucs): implement native Equals method for Statistics
return (self.has_min_max == other.has_min_max and
self.min == other.min and
self.max == other.max and
@@ -99,50 +110,25 @@ cdef class RowGroupStatistics:
self.num_values == other.num_values and
self.physical_type == other.physical_type)
- cdef inline _cast_statistic(self, object value):
- # Input value is bytes
- cdef ParquetType physical_type = self.statistics.get().physical_type()
- if physical_type == ParquetType_BOOLEAN:
- return bool(int(value))
- elif physical_type == ParquetType_INT32:
- return int(value)
- elif physical_type == ParquetType_INT64:
- return int(value)
- elif physical_type == ParquetType_INT96:
- # Leave as PyBytes
- return value
- elif physical_type == ParquetType_FLOAT:
- return float(value)
- elif physical_type == ParquetType_DOUBLE:
- return float(value)
- elif physical_type == ParquetType_BYTE_ARRAY:
- # Leave as PyBytes
- return value
- elif physical_type == ParquetType_FIXED_LEN_BYTE_ARRAY:
- # Leave as PyBytes
- return value
- else:
- raise ValueError('Unknown physical ParquetType')
-
@property
def has_min_max(self):
return self.statistics.get().HasMinMax()
@property
- def min(self):
- raw_physical_type = self.statistics.get().physical_type()
- encode_min = self.statistics.get().EncodeMin()
+ def min_raw(self):
+ return _cast_statistic_raw_min(self.statistics.get())
- min_value = FormatStatValue(raw_physical_type, encode_min)
- return self._cast_statistic(min_value)
+ @property
+ def max_raw(self):
+ return _cast_statistic_raw_max(self.statistics.get())
@property
- def max(self):
- raw_physical_type = self.statistics.get().physical_type()
- encode_max = self.statistics.get().EncodeMax()
+ def min(self):
+ return _cast_statistic_min(self.statistics.get())
- max_value = FormatStatValue(raw_physical_type, encode_max)
- return self._cast_statistic(max_value)
+ @property
+ def max(self):
+ return _cast_statistic_max(self.statistics.get())
@property
def null_count(self):
@@ -161,6 +147,145 @@ cdef class RowGroupStatistics:
raw_physical_type = self.statistics.get().physical_type()
return physical_type_name_from_enum(raw_physical_type)
+ @property
+ def logical_type(self):
+ return wrap_logical_type(self.statistics.get().descr().logical_type())
+
+ @property
+ def converted_type(self):
+ raw_converted_type = self.statistics.get().descr().converted_type()
+ return converted_type_name_from_enum(raw_converted_type)
+
+
+cdef class ParquetLogicalType:
+ cdef:
+ shared_ptr[const CParquetLogicalType] type
+
+ def __cinit__(self):
+ pass
+
+ cdef init(self, const shared_ptr[const CParquetLogicalType]& type):
+ self.type = type
+
+ def __str__(self):
+ return frombytes(self.type.get().ToString())
+
+ def to_json(self):
+ return frombytes(self.type.get().ToJSON())
+
+ @property
+ def type(self):
+ return logical_type_name_from_enum(self.type.get().type())
+
+
+cdef wrap_logical_type(const shared_ptr[const CParquetLogicalType]& type):
+ cdef ParquetLogicalType out = ParquetLogicalType()
+ out.init(type)
+ return out
+
+
+cdef _cast_statistic_raw_min(CStatistics* statistics):
+ cdef ParquetType physical_type = statistics.physical_type()
+ cdef uint32_t type_length = statistics.descr().type_length()
+ if physical_type == ParquetType_BOOLEAN:
+ return (<CBoolStatistics*> statistics).min()
+ elif physical_type == ParquetType_INT32:
+ return (<CInt32Statistics*> statistics).min()
+ elif physical_type == ParquetType_INT64:
+ return (<CInt64Statistics*> statistics).min()
+ elif physical_type == ParquetType_FLOAT:
+ return (<CFloatStatistics*> statistics).min()
+ elif physical_type == ParquetType_DOUBLE:
+ return (<CDoubleStatistics*> statistics).min()
+ elif physical_type == ParquetType_BYTE_ARRAY:
+ return _box_byte_array((<CByteArrayStatistics*> statistics).min())
+ elif physical_type == ParquetType_FIXED_LEN_BYTE_ARRAY:
+ return _box_flba((<CFLBAStatistics*> statistics).min(), type_length)
+
+
+cdef _cast_statistic_raw_max(CStatistics* statistics):
+ cdef ParquetType physical_type = statistics.physical_type()
+ cdef uint32_t type_length = statistics.descr().type_length()
+ if physical_type == ParquetType_BOOLEAN:
+ return (<CBoolStatistics*> statistics).max()
+ elif physical_type == ParquetType_INT32:
+ return (<CInt32Statistics*> statistics).max()
+ elif physical_type == ParquetType_INT64:
+ return (<CInt64Statistics*> statistics).max()
+ elif physical_type == ParquetType_FLOAT:
+ return (<CFloatStatistics*> statistics).max()
+ elif physical_type == ParquetType_DOUBLE:
+ return (<CDoubleStatistics*> statistics).max()
+ elif physical_type == ParquetType_BYTE_ARRAY:
+ return _box_byte_array((<CByteArrayStatistics*> statistics).max())
+ elif physical_type == ParquetType_FIXED_LEN_BYTE_ARRAY:
+ return _box_flba((<CFLBAStatistics*> statistics).max(), type_length)
+
+
+cdef _cast_statistic_min(CStatistics* statistics):
+ min_raw = _cast_statistic_raw_min(statistics)
+ return _box_logical_type_value(min_raw, statistics.descr())
+
+
+cdef _cast_statistic_max(CStatistics* statistics):
+ max_raw = _cast_statistic_raw_max(statistics)
+ return _box_logical_type_value(max_raw, statistics.descr())
+
+
+cdef _box_logical_type_value(object value, const ColumnDescriptor* descr):
+ cdef:
+ const CParquetLogicalType* ltype = descr.logical_type().get()
+ ParquetTimeUnit time_unit
+ const CParquetIntType* itype
+ const CParquetTimestampType* ts_type
+
+ if ltype.type() == ParquetLogicalType_STRING:
+ return value.decode('utf8')
+ elif ltype.type() == ParquetLogicalType_TIME:
+ time_unit = (<const CParquetTimeType*> ltype).time_unit()
+ if time_unit == ParquetTimeUnit_MILLIS:
+ return _box_time_milli(value)
+ else:
+ return _box_time_micro(value)
+ elif ltype.type() == ParquetLogicalType_TIMESTAMP:
+ ts_type = <const CParquetTimestampType*> ltype
+ time_unit = ts_type.time_unit()
+ if time_unit == ParquetTimeUnit_MILLIS:
+ converter = _datetime_conversion_functions()[TimeUnit_MILLI]
+ elif time_unit == ParquetTimeUnit_MICROS:
+ converter = _datetime_conversion_functions()[TimeUnit_MICRO]
+ elif time_unit == ParquetTimeUnit_NANOS:
+ converter = _datetime_conversion_functions()[TimeUnit_NANO]
+ else:
+ raise ValueError("Unsupported time unit")
+
+ if ts_type.is_adjusted_to_utc():
+ import pytz
+ tzinfo = pytz.utc
+ else:
+ tzinfo = None
+
+ return converter(value, tzinfo)
+ elif ltype.type() == ParquetLogicalType_INT:
+ itype = <const CParquetIntType*> ltype
+ if not itype.is_signed() and itype.bit_width() == 32:
+ return int(np.int32(value).view(np.uint32))
+ elif not itype.is_signed() and itype.bit_width() == 64:
+ return int(np.int64(value).view(np.uint64))
+ else:
+ return value
+ else:
+ # No logical boxing defined
+ return value
+
+
+cdef _box_byte_array(ParquetByteArray val):
+ return cp.PyBytes_FromStringAndSize(<char*> val.ptr, <Py_ssize_t> val.len)
+
+
+cdef _box_flba(ParquetFLBA val, uint32_t len):
+ return cp.PyBytes_FromStringAndSize(<char*> val.ptr, <Py_ssize_t> len)
+
cdef class ColumnChunkMetaData:
cdef:
@@ -278,7 +403,7 @@ cdef class ColumnChunkMetaData:
def statistics(self):
if not self.metadata.is_stats_set():
return None
- statistics = RowGroupStatistics()
+ statistics = Statistics()
statistics.init(self.metadata.statistics())
return statistics
@@ -581,10 +706,10 @@ cdef class ParquetSchema:
elements = []
for i in range(self.schema.num_columns()):
col = self.column(i)
- converted_type = col.converted_type
+ logical_type = col.logical_type
formatted = '{0}: {1}'.format(col.path, col.physical_type)
- if converted_type != 'NONE':
- formatted += ' {0}'.format(converted_type)
+ if logical_type.type != 'NONE':
+ formatted += ' {0}'.format(str(logical_type))
elements.append(formatted)
return """{0}
@@ -682,10 +807,13 @@ cdef class ColumnSchema:
max_definition_level: {2}
max_repetition_level: {3}
physical_type: {4}
- converted_type: {5}""".format(self.name, self.path,
- self.max_definition_level,
- self.max_repetition_level, physical_type,
- converted_type)
+ logical_type: {5}
+ converted_type (legacy): {6}""".format(self.name, self.path,
+ self.max_definition_level,
+ self.max_repetition_level,
+ physical_type,
+ str(self.logical_type),
+ converted_type)
@property
def name(self):
@@ -708,13 +836,16 @@ cdef class ColumnSchema:
return physical_type_name_from_enum(self.descr.physical_type())
@property
+ def logical_type(self):
+ return wrap_logical_type(self.descr.logical_type())
+
+ @property
def converted_type(self):
return converted_type_name_from_enum(self.descr.converted_type())
@property
def logical_type(self):
- # TODO: wrap new LogicalType objects
- return self.converted_type
+ return wrap_logical_type(self.descr.logical_type())
# FIXED_LEN_BYTE_ARRAY attribute
@property
@@ -744,6 +875,25 @@ cdef physical_type_name_from_enum(ParquetType type_):
}.get(type_, 'UNKNOWN')
+cdef logical_type_name_from_enum(ParquetLogicalTypeId type_):
+ return {
+ ParquetLogicalType_UNKNOWN: 'UNKNOWN',
+ ParquetLogicalType_STRING: 'STRING',
+ ParquetLogicalType_MAP: 'MAP',
+ ParquetLogicalType_LIST: 'LIST',
+ ParquetLogicalType_ENUM: 'ENUM',
+ ParquetLogicalType_DECIMAL: 'DECIMAL',
+ ParquetLogicalType_DATE: 'DATE',
+ ParquetLogicalType_TIME: 'TIME',
+ ParquetLogicalType_TIMESTAMP: 'TIMESTAMP',
+ ParquetLogicalType_INT: 'INT',
+ ParquetLogicalType_JSON: 'JSON',
+ ParquetLogicalType_BSON: 'BSON',
+ ParquetLogicalType_UUID: 'UUID',
+ ParquetLogicalType_NONE: 'NONE',
+ }.get(type_, 'UNKNOWN')
+
+
cdef converted_type_name_from_enum(ParquetConvertedType type_):
return {
ParquetConvertedType_NONE: 'NONE',
diff --git a/python/pyarrow/parquet.py b/python/pyarrow/parquet.py
index 108bc9d..2d780af 100644
--- a/python/pyarrow/parquet.py
+++ b/python/pyarrow/parquet.py
@@ -33,7 +33,7 @@ import pyarrow as pa
import pyarrow.lib as lib
import pyarrow._parquet as _parquet
-from pyarrow._parquet import (ParquetReader, RowGroupStatistics, # noqa
+from pyarrow._parquet import (ParquetReader, Statistics, # noqa
FileMetaData, RowGroupMetaData,
ColumnChunkMetaData,
ParquetSchema, ColumnSchema)
diff --git a/python/pyarrow/scalar.pxi b/python/pyarrow/scalar.pxi
index aa86d9e..326aefa 100644
--- a/python/pyarrow/scalar.pxi
+++ b/python/pyarrow/scalar.pxi
@@ -258,8 +258,7 @@ cdef class Time32Value(ArrayValue):
delta = datetime.timedelta(seconds=ap.Value(self.index))
return (datetime.datetime(1970, 1, 1) + delta).time()
else:
- delta = datetime.timedelta(milliseconds=ap.Value(self.index))
- return (datetime.datetime(1970, 1, 1) + delta).time()
+ return _box_time_milli(ap.Value(self.index))
cdef class Time64Value(ArrayValue):
@@ -277,17 +276,27 @@ cdef class Time64Value(ArrayValue):
cdef int64_t val = ap.Value(self.index)
if dtype.unit() == TimeUnit_MICRO:
- return (datetime.datetime(1970, 1, 1) +
- datetime.timedelta(microseconds=val)).time()
+ return _box_time_micro(val)
else:
return (datetime.datetime(1970, 1, 1) +
datetime.timedelta(microseconds=val / 1000)).time()
+cpdef _box_time_milli(int64_t val):
+ delta = datetime.timedelta(milliseconds=val)
+ return (datetime.datetime(1970, 1, 1) + delta).time()
+
+
+cpdef _box_time_micro(int64_t val):
+ return (datetime.datetime(1970, 1, 1) +
+ datetime.timedelta(microseconds=val)).time()
+
+
cdef dict _DATETIME_CONVERSION_FUNCTIONS = {}
cdef c_bool _datetime_conversion_initialized = False
-cdef _datetime_conversion_functions():
+
+def _datetime_conversion_functions():
global _datetime_conversion_initialized
if _datetime_conversion_initialized:
return _DATETIME_CONVERSION_FUNCTIONS
diff --git a/python/pyarrow/tests/test_parquet.py b/python/pyarrow/tests/test_parquet.py
index b622e40..4891907 100644
--- a/python/pyarrow/tests/test_parquet.py
+++ b/python/pyarrow/tests/test_parquet.py
@@ -1,3 +1,4 @@
+# -*- coding: utf-8 -*-
# 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
@@ -669,7 +670,7 @@ def test_parquet_metadata_api():
assert col_meta.num_values == 10000
assert col_meta.path_in_schema == 'bool'
assert col_meta.is_stats_set is True
- assert isinstance(col_meta.statistics, pq.RowGroupStatistics)
+ assert isinstance(col_meta.statistics, pq.Statistics)
assert col_meta.compression == 'SNAPPY'
assert col_meta.encodings == ('PLAIN', 'RLE')
assert col_meta.has_dictionary_page is False
@@ -741,8 +742,8 @@ def test_parquet_column_statistics_api(data, type, physical_type, min_value,
stat = col_meta.statistics
assert stat.has_min_max
- assert stat.min == min_value
- assert stat.max == max_value
+ assert _close(type, stat.min, min_value)
+ assert _close(type, stat.max, max_value)
assert stat.null_count == null_count
assert stat.num_values == num_values
# TODO(kszucs) until parquet-cpp API doesn't expose HasDistinctCount
@@ -751,6 +752,44 @@ def test_parquet_column_statistics_api(data, type, physical_type, min_value,
assert stat.physical_type == physical_type
+def _close(type, left, right):
+ if type == pa.float32():
+ return abs(left - right) < 1E-7
+ elif type == pa.float64():
+ return abs(left - right) < 1E-13
+ else:
+ return left == right
+
+
+def test_statistics_convert_logical_types(tempdir):
+ # ARROW-5166, ARROW-4139
+
+ # (min, max, type)
+ cases = [(10, 11164359321221007157, pa.uint64()),
+ (10, 4294967295, pa.uint32()),
+ (u"ähnlich", u"öffentlich", pa.utf8()),
+ (datetime.time(10, 30, 0, 1000), datetime.time(15, 30, 0, 1000),
+ pa.time32('ms')),
+ (datetime.time(10, 30, 0, 1000), datetime.time(15, 30, 0, 1000),
+ pa.time64('us')),
+ (datetime.datetime(2019, 6, 24, 0, 0, 0, 1000),
+ datetime.datetime(2019, 6, 25, 0, 0, 0, 1000),
+ pa.timestamp('ms')),
+ (datetime.datetime(2019, 6, 24, 0, 0, 0, 1000),
+ datetime.datetime(2019, 6, 25, 0, 0, 0, 1000),
+ pa.timestamp('us'))]
+
+ for i, (min_val, max_val, typ) in enumerate(cases):
+ t = pa.Table.from_arrays([pa.array([min_val, max_val], type=typ)],
+ ['col'])
+ path = str(tempdir / ('example{}.parquet'.format(i)))
+ pq.write_table(t, path, version='2.0')
+ pf = pq.ParquetFile(path)
+ stats = pf.metadata.row_group(0).column(0).statistics
+ assert stats.min == min_val
+ assert stats.max == max_val
+
+
def test_parquet_write_disable_statistics(tempdir):
table = pa.Table.from_pydict(
{'a': pa.array([1, 2, 3]), 'b': pa.array(['a', 'b', 'c'])})