Add unit tests for legacy Parquet input/output wrappers

cpp/src/parquet/CMakeLists.txt

@@ -271,7 +271,6 @@ endif()
 
 add_subdirectory(api)
 add_subdirectory(arrow)
-add_subdirectory(util)
 
 arrow_install_all_headers("parquet")
 

cpp/src/parquet/arrow/reader.cc

@@ -47,9 +47,9 @@
 #include "parquet/file_reader.h"
 #include "parquet/metadata.h"
 #include "parquet/properties.h"
+#include "parquet/schema-internal.h"
 #include "parquet/schema.h"
 #include "parquet/types.h"
-#include "parquet/util/schema-util.h"
 
 using arrow::Array;
 using arrow::BooleanArray;
@@ -399,7 +399,7 @@ Status FileReader::Impl::GetReaderForNode(
     std::unique_ptr<ColumnReader::ColumnReaderImpl>* out) {
   *out = nullptr;
 
-  if (IsSimpleStruct(node)) {
+  if (schema::IsSimpleStruct(node)) {
     const schema::GroupNode* group = static_cast<const schema::GroupNode*>(node);
     std::vector<std::shared_ptr<ColumnReader::ColumnReaderImpl>> children;
     for (int i = 0; i < group->field_count(); i++) {
@@ -425,7 +425,7 @@ Status FileReader::Impl::GetReaderForNode(
     const Node* walker = node;
     while (!walker->is_primitive()) {
       DCHECK(walker->is_group());
-      auto group = static_cast<const GroupNode*>(walker);
+      auto group = static_cast<const schema::GroupNode*>(walker);
       if (group->field_count() != 1) {
         return Status::NotImplemented("lists with structs are not supported.");
       }
@@ -433,7 +433,7 @@ Status FileReader::Impl::GetReaderForNode(
     }
     auto column_index = reader_->metadata()->schema()->ColumnIndex(*walker);
 
-    // If the index of the column is found then a reader for the coliumn is needed.
+    // If the index of the column is found then a reader for the column is needed.
     // Otherwise *out keeps the nullptr value.
     if (std::find(indices.begin(), indices.end(), column_index) != indices.end()) {
       std::unique_ptr<ColumnReader> reader;
@@ -555,8 +555,8 @@ Status FileReader::Impl::ReadRowGroup(int row_group_index,
   // We only need to read schema fields which have columns indicated
   // in the indices vector
   std::vector<int> field_indices;
-  if (!ColumnIndicesToFieldIndices(*reader_->metadata()->schema(), indices,
-                                   &field_indices)) {
+  if (!schema::ColumnIndicesToFieldIndices(*reader_->metadata()->schema(), indices,
+                                           &field_indices)) {
     return Status::Invalid("Invalid column index");
   }
   int num_fields = static_cast<int>(field_indices.size());
@@ -612,8 +612,8 @@ Status FileReader::Impl::ReadTable(const std::vector<int>& indices,
   // We only need to read schema fields which have columns indicated
   // in the indices vector
   std::vector<int> field_indices;
-  if (!ColumnIndicesToFieldIndices(*reader_->metadata()->schema(), indices,
-                                   &field_indices)) {
+  if (!schema::ColumnIndicesToFieldIndices(*reader_->metadata()->schema(), indices,
+                                           &field_indices)) {
     return Status::Invalid("Invalid column index");
   }
 

cpp/src/parquet/arrow/schema.cc

@@ -30,8 +30,8 @@
 #include "parquet/arrow/writer.h"
 #include "parquet/exception.h"
 #include "parquet/properties.h"
+#include "parquet/schema-internal.h"
 #include "parquet/types.h"
-#include "parquet/util/schema-util.h"
 
 using arrow::Field;
 using arrow::Status;
@@ -255,7 +255,7 @@ Status NodeToList(const GroupNode& group,
       // Special case mentioned in the format spec:
       //   If the name is array or ends in _tuple, this should be a list of struct
       //   even for single child elements.
-      if (list_group.field_count() == 1 && !HasStructListName(list_group)) {
+      if (list_group.field_count() == 1 && !schema::HasStructListName(list_group)) {
         // List of primitive type
         std::shared_ptr<Field> item_field;
         RETURN_NOT_OK(

cpp/src/parquet/arrow/writer.h

@@ -46,7 +46,6 @@ class OutputStream;
 namespace parquet {
 
 class FileMetaData;
-class OutputStream;
 class ParquetFileWriter;
 
 namespace arrow {

cpp/src/parquet/deprecated_io-test.cc

@@ -15,6 +15,7 @@
 // specific language governing permissions and limitations
 // under the License.
 
+#include <algorithm>
 #include <cstdint>
 #include <cstdio>
 #include <memory>
@@ -23,6 +24,9 @@
 
 #include <gtest/gtest.h>
 
+#include "arrow/testing/gtest_util.h"
+
+#include "parquet/deprecated_io.h"
 #include "parquet/exception.h"
 #include "parquet/platform.h"
 #include "parquet/test-util.h"
@@ -30,4 +34,171 @@
 using arrow::default_memory_pool;
 using arrow::MemoryPool;
 
-namespace parquet {}  // namespace parquet
+namespace parquet {
+
+class MockRandomAccessSource : public RandomAccessSource {
+ public:
+  MockRandomAccessSource(const uint8_t* data, int64_t size)
+      : data_(data), position_(0), size_(size) {}
+
+  int64_t Size() const override { return size_; }
+
+  int64_t Read(int64_t nbytes, uint8_t* out) override {
+    ThrowIfClosed();
+    int64_t bytes_to_read = std::min(nbytes, size_ - position_);
+    if (bytes_to_read == 0) {
+      return 0;
+    }
+    memcpy(out, data_ + position_, bytes_to_read);
+    position_ += bytes_to_read;
+    return bytes_to_read;
+  }
+
+  std::shared_ptr<Buffer> Read(int64_t nbytes) override {
+    ThrowIfClosed();
+    int64_t bytes_to_read = std::min(nbytes, size_ - position_);
+    std::shared_ptr<ResizableBuffer> out =
+        AllocateBuffer(::arrow::default_memory_pool(), bytes_to_read);
+    Read(bytes_to_read, out->mutable_data());
+    return std::move(out);
+  }
+
+  std::shared_ptr<Buffer> ReadAt(int64_t position, int64_t nbytes) override {
+    ThrowIfClosed();
+    position_ = position;
+    return Read(nbytes);
+  }
+
+  int64_t ReadAt(int64_t position, int64_t nbytes, uint8_t* out) override {
+    ThrowIfClosed();
+    position_ = position;
+    return Read(nbytes, out);
+  }
+
+  void Close() override { closed_ = true; }
+
+  int64_t Tell() override {
+    ThrowIfClosed();
+    return position_;
+  }
+
+  bool closed() const { return closed_; }
+
+ private:
+  const uint8_t* data_;
+  int64_t position_;
+  int64_t size_;
+  bool closed_ = false;
+
+  void ThrowIfClosed() {
+    if (closed_) {
+      throw ParquetException("file is closed");
+    }
+  }
+};
+
+TEST(ParquetInputWrapper, BasicOperation) {
+  std::string data = "some example data";
+
+  auto source = std::unique_ptr<RandomAccessSource>(new MockRandomAccessSource(
+      reinterpret_cast<const uint8_t*>(data.data()), static_cast<int64_t>(data.size())));
+  ParquetInputWrapper wrapper(std::move(source));
+
+  ASSERT_FALSE(wrapper.closed());
+
+  int64_t position = -1;
+  ASSERT_OK(wrapper.Tell(&position));
+  ASSERT_EQ(0, position);
+
+  // Read into memory
+  uint8_t buf[4] = {0};
+  int64_t bytes_read = -1;
+  ASSERT_OK(wrapper.Read(4, &bytes_read, buf));
+  ASSERT_EQ(4, bytes_read);
+  ASSERT_EQ(0, memcmp(buf, data.data(), 4));
+
+  ASSERT_OK(wrapper.Tell(&position));
+  ASSERT_EQ(4, position);
+
+  // Seek
+  ASSERT_RAISES(NotImplemented, wrapper.Seek(5));
+
+  // Read buffer
+  std::shared_ptr<Buffer> buffer;
+  ASSERT_OK(wrapper.Read(7, &buffer));
+  ASSERT_EQ(0, memcmp(buffer->data(), data.data() + 4, 7));
+
+  // ReadAt
+  ASSERT_OK(wrapper.ReadAt(13, 4, &buffer));
+  ASSERT_EQ(4, buffer->size());
+  ASSERT_EQ(0, memcmp(buffer->data(), data.data() + 13, 4));
+
+  // GetSize
+  int64_t size = -1;
+  ASSERT_OK(wrapper.GetSize(&size));
+  ASSERT_EQ(static_cast<int64_t>(data.size()), size);
+
+  // Close
+  ASSERT_OK(wrapper.Close());
+  ASSERT_TRUE(wrapper.closed());
+}
+
+class MockOutputStream : public OutputStream {
+ public:
+  MockOutputStream() {}
+
+  void Write(const uint8_t* data, int64_t length) override {
+    ThrowIfClosed();
+    size_ += length;
+  }
+
+  void Close() override { closed_ = true; }
+
+  int64_t Tell() override {
+    ThrowIfClosed();
+    return size_;
+  }
+
+  bool closed() const { return closed_; }
+
+ private:
+  int64_t size_ = 0;
+  bool closed_ = false;
+
+  void ThrowIfClosed() {
+    if (closed_) {
+      throw ParquetException("file is closed");
+    }
+  }
+};
+
+TEST(ParquetOutputWrapper, BasicOperation) {
+  auto stream = std::unique_ptr<OutputStream>(new MockOutputStream);
+  ParquetOutputWrapper wrapper(std::move(stream));
+
+  int64_t position = -1;
+  ASSERT_OK(wrapper.Tell(&position));
+  ASSERT_EQ(0, position);
+
+  std::string data = "food";
+
+  ASSERT_OK(wrapper.Write(reinterpret_cast<const uint8_t*>(data.data()), 4));
+  ASSERT_OK(wrapper.Tell(&position));
+  ASSERT_EQ(4, position);
+
+  // Close
+  ASSERT_OK(wrapper.Close());
+  ASSERT_TRUE(wrapper.closed());
+
+  // Test catch exceptions
+  ASSERT_RAISES(IOError, wrapper.Tell(&position));
+  ASSERT_RAISES(IOError, wrapper.Write(reinterpret_cast<const uint8_t*>(data.data()), 4));
+}
+
+TEST(ParquetOutputWrapper, DtorCloses) {
+  MockOutputStream stream;
+  { ParquetOutputWrapper wrapper(&stream); }
+  ASSERT_TRUE(stream.closed());
+}
+
+}  // namespace parquet

cpp/src/parquet/deprecated_io.cc

@@ -40,7 +40,10 @@ ParquetInputWrapper::ParquetInputWrapper(RandomAccessSource* source)
 
 ParquetInputWrapper::~ParquetInputWrapper() {
   if (!closed_) {
-    source_->Close();
+    try {
+      source_->Close();
+    } catch (...) {
+    }
     closed_ = true;
   }
 }
@@ -101,7 +104,10 @@ ParquetOutputWrapper::ParquetOutputWrapper(::parquet::OutputStream* sink)
 
 ParquetOutputWrapper::~ParquetOutputWrapper() {
   if (!closed_) {
-    sink_->Close();
+    try {
+      sink_->Close();
+    } catch (...) {
+    }
     closed_ = true;
   }
 }

cpp/src/parquet/file_writer.h

@@ -165,6 +165,7 @@ class PARQUET_EXPORT ParquetFileWriter {
       const std::shared_ptr<WriterProperties>& properties = default_writer_properties(),
       const std::shared_ptr<const KeyValueMetadata>& key_value_metadata = NULLPTR);
 
+  ARROW_DEPRECATED("Use version with arrow::io::OutputStream")
   static std::unique_ptr<ParquetFileWriter> Open(
       const std::shared_ptr<OutputStream>& sink,
       const std::shared_ptr<schema::GroupNode>& schema,

cpp/src/parquet/schema-internal.h

@@ -23,6 +23,8 @@
 
 #include <cstdint>
 #include <memory>
+#include <string>
+#include <unordered_set>
 #include <vector>
 
 #include "parquet/platform.h"
@@ -37,6 +39,57 @@ class SchemaElement;
 
 namespace schema {
 
+inline bool str_endswith_tuple(const std::string& str) {
+  if (str.size() >= 6) {
+    return str.substr(str.size() - 6, 6) == "_tuple";
+  }
+  return false;
+}
+
+// Special case mentioned in the format spec:
+//   If the name is array or ends in _tuple, this should be a list of struct
+//   even for single child elements.
+inline bool HasStructListName(const GroupNode& node) {
+  return (node.name() == "array" || str_endswith_tuple(node.name()));
+}
+
+// TODO(itaiin): This aux. function is to be deleted once repeated structs are supported
+inline bool IsSimpleStruct(const Node* node) {
+  if (!node->is_group()) return false;
+  if (node->is_repeated()) return false;
+  if (node->logical_type() == LogicalType::LIST) return false;
+  // Special case mentioned in the format spec:
+  //   If the name is array or ends in _tuple, this should be a list of struct
+  //   even for single child elements.
+  auto group = static_cast<const GroupNode*>(node);
+  if (group->field_count() == 1 && HasStructListName(*group)) return false;
+
+  return true;
+}
+
+// Coalesce a list of schema fields indices which are the roots of the
+// columns referred by a list of column indices
+inline bool ColumnIndicesToFieldIndices(const SchemaDescriptor& descr,
+                                        const std::vector<int>& column_indices,
+                                        std::vector<int>* out) {
+  const GroupNode* group = descr.group_node();
+  std::unordered_set<int> already_added;
+  out->clear();
+  for (auto& column_idx : column_indices) {
+    auto field_node = descr.GetColumnRoot(column_idx);
+    auto field_idx = group->FieldIndex(*field_node);
+    if (field_idx < 0) {
+      return false;
+    }
+    auto insertion = already_added.insert(field_idx);
+    if (insertion.second) {
+      out->push_back(field_idx);
+    }
+  }
+
+  return true;
+}
+
 // ----------------------------------------------------------------------
 // Conversion from Parquet Thrift metadata
 

cpp/src/parquet/util/CMakeLists.txt

@@ -1,19 +0,0 @@
-# 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.
-
-# Headers: util
-arrow_install_all_headers("parquet/util")

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

@@ -1,87 +0,0 @@
-// 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.
-
-#ifndef PARQUET_SCHEMA_UTIL_H
-#define PARQUET_SCHEMA_UTIL_H
-
-#include <string>
-#include <unordered_set>
-#include <vector>
-
-#include "parquet/exception.h"
-#include "parquet/schema.h"
-#include "parquet/types.h"
-
-using parquet::LogicalType;
-using parquet::ParquetException;
-using parquet::SchemaDescriptor;
-using parquet::schema::GroupNode;
-using parquet::schema::Node;
-using parquet::schema::NodePtr;
-
-inline bool str_endswith_tuple(const std::string& str) {
-  if (str.size() >= 6) {
-    return str.substr(str.size() - 6, 6) == "_tuple";
-  }
-  return false;
-}
-
-// Special case mentioned in the format spec:
-//   If the name is array or ends in _tuple, this should be a list of struct
-//   even for single child elements.
-inline bool HasStructListName(const GroupNode& node) {
-  return (node.name() == "array" || str_endswith_tuple(node.name()));
-}
-
-// TODO(itaiin): This aux. function is to be deleted once repeated structs are supported
-inline bool IsSimpleStruct(const Node* node) {
-  if (!node->is_group()) return false;
-  if (node->is_repeated()) return false;
-  if (node->logical_type() == LogicalType::LIST) return false;
-  // Special case mentioned in the format spec:
-  //   If the name is array or ends in _tuple, this should be a list of struct
-  //   even for single child elements.
-  auto group = static_cast<const GroupNode*>(node);
-  if (group->field_count() == 1 && HasStructListName(*group)) return false;
-
-  return true;
-}
-
-// Coalesce a list of schema fields indices which are the roots of the
-// columns referred by a list of column indices
-inline bool ColumnIndicesToFieldIndices(const SchemaDescriptor& descr,
-                                        const std::vector<int>& column_indices,
-                                        std::vector<int>* out) {
-  const GroupNode* group = descr.group_node();
-  std::unordered_set<int> already_added;
-  out->clear();
-  for (auto& column_idx : column_indices) {
-    auto field_node = descr.GetColumnRoot(column_idx);
-    auto field_idx = group->FieldIndex(*field_node);
-    if (field_idx < 0) {
-      return false;
-    }
-    auto insertion = already_added.insert(field_idx);
-    if (insertion.second) {
-      out->push_back(field_idx);
-    }
-  }
-
-  return true;
-}
-
-#endif  // PARQUET_SCHEMA_UTIL_H