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) +}