Copilot commented on code in PR #698:
URL:
https://github.com/apache/doris-flink-connector/pull/698#discussion_r4023939447
##########
flink-doris-connector/flink-doris-connector-base/src/main/java/org/apache/doris/flink/cfg/DorisExecutionOptions.java:
##########
@@ -601,9 +601,7 @@ public DorisExecutionOptions build() {
}
// Enable gz compression by default
- if (writeMode != WriteMode.TVF
- && streamLoadProp != null
- && !streamLoadProp.containsKey(COMPRESS_TYPE)) {
+ if (streamLoadProp != null &&
!streamLoadProp.containsKey(COMPRESS_TYPE)) {
streamLoadProp.put(COMPRESS_TYPE, COMPRESS_TYPE_GZ);
Review Comment:
For programmatic TVF sinks, an explicit `compress_type=""` is ignored.
`DorisExecutionOptions` preserves that value, but both writer adapters and the
SQL builder read the independent `S3TvfOptions.gzipEnabled` flag, whose builder
defaults to `true`; only the table-factory path synchronizes the two.
Consequently, DataStream users still upload `.gz` objects and declare
`compress_type='gz'`. Make one setting authoritative or synchronize
`S3TvfOptions` from this property when building TVF options.
##########
flink-doris-connector/flink-doris-connector-base/src/main/java/org/apache/doris/flink/cfg/S3TvfOptions.java:
##########
@@ -71,10 +77,30 @@ public String getSecretKey() {
return secretKey;
}
+ public String getRoleArn() {
+ return roleArn;
+ }
+
+ public String getExternalId() {
+ return externalId;
+ }
+
+ public boolean hasRoleArn() {
+ return roleArn != null && !roleArn.isEmpty();
+ }
+
+ public boolean hasStaticCredentials() {
+ return accessKey != null && !accessKey.isEmpty();
+ }
Review Comment:
This predicate treats an access key alone as complete credentials.
Public-builder users can therefore configure a role plus only an access key,
after which writer construction calls `AwsBasicCredentials.create(accessKey,
null)` and fails far from configuration. The table factory validates the pair,
but the public builder does not; validate that access and secret keys are both
present or both absent (and that credentials or a role exists) in
`Builder.build()`.
##########
flink-doris-connector/flink-doris-connector-base/src/main/java/org/apache/doris/flink/sink/writer/tvf/S3TvfWriter.java:
##########
@@ -184,14 +193,15 @@ private void uploadBuffer() throws IOException {
}
long uploadStartedAtNanos = System.nanoTime();
try {
- objectStore.put(objectKey, content);
+ byte[] uploadContent = gzipEnabled ? gzip(content) :
content;
+ objectStore.put(objectKey, uploadContent);
Review Comment:
Gzip is performed after the uncompressed array has been queued, so every
pending upload retains up to 100 MiB of uncompressed data and the active upload
additionally allocates a compressed array. With the default queue size of 2 and
a concurrently refilled 100 MiB buffer, one TVF subtask can transiently retain
roughly 500 MiB. Since gzip is now enabled by default, compress before queuing
or stream into a compressed staging buffer so queued tasks retain only the
upload representation.
--
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]
