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.

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

ARROW-22: [C++] Convert flat Parquet schemas to Arrow schemas #41

Closed
wants to merge 7 commits into from
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
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
11 changes: 11 additions & 0 deletions cpp/CMakeLists.txt
Original file line number Diff line number Diff line change
Expand Up @@ -378,6 +378,16 @@ function(ADD_ARROW_TEST_DEPENDENCIES REL_TEST_NAME)
add_dependencies(${TEST_NAME} ${ARGN})
endfunction()

# A wrapper for target_link_libraries() that is compatible with NO_TESTS.
function(ARROW_TEST_LINK_LIBRARIES REL_TEST_NAME)
if(NO_TESTS)
return()
endif()
get_filename_component(TEST_NAME ${REL_TEST_NAME} NAME_WE)

target_link_libraries(${TEST_NAME} ${ARGN})
endfunction()

enable_testing()

############################################################
Expand Down Expand Up @@ -528,6 +538,7 @@ set(ARROW_SRCS
src/arrow/ipc/metadata-internal.cc

src/arrow/types/construct.cc
src/arrow/types/decimal.cc
src/arrow/types/json.cc
src/arrow/types/list.cc
src/arrow/types/primitive.cc
Expand Down
8 changes: 7 additions & 1 deletion cpp/src/arrow/parquet/CMakeLists.txt
Original file line number Diff line number Diff line change
Expand Up @@ -19,17 +19,23 @@
# arrow_parquet : Arrow <-> Parquet adapter

set(PARQUET_SRCS
schema.cc
)

set(PARQUET_LIBS
arrow
${PARQUET_SHARED_LIB}
)

add_library(arrow_parquet STATIC
add_library(arrow_parquet SHARED
${PARQUET_SRCS}
)
target_link_libraries(arrow_parquet ${PARQUET_LIBS})
SET_TARGET_PROPERTIES(arrow_parquet PROPERTIES LINKER_LANGUAGE CXX)

ADD_ARROW_TEST(parquet-schema-test)
ARROW_TEST_LINK_LIBRARIES(parquet-schema-test arrow_parquet)

# Headers: top level
install(FILES
DESTINATION include/arrow/parquet)
147 changes: 147 additions & 0 deletions cpp/src/arrow/parquet/parquet-schema-test.cc
Original file line number Diff line number Diff line change
@@ -0,0 +1,147 @@
// 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.

#include <memory>
#include <vector>

#include "gtest/gtest.h"

#include "arrow/test-util.h"
#include "arrow/type.h"
#include "arrow/util/status.h"

#include "arrow/parquet/schema.h"

namespace arrow {

namespace parquet {

using parquet_cpp::Repetition;
using parquet_cpp::schema::NodePtr;
using parquet_cpp::schema::GroupNode;
using parquet_cpp::schema::PrimitiveNode;

const auto BOOL = std::make_shared<BooleanType>();
const auto UINT8 = std::make_shared<UInt8Type>();
const auto INT32 = std::make_shared<Int32Type>();
const auto INT64 = std::make_shared<Int64Type>();
const auto FLOAT = std::make_shared<FloatType>();
const auto DOUBLE = std::make_shared<DoubleType>();
const auto UTF8 = std::make_shared<StringType>();
const auto BINARY = std::make_shared<ListType>(
std::make_shared<Field>("", UINT8));

class TestConvertParquetSchema : public ::testing::Test {
public:
virtual void SetUp() {}

void CheckFlatSchema(const std::shared_ptr<Schema>& expected_schema) {
ASSERT_EQ(expected_schema->num_fields(), result_schema_->num_fields());
for (int i = 0; i < expected_schema->num_fields(); ++i) {
auto lhs = result_schema_->field(i);
auto rhs = expected_schema->field(i);
EXPECT_TRUE(lhs->Equals(rhs))
<< i << " " << lhs->ToString() << " != " << rhs->ToString();
}
}

Status ConvertSchema(const std::vector<NodePtr>& nodes) {
NodePtr schema = GroupNode::Make("schema", Repetition::REPEATED, nodes);
descr_.Init(schema);
return FromParquetSchema(&descr_, &result_schema_);
}

protected:
parquet_cpp::SchemaDescriptor descr_;
std::shared_ptr<Schema> result_schema_;
};

TEST_F(TestConvertParquetSchema, ParquetFlatPrimitives) {
std::vector<NodePtr> parquet_fields;
std::vector<std::shared_ptr<Field>> arrow_fields;

parquet_fields.push_back(
PrimitiveNode::Make("boolean", Repetition::REQUIRED, parquet_cpp::Type::BOOLEAN));
arrow_fields.push_back(std::make_shared<Field>("boolean", BOOL, false));

parquet_fields.push_back(
PrimitiveNode::Make("int32", Repetition::REQUIRED, parquet_cpp::Type::INT32));
arrow_fields.push_back(std::make_shared<Field>("int32", INT32, false));

parquet_fields.push_back(
PrimitiveNode::Make("int64", Repetition::REQUIRED, parquet_cpp::Type::INT64));
arrow_fields.push_back(std::make_shared<Field>("int64", INT64, false));

parquet_fields.push_back(
PrimitiveNode::Make("float", Repetition::OPTIONAL, parquet_cpp::Type::FLOAT));
arrow_fields.push_back(std::make_shared<Field>("float", FLOAT));

parquet_fields.push_back(
PrimitiveNode::Make("double", Repetition::OPTIONAL, parquet_cpp::Type::DOUBLE));
arrow_fields.push_back(std::make_shared<Field>("double", DOUBLE));

parquet_fields.push_back(
PrimitiveNode::Make("binary", Repetition::OPTIONAL,
parquet_cpp::Type::BYTE_ARRAY));
arrow_fields.push_back(std::make_shared<Field>("binary", BINARY));

parquet_fields.push_back(
PrimitiveNode::Make("string", Repetition::OPTIONAL,
parquet_cpp::Type::BYTE_ARRAY,
parquet_cpp::LogicalType::UTF8));
arrow_fields.push_back(std::make_shared<Field>("string", UTF8));

parquet_fields.push_back(
PrimitiveNode::Make("flba-binary", Repetition::OPTIONAL,
parquet_cpp::Type::FIXED_LEN_BYTE_ARRAY,
parquet_cpp::LogicalType::NONE, 12));
arrow_fields.push_back(std::make_shared<Field>("flba-binary", BINARY));

auto arrow_schema = std::make_shared<Schema>(arrow_fields);
ASSERT_OK(ConvertSchema(parquet_fields));

CheckFlatSchema(arrow_schema);
}

TEST_F(TestConvertParquetSchema, UnsupportedThings) {
std::vector<NodePtr> unsupported_nodes;

unsupported_nodes.push_back(
PrimitiveNode::Make("int96", Repetition::REQUIRED, parquet_cpp::Type::INT96));

unsupported_nodes.push_back(
GroupNode::Make("repeated-group", Repetition::REPEATED, {}));

unsupported_nodes.push_back(
PrimitiveNode::Make("int32", Repetition::OPTIONAL,
parquet_cpp::Type::INT32, parquet_cpp::LogicalType::DATE));

unsupported_nodes.push_back(
PrimitiveNode::Make("int64", Repetition::OPTIONAL,
parquet_cpp::Type::INT64, parquet_cpp::LogicalType::TIMESTAMP_MILLIS));

for (const NodePtr& node : unsupported_nodes) {
ASSERT_RAISES(NotImplemented, ConvertSchema({node}));
}
}

TEST(TestNodeConversion, DateAndTime) {
}

} // namespace parquet

} // namespace arrow
178 changes: 178 additions & 0 deletions cpp/src/arrow/parquet/schema.cc
Original file line number Diff line number Diff line change
@@ -0,0 +1,178 @@
// 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.

#include "arrow/parquet/schema.h"

#include <vector>

#include "parquet/api/schema.h"

#include "arrow/util/status.h"
#include "arrow/types/decimal.h"

using parquet_cpp::schema::Node;
using parquet_cpp::schema::NodePtr;
using parquet_cpp::schema::GroupNode;
using parquet_cpp::schema::PrimitiveNode;

using parquet_cpp::LogicalType;

namespace arrow {

namespace parquet {

const auto BOOL = std::make_shared<BooleanType>();
const auto UINT8 = std::make_shared<UInt8Type>();
const auto INT32 = std::make_shared<Int32Type>();
const auto INT64 = std::make_shared<Int64Type>();
const auto FLOAT = std::make_shared<FloatType>();
const auto DOUBLE = std::make_shared<DoubleType>();
const auto UTF8 = std::make_shared<StringType>();
const auto BINARY = std::make_shared<ListType>(
std::make_shared<Field>("", UINT8));

TypePtr MakeDecimalType(const PrimitiveNode* node) {
int precision = node->decimal_metadata().precision;
int scale = node->decimal_metadata().scale;
return std::make_shared<DecimalType>(precision, scale);
}

static Status FromByteArray(const PrimitiveNode* node, TypePtr* out) {
switch (node->logical_type()) {
case LogicalType::UTF8:
*out = UTF8;
break;
default:
// BINARY
*out = BINARY;
break;
}
return Status::OK();
}

static Status FromFLBA(const PrimitiveNode* node, TypePtr* out) {
switch (node->logical_type()) {
case LogicalType::NONE:
*out = BINARY;
break;
case LogicalType::DECIMAL:
*out = MakeDecimalType(node);
break;
default:
return Status::NotImplemented("unhandled type");
break;
}

return Status::OK();
}

static Status FromInt32(const PrimitiveNode* node, TypePtr* out) {
switch (node->logical_type()) {
case LogicalType::NONE:
*out = INT32;
break;
default:
return Status::NotImplemented("Unhandled logical type for int32");
break;
}
return Status::OK();
}

static Status FromInt64(const PrimitiveNode* node, TypePtr* out) {
switch (node->logical_type()) {
case LogicalType::NONE:
*out = INT64;
break;
default:
return Status::NotImplemented("Unhandled logical type for int64");
break;
}
return Status::OK();
}

// TODO: Logical Type Handling
Status NodeToField(const NodePtr& node, std::shared_ptr<Field>* out) {
std::shared_ptr<DataType> type;

if (node->is_repeated()) {
return Status::NotImplemented("No support yet for repeated node types");
}

if (node->is_group()) {
const GroupNode* group = static_cast<const GroupNode*>(node.get());
std::vector<std::shared_ptr<Field>> fields(group->field_count());
for (int i = 0; i < group->field_count(); i++) {
RETURN_NOT_OK(NodeToField(group->field(i), &fields[i]));
}
type = std::make_shared<StructType>(fields);
} else {
// Primitive (leaf) node
const PrimitiveNode* primitive = static_cast<const PrimitiveNode*>(node.get());

switch (primitive->physical_type()) {
case parquet_cpp::Type::BOOLEAN:
type = BOOL;
break;
case parquet_cpp::Type::INT32:
RETURN_NOT_OK(FromInt32(primitive, &type));
break;
case parquet_cpp::Type::INT64:
RETURN_NOT_OK(FromInt64(primitive, &type));
break;
case parquet_cpp::Type::INT96:
// TODO: Do we have that type in Arrow?
// type = TypePtr(new Int96Type());
return Status::NotImplemented("int96");
case parquet_cpp::Type::FLOAT:
type = FLOAT;
break;
case parquet_cpp::Type::DOUBLE:
type = DOUBLE;
break;
case parquet_cpp::Type::BYTE_ARRAY:
// TODO: Do we have that type in Arrow?
RETURN_NOT_OK(FromByteArray(primitive, &type));
break;
case parquet_cpp::Type::FIXED_LEN_BYTE_ARRAY:
RETURN_NOT_OK(FromFLBA(primitive, &type));
break;
}
}

*out = std::make_shared<Field>(node->name(), type, !node->is_required());
return Status::OK();
}

Status FromParquetSchema(const parquet_cpp::SchemaDescriptor* parquet_schema,
std::shared_ptr<Schema>* out) {
// TODO(wesm): Consider adding an arrow::Schema name attribute, which comes
// from the root Parquet node
const GroupNode* schema_node = static_cast<const GroupNode*>(
parquet_schema->schema().get());

std::vector<std::shared_ptr<Field>> fields(schema_node->field_count());
for (int i = 0; i < schema_node->field_count(); i++) {
RETURN_NOT_OK(NodeToField(schema_node->field(i), &fields[i]));
}

*out = std::make_shared<Schema>(fields);
return Status::OK();
}

} // namespace parquet

} // namespace arrow
Loading