You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@doris.apache.org by "wuwenchi (via GitHub)" <gi...@apache.org> on 2023/11/09 12:40:21 UTC
[PR] [feture](stream load)Add arrow data type for stream load [doris]
wuwenchi opened a new pull request, #26709:
URL: https://github.com/apache/doris/pull/26709
## Proposed changes
By using the Arrow data format, we can reduce the streamload of data transferred and improve the data import performance
## 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
Re: [PR] [feture](stream load) (step one)Add arrow data type for stream load [doris]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on code in PR #26709:
URL: https://github.com/apache/doris/pull/26709#discussion_r1404709568
##########
be/src/vec/data_types/serde/data_type_datetimev2_serde.cpp:
##########
@@ -19,11 +19,17 @@
#include <arrow/builder.h>
+#include <chrono> // IWYU pragma: keep
#include <type_traits>
#include "vec/columns/column_const.h"
#include "vec/io/io_helper.h"
+#define DIVISOR_FOR_SECOND 1
Review Comment:
warning: macro 'DIVISOR_FOR_SECOND' defines an integral constant; prefer an enum instead [modernize-macro-to-enum]
```cpp
#define DIVISOR_FOR_SECOND 1
^
```
##########
be/src/vec/data_types/serde/data_type_datetimev2_serde.cpp:
##########
@@ -19,11 +19,17 @@
#include <arrow/builder.h>
+#include <chrono> // IWYU pragma: keep
#include <type_traits>
#include "vec/columns/column_const.h"
#include "vec/io/io_helper.h"
+#define DIVISOR_FOR_SECOND 1
+#define DIVISOR_FOR_MILLI 1000
+#define DIVISOR_FOR_MICRO 1000000
+#define DIVISOR_FOR_NANO 1000000000
Review Comment:
warning: macro 'DIVISOR_FOR_NANO' defines an integral constant; prefer an enum instead [modernize-macro-to-enum]
```cpp
#define DIVISOR_FOR_NANO 1000000000
^
```
##########
be/src/vec/data_types/serde/data_type_datetimev2_serde.cpp:
##########
@@ -112,6 +118,52 @@
}
}
+void DataTypeDateTimeV2SerDe::read_column_from_arrow(IColumn& column,
Review Comment:
warning: method 'read_column_from_arrow' can be made static [readability-convert-member-functions-to-static]
```suggestion
static void DataTypeDateTimeV2SerDe::read_column_from_arrow(IColumn& column,
```
be/src/vec/data_types/serde/data_type_datetimev2_serde.cpp:122:
```diff
- int end, const cctz::time_zone& ctz) const {
+ int end, const cctz::time_zone& ctz) {
```
##########
be/src/vec/data_types/serde/data_type_datetimev2_serde.cpp:
##########
@@ -19,11 +19,17 @@
#include <arrow/builder.h>
+#include <chrono> // IWYU pragma: keep
#include <type_traits>
#include "vec/columns/column_const.h"
#include "vec/io/io_helper.h"
+#define DIVISOR_FOR_SECOND 1
+#define DIVISOR_FOR_MILLI 1000
Review Comment:
warning: macro 'DIVISOR_FOR_MILLI' defines an integral constant; prefer an enum instead [modernize-macro-to-enum]
```cpp
#define DIVISOR_FOR_MILLI 1000
^
```
##########
be/src/vec/data_types/serde/data_type_datetimev2_serde.cpp:
##########
@@ -19,11 +19,17 @@
#include <arrow/builder.h>
+#include <chrono> // IWYU pragma: keep
#include <type_traits>
#include "vec/columns/column_const.h"
#include "vec/io/io_helper.h"
+#define DIVISOR_FOR_SECOND 1
+#define DIVISOR_FOR_MILLI 1000
+#define DIVISOR_FOR_MICRO 1000000
Review Comment:
warning: macro 'DIVISOR_FOR_MICRO' defines an integral constant; prefer an enum instead [modernize-macro-to-enum]
```cpp
#define DIVISOR_FOR_MICRO 1000000
^
```
##########
be/src/vec/data_types/serde/data_type_datetimev2_serde.cpp:
##########
@@ -19,11 +19,17 @@
#include <arrow/builder.h>
+#include <chrono> // IWYU pragma: keep
#include <type_traits>
#include "vec/columns/column_const.h"
#include "vec/io/io_helper.h"
+#define DIVISOR_FOR_SECOND 1
+#define DIVISOR_FOR_MILLI 1000
+#define DIVISOR_FOR_MICRO 1000000
+#define DIVISOR_FOR_NANO 1000000000
+
Review Comment:
warning: replace macro with enum [modernize-macro-to-enum]
```suggestion
enum {
DIVISOR_FOR_SECOND = 1,
DIVISOR_FOR_MILLI = 1000,
DIVISOR_FOR_MICRO = 1000000,
DIVISOR_FOR_NANO = 1000000000};
```
--
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] [feture](stream load) (step one)Add arrow data type for stream load [doris]
Posted by "wuwenchi (via GitHub)" <gi...@apache.org>.
wuwenchi commented on PR #26709:
URL: https://github.com/apache/doris/pull/26709#issuecomment-1837816991
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] [feture](stream load) (step one)Add arrow data type for stream load [doris]
Posted by "wuwenchi (via GitHub)" <gi...@apache.org>.
wuwenchi commented on PR #26709:
URL: https://github.com/apache/doris/pull/26709#issuecomment-1827263586
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] [feture](stream load)Add arrow data type for stream load [doris]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on code in PR #26709:
URL: https://github.com/apache/doris/pull/26709#discussion_r1391915522
##########
be/src/vec/exec/format/arrow/arrow_batch_reader.cpp:
##########
@@ -0,0 +1,138 @@
+// 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.
+
+#include "arrow_batch_reader.h"
+
+#include "arrow/array.h"
+#include "arrow/io/buffered.h"
+#include "arrow/io/stdio.h"
+#include "arrow/ipc/options.h"
+#include "arrow/ipc/reader.h"
+#include "arrow/record_batch.h"
+#include "arrow/result.h"
+#include "common/logging.h"
+#include "io/fs/stream_load_pipe.h"
+#include "olap/wal_manager.h"
+#include "runtime/runtime_state.h"
+
+#define DEFAULT_READ_BUF_CAP (4 * 1024 * 1024)
Review Comment:
warning: macro 'DEFAULT_READ_BUF_CAP' defines an integral constant; prefer an enum instead [modernize-macro-to-enum]
```cpp
#define DEFAULT_READ_BUF_CAP (4 * 1024 * 1024)
^
```
##########
be/src/vec/exec/format/arrow/arrow_batch_reader.cpp:
##########
@@ -0,0 +1,138 @@
+// 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.
+
+#include "arrow_batch_reader.h"
+
+#include "arrow/array.h"
+#include "arrow/io/buffered.h"
+#include "arrow/io/stdio.h"
+#include "arrow/ipc/options.h"
+#include "arrow/ipc/reader.h"
+#include "arrow/record_batch.h"
+#include "arrow/result.h"
+#include "common/logging.h"
+#include "io/fs/stream_load_pipe.h"
+#include "olap/wal_manager.h"
+#include "runtime/runtime_state.h"
+
+#define DEFAULT_READ_BUF_CAP (4 * 1024 * 1024)
+#define BATCH_SIZE_LENGTH (4)
+
Review Comment:
warning: replace macro with enum [modernize-macro-to-enum]
```suggestion
enum {
DEFAULT_READ_BUF_CAP = (4 * 1024 * 1024),
BATCH_SIZE_LENGTH = (4)};
```
##########
be/src/vec/exec/format/arrow/arrow_batch_reader.cpp:
##########
@@ -0,0 +1,138 @@
+// 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.
+
+#include "arrow_batch_reader.h"
+
+#include "arrow/array.h"
+#include "arrow/io/buffered.h"
+#include "arrow/io/stdio.h"
+#include "arrow/ipc/options.h"
+#include "arrow/ipc/reader.h"
+#include "arrow/record_batch.h"
+#include "arrow/result.h"
+#include "common/logging.h"
+#include "io/fs/stream_load_pipe.h"
+#include "olap/wal_manager.h"
+#include "runtime/runtime_state.h"
+
+#define DEFAULT_READ_BUF_CAP (4 * 1024 * 1024)
+#define BATCH_SIZE_LENGTH (4)
+
+namespace doris::vectorized {
+
+ArrowBatchReader::ArrowBatchReader(io::FileReaderSPtr file_reader)
+ : _file_reader(file_reader),
+ _read_buf(new uint8_t[DEFAULT_READ_BUF_CAP]),
+ _read_buf_cap(DEFAULT_READ_BUF_CAP),
+ _read_buf_pos(0),
+ _read_buf_len(0),
+ _batch_size(-1) {}
+
+ArrowBatchReader::~ArrowBatchReader() = default;
+
+Status ArrowBatchReader::get_one_batch(uint8_t** data, int* length) {
+ _init();
+ RETURN_IF_ERROR(_get_batch_size());
+ RETURN_IF_ERROR(_get_batch_value());
+
+ *data = &_read_buf[_read_buf_pos];
+ if (_batch_size <= 0) {
+ *length = 0;
+ } else {
+ *length = _read_buf_len;
+ }
+ return Status::OK();
+}
+
+void ArrowBatchReader::_init() {
+ _read_buf_pos = _read_buf_pos + _read_buf_len;
+ _batch_size = -1;
+ _read_buf_len = 0;
+}
+
+Status ArrowBatchReader::_get_batch_size() {
+ if (_batch_size >= 0) {
+ return Status::OK();
+ }
+
+ if (_get_valid_cap() < BATCH_SIZE_LENGTH) {
+ memmove(_read_buf, _read_buf + _read_buf_pos, _read_buf_len);
+ }
+
+ while (_read_buf_len < BATCH_SIZE_LENGTH) {
+ Slice file_slice(_read_buf + _read_buf_pos + _read_buf_len, _get_valid_cap());
+ size_t read_length = 0;
+ RETURN_IF_ERROR(_file_reader->read_at(0, file_slice, &read_length, NULL));
+ if (read_length == 0) {
+ return Status::OK();
+ }
+ _read_buf_len += read_length;
+ }
+
+ _batch_size = _convert_to_length(_read_buf);
+
+ if (_batch_size < 0) {
+ return Status::InternalError("Invalid batch size");
+ }
+
+ _read_buf_pos += BATCH_SIZE_LENGTH;
+ _read_buf_len -= BATCH_SIZE_LENGTH;
+
+ if (_batch_size > _read_buf_cap) {
+ while (_read_buf_cap < _batch_size) {
+ _read_buf_cap *= 2;
+ }
+
+ uint8_t* new_read_buf = new uint8_t[_read_buf_cap];
+ memmove(new_read_buf, _read_buf + _read_buf_pos, _read_buf_len);
+ delete[] _read_buf;
+ _read_buf = new_read_buf;
+ }
+ return Status::OK();
+}
+
+int ArrowBatchReader::_get_valid_cap() {
+ return _read_buf_cap - _read_buf_pos - _read_buf_len;
+}
+
+uint32_t ArrowBatchReader::_convert_to_length(uint8_t* data) {
Review Comment:
warning: method '_convert_to_length' can be made static [readability-convert-member-functions-to-static]
be/src/vec/exec/format/arrow/arrow_batch_reader.h:65:
```diff
- uint32_t _convert_to_length(uint8_t* data);
+ static uint32_t _convert_to_length(uint8_t* data);
```
##########
be/src/vec/exec/format/arrow/arrow_batch_reader.cpp:
##########
@@ -0,0 +1,138 @@
+// 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.
+
+#include "arrow_batch_reader.h"
+
+#include "arrow/array.h"
+#include "arrow/io/buffered.h"
+#include "arrow/io/stdio.h"
+#include "arrow/ipc/options.h"
+#include "arrow/ipc/reader.h"
+#include "arrow/record_batch.h"
+#include "arrow/result.h"
+#include "common/logging.h"
+#include "io/fs/stream_load_pipe.h"
+#include "olap/wal_manager.h"
+#include "runtime/runtime_state.h"
+
+#define DEFAULT_READ_BUF_CAP (4 * 1024 * 1024)
+#define BATCH_SIZE_LENGTH (4)
+
+namespace doris::vectorized {
+
+ArrowBatchReader::ArrowBatchReader(io::FileReaderSPtr file_reader)
+ : _file_reader(file_reader),
+ _read_buf(new uint8_t[DEFAULT_READ_BUF_CAP]),
+ _read_buf_cap(DEFAULT_READ_BUF_CAP),
+ _read_buf_pos(0),
+ _read_buf_len(0),
+ _batch_size(-1) {}
+
+ArrowBatchReader::~ArrowBatchReader() = default;
+
+Status ArrowBatchReader::get_one_batch(uint8_t** data, int* length) {
+ _init();
+ RETURN_IF_ERROR(_get_batch_size());
+ RETURN_IF_ERROR(_get_batch_value());
+
+ *data = &_read_buf[_read_buf_pos];
+ if (_batch_size <= 0) {
+ *length = 0;
+ } else {
+ *length = _read_buf_len;
+ }
+ return Status::OK();
+}
+
+void ArrowBatchReader::_init() {
+ _read_buf_pos = _read_buf_pos + _read_buf_len;
+ _batch_size = -1;
+ _read_buf_len = 0;
+}
+
+Status ArrowBatchReader::_get_batch_size() {
+ if (_batch_size >= 0) {
+ return Status::OK();
+ }
+
+ if (_get_valid_cap() < BATCH_SIZE_LENGTH) {
+ memmove(_read_buf, _read_buf + _read_buf_pos, _read_buf_len);
+ }
+
+ while (_read_buf_len < BATCH_SIZE_LENGTH) {
+ Slice file_slice(_read_buf + _read_buf_pos + _read_buf_len, _get_valid_cap());
+ size_t read_length = 0;
+ RETURN_IF_ERROR(_file_reader->read_at(0, file_slice, &read_length, NULL));
+ if (read_length == 0) {
+ return Status::OK();
+ }
+ _read_buf_len += read_length;
+ }
+
+ _batch_size = _convert_to_length(_read_buf);
+
+ if (_batch_size < 0) {
+ return Status::InternalError("Invalid batch size");
+ }
+
+ _read_buf_pos += BATCH_SIZE_LENGTH;
+ _read_buf_len -= BATCH_SIZE_LENGTH;
+
+ if (_batch_size > _read_buf_cap) {
+ while (_read_buf_cap < _batch_size) {
+ _read_buf_cap *= 2;
+ }
+
+ uint8_t* new_read_buf = new uint8_t[_read_buf_cap];
+ memmove(new_read_buf, _read_buf + _read_buf_pos, _read_buf_len);
+ delete[] _read_buf;
+ _read_buf = new_read_buf;
+ }
+ return Status::OK();
+}
+
+int ArrowBatchReader::_get_valid_cap() {
Review Comment:
warning: method '_get_valid_cap' can be made const [readability-make-member-function-const]
be/src/vec/exec/format/arrow/arrow_batch_reader.h:67:
```diff
- int _get_valid_cap();
+ int _get_valid_cap() const;
```
```suggestion
int ArrowBatchReader::_get_valid_cap() const {
```
##########
be/src/vec/exec/format/arrow/arrow_batch_reader.cpp:
##########
@@ -0,0 +1,138 @@
+// 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.
+
+#include "arrow_batch_reader.h"
+
+#include "arrow/array.h"
+#include "arrow/io/buffered.h"
+#include "arrow/io/stdio.h"
+#include "arrow/ipc/options.h"
+#include "arrow/ipc/reader.h"
+#include "arrow/record_batch.h"
+#include "arrow/result.h"
+#include "common/logging.h"
+#include "io/fs/stream_load_pipe.h"
+#include "olap/wal_manager.h"
+#include "runtime/runtime_state.h"
+
+#define DEFAULT_READ_BUF_CAP (4 * 1024 * 1024)
+#define BATCH_SIZE_LENGTH (4)
Review Comment:
warning: macro 'BATCH_SIZE_LENGTH' defines an integral constant; prefer an enum instead [modernize-macro-to-enum]
```cpp
#define BATCH_SIZE_LENGTH (4)
^
```
##########
be/src/vec/exec/format/arrow/arrow_batch_reader.cpp:
##########
@@ -0,0 +1,138 @@
+// 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.
+
+#include "arrow_batch_reader.h"
+
+#include "arrow/array.h"
+#include "arrow/io/buffered.h"
+#include "arrow/io/stdio.h"
+#include "arrow/ipc/options.h"
+#include "arrow/ipc/reader.h"
+#include "arrow/record_batch.h"
+#include "arrow/result.h"
+#include "common/logging.h"
+#include "io/fs/stream_load_pipe.h"
+#include "olap/wal_manager.h"
+#include "runtime/runtime_state.h"
+
+#define DEFAULT_READ_BUF_CAP (4 * 1024 * 1024)
+#define BATCH_SIZE_LENGTH (4)
+
+namespace doris::vectorized {
+
+ArrowBatchReader::ArrowBatchReader(io::FileReaderSPtr file_reader)
+ : _file_reader(file_reader),
+ _read_buf(new uint8_t[DEFAULT_READ_BUF_CAP]),
+ _read_buf_cap(DEFAULT_READ_BUF_CAP),
+ _read_buf_pos(0),
+ _read_buf_len(0),
+ _batch_size(-1) {}
+
+ArrowBatchReader::~ArrowBatchReader() = default;
+
+Status ArrowBatchReader::get_one_batch(uint8_t** data, int* length) {
+ _init();
+ RETURN_IF_ERROR(_get_batch_size());
+ RETURN_IF_ERROR(_get_batch_value());
+
+ *data = &_read_buf[_read_buf_pos];
+ if (_batch_size <= 0) {
+ *length = 0;
+ } else {
+ *length = _read_buf_len;
+ }
+ return Status::OK();
+}
+
+void ArrowBatchReader::_init() {
+ _read_buf_pos = _read_buf_pos + _read_buf_len;
+ _batch_size = -1;
+ _read_buf_len = 0;
+}
+
+Status ArrowBatchReader::_get_batch_size() {
+ if (_batch_size >= 0) {
+ return Status::OK();
+ }
+
+ if (_get_valid_cap() < BATCH_SIZE_LENGTH) {
+ memmove(_read_buf, _read_buf + _read_buf_pos, _read_buf_len);
+ }
+
+ while (_read_buf_len < BATCH_SIZE_LENGTH) {
+ Slice file_slice(_read_buf + _read_buf_pos + _read_buf_len, _get_valid_cap());
+ size_t read_length = 0;
+ RETURN_IF_ERROR(_file_reader->read_at(0, file_slice, &read_length, NULL));
+ if (read_length == 0) {
+ return Status::OK();
+ }
+ _read_buf_len += read_length;
+ }
+
+ _batch_size = _convert_to_length(_read_buf);
+
+ if (_batch_size < 0) {
+ return Status::InternalError("Invalid batch size");
+ }
+
+ _read_buf_pos += BATCH_SIZE_LENGTH;
+ _read_buf_len -= BATCH_SIZE_LENGTH;
+
+ if (_batch_size > _read_buf_cap) {
+ while (_read_buf_cap < _batch_size) {
+ _read_buf_cap *= 2;
+ }
+
+ uint8_t* new_read_buf = new uint8_t[_read_buf_cap];
+ memmove(new_read_buf, _read_buf + _read_buf_pos, _read_buf_len);
+ delete[] _read_buf;
+ _read_buf = new_read_buf;
+ }
+ return Status::OK();
+}
+
+int ArrowBatchReader::_get_valid_cap() {
+ return _read_buf_cap - _read_buf_pos - _read_buf_len;
+}
+
+uint32_t ArrowBatchReader::_convert_to_length(uint8_t* data) {
+ uint32_t len = (((uint32_t)data[0]) << 24) & 0xff000000;
+ len |= (((uint32_t)data[1]) << 16) & 0xff0000;
Review Comment:
warning: 16 is a magic number; consider replacing it with a named constant [readability-magic-numbers]
```cpp
len |= (((uint32_t)data[1]) << 16) & 0xff0000;
^
```
##########
be/src/vec/exec/format/arrow/arrow_batch_reader.cpp:
##########
@@ -0,0 +1,138 @@
+// 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.
+
+#include "arrow_batch_reader.h"
+
+#include "arrow/array.h"
+#include "arrow/io/buffered.h"
+#include "arrow/io/stdio.h"
+#include "arrow/ipc/options.h"
+#include "arrow/ipc/reader.h"
+#include "arrow/record_batch.h"
+#include "arrow/result.h"
+#include "common/logging.h"
+#include "io/fs/stream_load_pipe.h"
+#include "olap/wal_manager.h"
+#include "runtime/runtime_state.h"
+
+#define DEFAULT_READ_BUF_CAP (4 * 1024 * 1024)
+#define BATCH_SIZE_LENGTH (4)
+
+namespace doris::vectorized {
+
+ArrowBatchReader::ArrowBatchReader(io::FileReaderSPtr file_reader)
+ : _file_reader(file_reader),
+ _read_buf(new uint8_t[DEFAULT_READ_BUF_CAP]),
+ _read_buf_cap(DEFAULT_READ_BUF_CAP),
+ _read_buf_pos(0),
+ _read_buf_len(0),
+ _batch_size(-1) {}
+
+ArrowBatchReader::~ArrowBatchReader() = default;
+
+Status ArrowBatchReader::get_one_batch(uint8_t** data, int* length) {
+ _init();
+ RETURN_IF_ERROR(_get_batch_size());
+ RETURN_IF_ERROR(_get_batch_value());
+
+ *data = &_read_buf[_read_buf_pos];
+ if (_batch_size <= 0) {
+ *length = 0;
+ } else {
+ *length = _read_buf_len;
+ }
+ return Status::OK();
+}
+
+void ArrowBatchReader::_init() {
+ _read_buf_pos = _read_buf_pos + _read_buf_len;
+ _batch_size = -1;
+ _read_buf_len = 0;
+}
+
+Status ArrowBatchReader::_get_batch_size() {
+ if (_batch_size >= 0) {
+ return Status::OK();
+ }
+
+ if (_get_valid_cap() < BATCH_SIZE_LENGTH) {
+ memmove(_read_buf, _read_buf + _read_buf_pos, _read_buf_len);
+ }
+
+ while (_read_buf_len < BATCH_SIZE_LENGTH) {
+ Slice file_slice(_read_buf + _read_buf_pos + _read_buf_len, _get_valid_cap());
+ size_t read_length = 0;
+ RETURN_IF_ERROR(_file_reader->read_at(0, file_slice, &read_length, NULL));
+ if (read_length == 0) {
+ return Status::OK();
+ }
+ _read_buf_len += read_length;
+ }
+
+ _batch_size = _convert_to_length(_read_buf);
+
+ if (_batch_size < 0) {
+ return Status::InternalError("Invalid batch size");
+ }
+
+ _read_buf_pos += BATCH_SIZE_LENGTH;
+ _read_buf_len -= BATCH_SIZE_LENGTH;
+
+ if (_batch_size > _read_buf_cap) {
+ while (_read_buf_cap < _batch_size) {
+ _read_buf_cap *= 2;
+ }
+
+ uint8_t* new_read_buf = new uint8_t[_read_buf_cap];
+ memmove(new_read_buf, _read_buf + _read_buf_pos, _read_buf_len);
+ delete[] _read_buf;
+ _read_buf = new_read_buf;
+ }
+ return Status::OK();
+}
+
+int ArrowBatchReader::_get_valid_cap() {
+ return _read_buf_cap - _read_buf_pos - _read_buf_len;
+}
+
+uint32_t ArrowBatchReader::_convert_to_length(uint8_t* data) {
+ uint32_t len = (((uint32_t)data[0]) << 24) & 0xff000000;
+ len |= (((uint32_t)data[1]) << 16) & 0xff0000;
+ len |= (((uint32_t)data[2]) << 8) & 0xff00;
Review Comment:
warning: 0xff00 is a magic number; consider replacing it with a named constant [readability-magic-numbers]
```cpp
len |= (((uint32_t)data[2]) << 8) & 0xff00;
^
```
##########
be/src/vec/exec/format/arrow/arrow_batch_reader.cpp:
##########
@@ -0,0 +1,138 @@
+// 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.
+
+#include "arrow_batch_reader.h"
+
+#include "arrow/array.h"
+#include "arrow/io/buffered.h"
+#include "arrow/io/stdio.h"
+#include "arrow/ipc/options.h"
+#include "arrow/ipc/reader.h"
+#include "arrow/record_batch.h"
+#include "arrow/result.h"
+#include "common/logging.h"
+#include "io/fs/stream_load_pipe.h"
+#include "olap/wal_manager.h"
+#include "runtime/runtime_state.h"
+
+#define DEFAULT_READ_BUF_CAP (4 * 1024 * 1024)
+#define BATCH_SIZE_LENGTH (4)
+
+namespace doris::vectorized {
+
+ArrowBatchReader::ArrowBatchReader(io::FileReaderSPtr file_reader)
+ : _file_reader(file_reader),
+ _read_buf(new uint8_t[DEFAULT_READ_BUF_CAP]),
+ _read_buf_cap(DEFAULT_READ_BUF_CAP),
+ _read_buf_pos(0),
+ _read_buf_len(0),
+ _batch_size(-1) {}
+
+ArrowBatchReader::~ArrowBatchReader() = default;
+
+Status ArrowBatchReader::get_one_batch(uint8_t** data, int* length) {
+ _init();
+ RETURN_IF_ERROR(_get_batch_size());
+ RETURN_IF_ERROR(_get_batch_value());
+
+ *data = &_read_buf[_read_buf_pos];
+ if (_batch_size <= 0) {
+ *length = 0;
+ } else {
+ *length = _read_buf_len;
+ }
+ return Status::OK();
+}
+
+void ArrowBatchReader::_init() {
+ _read_buf_pos = _read_buf_pos + _read_buf_len;
+ _batch_size = -1;
+ _read_buf_len = 0;
+}
+
+Status ArrowBatchReader::_get_batch_size() {
+ if (_batch_size >= 0) {
+ return Status::OK();
+ }
+
+ if (_get_valid_cap() < BATCH_SIZE_LENGTH) {
+ memmove(_read_buf, _read_buf + _read_buf_pos, _read_buf_len);
+ }
+
+ while (_read_buf_len < BATCH_SIZE_LENGTH) {
+ Slice file_slice(_read_buf + _read_buf_pos + _read_buf_len, _get_valid_cap());
+ size_t read_length = 0;
+ RETURN_IF_ERROR(_file_reader->read_at(0, file_slice, &read_length, NULL));
+ if (read_length == 0) {
+ return Status::OK();
+ }
+ _read_buf_len += read_length;
+ }
+
+ _batch_size = _convert_to_length(_read_buf);
+
+ if (_batch_size < 0) {
+ return Status::InternalError("Invalid batch size");
+ }
+
+ _read_buf_pos += BATCH_SIZE_LENGTH;
+ _read_buf_len -= BATCH_SIZE_LENGTH;
+
+ if (_batch_size > _read_buf_cap) {
+ while (_read_buf_cap < _batch_size) {
+ _read_buf_cap *= 2;
+ }
+
+ uint8_t* new_read_buf = new uint8_t[_read_buf_cap];
+ memmove(new_read_buf, _read_buf + _read_buf_pos, _read_buf_len);
+ delete[] _read_buf;
+ _read_buf = new_read_buf;
+ }
+ return Status::OK();
+}
+
+int ArrowBatchReader::_get_valid_cap() {
+ return _read_buf_cap - _read_buf_pos - _read_buf_len;
+}
+
+uint32_t ArrowBatchReader::_convert_to_length(uint8_t* data) {
Review Comment:
warning: pointer parameter 'data' can be pointer to const [readability-non-const-parameter]
be/src/vec/exec/format/arrow/arrow_batch_reader.h:65:
```diff
- uint32_t _convert_to_length(uint8_t* data);
+ uint32_t _convert_to_length(const uint8_t* data);
```
```suggestion
uint32_t ArrowBatchReader::_convert_to_length(const uint8_t* data) {
```
##########
be/src/vec/exec/format/arrow/arrow_batch_reader.cpp:
##########
@@ -0,0 +1,138 @@
+// 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.
+
+#include "arrow_batch_reader.h"
+
+#include "arrow/array.h"
+#include "arrow/io/buffered.h"
+#include "arrow/io/stdio.h"
+#include "arrow/ipc/options.h"
+#include "arrow/ipc/reader.h"
+#include "arrow/record_batch.h"
+#include "arrow/result.h"
+#include "common/logging.h"
+#include "io/fs/stream_load_pipe.h"
+#include "olap/wal_manager.h"
+#include "runtime/runtime_state.h"
+
+#define DEFAULT_READ_BUF_CAP (4 * 1024 * 1024)
+#define BATCH_SIZE_LENGTH (4)
+
+namespace doris::vectorized {
+
+ArrowBatchReader::ArrowBatchReader(io::FileReaderSPtr file_reader)
+ : _file_reader(file_reader),
+ _read_buf(new uint8_t[DEFAULT_READ_BUF_CAP]),
+ _read_buf_cap(DEFAULT_READ_BUF_CAP),
+ _read_buf_pos(0),
+ _read_buf_len(0),
+ _batch_size(-1) {}
+
+ArrowBatchReader::~ArrowBatchReader() = default;
+
+Status ArrowBatchReader::get_one_batch(uint8_t** data, int* length) {
+ _init();
+ RETURN_IF_ERROR(_get_batch_size());
+ RETURN_IF_ERROR(_get_batch_value());
+
+ *data = &_read_buf[_read_buf_pos];
+ if (_batch_size <= 0) {
+ *length = 0;
+ } else {
+ *length = _read_buf_len;
+ }
+ return Status::OK();
+}
+
+void ArrowBatchReader::_init() {
+ _read_buf_pos = _read_buf_pos + _read_buf_len;
+ _batch_size = -1;
+ _read_buf_len = 0;
+}
+
+Status ArrowBatchReader::_get_batch_size() {
+ if (_batch_size >= 0) {
+ return Status::OK();
+ }
+
+ if (_get_valid_cap() < BATCH_SIZE_LENGTH) {
+ memmove(_read_buf, _read_buf + _read_buf_pos, _read_buf_len);
+ }
+
+ while (_read_buf_len < BATCH_SIZE_LENGTH) {
+ Slice file_slice(_read_buf + _read_buf_pos + _read_buf_len, _get_valid_cap());
+ size_t read_length = 0;
+ RETURN_IF_ERROR(_file_reader->read_at(0, file_slice, &read_length, NULL));
+ if (read_length == 0) {
+ return Status::OK();
+ }
+ _read_buf_len += read_length;
+ }
+
+ _batch_size = _convert_to_length(_read_buf);
+
+ if (_batch_size < 0) {
+ return Status::InternalError("Invalid batch size");
+ }
+
+ _read_buf_pos += BATCH_SIZE_LENGTH;
+ _read_buf_len -= BATCH_SIZE_LENGTH;
+
+ if (_batch_size > _read_buf_cap) {
+ while (_read_buf_cap < _batch_size) {
+ _read_buf_cap *= 2;
+ }
+
+ uint8_t* new_read_buf = new uint8_t[_read_buf_cap];
+ memmove(new_read_buf, _read_buf + _read_buf_pos, _read_buf_len);
+ delete[] _read_buf;
+ _read_buf = new_read_buf;
+ }
+ return Status::OK();
+}
+
+int ArrowBatchReader::_get_valid_cap() {
+ return _read_buf_cap - _read_buf_pos - _read_buf_len;
+}
+
+uint32_t ArrowBatchReader::_convert_to_length(uint8_t* data) {
+ uint32_t len = (((uint32_t)data[0]) << 24) & 0xff000000;
+ len |= (((uint32_t)data[1]) << 16) & 0xff0000;
+ len |= (((uint32_t)data[2]) << 8) & 0xff00;
+ len |= ((uint32_t)data[3]) & 0xff;
Review Comment:
warning: 0xff is a magic number; consider replacing it with a named constant [readability-magic-numbers]
```cpp
len |= ((uint32_t)data[3]) & 0xff;
^
```
##########
be/src/vec/exec/format/arrow/arrow_batch_reader.h:
##########
@@ -0,0 +1,84 @@
+// 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.
+
+#pragma once
+
+#include <gen_cpp/PlanNodes_types.h>
+#include <gen_cpp/internal_service.pb.h>
+#include <stddef.h>
+#include <stdint.h>
Review Comment:
warning: inclusion of deprecated C++ header 'stdint.h'; consider using 'cstdint' instead [modernize-deprecated-headers]
```suggestion
#include <cstdint>
```
##########
be/src/vec/exec/format/arrow/arrow_batch_reader.h:
##########
@@ -0,0 +1,84 @@
+// 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.
+
+#pragma once
+
+#include <gen_cpp/PlanNodes_types.h>
+#include <gen_cpp/internal_service.pb.h>
+#include <stddef.h>
+#include <stdint.h>
+
+#include <cstddef>
+#include <memory>
+#include <string>
+#include <unordered_map>
+#include <unordered_set>
+#include <vector>
+
+#include "common/status.h"
+#include "io/file_factory.h"
+#include "io/fs/file_reader_writer_fwd.h"
+#include "util/slice.h"
+#include "vec/data_types/data_type.h"
+#include "vec/exec/format/file_reader/new_plain_text_line_reader.h"
+#include "vec/exec/format/generic_reader.h"
+
+namespace doris {
+
+namespace io {
+class FileSystem;
+struct IOContext;
+} // namespace io
+
+namespace vectorized {
+
+struct ScannerCounter;
+class Block;
+
+class ArrowBatchReader {
+ ENABLE_FACTORY_CREATOR(ArrowBatchReader);
+
+public:
+ ArrowBatchReader(io::FileReaderSPtr file_reader);
+
+ ~ArrowBatchReader();
+
+ // Status read_one_batch(std::unique_ptr<uint8_t[]>* data, size_t* length);
+ Status get_one_batch(uint8_t** data, int* length);
+
+private:
+ void _init();
+ Status _get_batch_size();
+ uint32_t _convert_to_length(uint8_t* data);
+ Status _get_batch_value();
+ int _get_valid_cap();
+
+private:
Review Comment:
warning: redundant access specifier has the same accessibility as the previous access specifier [readability-redundant-access-specifiers]
```suggestion
```
<details>
<summary>Additional context</summary>
**be/src/vec/exec/format/arrow/arrow_batch_reader.h:62:** previously declared here
```cpp
private:
^
```
</details>
##########
be/src/vec/exec/format/arrow/arrow_batch_reader.h:
##########
@@ -0,0 +1,84 @@
+// 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.
+
+#pragma once
+
+#include <gen_cpp/PlanNodes_types.h>
+#include <gen_cpp/internal_service.pb.h>
+#include <stddef.h>
Review Comment:
warning: inclusion of deprecated C++ header 'stddef.h'; consider using 'cstddef' instead [modernize-deprecated-headers]
```suggestion
#include <cstddef>
```
##########
be/src/vec/exec/format/arrow/arrow_batch_reader.cpp:
##########
@@ -0,0 +1,138 @@
+// 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.
+
+#include "arrow_batch_reader.h"
+
+#include "arrow/array.h"
+#include "arrow/io/buffered.h"
+#include "arrow/io/stdio.h"
+#include "arrow/ipc/options.h"
+#include "arrow/ipc/reader.h"
+#include "arrow/record_batch.h"
+#include "arrow/result.h"
+#include "common/logging.h"
+#include "io/fs/stream_load_pipe.h"
+#include "olap/wal_manager.h"
+#include "runtime/runtime_state.h"
+
+#define DEFAULT_READ_BUF_CAP (4 * 1024 * 1024)
+#define BATCH_SIZE_LENGTH (4)
+
+namespace doris::vectorized {
+
+ArrowBatchReader::ArrowBatchReader(io::FileReaderSPtr file_reader)
+ : _file_reader(file_reader),
+ _read_buf(new uint8_t[DEFAULT_READ_BUF_CAP]),
+ _read_buf_cap(DEFAULT_READ_BUF_CAP),
+ _read_buf_pos(0),
+ _read_buf_len(0),
+ _batch_size(-1) {}
+
+ArrowBatchReader::~ArrowBatchReader() = default;
+
+Status ArrowBatchReader::get_one_batch(uint8_t** data, int* length) {
+ _init();
+ RETURN_IF_ERROR(_get_batch_size());
+ RETURN_IF_ERROR(_get_batch_value());
+
+ *data = &_read_buf[_read_buf_pos];
+ if (_batch_size <= 0) {
+ *length = 0;
+ } else {
+ *length = _read_buf_len;
+ }
+ return Status::OK();
+}
+
+void ArrowBatchReader::_init() {
+ _read_buf_pos = _read_buf_pos + _read_buf_len;
+ _batch_size = -1;
+ _read_buf_len = 0;
+}
+
+Status ArrowBatchReader::_get_batch_size() {
+ if (_batch_size >= 0) {
+ return Status::OK();
+ }
+
+ if (_get_valid_cap() < BATCH_SIZE_LENGTH) {
+ memmove(_read_buf, _read_buf + _read_buf_pos, _read_buf_len);
+ }
+
+ while (_read_buf_len < BATCH_SIZE_LENGTH) {
+ Slice file_slice(_read_buf + _read_buf_pos + _read_buf_len, _get_valid_cap());
+ size_t read_length = 0;
+ RETURN_IF_ERROR(_file_reader->read_at(0, file_slice, &read_length, NULL));
+ if (read_length == 0) {
+ return Status::OK();
+ }
+ _read_buf_len += read_length;
+ }
+
+ _batch_size = _convert_to_length(_read_buf);
+
+ if (_batch_size < 0) {
+ return Status::InternalError("Invalid batch size");
+ }
+
+ _read_buf_pos += BATCH_SIZE_LENGTH;
+ _read_buf_len -= BATCH_SIZE_LENGTH;
+
+ if (_batch_size > _read_buf_cap) {
+ while (_read_buf_cap < _batch_size) {
+ _read_buf_cap *= 2;
+ }
+
+ uint8_t* new_read_buf = new uint8_t[_read_buf_cap];
Review Comment:
warning: use auto when initializing with new to avoid duplicating the type name [modernize-use-auto]
```suggestion
auto* new_read_buf = new uint8_t[_read_buf_cap];
```
##########
be/src/vec/exec/format/arrow/arrow_batch_reader.h:
##########
@@ -0,0 +1,84 @@
+// 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.
+
+#pragma once
+
+#include <gen_cpp/PlanNodes_types.h>
Review Comment:
warning: 'gen_cpp/PlanNodes_types.h' file not found [clang-diagnostic-error]
```cpp
#include <gen_cpp/PlanNodes_types.h>
^
```
##########
be/src/vec/exec/format/arrow/arrow_batch_reader.cpp:
##########
@@ -0,0 +1,138 @@
+// 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.
+
+#include "arrow_batch_reader.h"
+
+#include "arrow/array.h"
+#include "arrow/io/buffered.h"
+#include "arrow/io/stdio.h"
+#include "arrow/ipc/options.h"
+#include "arrow/ipc/reader.h"
+#include "arrow/record_batch.h"
+#include "arrow/result.h"
+#include "common/logging.h"
+#include "io/fs/stream_load_pipe.h"
+#include "olap/wal_manager.h"
+#include "runtime/runtime_state.h"
+
+#define DEFAULT_READ_BUF_CAP (4 * 1024 * 1024)
+#define BATCH_SIZE_LENGTH (4)
+
+namespace doris::vectorized {
+
+ArrowBatchReader::ArrowBatchReader(io::FileReaderSPtr file_reader)
+ : _file_reader(file_reader),
+ _read_buf(new uint8_t[DEFAULT_READ_BUF_CAP]),
+ _read_buf_cap(DEFAULT_READ_BUF_CAP),
+ _read_buf_pos(0),
+ _read_buf_len(0),
+ _batch_size(-1) {}
+
+ArrowBatchReader::~ArrowBatchReader() = default;
+
+Status ArrowBatchReader::get_one_batch(uint8_t** data, int* length) {
+ _init();
+ RETURN_IF_ERROR(_get_batch_size());
+ RETURN_IF_ERROR(_get_batch_value());
+
+ *data = &_read_buf[_read_buf_pos];
+ if (_batch_size <= 0) {
+ *length = 0;
+ } else {
+ *length = _read_buf_len;
+ }
+ return Status::OK();
+}
+
+void ArrowBatchReader::_init() {
+ _read_buf_pos = _read_buf_pos + _read_buf_len;
+ _batch_size = -1;
+ _read_buf_len = 0;
+}
+
+Status ArrowBatchReader::_get_batch_size() {
+ if (_batch_size >= 0) {
+ return Status::OK();
+ }
+
+ if (_get_valid_cap() < BATCH_SIZE_LENGTH) {
+ memmove(_read_buf, _read_buf + _read_buf_pos, _read_buf_len);
+ }
+
+ while (_read_buf_len < BATCH_SIZE_LENGTH) {
+ Slice file_slice(_read_buf + _read_buf_pos + _read_buf_len, _get_valid_cap());
+ size_t read_length = 0;
+ RETURN_IF_ERROR(_file_reader->read_at(0, file_slice, &read_length, NULL));
+ if (read_length == 0) {
+ return Status::OK();
+ }
+ _read_buf_len += read_length;
+ }
+
+ _batch_size = _convert_to_length(_read_buf);
+
+ if (_batch_size < 0) {
+ return Status::InternalError("Invalid batch size");
+ }
+
+ _read_buf_pos += BATCH_SIZE_LENGTH;
+ _read_buf_len -= BATCH_SIZE_LENGTH;
+
+ if (_batch_size > _read_buf_cap) {
+ while (_read_buf_cap < _batch_size) {
+ _read_buf_cap *= 2;
+ }
+
+ uint8_t* new_read_buf = new uint8_t[_read_buf_cap];
+ memmove(new_read_buf, _read_buf + _read_buf_pos, _read_buf_len);
+ delete[] _read_buf;
+ _read_buf = new_read_buf;
+ }
+ return Status::OK();
+}
+
+int ArrowBatchReader::_get_valid_cap() {
+ return _read_buf_cap - _read_buf_pos - _read_buf_len;
+}
+
+uint32_t ArrowBatchReader::_convert_to_length(uint8_t* data) {
+ uint32_t len = (((uint32_t)data[0]) << 24) & 0xff000000;
Review Comment:
warning: 0xff000000 is a magic number; consider replacing it with a named constant [readability-magic-numbers]
```cpp
uint32_t len = (((uint32_t)data[0]) << 24) & 0xff000000;
^
```
##########
be/src/vec/exec/format/arrow/arrow_batch_reader.cpp:
##########
@@ -0,0 +1,138 @@
+// 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.
+
+#include "arrow_batch_reader.h"
+
+#include "arrow/array.h"
+#include "arrow/io/buffered.h"
+#include "arrow/io/stdio.h"
+#include "arrow/ipc/options.h"
+#include "arrow/ipc/reader.h"
+#include "arrow/record_batch.h"
+#include "arrow/result.h"
+#include "common/logging.h"
+#include "io/fs/stream_load_pipe.h"
+#include "olap/wal_manager.h"
+#include "runtime/runtime_state.h"
+
+#define DEFAULT_READ_BUF_CAP (4 * 1024 * 1024)
+#define BATCH_SIZE_LENGTH (4)
+
+namespace doris::vectorized {
+
+ArrowBatchReader::ArrowBatchReader(io::FileReaderSPtr file_reader)
+ : _file_reader(file_reader),
+ _read_buf(new uint8_t[DEFAULT_READ_BUF_CAP]),
+ _read_buf_cap(DEFAULT_READ_BUF_CAP),
+ _read_buf_pos(0),
+ _read_buf_len(0),
+ _batch_size(-1) {}
+
+ArrowBatchReader::~ArrowBatchReader() = default;
+
+Status ArrowBatchReader::get_one_batch(uint8_t** data, int* length) {
+ _init();
+ RETURN_IF_ERROR(_get_batch_size());
+ RETURN_IF_ERROR(_get_batch_value());
+
+ *data = &_read_buf[_read_buf_pos];
+ if (_batch_size <= 0) {
+ *length = 0;
+ } else {
+ *length = _read_buf_len;
+ }
+ return Status::OK();
+}
+
+void ArrowBatchReader::_init() {
+ _read_buf_pos = _read_buf_pos + _read_buf_len;
+ _batch_size = -1;
+ _read_buf_len = 0;
+}
+
+Status ArrowBatchReader::_get_batch_size() {
+ if (_batch_size >= 0) {
+ return Status::OK();
+ }
+
+ if (_get_valid_cap() < BATCH_SIZE_LENGTH) {
+ memmove(_read_buf, _read_buf + _read_buf_pos, _read_buf_len);
+ }
+
+ while (_read_buf_len < BATCH_SIZE_LENGTH) {
+ Slice file_slice(_read_buf + _read_buf_pos + _read_buf_len, _get_valid_cap());
+ size_t read_length = 0;
+ RETURN_IF_ERROR(_file_reader->read_at(0, file_slice, &read_length, NULL));
+ if (read_length == 0) {
+ return Status::OK();
+ }
+ _read_buf_len += read_length;
+ }
+
+ _batch_size = _convert_to_length(_read_buf);
+
+ if (_batch_size < 0) {
+ return Status::InternalError("Invalid batch size");
+ }
+
+ _read_buf_pos += BATCH_SIZE_LENGTH;
+ _read_buf_len -= BATCH_SIZE_LENGTH;
+
+ if (_batch_size > _read_buf_cap) {
+ while (_read_buf_cap < _batch_size) {
+ _read_buf_cap *= 2;
+ }
+
+ uint8_t* new_read_buf = new uint8_t[_read_buf_cap];
+ memmove(new_read_buf, _read_buf + _read_buf_pos, _read_buf_len);
+ delete[] _read_buf;
+ _read_buf = new_read_buf;
+ }
+ return Status::OK();
+}
+
+int ArrowBatchReader::_get_valid_cap() {
+ return _read_buf_cap - _read_buf_pos - _read_buf_len;
+}
+
+uint32_t ArrowBatchReader::_convert_to_length(uint8_t* data) {
+ uint32_t len = (((uint32_t)data[0]) << 24) & 0xff000000;
+ len |= (((uint32_t)data[1]) << 16) & 0xff0000;
+ len |= (((uint32_t)data[2]) << 8) & 0xff00;
Review Comment:
warning: 8 is a magic number; consider replacing it with a named constant [readability-magic-numbers]
```cpp
len |= (((uint32_t)data[2]) << 8) & 0xff00;
^
```
##########
be/src/vec/exec/format/arrow/arrow_batch_reader.cpp:
##########
@@ -0,0 +1,138 @@
+// 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.
+
+#include "arrow_batch_reader.h"
+
+#include "arrow/array.h"
+#include "arrow/io/buffered.h"
+#include "arrow/io/stdio.h"
+#include "arrow/ipc/options.h"
+#include "arrow/ipc/reader.h"
+#include "arrow/record_batch.h"
+#include "arrow/result.h"
+#include "common/logging.h"
+#include "io/fs/stream_load_pipe.h"
+#include "olap/wal_manager.h"
+#include "runtime/runtime_state.h"
+
+#define DEFAULT_READ_BUF_CAP (4 * 1024 * 1024)
+#define BATCH_SIZE_LENGTH (4)
+
+namespace doris::vectorized {
+
+ArrowBatchReader::ArrowBatchReader(io::FileReaderSPtr file_reader)
+ : _file_reader(file_reader),
+ _read_buf(new uint8_t[DEFAULT_READ_BUF_CAP]),
+ _read_buf_cap(DEFAULT_READ_BUF_CAP),
+ _read_buf_pos(0),
+ _read_buf_len(0),
+ _batch_size(-1) {}
+
+ArrowBatchReader::~ArrowBatchReader() = default;
+
+Status ArrowBatchReader::get_one_batch(uint8_t** data, int* length) {
+ _init();
+ RETURN_IF_ERROR(_get_batch_size());
+ RETURN_IF_ERROR(_get_batch_value());
+
+ *data = &_read_buf[_read_buf_pos];
+ if (_batch_size <= 0) {
+ *length = 0;
+ } else {
+ *length = _read_buf_len;
+ }
+ return Status::OK();
+}
+
+void ArrowBatchReader::_init() {
+ _read_buf_pos = _read_buf_pos + _read_buf_len;
+ _batch_size = -1;
+ _read_buf_len = 0;
+}
+
+Status ArrowBatchReader::_get_batch_size() {
+ if (_batch_size >= 0) {
+ return Status::OK();
+ }
+
+ if (_get_valid_cap() < BATCH_SIZE_LENGTH) {
+ memmove(_read_buf, _read_buf + _read_buf_pos, _read_buf_len);
+ }
+
+ while (_read_buf_len < BATCH_SIZE_LENGTH) {
+ Slice file_slice(_read_buf + _read_buf_pos + _read_buf_len, _get_valid_cap());
+ size_t read_length = 0;
+ RETURN_IF_ERROR(_file_reader->read_at(0, file_slice, &read_length, NULL));
+ if (read_length == 0) {
+ return Status::OK();
+ }
+ _read_buf_len += read_length;
+ }
+
+ _batch_size = _convert_to_length(_read_buf);
+
+ if (_batch_size < 0) {
+ return Status::InternalError("Invalid batch size");
+ }
+
+ _read_buf_pos += BATCH_SIZE_LENGTH;
+ _read_buf_len -= BATCH_SIZE_LENGTH;
+
+ if (_batch_size > _read_buf_cap) {
+ while (_read_buf_cap < _batch_size) {
+ _read_buf_cap *= 2;
+ }
+
+ uint8_t* new_read_buf = new uint8_t[_read_buf_cap];
+ memmove(new_read_buf, _read_buf + _read_buf_pos, _read_buf_len);
+ delete[] _read_buf;
+ _read_buf = new_read_buf;
+ }
+ return Status::OK();
+}
+
+int ArrowBatchReader::_get_valid_cap() {
+ return _read_buf_cap - _read_buf_pos - _read_buf_len;
+}
+
+uint32_t ArrowBatchReader::_convert_to_length(uint8_t* data) {
+ uint32_t len = (((uint32_t)data[0]) << 24) & 0xff000000;
+ len |= (((uint32_t)data[1]) << 16) & 0xff0000;
Review Comment:
warning: 0xff0000 is a magic number; consider replacing it with a named constant [readability-magic-numbers]
```cpp
len |= (((uint32_t)data[1]) << 16) & 0xff0000;
^
```
##########
be/src/vec/exec/format/arrow/arrow_batch_reader.cpp:
##########
@@ -0,0 +1,138 @@
+// 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.
+
+#include "arrow_batch_reader.h"
+
+#include "arrow/array.h"
+#include "arrow/io/buffered.h"
+#include "arrow/io/stdio.h"
+#include "arrow/ipc/options.h"
+#include "arrow/ipc/reader.h"
+#include "arrow/record_batch.h"
+#include "arrow/result.h"
+#include "common/logging.h"
+#include "io/fs/stream_load_pipe.h"
+#include "olap/wal_manager.h"
+#include "runtime/runtime_state.h"
+
+#define DEFAULT_READ_BUF_CAP (4 * 1024 * 1024)
+#define BATCH_SIZE_LENGTH (4)
+
+namespace doris::vectorized {
+
+ArrowBatchReader::ArrowBatchReader(io::FileReaderSPtr file_reader)
+ : _file_reader(file_reader),
+ _read_buf(new uint8_t[DEFAULT_READ_BUF_CAP]),
+ _read_buf_cap(DEFAULT_READ_BUF_CAP),
+ _read_buf_pos(0),
+ _read_buf_len(0),
+ _batch_size(-1) {}
+
+ArrowBatchReader::~ArrowBatchReader() = default;
+
+Status ArrowBatchReader::get_one_batch(uint8_t** data, int* length) {
+ _init();
+ RETURN_IF_ERROR(_get_batch_size());
+ RETURN_IF_ERROR(_get_batch_value());
+
+ *data = &_read_buf[_read_buf_pos];
+ if (_batch_size <= 0) {
+ *length = 0;
+ } else {
+ *length = _read_buf_len;
+ }
+ return Status::OK();
+}
+
+void ArrowBatchReader::_init() {
+ _read_buf_pos = _read_buf_pos + _read_buf_len;
+ _batch_size = -1;
+ _read_buf_len = 0;
+}
+
+Status ArrowBatchReader::_get_batch_size() {
+ if (_batch_size >= 0) {
+ return Status::OK();
+ }
+
+ if (_get_valid_cap() < BATCH_SIZE_LENGTH) {
+ memmove(_read_buf, _read_buf + _read_buf_pos, _read_buf_len);
+ }
+
+ while (_read_buf_len < BATCH_SIZE_LENGTH) {
+ Slice file_slice(_read_buf + _read_buf_pos + _read_buf_len, _get_valid_cap());
+ size_t read_length = 0;
+ RETURN_IF_ERROR(_file_reader->read_at(0, file_slice, &read_length, NULL));
+ if (read_length == 0) {
+ return Status::OK();
+ }
+ _read_buf_len += read_length;
+ }
+
+ _batch_size = _convert_to_length(_read_buf);
+
+ if (_batch_size < 0) {
+ return Status::InternalError("Invalid batch size");
+ }
+
+ _read_buf_pos += BATCH_SIZE_LENGTH;
+ _read_buf_len -= BATCH_SIZE_LENGTH;
+
+ if (_batch_size > _read_buf_cap) {
+ while (_read_buf_cap < _batch_size) {
+ _read_buf_cap *= 2;
+ }
+
+ uint8_t* new_read_buf = new uint8_t[_read_buf_cap];
+ memmove(new_read_buf, _read_buf + _read_buf_pos, _read_buf_len);
+ delete[] _read_buf;
+ _read_buf = new_read_buf;
+ }
+ return Status::OK();
+}
+
+int ArrowBatchReader::_get_valid_cap() {
+ return _read_buf_cap - _read_buf_pos - _read_buf_len;
+}
+
+uint32_t ArrowBatchReader::_convert_to_length(uint8_t* data) {
+ uint32_t len = (((uint32_t)data[0]) << 24) & 0xff000000;
Review Comment:
warning: 24 is a magic number; consider replacing it with a named constant [readability-magic-numbers]
```cpp
uint32_t len = (((uint32_t)data[0]) << 24) & 0xff000000;
^
```
--
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] [feture](stream load) (step one)Add arrow data type for stream load [doris]
Posted by "wuwenchi (via GitHub)" <gi...@apache.org>.
wuwenchi commented on PR #26709:
URL: https://github.com/apache/doris/pull/26709#issuecomment-1817726590
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] [feture](stream load) (step one)Add arrow data type for stream load [doris]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on code in PR #26709:
URL: https://github.com/apache/doris/pull/26709#discussion_r1413351397
##########
be/src/vec/exec/format/arrow/arrow_stream_reader.cpp:
##########
@@ -0,0 +1,117 @@
+// 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.
+
+#include "arrow_stream_reader.h"
+
+#include "arrow/array.h"
+#include "arrow/io/buffered.h"
+#include "arrow/io/stdio.h"
+#include "arrow/ipc/options.h"
+#include "arrow/ipc/reader.h"
+#include "arrow/record_batch.h"
+#include "arrow/result.h"
+#include "arrow_pip_input_stream.h"
+#include "common/logging.h"
+#include "io/fs/stream_load_pipe.h"
+#include "olap/wal_manager.h"
+#include "runtime/descriptors.h"
+#include "runtime/runtime_state.h"
+#include "vec/utils/arrow_column_to_doris_column.h"
+
+namespace doris {
+class RuntimeProfile;
+} // namespace doris
+
+namespace doris::vectorized {
+
+ArrowStreamReader::ArrowStreamReader(RuntimeState* state, RuntimeProfile* profile,
+ ScannerCounter* counter, const TFileScanRangeParams& params,
+ const TFileRangeDesc& range,
+ const std::vector<SlotDescriptor*>& file_slot_descs,
+ io::IOContext* io_ctx)
+ : _state(state), _range(range), _file_slot_descs(file_slot_descs), _file_reader(nullptr) {
+ TimezoneUtils::find_cctz_time_zone(TimezoneUtils::default_time_zone, _ctzz);
+}
+
+ArrowStreamReader::~ArrowStreamReader() = default;
+
+Status ArrowStreamReader::init_reader() {
Review Comment:
warning: method 'init_reader' can be made static [readability-convert-member-functions-to-static]
be/src/vec/exec/format/arrow/arrow_stream_reader.h:59:
```diff
- Status init_reader();
+ static Status init_reader();
```
--
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](stream load) (step one)Add arrow data type for stream load [doris]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #26709:
URL: https://github.com/apache/doris/pull/26709#issuecomment-1842623091
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] [feture](stream load) (step one)Add arrow data type for stream load [doris]
Posted by "morningman (via GitHub)" <gi...@apache.org>.
morningman commented on PR #26709:
URL: https://github.com/apache/doris/pull/26709#issuecomment-1841979329
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] [feture](stream load) (step one)Add arrow data type for stream load [doris]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on code in PR #26709:
URL: https://github.com/apache/doris/pull/26709#discussion_r1405742045
##########
be/src/vec/data_types/serde/data_type_datetimev2_serde.cpp:
##########
@@ -112,6 +120,52 @@ void DataTypeDateTimeV2SerDe::write_column_to_arrow(const IColumn& column, const
}
}
+void DataTypeDateTimeV2SerDe::read_column_from_arrow(IColumn& column,
Review Comment:
warning: method 'read_column_from_arrow' can be made static [readability-convert-member-functions-to-static]
```suggestion
static void DataTypeDateTimeV2SerDe::read_column_from_arrow(IColumn& column,
```
be/src/vec/data_types/serde/data_type_datetimev2_serde.cpp:124:
```diff
- int end, const cctz::time_zone& ctz) const {
+ int end, const cctz::time_zone& ctz) {
```
--
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] [feture](stream load) (step one)Add arrow data type for stream load [doris]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on code in PR #26709:
URL: https://github.com/apache/doris/pull/26709#discussion_r1413304786
##########
be/src/vec/exec/format/arrow/arrow_pip_input_stream.h:
##########
@@ -0,0 +1,65 @@
+// 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.
+
+#pragma once
+
+#include <cstddef>
+#include <cstdint>
+#include <memory>
+#include <string>
+#include <unordered_map>
+#include <unordered_set>
+#include <vector>
+
+#include "io/file_factory.h"
+
+namespace doris {
+
+namespace io {
+class FileSystem;
+struct IOContext;
+} // namespace io
+
+namespace vectorized {
+
+class ArrowPipInputStream : public arrow::io::InputStream {
+ ENABLE_FACTORY_CREATOR(ArrowPipInputStream);
+
+public:
+ ArrowPipInputStream(io::FileReaderSPtr file_reader);
+ ~ArrowPipInputStream() override {}
Review Comment:
warning: use '= default' to define a trivial destructor [modernize-use-equals-default]
```suggestion
~ArrowPipInputStream() override = default;
```
##########
be/src/vec/exec/format/arrow/arrow_pip_input_stream.cpp:
##########
@@ -0,0 +1,101 @@
+// 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.
+
+#include "arrow_pip_input_stream.h"
+
+#include "arrow/array.h"
+#include "arrow/buffer.h"
+#include "arrow/io/buffered.h"
+#include "arrow/io/stdio.h"
+#include "arrow/ipc/options.h"
+#include "arrow/ipc/reader.h"
+#include "arrow/record_batch.h"
+#include "arrow/result.h"
+#include "common/logging.h"
+#include "io/fs/stream_load_pipe.h"
+#include "olap/wal_manager.h"
+#include "runtime/runtime_state.h"
+
+namespace doris::vectorized {
+
+ArrowPipInputStream::ArrowPipInputStream(io::FileReaderSPtr file_reader)
+ : _file_reader(file_reader), _pos(0), _begin(true), _read_buf(new uint8_t[4]) {
+ set_mode(arrow::io::FileMode::READ);
+}
+
+arrow::Status ArrowPipInputStream::Close() {
+ return arrow::Status::OK();
+}
+
+bool ArrowPipInputStream::closed() const {
+ return false;
+}
+
+arrow::Result<int64_t> ArrowPipInputStream::Tell() const {
+ return _pos;
+}
+
+Status ArrowPipInputStream::HasNext(bool* get) {
+ // 1. Arrow's serialization uses a 4-byte data to specify the length of the data that follows,
+ // so there must be 4-byte data here.
+ // 2. If it is not determined whether there is a next batch of data (the data has already been transmitted),
+ // then the `_file_reader->read_at` will return a buff with a read length of 0,
+ // and the `RecordBatchStreamReader::Open` function will directly report an error when it gets this buff
+ Slice file_slice(_read_buf, 4);
+ size_t read_length = 0;
+ RETURN_IF_ERROR(_file_reader->read_at(0, file_slice, &read_length, NULL));
+ if (read_length == 0) {
+ *get = false;
+ } else {
+ *get = true;
+ }
+ return Status::OK();
+}
+
+arrow::Result<int64_t> ArrowPipInputStream::Read(int64_t nbytes, void* out) {
+ // RecordBatchStreamReader::Open will create a new reader that will stream a batch of arrow data.
+ // But the first four bytes of this batch of data were taken by the HasNext function, so they need to be copied back here.
+ uint8_t* out_ptr = (uint8_t*)out;
Review Comment:
warning: use auto when initializing with a cast to avoid duplicating the type name [modernize-use-auto]
```suggestion
auto* out_ptr = (uint8_t*)out;
```
##########
be/src/vec/exec/format/arrow/arrow_pip_input_stream.cpp:
##########
@@ -0,0 +1,101 @@
+// 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.
+
+#include "arrow_pip_input_stream.h"
+
+#include "arrow/array.h"
+#include "arrow/buffer.h"
+#include "arrow/io/buffered.h"
+#include "arrow/io/stdio.h"
+#include "arrow/ipc/options.h"
+#include "arrow/ipc/reader.h"
+#include "arrow/record_batch.h"
+#include "arrow/result.h"
+#include "common/logging.h"
+#include "io/fs/stream_load_pipe.h"
+#include "olap/wal_manager.h"
+#include "runtime/runtime_state.h"
+
+namespace doris::vectorized {
+
+ArrowPipInputStream::ArrowPipInputStream(io::FileReaderSPtr file_reader)
+ : _file_reader(file_reader), _pos(0), _begin(true), _read_buf(new uint8_t[4]) {
+ set_mode(arrow::io::FileMode::READ);
+}
+
+arrow::Status ArrowPipInputStream::Close() {
+ return arrow::Status::OK();
+}
+
+bool ArrowPipInputStream::closed() const {
Review Comment:
warning: method 'closed' can be made static [readability-convert-member-functions-to-static]
be/src/vec/exec/format/arrow/arrow_pip_input_stream.h:46:
```diff
- bool closed() const override;
+ static bool closed() override;
```
```suggestion
bool ArrowPipInputStream::closed() {
```
--
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] [feture](stream load) (step one)Add arrow data type for stream load [doris]
Posted by "wuwenchi (via GitHub)" <gi...@apache.org>.
wuwenchi commented on PR #26709:
URL: https://github.com/apache/doris/pull/26709#issuecomment-1837816900
run compile
--
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] [feture](stream load) (step one)Add arrow data type for stream load [doris]
Posted by "morningman (via GitHub)" <gi...@apache.org>.
morningman commented on code in PR #26709:
URL: https://github.com/apache/doris/pull/26709#discussion_r1398411551
##########
be/src/vec/exec/format/arrow/arrow_reader.h:
##########
@@ -0,0 +1,78 @@
+// 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.
+
+#pragma once
+
+#include <gen_cpp/PlanNodes_types.h>
+#include <gen_cpp/internal_service.pb.h>
+#include <stddef.h>
+#include <stdint.h>
+
+#include <cstddef>
+#include <memory>
+#include <string>
+#include <unordered_map>
+#include <unordered_set>
+#include <vector>
+
+#include "arrow_pip_input_stream.h"
+#include "common/status.h"
+#include "io/file_factory.h"
+#include "io/fs/file_reader_writer_fwd.h"
+#include "util/slice.h"
+#include "vec/data_types/data_type.h"
+#include "vec/exec/format/file_reader/new_plain_text_line_reader.h"
+#include "vec/exec/format/generic_reader.h"
+
+namespace doris {
+
+namespace io {
+class FileSystem;
+struct IOContext;
+} // namespace io
+
+namespace vectorized {
+
+struct ScannerCounter;
+class Block;
+
+class ArrowReader : public GenericReader {
Review Comment:
Better name it as `ArrowStreamReader`? Because it only support reading from stream
##########
be/src/vec/exec/format/arrow/arrow_pip_input_stream.cpp:
##########
@@ -0,0 +1,95 @@
+// 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.
+
+#include "arrow_pip_input_stream.h"
+
+#include "arrow/array.h"
+#include "arrow/buffer.h"
+#include "arrow/io/buffered.h"
+#include "arrow/io/stdio.h"
+#include "arrow/ipc/options.h"
+#include "arrow/ipc/reader.h"
+#include "arrow/record_batch.h"
+#include "arrow/result.h"
+#include "common/logging.h"
+#include "io/fs/stream_load_pipe.h"
+#include "olap/wal_manager.h"
+#include "runtime/runtime_state.h"
+
+namespace doris::vectorized {
+
+ArrowPipInputStream::ArrowPipInputStream(io::FileReaderSPtr file_reader)
+ : _file_reader(file_reader), _pos(0), _begin(true), _read_buf(new uint8_t[4]) {
+ set_mode(arrow::io::FileMode::READ);
+}
+
+arrow::Status ArrowPipInputStream::Close() {
+ return arrow::Status::OK();
+}
+
+bool ArrowPipInputStream::closed() const {
+ return false;
+}
+
+arrow::Result<int64_t> ArrowPipInputStream::Tell() const {
+ return _pos;
+}
+
+Status ArrowPipInputStream::HasNext(bool* get) {
+ Slice file_slice(_read_buf, 4);
+ size_t read_length = 0;
+ RETURN_IF_ERROR(_file_reader->read_at(0, file_slice, &read_length, NULL));
+ if (read_length == 0) {
+ *get = false;
+ } else {
+ *get = true;
+ }
+ return Status::OK();
+}
+
+arrow::Result<int64_t> ArrowPipInputStream::Read(int64_t nbytes, void* out) {
+ uint8_t* out_ptr = (uint8_t*)out;
+ if (_begin) {
+ memmove(out_ptr, _read_buf, 4);
+ out_ptr += 4;
+ nbytes -= 4;
+ }
+
+ Slice file_slice(out_ptr, nbytes);
+ size_t read_length = 0;
+ Status status = _file_reader->read_at(0, file_slice, &read_length, NULL);
+ if (UNLIKELY(!status.ok())) {
+ return arrow::Status::IOError("Error to read data from pip");
+ }
+ // pos_ += (int64_t)read_length;
Review Comment:
remove unused code
##########
be/src/vec/exec/format/arrow/batch_with_length_reader.cpp:
##########
@@ -0,0 +1,156 @@
+// 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.
+
+#include "batch_with_length_reader.h"
+
+#include "arrow/array.h"
+#include "arrow/io/buffered.h"
+#include "arrow/io/stdio.h"
+#include "arrow/ipc/options.h"
+#include "arrow/ipc/reader.h"
+#include "arrow/record_batch.h"
+#include "arrow/result.h"
+#include "common/logging.h"
+#include "io/fs/stream_load_pipe.h"
+#include "olap/wal_manager.h"
+#include "runtime/runtime_state.h"
+
+#define DEFAULT_READ_BUF_CAP (4 * 1024 * 1024)
+#define BATCH_SIZE_LENGTH (4)
+
+
+// BatchWithLengthReader used in fixed format: <len><value><len><value>...
+// The <len> is of type int and represents the length of the <value>.
Review Comment:
better use int64 to support batch size larger than 2GB in future
##########
be/src/vec/exec/format/arrow/batch_with_length_reader.cpp:
##########
@@ -0,0 +1,226 @@
+// 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.
+
+#include "batch_with_length_reader.h"
+
+#include "arrow/array.h"
+#include "arrow/io/buffered.h"
+#include "arrow/io/stdio.h"
+#include "arrow/ipc/options.h"
+#include "arrow/ipc/reader.h"
+#include "arrow/record_batch.h"
+#include "arrow/result.h"
+#include "common/logging.h"
+#include "io/fs/stream_load_pipe.h"
+#include "olap/wal_manager.h"
+#include "runtime/runtime_state.h"
+#include "arrow/result.h"
Review Comment:
duplicate
##########
be/src/vec/exec/format/arrow/arrow_reader.cpp:
##########
@@ -0,0 +1,115 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include "arrow_reader.h"
+
+#include "arrow/array.h"
+#include "arrow/io/buffered.h"
+#include "arrow/io/stdio.h"
+#include "arrow/ipc/options.h"
+#include "arrow/ipc/reader.h"
+#include "arrow/record_batch.h"
+#include "arrow/result.h"
+#include "arrow_pip_input_stream.h"
+#include "common/logging.h"
+#include "io/fs/stream_load_pipe.h"
+#include "olap/wal_manager.h"
+#include "runtime/descriptors.h"
+#include "runtime/runtime_state.h"
+#include "vec/utils/arrow_column_to_doris_column.h"
+
+namespace doris {
+class RuntimeProfile;
+} // namespace doris
+
+namespace doris::vectorized {
+
+ArrowReader::ArrowReader(RuntimeState* state, RuntimeProfile* profile, ScannerCounter* counter,
+ const TFileScanRangeParams& params, const TFileRangeDesc& range,
+ const std::vector<SlotDescriptor*>& file_slot_descs, io::IOContext* io_ctx)
+ : _state(state), _range(range), _file_slot_descs(file_slot_descs), _file_reader(nullptr) {}
+
+ArrowReader::~ArrowReader() = default;
+
+Status ArrowReader::init_reader() {
+ RETURN_IF_ERROR(FileFactory::create_pipe_reader(_range.load_id, &_file_reader, _state, false));
+ _pip_stream = ArrowPipInputStream::create_unique(_file_reader);
+ return Status::OK();
+}
+
+Status ArrowReader::get_next_block(Block* block, size_t* read_rows, bool* eof) {
+ bool has_next = false;
+ RETURN_IF_ERROR(_pip_stream->HasNext(&has_next));
+ if (!has_next) {
+ *read_rows = 0;
+ *eof = true;
+ return Status::OK();
+ }
+
+ // create a reader to read data
+ arrow::Result<std::shared_ptr<arrow::ipc::RecordBatchStreamReader>> tRet =
+ arrow::ipc::RecordBatchStreamReader::Open(_pip_stream.get(),
+ arrow::ipc::IpcReadOptions::Defaults());
+ if (!tRet.ok()) {
+ LOG(ERROR) << "failed to open stream reader: " << tRet.status().message();
+ return Status::InternalError("failed to open stream reader: {}", tRet.status().message());
+ }
+ auto reader = std::move(tRet).ValueUnsafe();
+
+ // get arrow data from reader
+ arrow::Result<arrow::RecordBatchVector> tRet2 = reader->ToRecordBatches();
+ if (!tRet2.ok()) {
+ LOG(ERROR) << "failed to read batch: " << tRet2.status().message();
+ return Status::InternalError("failed to read batch: {}", tRet2.status().message());
+ }
+ std::vector<std::shared_ptr<arrow::RecordBatch>> out_batches = std::move(tRet2).ValueUnsafe();
+
+ // convert arrow batch to block
+ cctz::time_zone ctzz;
+ TimezoneUtils::find_cctz_time_zone("UTC", ctzz);
Review Comment:
2 questions:
1. Why using UTC
2. If always be UTC, this should be done once when init the arrow reader
##########
be/src/vec/exec/format/arrow/arrow_reader.cpp:
##########
@@ -0,0 +1,115 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include "arrow_reader.h"
+
+#include "arrow/array.h"
+#include "arrow/io/buffered.h"
+#include "arrow/io/stdio.h"
+#include "arrow/ipc/options.h"
+#include "arrow/ipc/reader.h"
+#include "arrow/record_batch.h"
+#include "arrow/result.h"
+#include "arrow_pip_input_stream.h"
+#include "common/logging.h"
+#include "io/fs/stream_load_pipe.h"
+#include "olap/wal_manager.h"
+#include "runtime/descriptors.h"
+#include "runtime/runtime_state.h"
+#include "vec/utils/arrow_column_to_doris_column.h"
+
+namespace doris {
+class RuntimeProfile;
+} // namespace doris
+
+namespace doris::vectorized {
+
+ArrowReader::ArrowReader(RuntimeState* state, RuntimeProfile* profile, ScannerCounter* counter,
+ const TFileScanRangeParams& params, const TFileRangeDesc& range,
+ const std::vector<SlotDescriptor*>& file_slot_descs, io::IOContext* io_ctx)
+ : _state(state), _range(range), _file_slot_descs(file_slot_descs), _file_reader(nullptr) {}
+
+ArrowReader::~ArrowReader() = default;
+
+Status ArrowReader::init_reader() {
+ RETURN_IF_ERROR(FileFactory::create_pipe_reader(_range.load_id, &_file_reader, _state, false));
+ _pip_stream = ArrowPipInputStream::create_unique(_file_reader);
+ return Status::OK();
+}
+
+Status ArrowReader::get_next_block(Block* block, size_t* read_rows, bool* eof) {
+ bool has_next = false;
+ RETURN_IF_ERROR(_pip_stream->HasNext(&has_next));
+ if (!has_next) {
+ *read_rows = 0;
+ *eof = true;
+ return Status::OK();
+ }
+
+ // create a reader to read data
+ arrow::Result<std::shared_ptr<arrow::ipc::RecordBatchStreamReader>> tRet =
Review Comment:
Why need to create stream reader for each call?
Can it be reused?
--
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] [feture](stream load) (step one)Add arrow data type for stream load [doris]
Posted by "wuwenchi (via GitHub)" <gi...@apache.org>.
wuwenchi commented on PR #26709:
URL: https://github.com/apache/doris/pull/26709#issuecomment-1837382058
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](stream load) (step one)Add arrow data type for stream load [doris]
Posted by "kaka11chen (via GitHub)" <gi...@apache.org>.
kaka11chen commented on code in PR #26709:
URL: https://github.com/apache/doris/pull/26709#discussion_r1417092707
##########
be/src/vec/data_types/serde/data_type_decimal_serde.cpp:
##########
@@ -167,6 +167,8 @@ void DataTypeDecimalSerDe<T>::read_column_from_arrow(IColumn& column,
static_cast<const arrow::DecimalType*>(arrow_array->type().get());
const auto arrow_scale = arrow_decimal_type->scale();
auto& column_data = static_cast<ColumnDecimal<T>&>(column).get_data();
+ // Decimal<Int128> for decimalv2
+ // Decimal<Int128I> for deicmalv3
if constexpr (std::is_same_v<T, Decimal<Int128>>) {
Review Comment:
Use `Decimal128` better.
--
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](stream load) (step one)Add arrow data type for stream load [doris]
Posted by "morningman (via GitHub)" <gi...@apache.org>.
morningman merged PR #26709:
URL: https://github.com/apache/doris/pull/26709
--
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] [feture](stream load) (step one)Add arrow data type for stream load [doris]
Posted by "doris-robot (via GitHub)" <gi...@apache.org>.
doris-robot commented on PR #26709:
URL: https://github.com/apache/doris/pull/26709#issuecomment-1842209135
<details>
<summary>TPC-H test result on machine: 'aliyun_ecs.c7a.8xlarge_32C64G'</summary>
```
Tpch sf100 test result on commit 9fb8797e38e8d0388b4e836cdc0015eaac21bda2, data reload: false
run tpch-sf100 query with default conf and session variables
q1 4698 4435 4440 4435
q2 371 160 154 154
q3 1479 1212 1258 1212
q4 1115 942 898 898
q5 3181 3203 3211 3203
q6 248 132 126 126
q7 982 506 492 492
q8 2242 2248 2197 2197
q9 6705 6806 6689 6689
q10 4519 3243 3289 3243
q11 325 215 198 198
q12 370 214 228 214
q13 17951 3762 3811 3762
q14 249 214 217 214
q15 566 525 521 521
q16 445 392 389 389
q17 1023 616 612 612
q18 7507 7515 7507 7507
q19 1523 1430 1416 1416
q20 1161 345 329 329
q21 3117 2659 2670 2659
q22 352 291 300 291
Total cold run time: 60129 ms
Total hot run time: 40761 ms
run tpch-sf100 query with default conf and set session variable runtime_filter_mode=off
q1 4373 4355 4373 4355
q2 268 161 173 161
q3 3523 3518 3543 3518
q4 2391 2376 2370 2370
q5 5795 5737 5758 5737
q6 241 122 123 122
q7 2387 1863 1844 1844
q8 3527 3530 3531 3530
q9 9096 9012 8988 8988
q10 3936 3980 3987 3980
q11 500 389 385 385
q12 768 590 612 590
q13 4278 3534 3544 3534
q14 292 255 266 255
q15 571 518 528 518
q16 503 451 462 451
q17 1871 1871 1866 1866
q18 8609 8064 8908 8064
q19 1755 1773 1777 1773
q20 2247 1948 1941 1941
q21 6548 6179 6113 6113
q22 518 418 434 418
Total cold run time: 63997 ms
Total hot run time: 60513 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] [feture](stream load) (step one)Add arrow data type for stream load [doris]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on code in PR #26709:
URL: https://github.com/apache/doris/pull/26709#discussion_r1399291548
##########
be/src/vec/exec/format/arrow/arrow_pip_input_stream.h:
##########
@@ -0,0 +1,74 @@
+// 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.
+
+#pragma once
+
+#include <cstddef>
+#include <cstdint>
+#include <memory>
+#include <string>
+#include <unordered_map>
+#include <unordered_set>
+#include <vector>
+
+#include "arrow/buffer.h"
Review Comment:
warning: 'arrow/buffer.h' file not found [clang-diagnostic-error]
```cpp
#include "arrow/buffer.h"
^
```
##########
be/src/vec/exec/format/arrow/arrow_stream_reader.cpp:
##########
@@ -0,0 +1,118 @@
+// 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.
+
+#include "arrow_stream_reader.h"
+
+#include "arrow/array.h"
+#include "arrow/io/buffered.h"
+#include "arrow/io/stdio.h"
+#include "arrow/ipc/options.h"
+#include "arrow/ipc/reader.h"
+#include "arrow/record_batch.h"
+#include "arrow/result.h"
+#include "arrow_pip_input_stream.h"
+#include "common/logging.h"
+#include "io/fs/stream_load_pipe.h"
+#include "olap/wal_manager.h"
+#include "runtime/descriptors.h"
+#include "runtime/runtime_state.h"
+#include "vec/utils/arrow_column_to_doris_column.h"
+
+namespace doris {
+class RuntimeProfile;
+} // namespace doris
+
+namespace doris::vectorized {
+
+ArrowStreamReader::ArrowStreamReader(RuntimeState* state, RuntimeProfile* profile,
+ ScannerCounter* counter, const TFileScanRangeParams& params,
+ const TFileRangeDesc& range,
+ const std::vector<SlotDescriptor*>& file_slot_descs,
+ io::IOContext* io_ctx)
+ : _state(state), _range(range), _file_slot_descs(file_slot_descs), _file_reader(nullptr) {}
+
+ArrowStreamReader::~ArrowStreamReader() = default;
+
+Status ArrowStreamReader::init_reader() {
+ RETURN_IF_ERROR(FileFactory::create_pipe_reader(_range.load_id, &_file_reader, _state, false));
+ _pip_stream = ArrowPipInputStream::create_unique(_file_reader);
+ return Status::OK();
+}
+
+Status ArrowStreamReader::get_next_block(Block* block, size_t* read_rows, bool* eof) {
+ bool has_next = false;
+ RETURN_IF_ERROR(_pip_stream->HasNext(&has_next));
+ if (!has_next) {
+ *read_rows = 0;
+ *eof = true;
+ return Status::OK();
+ }
+
+ // create a reader to read data
+ arrow::Result<std::shared_ptr<arrow::ipc::RecordBatchStreamReader>> tRet =
+ arrow::ipc::RecordBatchStreamReader::Open(_pip_stream.get(),
+ arrow::ipc::IpcReadOptions::Defaults());
+ if (!tRet.ok()) {
+ LOG(ERROR) << "failed to open stream reader: " << tRet.status().message();
+ return Status::InternalError("failed to open stream reader: {}", tRet.status().message());
+ }
+ auto reader = std::move(tRet).ValueUnsafe();
+
+ // get arrow data from reader
+ arrow::Result<arrow::RecordBatchVector> tRet2 = reader->ToRecordBatches();
+ if (!tRet2.ok()) {
+ LOG(ERROR) << "failed to read batch: " << tRet2.status().message();
+ return Status::InternalError("failed to read batch: {}", tRet2.status().message());
+ }
+ std::vector<std::shared_ptr<arrow::RecordBatch>> out_batches = std::move(tRet2).ValueUnsafe();
+
+ // convert arrow batch to block
+ // TODO get tz from schema
+ cctz::time_zone ctzz;
+ TimezoneUtils::find_cctz_time_zone("UTC", ctzz);
+ auto columns = block->mutate_columns();
+ int batch_size = out_batches.size();
+ for (int i = 0; i < batch_size; i++) {
+ arrow::RecordBatch& batch = *out_batches[i];
+ int num_rows = batch.num_rows();
+ int num_columns = batch.num_columns();
+ for (int c = 0; c < num_columns; ++c) {
+ arrow::Array* column = batch.column(c).get();
+
+ std::string column_name = batch.schema()->field(c)->name();
+
+ vectorized::ColumnWithTypeAndName& column_with_name = block->get_by_name(column_name);
+
+ column_with_name.type->get_serde()->read_column_from_arrow(
+ column_with_name.column->assume_mutable_ref(), column, 0, num_rows, ctzz);
+ }
+ *read_rows += batch.num_rows();
+ }
+
+ *eof = (*read_rows == 0);
+ return Status::OK();
+}
+
+Status ArrowStreamReader::get_columns(std::unordered_map<std::string, TypeDescriptor>* name_to_type,
Review Comment:
warning: method 'get_columns' can be made static [readability-convert-member-functions-to-static]
```suggestion
static Status ArrowStreamReader::get_columns(std::unordered_map<std::string, TypeDescriptor>* name_to_type,
```
##########
be/src/vec/exec/format/arrow/arrow_stream_reader.cpp:
##########
@@ -0,0 +1,118 @@
+// 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.
+
+#include "arrow_stream_reader.h"
+
+#include "arrow/array.h"
+#include "arrow/io/buffered.h"
+#include "arrow/io/stdio.h"
+#include "arrow/ipc/options.h"
+#include "arrow/ipc/reader.h"
+#include "arrow/record_batch.h"
+#include "arrow/result.h"
+#include "arrow_pip_input_stream.h"
+#include "common/logging.h"
+#include "io/fs/stream_load_pipe.h"
+#include "olap/wal_manager.h"
+#include "runtime/descriptors.h"
+#include "runtime/runtime_state.h"
+#include "vec/utils/arrow_column_to_doris_column.h"
+
+namespace doris {
+class RuntimeProfile;
+} // namespace doris
+
+namespace doris::vectorized {
+
+ArrowStreamReader::ArrowStreamReader(RuntimeState* state, RuntimeProfile* profile,
+ ScannerCounter* counter, const TFileScanRangeParams& params,
+ const TFileRangeDesc& range,
+ const std::vector<SlotDescriptor*>& file_slot_descs,
+ io::IOContext* io_ctx)
+ : _state(state), _range(range), _file_slot_descs(file_slot_descs), _file_reader(nullptr) {}
+
+ArrowStreamReader::~ArrowStreamReader() = default;
+
+Status ArrowStreamReader::init_reader() {
Review Comment:
warning: method 'init_reader' can be made static [readability-convert-member-functions-to-static]
be/src/vec/exec/format/arrow/arrow_stream_reader.h:58:
```diff
- Status init_reader();
+ static Status init_reader();
```
--
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] [feture](stream load) (step one)Add arrow data type for stream load [doris]
Posted by "wuwenchi (via GitHub)" <gi...@apache.org>.
wuwenchi commented on PR #26709:
URL: https://github.com/apache/doris/pull/26709#issuecomment-1827227910
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] [feture](stream load) (step one)Add arrow data type for stream load [doris]
Posted by "wuwenchi (via GitHub)" <gi...@apache.org>.
wuwenchi commented on PR #26709:
URL: https://github.com/apache/doris/pull/26709#issuecomment-1842153218
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] [feture](stream load) (step one)Add arrow data type for stream load [doris]
Posted by "doris-robot (via GitHub)" <gi...@apache.org>.
doris-robot commented on PR #26709:
URL: https://github.com/apache/doris/pull/26709#issuecomment-1842226940
(From new machine)TeamCity pipeline, clickbench performance test result:
the sum of best hot time: 44.36 seconds
stream load tsv: 563 seconds loaded 74807831229 Bytes, about 126 MB/s
stream load json: 19 seconds loaded 2358488459 Bytes, about 118 MB/s
stream load orc: 66 seconds loaded 1101869774 Bytes, about 15 MB/s
stream load parquet: 33 seconds loaded 861443392 Bytes, about 24 MB/s
insert into select: 28.7 seconds inserted 10000000 Rows, about 348K ops/s
storage size: 17162175113 Bytes
--
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] [feture](stream load) (step one)Add arrow data type for stream load [doris]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on code in PR #26709:
URL: https://github.com/apache/doris/pull/26709#discussion_r1413331684
##########
be/src/vec/exec/format/arrow/arrow_pip_input_stream.cpp:
##########
@@ -0,0 +1,101 @@
+// 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.
+
+#include "arrow_pip_input_stream.h"
+
+#include "arrow/array.h"
+#include "arrow/buffer.h"
+#include "arrow/io/buffered.h"
+#include "arrow/io/stdio.h"
+#include "arrow/ipc/options.h"
+#include "arrow/ipc/reader.h"
+#include "arrow/record_batch.h"
+#include "arrow/result.h"
+#include "common/logging.h"
+#include "io/fs/stream_load_pipe.h"
+#include "olap/wal_manager.h"
+#include "runtime/runtime_state.h"
+
+namespace doris::vectorized {
+
+ArrowPipInputStream::ArrowPipInputStream(io::FileReaderSPtr file_reader)
+ : _file_reader(file_reader), _pos(0), _begin(true), _read_buf(new uint8_t[4]) {
+ set_mode(arrow::io::FileMode::READ);
+}
+
+arrow::Status ArrowPipInputStream::Close() {
+ return arrow::Status::OK();
+}
+
+bool ArrowPipInputStream::closed() const {
Review Comment:
warning: method 'closed' can be made static [readability-convert-member-functions-to-static]
be/src/vec/exec/format/arrow/arrow_pip_input_stream.h:47:
```diff
- bool closed() const override;
+ static bool closed() override;
```
```suggestion
bool ArrowPipInputStream::closed() {
```
--
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] [feture](stream load) (step one)Add arrow data type for stream load [doris]
Posted by "wuwenchi (via GitHub)" <gi...@apache.org>.
wuwenchi commented on PR #26709:
URL: https://github.com/apache/doris/pull/26709#issuecomment-1838862360
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](stream load) (step one)Add arrow data type for stream load [doris]
Posted by "kaka11chen (via GitHub)" <gi...@apache.org>.
kaka11chen commented on code in PR #26709:
URL: https://github.com/apache/doris/pull/26709#discussion_r1417095035
##########
be/src/vec/exec/format/arrow/arrow_stream_reader.cpp:
##########
@@ -0,0 +1,117 @@
+// 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.
+
+#include "arrow_stream_reader.h"
+
+#include "arrow/array.h"
+#include "arrow/io/buffered.h"
+#include "arrow/io/stdio.h"
+#include "arrow/ipc/options.h"
+#include "arrow/ipc/reader.h"
+#include "arrow/record_batch.h"
+#include "arrow/result.h"
+#include "arrow_pip_input_stream.h"
+#include "common/logging.h"
+#include "io/fs/stream_load_pipe.h"
+#include "olap/wal_manager.h"
+#include "runtime/descriptors.h"
+#include "runtime/runtime_state.h"
+#include "vec/utils/arrow_column_to_doris_column.h"
+
+namespace doris {
+class RuntimeProfile;
+} // namespace doris
+
+namespace doris::vectorized {
+
+ArrowStreamReader::ArrowStreamReader(RuntimeState* state, RuntimeProfile* profile,
+ ScannerCounter* counter, const TFileScanRangeParams& params,
+ const TFileRangeDesc& range,
+ const std::vector<SlotDescriptor*>& file_slot_descs,
+ io::IOContext* io_ctx)
+ : _state(state), _range(range), _file_slot_descs(file_slot_descs), _file_reader(nullptr) {
+ TimezoneUtils::find_cctz_time_zone(TimezoneUtils::default_time_zone, _ctzz);
+}
+
+ArrowStreamReader::~ArrowStreamReader() = default;
+
+Status ArrowStreamReader::init_reader() {
+ RETURN_IF_ERROR(FileFactory::create_pipe_reader(_range.load_id, &_file_reader, _state, false));
+ _pip_stream = ArrowPipInputStream::create_unique(_file_reader);
+ return Status::OK();
+}
+
+Status ArrowStreamReader::get_next_block(Block* block, size_t* read_rows, bool* eof) {
+ bool has_next = false;
+ RETURN_IF_ERROR(_pip_stream->HasNext(&has_next));
+ if (!has_next) {
+ *read_rows = 0;
+ *eof = true;
+ return Status::OK();
+ }
+
+ // create a reader to read data
+ arrow::Result<std::shared_ptr<arrow::ipc::RecordBatchStreamReader>> tRet =
+ arrow::ipc::RecordBatchStreamReader::Open(_pip_stream.get(),
+ arrow::ipc::IpcReadOptions::Defaults());
+ if (!tRet.ok()) {
+ LOG(WARNING) << "failed to open stream reader: " << tRet.status().message();
+ return Status::InternalError("failed to open stream reader: {}", tRet.status().message());
+ }
+ auto reader = std::move(tRet).ValueUnsafe();
+
+ // get arrow data from reader
+ arrow::Result<arrow::RecordBatchVector> tRet2 = reader->ToRecordBatches();
Review Comment:
`tRet2` need a better 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] [feture](stream load) (step one)Add arrow data type for stream load [doris]
Posted by "wuwenchi (via GitHub)" <gi...@apache.org>.
wuwenchi commented on code in PR #26709:
URL: https://github.com/apache/doris/pull/26709#discussion_r1398634374
##########
be/src/vec/exec/format/arrow/arrow_reader.cpp:
##########
@@ -0,0 +1,115 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include "arrow_reader.h"
+
+#include "arrow/array.h"
+#include "arrow/io/buffered.h"
+#include "arrow/io/stdio.h"
+#include "arrow/ipc/options.h"
+#include "arrow/ipc/reader.h"
+#include "arrow/record_batch.h"
+#include "arrow/result.h"
+#include "arrow_pip_input_stream.h"
+#include "common/logging.h"
+#include "io/fs/stream_load_pipe.h"
+#include "olap/wal_manager.h"
+#include "runtime/descriptors.h"
+#include "runtime/runtime_state.h"
+#include "vec/utils/arrow_column_to_doris_column.h"
+
+namespace doris {
+class RuntimeProfile;
+} // namespace doris
+
+namespace doris::vectorized {
+
+ArrowReader::ArrowReader(RuntimeState* state, RuntimeProfile* profile, ScannerCounter* counter,
+ const TFileScanRangeParams& params, const TFileRangeDesc& range,
+ const std::vector<SlotDescriptor*>& file_slot_descs, io::IOContext* io_ctx)
+ : _state(state), _range(range), _file_slot_descs(file_slot_descs), _file_reader(nullptr) {}
+
+ArrowReader::~ArrowReader() = default;
+
+Status ArrowReader::init_reader() {
+ RETURN_IF_ERROR(FileFactory::create_pipe_reader(_range.load_id, &_file_reader, _state, false));
+ _pip_stream = ArrowPipInputStream::create_unique(_file_reader);
+ return Status::OK();
+}
+
+Status ArrowReader::get_next_block(Block* block, size_t* read_rows, bool* eof) {
+ bool has_next = false;
+ RETURN_IF_ERROR(_pip_stream->HasNext(&has_next));
+ if (!has_next) {
+ *read_rows = 0;
+ *eof = true;
+ return Status::OK();
+ }
+
+ // create a reader to read data
+ arrow::Result<std::shared_ptr<arrow::ipc::RecordBatchStreamReader>> tRet =
+ arrow::ipc::RecordBatchStreamReader::Open(_pip_stream.get(),
+ arrow::ipc::IpcReadOptions::Defaults());
+ if (!tRet.ok()) {
+ LOG(ERROR) << "failed to open stream reader: " << tRet.status().message();
+ return Status::InternalError("failed to open stream reader: {}", tRet.status().message());
+ }
+ auto reader = std::move(tRet).ValueUnsafe();
+
+ // get arrow data from reader
+ arrow::Result<arrow::RecordBatchVector> tRet2 = reader->ToRecordBatches();
+ if (!tRet2.ok()) {
+ LOG(ERROR) << "failed to read batch: " << tRet2.status().message();
+ return Status::InternalError("failed to read batch: {}", tRet2.status().message());
+ }
+ std::vector<std::shared_ptr<arrow::RecordBatch>> out_batches = std::move(tRet2).ValueUnsafe();
+
+ // convert arrow batch to block
+ cctz::time_zone ctzz;
+ TimezoneUtils::find_cctz_time_zone("UTC", ctzz);
Review Comment:
Since the timestamp type is not yet supported, this is just a placeholder. This value will be modified later based on the schema information.
We can add a TODO here.
--
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] [feture](stream load) (step one)Add arrow data type for stream load [doris]
Posted by "wuwenchi (via GitHub)" <gi...@apache.org>.
wuwenchi commented on PR #26709:
URL: https://github.com/apache/doris/pull/26709#issuecomment-1826199158
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] [feture](stream load) (step one)Add arrow data type for stream load [doris]
Posted by "wuwenchi (via GitHub)" <gi...@apache.org>.
wuwenchi commented on PR #26709:
URL: https://github.com/apache/doris/pull/26709#issuecomment-1837738102
run compile
--
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] [feture](stream load) (step one)Add arrow data type for stream load [doris]
Posted by "wuwenchi (via GitHub)" <gi...@apache.org>.
wuwenchi commented on PR #26709:
URL: https://github.com/apache/doris/pull/26709#issuecomment-1837797757
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] [feture](stream load) (step one)Add arrow data type for stream load [doris]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #26709:
URL: https://github.com/apache/doris/pull/26709#issuecomment-1831461449
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] [feture](stream load) (step one)Add arrow data type for stream load [doris]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on code in PR #26709:
URL: https://github.com/apache/doris/pull/26709#discussion_r1396836397
##########
be/src/vec/exec/format/arrow/batch_with_length_reader.cpp:
##########
@@ -0,0 +1,226 @@
+// 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.
+
+#include "batch_with_length_reader.h"
+
+#include "arrow/array.h"
+#include "arrow/io/buffered.h"
+#include "arrow/io/stdio.h"
+#include "arrow/ipc/options.h"
+#include "arrow/ipc/reader.h"
+#include "arrow/record_batch.h"
+#include "arrow/result.h"
+#include "common/logging.h"
+#include "io/fs/stream_load_pipe.h"
+#include "olap/wal_manager.h"
+#include "runtime/runtime_state.h"
+#include "arrow/result.h"
Review Comment:
warning: duplicate include [readability-duplicate-include]
be/src/vec/exec/format/arrow/batch_with_length_reader.cpp:29:
```diff
- #include "runtime/runtime_state.h"
- #include "arrow/result.h"
+ #include "runtime/runtime_state.h"
```
--
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] [feture](stream load) (step one)Add arrow data type for stream load [doris]
Posted by "morningman (via GitHub)" <gi...@apache.org>.
morningman commented on PR #26709:
URL: https://github.com/apache/doris/pull/26709#issuecomment-1831642117
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] [feture](stream load) (step one)Add arrow data type for stream load [doris]
Posted by "wuwenchi (via GitHub)" <gi...@apache.org>.
wuwenchi commented on PR #26709:
URL: https://github.com/apache/doris/pull/26709#issuecomment-1837782679
run compile
--
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] [feture](stream load) (step one)Add arrow data type for stream load [doris]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on code in PR #26709:
URL: https://github.com/apache/doris/pull/26709#discussion_r1398639981
##########
be/src/vec/exec/format/arrow/arrow_pip_input_stream.h:
##########
@@ -0,0 +1,76 @@
+// 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.
+
+#pragma once
+
+#include <gen_cpp/internal_service.pb.h>
Review Comment:
warning: 'gen_cpp/internal_service.pb.h' file not found [clang-diagnostic-error]
```cpp
#include <gen_cpp/internal_service.pb.h>
^
```
##########
be/src/vec/exec/format/arrow/arrow_stream_reader.h:
##########
@@ -0,0 +1,78 @@
+// 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.
+
+#pragma once
+
+#include <gen_cpp/PlanNodes_types.h>
+#include <gen_cpp/internal_service.pb.h>
+#include <stddef.h>
Review Comment:
warning: inclusion of deprecated C++ header 'stddef.h'; consider using 'cstddef' instead [modernize-deprecated-headers]
```suggestion
#include <cstddef>
```
##########
be/src/vec/exec/format/arrow/arrow_stream_reader.h:
##########
@@ -0,0 +1,78 @@
+// 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.
+
+#pragma once
+
+#include <gen_cpp/PlanNodes_types.h>
+#include <gen_cpp/internal_service.pb.h>
+#include <stddef.h>
+#include <stdint.h>
Review Comment:
warning: inclusion of deprecated C++ header 'stdint.h'; consider using 'cstdint' instead [modernize-deprecated-headers]
```suggestion
#include <cstdint>
```
##########
be/src/vec/exec/format/arrow/arrow_stream_reader.h:
##########
@@ -0,0 +1,78 @@
+// 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.
+
+#pragma once
+
+#include <gen_cpp/PlanNodes_types.h>
Review Comment:
warning: 'gen_cpp/PlanNodes_types.h' file not found [clang-diagnostic-error]
```cpp
#include <gen_cpp/PlanNodes_types.h>
^
```
##########
be/src/vec/exec/format/arrow/arrow_stream_reader.cpp:
##########
@@ -0,0 +1,118 @@
+// 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.
+
+#include "arrow_stream_reader.h"
+
+#include "arrow/array.h"
+#include "arrow/io/buffered.h"
+#include "arrow/io/stdio.h"
+#include "arrow/ipc/options.h"
+#include "arrow/ipc/reader.h"
+#include "arrow/record_batch.h"
+#include "arrow/result.h"
+#include "arrow_pip_input_stream.h"
+#include "common/logging.h"
+#include "io/fs/stream_load_pipe.h"
+#include "olap/wal_manager.h"
+#include "runtime/descriptors.h"
+#include "runtime/runtime_state.h"
+#include "vec/utils/arrow_column_to_doris_column.h"
+
+namespace doris {
+class RuntimeProfile;
+} // namespace doris
+
+namespace doris::vectorized {
+
+ArrowStreamReader::ArrowStreamReader(RuntimeState* state, RuntimeProfile* profile,
+ ScannerCounter* counter, const TFileScanRangeParams& params,
+ const TFileRangeDesc& range,
+ const std::vector<SlotDescriptor*>& file_slot_descs,
+ io::IOContext* io_ctx)
+ : _state(state), _range(range), _file_slot_descs(file_slot_descs), _file_reader(nullptr) {}
+
+ArrowStreamReader::~ArrowStreamReader() = default;
+
+Status ArrowStreamReader::init_reader() {
Review Comment:
warning: method 'init_reader' can be made static [readability-convert-member-functions-to-static]
be/src/vec/exec/format/arrow/arrow_stream_reader.h:62:
```diff
- Status init_reader();
+ static Status init_reader();
```
--
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] [feture](stream load)Add arrow data type for stream load [doris]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on code in PR #26709:
URL: https://github.com/apache/doris/pull/26709#discussion_r1387953417
##########
be/src/vec/exec/format/arrow/arrow_reader.cpp:
##########
@@ -0,0 +1,119 @@
+// 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.
+
+#include "arrow_reader.h"
+
+#include "common/logging.h"
+#include "olap/wal_manager.h"
+#include "runtime/runtime_state.h"
+#include "arrow/array.h"
+#include "arrow/ipc/reader.h"
+#include "arrow/io/buffered.h"
+#include "arrow/io/stdio.h"
+#include "arrow/result.h"
+#include "arrow/ipc/options.h"
+#include "arrow/record_batch.h"
+#include "io/fs/stream_load_pipe.h"
+
+namespace doris {
+class RuntimeProfile;
+} // namespace vectorized
+
+namespace doris::vectorized {
+
+ArrowReader::ArrowReader(RuntimeState* state, RuntimeProfile* profile, ScannerCounter* counter,
+ const TFileScanRangeParams& params, const TFileRangeDesc& range,
+ const std::vector<SlotDescriptor*>& file_slot_descs, io::IOContext* io_ctx)
+ // {
+ : _state(state),
+ // _profile(profile),
+ // _counter(counter),
+ // _params(params),
+ _range(range),
+ _file_reader(nullptr) {
+
+}
+
+ArrowReader::~ArrowReader() = default;
+
+Status ArrowReader::init_reader() {
Review Comment:
warning: method 'init_reader' can be made static [readability-convert-member-functions-to-static]
be/src/vec/exec/format/arrow/arrow_reader.h:62:
```diff
- Status init_reader();
+ static Status init_reader();
```
##########
be/src/vec/exec/format/arrow/arrow_reader.h:
##########
@@ -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.
+
+#pragma once
+
+#include <gen_cpp/PlanNodes_types.h>
Review Comment:
warning: 'gen_cpp/PlanNodes_types.h' file not found [clang-diagnostic-error]
```cpp
#include <gen_cpp/PlanNodes_types.h>
^
```
##########
be/src/vec/exec/format/arrow/arrow_reader.h:
##########
@@ -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.
+
+#pragma once
+
+#include <gen_cpp/PlanNodes_types.h>
+#include <gen_cpp/internal_service.pb.h>
+#include <stddef.h>
+#include <stdint.h>
Review Comment:
warning: inclusion of deprecated C++ header 'stdint.h'; consider using 'cstdint' instead [modernize-deprecated-headers]
```suggestion
#include <cstdint>
```
##########
be/src/vec/exec/format/arrow/arrow_reader.h:
##########
@@ -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.
+
+#pragma once
+
+#include <gen_cpp/PlanNodes_types.h>
+#include <gen_cpp/internal_service.pb.h>
+#include <stddef.h>
Review Comment:
warning: inclusion of deprecated C++ header 'stddef.h'; consider using 'cstddef' instead [modernize-deprecated-headers]
```suggestion
#include <cstddef>
```
--
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] [feture](stream load)Add arrow data type for stream load [doris]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on code in PR #26709:
URL: https://github.com/apache/doris/pull/26709#discussion_r1392553224
##########
be/src/vec/exec/format/arrow/batch_with_length_reader.cpp:
##########
@@ -0,0 +1,147 @@
+// 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.
+
+#include "batch_with_length_reader.h"
+
+#include "arrow/array.h"
+#include "arrow/io/buffered.h"
+#include "arrow/io/stdio.h"
+#include "arrow/ipc/options.h"
+#include "arrow/ipc/reader.h"
+#include "arrow/record_batch.h"
+#include "arrow/result.h"
+#include "common/logging.h"
+#include "io/fs/stream_load_pipe.h"
+#include "olap/wal_manager.h"
+#include "runtime/runtime_state.h"
+
+#define DEFAULT_READ_BUF_CAP (4 * 1024 * 1024)
+#define BATCH_SIZE_LENGTH (4)
Review Comment:
warning: macro 'BATCH_SIZE_LENGTH' defines an integral constant; prefer an enum instead [modernize-macro-to-enum]
```cpp
#define BATCH_SIZE_LENGTH (4)
^
```
##########
be/src/vec/exec/format/arrow/batch_with_length_reader.cpp:
##########
@@ -0,0 +1,147 @@
+// 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.
+
+#include "batch_with_length_reader.h"
+
+#include "arrow/array.h"
+#include "arrow/io/buffered.h"
+#include "arrow/io/stdio.h"
+#include "arrow/ipc/options.h"
+#include "arrow/ipc/reader.h"
+#include "arrow/record_batch.h"
+#include "arrow/result.h"
+#include "common/logging.h"
+#include "io/fs/stream_load_pipe.h"
+#include "olap/wal_manager.h"
+#include "runtime/runtime_state.h"
+
+#define DEFAULT_READ_BUF_CAP (4 * 1024 * 1024)
Review Comment:
warning: macro 'DEFAULT_READ_BUF_CAP' defines an integral constant; prefer an enum instead [modernize-macro-to-enum]
```cpp
#define DEFAULT_READ_BUF_CAP (4 * 1024 * 1024)
^
```
##########
be/src/vec/exec/format/arrow/batch_with_length_reader.cpp:
##########
@@ -0,0 +1,147 @@
+// 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.
+
+#include "batch_with_length_reader.h"
+
+#include "arrow/array.h"
+#include "arrow/io/buffered.h"
+#include "arrow/io/stdio.h"
+#include "arrow/ipc/options.h"
+#include "arrow/ipc/reader.h"
+#include "arrow/record_batch.h"
+#include "arrow/result.h"
+#include "common/logging.h"
+#include "io/fs/stream_load_pipe.h"
+#include "olap/wal_manager.h"
+#include "runtime/runtime_state.h"
+
+#define DEFAULT_READ_BUF_CAP (4 * 1024 * 1024)
+#define BATCH_SIZE_LENGTH (4)
+
Review Comment:
warning: replace macro with enum [modernize-macro-to-enum]
```suggestion
enum {
DEFAULT_READ_BUF_CAP = (4 * 1024 * 1024),
BATCH_SIZE_LENGTH = (4)};
```
##########
be/src/vec/exec/format/arrow/batch_with_length_reader.cpp:
##########
@@ -0,0 +1,147 @@
+// 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.
+
+#include "batch_with_length_reader.h"
+
+#include "arrow/array.h"
+#include "arrow/io/buffered.h"
+#include "arrow/io/stdio.h"
+#include "arrow/ipc/options.h"
+#include "arrow/ipc/reader.h"
+#include "arrow/record_batch.h"
+#include "arrow/result.h"
+#include "common/logging.h"
+#include "io/fs/stream_load_pipe.h"
+#include "olap/wal_manager.h"
+#include "runtime/runtime_state.h"
+
+#define DEFAULT_READ_BUF_CAP (4 * 1024 * 1024)
+#define BATCH_SIZE_LENGTH (4)
+
+namespace doris::vectorized {
+
+BatchWithLengthReader::BatchWithLengthReader(io::FileReaderSPtr file_reader)
+ : _file_reader(file_reader),
+ _read_buf(new uint8_t[DEFAULT_READ_BUF_CAP]),
+ _read_buf_cap(DEFAULT_READ_BUF_CAP),
+ _read_buf_pos(0),
+ _read_buf_len(0),
+ _batch_size(-1) {}
+
+BatchWithLengthReader::~BatchWithLengthReader() = default;
+
+Status BatchWithLengthReader::get_one_batch(uint8_t** data, int* length) {
+ _init();
+ RETURN_IF_ERROR(_get_batch_size());
+ RETURN_IF_ERROR(_get_batch_value());
+
+ *data = &_read_buf[_read_buf_pos];
+ if (_batch_size <= 0) {
+ *length = 0;
+ } else {
+ *length = _batch_size;
+ }
+ return Status::OK();
+}
+
+void BatchWithLengthReader::_init() {
+ if (_batch_size > 0) {
+ _read_buf_pos += _batch_size;
+ _read_buf_len -= _batch_size;
+ }
+ _batch_size = -1;
+}
+
+Status BatchWithLengthReader::_get_batch_size() {
+ if (_batch_size >= 0) {
+ return Status::OK();
+ }
+
+ _ensure_cap(BATCH_SIZE_LENGTH);
+
+ RETURN_IF_ERROR(_get_data_from_reader(BATCH_SIZE_LENGTH));
+ if (_read_buf_len < BATCH_SIZE_LENGTH) {
+ return Status::OK();
+ }
+
+ _batch_size = _convert_to_length(_read_buf + _read_buf_pos);
+
+ if (_batch_size < 0) {
+ return Status::InternalError("Invalid batch size");
+ }
+
+ _read_buf_pos += BATCH_SIZE_LENGTH;
+ _read_buf_len -= BATCH_SIZE_LENGTH;
+
+ if (_batch_size > _read_buf_cap) {
+ while (_read_buf_cap < _batch_size) {
+ _read_buf_cap *= 2;
+ }
+
+ // The current buff capacity is not enough and needs to be expanded
+ auto* new_read_buf = new uint8_t[_read_buf_cap];
+ memmove(new_read_buf, _read_buf + _read_buf_pos, _read_buf_len);
+ delete[] _read_buf;
+ _read_buf = new_read_buf;
+ _read_buf_pos = 0;
+ }
+ return Status::OK();
+}
+
+Status BatchWithLengthReader::_get_data_from_reader(int req_size) {
+ while (_read_buf_len < req_size) {
+ Slice file_slice(_read_buf + _read_buf_pos + _read_buf_len,
+ _read_buf_cap - _read_buf_pos - _read_buf_len);
+ size_t read_length = 0;
+ RETURN_IF_ERROR(_file_reader->read_at(0, file_slice, &read_length, NULL));
+ _read_buf_len += read_length;
+ if (read_length == 0) {
+ break;
+ }
+ }
+ return Status::OK();
+}
+
+void BatchWithLengthReader::_ensure_cap(int req_size) {
+ if (_read_buf_len < req_size && _read_buf_cap - _read_buf_pos < req_size) {
+ // there is not enough space for data,
+ // we need to move the data that has been used before and the useful data forward
+ memmove(_read_buf, _read_buf + _read_buf_pos, _read_buf_len);
+ _read_buf_pos = 0;
+ }
+}
+
+uint32_t BatchWithLengthReader::_convert_to_length(const uint8_t* data) {
+ uint32_t len = (((uint32_t)data[0]) << 24) & 0xff000000;
Review Comment:
warning: 24 is a magic number; consider replacing it with a named constant [readability-magic-numbers]
```cpp
uint32_t len = (((uint32_t)data[0]) << 24) & 0xff000000;
^
```
##########
be/src/vec/exec/format/arrow/batch_with_length_reader.cpp:
##########
@@ -0,0 +1,147 @@
+// 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.
+
+#include "batch_with_length_reader.h"
+
+#include "arrow/array.h"
+#include "arrow/io/buffered.h"
+#include "arrow/io/stdio.h"
+#include "arrow/ipc/options.h"
+#include "arrow/ipc/reader.h"
+#include "arrow/record_batch.h"
+#include "arrow/result.h"
+#include "common/logging.h"
+#include "io/fs/stream_load_pipe.h"
+#include "olap/wal_manager.h"
+#include "runtime/runtime_state.h"
+
+#define DEFAULT_READ_BUF_CAP (4 * 1024 * 1024)
+#define BATCH_SIZE_LENGTH (4)
+
+namespace doris::vectorized {
+
+BatchWithLengthReader::BatchWithLengthReader(io::FileReaderSPtr file_reader)
+ : _file_reader(file_reader),
+ _read_buf(new uint8_t[DEFAULT_READ_BUF_CAP]),
+ _read_buf_cap(DEFAULT_READ_BUF_CAP),
+ _read_buf_pos(0),
+ _read_buf_len(0),
+ _batch_size(-1) {}
+
+BatchWithLengthReader::~BatchWithLengthReader() = default;
+
+Status BatchWithLengthReader::get_one_batch(uint8_t** data, int* length) {
+ _init();
+ RETURN_IF_ERROR(_get_batch_size());
+ RETURN_IF_ERROR(_get_batch_value());
+
+ *data = &_read_buf[_read_buf_pos];
+ if (_batch_size <= 0) {
+ *length = 0;
+ } else {
+ *length = _batch_size;
+ }
+ return Status::OK();
+}
+
+void BatchWithLengthReader::_init() {
+ if (_batch_size > 0) {
+ _read_buf_pos += _batch_size;
+ _read_buf_len -= _batch_size;
+ }
+ _batch_size = -1;
+}
+
+Status BatchWithLengthReader::_get_batch_size() {
+ if (_batch_size >= 0) {
+ return Status::OK();
+ }
+
+ _ensure_cap(BATCH_SIZE_LENGTH);
+
+ RETURN_IF_ERROR(_get_data_from_reader(BATCH_SIZE_LENGTH));
+ if (_read_buf_len < BATCH_SIZE_LENGTH) {
+ return Status::OK();
+ }
+
+ _batch_size = _convert_to_length(_read_buf + _read_buf_pos);
+
+ if (_batch_size < 0) {
+ return Status::InternalError("Invalid batch size");
+ }
+
+ _read_buf_pos += BATCH_SIZE_LENGTH;
+ _read_buf_len -= BATCH_SIZE_LENGTH;
+
+ if (_batch_size > _read_buf_cap) {
+ while (_read_buf_cap < _batch_size) {
+ _read_buf_cap *= 2;
+ }
+
+ // The current buff capacity is not enough and needs to be expanded
+ auto* new_read_buf = new uint8_t[_read_buf_cap];
+ memmove(new_read_buf, _read_buf + _read_buf_pos, _read_buf_len);
+ delete[] _read_buf;
+ _read_buf = new_read_buf;
+ _read_buf_pos = 0;
+ }
+ return Status::OK();
+}
+
+Status BatchWithLengthReader::_get_data_from_reader(int req_size) {
+ while (_read_buf_len < req_size) {
+ Slice file_slice(_read_buf + _read_buf_pos + _read_buf_len,
+ _read_buf_cap - _read_buf_pos - _read_buf_len);
+ size_t read_length = 0;
+ RETURN_IF_ERROR(_file_reader->read_at(0, file_slice, &read_length, NULL));
+ _read_buf_len += read_length;
+ if (read_length == 0) {
+ break;
+ }
+ }
+ return Status::OK();
+}
+
+void BatchWithLengthReader::_ensure_cap(int req_size) {
+ if (_read_buf_len < req_size && _read_buf_cap - _read_buf_pos < req_size) {
+ // there is not enough space for data,
+ // we need to move the data that has been used before and the useful data forward
+ memmove(_read_buf, _read_buf + _read_buf_pos, _read_buf_len);
+ _read_buf_pos = 0;
+ }
+}
+
+uint32_t BatchWithLengthReader::_convert_to_length(const uint8_t* data) {
+ uint32_t len = (((uint32_t)data[0]) << 24) & 0xff000000;
+ len |= (((uint32_t)data[1]) << 16) & 0xff0000;
Review Comment:
warning: 0xff0000 is a magic number; consider replacing it with a named constant [readability-magic-numbers]
```cpp
len |= (((uint32_t)data[1]) << 16) & 0xff0000;
^
```
##########
be/src/vec/exec/format/arrow/batch_with_length_reader.cpp:
##########
@@ -0,0 +1,147 @@
+// 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.
+
+#include "batch_with_length_reader.h"
+
+#include "arrow/array.h"
+#include "arrow/io/buffered.h"
+#include "arrow/io/stdio.h"
+#include "arrow/ipc/options.h"
+#include "arrow/ipc/reader.h"
+#include "arrow/record_batch.h"
+#include "arrow/result.h"
+#include "common/logging.h"
+#include "io/fs/stream_load_pipe.h"
+#include "olap/wal_manager.h"
+#include "runtime/runtime_state.h"
+
+#define DEFAULT_READ_BUF_CAP (4 * 1024 * 1024)
+#define BATCH_SIZE_LENGTH (4)
+
+namespace doris::vectorized {
+
+BatchWithLengthReader::BatchWithLengthReader(io::FileReaderSPtr file_reader)
+ : _file_reader(file_reader),
+ _read_buf(new uint8_t[DEFAULT_READ_BUF_CAP]),
+ _read_buf_cap(DEFAULT_READ_BUF_CAP),
+ _read_buf_pos(0),
+ _read_buf_len(0),
+ _batch_size(-1) {}
+
+BatchWithLengthReader::~BatchWithLengthReader() = default;
+
+Status BatchWithLengthReader::get_one_batch(uint8_t** data, int* length) {
+ _init();
+ RETURN_IF_ERROR(_get_batch_size());
+ RETURN_IF_ERROR(_get_batch_value());
+
+ *data = &_read_buf[_read_buf_pos];
+ if (_batch_size <= 0) {
+ *length = 0;
+ } else {
+ *length = _batch_size;
+ }
+ return Status::OK();
+}
+
+void BatchWithLengthReader::_init() {
+ if (_batch_size > 0) {
+ _read_buf_pos += _batch_size;
+ _read_buf_len -= _batch_size;
+ }
+ _batch_size = -1;
+}
+
+Status BatchWithLengthReader::_get_batch_size() {
+ if (_batch_size >= 0) {
+ return Status::OK();
+ }
+
+ _ensure_cap(BATCH_SIZE_LENGTH);
+
+ RETURN_IF_ERROR(_get_data_from_reader(BATCH_SIZE_LENGTH));
+ if (_read_buf_len < BATCH_SIZE_LENGTH) {
+ return Status::OK();
+ }
+
+ _batch_size = _convert_to_length(_read_buf + _read_buf_pos);
+
+ if (_batch_size < 0) {
+ return Status::InternalError("Invalid batch size");
+ }
+
+ _read_buf_pos += BATCH_SIZE_LENGTH;
+ _read_buf_len -= BATCH_SIZE_LENGTH;
+
+ if (_batch_size > _read_buf_cap) {
+ while (_read_buf_cap < _batch_size) {
+ _read_buf_cap *= 2;
+ }
+
+ // The current buff capacity is not enough and needs to be expanded
+ auto* new_read_buf = new uint8_t[_read_buf_cap];
+ memmove(new_read_buf, _read_buf + _read_buf_pos, _read_buf_len);
+ delete[] _read_buf;
+ _read_buf = new_read_buf;
+ _read_buf_pos = 0;
+ }
+ return Status::OK();
+}
+
+Status BatchWithLengthReader::_get_data_from_reader(int req_size) {
+ while (_read_buf_len < req_size) {
+ Slice file_slice(_read_buf + _read_buf_pos + _read_buf_len,
+ _read_buf_cap - _read_buf_pos - _read_buf_len);
+ size_t read_length = 0;
+ RETURN_IF_ERROR(_file_reader->read_at(0, file_slice, &read_length, NULL));
+ _read_buf_len += read_length;
+ if (read_length == 0) {
+ break;
+ }
+ }
+ return Status::OK();
+}
+
+void BatchWithLengthReader::_ensure_cap(int req_size) {
+ if (_read_buf_len < req_size && _read_buf_cap - _read_buf_pos < req_size) {
+ // there is not enough space for data,
+ // we need to move the data that has been used before and the useful data forward
+ memmove(_read_buf, _read_buf + _read_buf_pos, _read_buf_len);
+ _read_buf_pos = 0;
+ }
+}
+
+uint32_t BatchWithLengthReader::_convert_to_length(const uint8_t* data) {
+ uint32_t len = (((uint32_t)data[0]) << 24) & 0xff000000;
+ len |= (((uint32_t)data[1]) << 16) & 0xff0000;
+ len |= (((uint32_t)data[2]) << 8) & 0xff00;
Review Comment:
warning: 0xff00 is a magic number; consider replacing it with a named constant [readability-magic-numbers]
```cpp
len |= (((uint32_t)data[2]) << 8) & 0xff00;
^
```
##########
be/src/vec/exec/format/arrow/batch_with_length_reader.cpp:
##########
@@ -0,0 +1,147 @@
+// 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.
+
+#include "batch_with_length_reader.h"
+
+#include "arrow/array.h"
+#include "arrow/io/buffered.h"
+#include "arrow/io/stdio.h"
+#include "arrow/ipc/options.h"
+#include "arrow/ipc/reader.h"
+#include "arrow/record_batch.h"
+#include "arrow/result.h"
+#include "common/logging.h"
+#include "io/fs/stream_load_pipe.h"
+#include "olap/wal_manager.h"
+#include "runtime/runtime_state.h"
+
+#define DEFAULT_READ_BUF_CAP (4 * 1024 * 1024)
+#define BATCH_SIZE_LENGTH (4)
+
+namespace doris::vectorized {
+
+BatchWithLengthReader::BatchWithLengthReader(io::FileReaderSPtr file_reader)
+ : _file_reader(file_reader),
+ _read_buf(new uint8_t[DEFAULT_READ_BUF_CAP]),
+ _read_buf_cap(DEFAULT_READ_BUF_CAP),
+ _read_buf_pos(0),
+ _read_buf_len(0),
+ _batch_size(-1) {}
+
+BatchWithLengthReader::~BatchWithLengthReader() = default;
+
+Status BatchWithLengthReader::get_one_batch(uint8_t** data, int* length) {
+ _init();
+ RETURN_IF_ERROR(_get_batch_size());
+ RETURN_IF_ERROR(_get_batch_value());
+
+ *data = &_read_buf[_read_buf_pos];
+ if (_batch_size <= 0) {
+ *length = 0;
+ } else {
+ *length = _batch_size;
+ }
+ return Status::OK();
+}
+
+void BatchWithLengthReader::_init() {
+ if (_batch_size > 0) {
+ _read_buf_pos += _batch_size;
+ _read_buf_len -= _batch_size;
+ }
+ _batch_size = -1;
+}
+
+Status BatchWithLengthReader::_get_batch_size() {
+ if (_batch_size >= 0) {
+ return Status::OK();
+ }
+
+ _ensure_cap(BATCH_SIZE_LENGTH);
+
+ RETURN_IF_ERROR(_get_data_from_reader(BATCH_SIZE_LENGTH));
+ if (_read_buf_len < BATCH_SIZE_LENGTH) {
+ return Status::OK();
+ }
+
+ _batch_size = _convert_to_length(_read_buf + _read_buf_pos);
+
+ if (_batch_size < 0) {
+ return Status::InternalError("Invalid batch size");
+ }
+
+ _read_buf_pos += BATCH_SIZE_LENGTH;
+ _read_buf_len -= BATCH_SIZE_LENGTH;
+
+ if (_batch_size > _read_buf_cap) {
+ while (_read_buf_cap < _batch_size) {
+ _read_buf_cap *= 2;
+ }
+
+ // The current buff capacity is not enough and needs to be expanded
+ auto* new_read_buf = new uint8_t[_read_buf_cap];
+ memmove(new_read_buf, _read_buf + _read_buf_pos, _read_buf_len);
+ delete[] _read_buf;
+ _read_buf = new_read_buf;
+ _read_buf_pos = 0;
+ }
+ return Status::OK();
+}
+
+Status BatchWithLengthReader::_get_data_from_reader(int req_size) {
+ while (_read_buf_len < req_size) {
+ Slice file_slice(_read_buf + _read_buf_pos + _read_buf_len,
+ _read_buf_cap - _read_buf_pos - _read_buf_len);
+ size_t read_length = 0;
+ RETURN_IF_ERROR(_file_reader->read_at(0, file_slice, &read_length, NULL));
+ _read_buf_len += read_length;
+ if (read_length == 0) {
+ break;
+ }
+ }
+ return Status::OK();
+}
+
+void BatchWithLengthReader::_ensure_cap(int req_size) {
+ if (_read_buf_len < req_size && _read_buf_cap - _read_buf_pos < req_size) {
+ // there is not enough space for data,
+ // we need to move the data that has been used before and the useful data forward
+ memmove(_read_buf, _read_buf + _read_buf_pos, _read_buf_len);
+ _read_buf_pos = 0;
+ }
+}
+
+uint32_t BatchWithLengthReader::_convert_to_length(const uint8_t* data) {
+ uint32_t len = (((uint32_t)data[0]) << 24) & 0xff000000;
+ len |= (((uint32_t)data[1]) << 16) & 0xff0000;
+ len |= (((uint32_t)data[2]) << 8) & 0xff00;
Review Comment:
warning: 8 is a magic number; consider replacing it with a named constant [readability-magic-numbers]
```cpp
len |= (((uint32_t)data[2]) << 8) & 0xff00;
^
```
##########
be/src/vec/exec/format/arrow/batch_with_length_reader.cpp:
##########
@@ -0,0 +1,147 @@
+// 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.
+
+#include "batch_with_length_reader.h"
+
+#include "arrow/array.h"
+#include "arrow/io/buffered.h"
+#include "arrow/io/stdio.h"
+#include "arrow/ipc/options.h"
+#include "arrow/ipc/reader.h"
+#include "arrow/record_batch.h"
+#include "arrow/result.h"
+#include "common/logging.h"
+#include "io/fs/stream_load_pipe.h"
+#include "olap/wal_manager.h"
+#include "runtime/runtime_state.h"
+
+#define DEFAULT_READ_BUF_CAP (4 * 1024 * 1024)
+#define BATCH_SIZE_LENGTH (4)
+
+namespace doris::vectorized {
+
+BatchWithLengthReader::BatchWithLengthReader(io::FileReaderSPtr file_reader)
+ : _file_reader(file_reader),
+ _read_buf(new uint8_t[DEFAULT_READ_BUF_CAP]),
+ _read_buf_cap(DEFAULT_READ_BUF_CAP),
+ _read_buf_pos(0),
+ _read_buf_len(0),
+ _batch_size(-1) {}
+
+BatchWithLengthReader::~BatchWithLengthReader() = default;
+
+Status BatchWithLengthReader::get_one_batch(uint8_t** data, int* length) {
+ _init();
+ RETURN_IF_ERROR(_get_batch_size());
+ RETURN_IF_ERROR(_get_batch_value());
+
+ *data = &_read_buf[_read_buf_pos];
+ if (_batch_size <= 0) {
+ *length = 0;
+ } else {
+ *length = _batch_size;
+ }
+ return Status::OK();
+}
+
+void BatchWithLengthReader::_init() {
+ if (_batch_size > 0) {
+ _read_buf_pos += _batch_size;
+ _read_buf_len -= _batch_size;
+ }
+ _batch_size = -1;
+}
+
+Status BatchWithLengthReader::_get_batch_size() {
+ if (_batch_size >= 0) {
+ return Status::OK();
+ }
+
+ _ensure_cap(BATCH_SIZE_LENGTH);
+
+ RETURN_IF_ERROR(_get_data_from_reader(BATCH_SIZE_LENGTH));
+ if (_read_buf_len < BATCH_SIZE_LENGTH) {
+ return Status::OK();
+ }
+
+ _batch_size = _convert_to_length(_read_buf + _read_buf_pos);
+
+ if (_batch_size < 0) {
+ return Status::InternalError("Invalid batch size");
+ }
+
+ _read_buf_pos += BATCH_SIZE_LENGTH;
+ _read_buf_len -= BATCH_SIZE_LENGTH;
+
+ if (_batch_size > _read_buf_cap) {
+ while (_read_buf_cap < _batch_size) {
+ _read_buf_cap *= 2;
+ }
+
+ // The current buff capacity is not enough and needs to be expanded
+ auto* new_read_buf = new uint8_t[_read_buf_cap];
+ memmove(new_read_buf, _read_buf + _read_buf_pos, _read_buf_len);
+ delete[] _read_buf;
+ _read_buf = new_read_buf;
+ _read_buf_pos = 0;
+ }
+ return Status::OK();
+}
+
+Status BatchWithLengthReader::_get_data_from_reader(int req_size) {
+ while (_read_buf_len < req_size) {
+ Slice file_slice(_read_buf + _read_buf_pos + _read_buf_len,
+ _read_buf_cap - _read_buf_pos - _read_buf_len);
+ size_t read_length = 0;
+ RETURN_IF_ERROR(_file_reader->read_at(0, file_slice, &read_length, NULL));
+ _read_buf_len += read_length;
+ if (read_length == 0) {
+ break;
+ }
+ }
+ return Status::OK();
+}
+
+void BatchWithLengthReader::_ensure_cap(int req_size) {
+ if (_read_buf_len < req_size && _read_buf_cap - _read_buf_pos < req_size) {
+ // there is not enough space for data,
+ // we need to move the data that has been used before and the useful data forward
+ memmove(_read_buf, _read_buf + _read_buf_pos, _read_buf_len);
+ _read_buf_pos = 0;
+ }
+}
+
+uint32_t BatchWithLengthReader::_convert_to_length(const uint8_t* data) {
+ uint32_t len = (((uint32_t)data[0]) << 24) & 0xff000000;
+ len |= (((uint32_t)data[1]) << 16) & 0xff0000;
Review Comment:
warning: 16 is a magic number; consider replacing it with a named constant [readability-magic-numbers]
```cpp
len |= (((uint32_t)data[1]) << 16) & 0xff0000;
^
```
##########
be/src/vec/exec/format/arrow/batch_with_length_reader.cpp:
##########
@@ -0,0 +1,147 @@
+// 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.
+
+#include "batch_with_length_reader.h"
+
+#include "arrow/array.h"
+#include "arrow/io/buffered.h"
+#include "arrow/io/stdio.h"
+#include "arrow/ipc/options.h"
+#include "arrow/ipc/reader.h"
+#include "arrow/record_batch.h"
+#include "arrow/result.h"
+#include "common/logging.h"
+#include "io/fs/stream_load_pipe.h"
+#include "olap/wal_manager.h"
+#include "runtime/runtime_state.h"
+
+#define DEFAULT_READ_BUF_CAP (4 * 1024 * 1024)
+#define BATCH_SIZE_LENGTH (4)
+
+namespace doris::vectorized {
+
+BatchWithLengthReader::BatchWithLengthReader(io::FileReaderSPtr file_reader)
+ : _file_reader(file_reader),
+ _read_buf(new uint8_t[DEFAULT_READ_BUF_CAP]),
+ _read_buf_cap(DEFAULT_READ_BUF_CAP),
+ _read_buf_pos(0),
+ _read_buf_len(0),
+ _batch_size(-1) {}
+
+BatchWithLengthReader::~BatchWithLengthReader() = default;
+
+Status BatchWithLengthReader::get_one_batch(uint8_t** data, int* length) {
+ _init();
+ RETURN_IF_ERROR(_get_batch_size());
+ RETURN_IF_ERROR(_get_batch_value());
+
+ *data = &_read_buf[_read_buf_pos];
+ if (_batch_size <= 0) {
+ *length = 0;
+ } else {
+ *length = _batch_size;
+ }
+ return Status::OK();
+}
+
+void BatchWithLengthReader::_init() {
+ if (_batch_size > 0) {
+ _read_buf_pos += _batch_size;
+ _read_buf_len -= _batch_size;
+ }
+ _batch_size = -1;
+}
+
+Status BatchWithLengthReader::_get_batch_size() {
+ if (_batch_size >= 0) {
+ return Status::OK();
+ }
+
+ _ensure_cap(BATCH_SIZE_LENGTH);
+
+ RETURN_IF_ERROR(_get_data_from_reader(BATCH_SIZE_LENGTH));
+ if (_read_buf_len < BATCH_SIZE_LENGTH) {
+ return Status::OK();
+ }
+
+ _batch_size = _convert_to_length(_read_buf + _read_buf_pos);
+
+ if (_batch_size < 0) {
+ return Status::InternalError("Invalid batch size");
+ }
+
+ _read_buf_pos += BATCH_SIZE_LENGTH;
+ _read_buf_len -= BATCH_SIZE_LENGTH;
+
+ if (_batch_size > _read_buf_cap) {
+ while (_read_buf_cap < _batch_size) {
+ _read_buf_cap *= 2;
+ }
+
+ // The current buff capacity is not enough and needs to be expanded
+ auto* new_read_buf = new uint8_t[_read_buf_cap];
+ memmove(new_read_buf, _read_buf + _read_buf_pos, _read_buf_len);
+ delete[] _read_buf;
+ _read_buf = new_read_buf;
+ _read_buf_pos = 0;
+ }
+ return Status::OK();
+}
+
+Status BatchWithLengthReader::_get_data_from_reader(int req_size) {
+ while (_read_buf_len < req_size) {
+ Slice file_slice(_read_buf + _read_buf_pos + _read_buf_len,
+ _read_buf_cap - _read_buf_pos - _read_buf_len);
+ size_t read_length = 0;
+ RETURN_IF_ERROR(_file_reader->read_at(0, file_slice, &read_length, NULL));
+ _read_buf_len += read_length;
+ if (read_length == 0) {
+ break;
+ }
+ }
+ return Status::OK();
+}
+
+void BatchWithLengthReader::_ensure_cap(int req_size) {
+ if (_read_buf_len < req_size && _read_buf_cap - _read_buf_pos < req_size) {
+ // there is not enough space for data,
+ // we need to move the data that has been used before and the useful data forward
+ memmove(_read_buf, _read_buf + _read_buf_pos, _read_buf_len);
+ _read_buf_pos = 0;
+ }
+}
+
+uint32_t BatchWithLengthReader::_convert_to_length(const uint8_t* data) {
+ uint32_t len = (((uint32_t)data[0]) << 24) & 0xff000000;
+ len |= (((uint32_t)data[1]) << 16) & 0xff0000;
+ len |= (((uint32_t)data[2]) << 8) & 0xff00;
+ len |= ((uint32_t)data[3]) & 0xff;
Review Comment:
warning: 0xff is a magic number; consider replacing it with a named constant [readability-magic-numbers]
```cpp
len |= ((uint32_t)data[3]) & 0xff;
^
```
##########
be/src/vec/exec/format/arrow/batch_with_length_reader.h:
##########
@@ -0,0 +1,77 @@
+// 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.
+
+#pragma once
+
+#include <gen_cpp/PlanNodes_types.h>
Review Comment:
warning: 'gen_cpp/PlanNodes_types.h' file not found [clang-diagnostic-error]
```cpp
#include <gen_cpp/PlanNodes_types.h>
^
```
##########
be/src/vec/exec/format/arrow/batch_with_length_reader.cpp:
##########
@@ -0,0 +1,147 @@
+// 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.
+
+#include "batch_with_length_reader.h"
+
+#include "arrow/array.h"
+#include "arrow/io/buffered.h"
+#include "arrow/io/stdio.h"
+#include "arrow/ipc/options.h"
+#include "arrow/ipc/reader.h"
+#include "arrow/record_batch.h"
+#include "arrow/result.h"
+#include "common/logging.h"
+#include "io/fs/stream_load_pipe.h"
+#include "olap/wal_manager.h"
+#include "runtime/runtime_state.h"
+
+#define DEFAULT_READ_BUF_CAP (4 * 1024 * 1024)
+#define BATCH_SIZE_LENGTH (4)
+
+namespace doris::vectorized {
+
+BatchWithLengthReader::BatchWithLengthReader(io::FileReaderSPtr file_reader)
+ : _file_reader(file_reader),
+ _read_buf(new uint8_t[DEFAULT_READ_BUF_CAP]),
+ _read_buf_cap(DEFAULT_READ_BUF_CAP),
+ _read_buf_pos(0),
+ _read_buf_len(0),
+ _batch_size(-1) {}
+
+BatchWithLengthReader::~BatchWithLengthReader() = default;
+
+Status BatchWithLengthReader::get_one_batch(uint8_t** data, int* length) {
+ _init();
+ RETURN_IF_ERROR(_get_batch_size());
+ RETURN_IF_ERROR(_get_batch_value());
+
+ *data = &_read_buf[_read_buf_pos];
+ if (_batch_size <= 0) {
+ *length = 0;
+ } else {
+ *length = _batch_size;
+ }
+ return Status::OK();
+}
+
+void BatchWithLengthReader::_init() {
+ if (_batch_size > 0) {
+ _read_buf_pos += _batch_size;
+ _read_buf_len -= _batch_size;
+ }
+ _batch_size = -1;
+}
+
+Status BatchWithLengthReader::_get_batch_size() {
+ if (_batch_size >= 0) {
+ return Status::OK();
+ }
+
+ _ensure_cap(BATCH_SIZE_LENGTH);
+
+ RETURN_IF_ERROR(_get_data_from_reader(BATCH_SIZE_LENGTH));
+ if (_read_buf_len < BATCH_SIZE_LENGTH) {
+ return Status::OK();
+ }
+
+ _batch_size = _convert_to_length(_read_buf + _read_buf_pos);
+
+ if (_batch_size < 0) {
+ return Status::InternalError("Invalid batch size");
+ }
+
+ _read_buf_pos += BATCH_SIZE_LENGTH;
+ _read_buf_len -= BATCH_SIZE_LENGTH;
+
+ if (_batch_size > _read_buf_cap) {
+ while (_read_buf_cap < _batch_size) {
+ _read_buf_cap *= 2;
+ }
+
+ // The current buff capacity is not enough and needs to be expanded
+ auto* new_read_buf = new uint8_t[_read_buf_cap];
+ memmove(new_read_buf, _read_buf + _read_buf_pos, _read_buf_len);
+ delete[] _read_buf;
+ _read_buf = new_read_buf;
+ _read_buf_pos = 0;
+ }
+ return Status::OK();
+}
+
+Status BatchWithLengthReader::_get_data_from_reader(int req_size) {
+ while (_read_buf_len < req_size) {
+ Slice file_slice(_read_buf + _read_buf_pos + _read_buf_len,
+ _read_buf_cap - _read_buf_pos - _read_buf_len);
+ size_t read_length = 0;
+ RETURN_IF_ERROR(_file_reader->read_at(0, file_slice, &read_length, NULL));
+ _read_buf_len += read_length;
+ if (read_length == 0) {
+ break;
+ }
+ }
+ return Status::OK();
+}
+
+void BatchWithLengthReader::_ensure_cap(int req_size) {
+ if (_read_buf_len < req_size && _read_buf_cap - _read_buf_pos < req_size) {
+ // there is not enough space for data,
+ // we need to move the data that has been used before and the useful data forward
+ memmove(_read_buf, _read_buf + _read_buf_pos, _read_buf_len);
+ _read_buf_pos = 0;
+ }
+}
+
+uint32_t BatchWithLengthReader::_convert_to_length(const uint8_t* data) {
+ uint32_t len = (((uint32_t)data[0]) << 24) & 0xff000000;
Review Comment:
warning: 0xff000000 is a magic number; consider replacing it with a named constant [readability-magic-numbers]
```cpp
uint32_t len = (((uint32_t)data[0]) << 24) & 0xff000000;
^
```
--
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] [feture](stream load) (step one)Add arrow data type for stream load [doris]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on code in PR #26709:
URL: https://github.com/apache/doris/pull/26709#discussion_r1404707056
##########
be/src/vec/data_types/serde/data_type_datetimev2_serde.cpp:
##########
@@ -112,6 +113,52 @@
}
}
+void DataTypeDateTimeV2SerDe::read_column_from_arrow(IColumn& column,
+ const arrow::Array* arrow_array, int start,
+ int end, const cctz::time_zone& ctz) const {
+ auto& col_data = static_cast<ColumnVector<Int64>&>(column).get_data();
+ int64_t divisor = 1;
+ if (arrow_array->type()->id() == arrow::Type::TIMESTAMP) {
+ auto concrete_array = dynamic_cast<const arrow::TimestampArray*>(arrow_array);
+ const auto type = std::static_pointer_cast<arrow::TimestampType>(arrow_array->type());
+ switch (type->unit()) {
+ case arrow::TimeUnit::type::SECOND: {
+ divisor = 1;
+ break;
+ }
+ case arrow::TimeUnit::type::MILLI: {
+ divisor = 1000;
Review Comment:
warning: 1000 is a magic number; consider replacing it with a named constant [readability-magic-numbers]
```cpp
divisor = 1000;
^
```
##########
be/src/vec/data_types/serde/data_type_datetimev2_serde.cpp:
##########
@@ -112,6 +113,52 @@ void DataTypeDateTimeV2SerDe::write_column_to_arrow(const IColumn& column, const
}
}
+void DataTypeDateTimeV2SerDe::read_column_from_arrow(IColumn& column,
Review Comment:
warning: method 'read_column_from_arrow' can be made static [readability-convert-member-functions-to-static]
```suggestion
static void DataTypeDateTimeV2SerDe::read_column_from_arrow(IColumn& column,
```
be/src/vec/data_types/serde/data_type_datetimev2_serde.cpp:117:
```diff
- int end, const cctz::time_zone& ctz) const {
+ int end, const cctz::time_zone& ctz) {
```
##########
be/src/vec/data_types/serde/data_type_datetimev2_serde.cpp:
##########
@@ -112,6 +113,52 @@
}
}
+void DataTypeDateTimeV2SerDe::read_column_from_arrow(IColumn& column,
+ const arrow::Array* arrow_array, int start,
+ int end, const cctz::time_zone& ctz) const {
+ auto& col_data = static_cast<ColumnVector<Int64>&>(column).get_data();
+ int64_t divisor = 1;
+ if (arrow_array->type()->id() == arrow::Type::TIMESTAMP) {
+ auto concrete_array = dynamic_cast<const arrow::TimestampArray*>(arrow_array);
+ const auto type = std::static_pointer_cast<arrow::TimestampType>(arrow_array->type());
+ switch (type->unit()) {
+ case arrow::TimeUnit::type::SECOND: {
+ divisor = 1;
+ break;
+ }
+ case arrow::TimeUnit::type::MILLI: {
+ divisor = 1000;
+ break;
+ }
+ case arrow::TimeUnit::type::MICRO: {
+ divisor = 1000000;
+ break;
+ }
+ case arrow::TimeUnit::type::NANO: {
+ divisor = 1000000000;
Review Comment:
warning: 1000000000 is a magic number; consider replacing it with a named constant [readability-magic-numbers]
```cpp
divisor = 1000000000;
^
```
##########
be/src/vec/data_types/serde/data_type_datetimev2_serde.cpp:
##########
@@ -112,6 +113,52 @@
}
}
+void DataTypeDateTimeV2SerDe::read_column_from_arrow(IColumn& column,
+ const arrow::Array* arrow_array, int start,
+ int end, const cctz::time_zone& ctz) const {
+ auto& col_data = static_cast<ColumnVector<Int64>&>(column).get_data();
+ int64_t divisor = 1;
+ if (arrow_array->type()->id() == arrow::Type::TIMESTAMP) {
+ auto concrete_array = dynamic_cast<const arrow::TimestampArray*>(arrow_array);
+ const auto type = std::static_pointer_cast<arrow::TimestampType>(arrow_array->type());
+ switch (type->unit()) {
+ case arrow::TimeUnit::type::SECOND: {
+ divisor = 1;
+ break;
+ }
+ case arrow::TimeUnit::type::MILLI: {
+ divisor = 1000;
+ break;
+ }
+ case arrow::TimeUnit::type::MICRO: {
+ divisor = 1000000;
Review Comment:
warning: 1000000 is a magic number; consider replacing it with a named constant [readability-magic-numbers]
```cpp
divisor = 1000000;
^
```
--
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] [feture](stream load) (step one)Add arrow data type for stream load [doris]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #26709:
URL: https://github.com/apache/doris/pull/26709#issuecomment-1831461377
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] [feture](stream load) (step one)Add arrow data type for stream load [doris]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on code in PR #26709:
URL: https://github.com/apache/doris/pull/26709#discussion_r1413010093
##########
be/src/vec/data_types/serde/data_type_datev2_serde.cpp:
##########
@@ -101,7 +101,7 @@ void DataTypeDateV2SerDe::read_column_from_arrow(IColumn& column, const arrow::A
int start, int end,
const cctz::time_zone& ctz) const {
auto& col_data = static_cast<ColumnVector<UInt32>&>(column).get_data();
- auto concrete_array = dynamic_cast<const arrow::Date64Array*>(arrow_array);
+ auto concrete_array = dynamic_cast<const arrow::Date32Array*>(arrow_array);
Review Comment:
warning: 'auto concrete_array' can be declared as 'const auto *concrete_array' [readability-qualified-auto]
```suggestion
const auto *concrete_array = dynamic_cast<const arrow::Date32Array*>(arrow_array);
```
##########
be/src/vec/data_types/serde/data_type_datetimev2_serde.cpp:
##########
@@ -110,6 +118,52 @@
}
}
+void DataTypeDateTimeV2SerDe::read_column_from_arrow(IColumn& column,
+ const arrow::Array* arrow_array, int start,
+ int end, const cctz::time_zone& ctz) const {
+ auto& col_data = static_cast<ColumnVector<Int64>&>(column).get_data();
+ int64_t divisor = 1;
+ if (arrow_array->type()->id() == arrow::Type::TIMESTAMP) {
+ auto concrete_array = dynamic_cast<const arrow::TimestampArray*>(arrow_array);
Review Comment:
warning: 'auto concrete_array' can be declared as 'const auto *concrete_array' [readability-qualified-auto]
```suggestion
const auto *concrete_array = dynamic_cast<const arrow::TimestampArray*>(arrow_array);
```
##########
be/src/vec/data_types/serde/data_type_datetimev2_serde.cpp:
##########
@@ -110,6 +118,52 @@ void DataTypeDateTimeV2SerDe::write_column_to_arrow(const IColumn& column, const
}
}
+void DataTypeDateTimeV2SerDe::read_column_from_arrow(IColumn& column,
Review Comment:
warning: method 'read_column_from_arrow' can be made static [readability-convert-member-functions-to-static]
be/src/vec/data_types/serde/data_type_datetimev2_serde.cpp:122:
```diff
- int end, const cctz::time_zone& ctz) const {
+ int end, const cctz::time_zone& ctz) {
```
be/src/vec/data_types/serde/data_type_datetimev2_serde.h:64:
```diff
- void read_column_from_arrow(IColumn& column, const arrow::Array* arrow_array, int start,
- int end, const cctz::time_zone& ctz) const override;
+ static void read_column_from_arrow(IColumn& column, const arrow::Array* arrow_array, int start,
+ int end, const cctz::time_zone& ctz) override;
```
--
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] [feture](stream load) (step one)Add arrow data type for stream load [doris]
Posted by "wuwenchi (via GitHub)" <gi...@apache.org>.
wuwenchi commented on code in PR #26709:
URL: https://github.com/apache/doris/pull/26709#discussion_r1398631548
##########
be/src/vec/exec/format/arrow/arrow_reader.h:
##########
@@ -0,0 +1,78 @@
+// 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.
+
+#pragma once
+
+#include <gen_cpp/PlanNodes_types.h>
+#include <gen_cpp/internal_service.pb.h>
+#include <stddef.h>
+#include <stdint.h>
+
+#include <cstddef>
+#include <memory>
+#include <string>
+#include <unordered_map>
+#include <unordered_set>
+#include <vector>
+
+#include "arrow_pip_input_stream.h"
+#include "common/status.h"
+#include "io/file_factory.h"
+#include "io/fs/file_reader_writer_fwd.h"
+#include "util/slice.h"
+#include "vec/data_types/data_type.h"
+#include "vec/exec/format/file_reader/new_plain_text_line_reader.h"
+#include "vec/exec/format/generic_reader.h"
+
+namespace doris {
+
+namespace io {
+class FileSystem;
+struct IOContext;
+} // namespace io
+
+namespace vectorized {
+
+struct ScannerCounter;
+class Block;
+
+class ArrowReader : public GenericReader {
Review Comment:
done
##########
be/src/vec/exec/format/arrow/arrow_pip_input_stream.cpp:
##########
@@ -0,0 +1,95 @@
+// 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.
+
+#include "arrow_pip_input_stream.h"
+
+#include "arrow/array.h"
+#include "arrow/buffer.h"
+#include "arrow/io/buffered.h"
+#include "arrow/io/stdio.h"
+#include "arrow/ipc/options.h"
+#include "arrow/ipc/reader.h"
+#include "arrow/record_batch.h"
+#include "arrow/result.h"
+#include "common/logging.h"
+#include "io/fs/stream_load_pipe.h"
+#include "olap/wal_manager.h"
+#include "runtime/runtime_state.h"
+
+namespace doris::vectorized {
+
+ArrowPipInputStream::ArrowPipInputStream(io::FileReaderSPtr file_reader)
+ : _file_reader(file_reader), _pos(0), _begin(true), _read_buf(new uint8_t[4]) {
+ set_mode(arrow::io::FileMode::READ);
+}
+
+arrow::Status ArrowPipInputStream::Close() {
+ return arrow::Status::OK();
+}
+
+bool ArrowPipInputStream::closed() const {
+ return false;
+}
+
+arrow::Result<int64_t> ArrowPipInputStream::Tell() const {
+ return _pos;
+}
+
+Status ArrowPipInputStream::HasNext(bool* get) {
+ Slice file_slice(_read_buf, 4);
+ size_t read_length = 0;
+ RETURN_IF_ERROR(_file_reader->read_at(0, file_slice, &read_length, NULL));
+ if (read_length == 0) {
+ *get = false;
+ } else {
+ *get = true;
+ }
+ return Status::OK();
+}
+
+arrow::Result<int64_t> ArrowPipInputStream::Read(int64_t nbytes, void* out) {
+ uint8_t* out_ptr = (uint8_t*)out;
+ if (_begin) {
+ memmove(out_ptr, _read_buf, 4);
+ out_ptr += 4;
+ nbytes -= 4;
+ }
+
+ Slice file_slice(out_ptr, nbytes);
+ size_t read_length = 0;
+ Status status = _file_reader->read_at(0, file_slice, &read_length, NULL);
+ if (UNLIKELY(!status.ok())) {
+ return arrow::Status::IOError("Error to read data from pip");
+ }
+ // pos_ += (int64_t)read_length;
Review Comment:
done
--
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] [feture](stream load) (step one)Add arrow data type for stream load [doris]
Posted by "doris-robot (via GitHub)" <gi...@apache.org>.
doris-robot commented on PR #26709:
URL: https://github.com/apache/doris/pull/26709#issuecomment-1818245747
TeamCity be ut coverage result:
Function Coverage: 36.48% (8448/23156)
Line Coverage: 28.85% (68667/238028)
Region Coverage: 27.83% (35512/127618)
Branch Coverage: 24.59% (18106/73644)
Coverage Report: http://coverage.selectdb-in.cc/coverage/9bd7b30f19a74525bd3a3e5642fb1eb3f9358019_9bd7b30f19a74525bd3a3e5642fb1eb3f9358019/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] [feture](stream load) (step one)Add arrow data type for stream load [doris]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on code in PR #26709:
URL: https://github.com/apache/doris/pull/26709#discussion_r1399313983
##########
be/src/vec/exec/format/arrow/arrow_pip_input_stream.h:
##########
@@ -0,0 +1,73 @@
+// 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.
+
+#pragma once
+
+#include <cstddef>
+#include <cstdint>
+#include <memory>
+#include <string>
+#include <unordered_map>
+#include <unordered_set>
+#include <vector>
+
+#include "arrow/io/interfaces.h"
Review Comment:
warning: 'arrow/io/interfaces.h' file not found [clang-diagnostic-error]
```cpp
#include "arrow/io/interfaces.h"
^
```
--
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] [feture](stream load) (step one)Add arrow data type for stream load [doris]
Posted by "doris-robot (via GitHub)" <gi...@apache.org>.
doris-robot commented on PR #26709:
URL: https://github.com/apache/doris/pull/26709#issuecomment-1817741554
TeamCity be ut coverage result:
Function Coverage: 36.71% (8431/22965)
Line Coverage: 29.22% (68585/234697)
Region Coverage: 27.83% (35459/127401)
Branch Coverage: 24.57% (18096/73640)
Coverage Report: http://coverage.selectdb-in.cc/coverage/3ad12f03a0f6cc9edc7387c4424635e3d050f245_3ad12f03a0f6cc9edc7387c4424635e3d050f245/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] [feture](stream load) (step one)Add arrow data type for stream load [doris]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on code in PR #26709:
URL: https://github.com/apache/doris/pull/26709#discussion_r1398312616
##########
be/src/vec/exec/format/arrow/arrow_pip_input_stream.h:
##########
@@ -0,0 +1,77 @@
+// 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.
+
+#pragma once
+
+#include <gen_cpp/PlanNodes_types.h>
Review Comment:
warning: 'gen_cpp/PlanNodes_types.h' file not found [clang-diagnostic-error]
```cpp
#include <gen_cpp/PlanNodes_types.h>
^
```
--
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] [feture](stream load) (step one)Add arrow data type for stream load [doris]
Posted by "wuwenchi (via GitHub)" <gi...@apache.org>.
wuwenchi commented on PR #26709:
URL: https://github.com/apache/doris/pull/26709#issuecomment-1837736845
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] [feture](stream load) (step one)Add arrow data type for stream load [doris]
Posted by "wuwenchi (via GitHub)" <gi...@apache.org>.
wuwenchi commented on code in PR #26709:
URL: https://github.com/apache/doris/pull/26709#discussion_r1398632291
##########
be/src/vec/exec/format/arrow/arrow_reader.cpp:
##########
@@ -0,0 +1,115 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include "arrow_reader.h"
+
+#include "arrow/array.h"
+#include "arrow/io/buffered.h"
+#include "arrow/io/stdio.h"
+#include "arrow/ipc/options.h"
+#include "arrow/ipc/reader.h"
+#include "arrow/record_batch.h"
+#include "arrow/result.h"
+#include "arrow_pip_input_stream.h"
+#include "common/logging.h"
+#include "io/fs/stream_load_pipe.h"
+#include "olap/wal_manager.h"
+#include "runtime/descriptors.h"
+#include "runtime/runtime_state.h"
+#include "vec/utils/arrow_column_to_doris_column.h"
+
+namespace doris {
+class RuntimeProfile;
+} // namespace doris
+
+namespace doris::vectorized {
+
+ArrowReader::ArrowReader(RuntimeState* state, RuntimeProfile* profile, ScannerCounter* counter,
+ const TFileScanRangeParams& params, const TFileRangeDesc& range,
+ const std::vector<SlotDescriptor*>& file_slot_descs, io::IOContext* io_ctx)
+ : _state(state), _range(range), _file_slot_descs(file_slot_descs), _file_reader(nullptr) {}
+
+ArrowReader::~ArrowReader() = default;
+
+Status ArrowReader::init_reader() {
+ RETURN_IF_ERROR(FileFactory::create_pipe_reader(_range.load_id, &_file_reader, _state, false));
+ _pip_stream = ArrowPipInputStream::create_unique(_file_reader);
+ return Status::OK();
+}
+
+Status ArrowReader::get_next_block(Block* block, size_t* read_rows, bool* eof) {
+ bool has_next = false;
+ RETURN_IF_ERROR(_pip_stream->HasNext(&has_next));
+ if (!has_next) {
+ *read_rows = 0;
+ *eof = true;
+ return Status::OK();
+ }
+
+ // create a reader to read data
+ arrow::Result<std::shared_ptr<arrow::ipc::RecordBatchStreamReader>> tRet =
Review Comment:
Because every batch data has schema and other metadata, this infomation will be parsed in `Open` func.
--
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] [feture](stream load) (step one)Add arrow data type for stream load [doris]
Posted by "morningman (via GitHub)" <gi...@apache.org>.
morningman commented on code in PR #26709:
URL: https://github.com/apache/doris/pull/26709#discussion_r1400123610
##########
be/src/vec/exec/format/arrow/arrow_pip_input_stream.cpp:
##########
@@ -0,0 +1,94 @@
+// 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.
+
+#include "arrow_pip_input_stream.h"
+
+#include "arrow/array.h"
+#include "arrow/buffer.h"
+#include "arrow/io/buffered.h"
+#include "arrow/io/stdio.h"
+#include "arrow/ipc/options.h"
+#include "arrow/ipc/reader.h"
+#include "arrow/record_batch.h"
+#include "arrow/result.h"
+#include "common/logging.h"
+#include "io/fs/stream_load_pipe.h"
+#include "olap/wal_manager.h"
+#include "runtime/runtime_state.h"
+
+namespace doris::vectorized {
+
+ArrowPipInputStream::ArrowPipInputStream(io::FileReaderSPtr file_reader)
+ : _file_reader(file_reader), _pos(0), _begin(true), _read_buf(new uint8_t[4]) {
+ set_mode(arrow::io::FileMode::READ);
+}
+
+arrow::Status ArrowPipInputStream::Close() {
+ return arrow::Status::OK();
+}
+
+bool ArrowPipInputStream::closed() const {
+ return false;
+}
+
+arrow::Result<int64_t> ArrowPipInputStream::Tell() const {
+ return _pos;
+}
+
+Status ArrowPipInputStream::HasNext(bool* get) {
+ Slice file_slice(_read_buf, 4);
Review Comment:
Add some comments to explain why we need to read 4 bytes
##########
be/src/vec/exec/format/arrow/arrow_stream_reader.cpp:
##########
@@ -0,0 +1,118 @@
+// 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.
+
+#include "arrow_stream_reader.h"
+
+#include "arrow/array.h"
+#include "arrow/io/buffered.h"
+#include "arrow/io/stdio.h"
+#include "arrow/ipc/options.h"
+#include "arrow/ipc/reader.h"
+#include "arrow/record_batch.h"
+#include "arrow/result.h"
+#include "arrow_pip_input_stream.h"
+#include "common/logging.h"
+#include "io/fs/stream_load_pipe.h"
+#include "olap/wal_manager.h"
+#include "runtime/descriptors.h"
+#include "runtime/runtime_state.h"
+#include "vec/utils/arrow_column_to_doris_column.h"
+
+namespace doris {
+class RuntimeProfile;
+} // namespace doris
+
+namespace doris::vectorized {
+
+ArrowStreamReader::ArrowStreamReader(RuntimeState* state, RuntimeProfile* profile,
+ ScannerCounter* counter, const TFileScanRangeParams& params,
+ const TFileRangeDesc& range,
+ const std::vector<SlotDescriptor*>& file_slot_descs,
+ io::IOContext* io_ctx)
+ : _state(state), _range(range), _file_slot_descs(file_slot_descs), _file_reader(nullptr) {}
+
+ArrowStreamReader::~ArrowStreamReader() = default;
+
+Status ArrowStreamReader::init_reader() {
+ RETURN_IF_ERROR(FileFactory::create_pipe_reader(_range.load_id, &_file_reader, _state, false));
+ _pip_stream = ArrowPipInputStream::create_unique(_file_reader);
+ return Status::OK();
+}
+
+Status ArrowStreamReader::get_next_block(Block* block, size_t* read_rows, bool* eof) {
+ bool has_next = false;
+ RETURN_IF_ERROR(_pip_stream->HasNext(&has_next));
+ if (!has_next) {
+ *read_rows = 0;
+ *eof = true;
+ return Status::OK();
+ }
+
+ // create a reader to read data
+ arrow::Result<std::shared_ptr<arrow::ipc::RecordBatchStreamReader>> tRet =
+ arrow::ipc::RecordBatchStreamReader::Open(_pip_stream.get(),
+ arrow::ipc::IpcReadOptions::Defaults());
+ if (!tRet.ok()) {
+ LOG(ERROR) << "failed to open stream reader: " << tRet.status().message();
Review Comment:
```suggestion
LOG(WARNING) << "failed to open stream reader: " << tRet.status().message();
```
##########
fe/fe-core/src/main/java/org/apache/doris/common/util/Util.java:
##########
@@ -565,6 +565,8 @@ public static TFileFormatType getFileFormatTypeFromName(String formatName) {
return TFileFormatType.FORMAT_CSV_PLAIN;
} else if (lowerFileFormat.equals("wal")) {
Review Comment:
define this `wal` to FileFormatConstants too
##########
be/src/vec/exec/format/arrow/arrow_stream_reader.cpp:
##########
@@ -0,0 +1,118 @@
+// 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.
+
+#include "arrow_stream_reader.h"
+
+#include "arrow/array.h"
+#include "arrow/io/buffered.h"
+#include "arrow/io/stdio.h"
+#include "arrow/ipc/options.h"
+#include "arrow/ipc/reader.h"
+#include "arrow/record_batch.h"
+#include "arrow/result.h"
+#include "arrow_pip_input_stream.h"
+#include "common/logging.h"
+#include "io/fs/stream_load_pipe.h"
+#include "olap/wal_manager.h"
+#include "runtime/descriptors.h"
+#include "runtime/runtime_state.h"
+#include "vec/utils/arrow_column_to_doris_column.h"
+
+namespace doris {
+class RuntimeProfile;
+} // namespace doris
+
+namespace doris::vectorized {
+
+ArrowStreamReader::ArrowStreamReader(RuntimeState* state, RuntimeProfile* profile,
+ ScannerCounter* counter, const TFileScanRangeParams& params,
+ const TFileRangeDesc& range,
+ const std::vector<SlotDescriptor*>& file_slot_descs,
+ io::IOContext* io_ctx)
+ : _state(state), _range(range), _file_slot_descs(file_slot_descs), _file_reader(nullptr) {}
+
+ArrowStreamReader::~ArrowStreamReader() = default;
+
+Status ArrowStreamReader::init_reader() {
+ RETURN_IF_ERROR(FileFactory::create_pipe_reader(_range.load_id, &_file_reader, _state, false));
+ _pip_stream = ArrowPipInputStream::create_unique(_file_reader);
+ return Status::OK();
+}
+
+Status ArrowStreamReader::get_next_block(Block* block, size_t* read_rows, bool* eof) {
+ bool has_next = false;
+ RETURN_IF_ERROR(_pip_stream->HasNext(&has_next));
+ if (!has_next) {
+ *read_rows = 0;
+ *eof = true;
+ return Status::OK();
+ }
+
+ // create a reader to read data
+ arrow::Result<std::shared_ptr<arrow::ipc::RecordBatchStreamReader>> tRet =
+ arrow::ipc::RecordBatchStreamReader::Open(_pip_stream.get(),
+ arrow::ipc::IpcReadOptions::Defaults());
+ if (!tRet.ok()) {
+ LOG(ERROR) << "failed to open stream reader: " << tRet.status().message();
+ return Status::InternalError("failed to open stream reader: {}", tRet.status().message());
+ }
+ auto reader = std::move(tRet).ValueUnsafe();
+
+ // get arrow data from reader
+ arrow::Result<arrow::RecordBatchVector> tRet2 = reader->ToRecordBatches();
+ if (!tRet2.ok()) {
+ LOG(ERROR) << "failed to read batch: " << tRet2.status().message();
Review Comment:
```suggestion
LOG(WARNING) << "failed to read batch: " << tRet2.status().message();
```
--
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] [feture](stream load) (step one)Add arrow data type for stream load [doris]
Posted by "wuwenchi (via GitHub)" <gi...@apache.org>.
wuwenchi commented on PR #26709:
URL: https://github.com/apache/doris/pull/26709#issuecomment-1818186721
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] [feture](stream load) (step one)Add arrow data type for stream load [doris]
Posted by "wuwenchi (via GitHub)" <gi...@apache.org>.
wuwenchi commented on PR #26709:
URL: https://github.com/apache/doris/pull/26709#issuecomment-1826194325
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] [feture](stream load) (step one)Add arrow data type for stream load [doris]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on code in PR #26709:
URL: https://github.com/apache/doris/pull/26709#discussion_r1405301762
##########
be/src/vec/data_types/serde/data_type_datetimev2_serde.cpp:
##########
@@ -112,6 +119,52 @@ void DataTypeDateTimeV2SerDe::write_column_to_arrow(const IColumn& column, const
}
}
+void DataTypeDateTimeV2SerDe::read_column_from_arrow(IColumn& column,
Review Comment:
warning: method 'read_column_from_arrow' can be made static [readability-convert-member-functions-to-static]
```suggestion
static void DataTypeDateTimeV2SerDe::read_column_from_arrow(IColumn& column,
```
be/src/vec/data_types/serde/data_type_datetimev2_serde.cpp:123:
```diff
- int end, const cctz::time_zone& ctz) const {
+ int end, const cctz::time_zone& ctz) {
```
--
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