Skip to content

Commit 696937a

Browse files
lxy-9602lxy-9602
committed
add more case for review
1 parent d44a2a5 commit 696937a

2 files changed

Lines changed: 192 additions & 0 deletions

File tree

‎test/inte/append_compaction_inte_test.cpp‎

Lines changed: 89 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -403,6 +403,95 @@ TEST_P(AppendCompactionInteTest, TestAppendTableStreamWriteFullCompactionWithMap
403403
}
404404
}
405405

406+
TEST_P(AppendCompactionInteTest,
407+
TestOrcAppendTableFullCompactionWithMapSharedShreddingStringValue) {
408+
auto file_format = GetParam();
409+
if (file_format != "orc") {
410+
return;
411+
}
412+
413+
auto dir = UniqueTestDirectory::Create();
414+
ASSERT_TRUE(dir);
415+
auto map_type = arrow::map(arrow::utf8(), arrow::utf8());
416+
arrow::FieldVector fields = {
417+
arrow::field("id", arrow::int32()),
418+
arrow::field("tags", map_type),
419+
};
420+
auto schema = arrow::schema(fields);
421+
422+
std::map<std::string, std::string> options = {
423+
{Options::FILE_FORMAT, "orc"},
424+
{Options::BUCKET, "1"},
425+
{Options::BUCKET_KEY, "id"},
426+
{Options::FILE_SYSTEM, "local"},
427+
{"orc.read.enable-lazy-decoding", "true"},
428+
{"orc.dictionary-key-size-threshold", "1"},
429+
{"fields.tags.map.storage-layout", "shared-shredding"},
430+
{"fields.tags.map.shared-shredding.max-columns", "1"},
431+
};
432+
ASSERT_OK_AND_ASSIGN(auto helper, TestHelper::Create(dir->Str(), schema, /*partition_keys=*/{},
433+
/*primary_keys=*/{}, options,
434+
/*is_streaming_mode=*/true));
435+
436+
int64_t commit_identifier = 0;
437+
ASSERT_OK_AND_ASSIGN(auto batch_0,
438+
TestHelper::MakeRecordBatch(arrow::struct_(fields),
439+
R"([
440+
[1, [["a", "shared"], ["b", "hot"]]],
441+
[2, [["c", "shared"]]]
442+
])",
443+
/*partition_map=*/{}, /*bucket=*/0, {}));
444+
ASSERT_OK(helper->WriteAndCommit(std::move(batch_0), commit_identifier++,
445+
/*expected_commit_messages=*/std::nullopt));
446+
447+
ASSERT_OK_AND_ASSIGN(auto batch_1,
448+
TestHelper::MakeRecordBatch(arrow::struct_(fields),
449+
R"([
450+
[3, [["a", "shared"], ["d", "hot"]]],
451+
[4, null]
452+
])",
453+
/*partition_map=*/{}, /*bucket=*/0, {}));
454+
ASSERT_OK(helper->WriteAndCommit(std::move(batch_1), commit_identifier++,
455+
/*expected_commit_messages=*/std::nullopt));
456+
457+
ASSERT_OK_AND_ASSIGN(auto batch_2,
458+
TestHelper::MakeRecordBatch(arrow::struct_(fields),
459+
R"([
460+
[5, [["e", "shared"], ["f", "hot"], ["g", "shared"]]]
461+
])",
462+
/*partition_map=*/{}, /*bucket=*/0, {}));
463+
ASSERT_OK(helper->WriteAndCommit(std::move(batch_2), commit_identifier++,
464+
/*expected_commit_messages=*/std::nullopt));
465+
466+
ASSERT_OK(helper->write_->Compact(/*partition=*/{}, /*bucket=*/0,
467+
/*full_compaction=*/true));
468+
ASSERT_OK_AND_ASSIGN(
469+
std::vector<std::shared_ptr<CommitMessage>> commit_messages,
470+
helper->write_->PrepareCommit(/*wait_compaction=*/true, commit_identifier));
471+
ASSERT_FALSE(commit_messages.empty());
472+
ASSERT_OK(helper->commit_->Commit(commit_messages, commit_identifier));
473+
ASSERT_OK_AND_ASSIGN(std::optional<Snapshot> snapshot, helper->LatestSnapshot());
474+
ASSERT_TRUE(snapshot);
475+
ASSERT_EQ(Snapshot::CommitKind::Compact(), snapshot.value().GetCommitKind());
476+
477+
ASSERT_OK_AND_ASSIGN(std::vector<std::shared_ptr<Split>> data_splits,
478+
helper->NewScan(StartupMode::LatestFull(), /*snapshot_id=*/std::nullopt));
479+
ASSERT_EQ(data_splits.size(), 1);
480+
arrow::FieldVector fields_with_row_kind = fields;
481+
fields_with_row_kind.insert(fields_with_row_kind.begin(),
482+
arrow::field("_VALUE_KIND", arrow::int8()));
483+
auto data_type = arrow::struct_(fields_with_row_kind);
484+
ASSERT_OK_AND_ASSIGN(bool success, helper->ReadAndCheckResult(data_type, data_splits,
485+
R"([
486+
[0, 1, [["a", "shared"], ["b", "hot"]]],
487+
[0, 2, [["c", "shared"]]],
488+
[0, 3, [["a", "shared"], ["d", "hot"]]],
489+
[0, 4, null],
490+
[0, 5, [["e", "shared"], ["f", "hot"], ["g", "shared"]]]
491+
])"));
492+
ASSERT_TRUE(success);
493+
}
494+
406495
TEST_P(AppendCompactionInteTest, TestAppendTableStreamWriteFullCompactionWithDv) {
407496
auto dir = UniqueTestDirectory::Create();
408497
ASSERT_TRUE(dir);

‎test/inte/write_and_read_inte_test.cpp‎

Lines changed: 103 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2116,6 +2116,109 @@ TEST_P(WriteAndReadInteTest, TestMapStorageLayoutSharedShreddingToDefault) {
21162116
ASSERT_TRUE(success);
21172117
}
21182118

2119+
TEST_P(WriteAndReadInteTest, TestAppendMapStorageLayoutSharedShreddingToDefaultCompaction) {
2120+
auto [file_format, file_system] = GetParam();
2121+
if (file_format != "parquet" && file_format != "orc") {
2122+
return;
2123+
}
2124+
2125+
arrow::FieldVector fields = {
2126+
arrow::field("id", arrow::int32()),
2127+
arrow::field("tags", arrow::map(arrow::utf8(), arrow::int64())),
2128+
};
2129+
std::map<std::string, std::string> options_v0 = {
2130+
{Options::MANIFEST_FORMAT, "avro"},
2131+
{Options::FILE_FORMAT, file_format},
2132+
{Options::TARGET_FILE_SIZE, "1024"},
2133+
{Options::BUCKET, "1"},
2134+
{Options::BUCKET_KEY, "id"},
2135+
{Options::FILE_SYSTEM, file_system},
2136+
{"fields.tags.map.storage-layout", "shared-shredding"},
2137+
{"fields.tags.map.shared-shredding.max-columns", "1"},
2138+
};
2139+
if (file_system == "jindo") {
2140+
options_v0 = AddOptionsForJindo(options_v0);
2141+
}
2142+
ASSERT_OK_AND_ASSIGN(auto helper,
2143+
TestHelper::Create(test_dir_, arrow::schema(fields),
2144+
/*partition_keys=*/{}, /*primary_keys=*/{}, options_v0,
2145+
/*is_streaming_mode=*/true));
2146+
std::string table_path = PathUtil::JoinPath(test_dir_, "foo.db/bar");
2147+
int64_t commit_identifier = 0;
2148+
2149+
ASSERT_OK_AND_ASSIGN(auto batch_v0_file1,
2150+
TestHelper::MakeRecordBatch(arrow::struct_(fields),
2151+
R"([
2152+
[1, [["a", 10], ["b", 11]]],
2153+
[2, [["c", 20]]]
2154+
])",
2155+
/*partition_map=*/{}, /*bucket=*/0, {}));
2156+
ASSERT_OK(helper->WriteAndCommit(std::move(batch_v0_file1), commit_identifier++,
2157+
/*expected_commit_messages=*/std::nullopt));
2158+
2159+
ASSERT_OK_AND_ASSIGN(auto batch_v0_file2,
2160+
TestHelper::MakeRecordBatch(arrow::struct_(fields),
2161+
R"([
2162+
[3, [["d", 30], ["e", 31]]]
2163+
])",
2164+
/*partition_map=*/{}, /*bucket=*/0, {}));
2165+
ASSERT_OK(helper->WriteAndCommit(std::move(batch_v0_file2), commit_identifier++,
2166+
/*expected_commit_messages=*/std::nullopt));
2167+
2168+
std::map<std::string, std::string> options_v1 = options_v0;
2169+
options_v1["fields.tags.map.storage-layout"] = "default";
2170+
options_v1.erase("fields.tags.map.shared-shredding.max-columns");
2171+
ASSERT_OK(WriteNextSchema({DataField(0, fields[0]), DataField(1, fields[1])},
2172+
/*highest_field_id=*/1, options_v1));
2173+
2174+
helper.reset();
2175+
ASSERT_OK_AND_ASSIGN(helper, TestHelper::Create(table_path, options_v1,
2176+
/*is_streaming_mode=*/true));
2177+
ASSERT_OK_AND_ASSIGN(auto batch_v1_file3,
2178+
TestHelper::MakeRecordBatch(arrow::struct_(fields),
2179+
R"([
2180+
[4, [["a", 40], ["f", 41]]],
2181+
[5, null]
2182+
])",
2183+
/*partition_map=*/{}, /*bucket=*/0, {}));
2184+
ASSERT_OK(helper->WriteAndCommit(std::move(batch_v1_file3), commit_identifier++,
2185+
/*expected_commit_messages=*/std::nullopt));
2186+
2187+
WriteContextBuilder write_context_builder(table_path, "commit_user");
2188+
ASSERT_OK_AND_ASSIGN(
2189+
auto write_context,
2190+
write_context_builder.SetOptions(options_v1).WithStreamingMode(true).Finish());
2191+
ASSERT_OK_AND_ASSIGN(auto file_store_write, FileStoreWrite::Create(std::move(write_context)));
2192+
ASSERT_OK(file_store_write->Compact(/*partition=*/{}, /*bucket=*/0,
2193+
/*full_compaction=*/true));
2194+
ASSERT_OK_AND_ASSIGN(auto compact_messages, file_store_write->PrepareCommit(
2195+
/*wait_compaction=*/true, commit_identifier));
2196+
ASSERT_FALSE(compact_messages.empty());
2197+
2198+
CommitContextBuilder commit_context_builder(table_path, "commit_user");
2199+
ASSERT_OK_AND_ASSIGN(auto commit_context,
2200+
commit_context_builder.SetOptions(options_v1).Finish());
2201+
ASSERT_OK_AND_ASSIGN(auto file_store_commit,
2202+
FileStoreCommit::Create(std::move(commit_context)));
2203+
ASSERT_OK(file_store_commit->Commit(compact_messages, commit_identifier));
2204+
2205+
arrow::FieldVector expected_fields = fields;
2206+
expected_fields.insert(expected_fields.begin(), arrow::field("_VALUE_KIND", arrow::int8()));
2207+
ASSERT_OK_AND_ASSIGN(auto splits,
2208+
helper->NewScan(StartupMode::LatestFull(), /*snapshot_id=*/std::nullopt));
2209+
ASSERT_EQ(1, splits.size());
2210+
ASSERT_OK_AND_ASSIGN(bool success,
2211+
helper->ReadAndCheckResult(arrow::struct_(expected_fields), splits,
2212+
R"([
2213+
[0, 1, [["a", 10], ["b", 11]]],
2214+
[0, 2, [["c", 20]]],
2215+
[0, 3, [["d", 30], ["e", 31]]],
2216+
[0, 4, [["a", 40], ["f", 41]]],
2217+
[0, 5, null]
2218+
])"));
2219+
ASSERT_TRUE(success);
2220+
}
2221+
21192222
// Nested map values through both selected physical columns and overflow.
21202223
TEST_P(WriteAndReadInteTest, TestSharedShreddingWithStructValue) {
21212224
auto [file_format, file_system] = GetParam();

0 commit comments

Comments
 (0)