This is an automated email from the ASF dual-hosted git repository.
Gabriel39 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new 962bd5b28c7 [Feature](file scanner) Support JNI and WAL in V2 (#66008)
962bd5b28c7 is described below
commit 962bd5b28c7d36fb6aab0328b7a56e533f0d2bb1
Author: Gabriel <[email protected]>
AuthorDate: Sun Jul 26 10:59:38 2026 +0800
[Feature](file scanner) Support JNI and WAL in V2 (#66008)
### What problem does this PR solve?
Problem Summary:
FileScannerV2 did not accept `FORMAT_JNI` or `FORMAT_WAL`, so these
scans remained on the legacy scanner path.
This change:
- routes supported `FORMAT_JNI` scans through FileScannerV2, including
compatible Paimon JNI/native split shapes;
- adds an independent V2 WAL reader with unique-column-id mapping,
block-version validation, checksum-backed WAL file reads, projection,
and filtering;
- does not include or call the legacy WAL reader implementation.
For Paimon compatibility, old FE plans may describe native Parquet/ORC
files as `FORMAT_JNI` without a `paimon_split`. A real JNI split also
carries physical `file_format` metadata, so native-format conversion
must first exclude ranges classified as JNI splits. This preserves old
native plans without misclassifying real JNI splits.
`FORMAT_AVRO` remains unsupported because the corresponding V1 reader is
no longer available on master.
### Release note
FileScannerV2 now supports JNI scans and WAL files.
### Check List (For Author)
- Test
- [x] Unit Test
Verified with:
- `./run-be-ut.sh --run
--filter='PaimonHybridReaderTest.*:FileScannerV2Test.*:WalReaderV2Test.*'
-j 8` (34 tests passed)
- Behavior changed:
- [x] Yes. Eligible JNI and WAL scans use FileScannerV2 when
`enable_file_scanner_v2` is enabled.
- Does this need documentation?
- [x] No.
### Check List (For Reviewer who merge this PR)
- [ ] Confirm the release note
- [ ] Confirm test cases
- [ ] Confirm document
- [ ] Add branch pick label
---
be/src/exec/operator/file_scan_operator.cpp | 9 +-
be/src/exec/scan/file_scanner_v2.cpp | 40 ++-
be/src/format_v2/file_reader.h | 1 +
be/src/format_v2/jni/jni_table_reader.cpp | 14 +-
be/src/format_v2/jni/paimon_jni_reader.cpp | 2 +-
be/src/format_v2/table/paimon_reader.cpp | 35 ++-
be/src/format_v2/table_reader.cpp | 2 +
be/src/format_v2/table_reader.h | 1 +
be/src/format_v2/wal/wal_reader.cpp | 301 +++++++++++++++++++++
be/src/format_v2/wal/wal_reader.h | 72 +++++
be/src/format_v2/wal/wal_table_reader.cpp | 47 ++++
be/src/format_v2/wal/wal_table_reader.h | 37 +++
be/test/exec/scan/file_scanner_v2_test.cpp | 35 +--
be/test/format_v2/jni/paimon_jni_reader_test.cpp | 20 +-
be/test/format_v2/table/paimon_reader_test.cpp | 59 +++-
be/test/format_v2/wal/wal_reader_test.cpp | 171 ++++++++++++
.../datasource/paimon/source/PaimonScanNode.java | 18 +-
.../paimon/source/PaimonScanNodeTest.java | 21 ++
18 files changed, 824 insertions(+), 61 deletions(-)
diff --git a/be/src/exec/operator/file_scan_operator.cpp
b/be/src/exec/operator/file_scan_operator.cpp
index 8121590ebf0..736c9647a2f 100644
--- a/be/src/exec/operator/file_scan_operator.cpp
+++ b/be/src/exec/operator/file_scan_operator.cpp
@@ -116,14 +116,9 @@ bool FileScanLocalState::_should_use_file_scanner_v2(const
TQueryOptions& query_
const bool is_transactional_hive =
scan_params.__isset.table_format_params &&
scan_params.table_format_params.table_format_type ==
"transactional_hive";
- // JNI reader selection is stored per split, but this scan-level selector
cannot inspect the
- // split yet. Older FEs may omit both the scan-level Paimon marker and
split-level reader_type,
- // so keep JNI scans on V1 until scanner selection can distinguish every
compatibility shape.
return query_options.__isset.enable_file_scanner_v2 &&
query_options.enable_file_scanner_v2 &&
- !is_load && scan_params.format_type != TFileFormatType::FORMAT_WAL
&&
- scan_params.format_type != TFileFormatType::FORMAT_ES_HTTP &&
- scan_params.format_type != TFileFormatType::FORMAT_LANCE &&
- scan_params.format_type != TFileFormatType::FORMAT_JNI &&
!is_transactional_hive;
+ !is_load && scan_params.format_type !=
TFileFormatType::FORMAT_ES_HTTP &&
+ scan_params.format_type != TFileFormatType::FORMAT_LANCE &&
!is_transactional_hive;
}
Status FileScanLocalState::_init_scanners(std::list<ScannerSPtr>* scanners) {
diff --git a/be/src/exec/scan/file_scanner_v2.cpp
b/be/src/exec/scan/file_scanner_v2.cpp
index cdaa3d2f3b8..6859fe8d543 100644
--- a/be/src/exec/scan/file_scanner_v2.cpp
+++ b/be/src/exec/scan/file_scanner_v2.cpp
@@ -61,6 +61,7 @@
#include "format_v2/table/paimon_reader.h"
#include "format_v2/table/remote_doris_reader.h"
#include "format_v2/table_reader.h"
+#include "format_v2/wal/wal_table_reader.h"
#include "io/cache/block_file_cache_profile.h"
#include "io/fs/file_meta_cache.h"
#include "io/io_common.h"
@@ -108,10 +109,26 @@ bool is_supported_arrow_table_format(const
TFileRangeDesc& range) {
bool is_supported_jni_table_format(const TFileRangeDesc& range) {
const auto table_format = table_format_name(range);
if (table_format == "paimon") {
- return range.__isset.table_format_params &&
- range.table_format_params.__isset.paimon_params &&
- range.table_format_params.paimon_params.__isset.reader_type &&
- range.table_format_params.paimon_params.reader_type ==
TPaimonReaderType::PAIMON_JNI;
+ if (!range.__isset.table_format_params ||
+ !range.table_format_params.__isset.paimon_params) {
+ return false;
+ }
+ const auto& params = range.table_format_params.paimon_params;
+ if (params.__isset.reader_type) {
+ if (params.reader_type == TPaimonReaderType::PAIMON_JNI) {
+ return params.__isset.paimon_split;
+ }
+ // V2 cannot pass a logical DataSplit through a raw native child
without silently
+ // dropping its multi-file semantics, so PAIMON_CPP must remain on
the V1 fallback.
+ return false;
+ }
+ if (params.__isset.paimon_split) {
+ // Before reader_type was added, an encoded split unambiguously
selected the Java
+ // reader; native scans carried only their physical Parquet or ORC
range.
+ return true;
+ }
+ return params.__isset.file_format &&
+ (params.file_format == "parquet" || params.file_format ==
"orc");
}
return table_format == "jdbc" || table_format == "iceberg" || table_format
== "hudi" ||
table_format == "max_compute" || table_format == "trino_connector";
@@ -155,6 +172,10 @@ bool is_native_format(TFileFormatType::type format_type) {
return format_type == TFileFormatType::FORMAT_NATIVE;
}
+bool is_wal_format(TFileFormatType::type format_type) {
+ return format_type == TFileFormatType::FORMAT_WAL;
+}
+
bool is_partition_slot(const TFileScanSlotInfo& slot_info, const std::string&
column_name) {
if (column_name.starts_with(BeConsts::GLOBAL_ROWID_COL) ||
column_name == BeConsts::ICEBERG_ROWID_COL) {
@@ -289,6 +310,8 @@ bool FileScannerV2::is_supported(const
TFileScanRangeParams& params, const TFile
return is_supported_arrow_table_format(range);
} else if (format_type == TFileFormatType::FORMAT_JNI) {
return is_supported_jni_table_format(range);
+ } else if (is_wal_format(format_type)) {
+ return table_format_name(range) == "NotSet";
} else if (is_csv_format(format_type) || is_text_format(format_type) ||
is_json_format(format_type) || is_native_format(format_type)) {
return is_supported_table_format(range);
@@ -563,6 +586,11 @@ Status FileScannerV2::_init_table_reader(const
TFileRangeDesc& range) {
Status FileScannerV2::_create_table_reader_for_format(
const TFileRangeDesc& range, std::unique_ptr<format::TableReader>*
reader) const {
DORIS_CHECK(reader != nullptr);
+ const auto file_format = get_range_format_type(*_params, range);
+ if (file_format == TFileFormatType::FORMAT_WAL) {
+ *reader = std::make_unique<format::wal::WalTableReader>();
+ return Status::OK();
+ }
const auto table_format = table_format_name(range);
if (table_format == "NotSet" || table_format == "tvf") {
*reader = std::make_unique<format::TableReader>();
@@ -738,6 +766,7 @@ Status FileScannerV2::_build_projected_columns(const
format::TableReader& table_
slot_info.slot_id);
}
auto column = _build_table_column(it->second);
+ build_context.slot_desc = it->second;
if (column.name.starts_with(BeConsts::GLOBAL_ROWID_COL)) {
_need_global_rowid_column = true;
}
@@ -851,6 +880,9 @@ Status FileScannerV2::_to_file_format(TFileFormatType::type
format_type,
case TFileFormatType::FORMAT_ARROW:
*file_format = format::FileFormat::ARROW;
return Status::OK();
+ case TFileFormatType::FORMAT_WAL:
+ *file_format = format::FileFormat::WAL;
+ return Status::OK();
default:
return Status::NotSupported("FileScannerV2 does not support file
format {}",
to_string(format_type));
diff --git a/be/src/format_v2/file_reader.h b/be/src/format_v2/file_reader.h
index 5f959c3e672..3ff512975d2 100644
--- a/be/src/format_v2/file_reader.h
+++ b/be/src/format_v2/file_reader.h
@@ -57,6 +57,7 @@ enum class FileFormat {
JNI,
NATIVE,
ARROW,
+ WAL,
};
struct FileScanRequest {
diff --git a/be/src/format_v2/jni/jni_table_reader.cpp
b/be/src/format_v2/jni/jni_table_reader.cpp
index 7658a52fc86..1c691fc3129 100644
--- a/be/src/format_v2/jni/jni_table_reader.cpp
+++ b/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()) {
+ if (_scanner_opened) {
+ return;
+ }
+ // Constructor-frozen readers must open with the stable query batch
size; a transient
+ // adaptive probe would otherwise remain their physical batch size for
the whole split.
+ if (_runtime_state != nullptr) {
+ TableReader::set_batch_size(_runtime_state->batch_size());
+ return;
+ }
}
TableReader::set_batch_size(batch_size);
if (!_scanner_opened) {
diff --git a/be/src/format_v2/jni/paimon_jni_reader.cpp
b/be/src/format_v2/jni/paimon_jni_reader.cpp
index 47c0ef6c7bc..86d16ce6f7d 100644
--- a/be/src/format_v2/jni/paimon_jni_reader.cpp
+++ b/be/src/format_v2/jni/paimon_jni_reader.cpp
@@ -59,7 +59,7 @@ Status PaimonJniReader::validate_scan_range(const
TFileRangeDesc& range) const {
"missing paimon_split for paimon jni reader, possibly caused
by FE/BE protocol "
"mismatch");
}
- if (!range.table_format_params.paimon_params.__isset.reader_type ||
+ if (range.table_format_params.paimon_params.__isset.reader_type &&
range.table_format_params.paimon_params.reader_type !=
TPaimonReaderType::PAIMON_JNI) {
return Status::InternalError(
"invalid reader_type for paimon jni reader, possibly caused by
FE/BE protocol "
diff --git a/be/src/format_v2/table/paimon_reader.cpp
b/be/src/format_v2/table/paimon_reader.cpp
index 5d8363848f3..a3f4092a470 100644
--- a/be/src/format_v2/table/paimon_reader.cpp
+++ b/be/src/format_v2/table/paimon_reader.cpp
@@ -104,6 +104,14 @@ Status PaimonHybridReader::prepare_split(const
format::SplitReadOptions& options
// timer around the first native or JNI child and double-count that
initialization.
RETURN_IF_ERROR(_ensure_current_split_reader(options));
DORIS_CHECK(_current_split_reader != nullptr);
+ if (!_is_jni_split(options.current_range)) {
+ auto native_options = options;
+ // Legacy FE plans wrap native files in FORMAT_JNI; normalize the
child contract so the
+ // physical reader does not overwrite its recovered Parquet/ORC format
with that wrapper.
+ RETURN_IF_ERROR(
+ _to_file_format(options.current_range,
&native_options.current_split_format));
+ return _current_split_reader->prepare_split(native_options);
+ }
return _current_split_reader->prepare_split(options);
}
@@ -178,7 +186,10 @@ Status
PaimonHybridReader::_ensure_current_split_reader(const format::SplitReadO
} else {
format::FileFormat file_format;
RETURN_IF_ERROR(_to_file_format(options.current_range, &file_format));
- DCHECK(options.current_split_format == file_format);
+ // Old FE plans encoded a native file as FORMAT_JNI without
paimon_split and carried the
+ // physical format only in paimon_params.file_format.
+ DCHECK(options.current_split_format == file_format ||
+ options.current_split_format == format::FileFormat::JNI);
DCHECK(file_format == format::FileFormat::PARQUET ||
file_format == format::FileFormat::ORC);
if (_native_reader == nullptr) {
@@ -236,16 +247,30 @@ Status
PaimonHybridReader::_clone_conjuncts(VExprContextSPtrs* conjuncts) const
}
bool PaimonHybridReader::_is_jni_split(const TFileRangeDesc& range) {
- return range.__isset.table_format_params &&
range.table_format_params.__isset.paimon_params &&
- range.table_format_params.paimon_params.__isset.reader_type &&
- range.table_format_params.paimon_params.reader_type ==
TPaimonReaderType::PAIMON_JNI;
+ if (!range.__isset.table_format_params ||
!range.table_format_params.__isset.paimon_params) {
+ return false;
+ }
+ const auto& params = range.table_format_params.paimon_params;
+ return params.__isset.paimon_split &&
+ (!params.__isset.reader_type || params.reader_type ==
TPaimonReaderType::PAIMON_JNI);
}
Status PaimonHybridReader::_to_file_format(const TFileRangeDesc& range,
format::FileFormat* file_format) {
DORIS_CHECK(file_format != nullptr);
- const auto format_type =
+ auto format_type =
range.__isset.format_type ? range.format_type :
TFileFormatType::FORMAT_PARQUET;
+ // JNI splits also carry file_format metadata; only a split without
paimon_split can use
+ // FORMAT_JNI as the legacy encoding of a native file.
+ if (format_type == TFileFormatType::FORMAT_JNI && !_is_jni_split(range) &&
+ range.__isset.table_format_params &&
range.table_format_params.__isset.paimon_params) {
+ const auto& params = range.table_format_params.paimon_params;
+ if (params.__isset.file_format && params.file_format == "orc") {
+ format_type = TFileFormatType::FORMAT_ORC;
+ } else if (params.__isset.file_format && params.file_format ==
"parquet") {
+ format_type = TFileFormatType::FORMAT_PARQUET;
+ }
+ }
switch (format_type) {
case TFileFormatType::FORMAT_PARQUET:
*file_format = format::FileFormat::PARQUET;
diff --git a/be/src/format_v2/table_reader.cpp
b/be/src/format_v2/table_reader.cpp
index 4beaf8c9ff5..164c0de6026 100644
--- a/be/src/format_v2/table_reader.cpp
+++ b/be/src/format_v2/table_reader.cpp
@@ -90,6 +90,8 @@ std::string file_format_to_string(FileFormat format) {
return "NATIVE";
case FileFormat::ARROW:
return "ARROW";
+ case FileFormat::WAL:
+ return "WAL";
}
return "UNKNOWN";
}
diff --git a/be/src/format_v2/table_reader.h b/be/src/format_v2/table_reader.h
index ea40280ee99..baf2feb3c4f 100644
--- a/be/src/format_v2/table_reader.h
+++ b/be/src/format_v2/table_reader.h
@@ -94,6 +94,7 @@ struct ProjectedColumnBuildContext {
const TFileScanRangeParams* scan_params = nullptr;
const TFileRangeDesc* range = nullptr;
RuntimeState* runtime_state = nullptr;
+ const SlotDescriptor* slot_desc = nullptr;
std::optional<ColumnDefinition> schema_column = std::nullopt;
size_t next_file_column_idx = 0;
};
diff --git a/be/src/format_v2/wal/wal_reader.cpp
b/be/src/format_v2/wal/wal_reader.cpp
new file mode 100644
index 00000000000..55e88796c98
--- /dev/null
+++ b/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));
+ return Status::OK();
+}
+
+Status WalReader::close() {
+ _request.reset();
+ _reader_eof = true;
+ _eof = true;
+ if (_wal_reader == nullptr) {
+ return Status::OK();
+ }
+ auto status = _wal_reader->finalize();
+ if (status.ok()) {
+ _wal_reader.reset();
+ }
+ return status;
+}
+
+Status WalReader::_ensure_schema_loaded() const {
+ if (_schema_inited) {
+ return Status::OK();
+ }
+
+ auto status = _wal_reader->read_block(_first_block);
+ if (status.is<ErrorCode::END_OF_FILE>()) {
+ // An empty WAL still has a complete unique-id header. Use only
matching projected types;
+ // there is no data block from which unprojected physical types could
be inferred.
+ return _init_schema_from_block(nullptr);
+ }
+ RETURN_IF_ERROR(status);
+ RETURN_IF_ERROR(_validate_block_version(_first_block));
+ _first_block_loaded = true;
+ return _init_schema_from_block(&_first_block);
+}
+
+Status WalReader::_validate_block_version(const PBlock& pblock) const {
+ const int version = pblock.has_be_exec_version() ?
pblock.be_exec_version() : 0;
+ if (!BeExecVersionManager::check_be_exec_version(version)) {
+ return Status::DataQualityError("unsupported BE execution version {}
in WAL", version);
+ }
+ return Status::OK();
+}
+
+Status WalReader::_init_schema_from_block(const PBlock* pblock) const {
+ if (pblock != nullptr && cast_set<size_t>(pblock->column_metas_size()) !=
_column_ids.size()) {
+ return Status::Corruption("WAL block schema has {} columns but header
declares {}",
+ pblock->column_metas_size(),
_column_ids.size());
+ }
+
+ _file_schema.clear();
+ for (size_t idx = 0; idx < _column_ids.size(); ++idx) {
+ ColumnDefinition field;
+ field.identifier = Field::create_field<TYPE_INT>(_column_ids[idx]);
+ field.local_id = cast_set<int32_t>(idx);
+ if (pblock != nullptr) {
+ const auto& meta = pblock->column_metas(cast_set<int>(idx));
+ // WAL metadata is the file-local schema; preserve its complete
nested shape so the
+ // mapper can validate ARRAY/MAP/STRUCT projections instead of
seeing an empty shell.
+ field = build_wal_column_definition(meta, cast_set<int32_t>(idx));
+ field.identifier = Field::create_field<TYPE_INT>(_column_ids[idx]);
+ } else {
+ const auto projected =
+ std::ranges::find_if(_projected_columns, [&](const auto&
candidate) {
+ return candidate.has_identifier_field_id() &&
+ candidate.get_identifier_field_id() ==
_column_ids[idx];
+ });
+ if (projected == _projected_columns.end()) {
+ continue;
+ }
+ field.name = projected->name;
+ field.type = projected->type;
+ }
+ _file_schema.push_back(std::move(field));
+ }
+ _schema_inited = true;
+ return Status::OK();
+}
+
+Status WalReader::_materialize_requested_columns(Block* source_block, Block*
file_block) const {
+ DORIS_CHECK(source_block != nullptr);
+ for (const auto& [file_column_id, block_position] :
_request->local_positions) {
+ const auto source_idx = file_column_id.value();
+ if (source_idx < 0 || cast_set<size_t>(source_idx) >=
source_block->columns()) {
+ return Status::Corruption("WAL request refers to invalid local
column {}", source_idx);
+ }
+ if (block_position.value() >= file_block->columns()) {
+ return Status::InternalError("WAL request has invalid block
position {}",
+ block_position.value());
+ }
+ const auto& target =
file_block->get_by_position(block_position.value());
+ // Deserialized WAL columns have a single owner. Move that ownership
into the output block
+ // before mutate() so wide payloads do not trigger copy-on-write deep
clones.
+ auto column =
std::move(source_block->get_by_position(source_idx).column);
+ column = make_column_nullable_if_needed(std::move(column),
target.type);
+ file_block->replace_by_position(block_position.value(),
IColumn::mutate(std::move(column)));
+ }
+ return Status::OK();
+}
+
+} // namespace doris::format::wal
diff --git a/be/src/format_v2/wal/wal_reader.h
b/be/src/format_v2/wal/wal_reader.h
new file mode 100644
index 00000000000..894822352fd
--- /dev/null
+++ b/be/src/format_v2/wal/wal_reader.h
@@ -0,0 +1,72 @@
+// 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>
+
+#include <cstdint>
+#include <memory>
+#include <string>
+#include <vector>
+
+#include "format_v2/file_reader.h"
+
+namespace doris {
+class WalFileReader;
+}
+
+namespace doris::format::wal {
+
+Status parse_wal_column_ids(const std::string& encoded, std::vector<int32_t>*
column_ids);
+
+class WalReader final : public FileReader {
+public:
+ 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);
+ ~WalReader() override;
+
+ Status init(RuntimeState* state) override;
+ Status get_schema(std::vector<ColumnDefinition>* file_schema) const
override;
+ std::unique_ptr<TableColumnMapper> create_column_mapper(
+ TableColumnMapperOptions options) const override;
+ Status open(std::shared_ptr<FileScanRequest> request) override;
+ Status get_block(Block* file_block, size_t* rows, bool* eof) override;
+ Status close() override;
+
+private:
+ Status _ensure_schema_loaded() const;
+ Status _validate_block_version(const PBlock& pblock) const;
+ Status _init_schema_from_block(const PBlock* pblock) const;
+ Status _materialize_requested_columns(Block* source_block, Block*
file_block) const;
+
+ const std::vector<ColumnDefinition> _projected_columns;
+ std::shared_ptr<doris::WalFileReader> _wal_reader;
+ std::string _wal_path;
+ uint32_t _version = 0;
+ std::vector<int32_t> _column_ids;
+ mutable std::vector<ColumnDefinition> _file_schema;
+ mutable PBlock _first_block;
+ mutable bool _first_block_loaded = false;
+ mutable bool _first_block_consumed = false;
+ mutable bool _schema_inited = false;
+ bool _reader_eof = false;
+};
+
+} // namespace doris::format::wal
diff --git a/be/src/format_v2/wal/wal_table_reader.cpp
b/be/src/format_v2/wal/wal_table_reader.cpp
new file mode 100644
index 00000000000..428ae7709af
--- /dev/null
+++ b/be/src/format_v2/wal/wal_table_reader.cpp
@@ -0,0 +1,47 @@
+// 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_table_reader.h"
+
+#include "format_v2/wal/wal_reader.h"
+#include "runtime/descriptors.h"
+
+namespace doris::format::wal {
+
+Status WalTableReader::annotate_projected_column(const TFileScanSlotInfo&,
+ ProjectedColumnBuildContext*
context,
+ ColumnDefinition* column)
const {
+ DORIS_CHECK(context != nullptr);
+ DORIS_CHECK(column != nullptr);
+ if (context->slot_desc == nullptr || context->slot_desc->col_unique_id() <
0) {
+ return Status::InternalError("WAL projected column {} has no valid
unique id",
+ column->name);
+ }
+ // WAL headers carry stable Doris column unique ids, so name-based
matching would return a
+ // renamed column from the wrong physical position.
+ column->identifier =
Field::create_field<TYPE_INT>(context->slot_desc->col_unique_id());
+ return Status::OK();
+}
+
+Status WalTableReader::create_file_reader(std::unique_ptr<FileReader>* reader)
{
+ DORIS_CHECK(reader != nullptr);
+ *reader = std::make_unique<WalReader>(_system_properties,
_current_task->data_file, _io_ctx,
+ _scanner_profile,
_projected_columns);
+ return Status::OK();
+}
+
+} // namespace doris::format::wal
diff --git a/be/src/format_v2/wal/wal_table_reader.h
b/be/src/format_v2/wal/wal_table_reader.h
new file mode 100644
index 00000000000..b172c21f87f
--- /dev/null
+++ b/be/src/format_v2/wal/wal_table_reader.h
@@ -0,0 +1,37 @@
+// 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 "format_v2/table_reader.h"
+
+namespace doris::format::wal {
+
+class WalTableReader final : public TableReader {
+public:
+ Status annotate_projected_column(const TFileScanSlotInfo& slot_info,
+ ProjectedColumnBuildContext* context,
+ ColumnDefinition* column) const override;
+
+protected:
+ Status create_file_reader(std::unique_ptr<FileReader>* reader) override;
+ TableColumnMappingMode mapping_mode() const override {
+ return TableColumnMappingMode::BY_FIELD_ID;
+ }
+};
+
+} // namespace doris::format::wal
diff --git a/be/test/exec/scan/file_scanner_v2_test.cpp
b/be/test/exec/scan/file_scanner_v2_test.cpp
index 87d4c92a776..0ca3213170d 100644
--- a/be/test/exec/scan/file_scanner_v2_test.cpp
+++ b/be/test/exec/scan/file_scanner_v2_test.cpp
@@ -86,6 +86,7 @@ TFileRangeDesc paimon_cpp_jni_range() {
auto range = range_with_format("paimon", TFileFormatType::FORMAT_JNI);
TPaimonFileDesc paimon_params;
paimon_params.__set_reader_type(TPaimonReaderType::PAIMON_CPP);
+ paimon_params.__set_file_format("parquet");
range.table_format_params.__set_paimon_params(std::move(paimon_params));
return range;
}
@@ -332,7 +333,7 @@ TEST(FileScannerV2Test, SupportedFormatMatrix) {
{"remote_doris", TFileFormatType::FORMAT_ARROW, std::nullopt,
true},
{"hive", TFileFormatType::FORMAT_ARROW, std::nullopt, false},
{"", TFileFormatType::FORMAT_ARROW, std::nullopt, false},
- {"", TFileFormatType::FORMAT_WAL, std::nullopt, false},
+ {"", TFileFormatType::FORMAT_WAL, std::nullopt, true},
{"", TFileFormatType::FORMAT_ES_HTTP, std::nullopt, false},
{"", TFileFormatType::FORMAT_LANCE, std::nullopt, false},
};
@@ -417,11 +418,13 @@ TEST(FileScannerV2Test,
FileScanLocalStateSelectsV2ForSupportedQueriesOnly) {
EXPECT_TRUE(FileScanLocalState::TEST_should_use_file_scanner_v2(query_options,
false, params));
EXPECT_FALSE(FileScanLocalState::TEST_should_use_file_scanner_v2(query_options,
true, params));
- const std::vector<TFileFormatType::type> unsupported_formats {
- TFileFormatType::FORMAT_WAL,
- TFileFormatType::FORMAT_ES_HTTP,
- TFileFormatType::FORMAT_LANCE,
- };
+ params.__set_format_type(TFileFormatType::FORMAT_WAL);
+
EXPECT_TRUE(FileScanLocalState::TEST_should_use_file_scanner_v2(query_options,
false, params));
+ params.__set_format_type(TFileFormatType::FORMAT_JNI);
+
EXPECT_TRUE(FileScanLocalState::TEST_should_use_file_scanner_v2(query_options,
false, params));
+
+ const std::vector<TFileFormatType::type> unsupported_formats
{TFileFormatType::FORMAT_ES_HTTP,
+
TFileFormatType::FORMAT_LANCE};
for (const auto format : unsupported_formats) {
params.__set_format_type(format);
EXPECT_FALSE(
@@ -441,24 +444,23 @@ TEST(FileScannerV2Test,
FileScanLocalStateSelectsV2ForSupportedQueriesOnly) {
EXPECT_FALSE(FileScanLocalState::TEST_should_use_file_scanner_v2(query_options,
false, params));
}
-TEST(FileScannerV2Test, JniCompatibilityShapesForceLegacyScanner) {
+TEST(FileScannerV2Test, JniCompatibilityShapesUseV2Scanner) {
TQueryOptions query_options;
query_options.__set_enable_file_scanner_v2(true);
query_options.__set_enable_paimon_cpp_reader(true);
TFileScanRangeParams params;
params.__set_format_type(TFileFormatType::FORMAT_JNI);
- // Rolling upgrades may carry the only Paimon marker and reader type on
each split. Since the
- // scan-level selector cannot inspect that split yet, JNI scans
conservatively stay on V1.
-
EXPECT_FALSE(FileScanLocalState::TEST_should_use_file_scanner_v2(query_options,
false, params));
- EXPECT_FALSE(FileScannerV2::is_supported(params, paimon_cpp_jni_range()));
+
EXPECT_TRUE(FileScanLocalState::TEST_should_use_file_scanner_v2(query_options,
false, params));
+ const auto cpp_range = paimon_cpp_jni_range();
+ EXPECT_FALSE(FileScannerV2::is_supported(params, cpp_range));
+ const auto cpp_status = FileScannerV2::TEST_validate_scan_range(params,
cpp_range);
+ EXPECT_TRUE(cpp_status.is<ErrorCode::NOT_IMPLEMENTED_ERROR>());
- // Older FEs can omit reader_type. The legacy scanner interprets this as
Paimon JNI when the C++
- // reader is disabled, so the scan-level choice must still stay on V1.
+ // Older FE plans without reader_type used Java whenever the C++ option
was disabled.
query_options.__set_enable_paimon_cpp_reader(false);
-
EXPECT_FALSE(FileScanLocalState::TEST_should_use_file_scanner_v2(query_options,
false, params));
- EXPECT_FALSE(
- FileScannerV2::is_supported(params,
legacy_paimon_jni_range_without_reader_type()));
+
EXPECT_TRUE(FileScanLocalState::TEST_should_use_file_scanner_v2(query_options,
false, params));
+ EXPECT_TRUE(FileScannerV2::is_supported(params,
legacy_paimon_jni_range_without_reader_type()));
}
TEST(FileScannerV2Test, FailedTableReaderCloseCanBeRetriedThroughScanner) {
@@ -563,6 +565,7 @@ TEST(FileScannerV2Test, FileFormatConversionMatrix) {
{TFileFormatType::FORMAT_JSON, format::FileFormat::JSON},
{TFileFormatType::FORMAT_NATIVE, format::FileFormat::NATIVE},
{TFileFormatType::FORMAT_ARROW, format::FileFormat::ARROW},
+ {TFileFormatType::FORMAT_WAL, format::FileFormat::WAL},
{TFileFormatType::FORMAT_ORC, format::FileFormat::ORC},
};
diff --git a/be/test/format_v2/jni/paimon_jni_reader_test.cpp
b/be/test/format_v2/jni/paimon_jni_reader_test.cpp
index 570977a5f7e..9c46aba05c7 100644
--- a/be/test/format_v2/jni/paimon_jni_reader_test.cpp
+++ b/be/test/format_v2/jni/paimon_jni_reader_test.cpp
@@ -25,6 +25,7 @@
#include "format_v2/table_reader.h"
#include "gen_cpp/PlanNodes_types.h"
+#include "runtime/runtime_state.h"
namespace doris::format::paimon {
namespace {
@@ -47,14 +48,15 @@ TFileScanRangeParams make_scan_params() {
return scan_params;
}
-Status init_reader(PaimonJniReader* reader, TFileScanRangeParams* scan_params)
{
+Status init_reader(PaimonJniReader* reader, TFileScanRangeParams* scan_params,
+ RuntimeState* runtime_state = nullptr) {
return reader->init({
.projected_columns = {},
.conjuncts = {},
.format = FileFormat::JNI,
.scan_params = scan_params,
.io_ctx = nullptr,
- .runtime_state = nullptr,
+ .runtime_state = runtime_state,
.scanner_profile = nullptr,
});
}
@@ -159,16 +161,20 @@ TEST(PaimonJniReaderTest,
ScanLevelOptionsOverrideLegacySplitFallbacks) {
EXPECT_EQ(params["hadoop.source"], "scan");
}
-TEST(PaimonJniReaderTest, KeepsInitialPhysicalBatchSizeAfterOpen) {
+TEST(PaimonJniReaderTest, UsesStableRuntimeBatchSizeBeforeAndAfterOpen) {
+ TQueryOptions query_options;
+ query_options.__set_batch_size(8160);
+ RuntimeState state {query_options, TQueryGlobals()};
+ auto scan_params = make_scan_params();
PaimonJniReader reader;
+ ASSERT_TRUE(init_reader(&reader, &scan_params, &state).ok());
+
reader.set_batch_size(32);
- EXPECT_EQ(reader.TEST_batch_size(), 32);
+ EXPECT_EQ(reader.TEST_batch_size(), 8160);
- // Paimon copies the constructor size into the RecordReader during Java
open. A later predictor
- // result cannot resize that physical reader, so keep the initial probe
size for the split.
reader.TEST_set_split_state(true, false);
reader.set_batch_size(1);
- EXPECT_EQ(reader.TEST_batch_size(), 32);
+ EXPECT_EQ(reader.TEST_batch_size(), 8160);
}
} // namespace
diff --git a/be/test/format_v2/table/paimon_reader_test.cpp
b/be/test/format_v2/table/paimon_reader_test.cpp
index 4186aa78f03..06301815b49 100644
--- a/be/test/format_v2/table/paimon_reader_test.cpp
+++ b/be/test/format_v2/table/paimon_reader_test.cpp
@@ -79,6 +79,16 @@ public:
}
};
+class SplitFormatTrackingTableReader final : public TableReader {
+public:
+ Status prepare_split(const SplitReadOptions& options) override {
+ prepared_format = options.current_split_format;
+ return Status::OK();
+ }
+
+ FileFormat prepared_format = FileFormat::JNI;
+};
+
DataTypePtr table_type(const DataTypePtr& type) {
return type->is_nullable() ? type : make_nullable(type);
}
@@ -322,8 +332,9 @@ TFileRangeDesc make_paimon_jni_range() {
return range;
}
-TFileRangeDesc make_paimon_range_without_reader_type(TFileFormatType::type
format_type) {
- TFileRangeDesc range = make_paimon_native_range(format_type);
+TFileRangeDesc make_legacy_paimon_native_range(TFileFormatType::type
physical_format_type) {
+ TFileRangeDesc range = make_paimon_native_range(physical_format_type);
+ range.__set_format_type(TFileFormatType::FORMAT_JNI);
range.table_format_params.paimon_params.__isset.reader_type = false;
return range;
}
@@ -663,7 +674,7 @@ TEST(PaimonHybridReaderTest,
ClassifiesJniSplitByReaderType) {
EXPECT_FALSE(paimon::PaimonHybridReader::TEST_is_jni_split(
make_paimon_native_range(TFileFormatType::FORMAT_PARQUET)));
EXPECT_FALSE(paimon::PaimonHybridReader::TEST_is_jni_split(
-
make_paimon_range_without_reader_type(TFileFormatType::FORMAT_JNI)));
+ make_legacy_paimon_native_range(TFileFormatType::FORMAT_PARQUET)));
EXPECT_TRUE(paimon::PaimonHybridReader::TEST_is_jni_split(make_paimon_jni_range()));
}
@@ -679,12 +690,54 @@ TEST(PaimonHybridReaderTest,
ConvertsNativeSplitFileFormat) {
.ok());
EXPECT_EQ(file_format, FileFormat::ORC);
+ ASSERT_TRUE(
+ paimon::PaimonHybridReader::TEST_to_file_format(
+
make_legacy_paimon_native_range(TFileFormatType::FORMAT_PARQUET), &file_format)
+ .ok());
+ EXPECT_EQ(file_format, FileFormat::PARQUET);
+
+ ASSERT_TRUE(paimon::PaimonHybridReader::TEST_to_file_format(
+
make_legacy_paimon_native_range(TFileFormatType::FORMAT_ORC), &file_format)
+ .ok());
+ EXPECT_EQ(file_format, FileFormat::ORC);
+
auto status =
paimon::PaimonHybridReader::TEST_to_file_format(make_paimon_jni_range(),
&file_format);
EXPECT_FALSE(status.ok());
EXPECT_NE(std::string::npos, status.to_string().find("Unsupported native
Paimon file format"));
}
+TEST(PaimonHybridReaderTest,
NormalizesLegacyNativeSplitFormatBeforeChildPrepare) {
+ RuntimeState state {TQueryOptions(), TQueryGlobals()};
+ auto scan_params = make_local_parquet_scan_params();
+ paimon::PaimonHybridReader reader;
+ SplitFormatTrackingTableReader* tracking_reader = nullptr;
+ reader.TEST_set_child_reader_factories(
+ [&] {
+ auto child =
std::make_unique<SplitFormatTrackingTableReader>();
+ tracking_reader = child.get();
+ return child;
+ },
+ [] { return std::make_unique<TableReader>(); });
+ ASSERT_TRUE(reader.init({
+ .projected_columns = {},
+ .conjuncts = {},
+ .format = FileFormat::JNI,
+ .scan_params = &scan_params,
+ .io_ctx = nullptr,
+ .runtime_state = &state,
+ .scanner_profile = nullptr,
+ })
+ .ok());
+
+ SplitReadOptions options;
+ options.current_range =
make_legacy_paimon_native_range(TFileFormatType::FORMAT_PARQUET);
+ options.current_split_format = FileFormat::JNI;
+ ASSERT_TRUE(reader.prepare_split(options).ok());
+ ASSERT_NE(tracking_reader, nullptr);
+ EXPECT_EQ(tracking_reader->prepared_format, FileFormat::PARQUET);
+}
+
TEST(PaimonHybridReaderTest, AdaptiveBatchSizeReachesBothChildReaders) {
paimon::PaimonHybridReader reader;
reader.TEST_install_batch_size_children();
diff --git a/be/test/format_v2/wal/wal_reader_test.cpp
b/be/test/format_v2/wal/wal_reader_test.cpp
new file mode 100644
index 00000000000..77c14afb12f
--- /dev/null
+++ b/be/test/format_v2/wal/wal_reader_test.cpp
@@ -0,0 +1,171 @@
+// 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 <gtest/gtest.h>
+
+#include <chrono>
+#include <filesystem>
+#include <memory>
+
+#include "agent/be_exec_version_manager.h"
+#include "core/block/block.h"
+#include "core/column/column_string.h"
+#include "core/data_type/data_type_array.h"
+#include "core/data_type/data_type_map.h"
+#include "core/data_type/data_type_nullable.h"
+#include "core/data_type/data_type_number.h"
+#include "core/data_type/data_type_string.h"
+#include "core/data_type/data_type_struct.h"
+#include "io/fs/local_file_system.h"
+#include "load/group_commit/wal/wal_file_reader.h"
+#include "load/group_commit/wal/wal_writer.h"
+
+namespace doris::format::wal {
+namespace {
+
+std::string temporary_wal_path() {
+ const auto root = std::filesystem::temp_directory_path() /
+ ("doris-wal-v2-" +
+
std::to_string(std::chrono::steady_clock::now().time_since_epoch().count()));
+ const auto path = root / "1" / "2" / "1_1_1_test";
+ std::filesystem::create_directories(path.parent_path());
+ return path.string();
+}
+
+PBlock serialize_block(const Block& block) {
+ PBlock pblock;
+ size_t uncompressed_bytes = 0;
+ size_t compressed_bytes = 0;
+ int64_t compress_time = 0;
+ EXPECT_TRUE(block.serialize(BeExecVersionManager::get_newest_version(),
&pblock,
+ &uncompressed_bytes, &compressed_bytes,
&compress_time,
+ segment_v2::CompressionTypePB::SNAPPY)
+ .ok());
+ return pblock;
+}
+
+} // namespace
+
+TEST(WalReaderV2Test, ParseColumnIdsPreservesHeaderOrder) {
+ std::vector<int32_t> column_ids;
+ ASSERT_TRUE(parse_wal_column_ids("17,4,99", &column_ids).ok());
+ EXPECT_EQ(column_ids, (std::vector<int32_t> {17, 4, 99}));
+}
+
+TEST(WalReaderV2Test, ParseColumnIdsRejectsMalformedOrAmbiguousHeaders) {
+ std::vector<int32_t> column_ids;
+ EXPECT_FALSE(parse_wal_column_ids("", &column_ids).ok());
+ EXPECT_FALSE(parse_wal_column_ids("17,,99", &column_ids).ok());
+ EXPECT_FALSE(parse_wal_column_ids("17,nope,99", &column_ids).ok());
+ EXPECT_FALSE(parse_wal_column_ids("17,4,17", &column_ids).ok());
+}
+
+TEST(WalReaderV2Test,
WriterBackedReaderPreservesNestedSchemaAndMovesFirstPayload) {
+ const auto wal_path = temporary_wal_path();
+ const auto int_type = make_nullable(std::make_shared<DataTypeInt32>());
+ const auto array_type =
make_nullable(std::make_shared<DataTypeArray>(int_type));
+ const auto string_type = make_nullable(std::make_shared<DataTypeString>());
+ const auto map_type =
make_nullable(std::make_shared<DataTypeMap>(string_type, int_type));
+ const auto struct_type = make_nullable(std::make_shared<DataTypeStruct>(
+ DataTypes {int_type, array_type}, Strings {"id", "nested_items"}));
+
+ Block source;
+ auto array_column = array_type->create_column();
+ array_column->insert_default();
+ source.insert({std::move(array_column), array_type, "items"});
+ auto string_column = string_type->create_column();
+ string_column->insert_data("value", 5);
+ source.insert({std::move(string_column), string_type, "renamed_later"});
+ auto map_column = map_type->create_column();
+ map_column->insert_default();
+ source.insert({std::move(map_column), map_type, "properties"});
+ auto struct_column = struct_type->create_column();
+ struct_column->insert_default();
+ source.insert({std::move(struct_column), struct_type, "record"});
+ auto pblock = serialize_block(source);
+
+ WalWriter writer(wal_path);
+ ASSERT_TRUE(writer.init(io::global_local_filesystem()).ok());
+ ASSERT_TRUE(writer.append_header("17,4,88,99").ok());
+ ASSERT_TRUE(writer.append_blocks({&pblock}).ok());
+ ASSERT_TRUE(writer.finalize().ok());
+
+ std::shared_ptr<io::FileSystemProperties> properties;
+ std::unique_ptr<io::FileDescription> description;
+ WalReader reader(properties, description, nullptr, nullptr, {});
+ reader._wal_reader = std::make_shared<WalFileReader>(wal_path);
+ ASSERT_TRUE(reader._wal_reader->init().ok());
+ std::string encoded_ids;
+ ASSERT_TRUE(reader._wal_reader->read_header(reader._version,
encoded_ids).ok());
+ ASSERT_TRUE(parse_wal_column_ids(encoded_ids, &reader._column_ids).ok());
+
+ std::vector<ColumnDefinition> schema;
+ ASSERT_TRUE(reader.get_schema(&schema).ok());
+ ASSERT_EQ(schema.size(), 4);
+ ASSERT_EQ(schema[0].children.size(), 1);
+ EXPECT_TRUE(schema[0].children[0].type->is_nullable());
+ ASSERT_EQ(schema[2].children.size(), 2);
+ EXPECT_TRUE(schema[2].children[0].type->is_nullable());
+ EXPECT_TRUE(schema[2].children[1].type->is_nullable());
+ ASSERT_EQ(schema[3].children.size(), 2);
+ EXPECT_EQ(schema[3].children[0].name, "id");
+ EXPECT_EQ(schema[3].children[1].name, "nested_items");
+ ASSERT_EQ(schema[3].children[1].children.size(), 1);
+
+ auto request = std::make_shared<FileScanRequest>();
+ request->local_positions.emplace(LocalColumnId(1), LocalIndex(0));
+ request->local_positions.emplace(LocalColumnId(0), LocalIndex(1));
+ ASSERT_TRUE(reader.open(std::move(request)).ok());
+ Block output({
+ {string_type->create_column(), string_type, "renamed"},
+ {array_type->create_column(), array_type, "items"},
+ });
+ size_t rows = 0;
+ bool eof = false;
+ ASSERT_TRUE(reader.get_block(&output, &rows, &eof).ok());
+ EXPECT_EQ(rows, 1);
+ EXPECT_FALSE(eof);
+ EXPECT_EQ(reader._first_block.ByteSizeLong(), 0);
+ EXPECT_EQ(output.get_by_position(0).column->get_data_at(0).to_string(),
"value");
+
+ ASSERT_TRUE(reader.close().ok());
+ std::filesystem::remove_all(
+
std::filesystem::path(wal_path).parent_path().parent_path().parent_path());
+}
+
+TEST(WalReaderV2Test, MaterializationTransfersColumnOwnership) {
+ std::shared_ptr<io::FileSystemProperties> properties;
+ std::unique_ptr<io::FileDescription> description;
+ WalReader reader(properties, description, nullptr, nullptr, {});
+ reader._request = std::make_shared<FileScanRequest>();
+ reader._request->local_positions.emplace(LocalColumnId(0), LocalIndex(0));
+
+ const auto type = std::make_shared<DataTypeString>();
+ auto source_column = ColumnString::create();
+ source_column->insert_data("payload", 7);
+ const auto* original = source_column.get();
+ Block source({{std::move(source_column), type, "value"}});
+ Block output({{type->create_column(), type, "value"}});
+
+ ASSERT_TRUE(reader._materialize_requested_columns(&source, &output).ok());
+ EXPECT_FALSE(static_cast<bool>(source.get_by_position(0).column));
+ EXPECT_EQ(output.get_by_position(0).column.get(), original);
+}
+
+} // namespace doris::format::wal
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/source/PaimonScanNode.java
b/fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/source/PaimonScanNode.java
index 990e79923f0..3bfada3e9ac 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/source/PaimonScanNode.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/source/PaimonScanNode.java
@@ -266,21 +266,11 @@ public class PaimonScanNode extends FileQueryScanNode {
String fileFormat = getFileFormat(paimonSplit.getPathString());
if (split != null) {
- // use jni reader or paimon-cpp reader
rangeDesc.setFormatType(TFileFormatType.FORMAT_JNI);
- // Use Paimon native serialization for paimon-cpp reader
- if (sessionVariable.isEnablePaimonCppReader() && split instanceof
DataSplit) {
- fileDesc.setReaderType(TPaimonReaderType.PAIMON_CPP);
-
fileDesc.setPaimonSplit(PaimonUtil.encodeDataSplitToString((DataSplit) split));
- } else {
- fileDesc.setReaderType(TPaimonReaderType.PAIMON_JNI);
-
fileDesc.setPaimonSplit(PaimonUtil.encodeObjectToString(split));
- }
- // Set table location for paimon-cpp reader
- String tableLocation = source.getTableLocation();
- if (tableLocation != null) {
- fileDesc.setPaimonTable(tableLocation);
- }
+ // A logical DataSplit may span multiple files, so keep it intact
for the JNI reader
+ // until the C++ path has a split-aware V2 adapter.
+ fileDesc.setReaderType(TPaimonReaderType.PAIMON_JNI);
+ fileDesc.setPaimonSplit(PaimonUtil.encodeObjectToString(split));
rangeDesc.setSelfSplitWeight(paimonSplit.getSelfSplitWeight());
} else {
// use native reader
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/datasource/paimon/source/PaimonScanNodeTest.java
b/fe/fe-core/src/test/java/org/apache/doris/datasource/paimon/source/PaimonScanNodeTest.java
index 9f8583ed1bc..30d79c47626 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/datasource/paimon/source/PaimonScanNodeTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/datasource/paimon/source/PaimonScanNodeTest.java
@@ -36,6 +36,7 @@ import org.apache.doris.planner.ScanContext;
import org.apache.doris.qe.SessionVariable;
import org.apache.doris.thrift.TFileRangeDesc;
import org.apache.doris.thrift.TFileScanRangeParams;
+import org.apache.doris.thrift.TPaimonReaderType;
import org.apache.doris.thrift.TPushAggOp;
import org.apache.paimon.data.BinaryRow;
@@ -658,6 +659,26 @@ public class PaimonScanNodeTest {
Assert.assertEquals(Arrays.asList(false, false),
rangeDesc.getColumnsFromPathIsNull());
}
+ @Test
+ public void testSetPaimonParamsUsesJniWhenCppOptionEnabled() throws
Exception {
+ Mockito.when(sv.isEnablePaimonCppReader()).thenReturn(true);
+ PaimonScanNode node = newTestNode(new PlanNodeId(0), new TupleId(0),
sv);
+ PaimonSource source = Mockito.mock(PaimonSource.class);
+
Mockito.when(source.getTableLocation()).thenReturn("file:///warehouse");
+ Table paimonTable =
mockPaimonTableWithPartitionKeys(Collections.emptyList());
+ Mockito.when(source.getPaimonTable()).thenReturn(paimonTable);
+ node.setSource(source);
+
+ TFileRangeDesc rangeDesc = new TFileRangeDesc();
+ invokePrivateMethod(node, "setPaimonParams",
+ new Class<?>[] {TFileRangeDesc.class, PaimonSplit.class},
+ rangeDesc, new
PaimonSplit(createDataSplit("jni-only.parquet")));
+
+ Assert.assertEquals(TPaimonReaderType.PAIMON_JNI,
+
rangeDesc.getTableFormatParams().getPaimonParams().getReaderType());
+
Assert.assertTrue(rangeDesc.getTableFormatParams().getPaimonParams().isSetPaimonSplit());
+ }
+
@Test
public void testGetFieldIndexMatchesMixedCaseColumns() {
List<String> fieldNames = Arrays.asList("data", "mIxEd_COL", "PART");
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
