[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: #67702 (closed)

Problem Summary:

A segment whose schema owns an inverted index always owns an index file,
even
when no logical index had anything to write: an all-NULL VARIANT column
extracts
no subcolumn, so no CLucene directory is ever opened and the container
is closed
with nothing in it. That empty container is committed by `close()`, not
by
`create_file()`, and `IndexFileWriter`'s empty-file path only closed
three named
implementations -- `StreamSinkFileWriter`, `S3FileWriter`,
`PackedFileWriter`.
`LocalFileWriter` and `HdfsFileWriter` were left open, and
`~LocalFileWriter`
aborts -- and DELETES -- a writer it was never asked to close. So on
local
storage the file silently disappeared and `show_nested_index_file`
reported
`E-6003 ... not found` where `E-6004 ... is empty` was the truth.

Closing through `FileWriter::state()` instead of a type whitelist fixes
that, but
on its own it is not enough: it makes one latent `IndexBuilder`
invariant
violation real, and it leaves a second `IndexBuilder` bug -- already
reachable
today on the sink-node and remote paths, where the empty file does
persist --
untouched.

**`IndexBuilder` writes a container for every segment it rewrites**,
including
when the ALTER leaves the schema with no index at all: dropping the last
index,
or adding an index whose column does not exist. Until now the local
writer's
destructor happened to delete those files. Once they are closed they
persist as
orphans that nothing ever links, copies, uploads, removes or checksums
--
`BetaRowset` decides whether to touch the container by asking
`has_inverted_or_ann_index()` of the rowset's own schema, and both
segment-writer
paths (`SegmentFlusher::_create_segment_writer`,
`VerticalBetaRowsetWriter`) already gate creation the same way.
`IndexBuilder`
was the one caller that did not.

**`IndexBuilder` refuses to read an empty container.**
`update_inverted_index_info()` tolerates `E-6003` ("this rowset predates
the
index") but not `E-6004` ("this rowset's index is empty"), even though
both mean
the same thing there -- there is nothing to carry over, build everything
from the
raw columns. `BUILD INDEX` / `ADD INDEX` / `DROP INDEX` on a table with
an
all-NULL VARIANT index therefore fails the whole ALTER with
`[E-6004]inverted index file ... is empty`.

So: three changes, one invariant.

> a rowset's schema owns an inverted/ANN index **iff** every one of its
segments
> owns an index file, possibly zero length

1. `IndexFileWriter::begin_close()` / `finish_close()` close the empty
container
through `FileWriter` for any non-null, non-CLOSED writer, preserving the
   two-phase asynchronous protocol and propagating close errors.
2. `IndexBuilder::handle_single_rowset()` writes no container when the
output
rowset schema owns no index -- on the drop path and the add path alike.
3. `IndexBuilder::update_inverted_index_info()` reads an empty container
exactly
   the way it reads a missing one.

Consumers already accept `E-6004`: queries downgrade unconditionally
(`SegmentIterator::_downgrade_without_index`), index compaction marks
the segment
ineligible and rebuilds, `CollectionStatistics` skips it. SNII is
untouched --
its container always carries a header, so it is never zero length, and
its
`has_container` probe stays `E-6003`-only.

#67702 made change 1 and then updated
`regression-test/plugins/plugin_curl_requester.groovy` and
`test_drop_index_with_format_v2.groovy` to expect `E-6004` after
dropping every
index. That expectation describes the orphan file, so this PR leaves
both at
`E-6003` and removes the orphan instead.

### Release note

Persist empty V2/V3 inverted index files through every FileWriter
implementation
instead of only the remote ones; stop failing ALTER on a segment whose
index file
is legitimately empty; stop leaving an unreferenced index file behind
when the
last index of a table is dropped.

be/src/storage/index/index_file_writer.cpp

@@ -27,7 +27,6 @@
 #include "common/config.h"
 #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"
@@ -508,10 +507,16 @@ Status IndexFileWriter::begin_close() {
         return _idx_v2_writer->close(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();
@@ -556,10 +561,11 @@ Status IndexFileWriter::finish_close() {
         return Status::OK();
     }
     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();

be/src/storage/task/index_builder.cpp

@@ -386,7 +386,14 @@ 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 -- an all-NULL VARIANT column) both mean the
+                // same thing here: there is nothing to carry over. Both leave the
+                // reader un-inited, so get_all_directories() yields an empty map
+                // and 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(
@@ -446,8 +453,23 @@ Status IndexBuilder::handle_single_rowset(RowsetMetaSharedPtr output_rowset_meta
                       << " rowset_id=" << output_rowset_meta->rowset_id().to_string();
             return Status::OK();
         }
-        if (output_rs_tablet_schema->get_inverted_index_storage_format() !=
-            InvertedIndexStorageFormatPB::V1) {
+        // Dropping the last index leaves a schema that owns no index file at all,
+        // and a rowset must not keep a file its own schema does not claim: link,
+        // copy, upload, remove and CRC all decide whether to touch the container
+        // by asking has_inverted_or_ann_index() of the rowset's own schema, so a
+        // file written here would be linked by nobody and deleted by nobody.
+        // Both segment writer paths already gate container creation the same way;
+        // this is the one caller that did not. It used to be masked by
+        // ~LocalFileWriter deleting a file that was never closed -- an accident,
+        // not a design, and one that never held for remote storage.
+        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_or_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();
@@ -516,12 +538,26 @@ 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();
+        // Same invariant as the drop branch above, and it holds for every storage
+        // format: no index in the output schema, no index file. Reached when
+        // nothing the request asked for survived schema resolution (every
+        // requested column is missing) and the input rowset carried no index
+        // either -- SNII would otherwise seal a header-only container here.
+        if (!output_rowset_schema->has_inverted_or_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();
+        }
         if (output_rowset_schema->get_inverted_index_storage_format() ==
             InvertedIndexStorageFormatPB::SNII) {
             return _handle_single_rowset_snii(output_rowset_meta, segments);
         }
+        // fs() looks the tablet up to pick the encryption algorithm, so resolve it
+        // only on the V2/V3 path that writes an index file.
+        const auto& fs = output_rowset_meta->fs();
         size_t inverted_index_size = 0;
         for (auto& seg_ptr : segments) {
             std::string index_path_prefix {

be/test/io/fs/s3_file_writer_test.cpp

@@ -1530,11 +1530,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<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";
@@ -1542,8 +1544,20 @@ TEST_F(S3FileWriterTest, test_empty_file) {
     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

be/test/storage/index/index_builder_test.cpp

@@ -21,6 +21,7 @@
 #include <gtest/gtest.h>
 
 #include <filesystem>
+#include <set>
 
 #include "common/config.h"
 #include "storage/index/index_file_reader.h"
@@ -534,6 +535,129 @@ protected:
         EXPECT_FALSE(index_exists);
     }
 
+    // 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.
+    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_storage_block();
+            auto columns = std::move(block).mutate_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));
+            }
+            block.set_columns(std::move(columns));
+            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;
@@ -706,7 +830,9 @@ 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 at all, so the output rowset must carry none --
+    // nothing would ever link, copy, upload or remove it.
     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";
@@ -1025,6 +1151,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);
@@ -3507,6 +3653,28 @@ TEST_F(IndexBuilderTest, DropOneIndexNotAffectOtherIndexesOnSameColumnTest) {
             << "Should have exactly 1 inverted index remaining after drop";
 }
 
+// SNII writes a header even when it holds no logical index, so the no-index gate
+// has to run before the SNII dispatch or a schema without an index gets a file.
+TEST_F(IndexBuilderTest, BuildSniiIndexWithNoSurvivingIndexWritesNoFile) {
+    const auto tablet_path = _absolute_dir + "/15697";
+    auto tablet_schema = create_snii_schema({});
+    TabletSharedPtr tablet;
+    ASSERT_TRUE(create_snii_drop_tablet(tablet_schema, tablet_path, &tablet).ok());
+    RowsetSharedPtr source_rowset;
+    ASSERT_TRUE(create_snii_source_rowset(tablet, tablet_schema, tablet_path, &source_rowset).ok());
+    ASSERT_EQ(source_rowset->index_disk_size(), 0);
+
+    // Requested on a column this schema does not have, so nothing survives schema
+    // resolution and the output schema still owns no index.
+    std::vector<RowsetSharedPtr> output_rowsets;
+    ASSERT_TRUE(build_snii_index(tablet,
+                                 {create_build_index(9, "idx_missing", "does_not_exist", -1, {})},
+                                 &output_rowsets)
+                        .ok());
+    ASSERT_EQ(output_rowsets.size(), 1);
+    assert_last_snii_index_dropped(source_rowset, output_rowsets[0]);
+}
+
 TEST_F(IndexBuilderTest, DropOneSniiIndexPreservesSurvivingPhysicalIndex) {
     const auto tablet_path = _absolute_dir + "/15691";
     auto tablet_schema = create_snii_drop_schema();

be/test/storage/index/inverted/empty_index_file_test.cpp

@@ -20,9 +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 {
@@ -30,7 +33,7 @@ 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";
+constexpr static std::string_view tmp_dir = "./ut_dir/empty_index_file";
 class EmptyIndexFileTest : public testing::TestWithParam<InvertedIndexStorageFormatPB> {
     struct WriteState {
         size_t bytes_appended = 0;
@@ -74,6 +77,58 @@ public:
     ~EmptyIndexFileTest() override = default;
 
 protected:
+    // 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<bool> 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<segment_v2::IndexFileWriter> make_index_writer(
+            io::FileWriterPtr file_writer, InvertedIndexStorageFormatPB format) {
+        return std::make_unique<segment_v2::IndexFileWriter>(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);
@@ -101,15 +156,17 @@ protected:
 };
 
 TEST_P(EmptyIndexFileTest, PreservesZeroByteFileWhenNoLogicalIndexes) {
-    io::FileWriterPtr file_writer = std::make_unique<io::StreamSinkFileWriter>(_streams);
-    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, GetParam(), std::move(file_writer), false);
-    EXPECT_TRUE(index_file_writer->begin_close().ok());
-    EXPECT_TRUE(index_file_writer->finish_close().ok());
+    auto file_writer = std::make_unique<io::StreamSinkFileWriter>(_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);
@@ -117,6 +174,123 @@ TEST_P(EmptyIndexFileTest, PreservesZeroByteFileWhenNoLogicalIndexes) {
     }
 }
 
+TEST_P(EmptyIndexFileTest, ClosesOpaqueWriterWithoutAppending) {
+    auto file_writer = std::make_unique<RecordingFileWriter>();
+    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<bool> {true}));
+    EXPECT_EQ(recording->state(), io::FileWriter::State::ASYNC_CLOSING);
+    ASSERT_TRUE(index_writer->finish_close().ok());
+    EXPECT_EQ(recording->close_calls, (std::vector<bool> {true, false}));
+    EXPECT_EQ(recording->state(), io::FileWriter::State::CLOSED);
+    ASSERT_TRUE(index_writer->finish_close().ok());
+    EXPECT_EQ(recording->close_calls, (std::vector<bool> {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<RecordingFileWriter>();
+    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<RecordingFileWriter>();
+    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<bool> {true}));
+}
+
+TEST_P(EmptyIndexFileTest, PropagatesBeginCloseError) {
+    auto file_writer = std::make_unique<RecordingFileWriter>();
+    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<bool> {true}));
+}
+
+TEST_P(EmptyIndexFileTest, PropagatesFinishCloseError) {
+    auto file_writer = std::make_unique<RecordingFileWriter>();
+    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<bool> {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();
+    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<segment_v2::IndexFileReader>(fs, index_path_prefix(), GetParam());
+    auto st = reader->init();
+    EXPECT_TRUE(st.is<ErrorCode::INVERTED_INDEX_BYPASS>()) << 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<segment_v2::IndexFileReader>(
+            io::global_local_filesystem(), std::string(tmp_dir) + "/absent_0", GetParam());
+    auto st = reader->init();
+    EXPECT_TRUE(st.is<ErrorCode::INVERTED_INDEX_FILE_NOT_FOUND>()) << st;
+}
+
 INSTANTIATE_TEST_SUITE_P(LegacyCompoundFormats, EmptyIndexFileTest,
                          testing::Values(InvertedIndexStorageFormatPB::V2,
                                          InvertedIndexStorageFormatPB::V3));

be/test/storage/variant/index_storage_variant_debug_point_test.cpp

@@ -147,11 +147,8 @@ protected:
         EXPECT_TRUE(probe.has_value()) << probe.error();
         if (probe.has_value()) {
             EXPECT_EQ(probe->contains_relative_path(std::string(kArrayPath)), !array_is_null);
-            if (array_is_null) {
-                EXPECT_EQ(probe->index_files.existing_files, 0);
-            } else {
-                expect_index_files(probe.value(), true);
-            }
+            // The schema owns the index, so a null array still leaves an index file.
+            expect_index_files(probe.value(), true);
         }
         return rowset.value();
     }

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

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"))
-
-    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"))
+    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"
+            );
+        """
     }
-}
+
+    // 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"))
+    }
+
+    // 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)
+}