Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Refactor LogicalType for Parquet #14264

Merged
merged 17 commits into from
Oct 20, 2023
Merged

Refactor LogicalType for Parquet #14264

Show file tree
Hide file tree
Changes from 8 commits
Commits
Show all changes
17 commits
Select commit Hold shift + click to select a range
1a0fd1f
refactor LogicalType
etseidl Oct 10, 2023
a0ed18d
pass optional to compare_binary
etseidl Oct 10, 2023
35c9b96
use enum for field number
etseidl Oct 10, 2023
09c0709
add helpers for nanoseconds
etseidl Oct 10, 2023
d1c9899
Merge remote-tracking branch 'origin/branch-23.12' into refactor_logi…
etseidl Oct 10, 2023
a854121
finish merge
etseidl Oct 10, 2023
cbdc193
add more helper functions
etseidl Oct 10, 2023
c4e4a9d
Merge remote-tracking branch 'origin/branch-23.12' into refactor_logi…
etseidl Oct 10, 2023
a018c60
fix for unknown converted type
etseidl Oct 11, 2023
9b0b36d
Merge branch 'branch-23.12' into refactor_logical_type
etseidl Oct 11, 2023
4513b94
lost a line
etseidl Oct 11, 2023
f5a7cb7
Merge branch 'branch-23.12' into refactor_logical_type
etseidl Oct 11, 2023
ebacc12
address some review comments
etseidl Oct 11, 2023
517460a
fail when writing invalid LogicalType
etseidl Oct 11, 2023
d9139ff
get rid of some superfluous braces
etseidl Oct 11, 2023
c2f57fe
make writers for ColumnOrder and TimeUnit match behavior of LogicalType
etseidl Oct 11, 2023
c56d399
Merge branch 'branch-23.12' into refactor_logical_type
vuule Oct 20, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
95 changes: 22 additions & 73 deletions cpp/src/io/parquet/compact_protocol_reader.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -339,61 +339,6 @@ struct parquet_field_struct_list : public parquet_field_list<T> {
}
};

// TODO(ets): replace current union handling (which mirrors thrift) to use std::optional fields
// in a struct
/**
* @brief Functor to read a union member from CompactProtocolReader
*
* @tparam is_empty True if tparam `T` type is empty type, else false.
*
* @return True if field types mismatch or if the process of reading a
* union member fails
*/
template <typename T, bool is_empty = false>
class ParquetFieldUnionFunctor : public parquet_field {
bool& is_set;
T& val;

public:
ParquetFieldUnionFunctor(int f, bool& b, T& v) : parquet_field(f), is_set(b), val(v) {}

inline bool operator()(CompactProtocolReader* cpr, int field_type)
{
if (field_type != ST_FLD_STRUCT) {
return true;
} else {
is_set = true;
return !cpr->read(&val);
}
}
};

template <typename T>
class ParquetFieldUnionFunctor<T, true> : public parquet_field {
bool& is_set;
T& val;

public:
ParquetFieldUnionFunctor(int f, bool& b, T& v) : parquet_field(f), is_set(b), val(v) {}

inline bool operator()(CompactProtocolReader* cpr, int field_type)
{
if (field_type != ST_FLD_STRUCT) {
return true;
} else {
is_set = true;
cpr->skip_struct_field(field_type);
return false;
}
}
};

template <typename T>
ParquetFieldUnionFunctor<T, std::is_empty_v<T>> ParquetFieldUnion(int f, bool& b, T& v)
{
return ParquetFieldUnionFunctor<T, std::is_empty_v<T>>(f, b, v);
}

/**
* @brief Functor to read a binary from CompactProtocolReader
*
Expand Down Expand Up @@ -595,34 +540,38 @@ bool CompactProtocolReader::read(FileMetaData* f)

bool CompactProtocolReader::read(SchemaElement* s)
{
using optional_converted_type =
parquet_field_optional<ConvertedType, parquet_field_enum<ConvertedType>>;
using optional_logical_type =
parquet_field_optional<LogicalType, parquet_field_struct<LogicalType>>;
auto op = std::make_tuple(parquet_field_enum<Type>(1, s->type),
parquet_field_int32(2, s->type_length),
parquet_field_enum<FieldRepetitionType>(3, s->repetition_type),
parquet_field_string(4, s->name),
parquet_field_int32(5, s->num_children),
parquet_field_enum<ConvertedType>(6, s->converted_type),
optional_converted_type(6, s->converted_type),
parquet_field_int32(7, s->decimal_scale),
parquet_field_int32(8, s->decimal_precision),
parquet_field_optional<int32_t, parquet_field_int32>(9, s->field_id),
parquet_field_struct(10, s->logical_type));
optional_logical_type(10, s->logical_type));
return function_builder(this, op);
}

bool CompactProtocolReader::read(LogicalType* l)
{
auto op =
std::make_tuple(ParquetFieldUnion(1, l->isset.STRING, l->STRING),
ParquetFieldUnion(2, l->isset.MAP, l->MAP),
ParquetFieldUnion(3, l->isset.LIST, l->LIST),
ParquetFieldUnion(4, l->isset.ENUM, l->ENUM),
ParquetFieldUnion(5, l->isset.DECIMAL, l->DECIMAL), // read the struct
ParquetFieldUnion(6, l->isset.DATE, l->DATE),
ParquetFieldUnion(7, l->isset.TIME, l->TIME), // read the struct
ParquetFieldUnion(8, l->isset.TIMESTAMP, l->TIMESTAMP), // read the struct
ParquetFieldUnion(10, l->isset.INTEGER, l->INTEGER), // read the struct
ParquetFieldUnion(11, l->isset.UNKNOWN, l->UNKNOWN),
ParquetFieldUnion(12, l->isset.JSON, l->JSON),
ParquetFieldUnion(13, l->isset.BSON, l->BSON));
auto op = std::make_tuple(
parquet_field_union_enumerator(1, l->type),
parquet_field_union_enumerator(2, l->type),
parquet_field_union_enumerator(3, l->type),
parquet_field_union_enumerator(4, l->type),
parquet_field_union_struct<LogicalType::Type, DecimalType>(5, l->type, l->decimal_type),
parquet_field_union_enumerator(6, l->type),
parquet_field_union_struct<LogicalType::Type, TimeType>(7, l->type, l->time_type),
parquet_field_union_struct<LogicalType::Type, TimestampType>(8, l->type, l->timestamp_type),
parquet_field_union_struct<LogicalType::Type, IntType>(10, l->type, l->int_type),
parquet_field_union_enumerator(11, l->type),
parquet_field_union_enumerator(12, l->type),
parquet_field_union_enumerator(13, l->type));
return function_builder(this, op);
}

Expand All @@ -648,9 +597,9 @@ bool CompactProtocolReader::read(TimestampType* t)

bool CompactProtocolReader::read(TimeUnit* u)
{
auto op = std::make_tuple(ParquetFieldUnion(1, u->isset.MILLIS, u->MILLIS),
ParquetFieldUnion(2, u->isset.MICROS, u->MICROS),
ParquetFieldUnion(3, u->isset.NANOS, u->NANOS));
auto op = std::make_tuple(parquet_field_union_enumerator(1, u->type),
parquet_field_union_enumerator(2, u->type),
parquet_field_union_enumerator(3, u->type));
return function_builder(this, op);
}

Expand Down
76 changes: 33 additions & 43 deletions cpp/src/io/parquet/compact_protocol_writer.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -46,13 +46,11 @@ size_t CompactProtocolWriter::write(DecimalType const& decimal)
size_t CompactProtocolWriter::write(TimeUnit const& time_unit)
{
CompactProtocolFieldWriter c(*this);
auto const isset = time_unit.isset;
if (isset.MILLIS) {
c.field_struct(1, time_unit.MILLIS);
} else if (isset.MICROS) {
c.field_struct(2, time_unit.MICROS);
} else if (isset.NANOS) {
c.field_struct(3, time_unit.NANOS);
switch (time_unit.type) {
case TimeUnit::MILLIS: c.field_empty_struct(TimeUnit::MILLIS); break;
case TimeUnit::MICROS: c.field_empty_struct(TimeUnit::MICROS); break;
case TimeUnit::NANOS: c.field_empty_struct(TimeUnit::NANOS); break;
default: break;
}
return c.value();
}
Expand Down Expand Up @@ -84,31 +82,28 @@ size_t CompactProtocolWriter::write(IntType const& integer)
size_t CompactProtocolWriter::write(LogicalType const& logical_type)
{
CompactProtocolFieldWriter c(*this);
auto const isset = logical_type.isset;
if (isset.STRING) {
c.field_struct(1, logical_type.STRING);
} else if (isset.MAP) {
c.field_struct(2, logical_type.MAP);
} else if (isset.LIST) {
c.field_struct(3, logical_type.LIST);
} else if (isset.ENUM) {
c.field_struct(4, logical_type.ENUM);
} else if (isset.DECIMAL) {
c.field_struct(5, logical_type.DECIMAL);
} else if (isset.DATE) {
c.field_struct(6, logical_type.DATE);
} else if (isset.TIME) {
c.field_struct(7, logical_type.TIME);
} else if (isset.TIMESTAMP) {
c.field_struct(8, logical_type.TIMESTAMP);
} else if (isset.INTEGER) {
c.field_struct(10, logical_type.INTEGER);
} else if (isset.UNKNOWN) {
c.field_struct(11, logical_type.UNKNOWN);
} else if (isset.JSON) {
c.field_struct(12, logical_type.JSON);
} else if (isset.BSON) {
c.field_struct(13, logical_type.BSON);
switch (logical_type.type) {
case LogicalType::STRING: c.field_empty_struct(LogicalType::STRING); break;
case LogicalType::MAP: c.field_empty_struct(LogicalType::MAP); break;
case LogicalType::LIST: c.field_empty_struct(LogicalType::LIST); break;
case LogicalType::ENUM: c.field_empty_struct(LogicalType::ENUM); break;
case LogicalType::DECIMAL:
c.field_struct(LogicalType::DECIMAL, logical_type.decimal_type.value());
break;
case LogicalType::DATE: c.field_empty_struct(LogicalType::DATE); break;
case LogicalType::TIME:
c.field_struct(LogicalType::TIME, logical_type.time_type.value());
break;
case LogicalType::TIMESTAMP:
c.field_struct(LogicalType::TIMESTAMP, logical_type.timestamp_type.value());
break;
case LogicalType::INTEGER:
c.field_struct(LogicalType::INTEGER, logical_type.int_type.value());
break;
case LogicalType::UNKNOWN: c.field_empty_struct(LogicalType::UNKNOWN); break;
case LogicalType::JSON: c.field_empty_struct(LogicalType::JSON); break;
case LogicalType::BSON: c.field_empty_struct(LogicalType::BSON); break;
vuule marked this conversation as resolved.
Show resolved Hide resolved
default: break;
}
return c.value();
}
Expand All @@ -124,20 +119,15 @@ size_t CompactProtocolWriter::write(SchemaElement const& s)
c.field_string(4, s.name);

if (s.type == UNDEFINED_TYPE) { c.field_int(5, s.num_children); }
if (s.converted_type != UNKNOWN) {
c.field_int(6, s.converted_type);
if (s.converted_type.has_value()) {
c.field_int(6, s.converted_type.value());
if (s.converted_type == DECIMAL) {
c.field_int(7, s.decimal_scale);
c.field_int(8, s.decimal_precision);
}
}
if (s.field_id) { c.field_int(9, s.field_id.value()); }
auto const isset = s.logical_type.isset;
// TODO: add handling for all logical types
// if (isset.STRING or isset.MAP or isset.LIST or isset.ENUM or isset.DECIMAL or isset.DATE or
// isset.TIME or isset.TIMESTAMP or isset.INTEGER or isset.UNKNOWN or isset.JSON or isset.BSON)
// {
if (isset.TIMESTAMP or isset.TIME) { c.field_struct(10, s.logical_type); }
if (s.field_id.has_value()) { c.field_int(9, s.field_id.value()); }
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Note: we could probably get rid of this if has_value -> write value pattern with SFINAE in ProtobufWriter, assuming that the field_xyz names map to the actual type of the parameter.
If all types were supported with an overload set, optional support could be a part of the set, and simply delegate to <T::value> implementation if has_value

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

That's a lot of template foo to avoid a few invocations of has_value. We can leave this for another day :D

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think it would have a larger impact than that, but maybe I'm mixing it up with the reader side. Either way it's not a suggestion for this PR.

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

(Not sure if I'm on the same page)
How about std::visit to deal with such template issue and optional?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

That can be in the next refactor of CompactProtocolReader/Writer 🤣 (or should I say compact_protocol_reader/writer?)

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I don't think std::visit is applicable here. Outside of reflection that would allow us to iterate over data members (which AFAIK does not exist), I think an overload set is as good as this gets. I'd be happy to learn about other solutions.

if (s.logical_type.has_value()) { c.field_struct(10, s.logical_type.value()); }
etseidl marked this conversation as resolved.
Show resolved Hide resolved
return c.value();
}

Expand Down Expand Up @@ -223,8 +213,8 @@ size_t CompactProtocolWriter::write(OffsetIndex const& s)
size_t CompactProtocolWriter::write(ColumnOrder const& co)
{
CompactProtocolFieldWriter c(*this);
switch (co) {
case ColumnOrder::TYPE_ORDER: c.field_empty_struct(1); break;
switch (co.type) {
case ColumnOrder::TYPE_ORDER: c.field_empty_struct(ColumnOrder::TYPE_ORDER); break;
default: break;
}
return c.value();
Expand Down
3 changes: 2 additions & 1 deletion cpp/src/io/parquet/page_decode.cuh
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1143,7 +1143,8 @@ inline __device__ bool setupLocalPageInfo(page_state_s* const s,
units = cudf::timestamp_ms::period::den;
} else if (s->col.converted_type == TIMESTAMP_MICROS) {
units = cudf::timestamp_us::period::den;
} else if (s->col.logical_type.TIMESTAMP.unit.isset.NANOS) {
} else if (s->col.logical_type.has_value() and
s->col.logical_type->is_timestamp_nanos()) {
units = cudf::timestamp_ns::period::den;
}
if (units and units != s->col.ts_clock_rate) {
Expand Down
Loading
Loading