Skip to content

Commit da29595

Browse files
cpcloudwesm
cpcloud
authored and
wesm
committed
PARQUET-1095: [C++] Read and write Arrow decimal values
This depends on: - [x] [ARROW-1607](apache#1128) - [x] [ARROW-1656](apache#1184) - [x] [ARROW-1588](apache#1211) - [x] Add tests for writing different sizes of values Author: Phillip Cloud <cpcloud@gmail.com> Author: Wes McKinney <wes.mckinney@twosigma.com> Closes apache#403 from cpcloud/PARQUET-1095 and squashes the following commits: 8c3d222 [Phillip Cloud] Remove loop from BytesToInteger 63018bc [Wes McKinney] Suppress C4996 due to arrow/util/variant.h e4b02d3 [Phillip Cloud] Refactor types.h 83948ec [Phillip Cloud] Add last_value_ init 51965cd [Phillip Cloud] Min commit that contains the unique kernel in arrow e25c59b [Phillip Cloud] Fix reader writer test for unique kernel addition da0a7eb [Phillip Cloud] Update for ARROW-1811 16935de [Phillip Cloud] Reverse operand order and explicit cast 6036ca5 [Phillip Cloud] ARROW-1811 c5c4294 [Phillip Cloud] Fix issues 32a4abe [Phillip Cloud] Cleanup iteration a bit 920832a [Phillip Cloud] Update arrow version 9f97c1d [Phillip Cloud] Update for ARROW-1794: rename DecimalArray to Decimal128Array b2e0290 [Phillip Cloud] IWYU 64748a8 [Phillip Cloud] Copy from arrow for now 6c9e2a7 [Phillip Cloud] Reduce the number of decimal test cases 7ab2e5c [Phillip Cloud] Parameterize on precision 30655d6 [Phillip Cloud] Use arrow random_decimals 9ff7eb4 [Phillip Cloud] Remove specific template parameters 1eee6a9 [Phillip Cloud] Remove specific randint call 8808e4c [Phillip Cloud] Bump arrow version 659fbc1 [Phillip Cloud] Fix deprecated API call e162ca1 [Phillip Cloud] Allocate scratch space to hold the byteswapped values 5c9292b [Phillip Cloud] Proper dcheck call 1782da0 [Phillip Cloud] Use arrow 3d243d5 [Phillip Cloud] Checkpoint [ci skip] 028fb03 [Phillip Cloud] Remove garbage values 46dff15 [Phillip Cloud] Clean up uint32 test 613255e [Phillip Cloud] Do not use std::copy when reinterpret_cast will suffice 2917a62 [Phillip Cloud] PARQUET-1095: [C++] Read and write Arrow decimal values Change-Id: Ibe81cd5a5961bbe86c66db811ec8b770ae48c38b
1 parent dcea0ab commit da29595

11 files changed

Lines changed: 721 additions & 152 deletions

File tree

‎cpp/src/parquet/arrow/arrow-reader-writer-test.cc‎

Lines changed: 141 additions & 34 deletions
Original file line numberDiff line numberDiff line change
@@ -24,6 +24,7 @@
2424
#include "gtest/gtest.h"
2525

2626
#include <sstream>
27+
#include <arrow/compute/api.h>
2728

2829
#include "parquet/api/reader.h"
2930
#include "parquet/api/writer.h"
@@ -37,20 +38,23 @@
3738

3839
#include "arrow/api.h"
3940
#include "arrow/test-util.h"
41+
#include "arrow/util/decimal.h"
4042

4143
using arrow::Array;
4244
using arrow::ArrayVisitor;
4345
using arrow::Buffer;
4446
using arrow::ChunkedArray;
4547
using arrow::Column;
46-
using arrow::EncodeArrayToDictionary;
4748
using arrow::ListArray;
4849
using arrow::PoolBuffer;
4950
using arrow::PrimitiveArray;
5051
using arrow::Status;
5152
using arrow::Table;
5253
using arrow::TimeUnit;
5354
using arrow::default_memory_pool;
55+
using arrow::compute::DictionaryEncode;
56+
using arrow::compute::FunctionContext;
57+
using arrow::compute::Datum;
5458
using arrow::io::BufferReader;
5559

5660
using arrow::test::randint;
@@ -68,10 +72,10 @@ using ColumnVector = std::vector<std::shared_ptr<arrow::Column>>;
6872
namespace parquet {
6973
namespace arrow {
7074

71-
const int SMALL_SIZE = 100;
72-
const int LARGE_SIZE = 10000;
75+
static constexpr int SMALL_SIZE = 100;
76+
static constexpr int LARGE_SIZE = 10000;
7377

74-
constexpr uint32_t kDefaultSeed = 0;
78+
static constexpr uint32_t kDefaultSeed = 0;
7579

7680
LogicalType::type get_logical_type(const ::arrow::DataType& type) {
7781
switch (type.id()) {
@@ -118,6 +122,8 @@ LogicalType::type get_logical_type(const ::arrow::DataType& type) {
118122
static_cast<const ::arrow::DictionaryType&>(type);
119123
return get_logical_type(*dict_type.dictionary()->type());
120124
}
125+
case ArrowId::DECIMAL:
126+
return LogicalType::DECIMAL;
121127
default:
122128
break;
123129
}
@@ -147,6 +153,7 @@ ParquetType::type get_physical_type(const ::arrow::DataType& type) {
147153
case ArrowId::STRING:
148154
return ParquetType::BYTE_ARRAY;
149155
case ArrowId::FIXED_SIZE_BINARY:
156+
case ArrowId::DECIMAL:
150157
return ParquetType::FIXED_LEN_BYTE_ARRAY;
151158
case ArrowId::DATE32:
152159
return ParquetType::INT32;
@@ -299,6 +306,7 @@ struct test_traits<::arrow::FixedSizeBinaryType> {
299306
const std::string test_traits<::arrow::StringType>::value("Test"); // NOLINT
300307
const std::string test_traits<::arrow::BinaryType>::value("\x00\x01\x02\x03"); // NOLINT
301308
const std::string test_traits<::arrow::FixedSizeBinaryType>::value("Fixed"); // NOLINT
309+
302310
template <typename T>
303311
using ParquetDataType = DataType<test_traits<T>::parquet_enum>;
304312

@@ -342,36 +350,52 @@ void DoSimpleRoundtrip(const std::shared_ptr<Table>& table, int num_threads,
342350

343351
static std::shared_ptr<GroupNode> MakeSimpleSchema(const ::arrow::DataType& type,
344352
Repetition::type repetition) {
345-
int byte_width;
346-
// Decimal is not implemented yet.
353+
int32_t byte_width = -1;
354+
int32_t precision = -1;
355+
int32_t scale = -1;
356+
347357
switch (type.id()) {
348358
case ::arrow::Type::DICTIONARY: {
349-
const ::arrow::DictionaryType& dict_type =
350-
static_cast<const ::arrow::DictionaryType&>(type);
359+
const auto& dict_type = static_cast<const ::arrow::DictionaryType&>(type);
351360
const ::arrow::DataType& values_type = *dict_type.dictionary()->type();
352-
if (values_type.id() == ::arrow::Type::FIXED_SIZE_BINARY) {
353-
byte_width =
354-
static_cast<const ::arrow::FixedSizeBinaryType&>(values_type).byte_width();
355-
} else {
356-
byte_width = -1;
361+
switch (values_type.id()) {
362+
case ::arrow::Type::FIXED_SIZE_BINARY:
363+
byte_width =
364+
static_cast<const ::arrow::FixedSizeBinaryType&>(values_type).byte_width();
365+
break;
366+
case ::arrow::Type::DECIMAL: {
367+
const auto& decimal_type =
368+
static_cast<const ::arrow::Decimal128Type&>(values_type);
369+
precision = decimal_type.precision();
370+
scale = decimal_type.scale();
371+
byte_width = DecimalSize(precision);
372+
} break;
373+
default:
374+
break;
357375
}
358376
} break;
359377
case ::arrow::Type::FIXED_SIZE_BINARY:
360378
byte_width = static_cast<const ::arrow::FixedSizeBinaryType&>(type).byte_width();
361379
break;
380+
case ::arrow::Type::DECIMAL: {
381+
const auto& decimal_type = static_cast<const ::arrow::Decimal128Type&>(type);
382+
precision = decimal_type.precision();
383+
scale = decimal_type.scale();
384+
byte_width = DecimalSize(precision);
385+
} break;
362386
default:
363-
byte_width = -1;
387+
break;
364388
}
365389
auto pnode = PrimitiveNode::Make("column1", repetition, get_physical_type(type),
366-
get_logical_type(type), byte_width);
390+
get_logical_type(type), byte_width, precision, scale);
367391
NodePtr node_ =
368392
GroupNode::Make("schema", Repetition::REQUIRED, std::vector<NodePtr>({pnode}));
369393
return std::static_pointer_cast<GroupNode>(node_);
370394
}
371395

372396
namespace internal {
373397

374-
void AssertArraysEqual(const Array &expected, const Array &actual) {
398+
void AssertArraysEqual(const Array& expected, const Array& actual) {
375399
if (!actual.Equals(expected)) {
376400
std::stringstream pp_result;
377401
std::stringstream pp_expected;
@@ -526,11 +550,19 @@ class TestParquetIO : public ::testing::Test {
526550
// There we write an UInt32 Array but receive an Int64 Array as result for
527551
// Parquet version 1.0.
528552

529-
typedef ::testing::Types<::arrow::BooleanType, ::arrow::UInt8Type, ::arrow::Int8Type,
530-
::arrow::UInt16Type, ::arrow::Int16Type, ::arrow::Int32Type,
531-
::arrow::UInt64Type, ::arrow::Int64Type, ::arrow::Date32Type,
532-
::arrow::FloatType, ::arrow::DoubleType, ::arrow::StringType,
533-
::arrow::BinaryType, ::arrow::FixedSizeBinaryType>
553+
typedef ::testing::Types<
554+
::arrow::BooleanType, ::arrow::UInt8Type, ::arrow::Int8Type, ::arrow::UInt16Type,
555+
::arrow::Int16Type, ::arrow::Int32Type, ::arrow::UInt64Type, ::arrow::Int64Type,
556+
::arrow::Date32Type, ::arrow::FloatType, ::arrow::DoubleType, ::arrow::StringType,
557+
::arrow::BinaryType, ::arrow::FixedSizeBinaryType, DecimalWithPrecisionAndScale<1>,
558+
DecimalWithPrecisionAndScale<3>, DecimalWithPrecisionAndScale<5>,
559+
DecimalWithPrecisionAndScale<7>, DecimalWithPrecisionAndScale<10>,
560+
DecimalWithPrecisionAndScale<12>, DecimalWithPrecisionAndScale<15>,
561+
DecimalWithPrecisionAndScale<17>, DecimalWithPrecisionAndScale<19>,
562+
DecimalWithPrecisionAndScale<22>, DecimalWithPrecisionAndScale<23>,
563+
DecimalWithPrecisionAndScale<24>, DecimalWithPrecisionAndScale<27>,
564+
DecimalWithPrecisionAndScale<29>, DecimalWithPrecisionAndScale<32>,
565+
DecimalWithPrecisionAndScale<34>, DecimalWithPrecisionAndScale<38>>
534566
TestTypes;
535567

536568
TYPED_TEST_CASE(TestParquetIO, TestTypes);
@@ -590,8 +622,10 @@ TYPED_TEST(TestParquetIO, SingleColumnOptionalDictionaryWrite) {
590622

591623
ASSERT_OK(NullableArray<TypeParam>(SMALL_SIZE, 10, kDefaultSeed, &values));
592624

593-
std::shared_ptr<Array> dict_values;
594-
ASSERT_OK(EncodeArrayToDictionary(*values, default_memory_pool(), &dict_values));
625+
Datum out;
626+
FunctionContext ctx(default_memory_pool());
627+
ASSERT_OK(DictionaryEncode(&ctx, Datum(values), &out));
628+
std::shared_ptr<Array> dict_values = MakeArray(out.array());
595629
std::shared_ptr<GroupNode> schema =
596630
MakeSimpleSchema(*dict_values->type(), Repetition::OPTIONAL);
597631
this->WriteColumn(schema, dict_values);
@@ -856,25 +890,43 @@ TEST_F(TestUInt32ParquetIO, Parquet_1_0_Compability) {
856890
ASSERT_OK_NO_THROW(
857891
WriteTable(*table, ::arrow::default_memory_pool(), this->sink_, 512, properties));
858892

859-
std::shared_ptr<Array> expected_values;
860893
std::shared_ptr<PoolBuffer> int64_data =
861894
std::make_shared<PoolBuffer>(::arrow::default_memory_pool());
862895
{
863896
ASSERT_OK(int64_data->Resize(sizeof(int64_t) * values->length()));
864-
int64_t* int64_data_ptr = reinterpret_cast<int64_t*>(int64_data->mutable_data());
865-
const uint32_t* uint32_data_ptr =
866-
reinterpret_cast<const uint32_t*>(values->values()->data());
867-
// std::copy might be faster but this is explicit on the casts)
868-
for (int64_t i = 0; i < values->length(); i++) {
869-
int64_data_ptr[i] = static_cast<int64_t>(uint32_data_ptr[i]);
870-
}
897+
auto int64_data_ptr = reinterpret_cast<int64_t*>(int64_data->mutable_data());
898+
auto uint32_data_ptr = reinterpret_cast<const uint32_t*>(values->values()->data());
899+
const auto cast_uint32_to_int64 = [](uint32_t value) {
900+
return static_cast<int64_t>(value);
901+
};
902+
std::transform(uint32_data_ptr, uint32_data_ptr + values->length(), int64_data_ptr,
903+
cast_uint32_to_int64);
871904
}
872905

873906
std::vector<std::shared_ptr<Buffer>> buffers{values->null_bitmap(), int64_data};
874907
auto arr_data = std::make_shared<::arrow::ArrayData>(::arrow::int64(), values->length(),
875908
buffers, values->null_count());
876-
ASSERT_OK(MakeArray(arr_data, &expected_values));
877-
this->ReadAndCheckSingleColumnTable(expected_values);
909+
std::shared_ptr<Array> expected_values = MakeArray(arr_data);
910+
ASSERT_NE(expected_values, NULLPTR);
911+
912+
const auto& expected = static_cast<const ::arrow::Int64Array&>(*expected_values);
913+
ASSERT_GT(values->length(), 0);
914+
ASSERT_EQ(values->length(), expected.length());
915+
916+
// TODO(phillipc): Is there a better way to compare these two arrays?
917+
// AssertArraysEqual requires the same type, but we only care about values in this case
918+
for (int i = 0; i < expected.length(); ++i) {
919+
const bool value_is_valid = values->IsValid(i);
920+
const bool expected_value_is_valid = expected.IsValid(i);
921+
922+
ASSERT_EQ(expected_value_is_valid, value_is_valid);
923+
924+
if (value_is_valid) {
925+
uint32_t value = values->Value(i);
926+
int64_t expected_value = expected.Value(i);
927+
ASSERT_EQ(expected_value, static_cast<int64_t>(value));
928+
}
929+
}
878930
}
879931

880932
using TestStringParquetIO = TestParquetIO<::arrow::StringType>;
@@ -1432,7 +1484,7 @@ void MakeListTable(int num_rows, std::shared_ptr<Table>* out) {
14321484
offset_values.push_back(total_elements);
14331485

14341486
std::vector<int8_t> value_draws;
1435-
randint<int8_t>(total_elements, 0, 100, &value_draws);
1487+
randint(total_elements, 0, 100, &value_draws);
14361488

14371489
std::vector<bool> is_valid;
14381490
random_is_valid(total_elements, 0.1, &is_valid);
@@ -1889,6 +1941,61 @@ TEST(TestArrowReaderAdHoc, Int96BadMemoryAccess) {
18891941
ASSERT_OK_NO_THROW(arrow_reader->ReadTable(&table));
18901942
}
18911943

1944+
class TestArrowReaderAdHocSpark
1945+
: public ::testing::TestWithParam<
1946+
std::tuple<std::string, std::shared_ptr<::arrow::DataType>>> {};
1947+
1948+
TEST_P(TestArrowReaderAdHocSpark, ReadDecimals) {
1949+
std::string path(std::getenv("PARQUET_TEST_DATA"));
1950+
1951+
std::string filename;
1952+
std::shared_ptr<::arrow::DataType> decimal_type;
1953+
std::tie(filename, decimal_type) = GetParam();
1954+
1955+
path += "/" + filename;
1956+
ASSERT_GT(path.size(), 0);
1957+
1958+
auto pool = ::arrow::default_memory_pool();
1959+
1960+
std::unique_ptr<FileReader> arrow_reader;
1961+
ASSERT_NO_THROW(
1962+
arrow_reader.reset(new FileReader(pool, ParquetFileReader::OpenFile(path, false))));
1963+
std::shared_ptr<::arrow::Table> table;
1964+
ASSERT_OK_NO_THROW(arrow_reader->ReadTable(&table));
1965+
1966+
ASSERT_EQ(1, table->num_columns());
1967+
1968+
constexpr int32_t expected_length = 24;
1969+
1970+
auto value_column = table->column(0);
1971+
ASSERT_EQ(expected_length, value_column->length());
1972+
1973+
auto raw_array = value_column->data();
1974+
ASSERT_EQ(1, raw_array->num_chunks());
1975+
1976+
auto chunk = raw_array->chunk(0);
1977+
1978+
std::shared_ptr<Array> expected_array;
1979+
1980+
::arrow::Decimal128Builder builder(decimal_type, pool);
1981+
1982+
for (int32_t i = 0; i < expected_length; ++i) {
1983+
::arrow::Decimal128 value((i + 1) * 100);
1984+
ASSERT_OK(builder.Append(value));
1985+
}
1986+
ASSERT_OK(builder.Finish(&expected_array));
1987+
1988+
internal::AssertArraysEqual(*expected_array, *chunk);
1989+
}
1990+
1991+
INSTANTIATE_TEST_CASE_P(
1992+
ReadDecimals, TestArrowReaderAdHocSpark,
1993+
::testing::Values(
1994+
std::make_tuple("int32_decimal.parquet", ::arrow::decimal(4, 2)),
1995+
std::make_tuple("int64_decimal.parquet", ::arrow::decimal(10, 2)),
1996+
std::make_tuple("fixed_length_decimal.parquet", ::arrow::decimal(25, 2)),
1997+
std::make_tuple("fixed_length_decimal_legacy.parquet", ::arrow::decimal(13, 2))));
1998+
18921999
} // namespace arrow
18932000

18942001
} // namespace parquet

‎cpp/src/parquet/arrow/arrow-schema-test.cc‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -51,7 +51,7 @@ const auto TIMESTAMP_MS = ::arrow::timestamp(TimeUnit::MILLI);
5151
const auto TIMESTAMP_US = ::arrow::timestamp(TimeUnit::MICRO);
5252
const auto TIMESTAMP_NS = ::arrow::timestamp(TimeUnit::NANO);
5353
const auto BINARY = ::arrow::binary();
54-
const auto DECIMAL_8_4 = std::make_shared<::arrow::DecimalType>(8, 4);
54+
const auto DECIMAL_8_4 = std::make_shared<::arrow::Decimal128Type>(8, 4);
5555

5656
class TestConvertParquetSchema : public ::testing::Test {
5757
public:
@@ -62,8 +62,8 @@ class TestConvertParquetSchema : public ::testing::Test {
6262
for (int i = 0; i < expected_schema->num_fields(); ++i) {
6363
auto lhs = result_schema_->field(i);
6464
auto rhs = expected_schema->field(i);
65-
EXPECT_TRUE(lhs->Equals(rhs)) << i << " " << lhs->ToString()
66-
<< " != " << rhs->ToString();
65+
EXPECT_TRUE(lhs->Equals(rhs))
66+
<< i << " " << lhs->ToString() << " != " << rhs->ToString();
6767
}
6868
}
6969

0 commit comments

Comments
 (0)