From ed58d2950138ac783e31feab93bb6d66ce3c672b Mon Sep 17 00:00:00 2001 From: Jack Date: Wed, 16 Sep 2026 15:13:36 +0800 Subject: [PATCH] branch-4.1: [fix](inverted index) Persist empty index files and stop writing orphan ones #67859 ### What problem does this PR solve? Issue Number: N/A Related PR: #67859 Problem Summary: An all-NULL VARIANT column can legitimately produce a zero-byte V2/V3 index container. Close empty containers through the FileWriter interface so local, HDFS, stream, S3, packed, and future implementations all persist the same state. Let IndexBuilder treat an empty source container like a missing one, and do not write an orphan container when the output rowset schema owns no inverted or ANN index. This branch does not contain SNII, so the source PR's SNII dispatch and test are not applicable. The V2/V3 behavior is preserved using the branch-4.1 TabletSchema predicates and APIs. Validation: - BUILD_TYPE=ASAN ./build.sh --be -j32 - ./build.sh --fe -j16 - Focused ASAN BE unit tests: 25 passed, 0 failed - test_empty_index_file_lifecycle: passed - test_variant_empty_index_file: passed - clang-format 16 and git diff --check: passed - Targeted changed-line clang-tidy checks: passed ### Release note Persist legitimate empty V2/V3 inverted-index files, allow index rebuilds to consume them, and avoid orphan index files after the last index is gone. ### Check List (For Author) - Test - [x] Regression test - [x] Unit Test - [ ] Manual test (add detailed scripts or steps below) - [ ] No need to test or manual test. Explain why: - [ ] This is a refactor/code format and no logic has been changed. - [ ] Previous test can cover this change. - [ ] No code files have been changed. - [ ] Other reason - Behavior changed: - [ ] No. - [x] Yes. Empty index containers are persisted, accepted by IndexBuilder, and omitted when the output schema owns no index. - Does this need documentation? - [x] No. This restores the existing rowset/index-file invariant. - [ ] Yes. (cherry picked from commit 8d9e9c3f6d0ff2186a6ac52bb572a6fbee457938) Conflicts: - Adapted TabletSchema index predicates and block construction to branch-4.1. - Omitted SNII-only code and coverage because branch-4.1 has no SNII format. - Preserved branch-specific VARIANT debug-point expectations and disabled the branch-4.1 file cache inside the isolated S3 unit test. --- be/src/storage/index/index_file_writer.cpp | 24 +- be/src/storage/task/index_builder.cpp | 33 ++- be/test/io/fs/s3_file_writer_test.cpp | 24 +- be/test/storage/index/index_builder_test.cpp | 153 ++++++++++- .../index/inverted/empty_index_file_test.cpp | 253 ++++++++++++++++-- ...index_storage_variant_debug_point_test.cpp | 1 + .../test_empty_index_file_lifecycle.groovy | 178 ++++++++++++ .../test_variant_empty_index_file.groovy | 95 ++++--- 8 files changed, 682 insertions(+), 79 deletions(-) create mode 100644 regression-test/suites/inverted_index_p0/test_empty_index_file_lifecycle.groovy diff --git a/be/src/storage/index/index_file_writer.cpp b/be/src/storage/index/index_file_writer.cpp index 3c34326cd88bde..6ee16f2819b7c0 100644 --- a/be/src/storage/index/index_file_writer.cpp +++ b/be/src/storage/index/index_file_writer.cpp @@ -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" @@ -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(_idx_v2_writer.get()) != nullptr || - dynamic_cast(_idx_v2_writer.get()) != nullptr || - dynamic_cast(_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(); @@ -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(_idx_v2_writer.get()) != nullptr || - dynamic_cast(_idx_v2_writer.get()) != nullptr || - dynamic_cast(_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(); diff --git a/be/src/storage/task/index_builder.cpp b/be/src/storage/task/index_builder.cpp index 172b868efc2f5c..8d8811b4b6f3a9 100644 --- a/be/src/storage/task/index_builder.cpp +++ b/be/src/storage/task/index_builder.cpp @@ -286,7 +286,12 @@ Status IndexBuilder::update_inverted_index_info() { st = Status::Error( "debug point: reader init error"); }) - if (!st.ok() && !st.is()) { + // 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() && + !st.is()) { return st; } _index_file_readers.emplace( @@ -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(); @@ -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 { diff --git a/be/test/io/fs/s3_file_writer_test.cpp b/be/test/io/fs/s3_file_writer_test.cpp index 5678381b26ba61..0bb6f3c615b46f 100644 --- a/be/test/io/fs/s3_file_writer_test.cpp +++ b/be/test/io/fs/s3_file_writer_test.cpp @@ -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 paths; paths.emplace_back(std::string("tmp_dir"), 1024000000); auto tmp_file_dirs = std::make_unique(paths); @@ -1537,11 +1541,13 @@ 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(S3ClientConf {}); auto mock_client = std::make_shared(); holder->_client = mock_client; dynamic_cast(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"; @@ -1549,8 +1555,20 @@ TEST_F(S3FileWriterTest, test_empty_file) { auto index_file_writer = std::make_unique( 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 diff --git a/be/test/storage/index/index_builder_test.cpp b/be/test/storage/index/index_builder_test.cpp index dd36ba3ab33159..bba51abc9c3bf1 100644 --- a/be/test/storage/index/index_builder_test.cpp +++ b/be/test/storage/index/index_builder_test.cpp @@ -20,6 +20,9 @@ #include #include +#include + +#include "storage/index/index_file_reader.h" #include "storage/olap_common.h" #include "storage/rowset/beta_rowset.h" #include "storage/rowset/rowset_factory.h" @@ -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 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( + io::global_local_filesystem(), source_prefix, InvertedIndexStorageFormatPB::V2); + auto st = reader->init(); + EXPECT_TRUE(leave_empty_index_file ? st.is() + : st.is()) + << 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 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( + 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; @@ -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"; @@ -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 {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 {2})); +} + TEST_F(IndexBuilderTest, BuildAnnIndexAfterWritingDataTest) { // 0. prepare tablet path auto tablet_path = _absolute_dir + "/" + std::to_string(14686); diff --git a/be/test/storage/index/inverted/empty_index_file_test.cpp b/be/test/storage/index/inverted/empty_index_file_test.cpp index 94f1a6493209bb..cf645b1ba8a0ae 100644 --- a/be/test/storage/index/inverted/empty_index_file_test.cpp +++ b/be/test/storage/index/inverted/empty_index_file_test.cpp @@ -20,8 +20,12 @@ #include "exec/sink/load_stream_stub.h" #include "gtest/gtest_pred_impl.h" +#include "io/fs/local_file_system.h" #include "io/fs/stream_sink_file_writer.h" +#include "storage/index/index_file_reader.h" #include "storage/index/index_file_writer.h" +#include "storage/index/index_writer.h" +#include "storage/index/inverted/inverted_index_desc.h" #include "storage/olap_common.h" namespace doris { @@ -29,36 +33,109 @@ namespace doris { constexpr int64_t LOAD_ID_LO = 1; constexpr int64_t LOAD_ID_HI = 2; constexpr int64_t NUM_STREAM = 3; -constexpr static std::string_view tmp_dir = "./ut_dir/tmp"; -class EmptyIndexFileTest : public testing::Test { +constexpr static std::string_view tmp_dir = "./ut_dir/empty_index_file"; +class EmptyIndexFileTest : public testing::TestWithParam { + struct WriteState { + size_t bytes_appended = 0; + int data_calls = 0; + int eos_calls = 0; + }; + class MockStreamStub : public LoadStreamStub { public: - MockStreamStub(PUniqueId load_id, int64_t src_id) + MockStreamStub(PUniqueId load_id, int64_t src_id, std::shared_ptr state) : LoadStreamStub(load_id, src_id, std::make_shared(), - std::make_shared()) {}; + std::make_shared()), + _state(std::move(state)) {}; - virtual ~MockStreamStub() = default; + ~MockStreamStub() override = default; // APPEND_DATA - virtual Status append_data(int64_t partition_id, int64_t index_id, int64_t tablet_id, - int32_t segment_id, uint64_t offset, std::span data, - bool segment_eos = false, - FileType file_type = FileType::SEGMENT_FILE) override { - EXPECT_TRUE(segment_eos); + Status append_data(int64_t partition_id, int64_t index_id, int64_t tablet_id, + int32_t segment_id, uint64_t offset, std::span data, + bool segment_eos = false, + FileType file_type = FileType::SEGMENT_FILE) override { + EXPECT_EQ(offset, _state->bytes_appended); + if (segment_eos) { + ++_state->eos_calls; + EXPECT_TRUE(data.empty()); + return Status::OK(); + } + ++_state->data_calls; + for (const auto& slice : data) { + _state->bytes_appended += slice.size; + } return Status::OK(); } + + private: + std::shared_ptr _state; }; public: EmptyIndexFileTest() = default; - ~EmptyIndexFileTest() = default; + ~EmptyIndexFileTest() override = default; protected: - virtual void SetUp() { + // Implements FileWriter and NOTHING else: no concrete writer type can be + // recognised here, so a close path that dispatches on the implementation + // instead of the interface leaves this writer open and fails the test. + class RecordingFileWriter final : public io::FileWriter { + public: + Status close(bool non_block = false) override { + close_calls.push_back(non_block); + EXPECT_NE(_state, State::CLOSED); + if (non_block) { + EXPECT_EQ(_state, State::OPENED); + RETURN_IF_ERROR(begin_status); + _state = close_synchronously ? State::CLOSED : State::ASYNC_CLOSING; + return Status::OK(); + } + EXPECT_EQ(_state, State::ASYNC_CLOSING); + _state = State::CLOSED; + return finish_status; + } + + Status appendv(const Slice* data, size_t data_cnt) override { + ++append_calls; + for (size_t i = 0; i < data_cnt; ++i) { + _bytes_appended += data[i].size; + } + return Status::OK(); + } + + const io::Path& path() const override { return _path; } + size_t bytes_appended() const override { return _bytes_appended; } + State state() const override { return _state; } + + std::vector close_calls; + int append_calls = 0; + bool close_synchronously = false; + Status begin_status = Status::OK(); + Status finish_status = Status::OK(); + + private: + io::Path _path {"recording_0.idx"}; + size_t _bytes_appended = 0; + State _state = State::OPENED; + }; + + static std::string index_path_prefix() { return std::string(tmp_dir) + "/empty_0"; } + + static std::unique_ptr make_index_writer( + io::FileWriterPtr file_writer, InvertedIndexStorageFormatPB format) { + return std::make_unique(io::global_local_filesystem(), + index_path_prefix(), "empty", 0, + format, std::move(file_writer), false); + } + + void SetUp() override { _load_id.set_hi(LOAD_ID_HI); _load_id.set_lo(LOAD_ID_LO); for (int src_id = 0; src_id < NUM_STREAM; src_id++) { - _streams.emplace_back(new MockStreamStub(_load_id, src_id)); + auto state = std::make_shared(); + _write_states.push_back(state); + _streams.emplace_back(new MockStreamStub(_load_id, src_id, std::move(state))); } EXPECT_TRUE(io::global_local_filesystem()->delete_directory(tmp_dir).ok()); EXPECT_TRUE(io::global_local_filesystem()->create_directory(tmp_dir).ok()); @@ -69,25 +146,153 @@ class EmptyIndexFileTest : public testing::Test { ExecEnv::GetInstance()->set_tmp_file_dir(std::move(tmp_file_dirs)); } - virtual void TearDown() { + void TearDown() override { EXPECT_TRUE(io::global_local_filesystem()->delete_directory(tmp_dir).ok()); } PUniqueId _load_id; std::vector> _streams; + std::vector> _write_states; }; -TEST_F(EmptyIndexFileTest, test_empty_index_file) { - io::FileWriterPtr file_writer = std::make_unique(_streams); +TEST_P(EmptyIndexFileTest, PreservesZeroByteFileWhenNoLogicalIndexes) { + auto file_writer = std::make_unique(_streams); + file_writer->init(_load_id, 1, 2, 3, 0, FileType::INVERTED_INDEX_FILE); + auto* stream_writer = file_writer.get(); + auto index_file_writer = make_index_writer(std::move(file_writer), GetParam()); + ASSERT_TRUE(index_file_writer->begin_close().ok()); + EXPECT_EQ(stream_writer->state(), io::FileWriter::State::ASYNC_CLOSING); + ASSERT_TRUE(index_file_writer->finish_close().ok()); + EXPECT_EQ(stream_writer->state(), io::FileWriter::State::CLOSED); + // Finishing an already closed empty file must not send a second EOS. + ASSERT_TRUE(index_file_writer->finish_close().ok()); + index_file_writer.reset(); + for (const auto& state : _write_states) { + EXPECT_EQ(state->bytes_appended, 0); + EXPECT_EQ(state->data_calls, 0); + EXPECT_EQ(state->eos_calls, 1); + } +} + +TEST_P(EmptyIndexFileTest, ClosesOpaqueWriterWithoutAppending) { + auto file_writer = std::make_unique(); + auto* recording = file_writer.get(); + auto index_writer = make_index_writer(std::move(file_writer), GetParam()); + + ASSERT_TRUE(index_writer->begin_close().ok()); + EXPECT_EQ(recording->close_calls, (std::vector {true})); + EXPECT_EQ(recording->state(), io::FileWriter::State::ASYNC_CLOSING); + ASSERT_TRUE(index_writer->finish_close().ok()); + EXPECT_EQ(recording->close_calls, (std::vector {true, false})); + EXPECT_EQ(recording->state(), io::FileWriter::State::CLOSED); + ASSERT_TRUE(index_writer->finish_close().ok()); + EXPECT_EQ(recording->close_calls, (std::vector {true, false})); + // An empty index file is empty: closing it must not write a header. + EXPECT_EQ(recording->append_calls, 0); + EXPECT_EQ(recording->bytes_appended(), 0); +} + +TEST_P(EmptyIndexFileTest, SkipsAlreadyClosedWriter) { + auto file_writer = std::make_unique(); + auto* recording = file_writer.get(); + ASSERT_TRUE(recording->close(true).ok()); + ASSERT_TRUE(recording->close(false).ok()); + recording->close_calls.clear(); + auto index_writer = make_index_writer(std::move(file_writer), GetParam()); + ASSERT_TRUE(index_writer->begin_close().ok()); + ASSERT_TRUE(index_writer->finish_close().ok()); + EXPECT_TRUE(recording->close_calls.empty()); +} + +TEST_P(EmptyIndexFileTest, SkipsFinishWhenBeginClosesSynchronously) { + auto file_writer = std::make_unique(); + auto* recording = file_writer.get(); + recording->close_synchronously = true; + auto index_writer = make_index_writer(std::move(file_writer), GetParam()); + ASSERT_TRUE(index_writer->begin_close().ok()); + EXPECT_EQ(recording->state(), io::FileWriter::State::CLOSED); + ASSERT_TRUE(index_writer->finish_close().ok()); + EXPECT_EQ(recording->close_calls, (std::vector {true})); +} + +TEST_P(EmptyIndexFileTest, PropagatesBeginCloseError) { + auto file_writer = std::make_unique(); + auto* recording = file_writer.get(); + recording->begin_status = Status::IOError("begin close failed"); + auto index_writer = make_index_writer(std::move(file_writer), GetParam()); + auto st = index_writer->begin_close(); + EXPECT_EQ(st.to_string(), recording->begin_status.to_string()); + EXPECT_EQ(recording->close_calls, (std::vector {true})); +} + +TEST_P(EmptyIndexFileTest, PropagatesFinishCloseError) { + auto file_writer = std::make_unique(); + auto* recording = file_writer.get(); + recording->finish_status = Status::IOError("finish close failed"); + auto index_writer = make_index_writer(std::move(file_writer), GetParam()); + ASSERT_TRUE(index_writer->begin_close().ok()); + auto st = index_writer->finish_close(); + EXPECT_EQ(st.to_string(), recording->finish_status.to_string()); + EXPECT_EQ(recording->close_calls, (std::vector {true, false})); +} + +TEST_P(EmptyIndexFileTest, AllowsNullWriter) { + // V1 keeps one file per logical index and never owns a container; V2/V3 own + // one but may be constructed without it (the drop path for a V1 rowset). + for (auto format : {InvertedIndexStorageFormatPB::V1, GetParam()}) { + auto index_writer = make_index_writer(nullptr, format); + ASSERT_TRUE(index_writer->begin_close().ok()); + ASSERT_TRUE(index_writer->finish_close().ok()); + } +} + +TEST_P(EmptyIndexFileTest, PreservesLocalEmptyFileAfterDestruction) { 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( - 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()); + const auto path = + segment_v2::InvertedIndexDescriptor::get_index_file_path_v2(index_path_prefix()); + io::FileWriterPtr file_writer; + ASSERT_TRUE(fs->create_file(path, &file_writer).ok()); + auto* local_writer = file_writer.get(); + bool exists = false; + ASSERT_TRUE(fs->exists(path, &exists).ok()); + ASSERT_TRUE(exists); // O_CREAT alone does not guarantee persistence. + auto index_writer = make_index_writer(std::move(file_writer), GetParam()); + ASSERT_TRUE(index_writer->begin_close().ok()); + EXPECT_EQ(local_writer->state(), io::FileWriter::State::ASYNC_CLOSING); + ASSERT_TRUE(index_writer->finish_close().ok()); + EXPECT_EQ(local_writer->state(), io::FileWriter::State::CLOSED); + EXPECT_EQ(local_writer->bytes_appended(), 0); + // ~LocalFileWriter aborts (and DELETES) a writer it was never asked to close. + index_writer.reset(); + + ASSERT_TRUE(fs->exists(path, &exists).ok()); + ASSERT_TRUE(exists); + int64_t file_size = -1; + ASSERT_TRUE(fs->file_size(path, &file_size).ok()); + EXPECT_EQ(file_size, 0); + + // The contract every reader of this file relies on: a zero-length container + // is not corruption, it is "this segment has no index data". Query falls back + // to a non-indexed evaluation on it and IndexBuilder treats it as "nothing to + // carry over"; both branch on this exact error code. + auto reader = + std::make_unique(fs, index_path_prefix(), GetParam()); + auto st = reader->init(); + EXPECT_TRUE(st.is()) << st; + EXPECT_NE(st.to_string().find(" is empty"), std::string::npos) << st; +} + +TEST_P(EmptyIndexFileTest, MissingLocalFileReadsAsFileNotFound) { + // The other half of the same contract: a rowset written before any index + // existed has no container at all, and that is distinct from an empty one. + auto reader = std::make_unique( + io::global_local_filesystem(), std::string(tmp_dir) + "/absent_0", GetParam()); + auto st = reader->init(); + EXPECT_TRUE(st.is()) << st; } +INSTANTIATE_TEST_SUITE_P(LegacyCompoundFormats, EmptyIndexFileTest, + testing::Values(InvertedIndexStorageFormatPB::V2, + InvertedIndexStorageFormatPB::V3)); + } // namespace doris diff --git a/be/test/storage/variant/index_storage_variant_debug_point_test.cpp b/be/test/storage/variant/index_storage_variant_debug_point_test.cpp index 720a83b342a4e7..180614adedb6ea 100644 --- a/be/test/storage/variant/index_storage_variant_debug_point_test.cpp +++ b/be/test/storage/variant/index_storage_variant_debug_point_test.cpp @@ -136,6 +136,7 @@ class IndexStorageVariantDebugPointTest : public IndexStorageTestFixture { auto probe = probe_rowset(rowset.value()); EXPECT_TRUE(probe.has_value()) << probe.error(); if (probe.has_value()) { + // The schema owns the index, so a null array still leaves an index file. expect_index_files(probe.value(), true); } return rowset.value(); diff --git a/regression-test/suites/inverted_index_p0/test_empty_index_file_lifecycle.groovy b/regression-test/suites/inverted_index_p0/test_empty_index_file_lifecycle.groovy new file mode 100644 index 00000000000000..04b53207ae9104 --- /dev/null +++ b/regression-test/suites/inverted_index_p0/test_empty_index_file_lifecycle.groovy @@ -0,0 +1,178 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +// Lifecycle of a segment whose index file is legitimately empty. +// +// An all-NULL VARIANT column extracts no subcolumn, so the only indexed column +// opens no logical index and the segment's index file is closed with nothing in +// it. The invariant, in both directions: +// +// rowset schema owns an inverted index <=> every segment owns an index file +// +// An empty index file (E-6004 "is empty") is therefore expected wherever a +// rowset's schema still has the index, and NO index file at all (E-6003 "not +// found") is expected once it has none. Every producer has to keep the first half +// true -- load, compaction, light and direct schema change, BUILD INDEX -- and +// every consumer has to accept an empty index file instead of failing on it. +suite("test_empty_index_file_lifecycle", "p0") { + def backendId_to_backendIP = [:] + def backendId_to_backendHttpPort = [:] + getBackendIpHttpPort(backendId_to_backendIP, backendId_to_backendHttpPort) + + // show_nested_index_file reports the first rowset it cannot open, so there is + // one status per tablet: "E-6004" empty index file, "E-6003" none at all, + // "OK" an index file with real indexes in it. + def index_file_status = { String tableName -> + def statuses = [] as Set + for (def tablet : sql_return_maparray(" show tablets from ${tableName} ")) { + String ip = backendId_to_backendIP.get(tablet.BackendId) + String port = backendId_to_backendHttpPort.get(tablet.BackendId) + def (code, out, err) = http_client("GET", String.format( + "http://%s:%s/api/show_nested_index_file?tablet_id=%s", ip, port, tablet.TabletId)) + logger.info("show_nested_index_file tablet=${tablet.TabletId}: code=${code}, out=${out}, err=${err}") + statuses.add(code == 500 ? parseJson(out.trim()).status : "OK") + } + return statuses + } + + // Both SHOW lists are empty for a table that never had such a job, so this + // waits for "nothing pending" rather than for a FINISHED row to appear. + def wait_alter_done = { String tableName -> + for (int i = 0; i < 600; i++) { + def jobs = sql_return_maparray(""" SHOW ALTER TABLE COLUMN WHERE TableName = "${tableName}" """) + + sql_return_maparray(""" SHOW BUILD INDEX WHERE TableName = "${tableName}" """) + def cancelled = jobs.findAll { it.State == "CANCELLED" } + assertTrue(cancelled.isEmpty(), "job cancelled on ${tableName}: ${cancelled}") + if (jobs.every { it.State == "FINISHED" }) { + return + } + sleep(1000) + } + assertTrue(false, "schema change or index job on ${tableName} did not finish") + } + + // The rewrite lands asynchronously on the BE, so poll the invariant instead + // of racing the FE job record. + def assert_index_file_status = { String tableName, String expected, String step -> + def observed = null + for (int i = 0; i < 180; i++) { + observed = index_file_status(tableName) + if (observed == ([expected] as Set)) { + return + } + sleep(1000) + } + assertEquals([expected] as Set, observed, "after ${step}") + } + + def assert_readable = { String tableName, int rows = 2 -> + assertEquals(rows, sql(" select count(*) from ${tableName} ")[0][0] as int) + assertEquals(1, sql(" select count(*) from ${tableName} where id = 1 ")[0][0] as int) + } + + // v2 covers both write paths -- memtable on the sink node streams the index + // file through StreamSinkFileWriter, the local path writes it through + // LocalFileWriter -- and v3 covers the other storage format. + [["v2", true], ["v2", false], ["v3", false]].each { fmt, sinkNode -> + def tableName = "test_empty_idx_lifecycle_${fmt}_sink${sinkNode ? 1 : 0}" + sql " drop table if exists ${tableName} " + sql """ + CREATE TABLE ${tableName} ( + `id` bigint NOT NULL, + `v` variant NULL, + INDEX v_idx (`v`) USING INVERTED + ) DUPLICATE KEY(`id`) + DISTRIBUTED BY HASH(`id`) BUCKETS 1 + PROPERTIES ( + "replication_allocation" = "tag.location.default: 1", + "inverted_index_storage_format" = "${fmt}", + "disable_auto_compaction" = "true" + ); + """ + sql " set enable_memtable_on_sink_node = ${sinkNode} " + + // 1. load: the only indexed column is all NULL, so the index file is empty + sql " insert into ${tableName} values (1, NULL) " + sql " insert into ${tableName} values (2, NULL) " + sql "sync" + assert_index_file_status(tableName, "E-6004", "load") + assert_readable(tableName) + + // The steps below do not depend on the write path that loaded the rows, so + // they run once. + if (sinkNode) { + return + } + + // 2. compaction: two empty-index inputs produce one empty-index output + trigger_and_wait_compaction(tableName, "full") + assert_index_file_status(tableName, "E-6004", "full compaction") + assert_readable(tableName) + + // 3. light schema change: a default-valued ADD COLUMN keeps the segments + // and their index files (locally it hard-links them) + sql """ ALTER TABLE ${tableName} ADD COLUMN c INT DEFAULT "7" """ + wait_alter_done(tableName) + assert_index_file_status(tableName, "E-6004", "light schema change") + assert_readable(tableName) + + // 4. direct schema change: a type change rewrites every rowset, so the + // output segments write their own empty index files. The default has + // to be repeated, or FE rejects the statement as a default value change. + sql """ ALTER TABLE ${tableName} MODIFY COLUMN c BIGINT DEFAULT "7" """ + wait_alter_done(tableName) + assert_index_file_status(tableName, "E-6004", "direct schema change") + assert_readable(tableName) + + if (!isCloudMode()) { + // 5. ADD INDEX + BUILD INDEX runs IndexBuilder over the existing index + // files. An empty one means "nothing to carry over", exactly like a + // rowset written before any index existed; failing on it aborts the + // whole ALTER with "[E-6004] ... is empty". + sql " ALTER TABLE ${tableName} ADD INDEX id_idx (`id`) USING INVERTED " + wait_alter_done(tableName) + build_index_on_table("id_idx", tableName) + wait_alter_done(tableName) + assert_index_file_status(tableName, "OK", "BUILD INDEX over an empty index file") + assert_readable(tableName) + + // 6. dropping one of two indexes leaves an index in the schema, so the + // rewritten segments keep an index file -- empty again, because the + // surviving index is the all-NULL VARIANT one + sql " ALTER TABLE ${tableName} DROP INDEX id_idx " + wait_alter_done(tableName) + assert_index_file_status(tableName, "E-6004", "dropping one of two indexes") + assert_readable(tableName) + } + + // 7. dropping the LAST index: IndexBuilder does not remove a VARIANT index + // from the rowsets it rewrites, so they keep their empty index file until + // full compaction rewrites them with a schema that owns no index. + sql " ALTER TABLE ${tableName} DROP INDEX v_idx " + wait_alter_done(tableName) + if (!isCloudMode()) { + assert_index_file_status(tableName, "E-6004", "dropping the last index") + } + // Full compaction needs a second rowset, or it has nothing to merge. + sql " insert into ${tableName} (id, v) values (3, NULL) " + if (!isCloudMode()) { + trigger_and_wait_compaction(tableName, "full") + assert_index_file_status(tableName, "E-6003", "compacting a table with no index") + } + assert_readable(tableName, 3) + } +} diff --git a/regression-test/suites/inverted_index_p0/test_variant_empty_index_file.groovy b/regression-test/suites/inverted_index_p0/test_variant_empty_index_file.groovy index 93e4cf7b521487..724eee97f737a7 100644 --- a/regression-test/suites/inverted_index_p0/test_variant_empty_index_file.groovy +++ b/regression-test/suites/inverted_index_p0/test_variant_empty_index_file.groovy @@ -16,47 +16,66 @@ // under the License. suite("test_variant_empty_index_file", "p0") { - def tableName = "test_variant_empty_index_file" - sql """ drop table if exists ${tableName} """ - // create table - sql """ - CREATE TABLE IF NOT EXISTS ${tableName} - ( - `id` bigint NOT NULL, - `v` variant NULL, - INDEX v_idx (`v`) USING INVERTED - ) DUPLICATE KEY(`id`) - DISTRIBUTED BY HASH (`id`) BUCKETS 1 - PROPERTIES ( - "replication_allocation" = "tag.location.default: 1", - "inverted_index_storage_format" = "v2", - "disable_auto_compaction" = "true" - ); - """ - - sql """ set enable_memtable_on_sink_node = true """ - sql """ insert into ${tableName} values (1, NULL) """ - qt_sql9 "select * from ${tableName}" - sql "sync" - def tablets = sql_return_maparray """ show tablets from ${tableName}; """ - def backendId_to_backendIP = [:] def backendId_to_backendHttpPort = [:] getBackendIpHttpPort(backendId_to_backendIP, backendId_to_backendHttpPort); - String tablet_id = tablets[0].TabletId - String backend_id = tablets[0].BackendId - String ip = backendId_to_backendIP.get(backend_id) - String port = backendId_to_backendHttpPort.get(backend_id) - def (code, out, err) = http_client("GET", String.format("http://%s:%s/api/show_nested_index_file?tablet_id=%s", ip, port, tablet_id)) - logger.info("Run show_nested_index_file_on_tablet: code=" + code + ", out=" + out + ", err=" + err) - assertEquals("E-6004", parseJson(out.trim()).status) - assertTrue(out.contains(" is empty")) + def create_table = { String table -> + sql """ drop table if exists ${table} """ + sql """ + CREATE TABLE IF NOT EXISTS ${table} + ( + `id` bigint NOT NULL, + `v` variant NULL, + INDEX v_idx (`v`) USING INVERTED + ) DUPLICATE KEY(`id`) + DISTRIBUTED BY HASH (`id`) BUCKETS 1 + PROPERTIES ( + "replication_allocation" = "tag.location.default: 1", + "inverted_index_storage_format" = "v2", + "disable_auto_compaction" = "true" + ); + """ + } - try { - sql """ select /*+ SET_VAR(enable_match_without_inverted_index = 0) */ * from ${tableName} where v match 'abcd'; """ - } catch (Exception e) { - log.info(e.getMessage()); - assertTrue(e.getMessage().contains("VARIANT root column does not support MATCH predicates")) + // The schema owns an index, so every segment owns an index file -- but an + // all-NULL VARIANT column extracts no subcolumn, so no logical index is ever + // opened and the file is closed empty. Reading it must say "is empty" + // (E-6004), never "not found" (E-6003): the file is there, it just has no + // index in it. + def assert_empty_index_file = { String table -> + sql "sync" + def tablets = sql_return_maparray """ show tablets from ${table}; """ + String tablet_id = tablets[0].TabletId + String backend_id = tablets[0].BackendId + String ip = backendId_to_backendIP.get(backend_id) + String port = backendId_to_backendHttpPort.get(backend_id) + def (code, out, err) = http_client("GET", String.format("http://%s:%s/api/show_nested_index_file?tablet_id=%s", ip, port, tablet_id)) + logger.info("Run show_nested_index_file_on_tablet: code=" + code + ", out=" + out + ", err=" + err) + assertEquals("E-6004", parseJson(out.trim()).status) + assertTrue(out.contains(" is empty")) } -} \ No newline at end of file + + // Memtable on the sink node streams the index file through StreamSinkFileWriter. + def tableName = "test_variant_empty_index_file" + create_table(tableName) + sql """ set enable_memtable_on_sink_node = true """ + sql """ insert into ${tableName} values (1, NULL) """ + qt_sql9 "select * from ${tableName}" + assert_empty_index_file(tableName) + + test { + sql """ select /*+ SET_VAR(enable_match_without_inverted_index = 0) */ + * from ${tableName} where v match 'abcd' """ + exception "VARIANT root column does not support MATCH predicates" + } + + // The local write path goes through LocalFileWriter, whose destructor ABORTS + // (and deletes) a writer that was never closed. Creating the file is not + // enough; it has to be closed like every other implementation. + def localTableName = "test_variant_empty_index_file_local" + create_table(localTableName) + sql """ set enable_memtable_on_sink_node = false """ + sql """ insert into ${localTableName} values (1, NULL) """ + assert_empty_index_file(localTableName) +}