github-actions[bot] commented on code in PR #66008:
URL: https://github.com/apache/doris/pull/66008#discussion_r3649613871
##########
be/src/format_v2/jni/jni_table_reader.cpp:
##########
@@ -464,10 +464,16 @@ Status JniTableReader::_open_jni_scanner() {
}
void JniTableReader::set_batch_size(size_t batch_size) {
- if (_scanner_opened && !supports_batch_size_update_after_open()) {
- // Some connectors bake the constructor batch size into an
already-open physical reader.
- // Keep C++ and Java on that initial size instead of pretending a
later resize took effect.
- return;
+ if (!supports_batch_size_update_after_open()) {
Review Comment:
[P1] Do not leave JDBC's physical fetch size at the probe value
V2 calls this method with the 32-row adaptive probe before `prepare_split()`
eagerly opens JDBC. `JdbcJniScanner.open()` passes that value to
`PreparedStatement.setFetchSize()` and executes the query, but the later JNI
`setBatchSize()` call only updates the Java output-block field; it never
changes the statement/result-set fetch size. Drivers using the default/Oracle
handler can therefore keep fetching remote rows in 32-row windows for the
entire split, whereas V1 opens JDBC with `RuntimeState::batch_size()`. Model
physical fetch mutability separately (or open JDBC at the stable batch size),
and add a scanner-level JDBC test that observes the statement/result-set fetch
size.
##########
be/src/format_v2/wal/wal_reader.cpp:
##########
@@ -0,0 +1,301 @@
+// 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 "format_v2/wal/wal_reader.h"
+
+#include <absl/strings/numbers.h>
+#include <absl/strings/str_split.h>
+
+#include <ranges>
+#include <unordered_set>
+#include <utility>
+
+#include "agent/be_exec_version_manager.h"
+#include "common/cast_set.h"
+#include "core/block/block.h"
+#include "core/data_type/data_type_factory.hpp"
+#include "core/data_type/data_type_nullable.h"
+#include "format_v2/column_mapper.h"
+#include "format_v2/materialized_reader_util.h"
+#include "load/group_commit/wal/wal_file_reader.h"
+#include "load/group_commit/wal/wal_manager.h"
+#include "runtime/exec_env.h"
+#include "runtime/runtime_state.h"
+
+namespace doris::format::wal {
+namespace {
+
+class WalColumnMapper final : public TableColumnMapper {
+public:
+ using TableColumnMapper::TableColumnMapper;
+
+ Status create_mapping(const std::vector<ColumnDefinition>&
projected_columns,
+ const std::map<std::string, Field>& partition_values,
+ const std::vector<ColumnDefinition>& file_schema)
override {
+ for (const auto& projected : projected_columns) {
+ if (!projected.has_identifier_field_id()) {
+ return Status::InternalError("WAL projected column {} has no
unique id",
+ projected.name);
+ }
+ const auto found = std::ranges::find_if(file_schema, [&](const
auto& file_column) {
+ return file_column.has_identifier_field_id() &&
+ file_column.get_identifier_field_id() ==
projected.get_identifier_field_id();
+ });
+ if (found == file_schema.end()) {
+ return Status::InternalError("WAL does not contain column
unique id {} ({})",
+
projected.get_identifier_field_id(), projected.name);
+ }
+ }
+ return TableColumnMapper::create_mapping(projected_columns,
partition_values, file_schema);
+ }
+
+protected:
+ bool enable_lazy_materialization() const override { return false; }
+ bool force_full_complex_scan_projection() const override { return true; }
+};
+
+ColumnDefinition build_wal_column_definition(const PColumnMeta& meta, int32_t
local_id) {
+ ColumnDefinition field;
+ field.local_id = local_id;
+ field.name = meta.name();
+ field.type =
make_nullable(DataTypeFactory::instance().create_data_type(meta));
+ field.children.reserve(meta.children_size());
+ for (int child_idx = 0; child_idx < meta.children_size(); ++child_idx) {
+
field.children.push_back(build_wal_column_definition(meta.children(child_idx),
child_idx));
+ }
+ return field;
+}
+
+} // namespace
+
+Status parse_wal_column_ids(const std::string& encoded, std::vector<int32_t>*
column_ids) {
+ DORIS_CHECK(column_ids != nullptr);
+ column_ids->clear();
+ if (encoded.empty()) {
+ return Status::Corruption("WAL header contains no column ids");
+ }
+
+ std::unordered_set<int32_t> seen;
+ for (const absl::string_view token : absl::StrSplit(encoded, ',')) {
+ int32_t column_id = 0;
+ if (token.empty() || !absl::SimpleAtoi(token, &column_id)) {
+ return Status::Corruption("invalid WAL column id '{}'",
std::string(token));
+ }
+ if (!seen.emplace(column_id).second) {
+ return Status::Corruption("duplicate WAL column id {}", column_id);
+ }
+ column_ids->push_back(column_id);
+ }
+ return Status::OK();
+}
+
+WalReader::WalReader(std::shared_ptr<io::FileSystemProperties>&
system_properties,
+ std::unique_ptr<io::FileDescription>& file_description,
+ std::shared_ptr<io::IOContext> io_ctx, RuntimeProfile*
profile,
+ const std::vector<ColumnDefinition>& projected_columns)
+ : FileReader(system_properties, file_description, std::move(io_ctx),
profile),
+ _projected_columns(projected_columns) {}
+
+WalReader::~WalReader() {
+ static_cast<void>(close());
+}
+
+Status WalReader::init(RuntimeState* state) {
+ if (state == nullptr || state->exec_env() == nullptr ||
+ state->exec_env()->wal_mgr() == nullptr) {
+ return Status::InvalidArgument("WAL v2 reader requires a runtime WAL
manager");
+ }
+
RETURN_IF_ERROR(state->exec_env()->wal_mgr()->get_wal_path(state->wal_id(),
_wal_path));
+ _wal_reader = std::make_shared<doris::WalFileReader>(_wal_path);
+ RETURN_IF_ERROR(_wal_reader->init());
+
+ std::string encoded_column_ids;
+ RETURN_IF_ERROR(_wal_reader->read_header(_version, encoded_column_ids));
+ RETURN_IF_ERROR(parse_wal_column_ids(encoded_column_ids, &_column_ids));
+ _reader_eof = false;
+ _eof = false;
+ return Status::OK();
+}
+
+Status WalReader::get_schema(std::vector<ColumnDefinition>* file_schema) const
{
+ if (file_schema == nullptr) {
+ return Status::InvalidArgument("WAL v2 file_schema is null");
+ }
+ RETURN_IF_ERROR(_ensure_schema_loaded());
+ *file_schema = _file_schema;
+ return Status::OK();
+}
+
+std::unique_ptr<TableColumnMapper> WalReader::create_column_mapper(
+ TableColumnMapperOptions options) const {
+ return std::make_unique<WalColumnMapper>(std::move(options));
+}
+
+Status WalReader::open(std::shared_ptr<FileScanRequest> request) {
+ RETURN_IF_ERROR(FileReader::open(std::move(request)));
+ _first_block_consumed = false;
+ _eof = false;
+ return Status::OK();
+}
+
+Status WalReader::get_block(Block* file_block, size_t* rows, bool* eof) {
+ DORIS_CHECK(file_block != nullptr);
+ DORIS_CHECK(rows != nullptr);
+ DORIS_CHECK(eof != nullptr);
+ if (_request == nullptr) {
+ return Status::InternalError("WAL v2 reader is not open");
+ }
+
+ *rows = 0;
+ *eof = false;
+ if (_reader_eof) {
+ *eof = true;
+ _eof = true;
+ return Status::OK();
+ }
+
+ PBlock pblock;
+ if (_first_block_loaded && !_first_block_consumed) {
+ // Schema discovery owns the first payload temporarily; transfer it
into the read path so
+ // the reader neither retains a second PBlock nor forces protobuf to
clone its buffers.
+ pblock.Swap(&_first_block);
+ _first_block_consumed = true;
+ _first_block_loaded = false;
+ } else {
+ auto status = _wal_reader->read_block(pblock);
+ if (status.is<ErrorCode::END_OF_FILE>()) {
+ _reader_eof = true;
+ *eof = true;
+ _eof = true;
+ return Status::OK();
+ }
+ RETURN_IF_ERROR(status);
+ }
+ RETURN_IF_ERROR(_validate_block_version(pblock));
+
+ Block source_block;
+ size_t uncompressed_size = 0;
+ int64_t decompress_time = 0;
+ RETURN_IF_ERROR(source_block.deserialize(pblock, &uncompressed_size,
&decompress_time));
+ if (source_block.columns() != _column_ids.size()) {
+ return Status::Corruption("WAL block has {} columns but header
declares {}",
+ source_block.columns(), _column_ids.size());
+ }
+ RETURN_IF_ERROR(_materialize_requested_columns(&source_block, file_block));
+ *rows = file_block->rows();
+ _record_scan_rows(cast_set<int64_t>(*rows));
+ RETURN_IF_ERROR(
+ apply_materialized_reader_filters(_request.get(), _io_ctx.get(),
file_block, rows));
Review Comment:
[P1] Keep unsafe WAL predicates on Scanner's ordered path
`TableColumnMapper::localize_filters()` accepts every deterministic local
predicate, even when `is_safe_to_execute_on_selected_rows()` is false or it
follows an earlier unsafe barrier. This call then runs those clones over the
full WAL block before Scanner evaluates the original conjunct order. For
example, with an always-false slotless `rand()` predicate followed by an
error-preserving predicate on a column, V1 filters every row before the latter
runs, while V2 can execute the latter here first and raise. Restrict file-local
WAL filtering to the original safe prefix (or leave WAL row filtering entirely
to Scanner), and cover the unsafe-barrier ordering case through FileScannerV2.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
