lxy-9602 commented on code in PR #248:
URL: https://github.com/apache/paimon-cpp/pull/248#discussion_r3869333085
##########
test/inte/write_and_read_inte_test.cpp:
##########
@@ -630,6 +676,652 @@ TEST_P(WriteAndReadInteTest, TestPKSimple) {
ASSERT_TRUE(success);
}
+TEST_P(WriteAndReadInteTest, TestInputChangelogStreamRead) {
+ arrow::FieldVector fields = {
+ arrow::field("pk", arrow::utf8()),
+ arrow::field("value", arrow::int32()),
+ };
+ auto [file_format, file_system] = GetParam();
+ std::map<std::string, std::string> options = {
+ {Options::MANIFEST_FORMAT, "avro"}, {Options::FILE_FORMAT,
file_format},
+ {Options::TARGET_FILE_SIZE, "1024"}, {Options::BUCKET, "1"},
+ {Options::FILE_SYSTEM, file_system}, {Options::CHANGELOG_PRODUCER,
"input"},
+ };
+ if (file_system == "jindo") {
+ options = AddOptionsForJindo(options);
+ }
+ ASSERT_OK_AND_ASSIGN(
+ auto helper,
+ TestHelper::Create(test_dir_, arrow::schema(fields),
/*partition_keys=*/{},
+ /*primary_keys=*/{"pk"}, options,
/*is_streaming_mode=*/true));
+
+ ASSERT_OK_AND_ASSIGN(std::vector<std::shared_ptr<Split>> initial_splits,
+ helper->NewScan(StartupMode::Latest(),
/*snapshot_id=*/std::nullopt));
+ ASSERT_TRUE(initial_splits.empty());
+
+ ASSERT_OK_AND_ASSIGN(
+ std::unique_ptr<RecordBatch> batch,
+ TestHelper::MakeRecordBatch(
+ arrow::struct_(fields), R"([["Alice", 10], ["Bob", 20], ["Alice",
11], ["Bob", 21]])",
+ /*partition_map=*/{}, /*bucket=*/0,
+ {RecordBatch::RowKind::INSERT, RecordBatch::RowKind::UPDATE_BEFORE,
+ RecordBatch::RowKind::UPDATE_AFTER,
RecordBatch::RowKind::DELETE}));
+ ASSERT_OK(helper->WriteAndCommit(std::move(batch), /*commit_identifier=*/0,
+
/*expected_commit_messages=*/std::nullopt));
+
+ ASSERT_OK_AND_ASSIGN(std::vector<std::shared_ptr<Split>> changelog_splits,
helper->Scan());
+ ASSERT_TRUE(changelog_splits.empty());
+ ASSERT_OK_AND_ASSIGN(changelog_splits, helper->Scan());
+ ASSERT_FALSE(changelog_splits.empty());
+ auto expected_type = arrow::struct_({
+ arrow::field("_VALUE_KIND", arrow::int8()),
+ fields[0],
+ fields[1],
+ });
+ ASSERT_OK_AND_ASSIGN(bool success,
+ helper->ReadAndCheckResult(expected_type,
changelog_splits,
+ R"([[0, "Alice", 10], [2,
"Alice", 11],
+ [1, "Bob", 20], [3, "Bob", 21]])"));
+ ASSERT_TRUE(success);
+}
+
+TEST_P(WriteAndReadInteTest, TestLookupChangelogStreamRead) {
+ auto [file_format, file_system] = GetParam();
+ arrow::FieldVector fields = {
+ arrow::field("pk", arrow::utf8()),
+ arrow::field("value", arrow::int32()),
+ };
+ std::map<std::string, std::string> options = {
+ {Options::MANIFEST_FORMAT, "avro"}, {Options::FILE_FORMAT,
file_format},
+ {Options::TARGET_FILE_SIZE, "1024"}, {Options::BUCKET, "1"},
+ {Options::FILE_SYSTEM, file_system}, {Options::CHANGELOG_PRODUCER,
"lookup"},
+ };
+ ASSERT_OK_AND_ASSIGN(auto helper,
CreateLookupTestHelper(arrow::schema(fields), options));
+
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<RecordBatch> initial_batch,
+ TestHelper::MakeRecordBatch(arrow::struct_(fields),
R"([["Alice", 10]])",
+ /*partition_map=*/{},
/*bucket=*/0, {}));
+ ASSERT_OK(helper->WriteAndCommit(std::move(initial_batch),
/*commit_identifier=*/0,
+
/*expected_commit_messages=*/std::nullopt));
+
+ std::string table_path = PathUtil::JoinPath(test_dir_, "foo.db/bar");
+ auto compact_and_commit = [this, &options, &table_path](int64_t
commit_identifier) -> Status {
+ WriteContextBuilder write_context_builder(table_path, "commit_user");
+ write_context_builder.WithTempDirectory(LookupTempDirectory());
+ PAIMON_ASSIGN_OR_RAISE(
+ std::unique_ptr<WriteContext> write_context,
+
write_context_builder.SetOptions(options).WithStreamingMode(true).Finish());
+ PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<FileStoreWrite>
file_store_write,
+
FileStoreWrite::Create(std::move(write_context)));
+ PAIMON_RETURN_NOT_OK(file_store_write->Compact(/*partition=*/{},
/*bucket=*/0,
+
/*full_compaction=*/true));
+ PAIMON_ASSIGN_OR_RAISE(
+ std::vector<std::shared_ptr<CommitMessage>> compact_messages,
+ file_store_write->PrepareCommit(/*wait_compaction=*/true,
commit_identifier));
+ PAIMON_RETURN_NOT_OK(file_store_write->Close());
+ if (compact_messages.empty()) {
+ return Status::Invalid("expected lookup compaction commit
messages");
+ }
+ CommitContextBuilder commit_context_builder(table_path, "commit_user");
+ PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<CommitContext> commit_context,
+
commit_context_builder.SetOptions(options).Finish());
+ PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<FileStoreCommit>
file_store_commit,
+
FileStoreCommit::Create(std::move(commit_context)));
+ return file_store_commit->Commit(compact_messages, commit_identifier);
+ };
+
+ // Move the initial value to a high level so the next compaction must look
it up.
+ ASSERT_OK(compact_and_commit(/*commit_identifier=*/1));
+ helper.reset();
+ ASSERT_OK_AND_ASSIGN(helper, CreateLookupTestHelper(table_path, options));
+
+ ASSERT_OK_AND_ASSIGN(std::vector<std::shared_ptr<Split>> initial_splits,
+ helper->NewScan(StartupMode::Latest(),
/*snapshot_id=*/std::nullopt));
+ ASSERT_TRUE(initial_splits.empty());
+
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<RecordBatch> update_batch,
+ TestHelper::MakeRecordBatch(arrow::struct_(fields),
R"([["Alice", 20]])",
+ /*partition_map=*/{},
/*bucket=*/0, {}));
+ ASSERT_OK(helper->WriteAndCommit(std::move(update_batch),
/*commit_identifier=*/2,
+
/*expected_commit_messages=*/std::nullopt));
+ ASSERT_OK(compact_and_commit(/*commit_identifier=*/3));
+
+ ASSERT_OK_AND_ASSIGN(std::vector<std::shared_ptr<Split>> changelog_splits,
helper->Scan());
+ ASSERT_FALSE(changelog_splits.empty());
+ auto expected_type = arrow::struct_({
+ arrow::field("_VALUE_KIND", arrow::int8()),
+ fields[0],
+ fields[1],
+ });
+ ASSERT_OK_AND_ASSIGN(bool success,
+ helper->ReadAndCheckResult(expected_type,
changelog_splits,
+ R"([[1, "Alice", 10], [2,
"Alice", 20]])"));
+ ASSERT_TRUE(success);
+}
+
+TEST_P(WriteAndReadInteTest,
TestLookupChangelogInitialFullScanExcludesLevelZero) {
+ auto [file_format, file_system] = GetParam();
+ arrow::FieldVector fields = {
+ arrow::field("pk", arrow::utf8()),
+ arrow::field("value", arrow::int32()),
+ };
+ std::map<std::string, std::string> options = {
+ {Options::MANIFEST_FORMAT, "avro"}, {Options::FILE_FORMAT,
file_format},
+ {Options::TARGET_FILE_SIZE, "1024"}, {Options::BUCKET, "1"},
+ {Options::FILE_SYSTEM, file_system}, {Options::CHANGELOG_PRODUCER,
"lookup"},
+ };
+ ASSERT_OK_AND_ASSIGN(auto helper,
CreateLookupTestHelper(arrow::schema(fields), options));
+
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<RecordBatch> initial_batch,
+ TestHelper::MakeRecordBatch(arrow::struct_(fields),
R"([["Alice", 10]])",
+ /*partition_map=*/{},
/*bucket=*/0, {}));
+ ASSERT_OK(helper->WriteAndCommit(std::move(initial_batch),
/*commit_identifier=*/0,
+
/*expected_commit_messages=*/std::nullopt));
+
+ std::string table_path = PathUtil::JoinPath(test_dir_, "foo.db/bar");
+ WriteContextBuilder write_context_builder(table_path, "commit_user");
+ write_context_builder.WithTempDirectory(LookupTempDirectory());
+ ASSERT_OK_AND_ASSIGN(
+ std::unique_ptr<WriteContext> write_context,
+
write_context_builder.SetOptions(options).WithStreamingMode(true).Finish());
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<FileStoreWrite> file_store_write,
+ FileStoreWrite::Create(std::move(write_context)));
+ ASSERT_OK(file_store_write->Compact(/*partition=*/{}, /*bucket=*/0,
+ /*full_compaction=*/true));
+ ASSERT_OK_AND_ASSIGN(
+ std::vector<std::shared_ptr<CommitMessage>> compact_messages,
+ file_store_write->PrepareCommit(/*wait_compaction=*/true,
/*commit_identifier=*/1));
+ ASSERT_OK(file_store_write->Close());
+ ASSERT_FALSE(compact_messages.empty());
+ CommitContextBuilder commit_context_builder(table_path, "commit_user");
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<CommitContext> commit_context,
+ commit_context_builder.SetOptions(options).Finish());
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<FileStoreCommit> file_store_commit,
+ FileStoreCommit::Create(std::move(commit_context)));
+ ASSERT_OK(file_store_commit->Commit(compact_messages,
/*commit_identifier=*/1));
+
Review Comment:
Why not use `CompactAndCommit`?
##########
test/inte/write_and_read_inte_test.cpp:
##########
@@ -630,6 +676,652 @@ TEST_P(WriteAndReadInteTest, TestPKSimple) {
ASSERT_TRUE(success);
}
+TEST_P(WriteAndReadInteTest, TestInputChangelogStreamRead) {
+ arrow::FieldVector fields = {
+ arrow::field("pk", arrow::utf8()),
+ arrow::field("value", arrow::int32()),
+ };
+ auto [file_format, file_system] = GetParam();
+ std::map<std::string, std::string> options = {
+ {Options::MANIFEST_FORMAT, "avro"}, {Options::FILE_FORMAT,
file_format},
+ {Options::TARGET_FILE_SIZE, "1024"}, {Options::BUCKET, "1"},
+ {Options::FILE_SYSTEM, file_system}, {Options::CHANGELOG_PRODUCER,
"input"},
+ };
+ if (file_system == "jindo") {
+ options = AddOptionsForJindo(options);
+ }
+ ASSERT_OK_AND_ASSIGN(
+ auto helper,
+ TestHelper::Create(test_dir_, arrow::schema(fields),
/*partition_keys=*/{},
+ /*primary_keys=*/{"pk"}, options,
/*is_streaming_mode=*/true));
+
+ ASSERT_OK_AND_ASSIGN(std::vector<std::shared_ptr<Split>> initial_splits,
+ helper->NewScan(StartupMode::Latest(),
/*snapshot_id=*/std::nullopt));
+ ASSERT_TRUE(initial_splits.empty());
+
+ ASSERT_OK_AND_ASSIGN(
+ std::unique_ptr<RecordBatch> batch,
+ TestHelper::MakeRecordBatch(
+ arrow::struct_(fields), R"([["Alice", 10], ["Bob", 20], ["Alice",
11], ["Bob", 21]])",
+ /*partition_map=*/{}, /*bucket=*/0,
+ {RecordBatch::RowKind::INSERT, RecordBatch::RowKind::UPDATE_BEFORE,
+ RecordBatch::RowKind::UPDATE_AFTER,
RecordBatch::RowKind::DELETE}));
+ ASSERT_OK(helper->WriteAndCommit(std::move(batch), /*commit_identifier=*/0,
+
/*expected_commit_messages=*/std::nullopt));
+
+ ASSERT_OK_AND_ASSIGN(std::vector<std::shared_ptr<Split>> changelog_splits,
helper->Scan());
+ ASSERT_TRUE(changelog_splits.empty());
+ ASSERT_OK_AND_ASSIGN(changelog_splits, helper->Scan());
+ ASSERT_FALSE(changelog_splits.empty());
+ auto expected_type = arrow::struct_({
+ arrow::field("_VALUE_KIND", arrow::int8()),
+ fields[0],
+ fields[1],
+ });
+ ASSERT_OK_AND_ASSIGN(bool success,
+ helper->ReadAndCheckResult(expected_type,
changelog_splits,
+ R"([[0, "Alice", 10], [2,
"Alice", 11],
+ [1, "Bob", 20], [3, "Bob", 21]])"));
+ ASSERT_TRUE(success);
+}
+
+TEST_P(WriteAndReadInteTest, TestLookupChangelogStreamRead) {
+ auto [file_format, file_system] = GetParam();
+ arrow::FieldVector fields = {
+ arrow::field("pk", arrow::utf8()),
+ arrow::field("value", arrow::int32()),
+ };
+ std::map<std::string, std::string> options = {
+ {Options::MANIFEST_FORMAT, "avro"}, {Options::FILE_FORMAT,
file_format},
+ {Options::TARGET_FILE_SIZE, "1024"}, {Options::BUCKET, "1"},
+ {Options::FILE_SYSTEM, file_system}, {Options::CHANGELOG_PRODUCER,
"lookup"},
+ };
+ ASSERT_OK_AND_ASSIGN(auto helper,
CreateLookupTestHelper(arrow::schema(fields), options));
+
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<RecordBatch> initial_batch,
+ TestHelper::MakeRecordBatch(arrow::struct_(fields),
R"([["Alice", 10]])",
+ /*partition_map=*/{},
/*bucket=*/0, {}));
+ ASSERT_OK(helper->WriteAndCommit(std::move(initial_batch),
/*commit_identifier=*/0,
+
/*expected_commit_messages=*/std::nullopt));
+
+ std::string table_path = PathUtil::JoinPath(test_dir_, "foo.db/bar");
+ auto compact_and_commit = [this, &options, &table_path](int64_t
commit_identifier) -> Status {
+ WriteContextBuilder write_context_builder(table_path, "commit_user");
Review Comment:
Why not use `CompactAndCommit`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
