Skip to content

branch-4.1: [fix](inverted index) Persist empty index files and stop writing orphan ones #67859 #68082

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
airborne12 wants to merge 1 commit into
base: branch-4.1
Choose a base branch
from
Open

branch-4.1: [fix](inverted index) Persist empty index files and stop writing orphan ones #67859 #68082

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
24 changes: 15 additions & 9 deletions be/src/storage/index/index_file_writer.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,7 +24,6 @@

#include "common/status.h"
#include "io/fs/packed_file_writer.h"
#include "io/fs/s3_file_writer.h"
#include "io/fs/stream_sink_file_writer.h"
#include "storage/index/ann/ann_index_files.h"
#include "storage/index/index_file_reader.h"
Expand Down Expand Up @@ -198,10 +197,16 @@ Status IndexFileWriter::begin_close() {
DCHECK(!_closed) << debug_string();
_closed = true;
if (_indices_dirs.empty()) {
// An empty file must still be created even if there are no indexes to write
if (dynamic_cast<io::StreamSinkFileWriter*>(_idx_v2_writer.get()) != nullptr ||
dynamic_cast<io::S3FileWriter*>(_idx_v2_writer.get()) != nullptr ||
dynamic_cast<io::PackedFileWriter*>(_idx_v2_writer.get()) != nullptr) {
// A schema that owns an index file always gets one, even when no logical
// index had anything to write (an all-NULL VARIANT column extracts no
// subcolumn, so no directory is ever opened). The file is committed by
// close(), not by create_file(): S3 turns a zero-byte writer into an empty
// object, StreamSink sends segment_eos, and LocalFileWriter's destructor
// ABORTS -- and deletes -- a writer it was never asked to close. Dispatch
// through FileWriter rather than naming implementations: the old whitelist
// silently dropped LocalFileWriter and HdfsFileWriter, and every new
// implementation would have had to remember to add itself here.
if (_idx_v2_writer != nullptr && _idx_v2_writer->state() != io::FileWriter::State::CLOSED) {
return _idx_v2_writer->close(true);
}
return Status::OK();
Expand Down Expand Up @@ -240,10 +245,11 @@ Status IndexFileWriter::begin_close() {
Status IndexFileWriter::finish_close() {
DCHECK(_closed) << debug_string();
if (_indices_dirs.empty()) {
// An empty file must still be created even if there are no indexes to write
if (dynamic_cast<io::StreamSinkFileWriter*>(_idx_v2_writer.get()) != nullptr ||
dynamic_cast<io::S3FileWriter*>(_idx_v2_writer.get()) != nullptr ||
dynamic_cast<io::PackedFileWriter*>(_idx_v2_writer.get()) != nullptr) {
// Second phase of the empty-file close begun in begin_close(). Skipping an
// already CLOSED writer keeps this idempotent: begin_close() may have
// closed synchronously, and a retried finish_close() must not send a
// second EOS or PUT a second empty object.
if (_idx_v2_writer != nullptr && _idx_v2_writer->state() != io::FileWriter::State::CLOSED) {
return _idx_v2_writer->close(false);
}
return Status::OK();
Expand Down
33 changes: 29 additions & 4 deletions be/src/storage/task/index_builder.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -286,7 +286,12 @@ Status IndexBuilder::update_inverted_index_info() {
st = Status::Error<ErrorCode::INIT_FAILED>(
"debug point: reader init error");
})
if (!st.ok() && !st.is<ErrorCode::INVERTED_INDEX_FILE_NOT_FOUND>()) {
// A missing container (the rowset predates every index) and an
// empty one (the schema owns an index, but no logical index had
// anything to write) both mean there is nothing to carry over.
// In both cases every requested index is built from the raw columns.
if (!st.ok() && !st.is<ErrorCode::INVERTED_INDEX_FILE_NOT_FOUND>() &&
!st.is<ErrorCode::INVERTED_INDEX_BYPASS>()) {
return st;
}
_index_file_readers.emplace(
Expand Down Expand Up @@ -339,8 +344,19 @@ Status IndexBuilder::handle_single_rowset(RowsetMetaSharedPtr output_rowset_meta

if (_is_drop_op) {
const auto& output_rs_tablet_schema = output_rowset_meta->tablet_schema();
if (output_rs_tablet_schema->get_inverted_index_storage_format() !=
InvertedIndexStorageFormatPB::V1) {
// A rowset must not keep an index file that its own schema does not claim:
// link, copy, upload, remove and checksum all consult that schema before
// touching the compound file. The old LocalFileWriter destructor happened
// to delete an unclosed orphan, but remote writers could preserve it.
const bool is_v1 = output_rs_tablet_schema->get_inverted_index_storage_format() ==
InvertedIndexStorageFormatPB::V1;
const bool output_has_index_file = output_rs_tablet_schema->has_inverted_index() ||
output_rs_tablet_schema->has_ann_index();
if (!is_v1 && !output_has_index_file) {
LOG(INFO) << "drop index removed the last index, no index file is written. tablet_id="
<< _tablet->tablet_id()
<< " rowset_id=" << output_rowset_meta->rowset_id().to_string();
} else if (!is_v1) {
const auto& fs = output_rowset_meta->fs();

const auto& output_rowset_schema = output_rowset_meta->tablet_schema();
Expand Down Expand Up @@ -409,8 +425,17 @@ Status IndexBuilder::handle_single_rowset(RowsetMetaSharedPtr output_rowset_meta
return Status::OK();
} else {
// create inverted or ann index writer
const auto& fs = output_rowset_meta->fs();
auto output_rowset_schema = output_rowset_meta->tablet_schema();
// If no requested index survives schema resolution and the input rowset
// owned none, the output schema must not gain an orphan compound file.
if (!output_rowset_schema->has_inverted_index() && !output_rowset_schema->has_ann_index()) {
LOG(INFO) << "no index in the output rowset schema, no index file is written."
<< " tablet_id=" << _tablet->tablet_id()
<< " rowset_id=" << output_rowset_meta->rowset_id().to_string()
<< " source_rows=" << output_rowset_meta->num_rows();
return Status::OK();
}
const auto& fs = output_rowset_meta->fs();
size_t inverted_index_size = 0;
for (auto& seg_ptr : segments) {
std::string index_path_prefix {
Expand Down
24 changes: 21 additions & 3 deletions be/test/io/fs/s3_file_writer_test.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1529,6 +1529,10 @@ TEST_F(S3FileWriterTest, write_buffer_boundary) {
}

TEST_F(S3FileWriterTest, test_empty_file) {
bool enable_file_cache = config::enable_file_cache;
config::enable_file_cache = false;
Defer defer {[&]() { config::enable_file_cache = enable_file_cache; }};

std::vector<StorePath> paths;
paths.emplace_back(std::string("tmp_dir"), 1024000000);
auto tmp_file_dirs = std::make_unique<segment_v2::TmpFileDirs>(paths);
Expand All @@ -1537,20 +1541,34 @@ TEST_F(S3FileWriterTest, test_empty_file) {
doris::io::FileWriterOptions opts;
io::FileWriterPtr file_writer;
auto st = s3_fs->create_file("test_empty_file.idx", &file_writer, &opts);
EXPECT_TRUE(st.ok()) << st;
ASSERT_TRUE(st.ok()) << st;
auto holder = std::make_shared<ObjClientHolder>(S3ClientConf {});
auto mock_client = std::make_shared<SimpleMockObjStorageClient>();
holder->_client = mock_client;
dynamic_cast<io::S3FileWriter*>(file_writer.get())->_obj_client = holder;
auto* s3_writer = file_writer.get();
const auto file_path = s3_writer->path().native();
auto fs = io::global_local_filesystem();
std::string index_path = "/tmp/empty_index_file_test";
std::string rowset_id = "1234567890";
int64_t seg_id = 1234567890;
auto index_file_writer = std::make_unique<segment_v2::IndexFileWriter>(
fs, index_path, rowset_id, seg_id, InvertedIndexStorageFormatPB::V2,
std::move(file_writer), false);
EXPECT_TRUE(index_file_writer->begin_close().ok());
EXPECT_TRUE(index_file_writer->finish_close().ok());
ASSERT_TRUE(index_file_writer->begin_close().ok());
EXPECT_EQ(s3_writer->state(), io::FileWriter::State::ASYNC_CLOSING);
ASSERT_TRUE(index_file_writer->finish_close().ok());
EXPECT_EQ(s3_writer->state(), io::FileWriter::State::CLOSED);
EXPECT_EQ(s3_writer->bytes_appended(), 0);
// Idempotent: a retried finish must not PUT a second object.
ASSERT_TRUE(index_file_writer->finish_close().ok());
index_file_writer.reset();
// An empty remote index file is one zero-byte object, not a multipart upload
// and not a missing key.
EXPECT_EQ(mock_client->put_object_count, 1);
EXPECT_EQ(mock_client->upload_part_count, 0);
ASSERT_EQ(mock_client->objects.count(file_path), 1);
EXPECT_TRUE(mock_client->objects.at(file_path).empty());
}

} // namespace doris
153 changes: 152 additions & 1 deletion be/test/storage/index/index_builder_test.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,9 @@
#include <gmock/gmock.h>
#include <gtest/gtest.h>

#include <set>

#include "storage/index/index_file_reader.h"
#include "storage/olap_common.h"
#include "storage/rowset/beta_rowset.h"
#include "storage/rowset/rowset_factory.h"
Expand Down Expand Up @@ -169,6 +172,133 @@ class IndexBuilderTest : public ::testing::Test {
rs_meta->set_tablet_schema(tablet_schema);
}

// Builds a k2 index over a rowset whose schema owns a k1 index that was never
// written, and returns the index ids present in the output rowset's index
// file. `leave_empty_index_file` picks between the two shapes a source
// segment with no index content can have on disk: a zero-byte file (read as
// INVERTED_INDEX_BYPASS) or no file at all (INVERTED_INDEX_FILE_NOT_FOUND).
//
// The rowset is written BEFORE the schema gains k1, so k1 has no content
// anywhere and nothing can be lost by either shape. That is not an artifact
// of the test: a zero-byte index file is only ever produced when no logical
// index opened a directory, and every non-VARIANT index opens its directory
// eagerly in IndexColumnWriter::create(). A zero-byte file therefore cannot
// be a former home of some other index's data.
// GTest assertions stay together so the two source-file shapes cannot drift.
// NOLINTNEXTLINE(readability-function-cognitive-complexity,readability-function-size)
std::set<int64_t> build_k2_index_over_unwritten_k1(int64_t tablet_id, int64_t rowset_id,
bool leave_empty_index_file) {
auto tablet_path = _absolute_dir + "/" + std::to_string(tablet_id);
_tablet->_tablet_path = tablet_path;
EXPECT_TRUE(io::global_local_filesystem()->delete_directory(tablet_path).ok());
EXPECT_TRUE(io::global_local_filesystem()->create_directory(tablet_path).ok());

RowsetSharedPtr rowset;
RowsetWriterContext writer_context;
writer_context.rowset_id.init(rowset_id);
writer_context.tablet_id = rowset_id;
writer_context.tablet_schema_hash = 567997577;
writer_context.partition_id = 10;
writer_context.rowset_type = BETA_ROWSET;
writer_context.tablet_path = _absolute_dir + "/" + std::to_string(rowset_id);
writer_context.rowset_state = VISIBLE;
writer_context.tablet_schema = _tablet_schema;
writer_context.version.first = 10;
writer_context.version.second = 10;
EXPECT_TRUE(
io::global_local_filesystem()->create_directory(writer_context.tablet_path).ok());

auto res = RowsetFactory::create_rowset_writer(*_engine_ref, writer_context, false);
EXPECT_TRUE(res.has_value()) << res.error();
auto rowset_writer = std::move(res).value();
{
Block block = _tablet_schema->create_block();
{
auto columns_guard = block.mutate_columns_scoped();
auto& columns = columns_guard.mutable_columns();
for (int i = 0; i < 1000; ++i) {
int32_t k1 = i * 10;
columns[0]->insert_data((const char*)&k1, sizeof(k1));
int32_t k2 = i % 100;
columns[1]->insert_data((const char*)&k2, sizeof(k2));
}
}
EXPECT_TRUE(rowset_writer->add_block(&block).ok());
EXPECT_TRUE(rowset_writer->flush().ok());
EXPECT_TRUE(rowset_writer->build(rowset).ok());
EXPECT_TRUE(_tablet->add_rowset(rowset).ok());
}

// The schema gains k1 only now, so the rowset owns an index whose content
// was never written -- the same state an all-NULL VARIANT column leaves.
TabletIndex k1_index;
k1_index._index_id = 1;
k1_index._index_name = "k1_index";
k1_index._index_type = IndexType::INVERTED;
k1_index._col_unique_ids.push_back(1);
_tablet_schema->append_index(std::move(k1_index));

auto segment_path = rowset->segment_path(0);
EXPECT_TRUE(segment_path.has_value()) << segment_path.error();
const std::string source_prefix {
segment_v2::InvertedIndexDescriptor::get_index_file_path_prefix(
segment_path.value())};
const auto source_index_path =
segment_v2::InvertedIndexDescriptor::get_index_file_path_v2(source_prefix);
bool exists = true;
EXPECT_TRUE(io::global_local_filesystem()->exists(source_index_path, &exists).ok());
EXPECT_FALSE(exists) << "an index-less rowset must not have written an index file";
if (leave_empty_index_file) {
io::FileWriterPtr empty;
EXPECT_TRUE(io::global_local_filesystem()->create_file(source_index_path, &empty).ok());
EXPECT_TRUE(empty->close().ok());
}
{
auto reader = std::make_unique<segment_v2::IndexFileReader>(
io::global_local_filesystem(), source_prefix, InvertedIndexStorageFormatPB::V2);
auto st = reader->init();
EXPECT_TRUE(leave_empty_index_file ? st.is<ErrorCode::INVERTED_INDEX_BYPASS>()
: st.is<ErrorCode::INVERTED_INDEX_FILE_NOT_FOUND>())
<< st;
}

TOlapTableIndex k2_index;
k2_index.index_id = 2;
k2_index.columns.emplace_back("k2");
k2_index.index_name = "k2_index";
k2_index.index_type = TIndexType::INVERTED;
_alter_indexes.clear();
_alter_indexes.push_back(k2_index);

IndexBuilder builder(ExecEnv::GetInstance()->storage_engine().to_local(), _tablet, _columns,
_alter_indexes, false);
EXPECT_TRUE(builder.init().ok());
auto status = builder.do_build_inverted_index();
EXPECT_TRUE(status.ok()) << status.to_string();

std::set<int64_t> output_index_ids;
EXPECT_EQ(builder._output_rowsets.size(), 1);
if (builder._output_rowsets.empty()) {
return output_index_ids;
}
auto output_segment_path = builder._output_rowsets[0]->segment_path(0);
EXPECT_TRUE(output_segment_path.has_value()) << output_segment_path.error();
auto reader = std::make_unique<segment_v2::IndexFileReader>(
io::global_local_filesystem(),
std::string {segment_v2::InvertedIndexDescriptor::get_index_file_path_prefix(
output_segment_path.value())},
InvertedIndexStorageFormatPB::V2);
EXPECT_TRUE(reader->init().ok());
auto dirs = reader->get_all_directories();
EXPECT_TRUE(dirs.has_value());
if (dirs.has_value()) {
for (const auto& [key, _] : dirs.value()) {
output_index_ids.insert(key.first);
}
}
return output_index_ids;
}

StorageEngine* _engine_ref = nullptr;
TabletSharedPtr _tablet;
TabletMetaSharedPtr _tablet_meta;
Expand Down Expand Up @@ -330,7 +460,8 @@ TEST_F(IndexBuilderTest, DropInvertedIndexTest) {
new_dat_file_count++;
}
}
// The index should have been removed
// The index should have been removed. Dropping the last index leaves a schema
// that owns no index file, so the output rowset must carry none.
EXPECT_EQ(old_idx_file_count, 1) << "Tablet path should have 1 .idx file before drop";
EXPECT_EQ(old_dat_file_count, 1) << "Tablet path should have 1 .dat file before drop";
EXPECT_EQ(new_idx_file_count, 0) << "Tablet path should have no .idx file after drop";
Expand Down Expand Up @@ -649,6 +780,26 @@ TEST_F(IndexBuilderTest, BuildInvertedIndexAfterWritingDataTest) {
//EXPECT_TRUE(tablet_schema->has_inverted_index_with_index_id(2));
}

// A schema can own an inverted index whose index file holds nothing: an all-NULL
// VARIANT column extracts no subcolumn, so no logical index directory is ever
// opened and the file is closed with nothing in it. ALTER on such a rowset must
// read that exactly like a rowset written before any index existed -- there is
// nothing to carry over, and every requested index is built from the raw columns.
TEST_F(IndexBuilderTest, BuildIndexOverEmptyIndexFileTest) {
// Without tolerating the empty index file this fails the whole ALTER with
// [E-6004]inverted index file ... is empty.
EXPECT_EQ(build_k2_index_over_unwritten_k1(14695, 15695, true), (std::set<int64_t> {2}));
}

// The same rowset with NO index file at all. This path has tolerated
// INVERTED_INDEX_FILE_NOT_FOUND for years, and it produces exactly the same
// output: the requested index is built, and an index the source never held is
// not invented. Pinning both together is the point -- an empty index file is
// being read the way a missing one already was, not given new semantics.
TEST_F(IndexBuilderTest, BuildIndexOverMissingIndexFileTest) {
EXPECT_EQ(build_k2_index_over_unwritten_k1(14696, 15696, false), (std::set<int64_t> {2}));
}

TEST_F(IndexBuilderTest, BuildAnnIndexAfterWritingDataTest) {
// 0. prepare tablet path
auto tablet_path = _absolute_dir + "/" + std::to_string(14686);
Expand Down
Loading
Loading