support to read binary as string in parquet (facebookincubator#10399)

Summary:
Some other Parquet-producing systems, in particular Impala, will write string-type data as binary into Parquet file. and Velox's ParquetReader does not support binaryAsString. this PR is intended to resolve this issue.

issue resolved:  [facebookincubator#10398]

Pull Request resolved: facebookincubator#10399

Reviewed By: pedroerp

Differential Revision: D60452520

Pulled By: Yuhta

fbshipit-source-id: b582117d0d64074bd56fdc52995d80d5eb489222

Author: kevincmchen

velox/dwio/parquet/reader/ParquetReader.cpp

@@ -99,9 +99,12 @@ class ReaderBase {
       uint32_t maxDefine,
       uint32_t parentSchemaIdx,
       uint32_t& schemaIdx,
-      uint32_t& columnIdx) const;
+      uint32_t& columnIdx,
+      const TypePtr& requestedType) const;
 
-  TypePtr convertType(const thrift::SchemaElement& schemaElement) const;
+  TypePtr convertType(
+      const thrift::SchemaElement& schemaElement,
+      const TypePtr& requestedType) const;
 
   template <typename T>
   static std::shared_ptr<const RowType> createRowType(
@@ -119,8 +122,6 @@ class ReaderBase {
   RowTypePtr schema_;
   std::shared_ptr<const dwio::common::TypeWithId> schemaWithId_;
 
-  const bool binaryAsString = false;
-
   // Map from row group index to pre-created loading BufferedInput.
   std::unordered_map<uint32_t, std::shared_ptr<dwio::common::BufferedInput>>
       inputs_;
@@ -220,7 +221,13 @@ void ReaderBase::initializeSchema() {
   // is the root itself. This is ok because it's never required to check the
   // parent of the root in getParquetColumnInfo().
   schemaWithId_ = getParquetColumnInfo(
-      maxSchemaElementIdx, maxRepeat, maxDefine, 0, schemaIdx, columnIdx);
+      maxSchemaElementIdx,
+      maxRepeat,
+      maxDefine,
+      0,
+      schemaIdx,
+      columnIdx,
+      options_.fileSchema());
   schema_ = createRowType(
       schemaWithId_->getChildren(), isFileColumnNamesReadAsLowerCase());
 }
@@ -231,7 +238,8 @@ std::unique_ptr<ParquetTypeWithId> ReaderBase::getParquetColumnInfo(
     uint32_t maxDefine,
     uint32_t parentSchemaIdx,
     uint32_t& schemaIdx,
-    uint32_t& columnIdx) const {
+    uint32_t& columnIdx,
+    const TypePtr& requestedType) const {
   VELOX_CHECK(fileMetaData_ != nullptr);
   VELOX_CHECK_LT(schemaIdx, fileMetaData_->schema.size());
 
@@ -265,18 +273,24 @@ std::unique_ptr<ParquetTypeWithId> ReaderBase::getParquetColumnInfo(
     VELOX_CHECK(
         schemaElement.__isset.num_children && schemaElement.num_children > 0,
         "Node has no children but should");
+    VELOX_CHECK(!requestedType || requestedType->isRow());
 
     std::vector<std::unique_ptr<ParquetTypeWithId::TypeWithId>> children;
 
     auto curSchemaIdx = schemaIdx;
     for (int32_t i = 0; i < schemaElement.num_children; i++) {
+      ++schemaIdx;
+      auto& childName = schema[schemaIdx].name;
+      auto childRequestedType =
+          requestedType ? requestedType->asRow().findChild(childName) : nullptr;
       auto child = getParquetColumnInfo(
           maxSchemaElementIdx,
           maxRepeat,
           maxDefine,
           curSchemaIdx,
-          ++schemaIdx,
-          columnIdx);
+          schemaIdx,
+          columnIdx,
+          childRequestedType);
       children.push_back(std::move(child));
     }
     VELOX_CHECK(!children.empty());
@@ -467,7 +481,7 @@ std::unique_ptr<ParquetTypeWithId> ReaderBase::getParquetColumnInfo(
       }
     }
   } else { // leaf node
-    const auto veloxType = convertType(schemaElement);
+    const auto veloxType = convertType(schemaElement, requestedType);
     int32_t precision =
         schemaElement.__isset.precision ? schemaElement.precision : 0;
     int32_t scale = schemaElement.__isset.scale ? schemaElement.scale : 0;
@@ -523,7 +537,8 @@ std::unique_ptr<ParquetTypeWithId> ReaderBase::getParquetColumnInfo(
 }
 
 TypePtr ReaderBase::convertType(
-    const thrift::SchemaElement& schemaElement) const {
+    const thrift::SchemaElement& schemaElement,
+    const TypePtr& requestedType) const {
   VELOX_CHECK(schemaElement.__isset.type && schemaElement.num_children == 0);
   VELOX_CHECK(
       schemaElement.type != thrift::Type::FIXED_LEN_BYTE_ARRAY ||
@@ -655,7 +670,7 @@ TypePtr ReaderBase::convertType(
         return DOUBLE();
       case thrift::Type::type::BYTE_ARRAY:
       case thrift::Type::type::FIXED_LEN_BYTE_ARRAY:
-        if (binaryAsString) {
+        if (requestedType && requestedType->isVarchar()) {
           return VARCHAR();
         } else {
           return VARBINARY();

velox/dwio/parquet/tests/reader/ParquetReaderTest.cpp

@@ -1150,6 +1150,36 @@ TEST_F(ParquetReaderTest, readVarbinaryFromFLBA) {
           0));
 }
 
+TEST_F(ParquetReaderTest, readBinaryAsStringFromNation) {
+  const std::string filename("nation.parquet");
+  const std::string sample(getExampleFilePath(filename));
+
+  dwio::common::ReaderOptions readerOptions{leafPool_.get()};
+  auto outputRowType =
+      ROW({"nationkey", "name", "regionkey", "comment"},
+          {BIGINT(), VARCHAR(), BIGINT(), VARCHAR()});
+
+  readerOptions.setFileSchema(outputRowType);
+  auto reader = createReader(sample, readerOptions);
+  EXPECT_EQ(reader->numberOfRows(), 25ULL);
+  auto rowType = reader->typeWithId();
+  EXPECT_EQ(rowType->type()->kind(), TypeKind::ROW);
+  EXPECT_EQ(rowType->size(), 4ULL);
+  EXPECT_EQ(rowType->childAt(1)->type()->kind(), TypeKind::VARCHAR);
+
+  auto rowReaderOpts = getReaderOpts(outputRowType);
+  rowReaderOpts.setScanSpec(makeScanSpec(outputRowType));
+  auto rowReader = reader->createRowReader(rowReaderOpts);
+
+  auto expected = std::string("ALGERIA");
+  VectorPtr result = BaseVector::create(outputRowType, 0, &(*leafPool_));
+  rowReader->next(1, result);
+  EXPECT_EQ(
+      expected,
+      result->as<RowVector>()->childAt(1)->asFlatVector<StringView>()->valueAt(
+          0));
+}
+
 TEST_F(ParquetReaderTest, testV2PageWithZeroMaxDefRep) {
   // enum_type.parquet contains 1 column (ENUM) with 3 rows.
   const std::string sample(getExampleFilePath("v2_page.parquet"));