Copilot commented on code in PR #688:
URL:
https://github.com/apache/doris-flink-connector/pull/688#discussion_r3774586113
##########
flink-doris-connector/flink-doris-connector-base/src/main/java/org/apache/doris/flink/cfg/DorisExecutionOptions.java:
##########
@@ -520,21 +573,34 @@ public Builder setIgnoreCommitError(boolean
ignoreCommitError) {
return this;
}
+ public Builder setS3TvfOptions(S3TvfOptions s3TvfOptions) {
+ this.s3TvfOptions = s3TvfOptions;
+ return this;
+ }
+
/**
* Build the {@link DorisExecutionOptions}.
*
* @return a DorisExecutionOptions with the settings made for this
builder.
*/
public DorisExecutionOptions build() {
+ Preconditions.checkArgument(
+ writeMode != WriteMode.TVF
+ || (labelPrefix != null &&
!labelPrefix.trim().isEmpty()),
+ "sink.label-prefix must be set for TVF write mode");
+
// If format=json is set but read_json_by_line is not set, record
may not be written.
- if (streamLoadProp != null
+ if (writeMode != WriteMode.TVF
+ && streamLoadProp != null
&& streamLoadProp.containsKey(FORMAT_KEY)
&& JSON.equals(streamLoadProp.getProperty(FORMAT_KEY))) {
streamLoadProp.put(READ_JSON_BY_LINE, true);
}
// Enable gz compression by default
- if (streamLoadProp != null &&
!streamLoadProp.containsKey(COMPRESS_TYPE)) {
+ if (writeMode != WriteMode.TVF
+ && streamLoadProp != null
+ && !streamLoadProp.containsKey(COMPRESS_TYPE)) {
Review Comment:
TVF mode still reaches the existing 10 MiB minimum check below, so the new
integration configurations using 64 bytes or 1024 bytes fail in `build()`
before a writer is created. Make the byte validation mode-aware (TVF only needs
a positive limit); otherwise the multiple-object and failover tests cannot run
as written.
##########
flink-doris-connector/flink-doris-connector-flink2/src/main/java/org/apache/doris/flink/table/DorisDynamicTableSink.java:
##########
@@ -83,43 +87,59 @@ public ChangelogMode getChangelogMode(ChangelogMode
changelogMode) {
@Override
public SinkRuntimeProvider getSinkRuntimeProvider(Context context) {
- Properties loadProperties = executionOptions.getStreamLoadProp();
boolean deletable =
executionOptions.getDeletable()
&& RestService.isUniqueKeyType(options, readOptions,
LOG);
- if (!loadProperties.containsKey(COLUMNS_KEY)) {
- String[] fieldNames = tableSchema.getFieldNames();
- Preconditions.checkState(fieldNames != null && fieldNames.length >
0);
- String columns =
- String.join(
- ",",
- Arrays.stream(fieldNames)
- .map(
- item ->
- String.format(
- "`%s`",
item.trim().replace("`", "")))
- .collect(Collectors.toList()));
- if (deletable) {
- columns = String.format("%s,%s", columns, DORIS_DELETE_SIGN);
+ DorisRecordSerializer<RowData> serializer;
+ if (WriteMode.TVF.equals(executionOptions.getWriteMode())) {
+ List<String> columns =
+ TvfColumnUtils.resolveColumns(
+ executionOptions.getStreamLoadProp(),
tableSchema.getFieldNames());
+ serializer =
+ new S3TvfRowDataSerializer(
+ tableSchema.getFieldNames(),
+ tableSchema.getFieldDataTypes(),
+ columns,
+ deletable);
Review Comment:
For TVF mode this flag is still gated by `isUniqueKeyType` above.
Consequently, DELETE rows targeting Duplicate or Aggregate tables are
serialized without `__DORIS_DELETE_SIGN__` and loaded as ordinary rows, which
contradicts the PR's stated delete support for those table models. Either
implement the promised semantics or reject those changelogs instead of silently
inserting them.
##########
flink-doris-connector/flink-doris-connector-base/src/main/java/org/apache/doris/flink/sink/writer/tvf/S3TvfSqlBuilder.java:
##########
@@ -0,0 +1,97 @@
+// 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.flink.sink.writer.tvf;
+
+import org.apache.flink.util.Preconditions;
+
+import org.apache.doris.flink.cfg.S3TvfOptions;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.StringJoiner;
+
+import static
org.apache.doris.flink.sink.writer.LoadConstants.DORIS_DELETE_SIGN;
+import static
org.apache.doris.flink.sink.writer.tvf.TvfSqlUtils.quoteIdentifier;
+import static org.apache.doris.flink.sink.writer.tvf.TvfSqlUtils.quoteLiteral;
+
+/** Builds one INSERT INTO SELECT FROM S3 TVF statement for a committable. */
+class S3TvfSqlBuilder {
+
+ private final S3TvfOptions options;
+
+ public S3TvfSqlBuilder(S3TvfOptions options) {
+ this.options = options;
+ }
+
+ public String buildInsertSql(S3TvfCommittable committable) {
+ Preconditions.checkArgument(!committable.getObjectKeys().isEmpty());
+ List<String> loadColumns = new ArrayList<>(committable.getColumns());
+ if (committable.isDeleteSignEnabled()) {
+ loadColumns.add(DORIS_DELETE_SIGN);
+ }
+ String columnSql = joinIdentifiers(loadColumns);
+ String uri = buildUri(committable.getObjectKeys());
+
+ return "INSERT INTO "
+ + quoteIdentifier(committable.getDatabase())
+ + "."
+ + quoteIdentifier(committable.getTable())
+ + " WITH LABEL "
+ + quoteIdentifier(committable.getLabel())
+ + " ("
+ + columnSql
+ + ") SELECT "
+ + columnSql
+ + " FROM S3("
+ + property("uri", uri)
+ + ","
+ + property("s3.access_key", options.getAccessKey())
+ + ","
+ + property("s3.secret_key", options.getSecretKey())
+ + ","
+ + property("s3.region", options.getRegion())
+ + ","
+ + property("s3.endpoint", options.getEndpoint())
+ + ","
+ + property("format", "json")
+ + ","
+ + property("read_json_by_line", "true")
+ + ","
+ + property("use_path_style",
Boolean.toString(options.isPathStyleAccess()))
+ + ")";
+ }
+
+ private String buildUri(List<String> objectKeys) {
+ if (objectKeys.size() == 1) {
+ return "s3://" + options.getBucket() + "/" + objectKeys.get(0);
+ }
+ return "s3://" + options.getBucket() + "/{" + String.join(",",
objectKeys) + "}";
Review Comment:
These raw object keys are inserted into Doris's brace/glob URI syntax. Valid
S3 keys containing commas, braces, or wildcard characters will therefore alter
the expansion and can load the wrong objects (for example, a comma in a shared
prefix splits each key into extra alternatives). Escape/encode the TVF path
metacharacters, or validate and reject such prefixes before upload.
##########
flink-doris-connector/flink-doris-connector-flink1/src/main/java/org/apache/doris/flink/table/DorisDynamicTableSink.java:
##########
@@ -83,43 +87,59 @@ public ChangelogMode getChangelogMode(ChangelogMode
changelogMode) {
@Override
public SinkRuntimeProvider getSinkRuntimeProvider(Context context) {
- Properties loadProperties = executionOptions.getStreamLoadProp();
boolean deletable =
executionOptions.getDeletable()
&& RestService.isUniqueKeyType(options, readOptions,
LOG);
- if (!loadProperties.containsKey(COLUMNS_KEY)) {
- String[] fieldNames = tableSchema.getFieldNames();
- Preconditions.checkState(fieldNames != null && fieldNames.length >
0);
- String columns =
- String.join(
- ",",
- Arrays.stream(fieldNames)
- .map(
- item ->
- String.format(
- "`%s`",
item.trim().replace("`", "")))
- .collect(Collectors.toList()));
- if (deletable) {
- columns = String.format("%s,%s", columns, DORIS_DELETE_SIGN);
+ DorisRecordSerializer<RowData> serializer;
+ if (WriteMode.TVF.equals(executionOptions.getWriteMode())) {
+ List<String> columns =
+ TvfColumnUtils.resolveColumns(
+ executionOptions.getStreamLoadProp(),
tableSchema.getFieldNames());
+ serializer =
+ new S3TvfRowDataSerializer(
+ tableSchema.getFieldNames(),
+ tableSchema.getFieldDataTypes(),
+ columns,
+ deletable);
Review Comment:
For TVF mode this flag is still gated by `isUniqueKeyType` above.
Consequently, DELETE rows targeting Duplicate or Aggregate tables are
serialized without `__DORIS_DELETE_SIGN__` and loaded as ordinary rows, which
contradicts the PR's stated delete support for those table models. Either
implement the promised semantics or reject those changelogs instead of silently
inserting them.
##########
flink-doris-connector/flink-doris-connector-base/src/main/java/org/apache/doris/flink/cfg/DorisExecutionOptions.java:
##########
@@ -520,21 +573,34 @@ public Builder setIgnoreCommitError(boolean
ignoreCommitError) {
return this;
}
+ public Builder setS3TvfOptions(S3TvfOptions s3TvfOptions) {
+ this.s3TvfOptions = s3TvfOptions;
+ return this;
+ }
+
/**
* Build the {@link DorisExecutionOptions}.
*
* @return a DorisExecutionOptions with the settings made for this
builder.
*/
public DorisExecutionOptions build() {
+ Preconditions.checkArgument(
+ writeMode != WriteMode.TVF
+ || (labelPrefix != null &&
!labelPrefix.trim().isEmpty()),
+ "sink.label-prefix must be set for TVF write mode");
Review Comment:
The TVF-specific validation only checks the label. The public builder still
accepts TVF mode with `s3TvfOptions == null`, producing an unusable options
object that fails later when the writer is initialized. Reject the missing S3
configuration here so programmatic users get a configuration error during
construction.
--
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]
