(cloud-merge) Support to write data into cache in cloud mode (#31934)

be/src/vec/sink/writer/vtablet_writer.cpp

@@ -422,6 +422,7 @@ void VNodeChannel::_open_internal(bool is_incremental) {
     request->set_enable_profile(_state->enable_profile());
     request->set_is_incremental(is_incremental);
     request->set_txn_expiration(_parent->_txn_expiration);
+    request->set_write_file_cache(_parent->_write_file_cache);
 
     auto open_callback = DummyBrpcCallback<PTabletWriterOpenResult>::create_shared();
     auto open_closure = AutoReleaseClosure<
@@ -1129,6 +1130,7 @@ Status VTabletWriter::_init(RuntimeState* state, RuntimeProfile* profile) {
     _txn_id = table_sink.txn_id;
     _num_replicas = table_sink.num_replicas;
     _tuple_desc_id = table_sink.tuple_id;
+    _write_file_cache = table_sink.write_file_cache;
     _schema.reset(new OlapTableSchemaParam());
     RETURN_IF_ERROR(_schema->init(table_sink.schema));
     _location = _pool->add(new OlapTableLocationParam(table_sink.location));

be/src/vec/sink/writer/vtablet_writer.h

@@ -654,6 +654,7 @@ class VTabletWriter final : public AsyncResultWriter {
     // for non-pipeline, if close() did something, close_wait() should wait it.
     bool _close_wait = false;
     bool _inited = false;
+    bool _write_file_cache = false;
 
     // User can change this config at runtime, avoid it being modified during query or loading process.
     bool _transfer_large_data_by_brpc = false;

fe/fe-core/src/main/java/org/apache/doris/planner/OlapTableSink.java

@@ -133,6 +133,9 @@ public void init(TUniqueId loadId, long txnId, long dbId, long loadChannelTimeou
         tSink.setBaseSchemaVersion(dstTable.getBaseSchemaVersion());
         tSink.setLoadChannelTimeoutS(loadChannelTimeoutS);
         tSink.setSendBatchParallelism(sendBatchParallelism);
+        tSink.setWriteFileCache(ConnectContext.get() != null
+                ? !ConnectContext.get().getSessionVariable().isDisableFileCache()
+                : false);
         this.isStrictMode = isStrictMode;
         this.txnId = txnId;
         if (loadToSingleTablet && !(dstTable.getDefaultDistributionInfo() instanceof RandomDistributionInfo)) {

fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java

@@ -374,6 +374,8 @@ public class SessionVariable implements Serializable, Writable {
 
     public static final String ENABLE_FILE_CACHE = "enable_file_cache";
 
+    public static final String DISABLE_FILE_CACHE = "disable_file_cache";
+
     public static final String FILE_CACHE_BASE_PATH = "file_cache_base_path";
 
     public static final String ENABLE_INVERTED_INDEX_QUERY = "enable_inverted_index_query";
@@ -1305,6 +1307,11 @@ public void setEnableLeftZigZag(boolean enableLeftZigZag) {
     @VariableMgr.VarAttr(name = GROUP_BY_AND_HAVING_USE_ALIAS_FIRST)
     public boolean groupByAndHavingUseAliasFirst = false;
 
+    // Whether disable block file cache. Block cache only works when FE's query options sets disableFileCache false
+    // along with BE's config `enable_file_cache` true
+    @VariableMgr.VarAttr(name = DISABLE_FILE_CACHE, needForward = true)
+    public boolean disableFileCache = false;
+
     // Whether enable block file cache. Only take effect when BE config item enable_file_cache is true.
     @VariableMgr.VarAttr(name = ENABLE_FILE_CACHE, needForward = true, description = {
             "是否启用file cache。该变量只有在be.conf中enable_file_cache=true时才有效，"
@@ -2131,6 +2138,14 @@ public void setResourceGroup(String resourceGroup) {
         this.resourceGroup = resourceGroup;
     }
 
+    public boolean isDisableFileCache() {
+        return Config.isCloudMode() ? disableFileCache : false;
+    }
+
+    public void setDisableFileCache(boolean disableFileCache) {
+        this.disableFileCache = disableFileCache;
+    }
+
     public boolean isDisableColocatePlan() {
         return disableColocatePlan;
     }
@@ -2982,6 +2997,7 @@ public TQueryOptions toThrift() {
         tResult.setParallelScanMaxScannersCount(parallelScanMaxScannersCount);
         tResult.setParallelScanMinRowsPerScanner(parallelScanMinRowsPerScanner);
         tResult.setSkipBadTablet(skipBadTablet);
+        tResult.setDisableFileCache(disableFileCache);
 
         return tResult;
     }