[fix](inverted index) Writing to the inverted index also writes to the file cache.

be/src/olap/compaction.cpp

@@ -686,6 +686,9 @@ Status Compaction::do_inverted_index_compaction() {
                        << st;
             return st;
         }
+        for (const auto& writer : inverted_index_file_writers) {
+            writer->set_file_writer_opts(ctx.get_file_writer_options());
+        }
     }
 
     // use tmp file dir to store index files

be/src/olap/rowset/beta_rowset_writer.cpp

@@ -846,13 +846,7 @@ Status BaseBetaRowsetWriter::_build_tmp(RowsetSharedPtr& rowset_ptr) {
 
 Status BaseBetaRowsetWriter::_create_file_writer(const std::string& path,
                                                  io::FileWriterPtr& file_writer) {
-    io::FileWriterOptions opts {
-            .write_file_cache = _context.write_file_cache,
-            .is_cold_data = _context.is_hot_data,
-            .file_cache_expiration =
-                    _context.file_cache_ttl_sec > 0 && _context.newest_write_timestamp > 0
-                            ? _context.newest_write_timestamp + _context.file_cache_ttl_sec
-                            : 0};
+    io::FileWriterOptions opts = _context.get_file_writer_options();
     Status st = _context.fs()->create_file(path, &file_writer, &opts);
     if (!st.ok()) {
         LOG(WARNING) << "failed to create writable file. path=" << path << ", err: " << st;

be/src/olap/rowset/rowset_writer_context.h

@@ -140,6 +140,16 @@ struct RowsetWriterContext {
             return *storage_resource->fs;
         }
     }
+
+    io::FileWriterOptions get_file_writer_options() const {
+        io::FileWriterOptions opts {
+                .write_file_cache = write_file_cache,
+                .is_cold_data = is_hot_data,
+                .file_cache_expiration = file_cache_ttl_sec > 0 && newest_write_timestamp > 0
+                                                 ? newest_write_timestamp + file_cache_ttl_sec
+                                                 : 0};
+        return opts;
+    }
 };
 
 } // namespace doris

be/src/olap/rowset/segment_v2/inverted_index_file_writer.cpp

@@ -283,6 +283,7 @@ size_t InvertedIndexFileWriter::write_v1() {
             ram_dir.close();
 
             auto* out_dir = DorisFSDirectoryFactory::getDirectory(_fs, idx_path.c_str());
+            out_dir->set_file_writer_opts(_opts);
 
             auto* out = out_dir->createOutput(idx_name.c_str());
             if (out == nullptr) {
@@ -348,6 +349,8 @@ size_t InvertedIndexFileWriter::write_v2() {
     io::Path index_path {InvertedIndexDescriptor::get_index_file_path_v2(_index_path_prefix)};
 
     auto* out_dir = DorisFSDirectoryFactory::getDirectory(_fs, index_path.parent_path().c_str());
+    out_dir->set_file_writer_opts(_opts);
+
     std::unique_ptr<lucene::store::IndexOutput> compound_file_output;
     // idx v2 writer != nullptr means memtable on sink node now
     if (_idx_v2_writer != nullptr) {

be/src/olap/rowset/segment_v2/inverted_index_file_writer.h

@@ -71,6 +71,8 @@ class InvertedIndexFileWriter {
                   lucene::store::IndexOutput* output, uint8_t* buffer, int64_t bufferLength);
     InvertedIndexStorageFormatPB get_storage_format() const { return _storage_format; }
 
+    void set_file_writer_opts(const io::FileWriterOptions& opts) { _opts = opts; }
+
 private:
     InvertedIndexDirectoryMap _indices_dirs;
     const io::FileSystemSPtr _fs;
@@ -81,6 +83,7 @@ class InvertedIndexFileWriter {
     size_t _file_size = 0;
     // write to disk or stream
     io::FileWriterPtr _idx_v2_writer;
+    io::FileWriterOptions _opts;
 };
 } // namespace segment_v2
 } // namespace doris

be/src/olap/rowset/segment_v2/inverted_index_fs_directory.cpp

@@ -84,9 +84,6 @@ namespace doris::segment_v2 {
 const char* const DorisFSDirectory::WRITE_LOCK_FILE = "write.lock";
 
 class DorisFSDirectory::FSIndexOutput : public lucene::store::BufferedIndexOutput {
-private:
-    io::FileWriterPtr _writer;
-
 protected:
     void flushBuffer(const uint8_t* b, const int32_t size) override;
 
@@ -96,6 +93,12 @@ class DorisFSDirectory::FSIndexOutput : public lucene::store::BufferedIndexOutpu
     ~FSIndexOutput() override;
     void close() override;
     int64_t length() const override;
+
+    void set_file_writer_opts(const io::FileWriterOptions& opts) { _opts = opts; }
+
+private:
+    io::FileWriterPtr _writer;
+    io::FileWriterOptions _opts;
 };
 
 class DorisFSDirectory::FSIndexOutputV2 : public lucene::store::BufferedIndexOutput {
@@ -242,7 +245,13 @@ void DorisFSDirectory::FSIndexInput::readInternal(uint8_t* b, const int32_t len)
 }
 
 void DorisFSDirectory::FSIndexOutput::init(const io::FileSystemSPtr& fs, const char* path) {
-    Status status = fs->create_file(path, &_writer);
+    DBUG_EXECUTE_IF("DorisFSDirectory::FSIndexOutput::init.file_cache", {
+        if (fs->type() == io::FileSystemType::S3 && _opts.write_file_cache == false) {
+            _CLTHROWA(CL_ERR_IO, "Inverted index failed to enter file cache");
+        }
+    });
+
+    Status status = fs->create_file(path, &_writer, &_opts);
     DBUG_EXECUTE_IF(
             "DorisFSDirectory::FSIndexOutput._throw_clucene_error_in_fsindexoutput_"
             "init",
@@ -579,6 +588,7 @@ lucene::store::IndexOutput* DorisFSDirectory::createOutput(const char* name) {
         assert(!exists);
     }
     auto* ret = _CLNEW FSIndexOutput();
+    ret->set_file_writer_opts(_opts);
     try {
         ret->init(_fs, fl);
     } catch (CLuceneError& err) {

be/src/olap/rowset/segment_v2/inverted_index_fs_directory.h

@@ -29,6 +29,7 @@
 #include "CLucene/SharedHeader.h"
 #include "io/fs/file_reader_writer_fwd.h"
 #include "io/fs/file_system.h"
+#include "io/fs/file_writer.h"
 #include "io/io_common.h"
 
 class CLuceneError;
@@ -46,8 +47,6 @@ class CLUCENE_EXPORT DorisFSDirectory : public lucene::store::Directory {
 public:
     static const char* const WRITE_LOCK_FILE;
     static const int64_t MAX_HEADER_DATA_SIZE = 1024 * 128; // 128k
-private:
-    int filemode;
 
 protected:
     mutable std::mutex _this_lock;
@@ -91,6 +90,12 @@ class CLUCENE_EXPORT DorisFSDirectory : public lucene::store::Directory {
 
     virtual void init(const io::FileSystemSPtr& fs, const char* path,
                       lucene::store::LockFactory* lock_factory = nullptr);
+
+    void set_file_writer_opts(const io::FileWriterOptions& opts) { _opts = opts; }
+
+private:
+    int32_t filemode;
+    io::FileWriterOptions _opts;
 };
 
 class CLUCENE_EXPORT DorisRAMFSDirectory : public DorisFSDirectory {

be/src/olap/rowset/segment_v2/segment_writer.cpp

@@ -139,6 +139,8 @@ SegmentWriter::SegmentWriter(io::FileWriter* file_writer, uint32_t segment_id,
                 _opts.rowset_ctx->rowset_id.to_string(), segment_id,
                 _tablet_schema->get_inverted_index_storage_format(),
                 std::move(inverted_file_writer));
+        _inverted_index_file_writer->set_file_writer_opts(
+                _opts.rowset_ctx->get_file_writer_options());
     }
 }
 

be/src/olap/rowset/segment_v2/vertical_segment_writer.cpp

@@ -120,6 +120,8 @@ VerticalSegmentWriter::VerticalSegmentWriter(io::FileWriter* file_writer, uint32
                 _opts.rowset_ctx->rowset_id.to_string(), segment_id,
                 _tablet_schema->get_inverted_index_storage_format(),
                 std::move(inverted_file_writer));
+        _inverted_index_file_writer->set_file_writer_opts(
+                _opts.rowset_ctx->get_file_writer_options());
     }
 }
 

be/src/olap/rowset/vertical_beta_rowset_writer.cpp

@@ -165,14 +165,7 @@ Status VerticalBetaRowsetWriter<T>::_create_segment_writer(
     int seg_id = this->_num_segment.fetch_add(1, std::memory_order_relaxed);
 
     io::FileWriterPtr file_writer;
-    io::FileWriterOptions opts {
-            .write_file_cache = this->_context.write_file_cache,
-            .is_cold_data = this->_context.is_hot_data,
-            .file_cache_expiration = this->_context.file_cache_ttl_sec > 0 &&
-                                                     this->_context.newest_write_timestamp > 0
-                                             ? this->_context.newest_write_timestamp +
-                                                       this->_context.file_cache_ttl_sec
-                                             : 0};
+    io::FileWriterOptions opts = this->_context.get_file_writer_options();
 
     auto path = context.segment_path(seg_id);
     auto& fs = context.fs_ref();

regression-test/suites/fault_injection_p0/test_index_writer_file_cache.groovy

@@ -0,0 +1,116 @@
+// 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.
+
+
+suite("test_index_writer_file_cache_fault_injection", "nonConcurrent") {
+    if (!isCloudMode()) {
+        return;
+    }
+
+    def backendId_to_backendIP = [:]
+    def backendId_to_backendHttpPort = [:]
+    getBackendIpHttpPort(backendId_to_backendIP, backendId_to_backendHttpPort);
+
+    def testTable1 = "test_index_writer_file_cache_fault_injection_1"
+    def testTable2 = "test_index_writer_file_cache_fault_injection_2"
+
+    sql "DROP TABLE IF EXISTS ${testTable1}"
+    sql """
+        CREATE TABLE ${testTable1} (
+          `@timestamp` int(11) NULL COMMENT "",
+          `clientip` string NULL COMMENT "",
+          `request` string NULL COMMENT "",
+          `status` int(11) NULL COMMENT "",
+          `size` int(11) NULL COMMENT "",
+          INDEX clientip_idx (`clientip`) USING INVERTED COMMENT '',
+          INDEX request_idx (`request`) USING INVERTED PROPERTIES("parser" = "unicode", "support_phrase" = "true") COMMENT '',
+          INDEX status_idx (`status`) USING INVERTED COMMENT '',
+          INDEX size_idx (`size`) USING INVERTED COMMENT ''
+          ) ENGINE=OLAP
+          DUPLICATE KEY(`@timestamp`)
+          COMMENT "OLAP"
+          DISTRIBUTED BY HASH(`@timestamp`) BUCKETS 1
+          PROPERTIES (
+          "replication_allocation" = "tag.location.default: 1",
+          "disable_auto_compaction" = "true"
+        );
+    """
+
+    sql "DROP TABLE IF EXISTS ${testTable2}"
+    sql """
+        CREATE TABLE ${testTable2} (
+          `@timestamp` int(11) NULL COMMENT "",
+          `clientip` string NULL COMMENT "",
+          `request` string NULL COMMENT "",
+          `status` int(11) NULL COMMENT "",
+          `size` int(11) NULL COMMENT "",
+          INDEX clientip_idx (`clientip`) USING INVERTED COMMENT '',
+          INDEX request_idx (`request`) USING INVERTED PROPERTIES("parser" = "unicode", "support_phrase" = "true") COMMENT '',
+          INDEX status_idx (`status`) USING INVERTED COMMENT '',
+          INDEX size_idx (`size`) USING INVERTED COMMENT ''
+          ) ENGINE=OLAP
+          DUPLICATE KEY(`@timestamp`)
+          COMMENT "OLAP"
+          DISTRIBUTED BY HASH(`@timestamp`) BUCKETS 1
+          PROPERTIES (
+          "replication_allocation" = "tag.location.default: 1",
+          "disable_auto_compaction" = "true"
+        );
+    """
+
+    def insert_and_compaction = { tableName ->
+      sql """ INSERT INTO ${tableName} VALUES (893964617, '40.135.0.0', 'GET /images/hm_bg.jpg HTTP/1.0', 200, 24736); """
+      sql """ INSERT INTO ${tableName} VALUES (893964653, '232.0.0.0', 'GET /images/hm_bg.jpg HTTP/1.0', 200, 3781); """
+      sql """ INSERT INTO ${tableName} VALUES (893964672, '26.1.0.0', 'GET /images/hm_bg.jpg HTTP/1.0', 304, 0); """
+
+      def tablets = sql_return_maparray """ show tablets from ${tableName}; """
+
+      for (def tablet in tablets) {
+        String tablet_id = tablet.TabletId
+        String backend_id = tablet.BackendId
+        def (code, out, err) = be_run_full_compaction(backendId_to_backendIP.get(backend_id), backendId_to_backendHttpPort.get(backend_id), tablet_id)
+        logger.info("Run compaction: code=" + code + ", out=" + out + ", err=" + err)
+        assertEquals(code, 0)
+        def compactJson = parseJson(out.trim())
+        assertEquals("success", compactJson.status.toLowerCase())
+      }
+
+      for (def tablet in tablets) {
+        boolean running = true
+        do {
+            Thread.sleep(1000)
+            String tablet_id = tablet.TabletId
+            String backend_id = tablet.BackendId
+            def (code, out, err) = be_get_compaction_status(backendId_to_backendIP.get(backend_id), backendId_to_backendHttpPort.get(backend_id), tablet_id)
+            logger.info("Get compaction status: code=" + code + ", out=" + out + ", err=" + err)
+            assertEquals(code, 0)
+            def compactionStatus = parseJson(out.trim())
+            assertEquals("success", compactionStatus.status.toLowerCase())
+            running = compactionStatus.run_status
+        } while (running)
+      }
+    }
+
+    try {
+        GetDebugPoint().enableDebugPointForAllBEs("DorisFSDirectory::FSIndexOutput::init.file_cache")
+
+        insert_and_compaction.call(testTable1);
+        insert_and_compaction.call(testTable2);
+    } finally {
+        GetDebugPoint().disableDebugPointForAllBEs("DorisFSDirectory::FSIndexOutput::init.file_cache")
+    }
+}