PARQUET-583: Parquet to Thrift schema conversion

Depends on apache#86

Author: Uwe L. Korn <uwelk@xhochy.com>

Closes apache#87 from xhochy/parquet-583 and squashes the following commits:

9f3f050 [Uwe L. Korn] Incoperate feedback
86aed44 [Uwe L. Korn] PARQUET-583: Parquet to Thrift schema conversion

Change-Id: I0b255221ad1c5f47db6842b0141da27dd649ff8e

cpp/src/parquet/schema/converter.cc

@@ -83,6 +83,45 @@ std::shared_ptr<SchemaDescriptor> FromParquet(const std::vector<SchemaElement>&
   return descr;
 }
 
+void ToParquet(const GroupNode* schema, std::vector<format::SchemaElement>* out) {
+  SchemaFlattener flattener(schema, out);
+  flattener.Flatten();
+}
+
+class SchemaVisitor : public Node::ConstVisitor {
+ public:
+  explicit SchemaVisitor(std::vector<format::SchemaElement>* elements)
+     : elements_(elements) {}
+  virtual ~SchemaVisitor() {}
+
+  void Visit(const Node* node) override {
+    format::SchemaElement element;
+    node->ToParquet(&element);
+    // Override field_id here as we can get user-generated Nodes without a valid id
+    element.__set_field_id(elements_->size());
+    elements_->push_back(element);
+
+    if (node->is_group()) {
+      const GroupNode* group_node = static_cast<const GroupNode*>(node);
+      for (int i = 0; i < group_node->field_count(); ++i) {
+        group_node->field(i)->VisitConst(this);
+      }
+    }
+  }
+
+ private:
+  std::vector<format::SchemaElement>* elements_;
+};
+
+SchemaFlattener::SchemaFlattener(const GroupNode* schema,
+        std::vector<format::SchemaElement>* out)
+    : root_(schema), elements_(out) {}
+
+void SchemaFlattener::Flatten() {
+  SchemaVisitor visitor(elements_);
+  root_->VisitConst(&visitor);
+}
+
 } // namespace schema
 
 } // namespace parquet

cpp/src/parquet/schema/converter.h

@@ -80,9 +80,11 @@ class SchemaFlattener {
  public:
   SchemaFlattener(const GroupNode* schema, std::vector<format::SchemaElement>* out);
 
+  void Flatten();
+
  private:
   const GroupNode* root_;
-  std::vector<format::SchemaElement>* schema_;
+  std::vector<format::SchemaElement>* elements_;
 };
 
 } // namespace schema

cpp/src/parquet/schema/printer.cc

@@ -27,14 +27,14 @@ namespace parquet {
 
 namespace schema {
 
-class SchemaPrinter : public Node::Visitor {
+class SchemaPrinter : public Node::ConstVisitor {
  public:
   explicit SchemaPrinter(std::ostream& stream, int indent_width) :
       stream_(stream),
       indent_(0),
       indent_width_(2) {}
 
-  virtual void Visit(const Node* node);
+  void Visit(const Node* node) override;
 
  private:
   void Visit(const PrimitiveNode* node);
@@ -108,7 +108,7 @@ void SchemaPrinter::Visit(const GroupNode* node) {
 
   indent_ += indent_width_;
   for (int i = 0; i < node->field_count(); ++i) {
-    node->field(i)->Visit(this);
+    node->field(i)->VisitConst(this);
   }
   indent_ -= indent_width_;
   Indent();

cpp/src/parquet/schema/schema-converter-test.cc

@@ -82,21 +82,21 @@ bool check_for_parent_consistency(const GroupNode* node) {
 TEST_F(TestSchemaConverter, NestedExample) {
   SchemaElement elt;
   std::vector<SchemaElement> elements;
-  elements.push_back(NewGroup(name_, FieldRepetitionType::REPEATED, 2));
+  elements.push_back(NewGroup(name_, FieldRepetitionType::REPEATED, 2, 0));
 
   // A primitive one
   elements.push_back(NewPrimitive("a", FieldRepetitionType::REQUIRED,
-          format::Type::INT32));
+          format::Type::INT32, 1));
 
   // A group
-  elements.push_back(NewGroup("bag", FieldRepetitionType::OPTIONAL, 1));
+  elements.push_back(NewGroup("bag", FieldRepetitionType::OPTIONAL, 1, 2));
 
   // 3-level list encoding, by hand
-  elt = NewGroup("b", FieldRepetitionType::REPEATED, 1);
+  elt = NewGroup("b", FieldRepetitionType::REPEATED, 1, 3);
   elt.__set_converted_type(ConvertedType::LIST);
   elements.push_back(elt);
   elements.push_back(NewPrimitive("item", FieldRepetitionType::OPTIONAL,
-          format::Type::INT64));
+          format::Type::INT64, 4));
 
   Convert(&elements[0], elements.size());
 
@@ -127,33 +127,86 @@ TEST_F(TestSchemaConverter, InvalidRoot) {
 
   SchemaElement elements[2];
   elements[0] = NewPrimitive("not-a-group", FieldRepetitionType::REQUIRED,
-      format::Type::INT32);
+      format::Type::INT32, 0);
   ASSERT_THROW(Convert(elements, 2), ParquetException);
 
   // While the Parquet spec indicates that the root group should have REPEATED
   // repetition type, some implementations may return REQUIRED or OPTIONAL
   // groups as the first element. These tests check that this is okay as a
   // practicality matter.
-  elements[0] = NewGroup("not-repeated", FieldRepetitionType::REQUIRED, 1);
+  elements[0] = NewGroup("not-repeated", FieldRepetitionType::REQUIRED, 1, 0);
   elements[1] = NewPrimitive("a", FieldRepetitionType::REQUIRED,
-      format::Type::INT32);
+      format::Type::INT32, 1);
   Convert(elements, 2);
 
-  elements[0] = NewGroup("not-repeated", FieldRepetitionType::OPTIONAL, 1);
+  elements[0] = NewGroup("not-repeated", FieldRepetitionType::OPTIONAL, 1, 0);
   Convert(elements, 2);
 }
 
 TEST_F(TestSchemaConverter, NotEnoughChildren) {
   // Throw a ParquetException, but don't core dump or anything
   SchemaElement elt;
   std::vector<SchemaElement> elements;
-  elements.push_back(NewGroup(name_, FieldRepetitionType::REPEATED, 2));
+  elements.push_back(NewGroup(name_, FieldRepetitionType::REPEATED, 2, 0));
   ASSERT_THROW(Convert(&elements[0], 1), ParquetException);
 }
 
 // ----------------------------------------------------------------------
 // Schema tree flatten / unflatten
 
+class TestSchemaFlatten : public ::testing::Test {
+ public:
+  void setUp() {
+    name_ = "parquet_schema";
+  }
+
+  void Flatten(const GroupNode* schema) {
+    ToParquet(schema, &elements_);
+  }
+
+ protected:
+  std::string name_;
+  std::vector<format::SchemaElement> elements_;
+};
+
+TEST_F(TestSchemaFlatten, NestedExample) {
+  SchemaElement elt;
+  std::vector<SchemaElement> elements;
+  elements.push_back(NewGroup(name_, FieldRepetitionType::REPEATED, 2, 0));
+
+  // A primitive one
+  elements.push_back(NewPrimitive("a", FieldRepetitionType::REQUIRED,
+          format::Type::INT32, 1));
+
+  // A group
+  elements.push_back(NewGroup("bag", FieldRepetitionType::OPTIONAL, 1, 2));
+
+  // 3-level list encoding, by hand
+  elt = NewGroup("b", FieldRepetitionType::REPEATED, 1, 3);
+  elt.__set_converted_type(ConvertedType::LIST);
+  elements.push_back(elt);
+  elements.push_back(NewPrimitive("item", FieldRepetitionType::OPTIONAL,
+          format::Type::INT64, 4));
+
+  // Construct the schema
+  NodeVector fields;
+  fields.push_back(Int32("a", Repetition::REQUIRED));
+
+  // 3-level list encoding
+  NodePtr item = Int64("item");
+  NodePtr list(GroupNode::Make("b", Repetition::REPEATED, {item}, LogicalType::LIST));
+  NodePtr bag(GroupNode::Make("bag", Repetition::OPTIONAL, {list}));
+  fields.push_back(bag);
+
+  NodePtr schema = GroupNode::Make(name_, Repetition::REPEATED, fields);
+
+  Flatten(static_cast<GroupNode*>(schema.get()));
+  ASSERT_EQ(elements_.size(), elements.size());
+  for (size_t i = 0; i < elements_.size(); i++) {
+    ASSERT_EQ(elements_[i], elements[i]);
+  }
+}
+
 } // namespace schema
 
 } // namespace parquet

cpp/src/parquet/schema/schema-types-test.cc

@@ -112,7 +112,7 @@ TEST_F(TestPrimitiveNode, Attrs) {
 
 TEST_F(TestPrimitiveNode, FromParquet) {
   SchemaElement elt = NewPrimitive(name_, FieldRepetitionType::OPTIONAL,
-      format::Type::INT32);
+      format::Type::INT32, 0);
   Convert(&elt);
   ASSERT_EQ(name_, prim_node_->name());
   ASSERT_EQ(id_, prim_node_->id());
@@ -121,7 +121,7 @@ TEST_F(TestPrimitiveNode, FromParquet) {
   ASSERT_EQ(LogicalType::NONE, prim_node_->logical_type());
 
   // Test a logical type
-  elt = NewPrimitive(name_, FieldRepetitionType::REQUIRED, format::Type::BYTE_ARRAY);
+  elt = NewPrimitive(name_, FieldRepetitionType::REQUIRED, format::Type::BYTE_ARRAY, 0);
   elt.__set_converted_type(ConvertedType::UTF8);
 
   Convert(&elt);
@@ -131,7 +131,7 @@ TEST_F(TestPrimitiveNode, FromParquet) {
 
   // FIXED_LEN_BYTE_ARRAY
   elt = NewPrimitive(name_, FieldRepetitionType::OPTIONAL,
-      format::Type::FIXED_LEN_BYTE_ARRAY);
+      format::Type::FIXED_LEN_BYTE_ARRAY, 0);
   elt.__set_type_length(16);
 
   Convert(&elt);
@@ -143,7 +143,7 @@ TEST_F(TestPrimitiveNode, FromParquet) {
 
   // ConvertedType::Decimal
   elt = NewPrimitive(name_, FieldRepetitionType::OPTIONAL,
-      format::Type::FIXED_LEN_BYTE_ARRAY);
+      format::Type::FIXED_LEN_BYTE_ARRAY, 0);
   elt.__set_converted_type(ConvertedType::DECIMAL);
   elt.__set_type_length(6);
   elt.__set_scale(2);

cpp/src/parquet/schema/test-util.h

@@ -36,22 +36,28 @@ namespace parquet {
 namespace schema {
 
 static inline SchemaElement NewPrimitive(const std::string& name,
-    FieldRepetitionType::type repetition, format::Type::type type) {
+    FieldRepetitionType::type repetition, format::Type::type type, int id = 0) {
   SchemaElement result;
   result.__set_name(name);
   result.__set_repetition_type(repetition);
   result.__set_type(type);
   result.__set_num_children(0);
+  result.__set_field_id(id);
+  // Set default (non-set) values
+  result.__set_type_length(-1);
+  result.__set_precision(-1);
+  result.__set_scale(-1);
 
   return result;
 }
 
 static inline SchemaElement NewGroup(const std::string& name,
-    FieldRepetitionType::type repetition, int num_children) {
+    FieldRepetitionType::type repetition, int num_children, int id = 0) {
   SchemaElement result;
   result.__set_name(name);
   result.__set_repetition_type(repetition);
   result.__set_num_children(num_children);
+  result.__set_field_id(id);
 
   return result;
 }

cpp/src/parquet/schema/types.cc

@@ -203,6 +203,10 @@ void PrimitiveNode::Visit(Node::Visitor* visitor) {
   visitor->Visit(this);
 }
 
+void PrimitiveNode::VisitConst(Node::ConstVisitor* visitor) const {
+  visitor->Visit(this);
+}
+
 // ----------------------------------------------------------------------
 // Group node
 
@@ -232,6 +236,10 @@ void GroupNode::Visit(Node::Visitor* visitor) {
   visitor->Visit(this);
 }
 
+void GroupNode::VisitConst(Node::ConstVisitor* visitor) const {
+  visitor->Visit(this);
+}
+
 // ----------------------------------------------------------------------
 // Node construction from Parquet metadata
 
@@ -280,6 +288,35 @@ std::unique_ptr<Node> PrimitiveNode::FromParquet(const void* opaque_element,
   return std::unique_ptr<Node>(result.release());
 }
 
+void GroupNode::ToParquet(void* opaque_element) const {
+  format::SchemaElement* element =
+    static_cast<format::SchemaElement*>(opaque_element);
+  element->__set_name(name_);
+  element->__set_num_children(field_count());
+  element->__set_repetition_type(ToThrift(repetition_));
+  if (logical_type_ != LogicalType::NONE) {
+    element->__set_converted_type(ToThrift(logical_type_));
+  }
+  // FIXME: SchemaFlattener does this for us: element->__set_field_id(id_);
+}
+
+void PrimitiveNode::ToParquet(void* opaque_element) const {
+  format::SchemaElement* element =
+    static_cast<format::SchemaElement*>(opaque_element);
+
+  element->__set_name(name_);
+  element->__set_num_children(0);
+  element->__set_repetition_type(ToThrift(repetition_));
+  if (logical_type_ != LogicalType::NONE) {
+    element->__set_converted_type(ToThrift(logical_type_));
+  }
+  element->__set_type(ToThrift(physical_type_));
+  // FIXME: SchemaFlattener does this for us: element->__set_field_id(id_);
+  element->__set_type_length(type_length_);
+  element->__set_precision(decimal_metadata_.precision);
+  element->__set_scale(decimal_metadata_.scale);
+}
+
 } // namespace schema
 
 } // namespace parquet

cpp/src/parquet/schema/types.h

@@ -163,15 +163,26 @@ class Node {
     return parent_;
   }
 
+  // ToParquet returns an opaque void* to avoid exporting
+  // parquet::SchemaElement into the public API
+  virtual void ToParquet(void* opaque_element) const = 0;
+
   // Node::Visitor abstract class for walking schemas with the visitor pattern
   class Visitor {
    public:
     virtual ~Visitor() {}
 
+    virtual void Visit(Node* node) = 0;
+  };
+  class ConstVisitor {
+   public:
+    virtual ~ConstVisitor() {}
+
     virtual void Visit(const Node* node) = 0;
   };
 
   virtual void Visit(Visitor* visitor) = 0;
+  virtual void VisitConst(ConstVisitor* visitor) const = 0;
 
  protected:
   friend class GroupNode;
@@ -224,7 +235,9 @@ class PrimitiveNode : public Node {
     return decimal_metadata_;
   }
 
+  void ToParquet(void* opaque_element) const override;
   virtual void Visit(Visitor* visitor);
+  void VisitConst(ConstVisitor* visitor) const override;
 
  private:
   PrimitiveNode(const std::string& name, Repetition::type repetition,
@@ -278,7 +291,9 @@ class GroupNode : public Node {
     return fields_.size();
   }
 
+  void ToParquet(void* opaque_element) const override;
   virtual void Visit(Visitor* visitor);
+  void VisitConst(ConstVisitor* visitor) const override;
 
  private:
   GroupNode(const std::string& name, Repetition::type repetition,