branch-4.0: [enhance](load) only set brokerLoadBatchSize when enableMemtableOnSinkNode is true (#63801)

pick #60301

Author: sollhui

fe/fe-core/src/main/java/org/apache/doris/load/loadv2/BrokerLoadJob.java

@@ -18,6 +18,7 @@
 package org.apache.doris.load.loadv2;
 
 import org.apache.doris.analysis.BrokerDesc;
+import org.apache.doris.analysis.IndexDef;
 import org.apache.doris.analysis.StorageBackend;
 import org.apache.doris.analysis.UserIdentity;
 import org.apache.doris.catalog.Database;
@@ -307,6 +308,13 @@ private void createLoadingTask(Database db, BrokerPendingTaskAttachment attachme
                 }
                 boolean isEnableMemtableOnSinkNode =
                         table.getTableProperty().getUseSchemaLightChange() && this.enableMemTableOnSinkNode;
+                boolean hasInvertedIndexV1 = table.hasIndexOfType(IndexDef.IndexType.INVERTED)
+                        && table.getInvertedIndexFileStorageFormat()
+                        == org.apache.doris.thrift.TInvertedIndexFileStorageFormat.V1;
+                if (isPartialUpdate() || hasInvertedIndexV1 || Config.isCloudMode()) {
+                    isEnableMemtableOnSinkNode = false;
+                }
+
                 // Generate loading task and init the plan of task
                 LoadLoadingTask task = createTask(db, table, brokerFileGroups,
                         isEnableMemtableOnSinkNode, batchSize, aggKey, attachment);

fe/fe-core/src/main/java/org/apache/doris/load/loadv2/LoadLoadingTask.java

@@ -171,7 +171,9 @@ protected void executeOnce() throws Exception {
         curCoordinator.setExecMemoryLimit(execMemLimit);
 
         curCoordinator.setMemTableOnSinkNode(enableMemTableOnSinkNode);
-        curCoordinator.setBatchSize(batchSize);
+        if (enableMemTableOnSinkNode) {
+            curCoordinator.setBatchSize(batchSize);
+        }
 
         long leftTimeMs = getLeftTimeMs();
         if (leftTimeMs <= 0) {

fe/fe-core/src/test/java/org/apache/doris/load/loadv2/LoadLoadingTaskTest.java

@@ -0,0 +1,57 @@
+// 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.
+
+package org.apache.doris.load.loadv2;
+
+import org.apache.doris.thrift.TQueryOptions;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+public class LoadLoadingTaskTest {
+
+    /**
+     * Test the batch size setting logic in LoadLoadingTask.executeOnce().
+     * When enableMemTableOnSinkNode = true, setBatchSize() should be called.
+     * When enableMemTableOnSinkNode = false, setBatchSize() should NOT be called.
+     */
+    @Test
+    public void testBatchSizeSettingLogic() {
+        int brokerLoadBatchSize = 16352;
+
+        // Case 1: enableMemTableOnSinkNode = true, setBatchSize should be called
+        TQueryOptions queryOptionsWithMemTable = new TQueryOptions();
+        boolean enableMemTableOnSinkNode1 = true;
+        queryOptionsWithMemTable.setEnableMemtableOnSinkNode(enableMemTableOnSinkNode1);
+        if (enableMemTableOnSinkNode1) {
+            queryOptionsWithMemTable.setBatchSize(brokerLoadBatchSize);
+        }
+        Assert.assertTrue(queryOptionsWithMemTable.isEnableMemtableOnSinkNode());
+        Assert.assertEquals(brokerLoadBatchSize, queryOptionsWithMemTable.getBatchSize());
+
+        // Case 2: enableMemTableOnSinkNode = false, setBatchSize should NOT be called
+        TQueryOptions queryOptionsWithoutMemTable = new TQueryOptions();
+        boolean enableMemTableOnSinkNode2 = false;
+        queryOptionsWithoutMemTable.setEnableMemtableOnSinkNode(enableMemTableOnSinkNode2);
+        if (enableMemTableOnSinkNode2) {
+            queryOptionsWithoutMemTable.setBatchSize(brokerLoadBatchSize);
+        }
+        Assert.assertFalse(queryOptionsWithoutMemTable.isEnableMemtableOnSinkNode());
+        // batch_size should remain 0 (unset), BE will use DEFAULT_BATCH_SIZE (4062)
+        Assert.assertEquals(0, queryOptionsWithoutMemTable.getBatchSize());
+    }
+}