github-actions[bot] commented on code in PR #65840:
URL: https://github.com/apache/doris/pull/65840#discussion_r3620500745
##########
be/test/storage/cloud_file_cache_write_index_only_test.cpp:
##########
@@ -796,4 +803,144 @@ TEST_F(CloudFileCacheWriteIndexOnlyTest,
expect_segment_write_bypasses_file_cache(created_files);
}
+class SegmentWriterFileCacheAlignmentTest : public
CloudFileCacheWriteIndexOnlyTest {};
+
+TEST_F(SegmentWriterFileCacheAlignmentTest,
+ FinalizeCreatesCanonicalCacheBlocksBeforeAsyncUploadCompletes) {
+ config::enable_file_cache_write_index_file_only = false;
+
+ auto* sync_point = SyncPoint::get_instance();
+ sync_point->clear_all_call_backs();
+ sync_point->enable_processing();
+ SyncPoint::CallbackGuard s3_client_guard;
+ sync_point->set_call_back(
+ "s3_client_factory::create",
+ [](auto&& args) {
+ auto* ret =
try_any_cast_ret<std::shared_ptr<io::S3ObjStorageClient>>(args);
+ ret->second = true;
+ },
+ &s3_client_guard);
+ SyncPoint::CallbackGuard s3_put_guard;
+ sync_point->set_call_back(
+ "S3FileWriter::_put_object",
+ [](auto&& args) {
+ auto* should_return = try_any_cast<bool*>(args.back());
+ *should_return = true;
+ },
+ &s3_put_guard);
+
+ std::mutex upload_mutex;
+ std::condition_variable upload_cv;
+ bool upload_entered_cache_stage = false;
+ bool release_upload = false;
+ SyncPoint::CallbackGuard upload_cache_guard;
+ sync_point->set_call_back(
+ "UploadFileBuffer::upload_to_local_file_cache",
+ [&](auto&& /*args*/) {
+ std::unique_lock lock(upload_mutex);
+ upload_entered_cache_stage = true;
+ upload_cv.notify_all();
+ upload_cv.wait(lock, [&]() { return release_upload; });
+ },
+ &upload_cache_guard);
+ const auto unblock_upload = [&]() {
+ {
+ std::lock_guard lock(upload_mutex);
+ release_upload = true;
+ }
+ upload_cv.notify_all();
+ };
+
+ const std::string segment_path =
"segment_writer_finalize_cache_alignment_0.dat";
+ io::FileWriterOptions file_options;
+ file_options.write_file_cache = true;
+ io::FileWriterPtr file_writer;
+ auto status = _remote_fs->create_file(segment_path, &file_writer,
&file_options);
+ ASSERT_TRUE(status.ok()) << status;
+ ASSERT_NE(file_writer->cache_builder(), nullptr);
+ Defer unblock_upload_on_exit {unblock_upload};
+ const auto cache_hash = file_writer->cache_builder()->_cache_hash;
+ auto* cache = file_writer->cache_builder()->_cache;
+ bool observed_index_cache_holder = false;
+ SyncPoint::CallbackGuard index_cache_holder_guard;
+ sync_point->set_call_back(
+ "SegmentWriter::finalize::index_cache_holder",
+ [&](auto&& args) {
+ auto* holder = try_any_cast<io::FileBlocksHolder*>(args[0]);
+ ASSERT_NE(holder, nullptr);
+ ASSERT_EQ(holder->file_blocks.size(), 1);
+ const auto& block = holder->file_blocks.front();
+ const size_t cache_block_size =
+
static_cast<size_t>(config::file_cache_each_block_size);
+ EXPECT_EQ(block->range().left, 0);
+ EXPECT_EQ(block->range().right, cache_block_size - 1);
+ EXPECT_EQ(block->state(), io::FileBlock::State::EMPTY);
+ EXPECT_EQ(block->cache_type(), io::FileCacheType::INDEX);
+ observed_index_cache_holder = true;
+ },
+ &index_cache_holder_guard);
+
+ auto tablet_schema = create_schema();
+ TabletMetaPB tablet_meta_pb;
+ tablet_meta_pb.set_tablet_id(kIndexOnlyTabletId);
+ tablet_meta_pb.set_schema_hash(kIndexOnlyTabletSchemaHash);
+ tablet_meta_pb.set_tablet_state(PB_RUNNING);
+ tablet_schema->to_schema_pb(tablet_meta_pb.mutable_schema());
+ auto tablet_meta = std::make_shared<TabletMeta>();
+ tablet_meta->init_from_pb(tablet_meta_pb);
+ auto tablet = std::make_shared<Tablet>(*_engine, tablet_meta, nullptr);
+ auto rowset_context = create_context(tablet_schema);
+ segment_v2::SegmentWriterOptions writer_options;
+ writer_options.compression_type = NO_COMPRESSION;
+ writer_options.rowset_ctx = &rowset_context;
+ writer_options.write_type = rowset_context.write_type;
+ segment_v2::SegmentWriter writer(file_writer.get(), 0, tablet_schema,
tablet, nullptr,
+ writer_options, nullptr);
+ status = writer.init();
+ ASSERT_TRUE(status.ok()) << status;
+ auto block = create_full_block(tablet_schema);
+ status = writer.append_block(&block, 0, block.rows());
+ ASSERT_TRUE(status.ok()) << status;
+
+ uint64_t segment_file_size = 0;
+ uint64_t index_size = 0;
+ segment_v2::SegmentIndexFileCacheInfo index_cache_info;
+ status = writer.finalize(&segment_file_size, &index_size,
&index_cache_info);
+ ASSERT_TRUE(status.ok()) << status;
+
+ {
+ std::unique_lock lock(upload_mutex);
+ ASSERT_TRUE(upload_cv.wait_for(lock, std::chrono::seconds(10),
+ [&]() { return
upload_entered_cache_stage; }));
+ }
+
+ const size_t cache_block_size =
static_cast<size_t>(config::file_cache_each_block_size);
+ ASSERT_GT(cache_block_size, 0);
+ ASSERT_GT(segment_file_size, 0);
+ ASSERT_LT(segment_file_size, cache_block_size);
+ ASSERT_GT(index_size, 0);
+ ASSERT_GT(index_cache_info.cache_start_offset(), 0);
+ ASSERT_LT(index_cache_info.cache_start_offset(), segment_file_size);
+ ASSERT_NE(index_cache_info.cache_start_offset() % cache_block_size, 0);
+ EXPECT_TRUE(observed_index_cache_holder);
+
+ unblock_upload();
+ const auto close_deadline = std::chrono::steady_clock::now() +
std::chrono::seconds(10);
+ do {
+ status = file_writer->try_finish_close();
+ if (status.is<ErrorCode::NEED_SEND_AGAIN>()) {
+ std::this_thread::sleep_for(std::chrono::milliseconds(1));
+ }
+ } while (status.is<ErrorCode::NEED_SEND_AGAIN>() &&
+ std::chrono::steady_clock::now() < close_deadline);
+ ASSERT_TRUE(status.ok()) << status;
+
+ auto cache_blocks = cache->get_blocks_by_key(cache_hash);
+ ASSERT_FALSE(cache_blocks.empty());
+ const auto& downloaded_block = cache_blocks.begin()->second;
+ EXPECT_EQ(downloaded_block->range().left, 0);
Review Comment:
[P2] Exercise reuse of the INDEX cell in this test
This callback fires before `UploadFileBuffer` executes `_holder =
_alloc_holder()`. Consequently, `SegmentWriter::finalize` creates the only
holder for an EMPTY INDEX cell, then destroys it;
`FileBlocksHolder::~FileBlocksHolder` removes that cell, and the unblocked
uploader later creates a fresh NORMAL cell. These final assertions therefore
pass without exercising the compatible-cell reuse this PR is meant to enable,
and without preserving the intended INDEX classification. Please synchronize
after the uploader has acquired its holder (or otherwise retain the
reservation), then assert the final block is INDEX and that its cached bytes
match the segment.
##########
be/src/io/cache/file_cache_common.cpp:
##########
@@ -133,16 +133,27 @@ std::string UInt128Wrapper::to_string() const {
return get_hex_uint_lowercase(value_);
}
+FileCacheRange align_file_cache_range(size_t offset, size_t size, size_t
block_size) {
+ const size_t range_end = offset + size;
+ const size_t aligned_offset = offset - offset % block_size;
+ const size_t end_remainder = range_end % block_size;
+ const size_t aligned_end =
+ end_remainder == 0 ? range_end : range_end + block_size -
end_remainder;
+ return {.offset = aligned_offset, .size = aligned_end - aligned_offset};
+}
+
FileBlocksHolderPtr FileCacheAllocatorBuilder::allocate_cache_holder(size_t
offset, size_t size,
int64_t
tablet_id) const {
+ const auto aligned_range = align_file_cache_range(offset, size,
_cache->max_file_block_size());
Review Comment:
[P1] Preserve writer offsets when expanding this shared holder
This builder is also used for HDFS and S3 data buffers, but those copy loops
still treat byte 0 of the first returned block as byte 0 of the current buffer.
HDFS has no buffer/cache-block divisibility invariant. For example, with a 640
KiB cache block (a valid divisor of the default 5 MiB S3 buffer) and a 1 MiB
HDFS buffer, the first batch leaves a downloaded `[640,1024 KiB)` tail. The
next `[1024,2048 KiB)` request is expanded back to 640 KiB; the HDFS loop
consumes that 384 KiB downloaded prefix while advancing `pos`, then stores
remote `[1408,2048 KiB)` in a cell advertised as `[1024,1664 KiB)`. The cell
finalizes successfully, so a cache hit can return shifted bytes. Please
preserve the original requested interval in every data-writer copy, restrict
this alignment to the metadata reservation, or enforce alignment for every
caller, and add an adjacent-buffer byte-readback test.
--
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
