update

JNSimba · Sep 3, 2024 · 25c9ec9 · 25c9ec9
1 parent 4ceb2fd
commit 25c9ec9
Show file tree

Hide file tree

Showing 3 changed files with 3 additions and 6 deletions.
diff --git a/flink-doris-connector/src/main/java/org/apache/doris/flink/cfg/ConfigurationOptions.java b/flink-doris-connector/src/main/java/org/apache/doris/flink/cfg/ConfigurationOptions.java
@@ -43,6 +43,7 @@ public interface ConfigurationOptions {
 
     String DORIS_BATCH_SIZE = "doris.batch.size";
     Integer DORIS_BATCH_SIZE_DEFAULT = 4064;
+    Integer DORIS_BATCH_SIZE_MAX = 65535;
 
     String DORIS_EXEC_MEM_LIMIT = "doris.exec.mem.limit";
     Long DORIS_EXEC_MEM_LIMIT_DEFAULT = 8589934592L;

diff --git a/flink-doris-connector/src/main/java/org/apache/doris/flink/cfg/DorisReadOptions.java b/flink-doris-connector/src/main/java/org/apache/doris/flink/cfg/DorisReadOptions.java
@@ -17,8 +17,6 @@
 
 package org.apache.doris.flink.cfg;
 
-import org.apache.flink.util.Preconditions;
-
 import java.io.Serializable;
 import java.util.Objects;
 
@@ -278,9 +276,6 @@ public Builder setFlightSqlPort(Integer flightSqlPort) {
         }
 
         public DorisReadOptions build() {
-            Preconditions.checkArgument(
-                    requestBatchSize >= 1 && requestBatchSize <= 65535,
-                    "batchSize should be between 1 and 65535");
             return new DorisReadOptions(
                     readFields,
                     filterQuery,

diff --git a/...-doris-connector/src/main/java/org/apache/doris/flink/source/reader/DorisValueReader.java b/...-doris-connector/src/main/java/org/apache/doris/flink/source/reader/DorisValueReader.java
@@ -45,6 +45,7 @@
 import java.util.concurrent.locks.ReentrantLock;
 
 import static org.apache.doris.flink.cfg.ConfigurationOptions.DORIS_BATCH_SIZE_DEFAULT;
+import static org.apache.doris.flink.cfg.ConfigurationOptions.DORIS_BATCH_SIZE_MAX;
 import static org.apache.doris.flink.cfg.ConfigurationOptions.DORIS_DEFAULT_CLUSTER;
 import static org.apache.doris.flink.cfg.ConfigurationOptions.DORIS_DESERIALIZE_ARROW_ASYNC_DEFAULT;
 import static org.apache.doris.flink.cfg.ConfigurationOptions.DORIS_DESERIALIZE_QUEUE_SIZE_DEFAULT;
@@ -130,7 +131,7 @@ private TScanOpenParams openParams() {
         Integer batchSize =
                 readOptions.getRequestBatchSize() == null
                         ? DORIS_BATCH_SIZE_DEFAULT
-                        : readOptions.getRequestBatchSize();
+                        : Math.min(readOptions.getRequestBatchSize(), DORIS_BATCH_SIZE_MAX);
         Integer queryDorisTimeout =
                 readOptions.getRequestQueryTimeoutS() == null
                         ? DORIS_REQUEST_QUERY_TIMEOUT_S_DEFAULT