You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@doris.apache.org by GitBox <gi...@apache.org> on 2022/11/03 15:56:29 UTC
[GitHub] [doris] BePPPower opened a new pull request, #13959: [feature](table-valued-function) Support S3 tvf
BePPPower opened a new pull request, #13959:
URL: https://github.com/apache/doris/pull/13959
# Proposed changes
Issue Number: close #10217
pr Number: close #10214
This pr does three things：
1. Modified the framework of table-valued-function(tvf).
2. be support `fetch_table_schema` rpc.
3. Implemented `S3(path, AK, SK, format)` table-valued-function.
## Problem summary
1. Modified the framework of table-valued-function(tvf).
Each implementation of tvf requires a new `scan node` in the old framework of pr #10214 . Actually, different `scan nodes` should correspond to different data sources. I modified the tvf framework so that it can generate different `scan node` according to data source of this tvf.
For example, the data of `numbers(n,m)` tvf is derived from memory, so it generate `DataGenScanNode`. And the data of `s3(...)` tvf is derived from S3 Storage, so it generate `ExternalFileScanNode`.
2. be support `fetch_table_schema` rpc.
There is no table schema in `s3()` advance, so we need `fetch_table_schema` rpc to parse table schema from file in S3 Storage.
3. `S3(path, AK, SK, format)`
The initial implement of `s3(..)` tvf, use like:
```
MySQL [(none)]> select * from s3("http://127.0.0.1:9312/test2/student1.csv", "minioadmin", "minioadmin", "csv");
+------+------+------+
| c1 | c2 | c3 |
+------+------+------+
| 1 | ftw | 12 |
| 2 | zs | 18 |
| 3 | ww | 20 |
+------+------+------+
3 rows in set (0.04 sec)
MySQL [(none)]> select * from s3("http://127.0.0.1:9312/test2/student_with_names.csv", "minioadmin", "minioadmin", "csv_with_names");
+------+------+------+
| id | name | age |
+------+------+------+
| 1 | ftw | 12 |
| 2 | zs | 18 |
| 3 | ww | 20 |
+------+------+------+
3 rows in set (0.04 sec)
```
## Checklist(Required)
1. Does it affect the original behavior:
- [x] Yes
- [ ] No
- [ ] I don't know
5. Has unit tests been added:
- [ ] Yes
- [x] No
- [ ] No Need
6. Has document been added or modified:
- [ ] Yes
- [x] No
- [ ] No Need
7. Does it need to update dependencies:
- [ ] Yes
- [x] No
8. Are there any changes that cannot be rolled back:
- [ ] Yes (If Yes, please explain WHY)
- [x] No
## Further comments
If this is a relatively large or complex change, kick off the discussion at [dev@doris.apache.org](mailto:dev@doris.apache.org) by explaining why you chose the solution you did and what alternatives you considered, etc...
--
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
[GitHub] [doris] morningman commented on pull request #13959: [feature](table-valued-function) Support S3 tvf
Posted by GitBox <gi...@apache.org>.
morningman commented on PR #13959:
URL: https://github.com/apache/doris/pull/13959#issuecomment-1303174238
Need to add some regression test in next PR
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: 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
[GitHub] [doris] BePPPower commented on a diff in pull request #13959: [feature](table-valued-function) Support S3 tvf
Posted by GitBox <gi...@apache.org>.
BePPPower commented on code in PR #13959:
URL: https://github.com/apache/doris/pull/13959#discussion_r1013716581
##########
be/src/service/internal_service.cpp:
##########
@@ -407,6 +410,78 @@ void PInternalServiceImpl::fetch_data(google::protobuf::RpcController* cntl_base
_exec_env->result_mgr()->fetch_data(request->finst_id(), ctx);
}
+void PInternalServiceImpl::fetch_table_schema(google::protobuf::RpcController* controller,
+ const PFetchTableSchemaRequest* request,
+ PFetchTableSchemaResult* result,
+ google::protobuf::Closure* done) {
+ VLOG_RPC << "fetch table schema";
+ brpc::ClosureGuard closure_guard(done);
+ TFileScanRange file_scan_range;
+ Status st = Status::OK();
+ {
+ const uint8_t* buf = (const uint8_t*)(request->file_scan_range().data());
+ uint32_t len = request->file_scan_range().size();
+ st = deserialize_thrift_msg(buf, &len, false, &file_scan_range);
+ if (!st.ok()) {
+ LOG(WARNING) << "fetch table schema failed, errmsg=" << st.get_error_msg();
+ st.to_protobuf(result->mutable_status());
+ return;
+ }
+ }
+
+ if (file_scan_range.__isset.ranges == false) {
+ st = Status::InternalError("can not get TFileRangeDesc.");
+ st.to_protobuf(result->mutable_status());
+ return;
+ }
+ if (file_scan_range.__isset.params == false) {
+ st = Status::InternalError("can not get TFileScanRangeParams.");
+ st.to_protobuf(result->mutable_status());
+ return;
+ }
+ const TFileRangeDesc& range = file_scan_range.ranges.at(0);
+ const TFileScanRangeParams& params = file_scan_range.params;
+ // file_slots is no use
+ std::vector<SlotDescriptor*> file_slots;
+ std::unique_ptr<vectorized::GenericReader> reader(nullptr);
+ std::unique_ptr<RuntimeProfile> profile(new RuntimeProfile("FetchTableSchema"));
+ switch (params.format_type) {
+ case TFileFormatType::FORMAT_CSV_PLAIN:
+ case TFileFormatType::FORMAT_CSV_GZ:
+ case TFileFormatType::FORMAT_CSV_BZ2:
+ case TFileFormatType::FORMAT_CSV_LZ4FRAME:
+ case TFileFormatType::FORMAT_CSV_LZOP:
+ case TFileFormatType::FORMAT_CSV_DEFLATE: {
Review Comment:
Parquet and ORC will be added after implementing `get_parsered_schema` method of ParquetReader and OrcReader.
--
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
[GitHub] [doris] morningman commented on a diff in pull request #13959: [feature](table-valued-function) Support S3 tvf
Posted by GitBox <gi...@apache.org>.
morningman commented on code in PR #13959:
URL: https://github.com/apache/doris/pull/13959#discussion_r1013701332
##########
fe/fe-core/src/main/java/org/apache/doris/planner/DataGenScanNode.java:
##########
@@ -40,15 +40,15 @@
/**
* This scan node is used for table valued function.
Review Comment:
Change the comment
##########
be/src/service/internal_service.cpp:
##########
@@ -407,6 +410,78 @@ void PInternalServiceImpl::fetch_data(google::protobuf::RpcController* cntl_base
_exec_env->result_mgr()->fetch_data(request->finst_id(), ctx);
}
+void PInternalServiceImpl::fetch_table_schema(google::protobuf::RpcController* controller,
+ const PFetchTableSchemaRequest* request,
+ PFetchTableSchemaResult* result,
+ google::protobuf::Closure* done) {
+ VLOG_RPC << "fetch table schema";
+ brpc::ClosureGuard closure_guard(done);
+ TFileScanRange file_scan_range;
+ Status st = Status::OK();
+ {
+ const uint8_t* buf = (const uint8_t*)(request->file_scan_range().data());
+ uint32_t len = request->file_scan_range().size();
+ st = deserialize_thrift_msg(buf, &len, false, &file_scan_range);
+ if (!st.ok()) {
+ LOG(WARNING) << "fetch table schema failed, errmsg=" << st.get_error_msg();
+ st.to_protobuf(result->mutable_status());
+ return;
+ }
+ }
+
+ if (file_scan_range.__isset.ranges == false) {
+ st = Status::InternalError("can not get TFileRangeDesc.");
+ st.to_protobuf(result->mutable_status());
+ return;
+ }
+ if (file_scan_range.__isset.params == false) {
+ st = Status::InternalError("can not get TFileScanRangeParams.");
+ st.to_protobuf(result->mutable_status());
+ return;
+ }
+ const TFileRangeDesc& range = file_scan_range.ranges.at(0);
+ const TFileScanRangeParams& params = file_scan_range.params;
+ // file_slots is no use
+ std::vector<SlotDescriptor*> file_slots;
+ std::unique_ptr<vectorized::GenericReader> reader(nullptr);
+ std::unique_ptr<RuntimeProfile> profile(new RuntimeProfile("FetchTableSchema"));
+ switch (params.format_type) {
+ case TFileFormatType::FORMAT_CSV_PLAIN:
+ case TFileFormatType::FORMAT_CSV_GZ:
+ case TFileFormatType::FORMAT_CSV_BZ2:
+ case TFileFormatType::FORMAT_CSV_LZ4FRAME:
+ case TFileFormatType::FORMAT_CSV_LZOP:
+ case TFileFormatType::FORMAT_CSV_DEFLATE: {
+ reader.reset(new vectorized::CsvReader(profile.get(), params, range, file_slots));
+ break;
+ }
+ default:
+ st = Status::InternalError("Not supported file format in fetch table schema: {}",
+ params.format_type);
+ st.to_protobuf(result->mutable_status());
+ return;
+ }
+ std::unordered_map<std::string, TypeDescriptor> name_to_col_type;
+ std::vector<std::string> col_names;
+ std::vector<TypeDescriptor> col_types;
+ st = reader->get_parsered_schema(&col_names, &col_types);
+ if (!st.ok()) {
+ LOG(WARNING) << "fetch table schema failed, errmsg=" << st.get_error_msg();
+ st.to_protobuf(result->mutable_status());
+ return;
+ }
+ result->set_column_nums(col_names.size());
+ for (size_t idx = 0; idx < col_names.size(); ++idx) {
+ result->add_column_names(col_names[idx]);
+ }
+ for (size_t idx = 0; idx < col_types.size(); ++idx) {
+ PTypeDesc* type_desc = result->add_column_types();
+ col_types[idx].to_protobuf(type_desc);
+ }
+ LOG(INFO) << "complete parse, status: " << st;
Review Comment:
Remove this log, meaningless
##########
be/src/service/internal_service.cpp:
##########
@@ -407,6 +410,78 @@ void PInternalServiceImpl::fetch_data(google::protobuf::RpcController* cntl_base
_exec_env->result_mgr()->fetch_data(request->finst_id(), ctx);
}
+void PInternalServiceImpl::fetch_table_schema(google::protobuf::RpcController* controller,
+ const PFetchTableSchemaRequest* request,
+ PFetchTableSchemaResult* result,
+ google::protobuf::Closure* done) {
+ VLOG_RPC << "fetch table schema";
+ brpc::ClosureGuard closure_guard(done);
+ TFileScanRange file_scan_range;
+ Status st = Status::OK();
+ {
+ const uint8_t* buf = (const uint8_t*)(request->file_scan_range().data());
+ uint32_t len = request->file_scan_range().size();
+ st = deserialize_thrift_msg(buf, &len, false, &file_scan_range);
+ if (!st.ok()) {
+ LOG(WARNING) << "fetch table schema failed, errmsg=" << st.get_error_msg();
+ st.to_protobuf(result->mutable_status());
+ return;
+ }
+ }
+
+ if (file_scan_range.__isset.ranges == false) {
+ st = Status::InternalError("can not get TFileRangeDesc.");
+ st.to_protobuf(result->mutable_status());
+ return;
+ }
+ if (file_scan_range.__isset.params == false) {
+ st = Status::InternalError("can not get TFileScanRangeParams.");
+ st.to_protobuf(result->mutable_status());
+ return;
+ }
+ const TFileRangeDesc& range = file_scan_range.ranges.at(0);
+ const TFileScanRangeParams& params = file_scan_range.params;
+ // file_slots is no use
+ std::vector<SlotDescriptor*> file_slots;
+ std::unique_ptr<vectorized::GenericReader> reader(nullptr);
+ std::unique_ptr<RuntimeProfile> profile(new RuntimeProfile("FetchTableSchema"));
+ switch (params.format_type) {
+ case TFileFormatType::FORMAT_CSV_PLAIN:
+ case TFileFormatType::FORMAT_CSV_GZ:
+ case TFileFormatType::FORMAT_CSV_BZ2:
+ case TFileFormatType::FORMAT_CSV_LZ4FRAME:
+ case TFileFormatType::FORMAT_CSV_LZOP:
+ case TFileFormatType::FORMAT_CSV_DEFLATE: {
Review Comment:
Missing Parquet and ORC?
##########
fe/fe-core/src/main/java/org/apache/doris/planner/external/ExternalFileScanNode.java:
##########
@@ -202,6 +187,39 @@ public void init(Analyzer analyzer) throws UserException {
initParamCreateContexts(analyzer);
}
+ private void initHMSExternalTable(HMSExternalTable hmsTable) throws UserException {
+ Preconditions.checkNotNull(hmsTable);
+
+ if (hmsTable.isView()) {
+ throw new AnalysisException(
+ String.format("Querying external view '[%s].%s.%s' is not supported", hmsTable.getDlaType(),
+ hmsTable.getDbName(), hmsTable.getName()));
+ }
+
+ FileScanProviderIf scanProvider;
+ switch (hmsTable.getDlaType()) {
+ case HUDI:
+ scanProvider = new HudiScanProvider(hmsTable, desc);
+ break;
+ case ICEBERG:
+ scanProvider = new IcebergScanProvider(hmsTable, desc);
+ break;
+ case HIVE:
+ scanProvider = new HiveScanProvider(hmsTable, desc);
+ break;
+ default:
+ throw new UserException("Unknown table type: " + hmsTable.getDlaType());
+ }
+ this.scanProviders.add(scanProvider);
+ }
+
+ private void initFunctionGenTable(FunctionGenTable table, ExternalFileTableValuedFunction tvf) {
+ Preconditions.checkNotNull(table);
+ FileScanProviderIf scanProvider;
Review Comment:
```suggestion
FileScanProviderIf scanProvider = new TVFScanProvider(table, desc, tvf);;
```
--
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
[GitHub] [doris] github-actions[bot] commented on pull request #13959: [feature](table-valued-function) Support S3 tvf
Posted by GitBox <gi...@apache.org>.
github-actions[bot] commented on PR #13959:
URL: https://github.com/apache/doris/pull/13959#issuecomment-1304557141
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
[GitHub] [doris] morningman merged pull request #13959: [feature](table-valued-function) Support S3 tvf
Posted by GitBox <gi...@apache.org>.
morningman merged PR #13959:
URL: https://github.com/apache/doris/pull/13959
--
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
[GitHub] [doris] hello-stephen commented on pull request #13959: [feature](table-valued-function) Support S3 tvf
Posted by GitBox <gi...@apache.org>.
hello-stephen commented on PR #13959:
URL: https://github.com/apache/doris/pull/13959#issuecomment-1303236731
TeamCity pipeline, clickbench performance test result:
the sum of best hot time: 34.44 seconds
load time: 432 seconds
storage size: 17154655363 Bytes
https://doris-community-test-1308700295.cos.ap-hongkong.myqcloud.com/tmp/20221104103500_clickbench_pr_40058.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
[GitHub] [doris] github-actions[bot] commented on pull request #13959: [feature](table-valued-function) Support S3 tvf
Posted by GitBox <gi...@apache.org>.
github-actions[bot] commented on PR #13959:
URL: https://github.com/apache/doris/pull/13959#issuecomment-1304557133
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