github-actions[bot] commented on code in PR #69623:
URL: https://github.com/apache/doris/pull/69623#discussion_r4375813630
##########
be/src/cloud/cloud_stream_load_executor.cpp:
##########
@@ -33,12 +33,6 @@ bvar::Adder<uint64_t>
stream_load_commit_retry_counter("stream_load_commit_retry
bvar::Window<bvar::Adder<uint64_t>> stream_load_commit_retry_counter_minute(
"stream_load_commit_retry_counter", "1m",
&stream_load_commit_retry_counter, 60);
-enum class TxnOpParamType : int {
Review Comment:
[P1] Keep the enum used by cloud rollback
This removes the only declaration of `TxnOpParamType`, but
`CloudStreamLoadExecutor::rollback_txn` below still declares `TxnOpParamType
topt` and references its `WITH_TXN_ID`, `WITH_LABEL`, and `ILLEGAL` values. The
type is not declared in the header or elsewhere under `be/`, so this
translation unit no longer compiles and blocks the BE build. Keep the enum or
simplify the remaining rollback dispatch before removing it.
##########
regression-test/suites/auth_call/test_stream_load_2pc_owner.groovy:
##########
@@ -0,0 +1,115 @@
+// 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.
+
+import java.util.concurrent.TimeUnit
+
+suite("test_stream_load_2pc_owner", "p0,auth_call") {
+ String owner = "test_stream_load_2pc_owner_user"
+ String other = "test_stream_load_2pc_other_user"
+ String password = "C123_567p"
+ String db = "test_stream_load_2pc_owner_db"
+ String feAddress = context.config.feHttpAddress
+ String dataFile =
"${context.file.parent}/../../data/auth_call/stream_load_data.csv"
+ String computeGroup = ""
+
+ // Exercise the authenticated FE entry and its actual BE redirect in both
deployment modes.
+ def authConfig = sql "ADMIN SHOW FRONTEND CONFIG LIKE
'enable_all_http_auth'"
+ assertEquals("true", authConfig[0][1].toString().toLowerCase())
+ if (isCloudMode()) {
+ computeGroup = (sql "SHOW CLUSTERS")[0][0].toString()
+ }
+
+ def curl = { List args ->
+ def process = (["curl", "--noproxy", "*", "--silent", "--show-error",
+ "--location-trusted", "--max-time", "120"] +
args).collect { it.toString() }.execute()
+ def stdout = new ByteArrayOutputStream()
+ def stderr = new ByteArrayOutputStream()
+ def outThread = process.consumeProcessOutputStream(stdout)
+ def errThread = process.consumeProcessErrorStream(stderr)
+ boolean finished = process.waitFor(150, TimeUnit.SECONDS)
+ if (!finished) {
+ process.destroyForcibly()
+ process.waitFor()
+ }
+ outThread.join()
+ errThread.join()
+ assertTrue(finished, "curl timed out")
+ assertEquals(0, process.exitValue(), stderr.toString("UTF-8"))
+ return parseJson(stdout.toString("UTF-8"))
+ }
+ def headers = {
+ return computeGroup.isEmpty() ? [] : ["-H",
"cloud_cluster:${computeGroup}"]
+ }
+ def prepare = { boolean sqlHttp, String label ->
+ def args = ["-u", "${owner}:${password}", "-H",
"two_phase_commit:true",
+ "-H", "label:${label}", "-H", "column_separator:,"] +
headers()
Review Comment:
[P2] Send the required Expect header in the regression loads
`prepare` uploads the 18-byte `stream_load_data.csv` fixture with `curl -T`
but never sets `Expect: 100-continue`. With curl 8.5.0 (the runner version),
PUT uploads this small do not get an automatic Expect header; libcurl adds it
only for bodies over 1 MiB or unknown size ([curl
implementation](https://github.com/curl/curl/blob/curl-8_5_0/lib/http.c#L2360-L2411),
[threshold](https://github.com/curl/curl/blob/curl-8_5_0/lib/http.h#L167-L174)).
Both FE producer endpoints explicitly reject requests missing this header with
`There is no 100-continue header`, so the first `prepare(false, ...)` fails its
Success assertion before any ownership case executes. Add `-H`,
`Expect:100-continue` to the shared producer arguments.
##########
regression-test/suites/auth_call/test_stream_load_2pc_owner.groovy:
##########
@@ -0,0 +1,115 @@
+// 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.
+
+import java.util.concurrent.TimeUnit
+
+suite("test_stream_load_2pc_owner", "p0,auth_call") {
+ String owner = "test_stream_load_2pc_owner_user"
+ String other = "test_stream_load_2pc_other_user"
+ String password = "C123_567p"
+ String db = "test_stream_load_2pc_owner_db"
+ String feAddress = context.config.feHttpAddress
+ String dataFile =
"${context.file.parent}/../../data/auth_call/stream_load_data.csv"
+ String computeGroup = ""
+
+ // Exercise the authenticated FE entry and its actual BE redirect in both
deployment modes.
+ def authConfig = sql "ADMIN SHOW FRONTEND CONFIG LIKE
'enable_all_http_auth'"
+ assertEquals("true", authConfig[0][1].toString().toLowerCase())
+ if (isCloudMode()) {
+ computeGroup = (sql "SHOW CLUSTERS")[0][0].toString()
+ }
+
+ def curl = { List args ->
+ def process = (["curl", "--noproxy", "*", "--silent", "--show-error",
+ "--location-trusted", "--max-time", "120"] +
args).collect { it.toString() }.execute()
+ def stdout = new ByteArrayOutputStream()
+ def stderr = new ByteArrayOutputStream()
+ def outThread = process.consumeProcessOutputStream(stdout)
+ def errThread = process.consumeProcessErrorStream(stderr)
+ boolean finished = process.waitFor(150, TimeUnit.SECONDS)
+ if (!finished) {
+ process.destroyForcibly()
+ process.waitFor()
+ }
+ outThread.join()
+ errThread.join()
+ assertTrue(finished, "curl timed out")
+ assertEquals(0, process.exitValue(), stderr.toString("UTF-8"))
+ return parseJson(stdout.toString("UTF-8"))
+ }
+ def headers = {
+ return computeGroup.isEmpty() ? [] : ["-H",
"cloud_cluster:${computeGroup}"]
+ }
+ def prepare = { boolean sqlHttp, String label ->
+ def args = ["-u", "${owner}:${password}", "-H",
"two_phase_commit:true",
+ "-H", "label:${label}", "-H", "column_separator:,"] +
headers()
+ if (sqlHttp) {
+ args.addAll(["-H", "sql:insert into
${db}.test_stream_load_2pc_owner "
+ + "select c1, c2 from
http_stream(\"format\"=\"csv\",\"column_separator\"=\",\")",
+ "-T", dataFile, "http://${feAddress}/api/_http_stream";])
+ } else {
+ args.addAll(["-T", dataFile,
+
"http://${feAddress}/api/${db}/test_stream_load_2pc_owner/_stream_load";])
+ }
+ def result = curl(args)
+ assertEquals("Success", result.Status)
+ assertTrue(result.TwoPhaseCommit.toString().equalsIgnoreCase("true"))
+ return result
+ }
+ def finish = { String user, def txn, boolean byLabel, String operation ->
+ def key = byLabel ? "label:${txn.Label}" : "txn_id:${txn.TxnId}"
+ return curl(["-X", "PUT", "-u", "${user}:${password}", "-H", key,
+ "-H", "txn_operation:${operation}"] + headers()
+ + ["http://${feAddress}/api/${db}/_stream_load_2pc";])
+ }
+
+ try {
+ [owner, other].each { user ->
+ try_sql "DROP USER ${user}"
+ sql "CREATE USER '${user}' IDENTIFIED BY '${password}'"
+ if (!computeGroup.isEmpty()) {
+ sql "GRANT USAGE_PRIV ON CLUSTER `${computeGroup}` TO ${user}"
+ }
+ }
+ sql "CREATE DATABASE IF NOT EXISTS ${db}"
+ sql "DROP TABLE IF EXISTS ${db}.test_stream_load_2pc_owner"
+ sql """CREATE TABLE ${db}.test_stream_load_2pc_owner (id BIGINT,
username VARCHAR(20))
+ DUPLICATE KEY(id) DISTRIBUTED BY HASH(id) BUCKETS 1
+ PROPERTIES ("replication_num" = "1")"""
+ [owner, other].each { user ->
+ sql "GRANT LOAD_PRIV ON ${db}.test_stream_load_2pc_owner TO
${user}"
Review Comment:
[P2] Give the SQL HTTP producer a valid non-ADMIN setup
Even after adding the missing Expect header, `prepare(true, ...)` cannot
succeed with this table-only LOAD grant. `HttpStreamAction::process_put` sends
`TStreamLoadPutRequest` without setting `db`/`tbl`, so they arrive as empty
required strings. Before planning the SQL, `httpStreamPutImpl` checks LOAD
using those request fields; the named table grant here cannot match that empty
resource under the built-in authorizer. The SQL half of the new regression
therefore fails at its preparation Success assertion and never tests owner
propagation. Use a valid producer setup, for example grant the creator global
LOAD for the SQL HTTP branch while keeping it non-ADMIN, or resolve/check the
actual SQL target in the producer. This concerns the new regression setup; the
empty-name production check itself predates this PR.
--
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]
