You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@doris.apache.org by "JNSimba (via GitHub)" <gi...@apache.org> on 2026/01/21 09:52:48 UTC
[PR] [Feature](tvf) Support cdc stream tvf for mysql and pg [doris]
JNSimba opened a new pull request, #60116:
URL: https://github.com/apache/doris/pull/60116
### What problem does this PR solve?
#58898 #59461 These two features support consuming data from MySQL and PostgreSQL.
Based on this, this PR introduces cdc_stream tvf, which allows consuming data from MySQL and PostgreSQL in TVF format.
Example
```
select * from cdc_stream(
"type" = "mysql",
"jdbc_url" = "jdbc:mysql://127.0.0.1:3306",
"driver_url" = "mysql-connector-j-8.0.31.jar",
"driver_class" = "com.mysql.cj.jdbc.Driver",
"user" = "root",
"password" = "123456",
"database" = "test_cdc_db",
"table" = "user_info",
"offset" = '{"file":"binlog.000003","pos":"496"}'
)
```
### Release note
None
### Check List (For Author)
- Test <!-- At least one of them must be included. -->
- [ ] Regression test
- [ ] Unit Test
- [ ] Manual test (add detailed scripts or steps below)
- [ ] No need to test or manual test. Explain why:
- [ ] This is a refactor/code format and no logic has been changed.
- [ ] Previous test can cover this change.
- [ ] No code files have been changed.
- [ ] Other reason <!-- Add your reason? -->
- Behavior changed:
- [ ] No.
- [ ] Yes. <!-- Explain the behavior change -->
- Does this need documentation?
- [ ] No.
- [ ] Yes. <!-- Add document PR link here. eg: https://github.com/apache/doris-website/pull/1214 -->
### Check List (For Reviewer who merge this PR)
- [ ] Confirm the release note
- [ ] Confirm test cases
- [ ] Confirm document
- [ ] Add branch pick label <!-- Add branch pick label that this PR should merge into -->
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [Feature](tvf) Support cdc stream tvf for mysql and pg [doris]
Posted by "doris-robot (via GitHub)" <gi...@apache.org>.
doris-robot commented on PR #60116:
URL: https://github.com/apache/doris/pull/60116#issuecomment-3777593969
<details>
<summary>TPC-DS: <b>Total hot run time: 174132 ms</b></summary>
```
machine: 'aliyun_ecs.c7a.8xlarge_32C64G'
scripts: https://github.com/apache/doris/tree/master/tools/tpcds-tools
TPC-DS sf100 test result on commit d84fcddbe4aa9142e912da69ab6bafe20075ca2e, data reload: false
query5 4430 629 510 510
query6 335 228 220 220
query7 4204 471 258 258
query8 347 242 237 237
query9 8640 2905 2870 2870
query10 523 382 353 353
query11 15309 15088 14903 14903
query12 176 119 121 119
query13 1266 471 392 392
query14 6338 3048 2824 2824
query14_1 2704 2657 2697 2657
query15 205 191 170 170
query16 1004 503 455 455
query17 1106 667 581 581
query18 2509 437 337 337
query19 224 225 197 197
query20 130 117 118 117
query21 232 137 119 119
query22 3944 4032 3935 3935
query23 16131 15574 15325 15325
query23_1 15411 15553 15299 15299
query24 7160 1574 1182 1182
query24_1 1181 1197 1227 1197
query25 545 448 411 411
query26 1241 267 155 155
query27 2750 464 288 288
query28 4502 2170 2160 2160
query29 799 549 451 451
query30 313 231 206 206
query31 804 637 577 577
query32 86 77 75 75
query33 538 366 329 329
query34 897 871 562 562
query35 747 759 685 685
query36 869 877 861 861
query37 137 99 89 89
query38 2791 2765 2672 2672
query39 763 737 739 737
query39_1 715 736 710 710
query40 224 140 128 128
query41 79 69 68 68
query42 107 106 106 106
query43 458 450 423 423
query44 1372 751 750 750
query45 188 204 188 188
query46 860 1011 582 582
query47 1458 1444 1388 1388
query48 311 339 241 241
query49 616 436 341 341
query50 625 275 215 215
query51 3786 3758 3711 3711
query52 105 108 98 98
query53 294 325 273 273
query54 295 279 255 255
query55 84 80 79 79
query56 318 329 300 300
query57 1018 1011 950 950
query58 269 263 268 263
query59 2015 2095 1964 1964
query60 346 337 317 317
query61 148 143 141 141
query62 387 362 286 286
query63 304 270 263 263
query64 4923 1256 955 955
query65 3774 3722 3738 3722
query66 1444 421 322 322
query67 15653 15616 15430 15430
query68 2467 1106 754 754
query69 441 372 372 372
query70 988 971 951 951
query71 332 329 299 299
query72 5313 3193 3290 3193
query73 601 740 319 319
query74 8734 8778 8606 8606
query75 2760 2835 2524 2524
query76 2289 1068 672 672
query77 375 392 300 300
query78 9722 9937 9162 9162
query79 1101 934 588 588
query80 1293 583 494 494
query81 551 264 233 233
query82 991 146 110 110
query83 337 265 250 250
query84 258 111 95 95
query85 922 489 425 425
query86 411 285 320 285
query87 2912 2926 2786 2786
query88 3521 2586 2570 2570
query89 405 356 328 328
query90 1960 180 168 168
query91 169 154 133 133
query92 72 71 71 71
query93 1039 940 545 545
query94 650 331 300 300
query95 586 340 320 320
query96 653 520 234 234
query97 2366 2431 2349 2349
query98 211 204 203 203
query99 587 569 471 471
Total cold run time: 247625 ms
Total hot run time: 174132 ms
```
</details>
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [Feature](tvf) Support cdc stream tvf for mysql and pg [doris]
Posted by "doris-robot (via GitHub)" <gi...@apache.org>.
doris-robot commented on PR #60116:
URL: https://github.com/apache/doris/pull/60116#issuecomment-4125030737
<details>
<summary>TPC-H: <b>Total hot run time: 26241 ms</b></summary>
```
machine: 'aliyun_ecs.c7a.8xlarge_32C64G'
scripts: https://github.com/apache/doris/tree/master/tools/tpch-tools
Tpch sf100 test result on commit 9ea6fe06a5cddc20a17b3f4565082396852f0a56, data reload: false
------ Round 1 ----------------------------------
orders Doris NULL NULL 0 0 0 NULL 0 NULL NULL 2023-12-26 18:27:23 2023-12-26 18:42:55 NULL utf-8 NULL NULL
============================================
q1 17633 4446 4292 4292
q2 q3 10643 789 515 515
q4 4679 360 246 246
q5 7567 1233 1032 1032
q6 174 176 146 146
q7 765 840 672 672
q8 9302 1480 1304 1304
q9 4833 4746 4669 4669
q10 6241 1900 1642 1642
q11 461 261 232 232
q12 694 585 463 463
q13 18021 2691 1912 1912
q14 222 230 214 214
q15 q16 735 737 669 669
q17 726 846 418 418
q18 6042 5435 5170 5170
q19 1150 993 601 601
q20 540 507 385 385
q21 4469 1844 1419 1419
q22 346 296 240 240
Total cold run time: 95243 ms
Total hot run time: 26241 ms
----- Round 2, with runtime_filter_mode=off -----
orders Doris NULL NULL 150000000 42 6422171781 NULL 22778155 NULL NULL 2023-12-26 18:27:23 2023-12-26 18:42:55 NULL utf-8 NULL NULL
============================================
q1 4990 4707 4609 4609
q2 q3 3828 4320 3830 3830
q4 875 1202 797 797
q5 4076 4351 4328 4328
q6 183 175 143 143
q7 1761 1657 1515 1515
q8 2478 2724 2569 2569
q9 7592 7306 7388 7306
q10 3739 4023 3618 3618
q11 504 425 411 411
q12 497 608 458 458
q13 2422 2833 2018 2018
q14 282 305 299 299
q15 q16 749 795 759 759
q17 1166 1357 1388 1357
q18 7246 6791 6631 6631
q19 895 925 936 925
q20 2054 2209 2056 2056
q21 3928 3743 3452 3452
q22 432 407 381 381
Total cold run time: 49697 ms
Total hot run time: 47462 ms
```
</details>
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [Feature](tvf) Support cdc stream tvf for mysql and pg [doris]
Posted by "hello-stephen (via GitHub)" <gi...@apache.org>.
hello-stephen commented on PR #60116:
URL: https://github.com/apache/doris/pull/60116#issuecomment-3779131241
# FE Regression Coverage Report
Increment line coverage ` 3.08% (2/65)` :tada:
[Increment coverage report](http://coverage.selectdb-in.cc/coverage/60116_d84fcddbe4aa9142e912da69ab6bafe20075ca2e_merge_fe/increment_report/index.html)
[Complete coverage report](http://coverage.selectdb-in.cc/coverage/60116_d84fcddbe4aa9142e912da69ab6bafe20075ca2e_merge_fe/report/index.html)
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [Feature](tvf) Support cdc stream tvf for mysql and pg [doris]
Posted by "JNSimba (via GitHub)" <gi...@apache.org>.
JNSimba commented on PR #60116:
URL: https://github.com/apache/doris/pull/60116#issuecomment-4116404305
run buildall
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [Feature](tvf) Support cdc stream tvf for mysql and pg [doris]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #60116:
URL: https://github.com/apache/doris/pull/60116#discussion_r2980526691
##########
fe/fe-core/src/main/java/org/apache/doris/tablefunction/CdcStreamTableValuedFunction.java:
##########
@@ -0,0 +1,140 @@
+// 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.tablefunction;
+
+import org.apache.doris.analysis.BrokerDesc;
+import org.apache.doris.analysis.StorageBackend.StorageType;
+import org.apache.doris.catalog.Column;
+import org.apache.doris.common.AnalysisException;
+import org.apache.doris.datasource.jdbc.client.JdbcClient;
+import org.apache.doris.job.cdc.DataSourceConfigKeys;
+import org.apache.doris.job.cdc.request.FetchRecordRequest;
+import org.apache.doris.job.common.DataSourceType;
+import org.apache.doris.job.util.StreamingJobUtils;
+import org.apache.doris.thrift.TBrokerFileStatus;
+import org.apache.doris.thrift.TFileType;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+
+public class CdcStreamTableValuedFunction extends ExternalFileTableValuedFunction {
+ private static final ObjectMapper objectMapper = new ObjectMapper();
+ private static final String URI = "http://127.0.0.1:CDC_CLIENT_PORT/api/fetchRecordStream";
+
+ public CdcStreamTableValuedFunction(Map<String, String> properties) throws AnalysisException {
+ validate(properties);
+ processProps(properties);
+ }
+
+ private void processProps(Map<String, String> properties) throws AnalysisException {
+ Map<String, String> copyProps = new HashMap<>(properties);
+ copyProps.put("format", "json");
+ super.parseCommonProperties(copyProps);
+ this.processedParams.put("enable_cdc_client", "true");
+ this.processedParams.put("uri", URI);
+ this.processedParams.put("http.enable.range.request", "false");
+ this.processedParams.put("http.enable.chunk.response", "true");
+ this.processedParams.put("http.method", "POST");
+
+ String payload = generateParams(properties);
+ this.processedParams.put("http.payload", payload);
+ this.backendConnectProperties.putAll(processedParams);
+ generateFileStatus();
+ }
+
+ private String generateParams(Map<String, String> properties) throws AnalysisException {
+ FetchRecordRequest recordRequest = new FetchRecordRequest();
+ recordRequest.setJobId(UUID.randomUUID().toString().replace("-", ""));
+ recordRequest.setDataSource(properties.get(DataSourceConfigKeys.TYPE));
+ recordRequest.setConfig(properties);
+ try {
+ return objectMapper.writeValueAsString(recordRequest);
+ } catch (IOException e) {
+ LOG.info("Failed to serialize fetch record request," + e.getMessage());
+ throw new AnalysisException(e.getMessage());
Review Comment:
`generateParams` logs at INFO and drops the exception cause, then throws a new `AnalysisException` with only `e.getMessage()`. This makes diagnosing serialization failures harder (no stack trace / context). Consider logging with the throwable and rethrowing an `AnalysisException` that includes a clear message plus the original cause.
```suggestion
LOG.warn("Failed to serialize fetch record request", e);
throw new AnalysisException("Failed to serialize fetch record request: " + e.getMessage(), e);
```
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/table/CdcStream.java:
##########
@@ -0,0 +1,59 @@
+// 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.nereids.trees.expressions.functions.table;
+
+import org.apache.doris.catalog.FunctionSignature;
+import org.apache.doris.nereids.exceptions.AnalysisException;
+import org.apache.doris.nereids.trees.expressions.Properties;
+import org.apache.doris.nereids.trees.expressions.visitor.ExpressionVisitor;
+import org.apache.doris.nereids.types.coercion.AnyDataType;
+import org.apache.doris.tablefunction.CdcStreamTableValuedFunction;
+import org.apache.doris.tablefunction.TableValuedFunctionIf;
+
+import java.util.Map;
+
+/**
+ * CdcStream TVF.
+ */
+public class CdcStream extends TableValuedFunction {
+
+ public CdcStream(Properties tvfProperties) {
+ super("cdc_stream", tvfProperties);
+ }
+
+ @Override
+ public FunctionSignature customSignature() {
+ return FunctionSignature.of(AnyDataType.INSTANCE_WITHOUT_INDEX, getArgumentsTypes());
+ }
+
+ @Override
+ protected TableValuedFunctionIf toCatalogFunction() {
+ try {
+ Map<String, String> arguments = getTVFProperties().getMap();
+ return new CdcStreamTableValuedFunction(arguments);
+ } catch (Throwable t) {
+ throw new AnalysisException("Can not build CdcStreamTableValuedFunction by "
Review Comment:
Spelling/grammar: prefer “Cannot” over “Can not” in user-facing exception text.
```suggestion
throw new AnalysisException("Cannot build CdcStreamTableValuedFunction by "
```
##########
be/src/io/fs/http_file_reader.cpp:
##########
@@ -224,9 +276,29 @@ Status HttpFileReader::read_at_impl(size_t offset, Slice result, size_t* bytes_r
VLOG(2) << "Issuing HTTP GET request: offset=" << offset << " req_len=" << req_len
<< " with_range=" << _range_supported;
- // Prepare and initialize the HTTP client for GET request
+ // Prepare and initialize the HTTP client for request
RETURN_IF_ERROR(prepare_client(/*set_fail_on_error=*/false));
- _client->set_method(HttpMethod::GET);
+
+ // Determine HTTP method from configuration (default: GET)
+ HttpMethod method = HttpMethod::GET;
+ auto method_iter = _extend_kv.find("http.method");
+ if (method_iter != _extend_kv.end()) {
+ method = to_http_method(method_iter->second.c_str());
+ if (method == HttpMethod::UNKNOWN) {
+ LOG(WARNING) << "Invalid http.method value: " << method_iter->second
+ << ", falling back to GET";
+ method = HttpMethod::GET;
+ }
+ }
+ _client->set_method(method);
+
+ // Set payload if configured (supports POST, PUT, DELETE, etc.)
+ auto payload_iter = _extend_kv.find("http.payload");
+ if (payload_iter != _extend_kv.end() && !payload_iter->second.empty()) {
+ _client->set_payload(payload_iter->second);
+ _client->set_content_type("application/json");
+ VLOG(2) << "HTTP request with payload, size=" << payload_iter->second.size();
+ }
Review Comment:
`HttpFileReader` now honors user-supplied `http.method` and `http.payload`, which allows arbitrary non-GET requests (with bodies) from BE to any HTTP endpoint. This expands the attack surface beyond read-only SSRF (e.g. POST/DELETE side effects). Consider restricting these options to the CDC path only (e.g. require `enable_cdc_client=true` and/or only allow loopback URLs), and/or whitelist allowed methods (likely just POST) and reject others explicitly.
##########
fe/fe-core/src/main/java/org/apache/doris/job/util/StreamingJobUtils.java:
##########
@@ -461,7 +461,7 @@ private static String getRemoteDbName(DataSourceType sourceType, Map<String, Str
Preconditions.checkArgument(StringUtils.isNotEmpty(remoteDb), "schema is required");
break;
default:
- throw new JobException("Unsupported source type " + sourceType);
+ throw new RuntimeException("Unsupported source type " + sourceType);
}
Review Comment:
`getRemoteDbName` used to throw `JobException` (checked) but now throws/returns `RuntimeException`, and it’s called from job code paths like `generateCreateTableCmds(...)` that declare `throws JobException`. This can bypass existing error handling/reporting for streaming jobs. Consider keeping `JobException` here (or wrapping unsupported source type / missing schema|database in `JobException`) and let the TVF convert that to an `AnalysisException` at its boundary.
##########
fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/utils/ConfigUtil.java:
##########
@@ -107,6 +109,21 @@ public static Properties getDefaultDebeziumProps() {
return properties;
}
+ public static String[] getTableList(String schema, Map<String, String> cdcConfig) {
+ String includingTables = cdcConfig.get(DataSourceConfigKeys.INCLUDE_TABLES);
+ String table = cdcConfig.get(DataSourceConfigKeys.TABLE);
+ if (StringUtils.isNotEmpty(includingTables)) {
+ return Arrays.stream(includingTables.split(","))
+ .map(t -> schema + "." + t.trim())
+ .toArray(String[]::new);
+ } else if (StringUtils.isNotEmpty(table)) {
+ Preconditions.checkArgument(!table.contains(","), "table only supports one table");
+ return new String[] {schema + "." + table.trim()};
Review Comment:
`getTableList()` trims tokens but doesn’t filter out empty entries (e.g. `include_tables="t1, ,t2"` will produce `schema.""`). Consider filtering blank table names after `trim()` (and possibly validating against `.` to prevent already-qualified names) to fail fast with a clearer error.
```suggestion
String[] tableNames = Arrays.stream(includingTables.split(","))
.map(String::trim)
.filter(t -> !t.isEmpty())
.toArray(String[]::new);
Preconditions.checkArgument(tableNames.length > 0,
"include_tables must contain at least one non-blank table name");
for (String t : tableNames) {
Preconditions.checkArgument(!t.contains("."),
"Table name should not be schema-qualified in include_tables: %s", t);
}
return Arrays.stream(tableNames)
.map(t -> schema + "." + t)
.toArray(String[]::new);
} else if (StringUtils.isNotEmpty(table)) {
Preconditions.checkArgument(!table.contains(","), "table only supports one table");
String trimmedTable = table.trim();
Preconditions.checkArgument(!trimmedTable.isEmpty(),
"table must not be blank");
Preconditions.checkArgument(!trimmedTable.contains("."),
"Table name should not be schema-qualified in table: %s", trimmedTable);
return new String[] {schema + "." + trimmedTable};
```
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [Feature](tvf) Support cdc stream tvf for mysql and pg [doris]
Posted by "doris-robot (via GitHub)" <gi...@apache.org>.
doris-robot commented on PR #60116:
URL: https://github.com/apache/doris/pull/60116#issuecomment-4124594728
<details>
<summary>TPC-DS: <b>Total hot run time: 169257 ms</b></summary>
```
machine: 'aliyun_ecs.c7a.8xlarge_32C64G'
scripts: https://github.com/apache/doris/tree/master/tools/tpcds-tools
TPC-DS sf100 test result on commit 7afa17eeb0f2d0735b05ec72eeaf3bf3cb6a96a4, data reload: false
query5 4355 630 542 542
query6 343 249 227 227
query7 4224 471 270 270
query8 354 251 231 231
query9 8749 2736 2728 2728
query10 538 382 346 346
query11 6913 5151 4894 4894
query12 196 133 125 125
query13 1278 451 349 349
query14 5773 3757 3523 3523
query14_1 2868 2846 2830 2830
query15 211 196 177 177
query16 985 472 435 435
query17 891 719 624 624
query18 2461 448 355 355
query19 221 218 195 195
query20 141 127 129 127
query21 221 140 109 109
query22 13273 14174 14694 14174
query23 16889 16157 15966 15966
query23_1 16203 15609 15736 15609
query24 7111 1630 1214 1214
query24_1 1231 1251 1213 1213
query25 553 466 404 404
query26 1237 264 148 148
query27 2785 489 292 292
query28 4460 1824 1815 1815
query29 847 565 475 475
query30 303 233 193 193
query31 1015 950 871 871
query32 83 71 73 71
query33 511 337 293 293
query34 881 872 511 511
query35 646 663 596 596
query36 1107 1139 1003 1003
query37 135 97 87 87
query38 2957 2923 2881 2881
query39 873 842 830 830
query39_1 797 807 835 807
query40 234 159 136 136
query41 63 61 61 61
query42 257 254 255 254
query43 244 245 224 224
query44
query45 202 188 183 183
query46 872 986 611 611
query47 2120 2156 2050 2050
query48 330 310 226 226
query49 641 472 378 378
query50 679 272 209 209
query51 4053 4091 4093 4091
query52 262 272 260 260
query53 299 336 289 289
query54 304 275 264 264
query55 91 87 85 85
query56 308 331 315 315
query57 1917 1684 1686 1684
query58 282 276 272 272
query59 2791 2977 2750 2750
query60 352 340 327 327
query61 164 150 159 150
query62 641 589 537 537
query63 315 278 278 278
query64 5098 1324 999 999
query65
query66 1466 451 356 356
query67 24375 24305 24271 24271
query68
query69 414 311 298 298
query70 968 992 961 961
query71 354 302 295 295
query72 2823 2731 2456 2456
query73 544 551 319 319
query74 9662 9605 9433 9433
query75 2885 2788 2483 2483
query76 2289 1019 667 667
query77 364 428 314 314
query78 10956 11113 10462 10462
query79 1109 787 578 578
query80 1342 625 539 539
query81 553 265 234 234
query82 970 157 122 122
query83 337 262 247 247
query84 304 120 102 102
query85 929 494 463 463
query86 442 310 322 310
query87 3163 3181 3095 3095
query88 3523 2648 2624 2624
query89 441 369 349 349
query90 2018 169 178 169
query91 175 180 135 135
query92 76 72 76 72
query93 914 852 494 494
query94 657 328 310 310
query95 592 403 317 317
query96 648 509 227 227
query97 2459 2495 2383 2383
query98 246 231 227 227
query99 973 995 916 916
Total cold run time: 250262 ms
Total hot run time: 169257 ms
```
</details>
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [Feature](tvf) Support cdc stream tvf for mysql and pg [doris]
Posted by "JNSimba (via GitHub)" <gi...@apache.org>.
JNSimba commented on PR #60116:
URL: https://github.com/apache/doris/pull/60116#issuecomment-3777223851
run buildall
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [Feature](tvf) Support cdc stream tvf for mysql and pg [doris]
Posted by "doris-robot (via GitHub)" <gi...@apache.org>.
doris-robot commented on PR #60116:
URL: https://github.com/apache/doris/pull/60116#issuecomment-3782468391
<details>
<summary>TPC-H: <b>Total hot run time: 31405 ms</b></summary>
```
machine: 'aliyun_ecs.c7a.8xlarge_32C64G'
scripts: https://github.com/apache/doris/tree/master/tools/tpch-tools
Tpch sf100 test result on commit b56d1d11f9b7fee728fc06d90ddd84506f443adf, data reload: false
------ Round 1 ----------------------------------
q1 17687 4722 4601 4601
q2 2044 330 197 197
q3 10188 1326 733 733
q4 10220 908 315 315
q5 7608 2118 1897 1897
q6 215 171 142 142
q7 903 724 599 599
q8 9279 1402 1126 1126
q9 5362 4603 4754 4603
q10 6826 1662 1276 1276
q11 525 281 266 266
q12 372 388 225 225
q13 17787 3834 3108 3108
q14 235 233 213 213
q15 615 518 526 518
q16 671 665 583 583
q17 653 793 503 503
q18 6645 6460 6819 6460
q19 1625 997 671 671
q20 436 377 241 241
q21 3109 2363 2125 2125
q22 1089 1071 1003 1003
Total cold run time: 104094 ms
Total hot run time: 31405 ms
----- Round 2, with runtime_filter_mode=off -----
q1 5195 4900 5000 4900
q2 326 394 346 346
q3 2416 2956 2469 2469
q4 1392 1887 1461 1461
q5 4568 4452 4299 4299
q6 230 176 133 133
q7 2129 1970 1813 1813
q8 2591 2444 2469 2444
q9 7191 7151 7154 7151
q10 2777 2685 2356 2356
q11 568 474 451 451
q12 726 756 566 566
q13 3407 3856 3081 3081
q14 273 287 259 259
q15 547 509 497 497
q16 634 664 612 612
q17 1107 1282 1321 1282
q18 7268 7312 7141 7141
q19 866 807 828 807
q20 1935 1978 1862 1862
q21 4543 4244 4125 4125
q22 1085 1056 1014 1014
Total cold run time: 51774 ms
Total hot run time: 49069 ms
```
</details>
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [Feature](tvf) Support cdc stream tvf for mysql and pg [doris]
Posted by "JNSimba (via GitHub)" <gi...@apache.org>.
JNSimba commented on PR #60116:
URL: https://github.com/apache/doris/pull/60116#issuecomment-4117268860
run buildall
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [Feature](tvf) Support cdc stream tvf for mysql and pg [doris]
Posted by "JNSimba (via GitHub)" <gi...@apache.org>.
JNSimba commented on PR #60116:
URL: https://github.com/apache/doris/pull/60116#issuecomment-4124362317
run buildall
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [Feature](tvf) Support cdc stream tvf for mysql and pg [doris]
Posted by "doris-robot (via GitHub)" <gi...@apache.org>.
doris-robot commented on PR #60116:
URL: https://github.com/apache/doris/pull/60116#issuecomment-4124539361
<details>
<summary>TPC-H: <b>Total hot run time: 26636 ms</b></summary>
```
machine: 'aliyun_ecs.c7a.8xlarge_32C64G'
scripts: https://github.com/apache/doris/tree/master/tools/tpch-tools
Tpch sf100 test result on commit 7afa17eeb0f2d0735b05ec72eeaf3bf3cb6a96a4, data reload: false
------ Round 1 ----------------------------------
orders Doris NULL NULL 0 0 0 NULL 0 NULL NULL 2023-12-26 18:27:23 2023-12-26 18:42:55 NULL utf-8 NULL NULL
============================================
q1 17607 4544 4339 4339
q2 q3 10649 825 547 547
q4 4678 347 249 249
q5 7581 1197 1016 1016
q6 179 174 146 146
q7 788 850 670 670
q8 9313 1501 1309 1309
q9 4978 4768 4745 4745
q10 6246 1935 1650 1650
q11 462 250 251 250
q12 700 581 464 464
q13 18038 2730 1948 1948
q14 225 236 207 207
q15 q16 759 734 665 665
q17 722 856 454 454
q18 6143 5501 5185 5185
q19 1230 1006 601 601
q20 545 513 382 382
q21 4754 1855 1519 1519
q22 473 366 290 290
Total cold run time: 96070 ms
Total hot run time: 26636 ms
----- Round 2, with runtime_filter_mode=off -----
orders Doris NULL NULL 150000000 42 6422171781 NULL 22778155 NULL NULL 2023-12-26 18:27:23 2023-12-26 18:42:55 NULL utf-8 NULL NULL
============================================
q1 4743 4641 4595 4595
q2 q3 3920 4443 3808 3808
q4 892 1209 765 765
q5 4084 4426 4344 4344
q6 198 179 140 140
q7 1818 1663 1529 1529
q8 2532 2750 2629 2629
q9 8083 7527 7441 7441
q10 3788 4024 3607 3607
q11 544 432 417 417
q12 488 571 461 461
q13 2502 3233 2076 2076
q14 283 300 283 283
q15 q16 728 775 725 725
q17 1194 1411 1366 1366
q18 7124 6911 6686 6686
q19 927 883 921 883
q20 2067 2167 2093 2093
q21 4028 3526 3380 3380
q22 497 427 388 388
Total cold run time: 50440 ms
Total hot run time: 47616 ms
```
</details>
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [Feature](tvf) Support cdc stream tvf for mysql and pg [doris]
Posted by "doris-robot (via GitHub)" <gi...@apache.org>.
doris-robot commented on PR #60116:
URL: https://github.com/apache/doris/pull/60116#issuecomment-3782497599
<details>
<summary>ClickBench: <b>Total hot run time: 26.8 s</b></summary>
```
machine: 'aliyun_ecs.c7a.8xlarge_32C64G'
scripts: https://github.com/apache/doris/tree/master/tools/clickbench-tools
ClickBench test result on commit b56d1d11f9b7fee728fc06d90ddd84506f443adf, data reload: false
query1 0.04 0.05 0.05
query2 0.10 0.05 0.05
query3 0.26 0.09 0.08
query4 1.61 0.12 0.11
query5 0.28 0.25 0.25
query6 1.17 0.66 0.66
query7 0.03 0.03 0.03
query8 0.05 0.04 0.04
query9 0.55 0.50 0.49
query10 0.56 0.54 0.55
query11 0.14 0.10 0.10
query12 0.15 0.11 0.12
query13 0.60 0.58 0.58
query14 0.94 0.94 0.95
query15 0.78 0.79 0.79
query16 0.39 0.39 0.40
query17 1.04 1.05 1.04
query18 0.23 0.21 0.21
query19 1.81 1.79 1.85
query20 0.02 0.01 0.01
query21 15.44 0.28 0.14
query22 4.91 0.08 0.05
query23 15.74 0.29 0.11
query24 1.04 0.30 0.54
query25 0.12 0.08 0.05
query26 0.14 0.14 0.13
query27 0.07 0.05 0.08
query28 4.01 1.10 0.89
query29 12.60 3.90 3.11
query30 0.28 0.14 0.12
query31 2.82 0.64 0.40
query32 3.24 0.56 0.45
query33 2.94 3.09 3.05
query34 16.06 5.12 4.40
query35 4.49 4.54 4.45
query36 0.67 0.50 0.49
query37 0.12 0.06 0.07
query38 0.07 0.04 0.03
query39 0.04 0.03 0.03
query40 0.17 0.15 0.14
query41 0.09 0.03 0.03
query42 0.04 0.03 0.03
query43 0.06 0.03 0.03
Total cold run time: 95.91 s
Total hot run time: 26.8 s
```
</details>
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [Feature](tvf) Support cdc stream tvf for mysql and pg [doris]
Posted by "hello-stephen (via GitHub)" <gi...@apache.org>.
hello-stephen commented on PR #60116:
URL: https://github.com/apache/doris/pull/60116#issuecomment-3783038226
# FE UT Coverage Report
Increment line coverage ` 3.08% (2/65)` :tada:
[Increment coverage report](http://coverage.selectdb-in.cc/coverage/60116_e02811eaefda6a3a1631ee0683594c271015510b/fe_increment_report/index.html)
[Complete coverage report](http://coverage.selectdb-in.cc/coverage/60116_e02811eaefda6a3a1631ee0683594c271015510b/fe_report/index.html)
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [Feature](tvf) Support cdc stream tvf for mysql and pg [doris]
Posted by "hello-stephen (via GitHub)" <gi...@apache.org>.
hello-stephen commented on PR #60116:
URL: https://github.com/apache/doris/pull/60116#issuecomment-3777162956
Thank you for your contribution to Apache Doris.
Don't know what should be done next? See [How to process your PR](https://cwiki.apache.org/confluence/display/DORIS/How+to+process+your+PR).
Please clearly describe your PR:
1. What problem was fixed (it's best to include specific error reporting information). How it was fixed.
2. Which behaviors were modified. What was the previous behavior, what is it now, why was it modified, and what possible impacts might there be.
3. What features were added. Why was this function added?
4. Which code was refactored and why was this part of the code refactored?
5. Which functions were optimized and what is the difference before and after the optimization?
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [Feature](tvf) Support cdc stream tvf for mysql and pg [doris]
Posted by "JNSimba (via GitHub)" <gi...@apache.org>.
JNSimba commented on PR #60116:
URL: https://github.com/apache/doris/pull/60116#issuecomment-4124916326
run buildall
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [Feature](tvf) Support cdc stream tvf for mysql and pg [doris]
Posted by "doris-robot (via GitHub)" <gi...@apache.org>.
doris-robot commented on PR #60116:
URL: https://github.com/apache/doris/pull/60116#issuecomment-4116679409
<details>
<summary>TPC-H: <b>Total hot run time: 26168 ms</b></summary>
```
machine: 'aliyun_ecs.c7a.8xlarge_32C64G'
scripts: https://github.com/apache/doris/tree/master/tools/tpch-tools
Tpch sf100 test result on commit d4210f02b4944ba2c18879a12b940c093405187b, data reload: false
------ Round 1 ----------------------------------
orders Doris NULL NULL 0 0 0 NULL 0 NULL NULL 2023-12-26 18:27:23 2023-12-26 18:42:55 NULL utf-8 NULL NULL
============================================
q1 17628 4399 4267 4267
q2 q3 10646 759 526 526
q4 4681 354 246 246
q5 7544 1221 1017 1017
q6 173 179 149 149
q7 777 850 659 659
q8 9302 1461 1344 1344
q9 4921 4777 4489 4489
q10 6252 1926 1633 1633
q11 489 252 246 246
q12 704 587 468 468
q13 18055 2713 1920 1920
q14 230 231 223 223
q15 q16 722 715 659 659
q17 740 866 419 419
q18 6006 5344 5222 5222
q19 1259 977 622 622
q20 537 492 375 375
q21 4714 1843 1435 1435
q22 347 294 249 249
Total cold run time: 95727 ms
Total hot run time: 26168 ms
----- Round 2, with runtime_filter_mode=off -----
orders Doris NULL NULL 150000000 42 6422171781 NULL 22778155 NULL NULL 2023-12-26 18:27:23 2023-12-26 18:42:55 NULL utf-8 NULL NULL
============================================
q1 4963 4645 4542 4542
q2 q3 3862 4391 3815 3815
q4 893 1202 798 798
q5 4094 4362 4354 4354
q6 182 185 153 153
q7 1771 1643 1606 1606
q8 2501 2755 2525 2525
q9 7639 7601 7377 7377
q10 3748 4031 3596 3596
q11 533 441 417 417
q12 463 612 434 434
q13 2468 2978 2096 2096
q14 284 307 291 291
q15 q16 722 754 723 723
q17 1228 1313 1345 1313
q18 7225 6713 6733 6713
q19 929 900 913 900
q20 2047 2145 1957 1957
q21 3954 3557 3340 3340
q22 450 429 378 378
Total cold run time: 49956 ms
Total hot run time: 47328 ms
```
</details>
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [Feature](tvf) Support cdc stream tvf for mysql and pg [doris]
Posted by "hello-stephen (via GitHub)" <gi...@apache.org>.
hello-stephen commented on PR #60116:
URL: https://github.com/apache/doris/pull/60116#issuecomment-4117795773
# FE UT Coverage Report
Increment line coverage ` 2.90% (2/69)` :tada:
[Increment coverage report](http://coverage.selectdb-in.cc/coverage/60116_aa433727e52f6f04fafef418019df4bd5b6a3cd7/fe_increment_report/index.html)
[Complete coverage report](http://coverage.selectdb-in.cc/coverage/60116_aa433727e52f6f04fafef418019df4bd5b6a3cd7/fe_report/index.html)
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [Feature](tvf) Support cdc stream tvf for mysql and pg [doris]
Posted by "JNSimba (via GitHub)" <gi...@apache.org>.
JNSimba commented on PR #60116:
URL: https://github.com/apache/doris/pull/60116#issuecomment-4117773728
run p0
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [Feature](tvf) Support cdc stream tvf for mysql and pg [doris]
Posted by "doris-robot (via GitHub)" <gi...@apache.org>.
doris-robot commented on PR #60116:
URL: https://github.com/apache/doris/pull/60116#issuecomment-3782712077
# BE UT Coverage Report
Increment line coverage `0.00% (0/79)` :tada:
[Increment coverage report](http://coverage.selectdb-in.cc/coverage/b56d1d11f9b7fee728fc06d90ddd84506f443adf_b56d1d11f9b7fee728fc06d90ddd84506f443adf/increment_report/index.html)
[Complete coverage report](http://coverage.selectdb-in.cc/coverage/b56d1d11f9b7fee728fc06d90ddd84506f443adf_b56d1d11f9b7fee728fc06d90ddd84506f443adf/report/index.html)
| Category | Coverage |
|-------------------|--------------------|
| Function Coverage | 52.46% (19116/36440) |
| Line Coverage | 35.83% (177584/495676) |
| Region Coverage | 32.30% (137314/425099) |
| Branch Coverage | 33.24% (59437/178833) |
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [Feature](tvf) Support cdc stream tvf for mysql and pg [doris]
Posted by "doris-robot (via GitHub)" <gi...@apache.org>.
doris-robot commented on PR #60116:
URL: https://github.com/apache/doris/pull/60116#issuecomment-4116743431
<details>
<summary>TPC-DS: <b>Total hot run time: 168504 ms</b></summary>
```
machine: 'aliyun_ecs.c7a.8xlarge_32C64G'
scripts: https://github.com/apache/doris/tree/master/tools/tpcds-tools
TPC-DS sf100 test result on commit d4210f02b4944ba2c18879a12b940c093405187b, data reload: false
query5 4304 636 511 511
query6 340 222 201 201
query7 4209 464 262 262
query8 341 252 226 226
query9 8724 2702 2712 2702
query10 519 382 347 347
query11 6964 5093 4888 4888
query12 191 135 129 129
query13 1274 460 371 371
query14 5786 3677 3479 3479
query14_1 2798 2803 2816 2803
query15 207 194 173 173
query16 987 463 453 453
query17 896 729 644 644
query18 2438 464 350 350
query19 218 213 187 187
query20 139 129 128 128
query21 216 134 108 108
query22 13211 14088 14732 14088
query23 16845 16190 15871 15871
query23_1 15916 15556 15610 15556
query24 7326 1636 1237 1237
query24_1 1230 1235 1275 1235
query25 523 461 407 407
query26 1245 260 147 147
query27 2799 480 294 294
query28 4471 1827 1819 1819
query29 836 561 474 474
query30 305 230 189 189
query31 1032 945 863 863
query32 88 70 70 70
query33 505 344 284 284
query34 887 870 523 523
query35 642 687 607 607
query36 1118 1134 1019 1019
query37 128 93 77 77
query38 2926 2892 2922 2892
query39 865 904 825 825
query39_1 792 794 791 791
query40 231 152 134 134
query41 64 60 59 59
query42 262 257 253 253
query43 250 254 220 220
query44
query45 194 185 184 184
query46 882 969 595 595
query47 2125 2126 2076 2076
query48 315 318 231 231
query49 628 452 383 383
query50 691 287 215 215
query51 4037 4059 4002 4002
query52 262 269 255 255
query53 288 333 279 279
query54 293 268 267 267
query55 92 84 85 84
query56 316 314 321 314
query57 1911 1664 1710 1664
query58 279 277 272 272
query59 2829 2939 2742 2742
query60 340 340 335 335
query61 156 154 156 154
query62 632 581 545 545
query63 312 282 273 273
query64 5013 1265 1012 1012
query65
query66 1476 450 359 359
query67 24286 24323 24172 24172
query68
query69 411 312 286 286
query70 907 966 870 870
query71 346 315 303 303
query72 2846 2605 2448 2448
query73 535 539 319 319
query74 9629 9613 9396 9396
query75 2825 2755 2474 2474
query76 2268 1033 682 682
query77 354 372 300 300
query78 10954 11177 10506 10506
query79 1112 766 572 572
query80 687 617 542 542
query81 485 264 222 222
query82 1352 153 114 114
query83 366 260 249 249
query84 297 117 92 92
query85 853 490 448 448
query86 363 300 293 293
query87 3134 3122 2990 2990
query88 3529 2654 2644 2644
query89 426 366 351 351
query90 1982 181 174 174
query91 164 161 133 133
query92 77 80 70 70
query93 913 870 490 490
query94 444 335 290 290
query95 608 392 319 319
query96 651 511 232 232
query97 2467 2472 2399 2399
query98 235 221 217 217
query99 992 929 899 899
Total cold run time: 249229 ms
Total hot run time: 168504 ms
```
</details>
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [Feature](tvf) Support cdc stream tvf for mysql and pg [doris]
Posted by "JNSimba (via GitHub)" <gi...@apache.org>.
JNSimba commented on PR #60116:
URL: https://github.com/apache/doris/pull/60116#issuecomment-4117335226
run buildall
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [Feature](tvf) Support cdc stream tvf for mysql and pg [doris]
Posted by "doris-robot (via GitHub)" <gi...@apache.org>.
doris-robot commented on PR #60116:
URL: https://github.com/apache/doris/pull/60116#issuecomment-4125656262
# BE UT Coverage Report
Increment line coverage ` ` :tada:
[Increment coverage report](http://coverage.selectdb-in.cc/coverage/9ea6fe06a5cddc20a17b3f4565082396852f0a56_9ea6fe06a5cddc20a17b3f4565082396852f0a56/increment_report/index.html)
[Complete coverage report](http://coverage.selectdb-in.cc/coverage/9ea6fe06a5cddc20a17b3f4565082396852f0a56_9ea6fe06a5cddc20a17b3f4565082396852f0a56/report/index.html)
| Category | Coverage |
|-------------------|--------------------|
| Function Coverage | 52.82% (19872/37622) |
| Line Coverage | 36.32% (185652/511210) |
| Region Coverage | 32.55% (143743/441670) |
| Branch Coverage | 33.77% (62962/186442) |
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [Feature](tvf) Support cdc stream tvf for mysql and pg [doris]
Posted by "hello-stephen (via GitHub)" <gi...@apache.org>.
hello-stephen commented on PR #60116:
URL: https://github.com/apache/doris/pull/60116#issuecomment-4117757105
# BE UT Coverage Report
Increment line coverage ` ` :tada:
[Increment coverage report](http://coverage.selectdb-in.cc/coverage/aa433727e52f6f04fafef418019df4bd5b6a3cd7_aa433727e52f6f04fafef418019df4bd5b6a3cd7/increment_report/index.html)
[Complete coverage report](http://coverage.selectdb-in.cc/coverage/aa433727e52f6f04fafef418019df4bd5b6a3cd7_aa433727e52f6f04fafef418019df4bd5b6a3cd7/report/index.html)
| Category | Coverage |
|-------------------|--------------------|
| Function Coverage | 52.79% (19858/37615) |
| Line Coverage | 36.29% (185527/511253) |
| Region Coverage | 32.55% (143685/441478) |
| Branch Coverage | 33.77% (62934/186373) |
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [Feature](tvf) Support cdc stream tvf for mysql and pg [doris]
Posted by "dataroaring (via GitHub)" <gi...@apache.org>.
dataroaring commented on PR #60116:
URL: https://github.com/apache/doris/pull/60116#issuecomment-4124008983
## Test Coverage Analysis
The regression tests validate the golden-path end-to-end, which is valuable for integration confidence. However, the error surface is largely untested. Here's a detailed mapping of code paths vs. test coverage.
### What's Covered
| # | Scenario | Verdict |
|---|----------|---------|
| 1 | MySQL: read from specific binlog offset (file+pos) | **Covered** — asserts C1/D1 rows |
| 2 | MySQL: reject `initial` offset | **Covered** |
| 3 | PostgreSQL: reject `initial` offset | **Covered** |
| 4 | PostgreSQL: `latest` offset executes | **Weak** — result logged but never asserted |
### Gap 1: Zero validation error tests
`CdcStreamTableValuedFunction.validate()` has 4 required-field checks. None are tested:
```java
if (!properties.containsKey(DataSourceConfigKeys.JDBC_URL)) // ❌ not tested
if (!properties.containsKey(DataSourceConfigKeys.TYPE)) // ❌ not tested
if (!properties.containsKey(DataSourceConfigKeys.TABLE)) // ❌ not tested
if (!properties.containsKey(DataSourceConfigKeys.OFFSET)) // ❌ not tested
```
Suggest adding at least:
```groovy
test {
sql """select * from cdc_stream("type" = "mysql")"""
exception "jdbc_url is required"
}
```
Additionally, `database` is **not validated** but **required downstream** by `getRemoteDbName()` — a missing `database` produces a confusing NPE instead of a clear error. Both validation and test are missing.
### Gap 2: Only 1 of 5 offset modes tested for MySQL
| Offset Mode | MySQL | PostgreSQL |
|-------------|-------|------------|
| `initial` | ✅ (error) | ✅ (error) |
| `earliest` | ❌ | ❌ |
| `latest` | ❌ | ⚠️ (no assertion) |
| JSON `{file, pos}` | ✅ | ❌ |
| Malformed/invalid string | ❌ | ❌ |
`earliest` goes through a completely different code path and should have at least one test.
### Gap 3: PostgreSQL result never asserted
```groovy
def result = sql """select * from cdc_stream(... "offset" = 'latest')"""
log.info("result:", result) // logged but NEVER asserted
```
This only verifies "no exception thrown." Even if PG slot creation makes exact offset-based tests difficult, the result should at minimum be checked for non-null or expected shape.
### Gap 4: Only INSERT tested — no UPDATE/DELETE
Both tests only insert rows. CDC fundamentally captures INSERT, UPDATE, and DELETE. The Debezium deserialization path for `Envelope.Operation.UPDATE` and `DELETE` is completely untested through the TVF.
### Gap 5: No data type coverage
Test tables use only `varchar` + `int`. Not tested:
- `DATETIME`/`TIMESTAMP` — CDC timestamp handling is notoriously tricky
- `DECIMAL`/`NUMERIC` — precision loss risk
- `NULL` values
Compare with the existing `test_streaming_mysql_job_all_type.groovy` which has comprehensive type coverage.
### Gap 6: No error handling tests for connectivity/auth failures
| Error Scenario | Code Path | Test |
|---|---|---|
| Unreachable JDBC URL | `getTableColumns()` → connection failure | ❌ |
| Wrong credentials | `getTableColumns()` → auth failure | ❌ |
| Non-existent table | `throw new AnalysisException("Table does not exist")` | ❌ |
| Invalid `type` (e.g., "oracle") | `DataSourceType.valueOf()` → IllegalArgumentException | ❌ |
### Gap 7: No unit tests for new utility methods
| Method | Branches | Unit Test |
|---|---|---|
| `ConfigUtil.getTableList()` | 4 (include_tables / single table / table with comma / neither) | ❌ |
| `ConfigUtil.getServerId(String)` | signature changed from `long` to `String` | ❌ |
| `PipelineCoordinator.isLong()` | critical TVF-vs-job discriminator | ❌ |
| `PipelineCoordinator.generateMeta()` | 3 branches (latest/earliest, JSON, unsupported) | ❌ |
### Summary
| Category | Covered / Total | Coverage |
|---|---|---|
| Happy path (MySQL offset modes) | 1 / 5 | ~20% |
| Happy path (PostgreSQL offset modes) | 0 / 4 (1 runs but unasserted) | ~0% |
| Validation errors | 0 / 5 | 0% |
| DML operations (INSERT/UPDATE/DELETE) | 1 / 3 | 33% |
| Data types | 1 basic / many | ~10% |
| Connectivity/auth errors | 0 / 4 | 0% |
| New utility method unit tests | 0 / 4 methods | 0% |
For a feature that connects to external databases over JDBC and streams data through HTTP, the error paths are arguably more important to test than the happy paths — they're what users encounter when something goes wrong.
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [Feature](tvf) Support cdc stream tvf for mysql and pg [doris]
Posted by "JNSimba (via GitHub)" <gi...@apache.org>.
JNSimba merged PR #60116:
URL: https://github.com/apache/doris/pull/60116
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [Feature](tvf) Support cdc stream tvf for mysql and pg [doris]
Posted by "hello-stephen (via GitHub)" <gi...@apache.org>.
hello-stephen commented on PR #60116:
URL: https://github.com/apache/doris/pull/60116#issuecomment-4118705519
# BE Regression && UT Coverage Report
Increment line coverage `100% (0/0)` :tada:
[Increment coverage report](http://coverage.selectdb-in.cc/coverage/60116_aa433727e52f6f04fafef418019df4bd5b6a3cd7_merge/increment_report/index.html)
[Complete coverage report](http://coverage.selectdb-in.cc/coverage/60116_aa433727e52f6f04fafef418019df4bd5b6a3cd7_merge/report/index.html)
| Category | Coverage |
|-------------------|--------------------|
| Function Coverage | 73.33% (27009/36832) |
| Line Coverage | 56.79% (289483/509701) |
| Region Coverage | 54.03% (240749/445618) |
| Branch Coverage | 55.81% (104323/186941) |
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [Feature](tvf) Support cdc stream tvf for mysql and pg [doris]
Posted by "liaoxin01 (via GitHub)" <gi...@apache.org>.
liaoxin01 commented on PR #60116:
URL: https://github.com/apache/doris/pull/60116#issuecomment-3802772193
### Code review
No issues found. Checked for bugs and CLAUDE.md compliance.
🤖 Generated with [Claude Code](https://claude.ai/code)
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [Feature](tvf) Support cdc stream tvf for mysql and pg [doris]
Posted by "JNSimba (via GitHub)" <gi...@apache.org>.
JNSimba commented on code in PR #60116:
URL: https://github.com/apache/doris/pull/60116#discussion_r2715155832
##########
fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/source/reader/mysql/MySqlSourceReader.java:
##########
@@ -170,39 +169,14 @@ public SplitReadResult readSplitRecords(JobBaseRecordRequest baseReq) throws Exc
if (offsetMeta == null || offsetMeta.isEmpty()) {
throw new RuntimeException("miss meta offset");
}
- LOG.info("Job {} read split records with offset: {}", baseReq.getJobId(), offsetMeta);
-
- // If there is an active split being consumed, reuse it directly;
- // Otherwise, create a new snapshot/binlog split based on offset and start the reader.
- MySqlSplit split = null;
- SplitRecords currentSplitRecords = this.getCurrentSplitRecords();
- if (currentSplitRecords == null) {
- DebeziumReader<SourceRecords, MySqlSplit> currentReader = this.getCurrentReader();
- if (baseReq.isReload() || currentReader == null) {
- LOG.info(
- "No current reader or reload {}, create new split reader",
- baseReq.isReload());
- // build split
- Tuple2<MySqlSplit, Boolean> splitFlag = createMySqlSplit(offsetMeta, baseReq);
- split = splitFlag.f0;
- // reset binlog reader
- // closeBinlogReader();
- currentSplitRecords = pollSplitRecordsWithSplit(split, baseReq);
- this.setCurrentSplitRecords(currentSplitRecords);
- this.setCurrentSplit(split);
- } else if (currentReader instanceof BinlogSplitReader) {
- LOG.info("Continue poll records with current binlog reader");
- // only for binlog reader
- currentSplitRecords = pollSplitRecordsWithCurrentReader(currentReader);
- split = this.getCurrentSplit();
- } else {
- throw new RuntimeException("Should not happen");
- }
- } else {
- LOG.info(
- "Continue read records with current split records, splitId: {}",
- currentSplitRecords.getSplitId());
- }
+ // Create a new snapshot/binlog split based on offset and start the reader.
+ LOG.info("create new split reader for {} with offset {}", baseReq.getJobId(), offsetMeta);
+ // build split
+ Tuple2<MySqlSplit, Boolean> splitFlag = createMySqlSplit(offsetMeta, baseReq);
+ MySqlSplit split = splitFlag.f0;
+ // it's necessary to ensure that the binlog reader is already closed.
+ this.currentSplitRecords = pollSplitRecordsWithSplit(split, baseReq);
+ this.currentSplit = split;
Review Comment:
Currently, the reader is closed in finishSplitRecords after each read operation, so the other branches here are probably useless.
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [Feature](tvf) Support cdc stream tvf for mysql and pg [doris]
Posted by "hello-stephen (via GitHub)" <gi...@apache.org>.
hello-stephen commented on PR #60116:
URL: https://github.com/apache/doris/pull/60116#issuecomment-3783332292
# BE Regression && UT Coverage Report
Increment line coverage `65.82% (52/79)` :tada:
[Increment coverage report](http://coverage.selectdb-in.cc/coverage/60116_e02811eaefda6a3a1631ee0683594c271015510b_merge/increment_report/index.html)
[Complete coverage report](http://coverage.selectdb-in.cc/coverage/60116_e02811eaefda6a3a1631ee0683594c271015510b_merge/report/index.html)
| Category | Coverage |
|-------------------|--------------------|
| Function Coverage | 73.12% (26118/35719) |
| Line Coverage | 56.20% (278285/495144) |
| Region Coverage | 54.08% (232333/429612) |
| Branch Coverage | 55.65% (99946/179604) |
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [Feature](tvf) Support cdc stream tvf for mysql and pg [doris]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on code in PR #60116:
URL: https://github.com/apache/doris/pull/60116#discussion_r2979973254
##########
be/src/io/fs/http_file_reader.cpp:
##########
@@ -96,32 +111,64 @@ Status HttpFileReader::open(const FileReaderOptions& opts) {
return Status::OK();
}
- // Step 1: HEAD request to get file metadata
- RETURN_IF_ERROR(prepare_client(/*set_fail_on_error=*/true));
- _client->set_method(HttpMethod::HEAD);
- RETURN_IF_ERROR(_client->execute());
+ // start CDC client
+ auto enable_cdc_iter = _extend_kv.find("enable_cdc_client");
+ if (enable_cdc_iter != _extend_kv.end() && enable_cdc_iter->second == "true") {
+ LOG(INFO) << "CDC client is enabled, starting CDC client for " << _url;
+ ExecEnv* env = ExecEnv::GetInstance();
+ if (env == nullptr || env->cdc_client_mgr() == nullptr) {
+ return Status::InternalError("ExecEnv or CdcClientMgr is not initialized");
+ }
Review Comment:
**[Fragile `_file_size` when `_size_known=false`]**
When `_enable_chunk_response=true`, `_size_known` is set to `false` but `_file_size` remains at its default value `static_cast<size_t>(-1)` (i.e., `SIZE_MAX` ~18 exabytes). While the current CDC TVF path avoids the dangerous `_read_one_message()` allocation because `_read_json_by_line` defaults to `true`, this is **fragile**.
If anyone later adds `strip_outer_array=true` or `read_json_by_line=false` to the TVF, it would trigger an allocation of `SIZE_MAX` bytes in `NewJsonReader::_read_one_message()`, causing OOM/crash.
Consider updating `_file_size` after the chunked response completes, e.g., after caching in the non-Range download callback:
```cpp
_full_file_cache = buf;
_full_file_cached = true;
_file_size = _full_file_cache.size();
_size_known = true;
```
##########
fe/fe-core/src/main/java/org/apache/doris/tablefunction/CdcStreamTableValuedFunction.java:
##########
@@ -0,0 +1,130 @@
+// 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.tablefunction;
+
+import org.apache.doris.analysis.BrokerDesc;
+import org.apache.doris.analysis.StorageBackend.StorageType;
+import org.apache.doris.catalog.Column;
+import org.apache.doris.common.AnalysisException;
+import org.apache.doris.datasource.jdbc.client.JdbcClient;
+import org.apache.doris.job.cdc.DataSourceConfigKeys;
+import org.apache.doris.job.cdc.request.FetchRecordRequest;
+import org.apache.doris.job.common.DataSourceType;
+import org.apache.doris.job.util.StreamingJobUtils;
+import org.apache.doris.thrift.TBrokerFileStatus;
+import org.apache.doris.thrift.TFileType;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.base.Preconditions;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+
+public class CdcStreamTableValuedFunction extends ExternalFileTableValuedFunction {
+ private static final ObjectMapper objectMapper = new ObjectMapper();
+ private static final String URI = "http://127.0.0.1:CDC_CLIENT_PORT/api/fetchRecordStream";
+ private final Map<String, String> originProps;
+
+ public CdcStreamTableValuedFunction(Map<String, String> properties) throws AnalysisException {
+ this.originProps = properties;
+ validate(properties);
+ processProps(properties);
+ }
+
+ private void processProps(Map<String, String> properties) throws AnalysisException {
+ Map<String, String> copyProps = new HashMap<>(properties);
+ copyProps.put("format", "json");
+ super.parseCommonProperties(copyProps);
+ this.processedParams.put("enable_cdc_client", "true");
+ this.processedParams.put("uri", URI);
+ this.processedParams.put("http.enable.range.request", "false");
+ this.processedParams.put("http.enable.chunk.response", "true");
+ this.processedParams.put("http.method", "POST");
+
+ String payload = generateParams(properties);
+ this.processedParams.put("http.payload", payload);
+ this.backendConnectProperties.putAll(processedParams);
+ generateFileStatus();
+ }
+
+ private String generateParams(Map<String, String> properties) throws AnalysisException {
+ FetchRecordRequest recordRequest = new FetchRecordRequest();
+ recordRequest.setJobId(UUID.randomUUID().toString().replace("-", ""));
+ recordRequest.setDataSource(properties.get(DataSourceConfigKeys.TYPE));
+ recordRequest.setConfig(properties);
+ try {
+ return objectMapper.writeValueAsString(recordRequest);
+ } catch (IOException e) {
+ LOG.info("Failed to serialize fetch record request," + e.getMessage());
+ throw new AnalysisException(e.getMessage());
+ }
+ }
+
+ private void validate(Map<String, String> properties) {
+ Preconditions.checkArgument(properties.containsKey(DataSourceConfigKeys.JDBC_URL), "jdbc_url is required");
+ Preconditions.checkArgument(properties.containsKey(DataSourceConfigKeys.TYPE), "type is required");
+ Preconditions.checkArgument(properties.containsKey(DataSourceConfigKeys.TABLE), "table is required");
+ Preconditions.checkArgument(properties.containsKey(DataSourceConfigKeys.OFFSET), "offset is required");
Review Comment:
**[`validate()` throws `IllegalArgumentException`, not `AnalysisException`]**
`Preconditions.checkArgument` throws `IllegalArgumentException`, not `AnalysisException`. This is inconsistent with FE's exception model. The exception will propagate as a generic runtime exception rather than a proper analysis error with a user-friendly message.
Consider using `AnalysisException` directly:
```java
if (!properties.containsKey(DataSourceConfigKeys.JDBC_URL)) {
throw new AnalysisException("jdbc_url is required");
}
```
Alternatively, the `CdcStream.toCatalogFunction()` does catch `Throwable` and wraps it in `AnalysisException`, but that results in a less clear error message.
##########
fe/fe-core/src/main/java/org/apache/doris/job/util/StreamingJobUtils.java:
##########
@@ -448,8 +448,8 @@ public static List<Column> getColumns(JdbcClient jdbcClient,
* The remoteDB implementation differs for each data source;
* refer to the hierarchical mapping in the JDBC catalog.
*/
- private static String getRemoteDbName(DataSourceType sourceType, Map<String, String> properties)
- throws JobException {
+ public static String getRemoteDbName(DataSourceType sourceType, Map<String, String> properties)
Review Comment:
**[Exception type change from `JobException` to `RuntimeException`]**
`getRemoteDbName` previously threw `JobException` (a checked exception). Changing it to `RuntimeException` is a behavioral change that could silently break callers who previously caught `JobException` specifically. Existing callers that catch `JobException` will no longer catch errors from this method.
If the change is intentional (to avoid checked exceptions for the TVF path), consider documenting this or ensuring all callers are updated.
##########
fe/fe-core/src/main/java/org/apache/doris/tablefunction/CdcStreamTableValuedFunction.java:
##########
@@ -0,0 +1,130 @@
+// 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.tablefunction;
+
+import org.apache.doris.analysis.BrokerDesc;
+import org.apache.doris.analysis.StorageBackend.StorageType;
+import org.apache.doris.catalog.Column;
+import org.apache.doris.common.AnalysisException;
+import org.apache.doris.datasource.jdbc.client.JdbcClient;
+import org.apache.doris.job.cdc.DataSourceConfigKeys;
+import org.apache.doris.job.cdc.request.FetchRecordRequest;
+import org.apache.doris.job.common.DataSourceType;
+import org.apache.doris.job.util.StreamingJobUtils;
+import org.apache.doris.thrift.TBrokerFileStatus;
+import org.apache.doris.thrift.TFileType;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.base.Preconditions;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+
+public class CdcStreamTableValuedFunction extends ExternalFileTableValuedFunction {
+ private static final ObjectMapper objectMapper = new ObjectMapper();
+ private static final String URI = "http://127.0.0.1:CDC_CLIENT_PORT/api/fetchRecordStream";
+ private final Map<String, String> originProps;
+
+ public CdcStreamTableValuedFunction(Map<String, String> properties) throws AnalysisException {
+ this.originProps = properties;
+ validate(properties);
+ processProps(properties);
+ }
+
+ private void processProps(Map<String, String> properties) throws AnalysisException {
+ Map<String, String> copyProps = new HashMap<>(properties);
+ copyProps.put("format", "json");
+ super.parseCommonProperties(copyProps);
+ this.processedParams.put("enable_cdc_client", "true");
+ this.processedParams.put("uri", URI);
+ this.processedParams.put("http.enable.range.request", "false");
+ this.processedParams.put("http.enable.chunk.response", "true");
+ this.processedParams.put("http.method", "POST");
+
+ String payload = generateParams(properties);
+ this.processedParams.put("http.payload", payload);
+ this.backendConnectProperties.putAll(processedParams);
+ generateFileStatus();
+ }
+
+ private String generateParams(Map<String, String> properties) throws AnalysisException {
+ FetchRecordRequest recordRequest = new FetchRecordRequest();
+ recordRequest.setJobId(UUID.randomUUID().toString().replace("-", ""));
+ recordRequest.setDataSource(properties.get(DataSourceConfigKeys.TYPE));
+ recordRequest.setConfig(properties);
+ try {
+ return objectMapper.writeValueAsString(recordRequest);
+ } catch (IOException e) {
+ LOG.info("Failed to serialize fetch record request," + e.getMessage());
+ throw new AnalysisException(e.getMessage());
+ }
+ }
+
+ private void validate(Map<String, String> properties) {
+ Preconditions.checkArgument(properties.containsKey(DataSourceConfigKeys.JDBC_URL), "jdbc_url is required");
+ Preconditions.checkArgument(properties.containsKey(DataSourceConfigKeys.TYPE), "type is required");
+ Preconditions.checkArgument(properties.containsKey(DataSourceConfigKeys.TABLE), "table is required");
+ Preconditions.checkArgument(properties.containsKey(DataSourceConfigKeys.OFFSET), "offset is required");
+ }
+
+ private void generateFileStatus() {
+ this.fileStatuses.clear();
+ this.fileStatuses.add(new TBrokerFileStatus(URI, false, Integer.MAX_VALUE, false));
+ }
+
+ @Override
+ public List<Column> getTableColumns() throws AnalysisException {
+ DataSourceType dataSourceType =
+ DataSourceType.valueOf(processedParams.get(DataSourceConfigKeys.TYPE).toUpperCase());
+ JdbcClient jdbcClient = StreamingJobUtils.getJdbcClient(dataSourceType, processedParams);
Review Comment:
**[JdbcClient resource leak]**
`StreamingJobUtils.getJdbcClient()` creates a `JdbcClient` which holds JDBC connections. This method never closes it. The `JdbcClient` should be closed in a try-with-resources or finally block:
```java
JdbcClient jdbcClient = StreamingJobUtils.getJdbcClient(dataSourceType, processedParams);
try {
// use jdbcClient
} finally {
jdbcClient.closeClient();
}
```
##########
fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/service/PipelineCoordinator.java:
##########
@@ -51,26 +58,31 @@
import static org.apache.flink.cdc.connectors.base.utils.SourceRecordUtils.SCHEMA_HEARTBEAT_EVENT_KEY_NAME;
import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.common.base.Preconditions;
import io.debezium.data.Envelope;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;
+import org.springframework.web.servlet.mvc.method.annotation.StreamingResponseBody;
/** Pipeline coordinator. */
@Component
public class PipelineCoordinator {
private static final Logger LOG = LoggerFactory.getLogger(PipelineCoordinator.class);
private static final String SPLIT_ID = "splitId";
// jobId
- private final Map<Long, DorisBatchStreamLoad> batchStreamLoadMap = new ConcurrentHashMap<>();
+ private final Map<String, DorisBatchStreamLoad> batchStreamLoadMap = new ConcurrentHashMap<>();
+ // taskId, offset
Review Comment:
**[`taskOffsetCache` potential memory leak]**
The `taskOffsetCache` map is populated in `buildStreamRecords()` but only cleaned up when `getOffsetWithTaskId()` is called (which uses `remove()`). For the current TVF usage, `taskId` is null (not set in the TVF path per `generateParams()` in `CdcStreamTableValuedFunction`), so `StringUtils.isNotEmpty(fetchRecord.getTaskId())` is false and no entry is added — so this is safe for now.
However, if `taskId` is set (e.g., by a future caller), and `getOffsetWithTaskId()` is never called, entries accumulate indefinitely. Consider adding TTL-based eviction or a cleanup mechanism.
##########
be/src/io/fs/http_file_reader.cpp:
##########
@@ -224,9 +271,29 @@ Status HttpFileReader::read_at_impl(size_t offset, Slice result, size_t* bytes_r
VLOG(2) << "Issuing HTTP GET request: offset=" << offset << " req_len=" << req_len
<< " with_range=" << _range_supported;
- // Prepare and initialize the HTTP client for GET request
+ // Prepare and initialize the HTTP client for request
RETURN_IF_ERROR(prepare_client(/*set_fail_on_error=*/false));
- _client->set_method(HttpMethod::GET);
+
+ // Determine HTTP method from configuration (default: GET)
+ HttpMethod method = HttpMethod::GET;
+ auto method_iter = _extend_kv.find("http.method");
+ if (method_iter != _extend_kv.end()) {
+ method = to_http_method(method_iter->second.c_str());
+ if (method == HttpMethod::UNKNOWN) {
+ LOG(WARNING) << "Invalid http.method value: " << method_iter->second
+ << ", falling back to GET";
+ method = HttpMethod::GET;
+ }
+ }
+ _client->set_method(method);
+
+ // Set payload if configured (supports POST, PUT, DELETE, etc.)
+ auto payload_iter = _extend_kv.find("http.payload");
+ if (payload_iter != _extend_kv.end() && !payload_iter->second.empty()) {
+ _client->set_payload(payload_iter->second);
+ _client->set_content_type("application/json");
+ VLOG(2) << "HTTP request with payload, size=" << payload_iter->second.size();
Review Comment:
**[VLOG may expose credentials]**
The `http.payload` contains the full `FetchRecordRequest` JSON, which includes the `config` map with `user`, `password`, `jdbc_url`, `driver_url`, etc. If someone changes this to log the actual payload content, credentials would be exposed.
Consider either:
1. Not including credentials in the HTTP payload (pass them separately)
2. Adding a note/TODO that the payload contains sensitive data
3. Masking sensitive fields before serialization
##########
fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/utils/ConfigUtil.java:
##########
@@ -43,8 +43,8 @@ public class ConfigUtil {
private static ObjectMapper objectMapper = new ObjectMapper();
private static final Logger LOG = LoggerFactory.getLogger(ConfigUtil.class);
- public static String getServerId(long jobId) {
- return String.valueOf(Math.abs(String.valueOf(jobId).hashCode()));
+ public static String getServerId(String jobId) {
Review Comment:
**[`Math.abs(hashCode())` can return negative for `Integer.MIN_VALUE`]**
`String.hashCode()` can return `Integer.MIN_VALUE`, and `Math.abs(Integer.MIN_VALUE)` returns `Integer.MIN_VALUE` (a negative number). This could produce a negative server ID.
Use `Math.abs((long) jobId.hashCode())` or `(jobId.hashCode() & 0x7FFFFFFF)` instead.
##########
fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/sink/DorisBatchStreamLoad.java:
##########
@@ -511,7 +511,7 @@ public void commitOffset(
CommitOffsetRequest commitRequest =
CommitOffsetRequest.builder()
.offset(OBJECT_MAPPER.writeValueAsString(meta))
- .jobId(jobId)
+ .jobId(Long.parseLong(jobId))
.taskId(Long.parseLong(taskId))
Review Comment:
**[`Long.parseLong(jobId)` will fail for UUID-based job IDs]**
`commitOffset()` calls `Long.parseLong(jobId)`. For TVF calls, the jobId is a UUID string (e.g., `UUID.randomUUID().toString().replace("-", "")`), so this would throw `NumberFormatException`.
The TVF path likely doesn't reach `commitOffset` (only the streaming job path does, and streaming jobs use numeric IDs), but the `isLong()` guard in `PipelineCoordinator.buildStreamRecords()` is the only protection. This is fragile — if the code path changes, it will break.
Consider adding a comment here noting this constraint, or restructuring so that `commitOffset` doesn't need to parse the jobId.
##########
fe/fe-core/src/main/java/org/apache/doris/tablefunction/CdcStreamTableValuedFunction.java:
##########
@@ -0,0 +1,130 @@
+// 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.tablefunction;
+
+import org.apache.doris.analysis.BrokerDesc;
+import org.apache.doris.analysis.StorageBackend.StorageType;
+import org.apache.doris.catalog.Column;
+import org.apache.doris.common.AnalysisException;
+import org.apache.doris.datasource.jdbc.client.JdbcClient;
+import org.apache.doris.job.cdc.DataSourceConfigKeys;
+import org.apache.doris.job.cdc.request.FetchRecordRequest;
+import org.apache.doris.job.common.DataSourceType;
+import org.apache.doris.job.util.StreamingJobUtils;
+import org.apache.doris.thrift.TBrokerFileStatus;
+import org.apache.doris.thrift.TFileType;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.base.Preconditions;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+
+public class CdcStreamTableValuedFunction extends ExternalFileTableValuedFunction {
+ private static final ObjectMapper objectMapper = new ObjectMapper();
+ private static final String URI = "http://127.0.0.1:CDC_CLIENT_PORT/api/fetchRecordStream";
+ private final Map<String, String> originProps;
+
+ public CdcStreamTableValuedFunction(Map<String, String> properties) throws AnalysisException {
+ this.originProps = properties;
+ validate(properties);
+ processProps(properties);
+ }
+
+ private void processProps(Map<String, String> properties) throws AnalysisException {
+ Map<String, String> copyProps = new HashMap<>(properties);
+ copyProps.put("format", "json");
+ super.parseCommonProperties(copyProps);
+ this.processedParams.put("enable_cdc_client", "true");
+ this.processedParams.put("uri", URI);
+ this.processedParams.put("http.enable.range.request", "false");
+ this.processedParams.put("http.enable.chunk.response", "true");
+ this.processedParams.put("http.method", "POST");
+
+ String payload = generateParams(properties);
+ this.processedParams.put("http.payload", payload);
+ this.backendConnectProperties.putAll(processedParams);
+ generateFileStatus();
+ }
+
+ private String generateParams(Map<String, String> properties) throws AnalysisException {
+ FetchRecordRequest recordRequest = new FetchRecordRequest();
+ recordRequest.setJobId(UUID.randomUUID().toString().replace("-", ""));
+ recordRequest.setDataSource(properties.get(DataSourceConfigKeys.TYPE));
+ recordRequest.setConfig(properties);
+ try {
+ return objectMapper.writeValueAsString(recordRequest);
+ } catch (IOException e) {
+ LOG.info("Failed to serialize fetch record request," + e.getMessage());
+ throw new AnalysisException(e.getMessage());
+ }
Review Comment:
**[`LOG.info` for serialization failure should be `LOG.warn` or `LOG.error`]**
Failing to serialize the fetch record request is an error condition, not informational. This should use `LOG.warn` or `LOG.error`.
##########
regression-test/suites/job_p0/streaming_job/cdc/tvf/test_cdc_stream_tvf_mysql.groovy:
##########
@@ -0,0 +1,82 @@
+// 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.
+
+suite("test_cdc_stream_tvf_mysql", "p0,external,mysql,external_docker,external_docker_mysql,nondatalake") {
+ def currentDb = (sql "select database()")[0][0]
+ def table1 = "user_info_cdc_stream_tvf"
+ def mysqlDb = "test_cdc_db"
+
+ String enabled = context.config.otherConfigs.get("enableJdbcTest")
+ if (enabled != null && enabled.equalsIgnoreCase("true")) {
+ String mysql_port = context.config.otherConfigs.get("mysql_57_port");
+ String externalEnvIp = context.config.otherConfigs.get("externalEnvIp")
+ String s3_endpoint = getS3Endpoint()
+ String bucket = getS3BucketName()
+ String driver_url = "https://${bucket}.${s3_endpoint}/regression/jdbc_driver/mysql-connector-j-8.4.0.jar"
+ def offset = ""
+
+ // create test
+ connect("root", "123456", "jdbc:mysql://${externalEnvIp}:${mysql_port}") {
+ sql """CREATE DATABASE IF NOT EXISTS ${mysqlDb}"""
+ sql """DROP TABLE IF EXISTS ${mysqlDb}.${table1}"""
+ sql """CREATE TABLE ${mysqlDb}.${table1} (
+ `name` varchar(200) NOT NULL,
+ `age` int DEFAULT NULL,
+ PRIMARY KEY (`name`)
+ ) ENGINE=InnoDB"""
+ sql """INSERT INTO ${mysqlDb}.${table1} (name, age) VALUES ('A1', 1);"""
+ sql """INSERT INTO ${mysqlDb}.${table1} (name, age) VALUES ('B1', 2);"""
+
+ def result = sql_return_maparray "show master status"
+ def file = result[0]["File"]
+ def position = result[0]["Position"]
+ offset = """{"file":"${file}","pos":"${position}"}"""
+ sql """INSERT INTO ${mysqlDb}.${table1} (name, age) VALUES ('C1', 3);"""
+ sql """INSERT INTO ${mysqlDb}.${table1} (name, age) VALUES ('D1', 4);"""
+ }
+
+ log.info("offset: " + offset)
+ qt_select_tvf """select * from cdc_stream(
Review Comment:
**[Missing `order_qt_` prefix or explicit `ORDER BY`]**
Per testing standards, `qt_select_tvf` should use `order_qt_select_tvf` prefix or the SQL should have an explicit `ORDER BY` clause to ensure deterministic output. CDC data ordering may not be guaranteed across different environments.
Change to:
```groovy
order_qt_select_tvf """select * from cdc_stream(...) order by name"""
```
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [Feature](tvf) Support cdc stream tvf for mysql and pg [doris]
Posted by "JNSimba (via GitHub)" <gi...@apache.org>.
JNSimba commented on PR #60116:
URL: https://github.com/apache/doris/pull/60116#issuecomment-4116993992
run buildall
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [Feature](tvf) Support cdc stream tvf for mysql and pg [doris]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on code in PR #60116:
URL: https://github.com/apache/doris/pull/60116#discussion_r2980531230
##########
fe/fe-core/src/main/java/org/apache/doris/tablefunction/CdcStreamTableValuedFunction.java:
##########
@@ -0,0 +1,140 @@
+// 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.tablefunction;
+
+import org.apache.doris.analysis.BrokerDesc;
+import org.apache.doris.analysis.StorageBackend.StorageType;
+import org.apache.doris.catalog.Column;
+import org.apache.doris.common.AnalysisException;
+import org.apache.doris.datasource.jdbc.client.JdbcClient;
+import org.apache.doris.job.cdc.DataSourceConfigKeys;
+import org.apache.doris.job.cdc.request.FetchRecordRequest;
+import org.apache.doris.job.common.DataSourceType;
+import org.apache.doris.job.util.StreamingJobUtils;
+import org.apache.doris.thrift.TBrokerFileStatus;
+import org.apache.doris.thrift.TFileType;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+
+public class CdcStreamTableValuedFunction extends ExternalFileTableValuedFunction {
+ private static final ObjectMapper objectMapper = new ObjectMapper();
+ private static final String URI = "http://127.0.0.1:CDC_CLIENT_PORT/api/fetchRecordStream";
+
+ public CdcStreamTableValuedFunction(Map<String, String> properties) throws AnalysisException {
+ validate(properties);
+ processProps(properties);
+ }
+
+ private void processProps(Map<String, String> properties) throws AnalysisException {
+ Map<String, String> copyProps = new HashMap<>(properties);
+ copyProps.put("format", "json");
+ super.parseCommonProperties(copyProps);
+ this.processedParams.put("enable_cdc_client", "true");
+ this.processedParams.put("uri", URI);
+ this.processedParams.put("http.enable.range.request", "false");
+ this.processedParams.put("http.enable.chunk.response", "true");
+ this.processedParams.put("http.method", "POST");
+
+ String payload = generateParams(properties);
+ this.processedParams.put("http.payload", payload);
+ this.backendConnectProperties.putAll(processedParams);
+ generateFileStatus();
+ }
+
+ private String generateParams(Map<String, String> properties) throws AnalysisException {
+ FetchRecordRequest recordRequest = new FetchRecordRequest();
+ recordRequest.setJobId(UUID.randomUUID().toString().replace("-", ""));
+ recordRequest.setDataSource(properties.get(DataSourceConfigKeys.TYPE));
+ recordRequest.setConfig(properties);
+ try {
+ return objectMapper.writeValueAsString(recordRequest);
+ } catch (IOException e) {
+ LOG.info("Failed to serialize fetch record request," + e.getMessage());
+ throw new AnalysisException(e.getMessage());
+ }
+ }
+
+ private void validate(Map<String, String> properties) throws AnalysisException {
+ if (!properties.containsKey(DataSourceConfigKeys.JDBC_URL)) {
Review Comment:
**[Low]** `validate()` only checks for `jdbc_url`, `type`, `table`, and `offset`, but `getTableColumns()` and the CDC client also require `user`, `password`, `driver_url`, and `driver_class`. If these are missing, the user will get a cryptic JDBC connection error instead of a clear validation message.
Consider adding checks for these required properties, or at minimum `user` and `password`.
##########
fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/source/reader/postgres/PostgresSourceReader.java:
##########
@@ -257,7 +251,7 @@ private PostgresSourceConfig generatePostgresConfig(
return configFactory.create(subtaskId);
}
- private String getSlotName(Long jobId) {
+ private String getSlotName(String jobId) {
return "doris_cdc_" + jobId;
Review Comment:
**[Low]** `getSlotName()` now concatenates `"doris_cdc_" + jobId` where `jobId` can be a 32-character UUID (from TVF). This produces a 42-character slot name. PostgreSQL replication slot names have a max length of 63 characters (per `NAMEDATALEN`), so this is within bounds, but barely. Worth documenting the constraint here.
Also note that UUID-based slot names are cleaned up at the end of `buildStreamRecords()` via `sourceReader.close(fetchRecord)`, which should drop the slot. If the close fails or the process crashes, orphan slots will remain in PostgreSQL.
##########
fe/fe-core/src/main/java/org/apache/doris/tablefunction/CdcStreamTableValuedFunction.java:
##########
@@ -0,0 +1,140 @@
+// 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.tablefunction;
+
+import org.apache.doris.analysis.BrokerDesc;
+import org.apache.doris.analysis.StorageBackend.StorageType;
+import org.apache.doris.catalog.Column;
+import org.apache.doris.common.AnalysisException;
+import org.apache.doris.datasource.jdbc.client.JdbcClient;
+import org.apache.doris.job.cdc.DataSourceConfigKeys;
+import org.apache.doris.job.cdc.request.FetchRecordRequest;
+import org.apache.doris.job.common.DataSourceType;
+import org.apache.doris.job.util.StreamingJobUtils;
+import org.apache.doris.thrift.TBrokerFileStatus;
+import org.apache.doris.thrift.TFileType;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+
+public class CdcStreamTableValuedFunction extends ExternalFileTableValuedFunction {
+ private static final ObjectMapper objectMapper = new ObjectMapper();
+ private static final String URI = "http://127.0.0.1:CDC_CLIENT_PORT/api/fetchRecordStream";
+
+ public CdcStreamTableValuedFunction(Map<String, String> properties) throws AnalysisException {
+ validate(properties);
+ processProps(properties);
+ }
+
+ private void processProps(Map<String, String> properties) throws AnalysisException {
+ Map<String, String> copyProps = new HashMap<>(properties);
+ copyProps.put("format", "json");
+ super.parseCommonProperties(copyProps);
+ this.processedParams.put("enable_cdc_client", "true");
+ this.processedParams.put("uri", URI);
+ this.processedParams.put("http.enable.range.request", "false");
+ this.processedParams.put("http.enable.chunk.response", "true");
+ this.processedParams.put("http.method", "POST");
+
+ String payload = generateParams(properties);
+ this.processedParams.put("http.payload", payload);
+ this.backendConnectProperties.putAll(processedParams);
+ generateFileStatus();
+ }
+
+ private String generateParams(Map<String, String> properties) throws AnalysisException {
+ FetchRecordRequest recordRequest = new FetchRecordRequest();
+ recordRequest.setJobId(UUID.randomUUID().toString().replace("-", ""));
+ recordRequest.setDataSource(properties.get(DataSourceConfigKeys.TYPE));
+ recordRequest.setConfig(properties);
+ try {
+ return objectMapper.writeValueAsString(recordRequest);
+ } catch (IOException e) {
+ LOG.info("Failed to serialize fetch record request," + e.getMessage());
+ throw new AnalysisException(e.getMessage());
Review Comment:
**[Medium]** `LOG.info` should be `LOG.warn` or `LOG.error` for a failure path. Using `info` level for an error that triggers an `AnalysisException` is misleading — it will be filtered out in production log analysis:
```java
LOG.info("Failed to serialize fetch record request," + e.getMessage());
```
Should be:
```java
LOG.warn("Failed to serialize fetch record request: {}", e.getMessage(), e);
```
Also note the missing space after the comma in the message, and the use of string concatenation instead of parameterized logging.
##########
fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/service/PipelineCoordinator.java:
##########
@@ -51,26 +58,31 @@
import static org.apache.flink.cdc.connectors.base.utils.SourceRecordUtils.SCHEMA_HEARTBEAT_EVENT_KEY_NAME;
import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.common.base.Preconditions;
import io.debezium.data.Envelope;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;
+import org.springframework.web.servlet.mvc.method.annotation.StreamingResponseBody;
/** Pipeline coordinator. */
@Component
public class PipelineCoordinator {
private static final Logger LOG = LoggerFactory.getLogger(PipelineCoordinator.class);
private static final String SPLIT_ID = "splitId";
// jobId
- private final Map<Long, DorisBatchStreamLoad> batchStreamLoadMap = new ConcurrentHashMap<>();
+ private final Map<String, DorisBatchStreamLoad> batchStreamLoadMap = new ConcurrentHashMap<>();
+ // taskId, offset
+ private final Map<String, Map<String, String>> taskOffsetCache = new ConcurrentHashMap<>();
// taskId -> writeFailReason
Review Comment:
**[Low]** `taskOffsetCache` is a `ConcurrentHashMap` that is populated at the end of `buildStreamRecords` but only cleaned up by `getOffsetWithTaskId()` (via `remove()`). If the caller never retrieves the offset (e.g., due to an error in the query pipeline), entries will accumulate indefinitely. Consider adding a TTL-based eviction or a periodic cleanup mechanism to prevent unbounded memory growth in long-running CDC client processes.
##########
fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/sink/DorisBatchStreamLoad.java:
##########
@@ -511,7 +511,7 @@ public void commitOffset(
CommitOffsetRequest commitRequest =
CommitOffsetRequest.builder()
.offset(OBJECT_MAPPER.writeValueAsString(meta))
- .jobId(jobId)
+ .jobId(Long.parseLong(jobId))
.taskId(Long.parseLong(taskId))
Review Comment:
**[Low]** `Long.parseLong(jobId)` will throw `NumberFormatException` if `jobId` is a UUID string (as generated by the TVF path in `CdcStreamTableValuedFunction.generateParams()`). While the TVF streaming path does not go through `DorisBatchStreamLoad.commitOffset()`, this conversion is fragile — if the code paths ever converge, this will fail silently. Consider adding a guard or documenting the assumption clearly.
##########
fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/service/PipelineCoordinator.java:
##########
@@ -90,6 +102,159 @@ public PipelineCoordinator() {
new ThreadPoolExecutor.AbortPolicy());
}
+ /** return data for http_file_reader */
+ public StreamingResponseBody fetchRecordStream(FetchRecordRequest fetchReq) throws Exception {
+ SourceReader sourceReader;
+ SplitReadResult readResult;
+ try {
+ if (fetchReq.getTaskId() == null && fetchReq.getMeta() == null) {
+ LOG.info(
+ "Generate initial meta for fetch record request, jobId={}, taskId={}",
+ fetchReq.getJobId(),
+ fetchReq.getTaskId());
+ // means the request did not originate from the job, only tvf
+ Map<String, Object> meta = generateMeta(fetchReq.getConfig());
+ fetchReq.setMeta(meta);
+ }
+
+ sourceReader = Env.getCurrentEnv().getReader(fetchReq);
+ readResult = sourceReader.prepareAndSubmitSplit(fetchReq);
+ } catch (Exception ex) {
+ throw new CommonException(ex);
+ }
+
+ return outputStream -> {
+ try {
+ buildStreamRecords(sourceReader, fetchReq, readResult, outputStream);
+ } catch (Exception ex) {
+ LOG.error(
+ "Failed fetch record, jobId={}, taskId={}",
+ fetchReq.getJobId(),
+ fetchReq.getTaskId(),
+ ex);
+ throw new StreamException(ex);
+ }
+ };
+ }
+
+ private void buildStreamRecords(
+ SourceReader sourceReader,
+ FetchRecordRequest fetchRecord,
+ SplitReadResult readResult,
+ OutputStream rawOutputStream)
+ throws Exception {
+ SourceSplit split = readResult.getSplit();
+ Map<String, String> lastMeta = null;
+ int rowCount = 0;
+ BufferedOutputStream bos = new BufferedOutputStream(rawOutputStream);
+ try {
+ // Poll records using the existing mechanism
+ boolean shouldStop = false;
+ long startTime = System.currentTimeMillis();
+ while (!shouldStop) {
+ Iterator<SourceRecord> recordIterator = sourceReader.pollRecords();
+ if (!recordIterator.hasNext()) {
+ Thread.sleep(100);
+ long elapsedTime = System.currentTimeMillis() - startTime;
Review Comment:
**[Nit]** The `Thread.sleep(100)` busy-wait loop with timeout is acceptable for the bounded CDC streaming use case, but consider using `Thread.sleep` with an exponential backoff (e.g., 100ms → 200ms → 400ms up to a cap) to reduce CPU overhead when the source is slow. The timeout (`Constants.POLL_SPLIT_RECORDS_TIMEOUTS`) already bounds the overall wait.
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [Feature](tvf) Support cdc stream tvf for mysql and pg [doris]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #60116:
URL: https://github.com/apache/doris/pull/60116#issuecomment-4141250310
PR approved by at least one committer and no changes requested.
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [Feature](tvf) Support cdc stream tvf for mysql and pg [doris]
Posted by "hello-stephen (via GitHub)" <gi...@apache.org>.
hello-stephen commented on PR #60116:
URL: https://github.com/apache/doris/pull/60116#issuecomment-4127125309
# BE Regression && UT Coverage Report
Increment line coverage `100% (0/0)` :tada:
[Increment coverage report](http://coverage.selectdb-in.cc/coverage/60116_9ea6fe06a5cddc20a17b3f4565082396852f0a56_merge/increment_report/index.html)
[Complete coverage report](http://coverage.selectdb-in.cc/coverage/60116_9ea6fe06a5cddc20a17b3f4565082396852f0a56_merge/report/index.html)
| Category | Coverage |
|-------------------|--------------------|
| Function Coverage | 63.07% (23233/36839) |
| Line Coverage | 46.50% (236962/509647) |
| Region Coverage | 43.56% (194198/445796) |
| Branch Coverage | 44.87% (83913/187006) |
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [Feature](tvf) Support cdc stream tvf for mysql and pg [doris]
Posted by "JNSimba (via GitHub)" <gi...@apache.org>.
JNSimba commented on PR #60116:
URL: https://github.com/apache/doris/pull/60116#issuecomment-4125581449
run p0
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [Feature](tvf) Support cdc stream tvf for mysql and pg [doris]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #60116:
URL: https://github.com/apache/doris/pull/60116#discussion_r2711847911
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/table/CdcStream.java:
##########
@@ -0,0 +1,59 @@
+// 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.nereids.trees.expressions.functions.table;
+
+import org.apache.doris.catalog.FunctionSignature;
+import org.apache.doris.nereids.exceptions.AnalysisException;
+import org.apache.doris.nereids.trees.expressions.Properties;
+import org.apache.doris.nereids.trees.expressions.visitor.ExpressionVisitor;
+import org.apache.doris.nereids.types.coercion.AnyDataType;
+import org.apache.doris.tablefunction.CdcStreamTableValuedFunction;
+import org.apache.doris.tablefunction.TableValuedFunctionIf;
+
+import java.util.Map;
+
+/**
+ * CdcStream TVF.
+ */
+public class CdcStream extends TableValuedFunction {
+
+ public CdcStream(Properties tvfProperties) {
+ super("cdc_stream", tvfProperties);
+ }
+
+ @Override
+ public FunctionSignature customSignature() {
+ return FunctionSignature.of(AnyDataType.INSTANCE_WITHOUT_INDEX, getArgumentsTypes());
+ }
+
+ @Override
+ protected TableValuedFunctionIf toCatalogFunction() {
+ try {
+ Map<String, String> arguments = getTVFProperties().getMap();
+ return new CdcStreamTableValuedFunction(arguments);
+ } catch (Throwable t) {
+ throw new AnalysisException("Can not build CdcStreamTableValuedFunction by "
+ + this + ": " + t.getMessage(), t);
+ }
+ }
+
+ @Override
+ public <R, C> R accept(ExpressionVisitor<R, C> visitor, C context) {
+ return visitor.visitPostgresCdc(this, context);
Review Comment:
The method 'visitPostgresCdc' is misleadingly named as it handles both MySQL and PostgreSQL CDC streams, not just PostgreSQL. This naming inconsistency can confuse developers. The method should be renamed to 'visitCdcStream' to match the actual functionality.
```suggestion
return visitor.visitCdcStream(this, context);
```
##########
fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/service/PipelineCoordinator.java:
##########
@@ -81,19 +95,154 @@ public PipelineCoordinator() {
new ThreadPoolExecutor.AbortPolicy());
}
+ /** return data for http_file_reader */
+ public StreamingResponseBody fetchRecordStream(FetchRecordRequest fetchReq) throws Exception {
+ SourceReader sourceReader;
+ SplitReadResult readResult;
+ try {
+ if (fetchReq.getTaskId() == null && fetchReq.getMeta() == null) {
+ LOG.info(
+ "Generate initial meta for fetch record request, jobId={}, taskId={}",
+ fetchReq.getJobId(),
+ fetchReq.getTaskId());
+ // means the request did not originate from the job, only tvf
+ Map<String, Object> meta = generateMeta(fetchReq.getConfig());
+ fetchReq.setMeta(meta);
+ }
+
+ sourceReader = Env.getCurrentEnv().getReader(fetchReq);
+ readResult = sourceReader.readSplitRecords(fetchReq);
+ } catch (Exception ex) {
+ throw new CommonException(ex);
+ }
+
+ return outputStream -> {
+ try {
+ buildRecords(sourceReader, fetchReq, readResult, outputStream);
+ } catch (Exception ex) {
+ LOG.error(
+ "Failed fetch record, jobId={}, taskId={}",
+ fetchReq.getJobId(),
+ fetchReq.getTaskId(),
+ ex);
+ throw new StreamException(ex);
+ }
+ };
+ }
+
+ private void buildRecords(
+ SourceReader sourceReader,
+ FetchRecordRequest fetchRecord,
+ SplitReadResult readResult,
+ OutputStream rawOutputStream)
+ throws Exception {
+ SourceSplit split = readResult.getSplit();
+ Map<String, String> lastMeta = null;
+ int rowCount = 0;
+ BufferedOutputStream bos = new BufferedOutputStream(rawOutputStream);
+ try {
+ // Serialize records and add them to the response (collect from iterator)
+ Iterator<SourceRecord> iterator = readResult.getRecordIterator();
+ while (iterator != null && iterator.hasNext()) {
+ SourceRecord element = iterator.next();
+ List<String> serializedRecords =
+ sourceReader.deserialize(fetchRecord.getConfig(), element);
+ for (String record : serializedRecords) {
+ bos.write(record.getBytes(StandardCharsets.UTF_8));
+ bos.write(LINE_DELIMITER);
+ }
+ rowCount += serializedRecords.size();
+ }
+ // force flush buffer
+ bos.flush();
+ } finally {
+ // The LSN in the commit is the current offset, which is the offset from the last
+ // successful write.
+ // Therefore, even if a subsequent write fails, it will not affect the commit.
+ sourceReader.commitSourceOffset(fetchRecord.getJobId(), readResult.getSplit());
+
+ // This must be called after commitSourceOffset; otherwise,
+ // PG's confirmed lsn will not proceed.
+ sourceReader.finishSplitRecords();
+ }
+
+ LOG.info(
+ "Fetch records completed, jobId={}, taskId={}, splitId={}, rowCount={}",
+ fetchRecord.getJobId(),
+ fetchRecord.getTaskId(),
+ split.splitId(),
+ rowCount);
+
+ if (readResult.getSplitState() != null) {
+ // Set meta information for hw
+ if (sourceReader.isSnapshotSplit(split)) {
+ lastMeta = sourceReader.extractSnapshotStateOffset(readResult.getSplitState());
+ lastMeta.put(SPLIT_ID, split.splitId());
+ }
+
+ // set meta for binlog event
+ if (sourceReader.isBinlogSplit(split)) {
+ lastMeta = sourceReader.extractBinlogStateOffset(readResult.getSplitState());
+ lastMeta.put(SPLIT_ID, BinlogSplit.BINLOG_SPLIT_ID);
+ }
+ } else {
+ throw new RuntimeException("split state is null");
+ }
+
+ if (StringUtils.isNotEmpty(fetchRecord.getTaskId())) {
+ taskOffsetCache.put(fetchRecord.getTaskId(), lastMeta);
+ }
+
+ sourceReader.commitSourceOffset(fetchRecord.getJobId(), readResult.getSplit());
Review Comment:
The 'commitSourceOffset' method is called twice - once at line 162 within the try-finally block, and again at line 196 after the finally block. This appears to be a bug as committing the same offset twice could lead to incorrect offset tracking or unexpected behavior. Remove one of the duplicate calls.
```suggestion
```
##########
fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/utils/ConfigUtil.java:
##########
@@ -96,6 +98,21 @@ public static ZoneId getPostgresServerTimeZoneFromProps(java.util.Properties pro
return ZoneId.systemDefault();
}
+ public static String[] getTableList(String schema, Map<String, String> cdcConfig) {
+ String includingTables = cdcConfig.get(DataSourceConfigKeys.INCLUDE_TABLES);
+ String table = cdcConfig.get(DataSourceConfigKeys.TABLE);
+ if (StringUtils.isNotEmpty(includingTables)) {
+ return Arrays.stream(includingTables.split(","))
+ .map(t -> schema + "." + t.trim())
+ .toArray(String[]::new);
+ } else if (StringUtils.isNotEmpty(table)) {
+ Preconditions.checkArgument(!table.contains(","), "table only support one table");
Review Comment:
The error message "table only support one table" contains a grammatical error. It should be "table only supports one table" (with an 's' in 'supports').
```suggestion
Preconditions.checkArgument(!table.contains(","), "table only supports one table");
```
##########
be/src/io/fs/http_file_reader.cpp:
##########
@@ -269,6 +335,21 @@ Status HttpFileReader::read_at_impl(size_t offset, Slice result, size_t* bytes_r
long http_status = _client->get_http_status();
VLOG(2) << "HTTP response: status=" << http_status << " received_bytes=" << buf.size();
+ // Check for HTTP error status codes (4xx, 5xx)
+ if (http_status >= 400) {
+ std::string error_body;
+ if (buf.empty()) {
+ error_body = "(empty response body)";
+ } else {
+ // Limit error message to 1024 bytes to avoid excessive logging
+ size_t max_len = std::min(buf.size(), static_cast<size_t>(1024));
+ error_body = buf.substr(0, max_len);
+ }
+
+ return Status::InternalError("HTTP request failed with status {}: {}.", http_status,
+ error_body);
Review Comment:
Missing space after comma in the error message format string. The format should be "HTTP request failed with status {}: {}." with proper spacing.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/visitor/TableValuedFunctionVisitor.java:
##########
@@ -81,6 +82,10 @@ default R visitHttp(Http http, C context) {
return visitTableValuedFunction(http, context);
}
+ default R visitPostgresCdc(CdcStream cdcStream, C context) {
+ return visitTableValuedFunction(cdcStream, context);
+ }
+
Review Comment:
The method name 'visitPostgresCdc' is misleading as this TVF supports both MySQL and PostgreSQL CDC streaming, not just PostgreSQL. Consider renaming to 'visitCdcStream' to accurately reflect its purpose.
```suggestion
default R visitCdcStream(CdcStream cdcStream, C context) {
return visitTableValuedFunction(cdcStream, context);
}
/**
* @deprecated This method name is misleading because {@link CdcStream} supports multiple CDC sources.
* Use {@link #visitCdcStream(CdcStream, Object)} instead.
*/
@Deprecated
default R visitPostgresCdc(CdcStream cdcStream, C context) {
return visitCdcStream(cdcStream, context);
}
```
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [Feature](tvf) Support cdc stream tvf for mysql and pg [doris]
Posted by "doris-robot (via GitHub)" <gi...@apache.org>.
doris-robot commented on PR #60116:
URL: https://github.com/apache/doris/pull/60116#issuecomment-3777613896
<details>
<summary>ClickBench: <b>Total hot run time: 26.75 s</b></summary>
```
machine: 'aliyun_ecs.c7a.8xlarge_32C64G'
scripts: https://github.com/apache/doris/tree/master/tools/clickbench-tools
ClickBench test result on commit d84fcddbe4aa9142e912da69ab6bafe20075ca2e, data reload: false
query1 0.05 0.04 0.05
query2 0.09 0.05 0.04
query3 0.26 0.09 0.08
query4 1.61 0.12 0.11
query5 0.28 0.27 0.25
query6 1.14 0.67 0.65
query7 0.02 0.02 0.03
query8 0.05 0.04 0.04
query9 0.57 0.50 0.49
query10 0.55 0.54 0.53
query11 0.14 0.10 0.10
query12 0.14 0.11 0.11
query13 0.60 0.58 0.58
query14 0.95 0.93 0.93
query15 0.78 0.77 0.78
query16 0.38 0.41 0.41
query17 1.08 1.04 1.06
query18 0.22 0.21 0.22
query19 1.95 1.89 1.84
query20 0.02 0.01 0.01
query21 15.46 0.25 0.14
query22 5.32 0.05 0.04
query23 16.11 0.28 0.10
query24 1.62 0.24 0.52
query25 0.09 0.07 0.08
query26 0.14 0.14 0.13
query27 0.06 0.08 0.06
query28 3.74 1.07 0.88
query29 12.53 3.89 3.13
query30 0.28 0.13 0.12
query31 2.84 0.63 0.39
query32 3.23 0.55 0.46
query33 3.00 3.05 3.04
query34 15.85 5.07 4.43
query35 4.47 4.46 4.44
query36 0.66 0.50 0.49
query37 0.10 0.07 0.06
query38 0.07 0.03 0.03
query39 0.04 0.03 0.03
query40 0.17 0.13 0.13
query41 0.09 0.04 0.03
query42 0.05 0.03 0.03
query43 0.05 0.03 0.03
Total cold run time: 96.85 s
Total hot run time: 26.75 s
```
</details>
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [Feature](tvf) Support cdc stream tvf for mysql and pg [doris]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #60116:
URL: https://github.com/apache/doris/pull/60116#discussion_r2715144714
##########
regression-test/suites/job_p0/streaming_job/cdc/tvf/test_cdc_stream_tvf_postgres.groovy:
##########
@@ -0,0 +1,81 @@
+// 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.
+
+suite("test_cdc_stream_tvf_postgres", "p0,external,pg,external_docker,external_docker_pg,nondatalake") {
+ def currentDb = (sql "select database()")[0][0]
+ def table1 = "user_info_pg_normal1"
+ def pgDB = "postgres"
+ def pgSchema = "cdc_test"
+ def pgUser = "postgres"
+ def pgPassword = "123456"
+
+ String enabled = context.config.otherConfigs.get("enableJdbcTest")
+ if (enabled != null && enabled.equalsIgnoreCase("true")) {
+ String pg_port = context.config.otherConfigs.get("pg_14_port");
+ String externalEnvIp = context.config.otherConfigs.get("externalEnvIp")
+ String s3_endpoint = getS3Endpoint()
+ String bucket = getS3BucketName()
+ String driver_url = "https://${bucket}.${s3_endpoint}/regression/jdbc_driver/postgresql-42.5.0.jar"
+
+ // create test
+ connect("${pgUser}", "${pgPassword}", "jdbc:postgresql://${externalEnvIp}:${pg_port}/${pgDB}") {
+ // sql """CREATE SCHEMA IF NOT EXISTS ${pgSchema}"""
+ sql """DROP TABLE IF EXISTS ${pgDB}.${pgSchema}.${table1}"""
+ sql """CREATE TABLE ${pgDB}.${pgSchema}.${table1} (
+ "name" varchar(200),
+ "age" int2,
+ PRIMARY KEY ("name")
+ )"""
+ sql """INSERT INTO ${pgDB}.${pgSchema}.${table1} (name, age) VALUES ('A1', 1);"""
+ sql """INSERT INTO ${pgDB}.${pgSchema}.${table1} (name, age) VALUES ('B1', 2);"""
+ }
+
+ test {
+ sql """
+ select * from cdc_stream(
+ "type" = "postgres",
+ "jdbc_url" = "jdbc:postgresql://${externalEnvIp}:${pg_port}/${pgDB}",
+ "driver_url" = "${driver_url}",
+ "driver_class" = "org.postgresql.Driver",
+ "user" = "${pgUser}",
+ "password" = "${pgPassword}",
+ "database" = "${pgDB}",
+ "schema" = "${pgSchema}",
+ "table" = "${table1}",
+ "offset" = 'initial')
+ """
+ exception "Unsupported offset: initial"
+ }
+
+ // Here, because PG consumption requires creating a slot first,
+ // we only verify whether the execution can be successful.
+ def result = sql """
+ select * from cdc_stream(
+ "type" = "postgres",
+ "jdbc_url" = "jdbc:postgresql://${externalEnvIp}:${pg_port}/${pgDB}",
+ "driver_url" = "${driver_url}",
+ "driver_class" = "org.postgresql.Driver",
+ "user" = "${pgUser}",
+ "password" = "${pgPassword}",
+ "database" = "${pgDB}",
+ "schema" = "${pgSchema}",
+ "table" = "${table1}",
+ "offset" = 'latest')
+ """
+ log.info("result:", result)
Review Comment:
The test doesn't verify the actual data returned by the query. It only checks that the query executes successfully. Consider adding assertions to verify that the correct CDC data is returned, matching the expected output format and content (similar to the MySQL test which has an expected output file).
##########
fe/fe-common/src/main/java/org/apache/doris/job/cdc/request/JobBaseConfig.java:
##########
@@ -27,7 +27,7 @@
@AllArgsConstructor
@NoArgsConstructor
public class JobBaseConfig {
- private Long jobId;
+ private String jobId;
Review Comment:
Changing jobId from Long to String throughout the codebase is a significant API change that affects serialization, deserialization, and client compatibility. This is a breaking change that requires careful migration planning. Existing job IDs stored as Long values in databases or configurations will need conversion logic. Consider providing a migration path or backward compatibility layer.
```suggestion
private Long jobId;
```
##########
fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/source/reader/JdbcIncrementalSourceReader.java:
##########
@@ -154,39 +154,17 @@ public SplitReadResult readSplitRecords(JobBaseRecordRequest baseReq) throws Exc
if (offsetMeta == null || offsetMeta.isEmpty()) {
throw new RuntimeException("miss meta offset");
}
- LOG.info("Job {} read split records with offset: {}", baseReq.getJobId(), offsetMeta);
-
- // If there is an active split being consumed, reuse it directly;
- // Otherwise, create a new snapshot/stream split based on offset and start the reader.
- SourceSplitBase split = null;
- SplitRecords currentSplitRecords = this.getCurrentSplitRecords();
- if (currentSplitRecords == null) {
- Fetcher<SourceRecords, SourceSplitBase> currentReader = this.getCurrentReader();
- if (baseReq.isReload() || currentReader == null) {
- LOG.info(
- "No current reader or reload {}, create new split reader for job {}",
- baseReq.isReload(),
- baseReq.getJobId());
- // build split
- Tuple2<SourceSplitBase, Boolean> splitFlag = createSourceSplit(offsetMeta, baseReq);
- split = splitFlag.f0;
- // closeBinlogReader();
- currentSplitRecords = pollSplitRecordsWithSplit(split, baseReq);
- this.setCurrentSplitRecords(currentSplitRecords);
- this.setCurrentSplit(split);
- } else if (currentReader instanceof IncrementalSourceStreamFetcher) {
- LOG.info("Continue poll records with current binlog reader");
- // only for binlog reader
- currentSplitRecords = pollSplitRecordsWithCurrentReader(currentReader);
- split = this.getCurrentSplit();
- } else {
- throw new RuntimeException("Should not happen");
- }
- } else {
- LOG.info(
- "Continue read records with current split records, splitId: {}",
- currentSplitRecords.getSplitId());
- }
+ // Create a new snapshot/stream split based on offset and start the reader.
+ LOG.info(
+ "Create new split reader for job {} with offset {}",
+ baseReq.getJobId(),
+ offsetMeta);
+ // build split
+ Tuple2<SourceSplitBase, Boolean> splitFlag = createSourceSplit(offsetMeta, baseReq);
+ SourceSplitBase split = splitFlag.f0;
+ // it's necessary to ensure that the binlog reader is already closed.
+ this.currentSplitRecords = pollSplitRecordsWithSplit(split, baseReq);
+ this.currentSplit = split;
Review Comment:
Similar to the MySQL reader, the logic for reusing active splits and continuing with the current reader has been removed. This forces recreation of splits on every call, which could impact performance. The removed logic appeared to optimize for streaming scenarios by reusing the IncrementalSourceStreamFetcher. If this optimization is no longer needed for TVF use cases, consider documenting why.
##########
fe/fe-core/src/main/java/org/apache/doris/tablefunction/CdcStreamTableValuedFunction.java:
##########
@@ -0,0 +1,129 @@
+// 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.tablefunction;
+
+import org.apache.doris.analysis.BrokerDesc;
+import org.apache.doris.analysis.StorageBackend.StorageType;
+import org.apache.doris.catalog.Column;
+import org.apache.doris.common.AnalysisException;
+import org.apache.doris.datasource.jdbc.client.JdbcClient;
+import org.apache.doris.job.cdc.DataSourceConfigKeys;
+import org.apache.doris.job.cdc.request.FetchRecordRequest;
+import org.apache.doris.job.common.DataSourceType;
+import org.apache.doris.job.util.StreamingJobUtils;
+import org.apache.doris.thrift.TBrokerFileStatus;
+import org.apache.doris.thrift.TFileType;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.base.Preconditions;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+
+public class CdcStreamTableValuedFunction extends ExternalFileTableValuedFunction {
+ private static final ObjectMapper objectMapper = new ObjectMapper();
+ private static String URI = "http://127.0.0.1:{}/api/fetchRecordStream";
+ private final Map<String, String> originProps;
+
+ public CdcStreamTableValuedFunction(Map<String, String> properties) throws AnalysisException {
+ this.originProps = properties;
+ processProps(properties);
+ validate(properties);
+ }
+
+ private void processProps(Map<String, String> properties) throws AnalysisException {
+ Map<String, String> copyProps = new HashMap<>(properties);
+ copyProps.put("format", "json");
+ super.parseCommonProperties(copyProps);
+ this.processedParams.put("enable_cdc_client", "true");
+ this.processedParams.put("uri", URI);
+ this.processedParams.put("http.enable.range.request", "false");
+ this.processedParams.put("http.chunk.response", "true");
+ this.processedParams.put("http.method", "POST");
+
+ String payload = generateParams(properties);
+ this.processedParams.put("http.payload", payload);
+ this.backendConnectProperties.putAll(processedParams);
+ generateFileStatus();
+ }
+
+ private String generateParams(Map<String, String> properties) throws AnalysisException {
+ FetchRecordRequest recordRequest = new FetchRecordRequest();
+ recordRequest.setJobId(UUID.randomUUID().toString().replace("-", ""));
+ recordRequest.setDataSource(properties.get(DataSourceConfigKeys.TYPE));
+ recordRequest.setConfig(properties);
+ try {
+ return objectMapper.writeValueAsString(recordRequest);
+ } catch (IOException e) {
+ LOG.info("Failed to serialize fetch record request," + e.getMessage());
+ throw new AnalysisException(e.getMessage());
+ }
+ }
+
+ private void validate(Map<String, String> properties) {
+ Preconditions.checkArgument(properties.containsKey(DataSourceConfigKeys.JDBC_URL), "jdbc_url is required");
+ Preconditions.checkArgument(properties.containsKey(DataSourceConfigKeys.TYPE), "type is required");
+ Preconditions.checkArgument(properties.containsKey(DataSourceConfigKeys.TABLE), "table is required");
Review Comment:
The validation doesn't check for the OFFSET parameter, which is required based on the generateMeta logic in PipelineCoordinator. Without an offset, the TVF will fail at runtime. Add validation to ensure the offset parameter is present.
```suggestion
Preconditions.checkArgument(properties.containsKey(DataSourceConfigKeys.TABLE), "table is required");
Preconditions.checkArgument(properties.containsKey(DataSourceConfigKeys.OFFSET), "offset is required");
```
##########
fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/source/reader/postgres/PostgresSourceReader.java:
##########
@@ -167,29 +166,18 @@ private PostgresSourceConfig generatePostgresConfig(Map<String, String> cdcConfi
configFactory.includeSchemaChanges(false);
// Set table list
- String includingTables = cdcConfig.get(DataSourceConfigKeys.INCLUDE_TABLES);
- if (StringUtils.isNotEmpty(includingTables)) {
- String[] includingTbls =
- Arrays.stream(includingTables.split(","))
- .map(t -> schema + "." + t.trim())
- .toArray(String[]::new);
- configFactory.tableList(includingTbls);
- }
+ String[] tableList = ConfigUtil.getTableList(schema, cdcConfig);
+ Preconditions.checkArgument(tableList.length >= 1, "include_tables or table is required");
+ configFactory.tableList(tableList);
// Set startup options
String startupMode = cdcConfig.get(DataSourceConfigKeys.OFFSET);
if (DataSourceConfigKeys.OFFSET_INITIAL.equalsIgnoreCase(startupMode)) {
configFactory.startupOptions(StartupOptions.initial());
- } else if (DataSourceConfigKeys.OFFSET_EARLIEST.equalsIgnoreCase(startupMode)) {
- configFactory.startupOptions(StartupOptions.earliest());
} else if (DataSourceConfigKeys.OFFSET_LATEST.equalsIgnoreCase(startupMode)) {
configFactory.startupOptions(StartupOptions.latest());
} else if (ConfigUtil.isJson(startupMode)) {
throw new RuntimeException("Unsupported json offset " + startupMode);
- } else if (ConfigUtil.is13Timestamp(startupMode)) {
- // start from timestamp
- Long ts = Long.parseLong(startupMode);
- configFactory.startupOptions(StartupOptions.timestamp(ts));
} else {
throw new RuntimeException("Unknown offset " + startupMode);
Review Comment:
The startup mode validation logic has removed support for "earliest" and timestamp-based offsets that were previously supported. The code now throws "Unsupported json offset" for JSON offsets and "Unknown offset" for timestamp offsets, but the removed branches suggest these were intentional features. If these offset modes are no longer needed, consider documenting why in the commit message or code comments. If they should still be supported, this is a breaking change that removes functionality.
##########
fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/source/reader/mysql/MySqlSourceReader.java:
##########
@@ -568,18 +543,16 @@ private MySqlSourceConfig generateMySqlConfig(Map<String, String> cdcConfig, Str
configFactory.includeSchemaChanges(false);
- String includingTables = cdcConfig.get(DataSourceConfigKeys.INCLUDE_TABLES);
- String[] includingTbls =
- Arrays.stream(includingTables.split(","))
- .map(t -> databaseName + "." + t.trim())
- .toArray(String[]::new);
- configFactory.tableList(includingTbls);
+ // Set table list
+ String[] tableList = ConfigUtil.getTableList(databaseName, cdcConfig);
+ com.google.common.base.Preconditions.checkArgument(
+ tableList.length >= 1, "include_tables or table is required");
+ configFactory.tableList(tableList);
// setting startMode
String startupMode = cdcConfig.get(DataSourceConfigKeys.OFFSET);
if (DataSourceConfigKeys.OFFSET_INITIAL.equalsIgnoreCase(startupMode)) {
- // do not need set offset when initial
- // configFactory.startupOptions(StartupOptions.initial());
+ configFactory.startupOptions(StartupOptions.initial());
Review Comment:
The commented-out line for setting StartupOptions.initial() has been uncommented, but the comment above it "do not need set offset when initial" has been removed. This suggests there may have been a reason not to set this option initially. If this behavior change is intentional and correct, the change is fine; otherwise, this could introduce a bug if the "initial" mode was intentionally not being set.
```suggestion
// Do not explicitly set startupOptions for INITIAL mode; rely on the default
// behavior of MySqlSourceConfigFactory to start from the initial snapshot/binlog.
```
##########
fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/controller/ClientController.java:
##########
@@ -82,6 +83,17 @@ public Object fetchRecords(@RequestBody FetchRecordRequest recordReq) {
}
}
+ @RequestMapping(path = "/api/fetchRecordStream", method = RequestMethod.POST)
+ public StreamingResponseBody fetchRecordStream(@RequestBody FetchRecordRequest recordReq)
+ throws Exception {
+ return pipelineCoordinator.fetchRecordStream(recordReq);
+ }
+
+ @RequestMapping(path = "/api/getTaskOffset/{taskId}", method = RequestMethod.POST)
Review Comment:
The getTaskIdOffset endpoint uses POST method but doesn't require any body in the request - only the taskId from the path. This should be a GET request instead, as it's a read-only operation retrieving data. Using POST for read operations violates RESTful API design principles and could cause caching and semantic issues.
```suggestion
@RequestMapping(path = "/api/getTaskOffset/{taskId}", method = RequestMethod.GET)
```
##########
regression-test/suites/job_p0/streaming_job/cdc/tvf/test_cdc_stream_tvf_mysql.groovy:
##########
@@ -0,0 +1,82 @@
+// 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.
+
+suite("test_cdc_stream_tvf_mysql", "p0,external,mysql,external_docker,external_docker_mysql,nondatalake") {
+ def currentDb = (sql "select database()")[0][0]
+ def table1 = "user_info_cdc_stream_tvf"
+ def mysqlDb = "test_cdc_db"
+
+ String enabled = context.config.otherConfigs.get("enableJdbcTest")
+ if (enabled != null && enabled.equalsIgnoreCase("true")) {
+ String mysql_port = context.config.otherConfigs.get("mysql_57_port");
+ String externalEnvIp = context.config.otherConfigs.get("externalEnvIp")
+ String s3_endpoint = getS3Endpoint()
+ String bucket = getS3BucketName()
+ String driver_url = "https://${bucket}.${s3_endpoint}/regression/jdbc_driver/mysql-connector-j-8.4.0.jar"
+ def offset = ""
+
+ // create test
+ connect("root", "123456", "jdbc:mysql://${externalEnvIp}:${mysql_port}") {
Review Comment:
The test uses hardcoded credentials ("root", "123456") directly in the code. While this is acceptable for test code, consider using configuration variables or test fixtures for better maintainability and to avoid accidental exposure if these tests are run against non-test environments.
##########
fe/fe-core/src/main/java/org/apache/doris/tablefunction/CdcStreamTableValuedFunction.java:
##########
@@ -0,0 +1,129 @@
+// 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.tablefunction;
+
+import org.apache.doris.analysis.BrokerDesc;
+import org.apache.doris.analysis.StorageBackend.StorageType;
+import org.apache.doris.catalog.Column;
+import org.apache.doris.common.AnalysisException;
+import org.apache.doris.datasource.jdbc.client.JdbcClient;
+import org.apache.doris.job.cdc.DataSourceConfigKeys;
+import org.apache.doris.job.cdc.request.FetchRecordRequest;
+import org.apache.doris.job.common.DataSourceType;
+import org.apache.doris.job.util.StreamingJobUtils;
+import org.apache.doris.thrift.TBrokerFileStatus;
+import org.apache.doris.thrift.TFileType;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.base.Preconditions;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+
+public class CdcStreamTableValuedFunction extends ExternalFileTableValuedFunction {
+ private static final ObjectMapper objectMapper = new ObjectMapper();
+ private static String URI = "http://127.0.0.1:{}/api/fetchRecordStream";
Review Comment:
The URI field is defined as non-final but is never reassigned after initialization. It should be declared as a constant: private static final String URI = "http://127.0.0.1:{}/api/fetchRecordStream";
```suggestion
private static final String URI = "http://127.0.0.1:{}/api/fetchRecordStream";
```
##########
fe/fe-common/src/main/java/org/apache/doris/job/cdc/DataSourceConfigKeys.java:
##########
@@ -19,13 +19,15 @@
public class DataSourceConfigKeys {
public static final String JDBC_URL = "jdbc_url";
+ public static final String TYPE = "type";
public static final String DRIVER_URL = "driver_url";
public static final String DRIVER_CLASS = "driver_class";
public static final String USER = "user";
public static final String PASSWORD = "password";
public static final String DATABASE = "database";
public static final String SCHEMA = "schema";
public static final String INCLUDE_TABLES = "include_tables";
Review Comment:
The new constant TABLE is added without documentation. Consider adding a JavaDoc comment explaining the difference between TABLE and INCLUDE_TABLES (e.g., "Single table name for CDC streaming. Use INCLUDE_TABLES for multiple tables.") to clarify usage.
```suggestion
public static final String INCLUDE_TABLES = "include_tables";
/**
* Single table name for CDC streaming. Use INCLUDE_TABLES for multiple tables.
*/
```
##########
fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/source/reader/mysql/MySqlSourceReader.java:
##########
@@ -170,39 +169,14 @@ public SplitReadResult readSplitRecords(JobBaseRecordRequest baseReq) throws Exc
if (offsetMeta == null || offsetMeta.isEmpty()) {
throw new RuntimeException("miss meta offset");
}
- LOG.info("Job {} read split records with offset: {}", baseReq.getJobId(), offsetMeta);
-
- // If there is an active split being consumed, reuse it directly;
- // Otherwise, create a new snapshot/binlog split based on offset and start the reader.
- MySqlSplit split = null;
- SplitRecords currentSplitRecords = this.getCurrentSplitRecords();
- if (currentSplitRecords == null) {
- DebeziumReader<SourceRecords, MySqlSplit> currentReader = this.getCurrentReader();
- if (baseReq.isReload() || currentReader == null) {
- LOG.info(
- "No current reader or reload {}, create new split reader",
- baseReq.isReload());
- // build split
- Tuple2<MySqlSplit, Boolean> splitFlag = createMySqlSplit(offsetMeta, baseReq);
- split = splitFlag.f0;
- // reset binlog reader
- // closeBinlogReader();
- currentSplitRecords = pollSplitRecordsWithSplit(split, baseReq);
- this.setCurrentSplitRecords(currentSplitRecords);
- this.setCurrentSplit(split);
- } else if (currentReader instanceof BinlogSplitReader) {
- LOG.info("Continue poll records with current binlog reader");
- // only for binlog reader
- currentSplitRecords = pollSplitRecordsWithCurrentReader(currentReader);
- split = this.getCurrentSplit();
- } else {
- throw new RuntimeException("Should not happen");
- }
- } else {
- LOG.info(
- "Continue read records with current split records, splitId: {}",
- currentSplitRecords.getSplitId());
- }
+ // Create a new snapshot/binlog split based on offset and start the reader.
+ LOG.info("create new split reader for {} with offset {}", baseReq.getJobId(), offsetMeta);
+ // build split
+ Tuple2<MySqlSplit, Boolean> splitFlag = createMySqlSplit(offsetMeta, baseReq);
+ MySqlSplit split = splitFlag.f0;
+ // it's necessary to ensure that the binlog reader is already closed.
+ this.currentSplitRecords = pollSplitRecordsWithSplit(split, baseReq);
+ this.currentSplit = split;
Review Comment:
The logic that handles reusing an active split and continuing with the current reader has been removed (lines 173-206 in the original). This appears to be a significant behavioral change that forces a new split reader to be created on every call. This could impact performance and resource usage, especially for continuous streaming scenarios. If this change is intentional, it should be documented why the optimization for reusing readers was removed.
##########
be/src/io/fs/http_file_reader.cpp:
##########
@@ -95,34 +107,68 @@ Status HttpFileReader::open(const FileReaderOptions& opts) {
return Status::OK();
}
- // Step 1: HEAD request to get file metadata
- RETURN_IF_ERROR(prepare_client(/*set_fail_on_error=*/true));
- _client->set_method(HttpMethod::HEAD);
- RETURN_IF_ERROR(_client->execute());
+ // start CDC client
+ auto enable_cdc_iter = _extend_kv.find("enable_cdc_client");
+ if (enable_cdc_iter != _extend_kv.end() && enable_cdc_iter->second == "true") {
+ LOG(INFO) << "CDC client is enabled, starting CDC client for " << _url;
+ ExecEnv* env = ExecEnv::GetInstance();
+ if (env == nullptr || env->cdc_client_mgr() == nullptr) {
+ return Status::InternalError("ExecEnv or CdcClientMgr is not initialized");
+ }
+
+ PRequestCdcClientResult result;
+ Status start_st = env->cdc_client_mgr()->start_cdc_client(&result);
+ if (!start_st.ok()) {
+ LOG(ERROR) << "Failed to start CDC client, status=" << start_st.to_string();
+ return start_st;
+ }
+
+ _url = fmt::format(_url, doris::config::cdc_client_port);
+ _range_supported = false;
+ LOG(INFO) << "CDC client started successfully for " << _url;
+ }
Review Comment:
The CDC client is started during the file reader open operation, but there's no corresponding cleanup in the close() or destructor. If the CDC client has resources that need cleanup (connections, threads, etc.), they may leak. Consider adding cleanup logic in the close() method or ensuring the CDC client manager handles resource cleanup properly.
##########
fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/service/PipelineCoordinator.java:
##########
@@ -81,19 +95,153 @@ public PipelineCoordinator() {
new ThreadPoolExecutor.AbortPolicy());
}
+ /** return data for http_file_reader */
+ public StreamingResponseBody fetchRecordStream(FetchRecordRequest fetchReq) throws Exception {
+ SourceReader sourceReader;
+ SplitReadResult readResult;
+ try {
+ if (fetchReq.getTaskId() == null && fetchReq.getMeta() == null) {
+ LOG.info(
+ "Generate initial meta for fetch record request, jobId={}, taskId={}",
+ fetchReq.getJobId(),
+ fetchReq.getTaskId());
+ // means the request did not originate from the job, only tvf
+ Map<String, Object> meta = generateMeta(fetchReq.getConfig());
+ fetchReq.setMeta(meta);
+ }
+
+ sourceReader = Env.getCurrentEnv().getReader(fetchReq);
+ readResult = sourceReader.readSplitRecords(fetchReq);
+ } catch (Exception ex) {
+ throw new CommonException(ex);
+ }
+
+ return outputStream -> {
+ try {
+ buildRecords(sourceReader, fetchReq, readResult, outputStream);
+ } catch (Exception ex) {
+ LOG.error(
+ "Failed fetch record, jobId={}, taskId={}",
+ fetchReq.getJobId(),
+ fetchReq.getTaskId(),
+ ex);
+ throw new StreamException(ex);
+ }
+ };
+ }
+
+ private void buildRecords(
+ SourceReader sourceReader,
+ FetchRecordRequest fetchRecord,
+ SplitReadResult readResult,
+ OutputStream rawOutputStream)
+ throws Exception {
+ SourceSplit split = readResult.getSplit();
+ Map<String, String> lastMeta = null;
+ int rowCount = 0;
+ BufferedOutputStream bos = new BufferedOutputStream(rawOutputStream);
+ try {
+ // Serialize records and add them to the response (collect from iterator)
+ Iterator<SourceRecord> iterator = readResult.getRecordIterator();
+ while (iterator != null && iterator.hasNext()) {
+ SourceRecord element = iterator.next();
+ List<String> serializedRecords =
+ sourceReader.deserialize(fetchRecord.getConfig(), element);
+ for (String record : serializedRecords) {
+ bos.write(record.getBytes(StandardCharsets.UTF_8));
+ bos.write(LINE_DELIMITER);
+ }
+ rowCount += serializedRecords.size();
+ }
+ // force flush buffer
+ bos.flush();
+ } finally {
+ // The LSN in the commit is the current offset, which is the offset from the last
+ // successful write.
+ // Therefore, even if a subsequent write fails, it will not affect the commit.
+ sourceReader.commitSourceOffset(fetchRecord.getJobId(), readResult.getSplit());
+
+ // This must be called after commitSourceOffset; otherwise,
+ // PG's confirmed lsn will not proceed.
+ sourceReader.finishSplitRecords();
+ }
+
+ LOG.info(
+ "Fetch records completed, jobId={}, taskId={}, splitId={}, rowCount={}",
+ fetchRecord.getJobId(),
+ fetchRecord.getTaskId(),
+ split.splitId(),
+ rowCount);
+
+ if (readResult.getSplitState() != null) {
+ // Set meta information for hw
+ if (sourceReader.isSnapshotSplit(split)) {
+ lastMeta = sourceReader.extractSnapshotStateOffset(readResult.getSplitState());
+ lastMeta.put(SPLIT_ID, split.splitId());
+ }
+
+ // set meta for binlog event
+ if (sourceReader.isBinlogSplit(split)) {
+ lastMeta = sourceReader.extractBinlogStateOffset(readResult.getSplitState());
+ lastMeta.put(SPLIT_ID, BinlogSplit.BINLOG_SPLIT_ID);
+ }
+ } else {
+ throw new RuntimeException("split state is null");
+ }
+
+ if (StringUtils.isNotEmpty(fetchRecord.getTaskId())) {
+ taskOffsetCache.put(fetchRecord.getTaskId(), lastMeta);
+ }
+
+ if (!isLong(fetchRecord.getJobId())) {
+ // TVF requires closing the window after each execution,
+ // while PG requires dropping the slot.
+ sourceReader.close(fetchRecord);
+ }
+ }
+
+ private boolean isLong(String s) {
+ if (s == null || s.isEmpty()) return false;
+ try {
+ Long.parseLong(s);
+ return true;
+ } catch (NumberFormatException e) {
+ return false;
+ }
+ }
+
+ /** Generate split meta from request.offset */
+ private Map<String, Object> generateMeta(Map<String, String> cdcConfig)
+ throws JsonProcessingException {
+ Map<String, Object> meta = new HashMap<>();
+ String offset = cdcConfig.get(DataSourceConfigKeys.OFFSET);
+ if (DataSourceConfigKeys.OFFSET_LATEST.equalsIgnoreCase(offset)
+ || DataSourceConfigKeys.OFFSET_EARLIEST.equalsIgnoreCase(offset)) {
+ meta.put(SPLIT_ID, BinlogSplit.BINLOG_SPLIT_ID);
+ } else if (ConfigUtil.isJson(offset)) {
+ Map<String, String> startOffset =
+ objectMapper.readValue(offset, new TypeReference<>() {});
+ meta.put(SPLIT_ID, BinlogSplit.BINLOG_SPLIT_ID);
+ meta.put("startingOffset", startOffset);
+ } else {
+ throw new RuntimeException("Unsupported offset: " + offset);
Review Comment:
The generateMeta method only supports 'latest', 'earliest', and JSON offsets for TVF, but 'initial' is a valid startup mode according to the source reader implementations (MySQL and PostgreSQL both have code to handle it). This creates an inconsistency where 'initial' mode is supported in the source readers but explicitly rejected for TVF usage. Consider either supporting 'initial' in TVF or documenting why it's not supported.
##########
fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/service/PipelineCoordinator.java:
##########
@@ -42,26 +50,32 @@
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.common.base.Preconditions;
import io.debezium.data.Envelope;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;
+import org.springframework.web.servlet.mvc.method.annotation.StreamingResponseBody;
/** Pipeline coordinator. */
@Component
public class PipelineCoordinator {
private static final Logger LOG = LoggerFactory.getLogger(PipelineCoordinator.class);
private static final String SPLIT_ID = "splitId";
// jobId
- private final Map<Long, DorisBatchStreamLoad> batchStreamLoadMap = new ConcurrentHashMap<>();
+ private final Map<String, DorisBatchStreamLoad> batchStreamLoadMap = new ConcurrentHashMap<>();
// taskId -> writeFailReason
private final Map<String, String> taskErrorMaps = new ConcurrentHashMap<>();
+ // taskId, offset
Review Comment:
The taskOffsetCache Map is accessed without thread-safety considerations. While ConcurrentHashMap provides thread-safe operations for individual methods, the pattern of checking and retrieving data in getOffsetWithTaskId could still have race conditions if another thread removes the entry between check and retrieval in calling code. Consider documenting the thread-safety expectations or adding synchronization if needed.
```suggestion
/**
* Cache of task offsets keyed by taskId.
*
* <p>Thread-safety: this map is a {@link ConcurrentHashMap} and is accessed by multiple
* threads concurrently. Entries may be added and removed at any time by background tasks.
* Callers must treat the contents as a best-effort snapshot only and must be prepared for
* {@code null} or missing entries even if a previous check suggested that an entry existed.
*
* <p>In particular, callers of helper methods such as {@code getOffsetWithTaskId} must not
* assume that a "check then use" pattern on this cache is atomic. Any such logic must handle
* the possibility that another thread has removed or updated the entry between operations.
*/
```
##########
be/src/io/fs/http_file_reader.cpp:
##########
@@ -269,6 +335,21 @@ Status HttpFileReader::read_at_impl(size_t offset, Slice result, size_t* bytes_r
long http_status = _client->get_http_status();
VLOG(2) << "HTTP response: status=" << http_status << " received_bytes=" << buf.size();
+ // Check for HTTP error status codes (4xx, 5xx)
+ if (http_status >= 400) {
+ std::string error_body;
+ if (buf.empty()) {
+ error_body = "(empty response body)";
+ } else {
+ // Limit error message to 1024 bytes to avoid excessive logging
+ size_t max_len = std::min(buf.size(), static_cast<size_t>(1024));
+ error_body = buf.substr(0, max_len);
Review Comment:
The error handling for HTTP status codes >= 400 limits the error body to 1024 bytes. While this is reasonable for logging, the comment says "avoid excessive logging" but doesn't handle cases where the error message might be truncated in the middle of a multi-byte UTF-8 character. Consider using a character-aware truncation or noting that the message might be incomplete.
##########
fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/config/GlobalExceptionHandler.java:
##########
@@ -35,4 +39,17 @@ public Object exceptionHandler(HttpServletRequest request, Exception e) {
log.error("Unexpected exception", e);
return RestResponse.internalError(e.getMessage());
}
+
+ @ExceptionHandler(StreamException.class)
+ public Object streamExceptionHandler(StreamException e) {
+ // Directly throwing an exception allows curl to detect anomalies in the streaming response.
+ log.error("Exception in streaming response, re-throwing to client", e);
+ throw e;
+ }
+
Review Comment:
The StreamException handler re-throws the exception after logging. This is appropriate for streaming responses where you want the client to detect issues. However, the logging should include the exception details that were already logged by the lambda (lines 123-128 in PipelineCoordinator). This creates duplicate logging. Consider whether one of these log statements could be removed or consolidated.
```suggestion
throw e;
}
```
##########
fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/service/PipelineCoordinator.java:
##########
@@ -81,19 +95,153 @@ public PipelineCoordinator() {
new ThreadPoolExecutor.AbortPolicy());
}
+ /** return data for http_file_reader */
+ public StreamingResponseBody fetchRecordStream(FetchRecordRequest fetchReq) throws Exception {
+ SourceReader sourceReader;
+ SplitReadResult readResult;
+ try {
+ if (fetchReq.getTaskId() == null && fetchReq.getMeta() == null) {
+ LOG.info(
+ "Generate initial meta for fetch record request, jobId={}, taskId={}",
+ fetchReq.getJobId(),
+ fetchReq.getTaskId());
+ // means the request did not originate from the job, only tvf
+ Map<String, Object> meta = generateMeta(fetchReq.getConfig());
+ fetchReq.setMeta(meta);
+ }
+
+ sourceReader = Env.getCurrentEnv().getReader(fetchReq);
+ readResult = sourceReader.readSplitRecords(fetchReq);
+ } catch (Exception ex) {
+ throw new CommonException(ex);
+ }
+
+ return outputStream -> {
+ try {
+ buildRecords(sourceReader, fetchReq, readResult, outputStream);
+ } catch (Exception ex) {
+ LOG.error(
+ "Failed fetch record, jobId={}, taskId={}",
+ fetchReq.getJobId(),
+ fetchReq.getTaskId(),
+ ex);
+ throw new StreamException(ex);
+ }
+ };
+ }
+
+ private void buildRecords(
+ SourceReader sourceReader,
+ FetchRecordRequest fetchRecord,
+ SplitReadResult readResult,
+ OutputStream rawOutputStream)
+ throws Exception {
+ SourceSplit split = readResult.getSplit();
+ Map<String, String> lastMeta = null;
+ int rowCount = 0;
+ BufferedOutputStream bos = new BufferedOutputStream(rawOutputStream);
+ try {
+ // Serialize records and add them to the response (collect from iterator)
+ Iterator<SourceRecord> iterator = readResult.getRecordIterator();
+ while (iterator != null && iterator.hasNext()) {
+ SourceRecord element = iterator.next();
+ List<String> serializedRecords =
+ sourceReader.deserialize(fetchRecord.getConfig(), element);
+ for (String record : serializedRecords) {
+ bos.write(record.getBytes(StandardCharsets.UTF_8));
+ bos.write(LINE_DELIMITER);
+ }
+ rowCount += serializedRecords.size();
+ }
+ // force flush buffer
+ bos.flush();
+ } finally {
+ // The LSN in the commit is the current offset, which is the offset from the last
+ // successful write.
+ // Therefore, even if a subsequent write fails, it will not affect the commit.
+ sourceReader.commitSourceOffset(fetchRecord.getJobId(), readResult.getSplit());
+
+ // This must be called after commitSourceOffset; otherwise,
+ // PG's confirmed lsn will not proceed.
+ sourceReader.finishSplitRecords();
+ }
+
+ LOG.info(
+ "Fetch records completed, jobId={}, taskId={}, splitId={}, rowCount={}",
+ fetchRecord.getJobId(),
+ fetchRecord.getTaskId(),
+ split.splitId(),
+ rowCount);
+
+ if (readResult.getSplitState() != null) {
+ // Set meta information for hw
+ if (sourceReader.isSnapshotSplit(split)) {
+ lastMeta = sourceReader.extractSnapshotStateOffset(readResult.getSplitState());
+ lastMeta.put(SPLIT_ID, split.splitId());
+ }
+
+ // set meta for binlog event
+ if (sourceReader.isBinlogSplit(split)) {
+ lastMeta = sourceReader.extractBinlogStateOffset(readResult.getSplitState());
+ lastMeta.put(SPLIT_ID, BinlogSplit.BINLOG_SPLIT_ID);
+ }
+ } else {
+ throw new RuntimeException("split state is null");
+ }
+
+ if (StringUtils.isNotEmpty(fetchRecord.getTaskId())) {
+ taskOffsetCache.put(fetchRecord.getTaskId(), lastMeta);
+ }
+
+ if (!isLong(fetchRecord.getJobId())) {
+ // TVF requires closing the window after each execution,
+ // while PG requires dropping the slot.
+ sourceReader.close(fetchRecord);
+ }
+ }
+
+ private boolean isLong(String s) {
+ if (s == null || s.isEmpty()) return false;
+ try {
+ Long.parseLong(s);
+ return true;
+ } catch (NumberFormatException e) {
+ return false;
+ }
Review Comment:
The isLong method uses a try-catch for control flow to determine if a string is parseable as a Long. While this works, it's inefficient and considered an anti-pattern. Consider using a regex pattern or manual character validation instead. Example: return s.matches("^-?\\d+$");
##########
fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/utils/ConfigUtil.java:
##########
@@ -96,6 +98,21 @@ public static ZoneId getPostgresServerTimeZoneFromProps(java.util.Properties pro
return ZoneId.systemDefault();
}
+ public static String[] getTableList(String schema, Map<String, String> cdcConfig) {
+ String includingTables = cdcConfig.get(DataSourceConfigKeys.INCLUDE_TABLES);
+ String table = cdcConfig.get(DataSourceConfigKeys.TABLE);
+ if (StringUtils.isNotEmpty(includingTables)) {
+ return Arrays.stream(includingTables.split(","))
+ .map(t -> schema + "." + t.trim())
+ .toArray(String[]::new);
+ } else if (StringUtils.isNotEmpty(table)) {
+ Preconditions.checkArgument(!table.contains(","), "table only supports one table");
+ return new String[] {schema + "." + table.trim()};
+ } else {
+ return new String[0];
+ }
+ }
Review Comment:
The getTableList method doesn't validate that the schema parameter is not null or empty before using it in string concatenation. If schema is null or empty, this will create malformed table names like ".tablename". Add validation to ensure schema is not null/empty, or document that callers must validate this.
##########
fe/fe-common/src/main/java/org/apache/doris/job/cdc/DataSourceConfigKeys.java:
##########
@@ -19,13 +19,15 @@
public class DataSourceConfigKeys {
public static final String JDBC_URL = "jdbc_url";
Review Comment:
The new constant TYPE is added without documentation. Consider adding a JavaDoc comment explaining what this configuration key represents (e.g., "Data source type: 'mysql' or 'postgres'") to help users understand how to use it.
```suggestion
public static final String JDBC_URL = "jdbc_url";
/**
* Data source type, e.g. "mysql" or "postgres".
*/
```
##########
be/src/io/fs/http_file_reader.cpp:
##########
@@ -95,34 +107,68 @@ Status HttpFileReader::open(const FileReaderOptions& opts) {
return Status::OK();
}
- // Step 1: HEAD request to get file metadata
- RETURN_IF_ERROR(prepare_client(/*set_fail_on_error=*/true));
- _client->set_method(HttpMethod::HEAD);
- RETURN_IF_ERROR(_client->execute());
+ // start CDC client
+ auto enable_cdc_iter = _extend_kv.find("enable_cdc_client");
+ if (enable_cdc_iter != _extend_kv.end() && enable_cdc_iter->second == "true") {
+ LOG(INFO) << "CDC client is enabled, starting CDC client for " << _url;
+ ExecEnv* env = ExecEnv::GetInstance();
+ if (env == nullptr || env->cdc_client_mgr() == nullptr) {
+ return Status::InternalError("ExecEnv or CdcClientMgr is not initialized");
+ }
+
+ PRequestCdcClientResult result;
+ Status start_st = env->cdc_client_mgr()->start_cdc_client(&result);
+ if (!start_st.ok()) {
+ LOG(ERROR) << "Failed to start CDC client, status=" << start_st.to_string();
+ return start_st;
+ }
+
+ _url = fmt::format(_url, doris::config::cdc_client_port);
Review Comment:
The URI template placeholder "{}" expects to be formatted with the CDC client port, but there's no validation that the format operation succeeds or that the port value is valid before the HTTP request is made. If config::cdc_client_port is not set correctly, this could result in an invalid URL. Consider adding validation or error handling for the URL formatting.
##########
fe/fe-common/src/main/java/org/apache/doris/job/cdc/request/JobBaseRecordRequest.java:
##########
@@ -28,8 +28,4 @@
@EqualsAndHashCode(callSuper = true)
public abstract class JobBaseRecordRequest extends JobBaseConfig {
protected Map<String, Object> meta;
Review Comment:
The removal of isReload() and getFetchSize() abstract methods from JobBaseRecordRequest and their implementations in child classes appears to be a breaking change. If these methods were part of a public API or used by other components, this could cause compilation errors. Verify that no other code depends on these methods.
```suggestion
protected Map<String, Object> meta;
/**
* Whether this request is a reload.
* <p>
* Default implementation reads the "reload" flag from {@link #meta} if present,
* otherwise returns {@code false}.
*/
public boolean isReload() {
if (meta == null) {
return false;
}
Object value = meta.get("reload");
if (value instanceof Boolean) {
return (Boolean) value;
}
return false;
}
/**
* Fetch size for this request.
* <p>
* Default implementation reads the "fetchSize" value from {@link #meta} if present,
* otherwise returns {@code 0}.
*/
public int getFetchSize() {
if (meta == null) {
return 0;
}
Object value = meta.get("fetchSize");
if (value instanceof Number) {
return ((Number) value).intValue();
}
return 0;
}
```
##########
fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/source/reader/postgres/PostgresSourceReader.java:
##########
@@ -167,29 +166,18 @@ private PostgresSourceConfig generatePostgresConfig(Map<String, String> cdcConfi
configFactory.includeSchemaChanges(false);
// Set table list
- String includingTables = cdcConfig.get(DataSourceConfigKeys.INCLUDE_TABLES);
- if (StringUtils.isNotEmpty(includingTables)) {
- String[] includingTbls =
- Arrays.stream(includingTables.split(","))
- .map(t -> schema + "." + t.trim())
- .toArray(String[]::new);
- configFactory.tableList(includingTbls);
- }
+ String[] tableList = ConfigUtil.getTableList(schema, cdcConfig);
+ Preconditions.checkArgument(tableList.length >= 1, "include_tables or table is required");
+ configFactory.tableList(tableList);
// Set startup options
String startupMode = cdcConfig.get(DataSourceConfigKeys.OFFSET);
if (DataSourceConfigKeys.OFFSET_INITIAL.equalsIgnoreCase(startupMode)) {
configFactory.startupOptions(StartupOptions.initial());
- } else if (DataSourceConfigKeys.OFFSET_EARLIEST.equalsIgnoreCase(startupMode)) {
- configFactory.startupOptions(StartupOptions.earliest());
} else if (DataSourceConfigKeys.OFFSET_LATEST.equalsIgnoreCase(startupMode)) {
configFactory.startupOptions(StartupOptions.latest());
} else if (ConfigUtil.isJson(startupMode)) {
throw new RuntimeException("Unsupported json offset " + startupMode);
Review Comment:
The error message "Unsupported json offset" is misleading because it suggests JSON offsets are not supported, but the actual issue is that JSON offset parsing for PostgreSQL is not implemented. Consider changing to a more descriptive message like "JSON offset format is not yet supported for PostgreSQL" or "Specific LSN offsets are not supported, use 'initial' or 'latest'".
```suggestion
throw new RuntimeException(
"JSON offset format is not yet supported for PostgreSQL, use 'initial' or "
+ "'latest': "
+ startupMode);
```
##########
fe/fe-core/src/main/java/org/apache/doris/tablefunction/CdcStreamTableValuedFunction.java:
##########
@@ -0,0 +1,129 @@
+// 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.tablefunction;
+
+import org.apache.doris.analysis.BrokerDesc;
+import org.apache.doris.analysis.StorageBackend.StorageType;
+import org.apache.doris.catalog.Column;
+import org.apache.doris.common.AnalysisException;
+import org.apache.doris.datasource.jdbc.client.JdbcClient;
+import org.apache.doris.job.cdc.DataSourceConfigKeys;
+import org.apache.doris.job.cdc.request.FetchRecordRequest;
+import org.apache.doris.job.common.DataSourceType;
+import org.apache.doris.job.util.StreamingJobUtils;
+import org.apache.doris.thrift.TBrokerFileStatus;
+import org.apache.doris.thrift.TFileType;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.base.Preconditions;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+
+public class CdcStreamTableValuedFunction extends ExternalFileTableValuedFunction {
+ private static final ObjectMapper objectMapper = new ObjectMapper();
+ private static String URI = "http://127.0.0.1:{}/api/fetchRecordStream";
+ private final Map<String, String> originProps;
+
+ public CdcStreamTableValuedFunction(Map<String, String> properties) throws AnalysisException {
+ this.originProps = properties;
+ processProps(properties);
+ validate(properties);
+ }
+
+ private void processProps(Map<String, String> properties) throws AnalysisException {
+ Map<String, String> copyProps = new HashMap<>(properties);
+ copyProps.put("format", "json");
+ super.parseCommonProperties(copyProps);
+ this.processedParams.put("enable_cdc_client", "true");
+ this.processedParams.put("uri", URI);
+ this.processedParams.put("http.enable.range.request", "false");
+ this.processedParams.put("http.chunk.response", "true");
+ this.processedParams.put("http.method", "POST");
+
+ String payload = generateParams(properties);
+ this.processedParams.put("http.payload", payload);
+ this.backendConnectProperties.putAll(processedParams);
+ generateFileStatus();
+ }
+
+ private String generateParams(Map<String, String> properties) throws AnalysisException {
+ FetchRecordRequest recordRequest = new FetchRecordRequest();
+ recordRequest.setJobId(UUID.randomUUID().toString().replace("-", ""));
+ recordRequest.setDataSource(properties.get(DataSourceConfigKeys.TYPE));
+ recordRequest.setConfig(properties);
+ try {
+ return objectMapper.writeValueAsString(recordRequest);
+ } catch (IOException e) {
+ LOG.info("Failed to serialize fetch record request," + e.getMessage());
Review Comment:
The error logging at line 77 concatenates strings directly instead of using parameterized logging. This is less efficient and doesn't follow logging best practices. Change to: LOG.info("Failed to serialize fetch record request: {}", e.getMessage());
```suggestion
LOG.info("Failed to serialize fetch record request: {}", e.getMessage());
```
##########
fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/utils/ConfigUtil.java:
##########
@@ -96,6 +98,21 @@ public static ZoneId getPostgresServerTimeZoneFromProps(java.util.Properties pro
return ZoneId.systemDefault();
}
+ public static String[] getTableList(String schema, Map<String, String> cdcConfig) {
+ String includingTables = cdcConfig.get(DataSourceConfigKeys.INCLUDE_TABLES);
+ String table = cdcConfig.get(DataSourceConfigKeys.TABLE);
+ if (StringUtils.isNotEmpty(includingTables)) {
+ return Arrays.stream(includingTables.split(","))
+ .map(t -> schema + "." + t.trim())
+ .toArray(String[]::new);
+ } else if (StringUtils.isNotEmpty(table)) {
+ Preconditions.checkArgument(!table.contains(","), "table only supports one table");
Review Comment:
The table parameter check at line 109 uses Preconditions.checkArgument to ensure it doesn't contain commas, but this error won't be very helpful to users. Consider a more descriptive error message like "table parameter should contain a single table name, use include_tables for multiple tables" to guide users toward the correct parameter.
```suggestion
Preconditions.checkArgument(
!table.contains(","),
"table parameter should contain a single table name, use include_tables for multiple tables");
```
##########
fe/fe-core/src/main/java/org/apache/doris/tablefunction/CdcStreamTableValuedFunction.java:
##########
@@ -0,0 +1,129 @@
+// 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.tablefunction;
+
+import org.apache.doris.analysis.BrokerDesc;
+import org.apache.doris.analysis.StorageBackend.StorageType;
+import org.apache.doris.catalog.Column;
+import org.apache.doris.common.AnalysisException;
+import org.apache.doris.datasource.jdbc.client.JdbcClient;
+import org.apache.doris.job.cdc.DataSourceConfigKeys;
+import org.apache.doris.job.cdc.request.FetchRecordRequest;
+import org.apache.doris.job.common.DataSourceType;
+import org.apache.doris.job.util.StreamingJobUtils;
+import org.apache.doris.thrift.TBrokerFileStatus;
+import org.apache.doris.thrift.TFileType;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.base.Preconditions;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+
+public class CdcStreamTableValuedFunction extends ExternalFileTableValuedFunction {
+ private static final ObjectMapper objectMapper = new ObjectMapper();
+ private static String URI = "http://127.0.0.1:{}/api/fetchRecordStream";
+ private final Map<String, String> originProps;
+
+ public CdcStreamTableValuedFunction(Map<String, String> properties) throws AnalysisException {
+ this.originProps = properties;
+ processProps(properties);
+ validate(properties);
+ }
+
+ private void processProps(Map<String, String> properties) throws AnalysisException {
+ Map<String, String> copyProps = new HashMap<>(properties);
+ copyProps.put("format", "json");
+ super.parseCommonProperties(copyProps);
+ this.processedParams.put("enable_cdc_client", "true");
+ this.processedParams.put("uri", URI);
+ this.processedParams.put("http.enable.range.request", "false");
+ this.processedParams.put("http.chunk.response", "true");
+ this.processedParams.put("http.method", "POST");
+
+ String payload = generateParams(properties);
+ this.processedParams.put("http.payload", payload);
+ this.backendConnectProperties.putAll(processedParams);
+ generateFileStatus();
+ }
+
+ private String generateParams(Map<String, String> properties) throws AnalysisException {
+ FetchRecordRequest recordRequest = new FetchRecordRequest();
+ recordRequest.setJobId(UUID.randomUUID().toString().replace("-", ""));
+ recordRequest.setDataSource(properties.get(DataSourceConfigKeys.TYPE));
+ recordRequest.setConfig(properties);
+ try {
+ return objectMapper.writeValueAsString(recordRequest);
+ } catch (IOException e) {
+ LOG.info("Failed to serialize fetch record request," + e.getMessage());
+ throw new AnalysisException(e.getMessage());
+ }
+ }
+
+ private void validate(Map<String, String> properties) {
+ Preconditions.checkArgument(properties.containsKey(DataSourceConfigKeys.JDBC_URL), "jdbc_url is required");
+ Preconditions.checkArgument(properties.containsKey(DataSourceConfigKeys.TYPE), "type is required");
+ Preconditions.checkArgument(properties.containsKey(DataSourceConfigKeys.TABLE), "table is required");
+ }
+
+ private void generateFileStatus() {
+ this.fileStatuses.clear();
+ this.fileStatuses.add(new TBrokerFileStatus(URI, false, Integer.MAX_VALUE, false));
+ }
+
+ @Override
+ public List<Column> getTableColumns() throws AnalysisException {
+ DataSourceType dataSourceType =
+ DataSourceType.valueOf(processedParams.get(DataSourceConfigKeys.TYPE).toUpperCase());
Review Comment:
The getTableColumns method uses valueOf which will throw IllegalArgumentException for invalid data source types, but doesn't provide a user-friendly error message. Consider adding a try-catch to provide a more helpful error message listing the supported types (e.g., "Unsupported data source type: X. Supported types are: MYSQL, POSTGRES").
```suggestion
String typeStr = processedParams.get(DataSourceConfigKeys.TYPE);
DataSourceType dataSourceType;
try {
dataSourceType = DataSourceType.valueOf(typeStr.toUpperCase());
} catch (IllegalArgumentException e) {
DataSourceType[] supportedTypes = DataSourceType.values();
StringBuilder supported = new StringBuilder();
for (int i = 0; i < supportedTypes.length; i++) {
if (i > 0) {
supported.append(", ");
}
supported.append(supportedTypes[i].name());
}
throw new AnalysisException("Unsupported data source type: " + typeStr
+ ". Supported types are: " + supported);
}
```
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [Feature](tvf) Support cdc stream tvf for mysql and pg [doris]
Posted by "doris-robot (via GitHub)" <gi...@apache.org>.
doris-robot commented on PR #60116:
URL: https://github.com/apache/doris/pull/60116#issuecomment-3782488272
<details>
<summary>TPC-DS: <b>Total hot run time: 173013 ms</b></summary>
```
machine: 'aliyun_ecs.c7a.8xlarge_32C64G'
scripts: https://github.com/apache/doris/tree/master/tools/tpcds-tools
TPC-DS sf100 test result on commit b56d1d11f9b7fee728fc06d90ddd84506f443adf, data reload: false
query5 4379 658 525 525
query6 330 221 226 221
query7 4224 471 259 259
query8 342 260 242 242
query9 8688 2876 2904 2876
query10 464 349 295 295
query11 15207 15167 14874 14874
query12 185 126 117 117
query13 1281 472 383 383
query14 5639 3477 2953 2953
query14_1 2816 2838 2804 2804
query15 200 194 171 171
query16 955 499 394 394
query17 1108 660 564 564
query18 2429 430 331 331
query19 196 186 153 153
query20 120 117 115 115
query21 212 137 116 116
query22 3799 3890 3926 3890
query23 16020 15667 15351 15351
query23_1 15441 15325 15628 15325
query24 7088 1534 1187 1187
query24_1 1177 1188 1176 1176
query25 520 437 400 400
query26 1249 280 147 147
query27 2755 439 275 275
query28 4612 2152 2140 2140
query29 835 529 444 444
query30 318 236 199 199
query31 772 645 561 561
query32 84 74 74 74
query33 512 364 326 326
query34 920 891 553 553
query35 724 754 684 684
query36 831 916 803 803
query37 136 101 85 85
query38 2750 2706 2614 2614
query39 774 765 722 722
query39_1 706 739 749 739
query40 218 142 122 122
query41 68 65 63 63
query42 98 90 95 90
query43 460 434 401 401
query44 1381 742 745 742
query45 188 187 177 177
query46 866 950 586 586
query47 1402 1389 1311 1311
query48 324 327 245 245
query49 611 423 353 353
query50 707 276 224 224
query51 3748 3755 3756 3755
query52 94 95 85 85
query53 215 226 171 171
query54 284 257 244 244
query55 84 80 75 75
query56 306 319 309 309
query57 1022 984 920 920
query58 276 271 269 269
query59 2180 2077 2041 2041
query60 336 337 318 318
query61 146 147 146 146
query62 378 358 312 312
query63 203 171 177 171
query64 5037 1157 850 850
query65 3759 3699 3796 3699
query66 1454 409 323 323
query67 15578 15666 15601 15601
query68 2349 1068 727 727
query69 396 320 282 282
query70 997 875 882 875
query71 302 303 287 287
query72 5307 3240 3358 3240
query73 623 749 321 321
query74 8714 8790 8580 8580
query75 2370 2327 1913 1913
query76 2286 1097 666 666
query77 363 393 316 316
query78 9743 9840 9199 9199
query79 1073 934 596 596
query80 619 577 436 436
query81 474 262 228 228
query82 1372 150 122 122
query83 364 266 243 243
query84 260 124 99 99
query85 824 466 406 406
query86 369 296 321 296
query87 2932 2864 2809 2809
query88 3463 2587 2571 2571
query89 311 267 253 253
query90 1879 190 178 178
query91 168 161 129 129
query92 81 77 75 75
query93 1079 1093 638 638
query94 457 340 292 292
query95 590 348 345 345
query96 652 523 233 233
query97 2391 2383 2282 2282
query98 214 227 213 213
query99 609 619 509 509
Total cold run time: 245476 ms
Total hot run time: 173013 ms
```
</details>
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [Feature](tvf) Support cdc stream tvf for mysql and pg [doris]
Posted by "doris-robot (via GitHub)" <gi...@apache.org>.
doris-robot commented on PR #60116:
URL: https://github.com/apache/doris/pull/60116#issuecomment-3782898046
<details>
<summary>TPC-H: <b>Total hot run time: 30904 ms</b></summary>
```
machine: 'aliyun_ecs.c7a.8xlarge_32C64G'
scripts: https://github.com/apache/doris/tree/master/tools/tpch-tools
Tpch sf100 test result on commit e02811eaefda6a3a1631ee0683594c271015510b, data reload: false
------ Round 1 ----------------------------------
q1 17682 4836 4545 4545
q2 2077 315 199 199
q3 10199 1292 717 717
q4 10211 845 300 300
q5 7515 2086 1816 1816
q6 183 170 139 139
q7 829 710 588 588
q8 9271 1345 1101 1101
q9 4933 4566 4573 4566
q10 6785 1618 1278 1278
q11 491 289 276 276
q12 339 368 219 219
q13 18275 3825 3274 3274
q14 241 249 208 208
q15 594 529 527 527
q16 652 648 581 581
q17 654 783 507 507
q18 6733 6561 6354 6354
q19 1234 966 618 618
q20 388 342 229 229
q21 2636 1982 1916 1916
q22 1045 969 946 946
Total cold run time: 102967 ms
Total hot run time: 30904 ms
----- Round 2, with runtime_filter_mode=off -----
q1 4704 4688 4654 4654
q2 324 379 336 336
q3 2148 2663 2304 2304
q4 1313 1779 1309 1309
q5 4101 3963 4066 3963
q6 204 171 135 135
q7 1904 1807 1759 1759
q8 2739 2487 2464 2464
q9 7392 7104 7184 7104
q10 2487 2830 2308 2308
q11 556 472 458 458
q12 708 741 631 631
q13 3616 4143 3503 3503
q14 320 327 283 283
q15 558 522 513 513
q16 639 660 661 660
q17 1136 1318 1282 1282
q18 7832 7747 7842 7747
q19 860 853 856 853
q20 2041 2228 1857 1857
q21 4812 4420 4114 4114
q22 1094 1015 984 984
Total cold run time: 51488 ms
Total hot run time: 49221 ms
```
</details>
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [Feature](tvf) Support cdc stream tvf for mysql and pg [doris]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #60116:
URL: https://github.com/apache/doris/pull/60116#issuecomment-4117667462
PR approved by at least one committer and no changes requested.
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [Feature](tvf) Support cdc stream tvf for mysql and pg [doris]
Posted by "JNSimba (via GitHub)" <gi...@apache.org>.
JNSimba commented on PR #60116:
URL: https://github.com/apache/doris/pull/60116#issuecomment-4116995016
/review
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [Feature](tvf) Support cdc stream tvf for mysql and pg [doris]
Posted by "JNSimba (via GitHub)" <gi...@apache.org>.
JNSimba commented on PR #60116:
URL: https://github.com/apache/doris/pull/60116#issuecomment-4125195078
run beut
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [Feature](tvf) Support cdc stream tvf for mysql and pg [doris]
Posted by "doris-robot (via GitHub)" <gi...@apache.org>.
doris-robot commented on PR #60116:
URL: https://github.com/apache/doris/pull/60116#issuecomment-4117583467
<details>
<summary>TPC-DS: <b>Total hot run time: 169381 ms</b></summary>
```
machine: 'aliyun_ecs.c7a.8xlarge_32C64G'
scripts: https://github.com/apache/doris/tree/master/tools/tpcds-tools
TPC-DS sf100 test result on commit aa433727e52f6f04fafef418019df4bd5b6a3cd7, data reload: false
query5 4329 627 493 493
query6 328 233 212 212
query7 4206 452 266 266
query8 346 232 223 223
query9 8749 2755 2734 2734
query10 531 408 362 362
query11 7001 5115 4883 4883
query12 187 129 139 129
query13 1287 475 353 353
query14 5809 3691 3475 3475
query14_1 2864 2878 2853 2853
query15 207 195 176 176
query16 978 462 421 421
query17 890 733 638 638
query18 2442 450 353 353
query19 213 212 190 190
query20 136 129 129 129
query21 211 142 111 111
query22 13325 14116 14629 14116
query23 16949 16591 16242 16242
query23_1 16467 15780 15740 15740
query24 7141 1624 1232 1232
query24_1 1239 1222 1253 1222
query25 613 466 407 407
query26 1238 257 152 152
query27 2762 481 294 294
query28 5035 1838 1825 1825
query29 857 578 480 480
query30 293 226 190 190
query31 1036 950 881 881
query32 79 73 70 70
query33 522 327 290 290
query34 909 864 514 514
query35 640 674 612 612
query36 1104 1097 968 968
query37 135 95 82 82
query38 2912 2904 2853 2853
query39 858 825 805 805
query39_1 801 792 801 792
query40 228 153 135 135
query41 61 63 57 57
query42 264 255 255 255
query43 242 246 221 221
query44
query45 197 196 195 195
query46 879 985 608 608
query47 2124 2133 2049 2049
query48 309 318 239 239
query49 639 460 378 378
query50 693 271 209 209
query51 4111 4110 4007 4007
query52 263 272 254 254
query53 293 336 282 282
query54 313 270 266 266
query55 94 91 84 84
query56 329 326 318 318
query57 1910 1738 1627 1627
query58 285 278 280 278
query59 2784 2930 2725 2725
query60 348 333 326 326
query61 155 155 154 154
query62 645 578 546 546
query63 311 278 274 274
query64 5093 1283 991 991
query65
query66 1459 453 362 362
query67 24248 24292 24241 24241
query68
query69 409 308 284 284
query70 888 986 944 944
query71 344 318 284 284
query72 2873 2654 2456 2456
query73 563 565 336 336
query74 9719 9557 9452 9452
query75 2848 2774 2503 2503
query76 2279 1029 705 705
query77 364 393 324 324
query78 10905 11073 10499 10499
query79 2613 780 588 588
query80 1793 644 546 546
query81 544 266 220 220
query82 1011 153 117 117
query83 334 264 246 246
query84 298 131 101 101
query85 928 495 447 447
query86 415 307 282 282
query87 3137 3136 3022 3022
query88 3575 2681 2684 2681
query89 440 374 343 343
query90 2005 182 185 182
query91 167 164 146 146
query92 78 75 70 70
query93 1160 825 498 498
query94 640 323 292 292
query95 595 400 320 320
query96 657 526 230 230
query97 2477 2513 2432 2432
query98 237 221 214 214
query99 1039 990 917 917
Total cold run time: 253340 ms
Total hot run time: 169381 ms
```
</details>
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [Feature](tvf) Support cdc stream tvf for mysql and pg [doris]
Posted by "doris-robot (via GitHub)" <gi...@apache.org>.
doris-robot commented on PR #60116:
URL: https://github.com/apache/doris/pull/60116#issuecomment-4117511867
<details>
<summary>TPC-H: <b>Total hot run time: 26664 ms</b></summary>
```
machine: 'aliyun_ecs.c7a.8xlarge_32C64G'
scripts: https://github.com/apache/doris/tree/master/tools/tpch-tools
Tpch sf100 test result on commit aa433727e52f6f04fafef418019df4bd5b6a3cd7, data reload: false
------ Round 1 ----------------------------------
orders Doris NULL NULL 0 0 0 NULL 0 NULL NULL 2023-12-26 18:27:23 2023-12-26 18:42:55 NULL utf-8 NULL NULL
============================================
q1 17607 4538 4328 4328
q2 q3 10637 787 521 521
q4 4673 347 245 245
q5 7575 1208 1027 1027
q6 183 176 146 146
q7 789 836 669 669
q8 9705 1469 1325 1325
q9 5322 4769 4715 4715
q10 6346 1919 1647 1647
q11 475 264 248 248
q12 765 586 482 482
q13 18061 2716 1943 1943
q14 234 240 227 227
q15 q16 731 744 659 659
q17 752 848 412 412
q18 5882 5477 5272 5272
q19 1263 983 599 599
q20 545 496 373 373
q21 4521 1825 1520 1520
q22 470 340 306 306
Total cold run time: 96536 ms
Total hot run time: 26664 ms
----- Round 2, with runtime_filter_mode=off -----
orders Doris NULL NULL 150000000 42 6422171781 NULL 22778155 NULL NULL 2023-12-26 18:27:23 2023-12-26 18:42:55 NULL utf-8 NULL NULL
============================================
q1 4835 4711 4586 4586
q2 q3 4013 4345 3842 3842
q4 933 1212 867 867
q5 4210 4425 4307 4307
q6 183 181 144 144
q7 1801 1699 1547 1547
q8 2499 2712 2576 2576
q9 7621 7360 7377 7360
q10 3764 4116 3747 3747
q11 511 440 424 424
q12 524 694 457 457
q13 2464 2893 2108 2108
q14 290 328 299 299
q15 q16 728 761 712 712
q17 1217 1504 1410 1410
q18 7063 6781 6621 6621
q19 991 913 918 913
q20 2072 2207 2000 2000
q21 3990 3477 3347 3347
q22 476 435 387 387
Total cold run time: 50185 ms
Total hot run time: 47654 ms
```
</details>
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [Feature](tvf) Support cdc stream tvf for mysql and pg [doris]
Posted by "hello-stephen (via GitHub)" <gi...@apache.org>.
hello-stephen commented on PR #60116:
URL: https://github.com/apache/doris/pull/60116#issuecomment-4116869509
# FE UT Coverage Report
Increment line coverage ` 3.08% (2/65)` :tada:
[Increment coverage report](http://coverage.selectdb-in.cc/coverage/60116_d4210f02b4944ba2c18879a12b940c093405187b/fe_increment_report/index.html)
[Complete coverage report](http://coverage.selectdb-in.cc/coverage/60116_d4210f02b4944ba2c18879a12b940c093405187b/fe_report/index.html)
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [Feature](tvf) Support cdc stream tvf for mysql and pg [doris]
Posted by "doris-robot (via GitHub)" <gi...@apache.org>.
doris-robot commented on PR #60116:
URL: https://github.com/apache/doris/pull/60116#issuecomment-4125096255
<details>
<summary>TPC-DS: <b>Total hot run time: 167899 ms</b></summary>
```
machine: 'aliyun_ecs.c7a.8xlarge_32C64G'
scripts: https://github.com/apache/doris/tree/master/tools/tpcds-tools
TPC-DS sf100 test result on commit 9ea6fe06a5cddc20a17b3f4565082396852f0a56, data reload: false
query5 4335 648 512 512
query6 346 237 222 222
query7 4210 454 265 265
query8 348 247 233 233
query9 8720 2710 2710 2710
query10 529 398 365 365
query11 6956 5090 4856 4856
query12 181 139 130 130
query13 1302 458 359 359
query14 5769 3661 3456 3456
query14_1 2863 2822 2835 2822
query15 203 191 177 177
query16 974 501 453 453
query17 1130 721 613 613
query18 2439 450 355 355
query19 219 211 191 191
query20 138 127 128 127
query21 216 143 113 113
query22 13357 13221 13086 13086
query23 16236 15737 16314 15737
query23_1 16204 16212 16097 16097
query24 7395 1669 1254 1254
query24_1 1348 1275 1267 1267
query25 616 589 463 463
query26 1235 257 149 149
query27 2794 472 289 289
query28 4538 1830 1813 1813
query29 850 562 476 476
query30 299 225 190 190
query31 1005 956 860 860
query32 81 74 69 69
query33 510 328 283 283
query34 910 859 506 506
query35 660 664 606 606
query36 1074 1097 1004 1004
query37 133 92 83 83
query38 2931 2918 2839 2839
query39 855 838 808 808
query39_1 798 791 796 791
query40 234 152 135 135
query41 62 99 59 59
query42 257 256 259 256
query43 238 241 224 224
query44
query45 202 189 182 182
query46 875 991 605 605
query47 2087 2103 2068 2068
query48 315 326 229 229
query49 639 458 386 386
query50 683 276 211 211
query51 4053 4030 3938 3938
query52 259 267 260 260
query53 288 334 289 289
query54 309 276 266 266
query55 92 83 84 83
query56 327 319 315 315
query57 1911 1788 1764 1764
query58 298 282 269 269
query59 2783 2954 2718 2718
query60 334 339 320 320
query61 159 159 158 158
query62 620 587 535 535
query63 306 281 279 279
query64 5108 1285 991 991
query65
query66 1450 454 346 346
query67 24267 24270 24148 24148
query68
query69 423 327 296 296
query70 868 970 920 920
query71 350 319 300 300
query72 2864 2740 2486 2486
query73 548 542 313 313
query74 9663 9531 9350 9350
query75 2899 2760 2477 2477
query76 2281 1035 679 679
query77 358 374 315 315
query78 10935 11103 10468 10468
query79 1424 760 563 563
query80 1363 607 554 554
query81 546 265 233 233
query82 1003 153 125 125
query83 333 262 241 241
query84 246 120 99 99
query85 936 505 441 441
query86 410 309 324 309
query87 3084 3113 3048 3048
query88 3578 2660 2648 2648
query89 428 371 341 341
query90 2027 173 167 167
query91 166 162 139 139
query92 72 72 73 72
query93 1003 862 504 504
query94 635 323 291 291
query95 578 352 313 313
query96 645 515 228 228
query97 2478 2486 2377 2377
query98 232 245 224 224
query99 1025 996 897 897
Total cold run time: 250279 ms
Total hot run time: 167899 ms
```
</details>
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [Feature](tvf) Support cdc stream tvf for mysql and pg [doris]
Posted by "JNSimba (via GitHub)" <gi...@apache.org>.
JNSimba commented on PR #60116:
URL: https://github.com/apache/doris/pull/60116#issuecomment-4126031364
run p0
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [Feature](tvf) Support cdc stream tvf for mysql and pg [doris]
Posted by "JNSimba (via GitHub)" <gi...@apache.org>.
JNSimba commented on PR #60116:
URL: https://github.com/apache/doris/pull/60116#issuecomment-4126265075
run p0
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [Feature](tvf) Support cdc stream tvf for mysql and pg [doris]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #60116:
URL: https://github.com/apache/doris/pull/60116#issuecomment-3815897693
PR approved by anyone and no changes requested.
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [Feature](tvf) Support cdc stream tvf for mysql and pg [doris]
Posted by "JNSimba (via GitHub)" <gi...@apache.org>.
JNSimba commented on PR #60116:
URL: https://github.com/apache/doris/pull/60116#issuecomment-3782756898
run buildall
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [Feature](tvf) Support cdc stream tvf for mysql and pg [doris]
Posted by "doris-robot (via GitHub)" <gi...@apache.org>.
doris-robot commented on PR #60116:
URL: https://github.com/apache/doris/pull/60116#issuecomment-3782949037
<details>
<summary>ClickBench: <b>Total hot run time: 26.76 s</b></summary>
```
machine: 'aliyun_ecs.c7a.8xlarge_32C64G'
scripts: https://github.com/apache/doris/tree/master/tools/clickbench-tools
ClickBench test result on commit e02811eaefda6a3a1631ee0683594c271015510b, data reload: false
query1 0.05 0.05 0.05
query2 0.09 0.04 0.04
query3 0.26 0.09 0.08
query4 1.60 0.11 0.12
query5 0.27 0.26 0.25
query6 1.20 0.65 0.65
query7 0.03 0.03 0.03
query8 0.06 0.04 0.04
query9 0.57 0.49 0.50
query10 0.55 0.54 0.53
query11 0.16 0.10 0.10
query12 0.15 0.11 0.12
query13 0.60 0.59 0.58
query14 0.96 0.96 0.94
query15 0.78 0.78 0.78
query16 0.39 0.39 0.41
query17 1.04 1.04 1.05
query18 0.22 0.21 0.22
query19 1.96 1.89 1.90
query20 0.02 0.01 0.02
query21 15.41 0.26 0.14
query22 5.15 0.06 0.05
query23 16.00 0.29 0.11
query24 2.90 0.35 0.72
query25 0.09 0.09 0.05
query26 0.14 0.13 0.13
query27 0.08 0.05 0.04
query28 4.93 1.06 0.88
query29 12.58 3.85 3.14
query30 0.28 0.14 0.12
query31 2.81 0.63 0.40
query32 3.24 0.56 0.45
query33 2.97 2.95 3.03
query34 15.97 5.05 4.42
query35 4.41 4.39 4.46
query36 0.64 0.50 0.49
query37 0.12 0.07 0.06
query38 0.08 0.05 0.04
query39 0.05 0.03 0.03
query40 0.17 0.14 0.13
query41 0.09 0.03 0.02
query42 0.05 0.03 0.02
query43 0.04 0.03 0.04
Total cold run time: 99.16 s
Total hot run time: 26.76 s
```
</details>
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [Feature](tvf) Support cdc stream tvf for mysql and pg [doris]
Posted by "hello-stephen (via GitHub)" <gi...@apache.org>.
hello-stephen commented on PR #60116:
URL: https://github.com/apache/doris/pull/60116#issuecomment-3783057461
# BE UT Coverage Report
Increment line coverage `0.00% (0/79)` :tada:
[Increment coverage report](http://coverage.selectdb-in.cc/coverage/e02811eaefda6a3a1631ee0683594c271015510b_e02811eaefda6a3a1631ee0683594c271015510b/increment_report/index.html)
[Complete coverage report](http://coverage.selectdb-in.cc/coverage/e02811eaefda6a3a1631ee0683594c271015510b_e02811eaefda6a3a1631ee0683594c271015510b/report/index.html)
| Category | Coverage |
|-------------------|--------------------|
| Function Coverage | 52.46% (19116/36440) |
| Line Coverage | 35.83% (177582/495676) |
| Region Coverage | 32.26% (137141/425099) |
| Branch Coverage | 33.22% (59417/178833) |
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [Feature](tvf) Support cdc stream tvf for mysql and pg [doris]
Posted by "hello-stephen (via GitHub)" <gi...@apache.org>.
hello-stephen commented on PR #60116:
URL: https://github.com/apache/doris/pull/60116#issuecomment-3783390821
# FE Regression Coverage Report
Increment line coverage ` 3.08% (2/65)` :tada:
[Increment coverage report](http://coverage.selectdb-in.cc/coverage/60116_e02811eaefda6a3a1631ee0683594c271015510b_merge_fe/increment_report/index.html)
[Complete coverage report](http://coverage.selectdb-in.cc/coverage/60116_e02811eaefda6a3a1631ee0683594c271015510b_merge_fe/report/index.html)
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [Feature](tvf) Support cdc stream tvf for mysql and pg [doris]
Posted by "hello-stephen (via GitHub)" <gi...@apache.org>.
hello-stephen commented on PR #60116:
URL: https://github.com/apache/doris/pull/60116#issuecomment-3779011807
# BE Regression && UT Coverage Report
Increment line coverage `65.82% (52/79)` :tada:
[Increment coverage report](http://coverage.selectdb-in.cc/coverage/60116_d84fcddbe4aa9142e912da69ab6bafe20075ca2e_merge/increment_report/index.html)
[Complete coverage report](http://coverage.selectdb-in.cc/coverage/60116_d84fcddbe4aa9142e912da69ab6bafe20075ca2e_merge/report/index.html)
| Category | Coverage |
|-------------------|--------------------|
| Function Coverage | 73.15% (26126/35718) |
| Line Coverage | 56.24% (278395/494978) |
| Region Coverage | 54.02% (231974/429398) |
| Branch Coverage | 55.69% (99963/179498) |
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [Feature](tvf) Support cdc stream tvf for mysql and pg [doris]
Posted by "doris-robot (via GitHub)" <gi...@apache.org>.
doris-robot commented on PR #60116:
URL: https://github.com/apache/doris/pull/60116#issuecomment-3782934214
<details>
<summary>TPC-DS: <b>Total hot run time: 172737 ms</b></summary>
```
machine: 'aliyun_ecs.c7a.8xlarge_32C64G'
scripts: https://github.com/apache/doris/tree/master/tools/tpcds-tools
TPC-DS sf100 test result on commit e02811eaefda6a3a1631ee0683594c271015510b, data reload: false
query5 4427 642 490 490
query6 335 214 193 193
query7 4238 452 260 260
query8 337 240 229 229
query9 8697 2851 2844 2844
query10 476 346 298 298
query11 15121 15111 14900 14900
query12 177 121 116 116
query13 1239 459 373 373
query14 6094 3036 2780 2780
query14_1 2693 2662 2660 2660
query15 201 196 177 177
query16 968 478 450 450
query17 1113 676 584 584
query18 2439 436 339 339
query19 210 182 174 174
query20 132 129 115 115
query21 216 138 128 128
query22 3964 3849 3921 3849
query23 16043 15500 15406 15406
query23_1 15574 15489 15536 15489
query24 7152 1527 1171 1171
query24_1 1159 1162 1157 1157
query25 548 457 411 411
query26 1250 271 159 159
query27 2756 448 280 280
query28 4540 2147 2123 2123
query29 789 560 464 464
query30 309 238 205 205
query31 781 643 548 548
query32 83 77 75 75
query33 538 362 325 325
query34 887 880 528 528
query35 727 756 674 674
query36 871 914 865 865
query37 148 117 95 95
query38 2704 2707 2635 2635
query39 777 757 736 736
query39_1 718 712 710 710
query40 221 138 123 123
query41 76 71 68 68
query42 101 97 95 95
query43 443 476 415 415
query44 1328 746 753 746
query45 190 192 179 179
query46 837 939 577 577
query47 1434 1473 1335 1335
query48 311 326 238 238
query49 614 422 344 344
query50 671 264 201 201
query51 3686 3771 3715 3715
query52 94 99 84 84
query53 215 230 158 158
query54 276 264 273 264
query55 83 81 75 75
query56 296 306 297 297
query57 1014 1014 872 872
query58 268 256 268 256
query59 1873 2172 2075 2075
query60 320 325 314 314
query61 163 143 142 142
query62 375 359 319 319
query63 193 161 164 161
query64 4895 1173 853 853
query65 3834 3752 3728 3728
query66 1454 427 307 307
query67 15359 15508 15401 15401
query68 2409 1101 712 712
query69 390 308 271 271
query70 1008 945 929 929
query71 310 307 274 274
query72 5260 3055 3189 3055
query73 597 713 321 321
query74 8726 8798 8543 8543
query75 2342 2327 1885 1885
query76 2259 1043 661 661
query77 349 378 292 292
query78 9667 9843 9125 9125
query79 1082 917 619 619
query80 1298 540 436 436
query81 562 263 236 236
query82 1011 151 121 121
query83 324 274 243 243
query84 255 118 94 94
query85 872 480 408 408
query86 436 301 286 286
query87 2870 2834 2739 2739
query88 3550 2586 2558 2558
query89 300 259 238 238
query90 1961 169 170 169
query91 166 160 142 142
query92 75 70 67 67
query93 1084 1032 652 652
query94 650 333 293 293
query95 591 331 317 317
query96 650 510 230 230
query97 2323 2369 2346 2346
query98 214 199 207 199
query99 605 580 511 511
Total cold run time: 245391 ms
Total hot run time: 172737 ms
```
</details>
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [Feature](tvf) Support cdc stream tvf for mysql and pg [doris]
Posted by "JNSimba (via GitHub)" <gi...@apache.org>.
JNSimba commented on PR #60116:
URL: https://github.com/apache/doris/pull/60116#issuecomment-3782310810
run buildall
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [Feature](tvf) Support cdc stream tvf for mysql and pg [doris]
Posted by "doris-robot (via GitHub)" <gi...@apache.org>.
doris-robot commented on PR #60116:
URL: https://github.com/apache/doris/pull/60116#issuecomment-3777553703
<details>
<summary>TPC-H: <b>Total hot run time: 32113 ms</b></summary>
```
machine: 'aliyun_ecs.c7a.8xlarge_32C64G'
scripts: https://github.com/apache/doris/tree/master/tools/tpch-tools
Tpch sf100 test result on commit d84fcddbe4aa9142e912da69ab6bafe20075ca2e, data reload: false
------ Round 1 ----------------------------------
q1 17610 4269 4073 4073
q2 2069 361 255 255
q3 10100 1284 721 721
q4 10193 773 301 301
q5 7490 2100 1882 1882
q6 189 168 134 134
q7 940 791 644 644
q8 9260 1468 1110 1110
q9 5010 4541 4588 4541
q10 6807 1781 1393 1393
q11 527 299 274 274
q12 760 745 605 605
q13 17803 3861 3109 3109
q14 300 302 269 269
q15 597 522 515 515
q16 731 689 647 647
q17 665 736 598 598
q18 6856 6500 6937 6500
q19 1661 1078 702 702
q20 412 373 301 301
q21 3327 2643 2501 2501
q22 1113 1064 1038 1038
Total cold run time: 104420 ms
Total hot run time: 32113 ms
----- Round 2, with runtime_filter_mode=off -----
q1 4342 4275 4369 4275
q2 333 425 326 326
q3 2253 2871 2521 2521
q4 1426 1851 1408 1408
q5 4578 4358 4279 4279
q6 213 161 127 127
q7 1972 1936 1809 1809
q8 2547 2399 2357 2357
q9 7287 7500 7152 7152
q10 2518 2648 2296 2296
q11 540 480 461 461
q12 683 727 565 565
q13 3381 3845 3113 3113
q14 264 280 265 265
q15 533 486 476 476
q16 627 649 639 639
q17 1087 1255 1306 1255
q18 7352 7303 7056 7056
q19 846 792 833 792
q20 1872 1956 1797 1797
q21 4521 4234 4070 4070
q22 1070 1079 1009 1009
Total cold run time: 50245 ms
Total hot run time: 48048 ms
```
</details>
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [Feature](tvf) Support cdc stream tvf for mysql and pg [doris]
Posted by "hello-stephen (via GitHub)" <gi...@apache.org>.
hello-stephen commented on PR #60116:
URL: https://github.com/apache/doris/pull/60116#issuecomment-3778340001
# BE UT Coverage Report
Increment line coverage `0.00% (0/79)` :tada:
[Increment coverage report](http://coverage.selectdb-in.cc/coverage/d84fcddbe4aa9142e912da69ab6bafe20075ca2e_d84fcddbe4aa9142e912da69ab6bafe20075ca2e/increment_report/index.html)
[Complete coverage report](http://coverage.selectdb-in.cc/coverage/d84fcddbe4aa9142e912da69ab6bafe20075ca2e_d84fcddbe4aa9142e912da69ab6bafe20075ca2e/report/index.html)
| Category | Coverage |
|-------------------|--------------------|
| Function Coverage | 52.41% (19093/36431) |
| Line Coverage | 35.77% (177218/495488) |
| Region Coverage | 32.25% (137009/424848) |
| Branch Coverage | 33.17% (59277/178717) |
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [Feature](tvf) Support cdc stream tvf for mysql and pg [doris]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #60116:
URL: https://github.com/apache/doris/pull/60116#discussion_r2979959428
##########
fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/config/GlobalExceptionHandler.java:
##########
@@ -35,4 +39,17 @@ public Object exceptionHandler(HttpServletRequest request, Exception e) {
log.error("Unexpected exception", e);
return RestResponse.internalError(e.getMessage());
}
+
+ @ExceptionHandler(StreamException.class)
+ public Object streamExceptionHandler(StreamException e) {
+ // Directly throwing an exception allows curl to detect anomalies in the streaming response.
+ log.error("Exception in streaming response, re-throwing to client", e);
+ throw e;
+ }
+
+ @ExceptionHandler(CommonException.class)
+ public Object commonExceptionHandler(CommonException e) {
+ log.error("Unexpected common exception", e);
+ return ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR).body(e.getMessage());
Review Comment:
`CommonException` is handled differently from the existing `Exception` handler: it returns a plain `ResponseEntity<String>` instead of the standard `RestResponse` error envelope used by other endpoints. This makes error responses inconsistent for clients and complicates parsing. Consider returning `RestResponse.internalError(...)` (optionally with an HTTP 500 status) for `CommonException` as well, or otherwise document/standardize the response format differences for streaming vs non-streaming endpoints.
```suggestion
return RestResponse.internalError(e.getMessage());
```
##########
fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/service/PipelineCoordinator.java:
##########
@@ -90,6 +102,156 @@ public PipelineCoordinator() {
new ThreadPoolExecutor.AbortPolicy());
}
+ /** return data for http_file_reader */
+ public StreamingResponseBody fetchRecordStream(FetchRecordRequest fetchReq) throws Exception {
+ SourceReader sourceReader;
+ SplitReadResult readResult;
+ try {
+ if (fetchReq.getTaskId() == null && fetchReq.getMeta() == null) {
+ LOG.info(
+ "Generate initial meta for fetch record request, jobId={}, taskId={}",
+ fetchReq.getJobId(),
+ fetchReq.getTaskId());
+ // means the request did not originate from the job, only tvf
+ Map<String, Object> meta = generateMeta(fetchReq.getConfig());
+ fetchReq.setMeta(meta);
+ }
+
+ sourceReader = Env.getCurrentEnv().getReader(fetchReq);
+ readResult = sourceReader.prepareAndSubmitSplit(fetchReq);
+ } catch (Exception ex) {
+ throw new CommonException(ex);
+ }
+
+ return outputStream -> {
+ try {
+ buildStreamRecords(sourceReader, fetchReq, readResult, outputStream);
+ } catch (Exception ex) {
+ LOG.error(
+ "Failed fetch record, jobId={}, taskId={}",
+ fetchReq.getJobId(),
+ fetchReq.getTaskId(),
+ ex);
+ throw new StreamException(ex);
+ }
+ };
+ }
+
+ private void buildStreamRecords(
+ SourceReader sourceReader,
+ FetchRecordRequest fetchRecord,
+ SplitReadResult readResult,
+ OutputStream rawOutputStream)
+ throws Exception {
+ SourceSplit split = readResult.getSplit();
+ Map<String, String> lastMeta = null;
+ int rowCount = 0;
+ BufferedOutputStream bos = new BufferedOutputStream(rawOutputStream);
+ try {
+ // Poll records using the existing mechanism
+ boolean shouldStop = false;
+ long startTime = System.currentTimeMillis();
+ while (!shouldStop) {
+ Iterator<SourceRecord> recordIterator = sourceReader.pollRecords();
+ if (!recordIterator.hasNext()) {
+ Thread.sleep(100);
+ long elapsedTime = System.currentTimeMillis() - startTime;
+ if (elapsedTime > Constants.POLL_SPLIT_RECORDS_TIMEOUTS) {
+ break;
+ }
+ continue;
+ }
+ while (recordIterator.hasNext()) {
+ SourceRecord element = recordIterator.next();
+ if (isHeartbeatEvent(element)) {
+ shouldStop = true;
+ break;
+ }
+ DeserializeResult result =
+ sourceReader.deserialize(fetchRecord.getConfig(), element);
+ if (!CollectionUtils.isEmpty(result.getRecords())) {
+ for (String record : result.getRecords()) {
+ bos.write(record.getBytes(StandardCharsets.UTF_8));
+ bos.write(LINE_DELIMITER);
+ }
+ rowCount += result.getRecords().size();
+ }
+ }
+ }
+ // force flush buffer
+ bos.flush();
+ } finally {
+ // Commit offset and cleanup
+ sourceReader.commitSourceOffset(fetchRecord.getJobId(), readResult.getSplit());
+ sourceReader.finishSplitRecords();
+ }
+
+ LOG.info(
+ "Fetch records completed, jobId={}, taskId={}, splitId={}, rowCount={}",
+ fetchRecord.getJobId(),
+ fetchRecord.getTaskId(),
+ split.splitId(),
+ rowCount);
+
+ if (readResult.getSplitState() != null) {
+ if (sourceReader.isSnapshotSplit(split)) {
+ lastMeta = sourceReader.extractSnapshotStateOffset(readResult.getSplitState());
+ lastMeta.put(SPLIT_ID, split.splitId());
+ } else if (sourceReader.isBinlogSplit(split)) {
+ lastMeta = sourceReader.extractBinlogStateOffset(readResult.getSplitState());
+ lastMeta.put(SPLIT_ID, BinlogSplit.BINLOG_SPLIT_ID);
+ } else {
+ throw new RuntimeException(
+ "unknown split type: " + split.getClass().getSimpleName());
+ }
+ } else {
+ throw new RuntimeException("split state is null");
+ }
+
+ if (StringUtils.isNotEmpty(fetchRecord.getTaskId())) {
+ taskOffsetCache.put(fetchRecord.getTaskId(), lastMeta);
+ }
+
+ if (!isLong(fetchRecord.getJobId())) {
+ // TVF requires closing the window after each execution,
+ // while PG requires dropping the slot.
+ sourceReader.close(fetchRecord);
Review Comment:
For TVF requests (non-numeric `jobId`), `sourceReader.close(fetchRecord)` is called but the `Env` job context is never removed. Since TVF generates a new random jobId per execution, this will leak entries in `Env.jobContexts/jobLocks` over time. After closing the reader for TVF, also call `Env.getCurrentEnv().close(fetchRecord.getJobId())` (and consider also cleaning any related stream-load state if applicable).
```suggestion
sourceReader.close(fetchRecord);
Env.getCurrentEnv().close(fetchRecord.getJobId());
```
##########
be/src/io/fs/http_file_reader.cpp:
##########
@@ -96,32 +111,64 @@ Status HttpFileReader::open(const FileReaderOptions& opts) {
return Status::OK();
}
- // Step 1: HEAD request to get file metadata
- RETURN_IF_ERROR(prepare_client(/*set_fail_on_error=*/true));
- _client->set_method(HttpMethod::HEAD);
- RETURN_IF_ERROR(_client->execute());
+ // start CDC client
+ auto enable_cdc_iter = _extend_kv.find("enable_cdc_client");
+ if (enable_cdc_iter != _extend_kv.end() && enable_cdc_iter->second == "true") {
+ LOG(INFO) << "CDC client is enabled, starting CDC client for " << _url;
+ ExecEnv* env = ExecEnv::GetInstance();
+ if (env == nullptr || env->cdc_client_mgr() == nullptr) {
+ return Status::InternalError("ExecEnv or CdcClientMgr is not initialized");
+ }
- uint64_t content_length = 0;
- RETURN_IF_ERROR(_client->get_content_length(&content_length));
+ PRequestCdcClientResult result;
+ Status start_st = env->cdc_client_mgr()->start_cdc_client(&result);
+ if (!start_st.ok()) {
+ LOG(ERROR) << "Failed to start CDC client, status=" << start_st.to_string();
+ return start_st;
+ }
- _file_size = content_length;
- _size_known = true;
+ // Replace CDC_CLIENT_PORT placeholder with actual CDC client port
+ const std::string placeholder = "CDC_CLIENT_PORT";
+ size_t pos = _url.find(placeholder);
+ if (pos != std::string::npos) {
+ _url.replace(pos, placeholder.size(), std::to_string(doris::config::cdc_client_port));
+ }
+ LOG(INFO) << "CDC client started successfully for " << _url;
+ }
Review Comment:
`open()` returns immediately when `_initialized` is true (e.g. when FE provided a `file_size`). With CDC-stream TVF, FE provides `file_size`, so this early return will skip the new CDC-client startup and `CDC_CLIENT_PORT` placeholder replacement logic, leaving `_url` invalid and the CDC client potentially not started. Move the CDC-client initialization / placeholder replacement before the `_initialized` short-circuit (or refine what `_initialized` means) so this setup always runs when `enable_cdc_client=true`.
##########
fe/fe-core/src/main/java/org/apache/doris/tablefunction/CdcStreamTableValuedFunction.java:
##########
@@ -0,0 +1,130 @@
+// 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.tablefunction;
+
+import org.apache.doris.analysis.BrokerDesc;
+import org.apache.doris.analysis.StorageBackend.StorageType;
+import org.apache.doris.catalog.Column;
+import org.apache.doris.common.AnalysisException;
+import org.apache.doris.datasource.jdbc.client.JdbcClient;
+import org.apache.doris.job.cdc.DataSourceConfigKeys;
+import org.apache.doris.job.cdc.request.FetchRecordRequest;
+import org.apache.doris.job.common.DataSourceType;
+import org.apache.doris.job.util.StreamingJobUtils;
+import org.apache.doris.thrift.TBrokerFileStatus;
+import org.apache.doris.thrift.TFileType;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.base.Preconditions;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+
+public class CdcStreamTableValuedFunction extends ExternalFileTableValuedFunction {
+ private static final ObjectMapper objectMapper = new ObjectMapper();
+ private static final String URI = "http://127.0.0.1:CDC_CLIENT_PORT/api/fetchRecordStream";
+ private final Map<String, String> originProps;
+
+ public CdcStreamTableValuedFunction(Map<String, String> properties) throws AnalysisException {
+ this.originProps = properties;
+ validate(properties);
+ processProps(properties);
+ }
+
+ private void processProps(Map<String, String> properties) throws AnalysisException {
+ Map<String, String> copyProps = new HashMap<>(properties);
+ copyProps.put("format", "json");
+ super.parseCommonProperties(copyProps);
+ this.processedParams.put("enable_cdc_client", "true");
+ this.processedParams.put("uri", URI);
+ this.processedParams.put("http.enable.range.request", "false");
+ this.processedParams.put("http.enable.chunk.response", "true");
+ this.processedParams.put("http.method", "POST");
+
+ String payload = generateParams(properties);
+ this.processedParams.put("http.payload", payload);
+ this.backendConnectProperties.putAll(processedParams);
+ generateFileStatus();
+ }
+
+ private String generateParams(Map<String, String> properties) throws AnalysisException {
+ FetchRecordRequest recordRequest = new FetchRecordRequest();
+ recordRequest.setJobId(UUID.randomUUID().toString().replace("-", ""));
+ recordRequest.setDataSource(properties.get(DataSourceConfigKeys.TYPE));
+ recordRequest.setConfig(properties);
+ try {
+ return objectMapper.writeValueAsString(recordRequest);
+ } catch (IOException e) {
+ LOG.info("Failed to serialize fetch record request," + e.getMessage());
+ throw new AnalysisException(e.getMessage());
+ }
+ }
+
+ private void validate(Map<String, String> properties) {
+ Preconditions.checkArgument(properties.containsKey(DataSourceConfigKeys.JDBC_URL), "jdbc_url is required");
+ Preconditions.checkArgument(properties.containsKey(DataSourceConfigKeys.TYPE), "type is required");
+ Preconditions.checkArgument(properties.containsKey(DataSourceConfigKeys.TABLE), "table is required");
+ Preconditions.checkArgument(properties.containsKey(DataSourceConfigKeys.OFFSET), "offset is required");
+ }
+
+ private void generateFileStatus() {
+ this.fileStatuses.clear();
+ this.fileStatuses.add(new TBrokerFileStatus(URI, false, Integer.MAX_VALUE, false));
Review Comment:
`generateFileStatus()` hard-codes the file size to `Integer.MAX_VALUE`. On the BE side `HttpFileReader` treats a provided `file_size` as already-initialized metadata and returns early from `open()`, which means the CDC-client startup + `CDC_CLIENT_PORT` placeholder replacement logic will be skipped and the request URL may never become valid. Consider using an “unknown size” sentinel (or avoiding setting `file_size` at all) so BE `open()` still runs its CDC-client initialization, or adjust BE initialization logic to run regardless of `file_size`.
```suggestion
// Use unknown size sentinel (-1) so BE HttpFileReader still runs its initialization logic.
this.fileStatuses.add(new TBrokerFileStatus(URI, false, -1, false));
```
##########
fe/fe-core/src/main/java/org/apache/doris/tablefunction/CdcStreamTableValuedFunction.java:
##########
@@ -0,0 +1,130 @@
+// 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.tablefunction;
+
+import org.apache.doris.analysis.BrokerDesc;
+import org.apache.doris.analysis.StorageBackend.StorageType;
+import org.apache.doris.catalog.Column;
+import org.apache.doris.common.AnalysisException;
+import org.apache.doris.datasource.jdbc.client.JdbcClient;
+import org.apache.doris.job.cdc.DataSourceConfigKeys;
+import org.apache.doris.job.cdc.request.FetchRecordRequest;
+import org.apache.doris.job.common.DataSourceType;
+import org.apache.doris.job.util.StreamingJobUtils;
+import org.apache.doris.thrift.TBrokerFileStatus;
+import org.apache.doris.thrift.TFileType;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.base.Preconditions;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+
+public class CdcStreamTableValuedFunction extends ExternalFileTableValuedFunction {
+ private static final ObjectMapper objectMapper = new ObjectMapper();
+ private static final String URI = "http://127.0.0.1:CDC_CLIENT_PORT/api/fetchRecordStream";
+ private final Map<String, String> originProps;
+
+ public CdcStreamTableValuedFunction(Map<String, String> properties) throws AnalysisException {
+ this.originProps = properties;
+ validate(properties);
+ processProps(properties);
+ }
+
+ private void processProps(Map<String, String> properties) throws AnalysisException {
+ Map<String, String> copyProps = new HashMap<>(properties);
+ copyProps.put("format", "json");
+ super.parseCommonProperties(copyProps);
+ this.processedParams.put("enable_cdc_client", "true");
+ this.processedParams.put("uri", URI);
+ this.processedParams.put("http.enable.range.request", "false");
+ this.processedParams.put("http.enable.chunk.response", "true");
+ this.processedParams.put("http.method", "POST");
+
+ String payload = generateParams(properties);
+ this.processedParams.put("http.payload", payload);
+ this.backendConnectProperties.putAll(processedParams);
+ generateFileStatus();
+ }
+
+ private String generateParams(Map<String, String> properties) throws AnalysisException {
+ FetchRecordRequest recordRequest = new FetchRecordRequest();
+ recordRequest.setJobId(UUID.randomUUID().toString().replace("-", ""));
+ recordRequest.setDataSource(properties.get(DataSourceConfigKeys.TYPE));
+ recordRequest.setConfig(properties);
+ try {
+ return objectMapper.writeValueAsString(recordRequest);
+ } catch (IOException e) {
+ LOG.info("Failed to serialize fetch record request," + e.getMessage());
+ throw new AnalysisException(e.getMessage());
+ }
+ }
+
+ private void validate(Map<String, String> properties) {
+ Preconditions.checkArgument(properties.containsKey(DataSourceConfigKeys.JDBC_URL), "jdbc_url is required");
+ Preconditions.checkArgument(properties.containsKey(DataSourceConfigKeys.TYPE), "type is required");
+ Preconditions.checkArgument(properties.containsKey(DataSourceConfigKeys.TABLE), "table is required");
+ Preconditions.checkArgument(properties.containsKey(DataSourceConfigKeys.OFFSET), "offset is required");
+ }
+
+ private void generateFileStatus() {
+ this.fileStatuses.clear();
+ this.fileStatuses.add(new TBrokerFileStatus(URI, false, Integer.MAX_VALUE, false));
+ }
+
+ @Override
+ public List<Column> getTableColumns() throws AnalysisException {
+ DataSourceType dataSourceType =
+ DataSourceType.valueOf(processedParams.get(DataSourceConfigKeys.TYPE).toUpperCase());
+ JdbcClient jdbcClient = StreamingJobUtils.getJdbcClient(dataSourceType, processedParams);
+ String database = StreamingJobUtils.getRemoteDbName(dataSourceType, processedParams);
+ String table = processedParams.get(DataSourceConfigKeys.TABLE);
+ boolean tableExist = jdbcClient.isTableExist(database, table);
+ Preconditions.checkArgument(tableExist, "Table does not exist: " + table);
+ return jdbcClient.getColumnsFromJdbc(database, table);
+ }
+
+ @Override
+ public TFileType getTFileType() {
+ return TFileType.FILE_HTTP;
+ }
+
+ @Override
+ public String getFilePath() {
+ return URI;
+ }
+
+ @Override
+ public BrokerDesc getBrokerDesc() {
+ return new BrokerDesc("CdcStreamTvfBroker", StorageType.HTTP, originProps);
Review Comment:
`getBrokerDesc()` is built from `originProps`, which does not include the computed HTTP properties (e.g. `uri`, `http.method`, `http.payload`, `enable_cdc_client`). This will prevent the BE HTTP reader from receiving the necessary request configuration and can also cause `StorageProperties.createPrimary(...)` to fail because `fs.http.support=true` is not set. Build the `BrokerDesc` from the processed/derived properties (and set `fs.http.support` like `HttpTableValuedFunction` does) so HTTP storage is detected correctly and the BE sees the TVF-specific HTTP options.
```suggestion
Map<String, String> brokerProps = new HashMap<>(processedParams);
brokerProps.put("fs.http.support", "true");
return new BrokerDesc("CdcStreamTvfBroker", StorageType.HTTP, brokerProps);
```
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [Feature](tvf) Support cdc stream tvf for mysql and pg [doris]
Posted by "JNSimba (via GitHub)" <gi...@apache.org>.
JNSimba commented on PR #60116:
URL: https://github.com/apache/doris/pull/60116#issuecomment-4116382550
/review
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [Feature](tvf) Support cdc stream tvf for mysql and pg [doris]
Posted by "JNSimba (via GitHub)" <gi...@apache.org>.
JNSimba commented on PR #60116:
URL: https://github.com/apache/doris/pull/60116#issuecomment-4124317361
run buildall
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [Feature](tvf) Support cdc stream tvf for mysql and pg [doris]
Posted by "JNSimba (via GitHub)" <gi...@apache.org>.
JNSimba commented on PR #60116:
URL: https://github.com/apache/doris/pull/60116#issuecomment-4125408850
run cloud_p0
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [Feature](tvf) Support cdc stream tvf for mysql and pg [doris]
Posted by "JNSimba (via GitHub)" <gi...@apache.org>.
JNSimba commented on PR #60116:
URL: https://github.com/apache/doris/pull/60116#issuecomment-4125834330
run cloud_p0
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [Feature](tvf) Support cdc stream tvf for mysql and pg [doris]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #60116:
URL: https://github.com/apache/doris/pull/60116#issuecomment-3815897579
PR approved by at least one committer and no changes requested.
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org