Copilot commented on code in PR #371: URL: https://github.com/apache/doris-spark-connector/pull/371#discussion_r3802424839
########## spark-doris-connector/spark-doris-connector-base/src/main/java/org/apache/doris/spark/client/write/tvf/S3TvfWriter.java: ########## @@ -0,0 +1,178 @@ +// 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.write.tvf; + +import org.apache.doris.spark.config.S3TvfOptions; +import org.apache.doris.spark.util.RowConvertors; +import org.apache.spark.sql.catalyst.InternalRow; +import org.apache.spark.sql.catalyst.expressions.GenericInternalRow; +import org.apache.spark.sql.types.StructField; +import org.apache.spark.sql.types.StructType; + +import java.io.ByteArrayOutputStream; +import java.io.IOException; +import java.util.ArrayList; +import java.util.List; +import java.util.UUID; + +/** Writes one logical Spark partition as deterministic JSON Lines objects. */ +public final class S3TvfWriter implements AutoCloseable { + private static final byte NEW_LINE = '\n'; + + private final S3TvfOptions options; + private final String database; + private final String table; + private final String normalizedTable; + private final String labelPrefix; + private final List<String> columns; + private final StructType outputSchema; + private final int[] selectedIndexes; + private final boolean projectionRequired; + private final String batchUuid = UUID.randomUUID().toString(); + private final int partitionId; + private final int batchSize; + private final S3ObjectStore objectStore; + private final ByteArrayOutputStream buffer = new ByteArrayOutputStream(); + private final List<String> objectKeys = new ArrayList<>(); + + private int recordCount; + private int fileNumber; + + public S3TvfWriter( + S3TvfOptions options, + String database, + String table, + String labelPrefix, + StructType inputSchema, + List<String> columns, + int partitionId, + int batchSize, + S3ObjectStore objectStore) { + if (batchSize <= 0) { + throw new IllegalArgumentException("S3 TVF batch size must be greater than zero"); + } + this.options = options; + this.database = database; + this.table = table; + this.normalizedTable = normalizeTable(table); + this.labelPrefix = labelPrefix; + this.columns = new ArrayList<>(columns); + this.selectedIndexes = resolveIndexes(inputSchema, columns); + this.outputSchema = selectSchema(inputSchema, selectedIndexes); + this.projectionRequired = requiresProjection(selectedIndexes, inputSchema.fields().length); + this.partitionId = partitionId; + this.batchSize = batchSize; + this.objectStore = objectStore; + } + + public void write(InternalRow row) throws IOException { + byte[] json = RowConvertors.convertToJsonBytes(project(row), outputSchema); + buffer.write(json); + buffer.write(NEW_LINE); + recordCount++; + if (recordCount >= batchSize) { + uploadBuffer(); + } + } + + public S3TvfCommittable prepareCommit() throws IOException { + uploadBuffer(); + return new S3TvfCommittable(database, table, label(), objectKeys, columns); + } + + private void uploadBuffer() throws IOException { + if (recordCount == 0) { + return; + } + byte[] content = buffer.toByteArray(); + int currentFileNumber = fileNumber++; Review Comment: The default 500,000-row batch is accumulated entirely in heap and `toByteArray()` creates a second full-size copy during upload. With moderately wide rows this can consume hundreds of MB to multiple GB per concurrent Spark task and OOM executors. Stream each object from a bounded buffer/temp file (and preferably enforce a byte-size rollover) rather than buffering by row count alone. ########## spark-doris-connector/spark-doris-connector-base/src/main/java/org/apache/doris/spark/client/DorisFrontendClient.java: ########## @@ -218,6 +223,36 @@ public <T> T queryFrontends(Function<Connection, T> function) throws Exception { throw ex; } + /** + * Executes a JDBC action against one selected frontend without failover. This is intended for + * non-idempotent statements whose outcome may be ambiguous after a connection failure. + */ + public void executeFrontendOnce(JdbcAction action) throws Exception { + if (jdbcTlsAdapter == null) { + jdbcTlsAdapter = DorisJdbcTlsAdapter.create(tlsOptions); + } + Iterator<Frontend> iterator = frontends.iterator(); + if (!iterator.hasNext()) { + throw new DorisException("No frontend is available for JDBC query."); + } + Frontend frontEnd = iterator.next(); Review Comment: A new `DorisFrontendClient` is created for every task commit, so its load-balancer offset always starts at zero; taking only the first iterator element sends every task to the first configured FE. That FE becomes a connection bottleneck, and if it is unavailable all retries select the same failed node even when other FEs are healthy. Select one FE across the configured set before execution (while still avoiding post-execution failover for ambiguous outcomes). ########## .github/workflows/run-itcase.yml: ########## @@ -40,27 +43,27 @@ jobs: - name: Run ITCases for spark 2 run: | - cd spark-doris-connector && mvn clean test -Pspark-2-it,spark-2.4_2.11 -pl spark-doris-connector-it -am -DfailIfNoTests=false -Dtest="*ITCase" -Dimage="apache/doris:doris-all-in-one-2.1.0" + cd spark-doris-connector && mvn clean test -Pspark-2-it,spark-2.4_2.11 -pl spark-doris-connector-it -am -DfailIfNoTests=false -Dtest="*ITCase" -Dimage="jnsimba/doris-all-in-one:4.1.3" Review Comment: These jobs explicitly override the test default with a mutable image from a personal Docker Hub account; `DorisContainer` runs that image privileged. This allows an unverified publisher update to execute privileged code in CI. Use an ASF-controlled image pinned by digest for every matrix command. ########## spark-doris-connector/spark-doris-connector-base/src/main/java/org/apache/doris/spark/client/write/tvf/S3TvfSqlBuilder.java: ########## @@ -0,0 +1,95 @@ +// 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.write.tvf; + +import org.apache.doris.spark.config.S3TvfOptions; + +import java.util.List; +import java.util.StringJoiner; + +/** Builds one S3 TVF INSERT for one Spark partition. */ +public final class S3TvfSqlBuilder { + private final S3TvfOptions options; + + public S3TvfSqlBuilder(S3TvfOptions options) { + this.options = options; + } + + public String buildInsertSql(S3TvfCommittable committable) { + List<String> objectKeys = committable.getObjectKeys(); + List<String> columns = committable.getColumns(); + if (objectKeys.isEmpty()) { + throw new IllegalArgumentException("S3 TVF object keys must not be empty"); + } + if (columns.isEmpty()) { + throw new IllegalArgumentException("S3 TVF columns must not be empty"); + } + String columnSql = joinIdentifiers(columns); + String uri = buildUri(objectKeys); + return "INSERT INTO " + + TvfSqlUtils.quoteIdentifier(committable.getDatabase()) + + "." + + TvfSqlUtils.quoteIdentifier(committable.getTable()) + + " WITH LABEL " + + TvfSqlUtils.quoteIdentifier(committable.getLabel()) + + " (" + + columnSql + + ") SELECT " + + columnSql + + " FROM S3(" + + property("uri", uri) + + "," + + property("format", "json") + + "," + + property("read_json_by_line", "true") + + "," + + property("s3.endpoint", options.getEndpoint()) + + "," + + property("s3.region", options.getRegion()) + + "," + + property("s3.access_key", options.getAccessKey()) + + "," + + property("s3.secret_key", options.getSecretKey()) Review Comment: The generated SQL embeds `s3.secret_key`, while connector MySQL TLS is optional and disabled by default. On an FE connection without TLS, the object-store secret is transmitted as SQL over the network. TVF mode should require verified MySQL TLS (and reject configurations that exclude the MySQL protocol), or use a Doris-side credential mechanism that does not put the secret on this connection. ########## .github/workflows/run-e2ecase.yml: ########## @@ -40,27 +43,27 @@ jobs: - name: Run E2ECases for spark 2 run: | - cd spark-doris-connector && mvn clean test -Pspark-2-it,spark-2.4_2.11 -pl spark-doris-connector-it -am -DfailIfNoTests=false -Dtest="*E2ECase" -Dimage="apache/doris:doris-all-in-one-2.1.0" + cd spark-doris-connector && mvn clean test -Pspark-2-it,spark-2.4_2.11 -pl spark-doris-connector-it -am -DfailIfNoTests=false -Dtest="*E2ECase" -Dimage="jnsimba/doris-all-in-one:4.1.3" Review Comment: These jobs explicitly override the test default with a mutable image from a personal Docker Hub account; `DorisContainer` runs that image privileged. This allows an unverified publisher update to execute privileged code in CI. Use an ASF-controlled image pinned by digest for every matrix command. ########## spark-doris-connector/spark-doris-connector-base/src/main/java/org/apache/doris/spark/client/write/tvf/S3TvfWriter.java: ########## @@ -0,0 +1,178 @@ +// 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.write.tvf; + +import org.apache.doris.spark.config.S3TvfOptions; +import org.apache.doris.spark.util.RowConvertors; +import org.apache.spark.sql.catalyst.InternalRow; +import org.apache.spark.sql.catalyst.expressions.GenericInternalRow; +import org.apache.spark.sql.types.StructField; +import org.apache.spark.sql.types.StructType; + +import java.io.ByteArrayOutputStream; +import java.io.IOException; +import java.util.ArrayList; +import java.util.List; +import java.util.UUID; + +/** Writes one logical Spark partition as deterministic JSON Lines objects. */ +public final class S3TvfWriter implements AutoCloseable { + private static final byte NEW_LINE = '\n'; + + private final S3TvfOptions options; + private final String database; + private final String table; + private final String normalizedTable; + private final String labelPrefix; + private final List<String> columns; + private final StructType outputSchema; + private final int[] selectedIndexes; + private final boolean projectionRequired; + private final String batchUuid = UUID.randomUUID().toString(); + private final int partitionId; + private final int batchSize; + private final S3ObjectStore objectStore; + private final ByteArrayOutputStream buffer = new ByteArrayOutputStream(); + private final List<String> objectKeys = new ArrayList<>(); + + private int recordCount; + private int fileNumber; + + public S3TvfWriter( + S3TvfOptions options, + String database, + String table, + String labelPrefix, + StructType inputSchema, + List<String> columns, + int partitionId, + int batchSize, + S3ObjectStore objectStore) { + if (batchSize <= 0) { + throw new IllegalArgumentException("S3 TVF batch size must be greater than zero"); + } + this.options = options; + this.database = database; + this.table = table; + this.normalizedTable = normalizeTable(table); + this.labelPrefix = labelPrefix; Review Comment: `labelPrefix` is copied into every object key, but multi-file loads later place those keys inside a `{key1,key2}` glob. A configured prefix containing `*`, `?`, `[`, `{`, `}`, or `,` will therefore be interpreted as glob syntax and the uploaded objects will not be loaded. Validate this filename component just as `S3TvfOptions` validates the staging prefix. ########## spark-doris-connector/spark-doris-connector-base/src/main/java/org/apache/doris/spark/client/write/tvf/S3TvfSqlBuilder.java: ########## @@ -0,0 +1,95 @@ +// 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.write.tvf; + +import org.apache.doris.spark.config.S3TvfOptions; + +import java.util.List; +import java.util.StringJoiner; + +/** Builds one S3 TVF INSERT for one Spark partition. */ +public final class S3TvfSqlBuilder { + private final S3TvfOptions options; + + public S3TvfSqlBuilder(S3TvfOptions options) { + this.options = options; + } + + public String buildInsertSql(S3TvfCommittable committable) { + List<String> objectKeys = committable.getObjectKeys(); + List<String> columns = committable.getColumns(); Review Comment: The PR adds extensive branching, glob/literal escaping, and option validation in the TVF core, but there are no corresponding unit-test files under the base module; the new coverage only exercises safe values through container integration tests. Add focused unit tests for single/multi-file SQL generation, identifier/literal escaping, invalid prefixes/options, column projection, and commit/session-variable behavior so these edge cases run in the normal unit suite. ########## spark-doris-connector/spark-doris-connector-base/src/main/java/org/apache/doris/spark/client/write/tvf/S3TvfCommitter.java: ########## @@ -0,0 +1,94 @@ +// 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.write.tvf; + +import org.apache.doris.spark.config.DorisConfig; +import org.apache.doris.spark.config.S3TvfOptions; + +import java.io.IOException; +import java.sql.SQLException; +import java.util.LinkedHashMap; +import java.util.Map; +import java.util.TreeMap; + +/** Commits one Spark partition with one Doris INSERT. */ +public final class S3TvfCommitter implements AutoCloseable { + private static final String COLUMNS = "columns"; + private static final String PARTIAL_COLUMNS = "partial_columns"; + private static final String FORMAT = "format"; + private static final String READ_JSON_BY_LINE = "read_json_by_line"; + private static final String ENABLE_UNIQUE_KEY_PARTIAL_UPDATE = + "enable_unique_key_partial_update"; + + private final Map<String, String> sessionVariables; + private final S3TvfLoadClient loadClient; + private final S3TvfSqlBuilder sqlBuilder; + + public S3TvfCommitter(DorisConfig config) throws Exception { + this( + S3TvfOptions.fromConfig(config), + config.getSinkProperties(), + new JdbcS3TvfLoadClient(config)); + } + + S3TvfCommitter( + S3TvfOptions options, + Map<String, String> loadProperties, + S3TvfLoadClient loadClient) { + this.sessionVariables = toSessionVariables(loadProperties); + this.loadClient = loadClient; + this.sqlBuilder = new S3TvfSqlBuilder(options); + } + + public void commit(S3TvfCommittable committable) throws IOException { + if (committable.isEmpty()) { + return; + } + try { + loadClient.executeInsert( + sqlBuilder.buildInsertSql(committable), + sessionVariables); + } catch (SQLException e) { + throw new IOException( + "Doris INSERT failed for S3 TVF label " + committable.getLabel(), e); + } + } + + private static Map<String, String> toSessionVariables(Map<String, String> loadProperties) { + Map<String, String> values = new TreeMap<>(); + for (Map.Entry<String, String> entry : loadProperties.entrySet()) { + String name = entry.getKey(); + if (!COLUMNS.equals(name) + && !PARTIAL_COLUMNS.equals(name) + && !FORMAT.equals(name) + && !READ_JSON_BY_LINE.equals(name)) { + values.put(name, entry.getValue()); Review Comment: Every sink property except four special cases is issued as a Doris session variable. This namespace currently contains Stream Load properties such as `max_filter_ratio`, `strict_mode`, and `timeout`, whose names are not equivalent Doris session variables, so otherwise valid connector options make TVF commits fail at `SET SESSION`. Reject unsupported TVF properties or map an explicit allowlist to the corresponding INSERT session variables. ########## spark-doris-connector/spark-doris-connector-it/src/test/java/org/apache/doris/spark/container/instance/DorisContainer.java: ########## @@ -49,7 +49,7 @@ public class DorisContainer implements ContainerService { private static final Logger LOG = LoggerFactory.getLogger(DorisContainer.class); - private static final String DEFAULT_DOCKER_IMAGE = "apache/doris:doris-all-in-one-2.1.0"; + private static final String DEFAULT_DOCKER_IMAGE = "jnsimba/doris-all-in-one:4.1.3"; Review Comment: This changes a privileged Testcontainers workload (`withPrivilegedMode(true)` below) from an ASF-owned image to a mutable image published under a personal Docker Hub account. Running unverified third-party code with privileged container access creates a CI/host compromise risk. Use an ASF-controlled Doris image and pin it by digest. -- 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]
