Skip to content

Paimon reuse scanner #65404

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
zhangstar333 wants to merge 3 commits into
base: master
Choose a base branch
from
Open

Paimon reuse scanner #65404

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
026810f
[opt](paimon) control paimon jni reader async file reader
zhangstar333 Jul 8, 2026
9b5dcfa
update
zhangstar333 Jul 9, 2026
c273a9f
reuse PaimonJniScanner
zhangstar333 Jul 9, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 10 additions & 6 deletions be/src/format_v2/jni/jni_table_reader.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -48,13 +48,12 @@ Status JniTableReader::prepare_split(const SplitReadOptions& options) {
RETURN_IF_ERROR(validate_scan_range(options.current_range));
RETURN_IF_ERROR(TableReader::prepare_split(options));
DORIS_CHECK(!_closed);
DORIS_CHECK(!_scanner_opened);
if (_is_table_level_count_active()) {
return Status::OK();
}
// Subclasses populate split-specific scanner params before calling this method, so the Java
// scanner can be opened here instead of being lazily opened by the first get_block() call.
return _open_jni_scanner();
return open_jni_scanner_for_split();
}

Status JniTableReader::get_block(Block* output_block, bool* eos) {
Expand All @@ -79,7 +78,7 @@ Status JniTableReader::get_block(Block* output_block, bool* eos) {
RETURN_IF_ERROR(_get_next_jni_block(&current_rows, &current_eof));
if (current_eof) {
_eof = true;
RETURN_IF_ERROR(_close_jni_scanner());
RETURN_IF_ERROR(close_jni_scanner_for_split());
*eos = true;
return Status::OK();
}
Expand Down Expand Up @@ -280,11 +279,11 @@ Status JniTableReader::close() {
return Status::OK();
}
_closed = true;
RETURN_IF_ERROR(_close_jni_scanner());
RETURN_IF_ERROR(JniTableReader::close_jni_scanner_for_split());
return TableReader::close();
}

Status JniTableReader::_close_jni_scanner() {
Status JniTableReader::close_jni_scanner_for_split() {
if (!_scanner_opened) {
JNIEnv* env = nullptr;
if (!_jni_scanner_obj.uninitialized()) {
Expand Down Expand Up @@ -343,7 +342,8 @@ void JniTableReader::_reset_split_state(JNIEnv* env) {
_fill_block_watcher = 0;
}

Status JniTableReader::_open_jni_scanner() {
Status JniTableReader::open_jni_scanner_for_split() {
DORIS_CHECK(!_scanner_opened);
// subclasses build map<string,string> _scanner_params to JAVA side
RETURN_IF_ERROR(build_scanner_params(&_scanner_params));
// subclasses build _jni_columns info to JAVA side, including column name and column type
Expand Down Expand Up @@ -424,6 +424,10 @@ Status JniTableReader::_register_jni_class_functions_once(JNIEnv* env) {
&_jni_scanner_get_statistics));
RETURN_IF_ERROR(
_jni_scanner_cls.get_method(env, "setBatchSize", "(I)V", &_jni_scanner_set_batch_size));
RETURN_IF_ERROR(_jni_scanner_cls.get_method(env, "prepareForSplit", "(Ljava/util/Map;)V",
&_jni_scanner_prepare_for_split));
RETURN_IF_ERROR(_jni_scanner_cls.get_method(env, "resetCurrentSplit", "()V",
&_jni_scanner_reset_current_split));
return Status::OK();
}

Expand Down
17 changes: 13 additions & 4 deletions be/src/format_v2/jni/jni_table_reader.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -60,6 +60,17 @@ class JniTableReader : public TableReader {
// Subclasses can override this method when Java transfer types differ from output types.
virtual Status build_jni_columns(std::vector<JniColumn>* columns) const;
virtual Status finalize_jni_block(Block* jni_block, Block* output_block, size_t* rows);
virtual Status open_jni_scanner_for_split();
virtual Status close_jni_scanner_for_split();
void reset_jni_eof() { _eof = false; }
bool jni_scanner_opened() const { return _scanner_opened; }
Jni::GlobalObject& jni_scanner_obj() { return _jni_scanner_obj; }
const Jni::MethodId& jni_scanner_prepare_for_split() const {
return _jni_scanner_prepare_for_split;
}
const Jni::MethodId& jni_scanner_reset_current_split() const {
return _jni_scanner_reset_current_split;
}
// used for profile
virtual int64_t self_split_weight() const;
const std::vector<JniColumn>& jni_columns() const { return _jni_columns; }
Expand All @@ -69,8 +80,6 @@ class JniTableReader : public TableReader {
// init
void _init_profile();
std::string _connector_name() const;
// open
Status _open_jni_scanner();
void _reset_split_state(JNIEnv* env);
void _prepare_jni_scanner_schema();
Status _register_jni_class_functions_once(JNIEnv* env);
Expand All @@ -81,8 +90,6 @@ class JniTableReader : public TableReader {
Status _get_statistics(JNIEnv* env, std::map<std::string, std::string>* result);
void _collect_jni_scanner_profile(JNIEnv* env);

Status _close_jni_scanner();

std::map<std::string, std::string> _scanner_params;
std::vector<JniColumn> _jni_columns;
Block _jni_block_template;
Expand Down Expand Up @@ -114,6 +121,8 @@ class JniTableReader : public TableReader {
Jni::MethodId _jni_scanner_release_table;
Jni::MethodId _jni_scanner_get_statistics;
Jni::MethodId _jni_scanner_set_batch_size;
Jni::MethodId _jni_scanner_prepare_for_split;
Jni::MethodId _jni_scanner_reset_current_split;
};

} // namespace doris::format
52 changes: 51 additions & 1 deletion be/src/format_v2/jni/paimon_jni_reader.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -91,7 +91,6 @@ Status PaimonJniReader::build_scanner_params(std::map<std::string, std::string>*
const auto& paimon_params = _current_range.table_format_params.paimon_params;
const auto* paimon_predicate = get_paimon_predicate(_scan_params, paimon_params);
DORIS_CHECK(paimon_predicate != nullptr);
(*params)["paimon_split"] = paimon_params.paimon_split;
(*params)["paimon_predicate"] = *paimon_predicate;
(*params)["serialized_table"] = _scan_params->serialized_table;

Expand Down Expand Up @@ -125,6 +124,57 @@ Status PaimonJniReader::build_scanner_params(std::map<std::string, std::string>*
return Status::OK();
}

Status PaimonJniReader::open_jni_scanner_for_split() {
if (!jni_scanner_opened()) {
RETURN_IF_ERROR(JniTableReader::open_jni_scanner_for_split());
}
return _prepare_for_split();
}

Status PaimonJniReader::close_jni_scanner_for_split() {
RETURN_IF_ERROR(_reset_current_split());
reset_jni_eof();
return Status::OK();
}

Status PaimonJniReader::_prepare_for_split() {
DORIS_CHECK(jni_scanner_opened());
DORIS_CHECK(!_current_split_prepared);
std::map<std::string, std::string> split_params;
split_params["paimon_split"] = _current_range.table_format_params.paimon_params.paimon_split;

JNIEnv* env = nullptr;
RETURN_IF_ERROR(Jni::Env::Get(&env));
Jni::LocalObject hashmap_object;
RETURN_IF_ERROR(Jni::Util::convert_to_java_map(env, split_params, &hashmap_object));
RETURN_IF_ERROR(jni_scanner_obj()
.call_void_method(env, jni_scanner_prepare_for_split())
.with_arg(hashmap_object)
.call());
RETURN_ERROR_IF_EXC(env);
_current_split_prepared = true;
reset_jni_eof();
return Status::OK();
}

Status PaimonJniReader::close() {

@github-actions github-actions Bot Jul 9, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

close() should still run the base JNI cleanup even if resetting the current split reports an error. resetCurrentSplit() can throw when releasing an active batch or closing the Paimon reader, and Java close() is explicitly written to keep going with IOManager cleanup/metric accounting after that reset error. This C++ override returns immediately, so JniTableReader::close() never reaches the idempotent releaseTable() or Java close() call, and FileScannerV2::close() has already marked the scanner closed so it will not retry. Please preserve the first reset error but still attempt JniTableReader::close() before returning.

RETURN_IF_ERROR(_reset_current_split());
return JniTableReader::close();
}

Status PaimonJniReader::_reset_current_split() {
if (!_current_split_prepared) {
return Status::OK();
}
JNIEnv* env = nullptr;
RETURN_IF_ERROR(Jni::Env::Get(&env));
RETURN_IF_ERROR(
jni_scanner_obj().call_void_method(env, jni_scanner_reset_current_split()).call());
RETURN_ERROR_IF_EXC(env);
_current_split_prepared = false;
return Status::OK();
}

std::string PaimonJniReader::build_default_io_manager_tmp_dirs(
const std::vector<StorePath>& store_paths) {
std::vector<std::string> tmp_dirs;
Expand Down
7 changes: 7 additions & 0 deletions be/src/format_v2/jni/paimon_jni_reader.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,6 +32,7 @@ namespace doris::format::paimon {
class PaimonJniReader final : public format::JniTableReader {
public:
~PaimonJniReader() override = default;
Status close() override;

#ifdef BE_TEST
void TEST_set_scan_params(TFileScanRangeParams* params) { _scan_params = params; }
Expand All @@ -50,9 +51,15 @@ class PaimonJniReader final : public format::JniTableReader {
std::string connector_class() const override;
Status validate_scan_range(const TFileRangeDesc& range) const override;
Status build_scanner_params(std::map<std::string, std::string>* params) const override;
Status open_jni_scanner_for_split() override;
Status close_jni_scanner_for_split() override;

private:
Status _prepare_for_split();
Status _reset_current_split();
static std::string build_default_io_manager_tmp_dirs(const std::vector<StorePath>& store_paths);

bool _current_split_prepared = false;
};

} // namespace doris::format::paimon
8 changes: 8 additions & 0 deletions fe/be-java-extensions/java-common/src/main/java/org/apache/doris/common/jni/JniScanner.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -52,6 +52,14 @@ public abstract class JniScanner {
// Close JniScanner and release resources
public abstract void close() throws IOException;

public void prepareForSplit(Map<String, String> splitParams) throws IOException {
throw new UnsupportedOperationException(getClass().getName() + " does not support split preparation");
}

public void resetCurrentSplit() throws IOException {
throw new UnsupportedOperationException(getClass().getName() + " does not support split reset");
}

// Scan data and save as vector table
protected abstract int getNext() throws IOException;

Expand Down
Loading
Loading