Copilot commented on code in PR #367: URL: https://github.com/apache/doris-spark-connector/pull/367#discussion_r3840384941
########## spark-doris-connector/spark-doris-connector-base/src/main/java/org/apache/doris/spark/client/read/DorisReadModeResolver.java: ########## @@ -0,0 +1,76 @@ +// 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.spark.client.read; + +import org.apache.doris.spark.client.DorisFrontendClient; +import org.apache.doris.spark.config.DorisConfig; +import org.apache.doris.spark.config.DorisOptions; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.Locale; + +public class DorisReadModeResolver { + + private static final Logger LOG = LoggerFactory.getLogger(DorisReadModeResolver.class); + private static final String ARROW = "arrow"; + private static final String THRIFT = "thrift"; + + private DorisReadModeResolver() { + } + + public static String resolve(DorisConfig config) throws Exception { + return resolve(config, () -> new DorisFrontendClient(config).tryGetArrowFlightSqlPort()); + } Review Comment: The public `resolve(DorisConfig)` API declares `throws Exception`, which forces checked-exception handling for Java callers even though discovery failures are already handled internally (fallback to Thrift). Consider catching and wrapping here so callers aren't required to handle checked exceptions. ########## spark-doris-connector/spark-doris-connector-base/src/main/scala/org/apache/doris/spark/rdd/DorisRDD.scala: ########## @@ -39,7 +39,7 @@ private[spark] class ScalaDorisRDDIterator[T]( extends AbstractDorisRDDIterator[T](context, partition) { override def initReader(config: DorisConfig): Unit = { - config.getValue(DorisOptions.READ_MODE).toLowerCase match { + DorisReadModeResolver.resolve(config) match { case "thrift" => config.setProperty(DorisOptions.DORIS_VALUE_READER_CLASS, classOf[DorisThriftReader].getName) case "arrow" => config.setProperty(DorisOptions.DORIS_VALUE_READER_CLASS, classOf[DorisFlightSqlReader].getName) case rm: String => throw new IllegalArgumentException("Unknown read mode: " + rm) Review Comment: `DorisReadModeResolver.resolve(config)` may trigger Arrow Flight SQL port discovery (HTTP calls to FE) at task/partition initialization time. In the RDD path this runs per partition (via `partition.getConfig`), which can lead to many redundant FE requests and noisy logs when the port isn't preconfigured. Consider resolving once on the driver (e.g., when building `dorisCfg` / generating partitions) and persisting `doris.read.arrow-flight-sql.port` into the config that gets serialized to partitions, or otherwise caching the discovered port per executor. ########## spark-doris-connector/spark-doris-connector-base/src/main/java/org/apache/doris/spark/client/DorisFrontendClient.java: ########## @@ -155,6 +120,67 @@ private LoadBalanceList<Frontend> initFrontends(DorisConfig config) throws Excep } } + private List<Frontend> fetchFrontends(String[] frontendNodeArray) throws Exception { + List<Frontend> frontendList = null; + Exception ex = null; + for (String frontendNode : frontendNodeArray) { + String[] nodeDetails = frontendNode.split(":"); + try { + LoadBalanceList<Frontend> list = new LoadBalanceList<>( + Collections.singletonList(new Frontend(nodeDetails[0], + nodeDetails.length > 1 ? Integer.parseInt(nodeDetails[1]) : -1))); + frontendList = requestFrontends(list, (frontend, client) -> { + String url = URLs.getFrontEndNodes(frontend.getHost(), frontend.getHttpPort(), + isHttpsEnabled); + HttpGet httpGet = new HttpGet(url); + HttpUtils.setAuth(httpGet, username, password); + JsonNode dataNode; + try { + HttpResponse response = client.execute(httpGet); + dataNode = extractDataFromResponse(response, url); + } catch (IOException e) { + throw new RuntimeException("fetch fe failed", e); + } + ArrayNode columnNames = (ArrayNode) dataNode.get("columnNames"); + ArrayNode rows = (ArrayNode) dataNode.get("rows"); + return parseFrontends(columnNames, rows); + }); Review Comment: `fetchFrontends` continues looping over all configured FE nodes even after successfully fetching a non-empty frontend list. Since this method is now also used for Arrow Flight SQL port discovery, this can cause unnecessary extra HTTP requests. Break once a valid non-empty list has been obtained. This issue also appears on line 162 of the same file. ########## spark-doris-connector/spark-doris-connector-it/src/test/java/org/apache/doris/spark/sql/DorisReaderITCase.scala: ########## @@ -115,6 +115,31 @@ class DorisReaderITCase(readMode: String, flightSqlPort: Int) extends AbstractCo } } + @Test + @throws[Exception] + def testArrowFlightSqlPortAutoDiscovery(): Unit = { + if (!readMode.equals("arrow")) { + return + } + initializeTable(TABLE_READ_TBL, DataModel.UNIQUE) + val session = SparkSession.builder().master("local[*]").getOrCreate() + try { + val dorisSparkDF = session.read + .format("doris") + .option("doris.fenodes", getFenodes) + .option("doris.table.identifier", DATABASE + "." + TABLE_READ_TBL) + .option("doris.user", getDorisUsername) + .option("doris.password", getDorisPassword) + .option("doris.read.mode", "arrow") + .load() + + val result = dorisSparkDF.collect().toList.toString() + assert("List([doris,18], [spark,10])".equals(result)) Review Comment: This assertion relies on `collect().toList.toString()` producing a deterministic row order, which isn't guaranteed (and can become flaky across Spark versions / partitioning). Compare order-independently (e.g., as a Set) or explicitly sort before asserting. ########## spark-doris-connector/spark-doris-connector-base/src/main/scala/org/apache/doris/spark/sql/ScalaDorisRowRDD.scala: ########## @@ -38,7 +39,7 @@ private[spark] class ScalaDorisRowRDDIterator(context: TaskContext, extends AbstractDorisRDDIterator[Row](context, partition) { override def initReader(config: DorisConfig): Unit = { - config.getValue(DorisOptions.READ_MODE).toLowerCase match { + DorisReadModeResolver.resolve(config) match { case "thrift" => config.setProperty(DorisOptions.DORIS_VALUE_READER_CLASS, classOf[DorisRowThriftReader].getName) case "arrow" => config.setProperty(DorisOptions.DORIS_VALUE_READER_CLASS, classOf[DorisRowFlightSqlReader].getName) case rm: String => throw new IllegalArgumentException("Unknown read mode: " + rm) Review Comment: `DorisReadModeResolver.resolve(config)` may perform Arrow Flight SQL port discovery (HTTP calls to FE). In the RDD iterator path this runs per partition (via `partition.getConfig`), which can cause a burst of redundant FE requests when the port isn't set. Prefer resolving once driver-side (before partitions are serialized) and propagating the discovered port in the serialized config, or add caching to avoid repeated discovery per partition. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
