From 32e21c9f8d465d66f8ba12536516ed5bf33dc05c Mon Sep 17 00:00:00 2001 From: yingsu00 Date: Thu, 21 Dec 2023 16:34:57 -0800 Subject: [PATCH] Fix reading MAP_KEY_VALUE Parquet SchemaElement (#7966) Summary: This PR fixes issue https://github.com/facebookincubator/velox/issues/7777 In Parquet, the map type is annotated as MAP converted type nomally. It should contain a repeated group annotated with MAP_KEY_VALUE, which in turn contains two children key and value: group (MAP) { repeated group key_value (MAP_KEY_VALUE) { required key; value; } } But sometimes a group annotated with MAP_KEY_VALUE was incorrectly used in place of MAP. group my_map (MAP_KEY_VALUE) { repeated group map { required binary key (UTF8); optional int32 value; } } For backward-compatibility, a MAP_KEY_VALUE that is not contained by MAP should be treated as MAP. This commit makes the following changes: 1. Adds a parentSchemaIdx to Parquet reader's getParquetColumnInfo() function to pass the parent schema. 2. Differenciate the situations where a MAP_KEY_VALUE's parent is or is not a MAP. If it is, then it should be the repeated group that contains the key and value. If it is not, it should be treated the same as MAP. For more information please check https://github.com/apache/parquet-format/blob/master/LogicalTypes.md#maps Pull Request resolved: https://github.com/facebookincubator/velox/pull/7966 Reviewed By: pedroerp Differential Revision: D52263936 Pulled By: Yuhta fbshipit-source-id: 486b6167c76e613c604b309c02b785834ab050ac --- velox/dwio/parquet/reader/ParquetReader.cpp | 81 ++++++++++++------ .../tests/examples/map_key_value.parquet | Bin 0 -> 580 bytes .../tests/reader/ParquetReaderTest.cpp | 53 ++++++++++++ 3 files changed, 106 insertions(+), 28 deletions(-) create mode 100644 velox/dwio/parquet/tests/examples/map_key_value.parquet diff --git a/velox/dwio/parquet/reader/ParquetReader.cpp b/velox/dwio/parquet/reader/ParquetReader.cpp index 07471ba15e24..92b767ef6a5c 100644 --- a/velox/dwio/parquet/reader/ParquetReader.cpp +++ b/velox/dwio/parquet/reader/ParquetReader.cpp @@ -94,6 +94,7 @@ class ReaderBase { uint32_t maxSchemaElementIdx, uint32_t maxRepeat, uint32_t maxDefine, + uint32_t parentSchemaIdx, uint32_t& schemaIdx, uint32_t& columnIdx) const; @@ -212,8 +213,11 @@ void ReaderBase::initializeSchema() { uint32_t schemaIdx = 0; uint32_t columnIdx = 0; uint32_t maxSchemaElementIdx = fileMetaData_->schema.size() - 1; + // Setting the parent schema index of the root("hive_schema") to be 0, which + // 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, schemaIdx, columnIdx); + maxSchemaElementIdx, maxRepeat, maxDefine, 0, schemaIdx, columnIdx); schema_ = createRowType( schemaWithId_->getChildren(), isFileColumnNamesReadAsLowerCase()); } @@ -222,6 +226,7 @@ std::shared_ptr ReaderBase::getParquetColumnInfo( uint32_t maxSchemaElementIdx, uint32_t maxRepeat, uint32_t maxDefine, + uint32_t parentSchemaIdx, uint32_t& schemaIdx, uint32_t& columnIdx) const { VELOX_CHECK(fileMetaData_ != nullptr); @@ -253,22 +258,60 @@ std::shared_ptr ReaderBase::getParquetColumnInfo( std::vector> children; + auto curSchemaIdx = schemaIdx; for (int32_t i = 0; i < schemaElement.num_children; i++) { auto child = getParquetColumnInfo( - maxSchemaElementIdx, maxRepeat, maxDefine, ++schemaIdx, columnIdx); + maxSchemaElementIdx, + maxRepeat, + maxDefine, + curSchemaIdx, + ++schemaIdx, + columnIdx); children.push_back(child); } VELOX_CHECK(!children.empty()); if (schemaElement.__isset.converted_type) { switch (schemaElement.converted_type) { + case thrift::ConvertedType::MAP_KEY_VALUE: + // If the MAP_KEY_VALUE annotated group's parent is a MAP, it should + // be the repeated key_value group that directly contains the key and + // value children. + if (schema[parentSchemaIdx].converted_type == + thrift::ConvertedType::MAP) { + VELOX_CHECK_EQ( + schemaElement.repetition_type, + thrift::FieldRepetitionType::REPEATED); + VELOX_CHECK_EQ(children.size(), 2); + + auto childrenCopy = children; + return std::make_shared( + TypeFactory::create( + children[0]->type(), children[1]->type()), + std::move(childrenCopy), + curSchemaIdx, // TODO: there are holes in the ids + maxSchemaElementIdx, + ParquetTypeWithId::kNonLeaf, // columnIdx, + std::move(name), + std::nullopt, + std::nullopt, + maxRepeat, + maxDefine); + } + + // For backward-compatibility, a group annotated with MAP_KEY_VALUE + // that is not contained by a MAP-annotated group should be handled as + // a MAP-annotated group. + FOLLY_FALLTHROUGH; + case thrift::ConvertedType::LIST: case thrift::ConvertedType::MAP: { - auto element = children.at(0)->getChildren(); VELOX_CHECK_EQ(children.size(), 1); + const auto& child = children[0]; + auto grandChildren = child->getChildren(); return std::make_shared( - children[0]->type(), - std::move(element), + child->type(), + std::move(grandChildren), curSchemaIdx, // TODO: there are holes in the ids maxSchemaElementIdx, ParquetTypeWithId::kNonLeaf, // columnIdx, @@ -278,30 +321,12 @@ std::shared_ptr ReaderBase::getParquetColumnInfo( maxRepeat + 1, maxDefine); } - case thrift::ConvertedType::MAP_KEY_VALUE: { - // child of MAP - VELOX_CHECK_EQ( - schemaElement.repetition_type, - thrift::FieldRepetitionType::REPEATED); - assert(children.size() == 2); - auto childrenCopy = children; - return std::make_shared( - TypeFactory::create( - children[0]->type(), children[1]->type()), - std::move(childrenCopy), - curSchemaIdx, // TODO: there are holes in the ids - maxSchemaElementIdx, - ParquetTypeWithId::kNonLeaf, // columnIdx, - std::move(name), - std::nullopt, - std::nullopt, - maxRepeat, - maxDefine); - } + default: - VELOX_UNSUPPORTED( - "Unsupported SchemaElement type: {}", - schemaElement.converted_type); + VELOX_UNREACHABLE( + "Invalid SchemaElement converted_type: {}, name: {}", + schemaElement.converted_type, + schemaElement.name); } } else { if (schemaElement.repetition_type == diff --git a/velox/dwio/parquet/tests/examples/map_key_value.parquet b/velox/dwio/parquet/tests/examples/map_key_value.parquet new file mode 100644 index 0000000000000000000000000000000000000000..ae0755cf0b25ff43c38afa5f7eb14a36e309a639 GIT binary patch literal 580 zcmWG=3^EjD5X}I>1^a$?amt7a0J&_UYzz$Y-5d;Hz?__vkdW~2IExDF42i}ij2#9n zf^G_jBwS`Ra0Ck^vMHTmeD!_(W)q-dpxzmbqG_V%e!p17L&z-21r35;#*UdYSh%O8 zwO+~zm?i7Q%+SNZZR-OxN>Y$XQi9O{h|NWrB*mZtP$vO-5*+Lh=P^K~WyCl{r!fK* zGDymZ&0~bJfs!#E+!>i=sqw|h8L7F6qD%~?OcE?5sl_FtOrk6drYsW7xrqfpmMD)Z zg9LMSYNaZJj||8&EUHWrtYwKgrKt>JOp-E^I%*s;qCAqS3~Z7XU;{uJK^jC^#012E zc28qe*fl0mI|i|POkzxG9AayjL>L$}yb2PF3QJQ name (): + // + // 0: REQUIRED BOOLEAN hive_schema (UTF8) + // 1: OPTIONAL BOOLEAN test (MAP_KEY_VALUE) + // 2: REPEATED BOOLEAN map (UTF8) + // 3: REQUIRED BYTE_ARRAY key (UTF8) + // 4: OPTIONAL INT64 value (UTF8) + + const std::string sample(getExampleFilePath("map_key_value.parquet")); + + facebook::velox::dwio::common::ReaderOptions readerOptions{leafPool_.get()}; + auto reader = createReader(sample, readerOptions); + EXPECT_EQ(reader->numberOfRows(), 1ULL); + + auto rowType = reader->typeWithId(); + EXPECT_EQ(rowType->type()->kind(), TypeKind::ROW); + EXPECT_EQ(rowType->size(), 1ULL); + + auto mapColumnType = rowType->childAt(0); + EXPECT_EQ(mapColumnType->type()->kind(), TypeKind::MAP); + + auto mapKeyType = mapColumnType->childAt(0); + EXPECT_EQ(mapKeyType->type()->kind(), TypeKind::VARCHAR); + + auto mapValueType = mapColumnType->childAt(1); + EXPECT_EQ(mapValueType->type()->kind(), TypeKind::BIGINT); + + auto fileSchema = + ROW({"test"}, {createType({VARCHAR(), BIGINT()})}); + auto rowReaderOpts = getReaderOpts(fileSchema); + auto scanSpec = makeScanSpec(fileSchema); + rowReaderOpts.setScanSpec(scanSpec); + auto rowReader = reader->createRowReader(rowReaderOpts); + + auto expected = makeRowVector({vectorMaker_.mapVector( + {{{"0", 0}, + {"1", 1}, + {"2", 2}, + {"3", 3}, + {"4", 4}, + {"5", 5}, + {"6", 6}, + {"7", 7}}})}); + + assertReadWithReaderAndExpected(fileSchema, *rowReader, expected, *leafPool_); +} + TEST_F(ParquetReaderTest, readSampleBigintRangeFilter) { // Read sample.parquet with the int filter "a BETWEEN 16 AND 20". FilterMap filters;