This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-4.1
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-4.1 by this push:
new 8740cb4301f [opt](lance) add some profiles in lance node (#67005)
8740cb4301f is described below
commit 8740cb4301fa8bfc0736d85251e742c7a3161487
Author: zhangstar333 <[email protected]>
AuthorDate: Thu Aug 27 14:26:14 2026 +0800
[opt](lance) add some profiles in lance node (#67005)
### What problem does this PR solve?
Problem Summary:
1. update lance version to 0.1.7
2. add some profiles in lance node.
### Release note
None
### Check List (For Author)
- Test <!-- At least one of them must be included. -->
- [] Regression test
- [x] Unit Test
- [ ] Manual test (add detailed scripts or steps below)
- [ ] No need to test or manual test. Explain why:
- [ ] This is a refactor/code format and no logic has been changed.
- [ ] Previous test can cover this change.
- [ ] No code files have been changed.
- [ ] Other reason <!-- Add your reason? -->
- Behavior changed:
- [x] No.
- [ ] Yes. <!-- Explain the behavior change -->
- Does this need documentation?
- [ ] No.
- [ ] Yes. <!-- Add document PR link here. eg:
https://github.com/apache/doris-website/pull/1214 -->
### Check List (For Reviewer who merge this PR)
- [ ] Confirm the release note
- [ ] Confirm test cases
- [ ] Confirm document
- [ ] Add branch pick label <!-- Add branch pick label that this PR
should merge into -->
---
be/src/exec/operator/materialization_opertor.cpp | 15 +-
be/src/exec/rowid_fetcher.cpp | 31 +-
be/src/exec/rowid_fetcher.h | 7 +
be/src/format_v2/table/lance_reader.cpp | 224 ++-
be/src/format_v2/table/lance_reader.h | 20 +-
.../operator/materialization_shared_state_test.cpp | 27 +
be/test/format_v2/table/lance_reader_test.cpp | 77 +-
.../datasource/lance/source/LanceScanNode.java | 2 -
.../lance/test_lance_rest_catalog.out | 1 -
.../lance/test_lance_vector_search.groovy | 8 +-
.../test_lance_vector_search_two_phase.groovy | 2 -
thirdparty/download-thirdparty.sh | 6 +-
thirdparty/patches/lance-c-0.1.6-doris.patch | 632 --------
thirdparty/patches/lance-c-0.1.7-pr-64.patch | 1522 ++++++++++++++++++++
thirdparty/vars.sh | 8 +-
15 files changed, 1899 insertions(+), 683 deletions(-)
diff --git a/be/src/exec/operator/materialization_opertor.cpp
b/be/src/exec/operator/materialization_opertor.cpp
index 695efb5ef78..47f65cd8c5f 100644
--- a/be/src/exec/operator/materialization_opertor.cpp
+++ b/be/src/exec/operator/materialization_opertor.cpp
@@ -449,16 +449,20 @@ Status
MaterializationSharedState::validate_rpc_results(int node_id) {
void MaterializationSharedState::_update_profile_info(int64_t backend_id,
RuntimeProfile*
response_profile) {
+ DORIS_CHECK(response_profile != nullptr);
if (!backend_profile_info_string.contains(backend_id)) {
backend_profile_info_string.emplace(backend_id,
std::map<std::string,
fmt::memory_buffer> {});
}
auto& info_map = backend_profile_info_string[backend_id];
- auto update_profile_info_key = [&](const std::string& info_key) {
+ auto update_profile_info_key = [&](const std::string& info_key, bool
warn_if_missing = true) {
const auto* info_value = response_profile->get_info_string(info_key);
if (info_value == nullptr) [[unlikely]] {
- LOG(WARNING) << "Get row id fetch rpc profile success, but no info
key :" << info_key;
+ if (warn_if_missing) {
+ LOG(WARNING) << "Get row id fetch rpc profile success, but no
info key :"
+ << info_key;
+ }
return;
}
if (!info_map.contains(info_key)) {
@@ -473,6 +477,10 @@ void
MaterializationSharedState::_update_profile_info(int64_t backend_id,
update_profile_info_key(RowIdStorageReader::FileReadLinesProfile);
update_profile_info_key(FileScanner::FileReadBytesProfile);
update_profile_info_key(FileScanner::FileReadTimeProfile);
+ update_profile_info_key(RowIdStorageReader::LanceDatasetOpenTimeProfile,
false);
+ update_profile_info_key(RowIdStorageReader::LanceRowIdTakeReadTimeProfile,
false);
+
update_profile_info_key(RowIdStorageReader::LanceArrowToDorisBlockTimeProfile,
false);
+
update_profile_info_key(RowIdStorageReader::LanceRowIdFetchTotalTimeProfile,
false);
}
Status MaterializationSharedState::create_muiltget_result(const Columns&
columns, bool child_eos,
@@ -654,8 +662,7 @@ Status MaterializationOperator::pull(RuntimeState* state,
Block* output_block, b
local_state._materialization_state.backend_profile_info_string) {
auto* child_profile = local_state.operator_profile()->create_child(
"RowIDFetcher: BackendId:" + std::to_string(backend_id));
- for (const auto& [info_key, info_value] :
-
local_state._materialization_state.backend_profile_info_string[backend_id]) {
+ for (const auto& [info_key, info_value] : child_info) {
child_profile->add_info_string(info_key, "{" +
fmt::to_string(info_value) + "}");
}
local_state.operator_profile()->add_child(child_profile, true);
diff --git a/be/src/exec/rowid_fetcher.cpp b/be/src/exec/rowid_fetcher.cpp
index bddf2ec4ad0..a8c3689d578 100644
--- a/be/src/exec/rowid_fetcher.cpp
+++ b/be/src/exec/rowid_fetcher.cpp
@@ -78,6 +78,7 @@
#include "util/brpc_client_cache.h" // BrpcClientCache
#include "util/defer_op.h"
#include "util/jsonb/serialize.h"
+#include "util/pretty_printer.h"
namespace doris {
@@ -724,6 +725,11 @@ const std::string
RowIdStorageReader::ScannersRunningTimeProfile = "ScannersRunn
const std::string RowIdStorageReader::InitReaderAvgTimeProfile =
"InitReaderAvgTime";
const std::string RowIdStorageReader::GetBlockAvgTimeProfile =
"GetBlockAvgTime";
const std::string RowIdStorageReader::FileReadLinesProfile = "FileReadLines";
+const std::string RowIdStorageReader::LanceDatasetOpenTimeProfile =
"LanceDatasetOpenTime";
+const std::string RowIdStorageReader::LanceRowIdTakeReadTimeProfile =
"LanceRowIdTakeReadTime";
+const std::string RowIdStorageReader::LanceArrowToDorisBlockTimeProfile =
+ "LanceArrowToDorisBlockTime";
+const std::string RowIdStorageReader::LanceRowIdFetchTotalTimeProfile =
"LanceRowIdFetchTotalTime";
const std::string
RowIdStorageReader::TopNLazyMaterializationSecondPhaseLocalIOCount =
"TopNLazyMaterializationSecondPhaseLocalIOCount";
const std::string
RowIdStorageReader::TopNLazyMaterializationSecondPhaseLocalIOBytes =
@@ -784,7 +790,18 @@ Status RowIdStorageReader::read_lance_rows_by_row_ids(
RETURN_IF_ERROR(scope_timer_run(
[&]() { return reader.read_by_row_ids(scan_range_desc, row_ids,
block); },
&fetch_statistics->get_block_ms));
- return reader.close();
+ RETURN_IF_ERROR(reader.close());
+
+ const auto collect_lance_fetch_time = [&](const std::string& timer_name) {
+ if (const auto* counter = runtime_profile->get_counter(timer_name);
counter != nullptr) {
+ fetch_statistics->lance_fetch_times_ns.emplace(timer_name,
counter->value());
+ }
+ };
+ collect_lance_fetch_time(LanceDatasetOpenTimeProfile);
+ collect_lance_fetch_time(LanceRowIdTakeReadTimeProfile);
+ collect_lance_fetch_time(LanceArrowToDorisBlockTimeProfile);
+ collect_lance_fetch_time(LanceRowIdFetchTotalTimeProfile);
+ return Status::OK();
}
Status RowIdStorageReader::read_external_row_from_file_mapping(
@@ -1108,13 +1125,17 @@ Status RowIdStorageReader::read_batch_external_row(
fmt::memory_buffer file_read_times_buffer;
format_to(file_read_times_buffer, "[");
+ std::map<std::string, int64_t> lance_fetch_times_ns;
size_t idx = 0;
for (const auto& [_, scan_info] : scan_rows) {
format_to(file_read_lines_buffer, "{}, ", scan_info.first.size());
- *init_reader_avg_ms = fetch_statistics[idx].init_reader_ms;
+ *init_reader_avg_ms += fetch_statistics[idx].init_reader_ms;
*get_block_avg_ms += fetch_statistics[idx].get_block_ms;
format_to(file_read_bytes_buffer, "{}, ",
fetch_statistics[idx].file_read_bytes);
format_to(file_read_times_buffer, "{}, ",
fetch_statistics[idx].file_read_times);
+ for (const auto& [time_name, time_value] :
fetch_statistics[idx].lance_fetch_times_ns) {
+ lance_fetch_times_ns[time_name] += time_value;
+ }
idx++;
}
@@ -1127,13 +1148,17 @@ Status RowIdStorageReader::read_batch_external_row(
runtime_profile->add_info_string(InitReaderAvgTimeProfile,
std::to_string(*init_reader_avg_ms) +
"ms");
runtime_profile->add_info_string(GetBlockAvgTimeProfile,
- std::to_string(*init_reader_avg_ms) +
"ms");
+ std::to_string(*get_block_avg_ms) +
"ms");
runtime_profile->add_info_string(FileReadLinesProfile,
fmt::to_string(file_read_lines_buffer));
runtime_profile->add_info_string(FileScanner::FileReadBytesProfile,
fmt::to_string(file_read_bytes_buffer));
runtime_profile->add_info_string(FileScanner::FileReadTimeProfile,
fmt::to_string(file_read_times_buffer));
+ for (const auto& [time_name, time_value] : lance_fetch_times_ns) {
+ runtime_profile->add_info_string(time_name,
+ PrettyPrinter::print(time_value,
TUnit::TIME_NS));
+ }
}
runtime_profile->to_proto(pprofile, 2);
diff --git a/be/src/exec/rowid_fetcher.h b/be/src/exec/rowid_fetcher.h
index 41acea97c1d..ceb7a51c1a0 100644
--- a/be/src/exec/rowid_fetcher.h
+++ b/be/src/exec/rowid_fetcher.h
@@ -22,8 +22,10 @@
#include <gen_cpp/DataSinks_types.h>
#include <gen_cpp/internal_service.pb.h>
+#include <map>
#include <memory>
#include <semaphore>
+#include <string>
#include <utility>
#include <vector>
@@ -101,6 +103,10 @@ public:
static const std::string InitReaderAvgTimeProfile;
static const std::string GetBlockAvgTimeProfile;
static const std::string FileReadLinesProfile;
+ static const std::string LanceDatasetOpenTimeProfile;
+ static const std::string LanceRowIdTakeReadTimeProfile;
+ static const std::string LanceArrowToDorisBlockTimeProfile;
+ static const std::string LanceRowIdFetchTotalTimeProfile;
static const std::string TopNLazyMaterializationSecondPhaseLocalIOCount;
static const std::string TopNLazyMaterializationSecondPhaseLocalIOBytes;
static const std::string TopNLazyMaterializationSecondPhaseRemoteIOCount;
@@ -165,6 +171,7 @@ private:
struct ExternalFetchStatistics {
int64_t init_reader_ms = 0;
int64_t get_block_ms = 0;
+ std::map<std::string, int64_t> lance_fetch_times_ns;
std::string file_read_bytes;
std::string file_read_times;
};
diff --git a/be/src/format_v2/table/lance_reader.cpp
b/be/src/format_v2/table/lance_reader.cpp
index 29d334f4b98..e8ad0c5229a 100644
--- a/be/src/format_v2/table/lance_reader.cpp
+++ b/be/src/format_v2/table/lance_reader.cpp
@@ -30,6 +30,7 @@
#include <memory>
#include "common/consts.h"
+#include "common/logging.h"
#include "core/column/column_nullable.h"
#include "core/column/column_string.h"
#include "core/data_type/data_type_array.h"
@@ -38,6 +39,7 @@
#include "core/data_type/data_type_nothing.h"
#include "core/data_type/data_type_struct.h"
#include "exec/common/endian.h"
+#include "runtime/file_scan_profile.h"
#include "storage/utils.h"
namespace doris::format::lance {
@@ -58,6 +60,7 @@ struct LanceBatchDeleter {
constexpr std::string_view DISTANCE_COLUMN = "_distance";
constexpr std::string_view ROW_ID_COLUMN = "_rowid";
constexpr std::string_view ARROW_EXTENSION_NAME = "ARROW:extension:name";
+constexpr const char* LANCE_READER_PROFILE = "LanceReader";
size_t vector_element_width(TVectorElementType::type type) {
switch (type) {
@@ -303,21 +306,77 @@ Status LanceTableReader::init(TableReadOptions&& options)
{
_ctz = _runtime_state->timezone_obj();
const auto& lance_scan_params = _scan_params->lance_scan_params;
+ ADD_CHILD_TIMER_WITH_LEVEL(_scanner_profile, LANCE_READER_PROFILE,
+ file_scan_profile::TABLE_READER, 1);
+ _dataset_open_time = ADD_CHILD_TIMER_WITH_LEVEL(_scanner_profile,
"LanceDatasetOpenTime",
+ LANCE_READER_PROFILE, 1);
+ _scanner_configure_time = ADD_CHILD_TIMER_WITH_LEVEL(
+ _scanner_profile, "LanceScannerConfigureTime",
LANCE_READER_PROFILE, 1);
+ _scanner_read_time = ADD_CHILD_TIMER_WITH_LEVEL(_scanner_profile,
"LanceScannerReadTime",
+ LANCE_READER_PROFILE, 1);
+ _arrow_to_doris_block_time = ADD_CHILD_TIMER_WITH_LEVEL(
+ _scanner_profile, "LanceArrowToDorisBlockTime",
LANCE_READER_PROFILE, 1);
+ _execution_iops = ADD_CHILD_COUNTER_WITH_LEVEL(_scanner_profile,
"LanceExecutionIOOps",
+ TUnit::UNIT,
LANCE_READER_PROFILE, 1);
+ _execution_requests = ADD_CHILD_COUNTER_WITH_LEVEL(_scanner_profile,
"LanceExecutionIORequests",
+ TUnit::UNIT,
LANCE_READER_PROFILE, 1);
+ _execution_bytes_read = ADD_CHILD_COUNTER_WITH_LEVEL(
+ _scanner_profile, "LanceExecutionIOBytesRead", TUnit::BYTES,
LANCE_READER_PROFILE, 1);
+ _index_partition_cache_miss_loads =
+ ADD_CHILD_COUNTER_WITH_LEVEL(_scanner_profile,
"LanceIndexPartitionCacheMissLoads",
+ TUnit::UNIT, LANCE_READER_PROFILE, 1);
+ _index_comparisons = ADD_CHILD_COUNTER_WITH_LEVEL(_scanner_profile,
"LanceIndexComparisons",
+ TUnit::UNIT,
LANCE_READER_PROFILE, 1);
+ // These scan counts are emitted by Lance's FilteredRead execution node.
For vector searches
+ // with an explicit fragment set, they normally describe the fragments,
ranges, and rows read
+ // while applying the row-id prefilter. They are scan input counts, not
ANN result counts.
+ _lance_count_metrics = {
+ {"fragments_scanned",
+ ADD_CHILD_COUNTER_WITH_LEVEL(_scanner_profile,
"LanceFragmentsScanned", TUnit::UNIT,
+ LANCE_READER_PROFILE, 1)},
+ {"ranges_scanned",
+ ADD_CHILD_COUNTER_WITH_LEVEL(_scanner_profile,
"LanceRowOffsetRangesScanned",
+ TUnit::UNIT, LANCE_READER_PROFILE,
1)},
+ {"rows_scanned", ADD_CHILD_COUNTER_WITH_LEVEL(_scanner_profile,
"LanceRowsScanned",
+ TUnit::UNIT,
LANCE_READER_PROFILE, 1)},
+ {"partitions_ranked",
+ ADD_CHILD_COUNTER_WITH_LEVEL(_scanner_profile,
"LanceIVFPartitionsRanked", TUnit::UNIT,
+ LANCE_READER_PROFILE, 1)},
+ {"partitions_searched",
+ ADD_CHILD_COUNTER_WITH_LEVEL(_scanner_profile,
"LanceIVFPartitionsSearched",
+ TUnit::UNIT, LANCE_READER_PROFILE,
1)},
+ {"deltas_searched",
+ ADD_CHILD_COUNTER_WITH_LEVEL(_scanner_profile,
"LanceVectorIndexSegmentsSearched",
+ TUnit::UNIT, LANCE_READER_PROFILE,
1)},
+ };
+ _lance_time_metrics = {
+ // This is wait time reported by the same Lance scan execution
node described above,
+ // rather than Doris scanner scheduling wait time.
+ {"task_wait_time", ADD_CHILD_TIMER_WITH_LEVEL(_scanner_profile,
"LanceTaskWaitTime",
+
LANCE_READER_PROFILE, 1)},
+ {"find_partitions_elapsed",
+ ADD_CHILD_TIMER_WITH_LEVEL(_scanner_profile,
"LanceIVFPartitionRankingTime",
+ LANCE_READER_PROFILE, 1)},
+ };
_vector_search = _scan_params->__isset.lance_scan_params &&
lance_scan_params.__isset.external_search_request;
if (_vector_search) {
RETURN_IF_ERROR(_validate_external_search_request());
const auto& request = lance_scan_params.external_search_request;
const auto& vector = request.search_query.vector_search;
- const bool use_index = !request.__isset.vector_search_options ||
-
!request.vector_search_options.__isset.use_index ||
- request.vector_search_options.use_index;
- _scanner_profile->add_info_string("LanceFragmentTopK",
std::to_string(vector.top_k));
- _scanner_profile->add_info_string("LanceFragmentOffset",
std::to_string(vector.offset));
- _scanner_profile->add_info_string("LanceVectorDimension",
-
std::to_string(vector.query_vector.dimension));
- _scanner_profile->add_info_string("LanceUseIndex", use_index ? "true"
: "false");
- _fragment_count = ADD_COUNTER(_scanner_profile, "LanceFragmentCount",
TUnit::UNIT);
+ _scanner_profile->add_info_string("LanceTopK",
std::to_string(vector.top_k));
+ _scanner_profile->add_info_string("LanceOffset",
std::to_string(vector.offset));
+ _scanner_profile->add_info_string("LanceTopKPlusOffset",
+ std::to_string(vector.top_k +
vector.offset));
+ _planned_index_segment_count =
+ ADD_CHILD_COUNTER_WITH_LEVEL(_scanner_profile,
"LancePlannedIndexSegmentCount",
+ TUnit::UNIT,
LANCE_READER_PROFILE, 1);
+ _planned_indexed_fragment_count =
+ ADD_CHILD_COUNTER_WITH_LEVEL(_scanner_profile,
"LancePlannedIndexedFragmentCount",
+ TUnit::UNIT,
LANCE_READER_PROFILE, 1);
+ _planned_flat_search_fragment_count = ADD_CHILD_COUNTER_WITH_LEVEL(
+ _scanner_profile, "LancePlannedFlatSearchFragmentCount",
TUnit::UNIT,
+ LANCE_READER_PROFILE, 1);
}
if (_scan_params->__isset.lance_scan_params &&
lance_scan_params.__isset.lance_substrait_filter) {
@@ -415,7 +474,11 @@ Status LanceTableReader::get_block(Block* block, bool*
eos) {
}
LanceBatch* raw_batch = nullptr;
- const int32_t scan_status = lance_scanner_next(_scanner,
&raw_batch);
+ int32_t scan_status = 0;
+ {
+ SCOPED_TIMER(_scanner_read_time);
+ scan_status = lance_scanner_next(_scanner, &raw_batch);
+ }
if (scan_status == 1) {
_eof = true;
_close_scanner();
@@ -427,7 +490,10 @@ Status LanceTableReader::get_block(Block* block, bool*
eos) {
std::unique_ptr<LanceBatch, LanceBatchDeleter> batch(raw_batch);
size_t rows = 0;
- RETURN_IF_ERROR(_fill_block_from_lance_batch(batch.get(), block,
&rows));
+ {
+ SCOPED_TIMER(_arrow_to_doris_block_time);
+ RETURN_IF_ERROR(_fill_block_from_lance_batch(batch.get(),
block, &rows));
+ }
_record_scan_rows(rows);
raw_rows += rows;
}
@@ -457,6 +523,15 @@ Status LanceTableReader::read_by_row_ids(const
TFileRangeDesc& range,
if (row_ids.empty()) {
return Status::OK();
}
+ if (_row_id_take_read_time == nullptr) {
+ _row_id_take_read_time = ADD_CHILD_TIMER_WITH_LEVEL(
+ _scanner_profile, "LanceRowIdTakeReadTime",
LANCE_READER_PROFILE, 1);
+ }
+ if (_row_id_fetch_total_time == nullptr) {
+ _row_id_fetch_total_time = ADD_CHILD_TIMER_WITH_LEVEL(
+ _scanner_profile, "LanceRowIdFetchTotalTime",
LANCE_READER_PROFILE, 1);
+ }
+ SCOPED_TIMER(_row_id_fetch_total_time);
RETURN_IF_ERROR(_ensure_dataset_open(range));
std::vector<const char*> columns;
@@ -467,8 +542,13 @@ Status LanceTableReader::read_by_row_ids(const
TFileRangeDesc& range,
columns.emplace_back(nullptr);
ArrowArrayStream stream {};
- if (lance_dataset_take_rows(_dataset, row_ids.data(), row_ids.size(),
columns.data(),
- &stream) != 0) {
+ int32_t take_rows_status = 0;
+ {
+ SCOPED_TIMER(_row_id_take_read_time);
+ take_rows_status = lance_dataset_take_rows(_dataset, row_ids.data(),
row_ids.size(),
+ columns.data(), &stream);
+ }
+ if (take_rows_status != 0) {
if (stream.release != nullptr) {
stream.release(&stream);
}
@@ -487,7 +567,12 @@ Status LanceTableReader::read_by_row_ids(const
TFileRangeDesc& range,
auto batch_reader = std::move(imported_reader).ValueUnsafe();
while (true) {
std::shared_ptr<arrow::RecordBatch> record_batch;
- const auto read_status = batch_reader->ReadNext(&record_batch);
+ arrow::Status read_status;
+ {
+ // Lance may materialize take_rows lazily while its Arrow stream
is consumed.
+ SCOPED_TIMER(_row_id_take_read_time);
+ read_status = batch_reader->ReadNext(&record_batch);
+ }
if (!read_status.ok()) {
return Status::InternalError("read Lance take-rows batch failed:
{}",
read_status.message());
@@ -496,7 +581,10 @@ Status LanceTableReader::read_by_row_ids(const
TFileRangeDesc& range,
break;
}
size_t rows = 0;
- RETURN_IF_ERROR(_fill_block_from_record_batch(record_batch, block,
&rows));
+ {
+ SCOPED_TIMER(_arrow_to_doris_block_time);
+ RETURN_IF_ERROR(_fill_block_from_record_batch(record_batch, block,
&rows));
+ }
fetched_rows += rows;
}
if (fetched_rows != row_ids.size()) {
@@ -650,9 +738,12 @@ Status LanceTableReader::_open_dataset(const DatasetKey&
key) {
}
storage_option_ptrs.emplace_back(nullptr);
- _dataset = lance_dataset_open(
- key.uri.c_str(), key.storage_options.empty() ? nullptr :
storage_option_ptrs.data(),
- static_cast<uint64_t>(key.version));
+ {
+ SCOPED_TIMER(_dataset_open_time);
+ _dataset = lance_dataset_open(
+ key.uri.c_str(), key.storage_options.empty() ? nullptr :
storage_option_ptrs.data(),
+ static_cast<uint64_t>(key.version));
+ }
if (_dataset == nullptr) {
return _lance_error("open Lance dataset");
}
@@ -660,6 +751,7 @@ Status LanceTableReader::_open_dataset(const DatasetKey&
key) {
}
Status LanceTableReader::_open_scanner(const TFileRangeDesc& range) {
+ SCOPED_TIMER(_scanner_configure_time);
std::vector<const char*> columns;
columns.reserve(_projected_columns.size() + 1);
for (size_t idx = 0; idx < _projected_columns.size(); ++idx) {
@@ -691,6 +783,13 @@ Status LanceTableReader::_open_scanner(const
TFileRangeDesc& range) {
return _lance_error("create Lance scanner");
}
std::unique_ptr<LanceScanner, LanceScannerDeleter> scanner_guard(scanner);
+ const auto collect_scan_statistics = [](void* callback_ctx,
+ const LanceScanStatistics*
statistics) {
+ LanceTableReader::_collect_scan_statistics(callback_ctx, statistics);
+ };
+ if (lance_scanner_set_statistics_callback(scanner,
collect_scan_statistics, this) != 0) {
+ return _lance_error("set Lance scanner statistics callback");
+ }
if (_global_rowid_output_idx.has_value() &&
lance_scanner_with_row_id(scanner, true) != 0) {
return _lance_error("enable Lance row id output");
@@ -776,9 +875,16 @@ Status LanceTableReader::_open_scanner(const
TFileRangeDesc& range) {
return _lance_error("enable Lance vector prefilter");
}
RETURN_IF_ERROR(_configure_vector_search(scanner));
- DORIS_CHECK(_fragment_count != nullptr);
- if (lance_params.__isset.fragment_ids) {
- COUNTER_UPDATE(_fragment_count,
static_cast<int64_t>(lance_params.fragment_ids.size()));
+ const int64_t fragment_count =
+ lance_params.__isset.fragment_ids
+ ?
static_cast<int64_t>(lance_params.fragment_ids.size())
+ : 0;
+ if (lance_params.__isset.index_segment_uuids &&
!lance_params.index_segment_uuids.empty()) {
+ COUNTER_UPDATE(_planned_index_segment_count,
+
static_cast<int64_t>(lance_params.index_segment_uuids.size()));
+ COUNTER_UPDATE(_planned_indexed_fragment_count, fragment_count);
+ } else {
+ COUNTER_UPDATE(_planned_flat_search_fragment_count,
fragment_count);
}
}
_scanner = scanner_guard.release();
@@ -904,6 +1010,82 @@ Status
LanceTableReader::_configure_vector_search(LanceScanner* scanner) const {
return Status::OK();
}
+void LanceTableReader::_collect_scan_statistics(void* callback_ctx, const
void* opaque_statistics) {
+ const auto* statistics = static_cast<const
LanceScanStatistics*>(opaque_statistics);
+ if (callback_ctx == nullptr || statistics == nullptr) {
+ LOG(WARNING) << "Lance scan statistics callback received a null
argument";
+ return;
+ }
+
+ auto* reader = static_cast<LanceTableReader*>(callback_ctx);
+ const auto update_counter = [](RuntimeProfile::Counter* counter, uint64_t
value,
+ std::string_view metric_name) {
+ if (counter == nullptr) {
+ return;
+ }
+ if (value >
static_cast<uint64_t>(std::numeric_limits<int64_t>::max())) {
+ LOG(WARNING) << "Ignoring Lance scan metric '" << metric_name <<
"' with value "
+ << value << " because it exceeds INT64_MAX";
+ return;
+ }
+ COUNTER_UPDATE(counter, static_cast<int64_t>(value));
+ };
+
+ update_counter(reader->_execution_iops, statistics->iops, "iops");
+ update_counter(reader->_execution_requests, statistics->requests,
"requests");
+ update_counter(reader->_execution_bytes_read, statistics->bytes_read,
"bytes_read");
+ update_counter(reader->_index_partition_cache_miss_loads,
statistics->index_partitions_loaded,
+ "index_partitions_loaded");
+ update_counter(reader->_index_comparisons, statistics->index_comparisons,
"index_comparisons");
+
+ if (statistics->metrics_len != 0 && statistics->metrics == nullptr) {
+ LOG(WARNING) << "Ignoring malformed Lance scan statistics: metrics is
NULL while "
+ << "metrics_len is " << statistics->metrics_len;
+ return;
+ }
+ for (size_t index = 0; index < statistics->metrics_len; ++index) {
+ const auto& metric = statistics->metrics[index];
+ if (metric.name_len != 0 && metric.name == nullptr) {
+ LOG(WARNING) << "Ignoring malformed Lance scan metric at index "
<< index
+ << ": name is NULL while name_len is " <<
metric.name_len;
+ continue;
+ }
+ const std::string_view name(metric.name == nullptr ? "" : metric.name,
metric.name_len);
+ RuntimeProfile::Counter* counter = nullptr;
+ switch (metric.kind) {
+ case LANCE_SCAN_METRIC_COUNT: {
+ const auto found = reader->_lance_count_metrics.find(name);
+ if (found != reader->_lance_count_metrics.end()) {
+ counter = found->second;
+ }
+ break;
+ }
+ case LANCE_SCAN_METRIC_TIME_NANOSECONDS: {
+ const auto found = reader->_lance_time_metrics.find(name);
+ if (found != reader->_lance_time_metrics.end()) {
+ counter = found->second;
+ } else if (name == "search_time") {
+ // Scalar-index metrics exist only when Lance includes the
corresponding
+ // execution node in this scan plan.
+ counter = ADD_CHILD_TIMER_WITH_LEVEL(reader->_scanner_profile,
+
"LanceScalarIndexQueryTime",
+ LANCE_READER_PROFILE, 1);
+ } else if (name == "serialization_time") {
+ counter = ADD_CHILD_TIMER_WITH_LEVEL(reader->_scanner_profile,
+
"LanceScalarIndexResultSerializationTime",
+ LANCE_READER_PROFILE, 1);
+ }
+ break;
+ }
+ default:
+ break;
+ }
+ if (counter != nullptr) {
+ update_counter(counter, metric.value, name);
+ }
+ }
+}
+
void LanceTableReader::_close_scanner() {
if (_scanner != nullptr) {
lance_scanner_close(_scanner);
diff --git a/be/src/format_v2/table/lance_reader.h
b/be/src/format_v2/table/lance_reader.h
index bda465185cb..892aaf518e5 100644
--- a/be/src/format_v2/table/lance_reader.h
+++ b/be/src/format_v2/table/lance_reader.h
@@ -88,6 +88,9 @@ private:
Status _open_dataset(const DatasetKey& key);
Status _open_scanner(const TFileRangeDesc& range);
Status _configure_vector_search(LanceScanner* scanner) const;
+ // Keep lance-c's anonymous statistics typedef out of this header.
_open_scanner installs the
+ // strongly typed C callback adapter before forwarding the borrowed value
here.
+ static void _collect_scan_statistics(void* callback_ctx, const void*
opaque_statistics);
void _close_scanner();
void _close_dataset();
Status _fill_block_from_lance_batch(LanceBatch* batch, Block* block,
size_t* rows);
@@ -107,7 +110,22 @@ private:
std::optional<size_t> _global_rowid_output_idx;
cctz::time_zone _ctz;
size_t _scanner_batch_size = 0;
- RuntimeProfile::Counter* _fragment_count = nullptr;
+ RuntimeProfile::Counter* _planned_index_segment_count = nullptr;
+ RuntimeProfile::Counter* _planned_indexed_fragment_count = nullptr;
+ RuntimeProfile::Counter* _planned_flat_search_fragment_count = nullptr;
+ RuntimeProfile::Counter* _dataset_open_time = nullptr;
+ RuntimeProfile::Counter* _scanner_configure_time = nullptr;
+ RuntimeProfile::Counter* _scanner_read_time = nullptr;
+ RuntimeProfile::Counter* _arrow_to_doris_block_time = nullptr;
+ RuntimeProfile::Counter* _row_id_take_read_time = nullptr;
+ RuntimeProfile::Counter* _row_id_fetch_total_time = nullptr;
+ RuntimeProfile::Counter* _execution_iops = nullptr;
+ RuntimeProfile::Counter* _execution_requests = nullptr;
+ RuntimeProfile::Counter* _execution_bytes_read = nullptr;
+ RuntimeProfile::Counter* _index_partition_cache_miss_loads = nullptr;
+ RuntimeProfile::Counter* _index_comparisons = nullptr;
+ std::unordered_map<std::string_view, RuntimeProfile::Counter*>
_lance_count_metrics;
+ std::unordered_map<std::string_view, RuntimeProfile::Counter*>
_lance_time_metrics;
bool _vector_search = false;
bool _eof = false;
};
diff --git a/be/test/exec/operator/materialization_shared_state_test.cpp
b/be/test/exec/operator/materialization_shared_state_test.cpp
index ee0b067824c..95ec24d0879 100644
--- a/be/test/exec/operator/materialization_shared_state_test.cpp
+++ b/be/test/exec/operator/materialization_shared_state_test.cpp
@@ -37,6 +37,21 @@ void add_request_row(PRequestBlockDesc* request_block_desc,
uint64_t row_id, uin
request_block_desc->add_file_id(file_id);
}
+void set_lance_fetch_profile(PMultiGetBlockV2* response_block, int64_t scale) {
+ RuntimeProfile profile("ExternalRowIDFetcher");
+ profile.add_info_string("LanceDatasetOpenTime", std::to_string(scale) +
"ns");
+ profile.add_info_string("LanceRowIdTakeReadTime", std::to_string(2 *
scale) + "ns");
+ profile.add_info_string("LanceArrowToDorisBlockTime", std::to_string(3 *
scale) + "ns");
+ profile.add_info_string("LanceRowIdFetchTotalTime", std::to_string(4 *
scale) + "ns");
+ profile.add_info_string("ScannersRunningTime", "0ms");
+ profile.add_info_string("InitReaderAvgTime", "0ms");
+ profile.add_info_string("GetBlockAvgTime", "0ms");
+ profile.add_info_string("FileReadLines", "[]");
+ profile.add_info_string("FileReadBytes", "[]");
+ profile.add_info_string("FileReadTime", "[]");
+ profile.to_proto(response_block->mutable_profile(), 2);
+}
+
} // namespace
class MaterializationSharedStateTest : public testing::Test {
@@ -242,6 +257,7 @@ TEST_F(MaterializationSharedStateTest,
TestMergeMultiResponse) {
auto s = resp_block1.serialize(0, serialized_block,
&uncompressed_size, &compressed_size,
&compress_time, CompressionTypePB::LZ4);
EXPECT_TRUE(s.ok());
+ set_lance_fetch_profile(response_.mutable_blocks(0), 10);
_shared_state->rpc_struct_map[_backend_id1].response =
std::move(response_);
// init the response blocks
@@ -268,6 +284,7 @@ TEST_F(MaterializationSharedStateTest,
TestMergeMultiResponse) {
auto s = resp_block2.serialize(0, serialized_block,
&uncompressed_size, &compressed_size,
&compress_time, CompressionTypePB::LZ4);
EXPECT_TRUE(s.ok());
+ set_lance_fetch_profile(response_.mutable_blocks(0), 1);
_shared_state->rpc_struct_map[_backend_id2].response =
std::move(response_);
}
@@ -294,6 +311,16 @@ TEST_F(MaterializationSharedStateTest,
TestMergeMultiResponse) {
EXPECT_EQ(merged_value_col->get_data_at(1).data,
nullptr); // Second value from BE1, replace by null
EXPECT_EQ(*((int*)merged_value_col->get_data_at(2).data), 200); // Third
value from BE2
+ const auto& backend1_info =
_shared_state->backend_profile_info_string.at(_backend_id1);
+ EXPECT_EQ("10ns, ",
fmt::to_string(backend1_info.at("LanceDatasetOpenTime")));
+ EXPECT_EQ("20ns, ",
fmt::to_string(backend1_info.at("LanceRowIdTakeReadTime")));
+ EXPECT_EQ("30ns, ",
fmt::to_string(backend1_info.at("LanceArrowToDorisBlockTime")));
+ EXPECT_EQ("40ns, ",
fmt::to_string(backend1_info.at("LanceRowIdFetchTotalTime")));
+ const auto& backend2_info =
_shared_state->backend_profile_info_string.at(_backend_id2);
+ EXPECT_EQ("1ns, ",
fmt::to_string(backend2_info.at("LanceDatasetOpenTime")));
+ EXPECT_EQ("2ns, ",
fmt::to_string(backend2_info.at("LanceRowIdTakeReadTime")));
+ EXPECT_EQ("3ns, ",
fmt::to_string(backend2_info.at("LanceArrowToDorisBlockTime")));
+ EXPECT_EQ("4ns, ",
fmt::to_string(backend2_info.at("LanceRowIdFetchTotalTime")));
}
TEST_F(MaterializationSharedStateTest, TestMergeMultiResponseMultiBlocks) {
diff --git a/be/test/format_v2/table/lance_reader_test.cpp
b/be/test/format_v2/table/lance_reader_test.cpp
index a576f1f63a2..9a2e802217e 100644
--- a/be/test/format_v2/table/lance_reader_test.cpp
+++ b/be/test/format_v2/table/lance_reader_test.cpp
@@ -90,6 +90,24 @@ struct LanceFixtureInfo {
std::vector<int64_t> fragment_ids;
};
+void expect_lance_profile_hierarchy(RuntimeProfile* profile,
+ const std::vector<std::string>&
metric_names) {
+ TRuntimeProfileTree tree;
+ profile->to_thrift(&tree, 3);
+ ASSERT_FALSE(tree.nodes.empty());
+ const auto& children = tree.nodes[0].child_counters_map;
+ ASSERT_TRUE(children.contains(RuntimeProfile::ROOT_COUNTER));
+
EXPECT_TRUE(children.at(RuntimeProfile::ROOT_COUNTER).contains("FileScannerV2"));
+ ASSERT_TRUE(children.contains("FileScannerV2"));
+ EXPECT_TRUE(children.at("FileScannerV2").contains("TableReader"));
+ ASSERT_TRUE(children.contains("TableReader"));
+ EXPECT_TRUE(children.at("TableReader").contains("LanceReader"));
+ ASSERT_TRUE(children.contains("LanceReader"));
+ for (const auto& metric_name : metric_names) {
+ EXPECT_TRUE(children.at("LanceReader").contains(metric_name)) <<
metric_name;
+ }
+}
+
Status get_fixture_info(const std::filesystem::path& dataset_uri,
LanceFixtureInfo* info) {
std::unique_ptr<LanceDataset, decltype(&lance_dataset_close)> dataset(
lance_dataset_open(dataset_uri.c_str(), nullptr, 0),
lance_dataset_close);
@@ -344,6 +362,12 @@ TEST(LanceTableReaderVectorSearchTest,
SearchesWholeSnapshotWithOffsetAndDistanc
EXPECT_FLOAT_EQ(1.0F, rows[0].second);
EXPECT_EQ(4, rows[1].first);
EXPECT_FLOAT_EQ(8.25F, rows[1].second);
+ ASSERT_NE(profile.get_info_string("LanceTopK"), nullptr);
+ EXPECT_EQ("2", *profile.get_info_string("LanceTopK"));
+ ASSERT_NE(profile.get_info_string("LanceOffset"), nullptr);
+ EXPECT_EQ("1", *profile.get_info_string("LanceOffset"));
+ ASSERT_NE(profile.get_info_string("LanceTopKPlusOffset"), nullptr);
+ EXPECT_EQ("3", *profile.get_info_string("LanceTopKPlusOffset"));
EXPECT_TRUE(reader.close().ok());
}
@@ -399,8 +423,6 @@ TEST(LanceTableReaderVectorSearchTest,
SearchesMultipleFragmentSplits) {
LanceTableReader reader;
ASSERT_TRUE(init_reader(&reader, columns, &state, &profile,
&scan_params).ok());
- ASSERT_NE(profile.get_info_string("LanceUseIndex"), nullptr);
- EXPECT_EQ(*profile.get_info_string("LanceUseIndex"), "false");
std::vector<int64_t> row_ids;
for (const auto fragment_id : fixture.fragment_ids) {
ASSERT_TRUE(prepare_fixture(&reader, dataset_uri, fixture,
{fragment_id}).ok());
@@ -413,9 +435,49 @@ TEST(LanceTableReaderVectorSearchTest,
SearchesMultipleFragmentSplits) {
}
std::ranges::sort(row_ids);
EXPECT_EQ((std::vector<int64_t> {1, 2, 3, 4}), row_ids);
- ASSERT_NE(profile.get_counter("LanceFragmentCount"), nullptr);
- EXPECT_EQ(profile.get_counter("LanceFragmentCount")->value(),
+ ASSERT_NE(profile.get_counter("LancePlannedIndexSegmentCount"), nullptr);
+ EXPECT_EQ(0,
profile.get_counter("LancePlannedIndexSegmentCount")->value());
+ ASSERT_NE(profile.get_counter("LancePlannedIndexedFragmentCount"),
nullptr);
+ EXPECT_EQ(0,
profile.get_counter("LancePlannedIndexedFragmentCount")->value());
+ ASSERT_NE(profile.get_counter("LancePlannedFlatSearchFragmentCount"),
nullptr);
+
EXPECT_EQ(profile.get_counter("LancePlannedFlatSearchFragmentCount")->value(),
static_cast<int64_t>(fixture.fragment_ids.size()));
+ ASSERT_NE(profile.get_info_string("LanceTopK"), nullptr);
+ EXPECT_EQ("4", *profile.get_info_string("LanceTopK"));
+ ASSERT_NE(profile.get_info_string("LanceOffset"), nullptr);
+ EXPECT_EQ("0", *profile.get_info_string("LanceOffset"));
+ ASSERT_NE(profile.get_info_string("LanceTopKPlusOffset"), nullptr);
+ EXPECT_EQ("4", *profile.get_info_string("LanceTopKPlusOffset"));
+ EXPECT_NE(profile.get_counter("LanceDatasetOpenTime"), nullptr);
+ EXPECT_NE(profile.get_counter("LanceScannerConfigureTime"), nullptr);
+ EXPECT_NE(profile.get_counter("LanceScannerReadTime"), nullptr);
+ EXPECT_NE(profile.get_counter("LanceRowOffsetRangesScanned"), nullptr);
+ EXPECT_NE(profile.get_counter("LanceTaskWaitTime"), nullptr);
+ EXPECT_EQ(profile.get_counter("LanceExecutionIndexCacheMissLoads"),
nullptr);
+ EXPECT_EQ(profile.get_counter("LanceRowIdTakeReadTime"), nullptr);
+ EXPECT_EQ(profile.get_counter("LanceRowIdFetchTotalTime"), nullptr);
+ EXPECT_EQ(profile.get_counter("LanceScalarIndexQueryTime"), nullptr);
+ EXPECT_EQ(profile.get_counter("LanceScalarIndexResultSerializationTime"),
nullptr);
+ expect_lance_profile_hierarchy(&profile, {"LanceDatasetOpenTime",
+ "LanceScannerConfigureTime",
+ "LanceScannerReadTime",
+ "LanceArrowToDorisBlockTime",
+ "LanceExecutionIOOps",
+ "LanceExecutionIORequests",
+ "LanceExecutionIOBytesRead",
+
"LanceIndexPartitionCacheMissLoads",
+ "LanceIndexComparisons",
+ "LanceFragmentsScanned",
+ "LanceRowOffsetRangesScanned",
+ "LanceRowsScanned",
+ "LanceIVFPartitionsRanked",
+ "LanceIVFPartitionsSearched",
+
"LanceVectorIndexSegmentsSearched",
+ "LanceTaskWaitTime",
+ "LanceIVFPartitionRankingTime",
+ "LancePlannedIndexSegmentCount",
+
"LancePlannedIndexedFragmentCount",
+
"LancePlannedFlatSearchFragmentCount"});
EXPECT_TRUE(reader.close().ok());
}
@@ -519,6 +581,13 @@ TEST(LanceTableReaderVectorSearchTest,
ReturnsStableGlobalRowIdsAndFetchesPayloa
EXPECT_EQ("extra", label_values.get_data_at(0).to_string());
EXPECT_EQ("unit-x", label_values.get_data_at(1).to_string());
EXPECT_EQ("extra", label_values.get_data_at(2).to_string());
+ EXPECT_NE(fetch_profile.get_counter("LanceDatasetOpenTime"), nullptr);
+ EXPECT_NE(fetch_profile.get_counter("LanceRowIdTakeReadTime"), nullptr);
+ EXPECT_NE(fetch_profile.get_counter("LanceArrowToDorisBlockTime"),
nullptr);
+ EXPECT_NE(fetch_profile.get_counter("LanceRowIdFetchTotalTime"), nullptr);
+ expect_lance_profile_hierarchy(&fetch_profile,
+ {"LanceDatasetOpenTime",
"LanceRowIdTakeReadTime",
+ "LanceArrowToDorisBlockTime",
"LanceRowIdFetchTotalTime"});
EXPECT_TRUE(payload_reader.close().ok());
}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/datasource/lance/source/LanceScanNode.java
b/fe/fe-core/src/main/java/org/apache/doris/datasource/lance/source/LanceScanNode.java
index 38ab936bd74..394cb6ebf06 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/datasource/lance/source/LanceScanNode.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/datasource/lance/source/LanceScanNode.java
@@ -436,8 +436,6 @@ public class LanceScanNode extends FileQueryScanNode {
TVectorSearchParams vector =
externalSearchRequest.getSearchQuery().getVectorSearch();
result.append(prefix).append("externalSearchType=VECTOR\n");
result.append(prefix).append("lanceVectorColumn=").append(vector.getColumn()).append("\n");
-
result.append(prefix).append("lanceTopK=").append(vector.getTopK()).append("\n");
-
result.append(prefix).append("lanceOffset=").append(vector.getOffset()).append("\n");
result.append(prefix).append("lanceMetric=")
.append(vector.isSetMetric() ?
metricName(vector.getMetric()) : "default")
.append("\n");
diff --git
a/regression-test/data/external_table_p0/lance/test_lance_rest_catalog.out
b/regression-test/data/external_table_p0/lance/test_lance_rest_catalog.out
index cb8003936c2..5dc72192239 100644
--- a/regression-test/data/external_table_p0/lance/test_lance_rest_catalog.out
+++ b/regression-test/data/external_table_p0/lance/test_lance_rest_catalog.out
@@ -17,4 +17,3 @@ all_types_unprefixed
-- !rest_predicate_pushdown --
8
9
-
diff --git
a/regression-test/suites/external_table_p0/lance/test_lance_vector_search.groovy
b/regression-test/suites/external_table_p0/lance/test_lance_vector_search.groovy
index 7312a70bf29..4464b5ad35b 100644
---
a/regression-test/suites/external_table_p0/lance/test_lance_vector_search.groovy
+++
b/regression-test/suites/external_table_p0/lance/test_lance_vector_search.groovy
@@ -124,16 +124,12 @@ suite("test_lance_vector_search", "p0,external") {
FROM `${catalogName}`.`doris`.`vs_ivf_pq_f32`
"""
- // EXPLAIN asserts the logical search parameters Doris sends to the
backend. It does not
- // prove which physical Lance index served the query; that proof lives
in the fixture
- // generator's plan self-check plus the nprobes=1 discriminator below,
until lance-c
- // exposes the selected index at runtime.
+ // EXPLAIN reports the searched vector column, distance metric, fixed
snapshot and Doris
+ // physical split plan, while lance-c does not expose its selected
index at runtime.
explain {
sql("""SELECT row_id, label, _distance FROM ${indexedTopFive}
ORDER BY _distance, row_id""")
contains "externalSearchType=VECTOR"
contains "lanceVectorColumn=embedding"
- contains "lanceTopK=5"
- contains "lanceOffset=0"
contains "lanceMetric=l2"
contains "lanceVersion="
contains "lanceSearchFragments=2"
diff --git
a/regression-test/suites/external_table_p0/lance/test_lance_vector_search_two_phase.groovy
b/regression-test/suites/external_table_p0/lance/test_lance_vector_search_two_phase.groovy
index ad916a8d3bc..d7dde8053a8 100644
---
a/regression-test/suites/external_table_p0/lance/test_lance_vector_search_two_phase.groovy
+++
b/regression-test/suites/external_table_p0/lance/test_lance_vector_search_two_phase.groovy
@@ -83,8 +83,6 @@ suite("test_lance_vector_search_two_phase", "p0,external") {
assertTrue(explainString.contains("limit: 5"))
assertTrue(explainString.contains("offset: 1"))
assertTrue(explainString.contains("externalSearchType=VECTOR"))
- assertTrue(explainString.contains("lanceTopK=5"))
- assertTrue(explainString.contains("lanceOffset=1"))
assertTrue(explainString.contains("lanceSearchFragments=2"))
assertTrue(explainString.contains("lanceSearchUnindexedFragments=0"))
assertTrue(explainString.contains("lanceSearchIndexSegments=1"))
diff --git a/thirdparty/download-thirdparty.sh
b/thirdparty/download-thirdparty.sh
index 7fa675da700..bb9c5928809 100755
--- a/thirdparty/download-thirdparty.sh
+++ b/thirdparty/download-thirdparty.sh
@@ -774,12 +774,12 @@ if [[ " ${TP_ARCHIVES[*]} " =~ " PAIMON_CPP " ]]; then
echo "Finished patching ${PAIMON_CPP_SOURCE}"
fi
-# Patch lance-c for fragment-scoped nearest-neighbor search and row-ID-based
fetching.
+# Patch lance-c with the scan execution statistics API from upstream PR #64.
if [[ " ${TP_ARCHIVES[*]} " =~ " LANCE_C " ]]; then
- if [[ "${LANCE_C_SOURCE}" == "lance-c-0.1.6" ]]; then
+ if [[ "${LANCE_C_SOURCE}" == "lance-c-0.1.7" ]]; then
cd "${TP_SOURCE_DIR}/${LANCE_C_SOURCE}"
if [[ ! -f "${PATCHED_MARK}" ]]; then
- patch -p1 <"${TP_PATCH_DIR}/lance-c-0.1.6-doris.patch"
+ patch -p1 <"${TP_PATCH_DIR}/lance-c-0.1.7-pr-64.patch"
touch "${PATCHED_MARK}"
fi
cd -
diff --git a/thirdparty/patches/lance-c-0.1.6-doris.patch
b/thirdparty/patches/lance-c-0.1.6-doris.patch
deleted file mode 100644
index cc49293d9de..00000000000
--- a/thirdparty/patches/lance-c-0.1.6-doris.patch
+++ /dev/null
@@ -1,632 +0,0 @@
-diff --git a/include/lance/lance.h b/include/lance/lance.h
-index 1de72fa..a0a8b5e 100644
---- a/include/lance/lance.h
-+++ b/include/lance/lance.h
-@@ -752,0 +753,25 @@ int32_t lance_dataset_take(
-+/**
-+ * Take rows by dataset row IDs.
-+ *
-+ * Row IDs are values from the `_rowid` scanner column, not zero-based row
-+ * offsets. They must belong to the same dataset snapshot used for this read.
-+ * Missing or deleted row IDs may be omitted from the result. For found rows,
-+ * input order and duplicates are preserved.
-+ *
-+ * @param dataset Open dataset snapshot.
-+ * @param row_ids Array of dataset row IDs. May be NULL only when
-+ * `num_row_ids` is zero.
-+ * @param num_row_ids Length of `row_ids`.
-+ * @param columns NULL-terminated column names, or NULL for all. The
-+ * system column `_rowid` may be requested explicitly.
-+ * @param out Pointer to caller-allocated ArrowArrayStream.
-+ * @return 0 on success, -1 on error.
-+ */
-+int32_t lance_dataset_take_rows(
-+ const LanceDataset* dataset,
-+ const uint64_t* row_ids,
-+ size_t num_row_ids,
-+ const char* const* columns,
-+ struct ArrowArrayStream* out
-+);
-+
-diff --git a/include/lance/lance.hpp b/include/lance/lance.hpp
-index 40aa9e3..d54dc5a 100644
---- a/include/lance/lance.hpp
-+++ b/include/lance/lance.hpp
-@@ -654,0 +655,24 @@ public:
-+ /// Take rows by dataset row IDs. Results exported as ArrowArrayStream.
-+ void take_rows(const uint64_t* row_ids, size_t num_row_ids,
-+ const std::vector<std::string>& columns,
-+ ArrowArrayStream* out) const {
-+ std::vector<const char*> col_ptrs;
-+ for (auto& c : columns) col_ptrs.push_back(c.c_str());
-+ col_ptrs.push_back(nullptr);
-+ const char* const* cols_ptr = columns.empty() ? nullptr :
col_ptrs.data();
-+
-+ if (lance_dataset_take_rows(
-+ handle_.get(), row_ids, num_row_ids, cols_ptr, out) != 0) {
-+ check_error();
-+ }
-+ }
-+
-+ /// Take all columns by dataset row IDs.
-+ void take_rows(const uint64_t* row_ids, size_t num_row_ids,
-+ ArrowArrayStream* out) const {
-+ if (lance_dataset_take_rows(
-+ handle_.get(), row_ids, num_row_ids, nullptr, out) != 0) {
-+ check_error();
-+ }
-+ }
-+
-diff --git a/src/dataset.rs b/src/dataset.rs
-index 91a735f..364be09 100644
---- a/src/dataset.rs
-+++ b/src/dataset.rs
-@@ -44,0 +45,20 @@ impl LanceDataset {
-+fn projection_from_columns(
-+ dataset: &Dataset,
-+ columns: Option<&[String]>,
-+) -> Result<lance::dataset::ProjectionRequest> {
-+ match columns {
-+ Some(columns) => {
-+ let schema = dataset
-+ .schema()
-+ .project_preserve_system_columns(columns)
-+ .map_err(|err| {
-+ lance_core::Error::invalid_input(format!("invalid columns
{columns:?}: {err}"))
-+ })?;
-+ Ok(lance::dataset::ProjectionRequest::from_schema(schema))
-+ }
-+ None => Ok(lance::dataset::ProjectionRequest::from_schema(
-+ dataset.schema().clone(),
-+ )),
-+ }
-+}
-+
-@@ -247,4 +267 @@ unsafe fn dataset_take_inner(
-- let projection = match &col_names {
-- Some(cols) =>
lance::dataset::ProjectionRequest::from_columns(cols.iter(), snap.schema()),
-- None =>
lance::dataset::ProjectionRequest::from_schema(snap.schema().clone()),
-- };
-+ let projection = projection_from_columns(&snap, col_names.as_deref())?;
-@@ -263,0 +281,69 @@ unsafe fn dataset_take_inner(
-+/// Take rows by dataset row IDs, returning results as an ArrowArrayStream.
-+///
-+/// - `row_ids`: array of dataset row IDs, such as values returned in the
-+/// `_rowid` scanner column
-+/// - `num_row_ids`: length of the row ID array
-+/// - `columns`: NULL-terminated column name array, or NULL for all columns
-+/// - `out`: pointer to a stack-allocated `ArrowArrayStream`
-+///
-+/// `row_ids` may be NULL only when `num_row_ids` is zero. Row IDs must belong
-+/// to the same dataset snapshot used for this read. Missing or deleted row
IDs
-+/// may be omitted from the result by the upstream Lance implementation.
-+///
-+/// Returns 0 on success, -1 on error.
-+#[unsafe(no_mangle)]
-+pub unsafe extern "C" fn lance_dataset_take_rows(
-+ dataset: *const LanceDataset,
-+ row_ids: *const u64,
-+ num_row_ids: usize,
-+ columns: *const *const c_char,
-+ out: *mut FFI_ArrowArrayStream,
-+) -> i32 {
-+ ffi_try!(
-+ unsafe { dataset_take_rows_inner(dataset, row_ids, num_row_ids,
columns, out) },
-+ neg
-+ )
-+}
-+
-+unsafe fn dataset_take_rows_inner(
-+ dataset: *const LanceDataset,
-+ row_ids: *const u64,
-+ num_row_ids: usize,
-+ columns: *const *const c_char,
-+ out: *mut FFI_ArrowArrayStream,
-+) -> Result<i32> {
-+ if dataset.is_null() {
-+ return Err(lance_core::Error::invalid_input("dataset must not be
NULL"));
-+ }
-+ if out.is_null() {
-+ return Err(lance_core::Error::invalid_input("out must not be NULL"));
-+ }
-+ if num_row_ids > 0 && row_ids.is_null() {
-+ return Err(lance_core::Error::invalid_input(format!(
-+ "row_ids must not be NULL when num_row_ids = {num_row_ids}"
-+ )));
-+ }
-+
-+ let ds = unsafe { &*dataset };
-+ let row_id_slice = if num_row_ids == 0 {
-+ &[]
-+ } else {
-+ unsafe { std::slice::from_raw_parts(row_ids, num_row_ids) }
-+ };
-+ let col_names = unsafe { helpers::parse_c_string_array(columns)? };
-+
-+ let snap = ds.snapshot();
-+ let projection = projection_from_columns(&snap, col_names.as_deref())?;
-+
-+ let batch = block_on(snap.take_rows(row_id_slice, projection))?;
-+
-+ // Match lance_dataset_take: export the single RecordBatch as an Arrow
stream.
-+ let schema = batch.schema();
-+ let reader =
arrow::record_batch::RecordBatchIterator::new(vec![Ok(batch)], schema);
-+ let ffi_stream = FFI_ArrowArrayStream::new(Box::new(reader));
-+ unsafe {
-+ std::ptr::write_unaligned(out, ffi_stream);
-+ }
-+ Ok(0)
-+}
-+
-diff --git a/src/scanner.rs b/src/scanner.rs
-index c1c2f67..de646be 100644
---- a/src/scanner.rs
-+++ b/src/scanner.rs
-@@ -160,0 +161,6 @@ impl LanceScanner {
-+ // Lance validates fragment-scoped nearest searches when nearest() is
-+ // configured. Such searches are supported when the fragment scan is
-+ // the input to a prefilter, so this flag must be set first.
-+ if self.prefilter {
-+ scanner.prefilter(true);
-+ }
-@@ -178,3 +183,0 @@ impl LanceScanner {
-- if self.prefilter {
-- scanner.prefilter(true);
-- }
-@@ -221,0 +225,5 @@ impl LanceScanner {
-+ // nearest() checks the current prefilter setting before accepting a
-+ // fragment-scoped search. Enable it before installing the query.
-+ if self.prefilter {
-+ scanner.prefilter(true);
-+ }
-@@ -239,3 +246,0 @@ impl LanceScanner {
-- if self.prefilter {
-- scanner.prefilter(true);
-- }
-diff --git a/tests/c_api_test.rs b/tests/c_api_test.rs
-index 2babda6..9ee5af9 100644
---- a/tests/c_api_test.rs
-+++ b/tests/c_api_test.rs
-@@ -9,0 +10 @@ use std::ffi::{CString, c_char};
-+use std::process::Command;
-@@ -18 +19 @@ use arrow::record_batch::RecordBatchReader;
--use arrow_array::{Array, Float32Array, Int32Array, RecordBatch, StringArray};
-+use arrow_array::{Array, Float32Array, Int32Array, RecordBatch, StringArray,
UInt64Array};
-@@ -393,0 +395,94 @@ fn test_dataset_take() {
-+#[test]
-+fn test_dataset_take_rows_empty_and_null_validation() {
-+ let (_tmp, uri) = create_test_dataset();
-+ let c_uri = c_str(&uri);
-+ let ds = unsafe { lance_dataset_open(c_uri.as_ptr(), ptr::null(), 0) };
-+ assert!(!ds.is_null());
-+
-+ let mut empty_stream = FFI_ArrowArrayStream::empty();
-+ assert_eq!(
-+ unsafe { lance_dataset_take_rows(ds, ptr::null(), 0, ptr::null(),
&mut empty_stream) },
-+ 0
-+ );
-+ let reader = unsafe { ArrowArrayStreamReader::from_raw(&mut empty_stream)
}.unwrap();
-+ assert_eq!(
-+ reader.map(|batch| batch.unwrap().num_rows()).sum::<usize>(),
-+ 0
-+ );
-+
-+ let mut invalid_stream = FFI_ArrowArrayStream::empty();
-+ assert_eq!(
-+ unsafe { lance_dataset_take_rows(ds, ptr::null(), 1, ptr::null(),
&mut invalid_stream) },
-+ -1
-+ );
-+ let message = unsafe {
std::ffi::CStr::from_ptr(lance_last_error_message()) }.to_string_lossy();
-+ assert!(
-+ message.contains("row_ids must not be NULL when num_row_ids = 1"),
-+ "unexpected error: {message}"
-+ );
-+
-+ let row_id = 0_u64;
-+ assert_eq!(
-+ unsafe {
-+ lance_dataset_take_rows(ptr::null(), &row_id, 1, ptr::null(),
&mut invalid_stream)
-+ },
-+ -1
-+ );
-+ assert_eq!(
-+ unsafe { lance_dataset_take_rows(ds, &row_id, 1, ptr::null(),
ptr::null_mut()) },
-+ -1
-+ );
-+
-+ unsafe { lance_dataset_close(ds) };
-+}
-+
-+#[test]
-+fn test_dataset_take_rows_invalid_column() {
-+ const CHILD_ENV: &str = "LANCE_C_TEST_TAKE_ROWS_INVALID_COLUMN_CHILD";
-+
-+ if std::env::var_os(CHILD_ENV).is_some() {
-+ let (_tmp, uri) = create_test_dataset();
-+ let c_uri = c_str(&uri);
-+ let ds = unsafe { lance_dataset_open(c_uri.as_ptr(), ptr::null(), 0)
};
-+ assert!(!ds.is_null());
-+
-+ let row_id = 0_u64;
-+ let invalid_column = c_str("unknown_column");
-+ let columns = [invalid_column.as_ptr(), ptr::null()];
-+ let mut stream = FFI_ArrowArrayStream::empty();
-+ assert_eq!(
-+ unsafe { lance_dataset_take_rows(ds, &row_id, 1,
columns.as_ptr(), &mut stream) },
-+ -1
-+ );
-+ assert_eq!(lance_last_error_code(), LanceErrorCode::InvalidArgument);
-+
-+ let message_ptr = lance_last_error_message();
-+ assert!(!message_ptr.is_null());
-+ let message = unsafe { std::ffi::CStr::from_ptr(message_ptr) }
-+ .to_string_lossy()
-+ .into_owned();
-+ unsafe { lance_free_string(message_ptr) };
-+ assert!(
-+ message.contains("unknown_column"),
-+ "unexpected error: {message}"
-+ );
-+
-+ unsafe { lance_dataset_close(ds) };
-+ return;
-+ }
-+
-+ let output = Command::new(std::env::current_exe().unwrap())
-+ .arg("--exact")
-+ .arg("test_dataset_take_rows_invalid_column")
-+ .arg("--nocapture")
-+ .env(CHILD_ENV, "1")
-+ .output()
-+ .unwrap();
-+ assert!(
-+ output.status.success(),
-+ "invalid-column subprocess failed\nstdout:\n{}\nstderr:\n{}",
-+ String::from_utf8_lossy(&output.stdout),
-+ String::from_utf8_lossy(&output.stderr)
-+ );
-+}
-+
-@@ -2497,0 +2593,73 @@ fn create_vector_dataset(num_rows: i32, dim: i32) ->
(tempfile::TempDir, String)
-+/// Create two vector fragments with deterministic vectors. Every component of
-+/// row `id` is `id as f32`, making nearest-neighbor expectations unambiguous.
-+fn create_multi_fragment_vector_dataset(
-+ rows_per_fragment: i32,
-+ dim: i32,
-+ enable_stable_row_ids: bool,
-+) -> (tempfile::TempDir, String) {
-+ use arrow_array::builder::{FixedSizeListBuilder, Float32Builder};
-+
-+ let tmp = tempfile::tempdir().unwrap();
-+ let uri = tmp
-+ .path()
-+ .join("multi_fragment_vec_ds")
-+ .to_str()
-+ .unwrap()
-+ .to_string();
-+ let schema = Arc::new(Schema::new(vec![
-+ Field::new("id", DataType::Int32, false),
-+ Field::new(
-+ "embedding",
-+ DataType::FixedSizeList(Arc::new(Field::new("item",
DataType::Float32, true)), dim),
-+ false,
-+ ),
-+ ]));
-+
-+ let make_batch = |first_id: i32| {
-+ let ids: Vec<i32> = (first_id..first_id +
rows_per_fragment).collect();
-+ let mut embeddings = FixedSizeListBuilder::new(Float32Builder::new(),
dim);
-+ for id in &ids {
-+ for _ in 0..dim {
-+ embeddings.values().append_value(*id as f32);
-+ }
-+ embeddings.append(true);
-+ }
-+ RecordBatch::try_new(
-+ schema.clone(),
-+ vec![
-+ Arc::new(Int32Array::from(ids)),
-+ Arc::new(embeddings.finish()),
-+ ],
-+ )
-+ .unwrap()
-+ };
-+
-+ let first = make_batch(0);
-+ let second = make_batch(rows_per_fragment);
-+ lance_c::runtime::block_on(async {
-+ Dataset::write(
-+ arrow::record_batch::RecordBatchIterator::new(vec![Ok(first)],
schema.clone()),
-+ &uri,
-+ Some(lance::dataset::WriteParams {
-+ enable_stable_row_ids,
-+ ..Default::default()
-+ }),
-+ )
-+ .await
-+ .unwrap();
-+ Dataset::write(
-+ arrow::record_batch::RecordBatchIterator::new(vec![Ok(second)],
schema),
-+ &uri,
-+ Some(lance::dataset::WriteParams {
-+ mode: lance::dataset::WriteMode::Append,
-+ enable_stable_row_ids,
-+ ..Default::default()
-+ }),
-+ )
-+ .await
-+ .unwrap();
-+ });
-+
-+ (tmp, uri)
-+}
-+
-@@ -2723,0 +2892,99 @@ fn test_scanner_nearest_brute_force() {
-+fn
assert_dataset_take_rows_from_multi_fragment_ann_result(enable_stable_row_ids:
bool) {
-+ let (_tmp, uri) = create_multi_fragment_vector_dataset(32, 8,
enable_stable_row_ids);
-+ let uri_c = c_str(&uri);
-+ let ds = unsafe { lance_dataset_open(uri_c.as_ptr(), ptr::null(), 0) };
-+ assert!(!ds.is_null());
-+
-+ // A non-NULL array whose first element is NULL is an explicit empty
-+ // projection. The ANN result should therefore contain only _distance and
-+ // the explicitly requested _rowid system column.
-+ let no_columns: [*const c_char; 1] = [ptr::null()];
-+ let scanner = unsafe { lance_scanner_new(ds, no_columns.as_ptr(),
ptr::null()) };
-+ assert!(!scanner.is_null());
-+ assert_eq!(unsafe { lance_scanner_with_row_id(scanner, true) }, 0);
-+
-+ let column = c_str("embedding");
-+ let query = [40.0_f32; 8];
-+ assert_eq!(
-+ unsafe {
-+ lance_scanner_nearest(
-+ scanner,
-+ column.as_ptr(),
-+ query.as_ptr().cast(),
-+ query.len(),
-+ LanceDataType::Float32 as i32,
-+ 1,
-+ )
-+ },
-+ 0
-+ );
-+ assert_eq!(unsafe { lance_scanner_set_use_index(scanner, false) }, 0);
-+
-+ let mut ann_stream = FFI_ArrowArrayStream::empty();
-+ assert_eq!(
-+ unsafe { lance_scanner_to_arrow_stream(scanner, &mut ann_stream) },
-+ 0
-+ );
-+ let reader = unsafe { ArrowArrayStreamReader::from_raw(&mut ann_stream)
}.unwrap();
-+ let ann_batches = reader.map(|batch| batch.unwrap()).collect::<Vec<_>>();
-+ assert_eq!(
-+ ann_batches.iter().map(RecordBatch::num_rows).sum::<usize>(),
-+ 1
-+ );
-+ assert_eq!(ann_batches[0].num_columns(), 2);
-+
-+ let distance = ann_batches[0]
-+ .column_by_name("_distance")
-+ .unwrap()
-+ .as_any()
-+ .downcast_ref::<Float32Array>()
-+ .unwrap()
-+ .value(0);
-+ assert_eq!(distance, 0.0);
-+ let row_id = ann_batches[0]
-+ .column_by_name("_rowid")
-+ .unwrap()
-+ .as_any()
-+ .downcast_ref::<UInt64Array>()
-+ .unwrap()
-+ .value(0);
-+ if !enable_stable_row_ids {
-+ assert_ne!(
-+ row_id >> 32,
-+ 0,
-+ "expected an address-style row ID from the second fragment"
-+ );
-+ }
-+
-+ let id_column = c_str("id");
-+ let columns = [id_column.as_ptr(), ptr::null()];
-+ let mut take_stream = FFI_ArrowArrayStream::empty();
-+ assert_eq!(
-+ unsafe { lance_dataset_take_rows(ds, &row_id, 1, columns.as_ptr(),
&mut take_stream) },
-+ 0
-+ );
-+ let reader = unsafe { ArrowArrayStreamReader::from_raw(&mut take_stream)
}.unwrap();
-+ let batches = reader.map(|batch| batch.unwrap()).collect::<Vec<_>>();
-+ assert_eq!(batches.len(), 1);
-+ let ids = batches[0]
-+ .column_by_name("id")
-+ .unwrap()
-+ .as_any()
-+ .downcast_ref::<Int32Array>()
-+ .unwrap();
-+ assert_eq!(ids.values(), &[40]);
-+
-+ unsafe { lance_scanner_close(scanner) };
-+ unsafe { lance_dataset_close(ds) };
-+}
-+
-+#[test]
-+fn test_dataset_take_rows_from_multi_fragment_ann_result() {
-+ assert_dataset_take_rows_from_multi_fragment_ann_result(false);
-+}
-+
-+#[test]
-+fn
test_dataset_take_rows_from_multi_fragment_ann_result_with_stable_row_ids() {
-+ assert_dataset_take_rows_from_multi_fragment_ann_result(true);
-+}
-+
-@@ -2855,0 +3123,132 @@ fn test_scanner_nearest_filter_postfilter() {
-+#[test]
-+fn test_scanner_nearest_prefilter_with_fragment_ids_next() {
-+ let (_tmp, uri) = create_multi_fragment_vector_dataset(32, 8, false);
-+ let uri_c = c_str(&uri);
-+ let ds = unsafe { lance_dataset_open(uri_c.as_ptr(), ptr::null(), 0) };
-+ assert!(!ds.is_null());
-+
-+ let mut fragment_ids = vec![0; unsafe { lance_dataset_fragment_count(ds)
} as usize];
-+ assert_eq!(fragment_ids.len(), 2);
-+ assert_eq!(
-+ unsafe { lance_dataset_fragment_ids(ds, fragment_ids.as_mut_ptr()) },
-+ 0
-+ );
-+
-+ let filter = c_str("id >= 40");
-+ let scanner = unsafe { lance_scanner_new(ds, ptr::null(),
filter.as_ptr()) };
-+ assert_eq!(
-+ unsafe { lance_scanner_set_fragment_ids(scanner,
fragment_ids[1..].as_ptr(), 1) },
-+ 0
-+ );
-+
-+ // Match the Doris call order: nearest is configured before prefilter.
-+ let column = c_str("embedding");
-+ let query = [40.0_f32; 8];
-+ assert_eq!(
-+ unsafe {
-+ lance_scanner_nearest(
-+ scanner,
-+ column.as_ptr(),
-+ query.as_ptr().cast(),
-+ query.len(),
-+ LanceDataType::Float32 as i32,
-+ 5,
-+ )
-+ },
-+ 0
-+ );
-+ assert_eq!(unsafe { lance_scanner_set_prefilter(scanner, true) }, 0);
-+ assert_eq!(unsafe { lance_scanner_set_use_index(scanner, false) }, 0);
-+
-+ let batches = scan_all_rows_from_scanner(scanner);
-+ let mut ids = batches
-+ .iter()
-+ .flat_map(|batch| {
-+ batch
-+ .column_by_name("id")
-+ .unwrap()
-+ .as_any()
-+ .downcast_ref::<Int32Array>()
-+ .unwrap()
-+ .values()
-+ .iter()
-+ .copied()
-+ .collect::<Vec<_>>()
-+ })
-+ .collect::<Vec<_>>();
-+ ids.sort_unstable();
-+ assert_eq!(ids, vec![40, 41, 42, 43, 44]);
-+
-+ unsafe { lance_scanner_close(scanner) };
-+ unsafe { lance_dataset_close(ds) };
-+}
-+
-+#[test]
-+fn test_scanner_nearest_prefilter_with_fragment_ids_arrow_stream() {
-+ let (_tmp, uri) = create_multi_fragment_vector_dataset(32, 8, false);
-+ let uri_c = c_str(&uri);
-+ let ds = unsafe { lance_dataset_open(uri_c.as_ptr(), ptr::null(), 0) };
-+ assert!(!ds.is_null());
-+
-+ let mut fragment_ids = vec![0; unsafe { lance_dataset_fragment_count(ds)
} as usize];
-+ assert_eq!(fragment_ids.len(), 2);
-+ assert_eq!(
-+ unsafe { lance_dataset_fragment_ids(ds, fragment_ids.as_mut_ptr()) },
-+ 0
-+ );
-+
-+ let filter = c_str("id >= 60");
-+ let scanner = unsafe { lance_scanner_new(ds, ptr::null(),
filter.as_ptr()) };
-+ assert_eq!(
-+ unsafe { lance_scanner_set_fragment_ids(scanner,
fragment_ids[1..].as_ptr(), 1) },
-+ 0
-+ );
-+
-+ let column = c_str("embedding");
-+ let query = [60.0_f32; 8];
-+ assert_eq!(
-+ unsafe {
-+ lance_scanner_nearest(
-+ scanner,
-+ column.as_ptr(),
-+ query.as_ptr().cast(),
-+ query.len(),
-+ LanceDataType::Float32 as i32,
-+ 10,
-+ )
-+ },
-+ 0
-+ );
-+ assert_eq!(unsafe { lance_scanner_set_prefilter(scanner, true) }, 0);
-+ assert_eq!(unsafe { lance_scanner_set_use_index(scanner, false) }, 0);
-+
-+ let mut stream = FFI_ArrowArrayStream::empty();
-+ assert_eq!(
-+ unsafe { lance_scanner_to_arrow_stream(scanner, &mut stream) },
-+ 0,
-+ "{}",
-+ unsafe { std::ffi::CStr::from_ptr(lance_last_error_message())
}.to_string_lossy()
-+ );
-+ let reader = unsafe { ArrowArrayStreamReader::from_raw(&mut stream)
}.unwrap();
-+ let mut ids = reader
-+ .flat_map(|batch| {
-+ let batch = batch.unwrap();
-+ batch
-+ .column_by_name("id")
-+ .unwrap()
-+ .as_any()
-+ .downcast_ref::<Int32Array>()
-+ .unwrap()
-+ .values()
-+ .iter()
-+ .copied()
-+ .collect::<Vec<_>>()
-+ })
-+ .collect::<Vec<_>>();
-+ ids.sort_unstable();
-+ assert_eq!(ids, vec![60, 61, 62, 63]);
-+
-+ unsafe { lance_scanner_close(scanner) };
-+ unsafe { lance_dataset_close(ds) };
-+}
-+
-diff --git a/tests/cpp/test_cpp_api.cpp b/tests/cpp/test_cpp_api.cpp
-index 43491f2..444fbdc 100644
---- a/tests/cpp/test_cpp_api.cpp
-+++ b/tests/cpp/test_cpp_api.cpp
-@@ -124,0 +125,29 @@ static void test_dataset_take(const std::string& uri) {
-+static void test_dataset_take_rows(const std::string& uri) {
-+ TEST(test_dataset_take_rows);
-+
-+ auto ds = lance::Dataset::open(uri);
-+
-+ // The smoke fixture has one fragment, so its first row IDs are 0, 1, 2.
-+ uint64_t row_ids[] = {0, 1, 2};
-+ ArrowArrayStream stream;
-+ memset(&stream, 0, sizeof(stream));
-+ ds.take_rows(row_ids, 3, &stream);
-+
-+ uint64_t total = 0;
-+ while (true) {
-+ ArrowArray arr;
-+ memset(&arr, 0, sizeof(arr));
-+ int rc = stream.get_next(&stream, &arr);
-+ assert(rc == 0);
-+ if (!arr.release) break;
-+ total += (uint64_t)arr.length;
-+ arr.release(&arr);
-+ }
-+
-+ assert(total == 3);
-+ printf("rows=%llu... ", (unsigned long long)total);
-+
-+ if (stream.release) stream.release(&stream);
-+ PASS();
-+}
-+
-@@ -695,0 +725 @@ int main(int argc, char** argv) {
-+ test_dataset_take_rows(uri);
diff --git a/thirdparty/patches/lance-c-0.1.7-pr-64.patch
b/thirdparty/patches/lance-c-0.1.7-pr-64.patch
new file mode 100644
index 00000000000..17f91bc7c31
--- /dev/null
+++ b/thirdparty/patches/lance-c-0.1.7-pr-64.patch
@@ -0,0 +1,1522 @@
+From e3320c1e7d5c72234b3b44e9e7e9a93a72fe488c Mon Sep 17 00:00:00 2001
+From: zhangstar333 <[email protected]>
+Date: Mon, 24 Aug 2026 16:37:33 +0800
+Subject: [PATCH 1/3] update
+
+---
+ include/lance/lance.h | 73 ++++++++++++++++
+ include/lance/lance.hpp | 10 +++
+ src/scanner.rs | 179 +++++++++++++++++++++++++++++++++++++-
+ tests/c_api_test.rs | 188 +++++++++++++++++++++++++++++++++++++++-
+ 4 files changed, 448 insertions(+), 2 deletions(-)
+
+diff --git a/include/lance/lance.h b/include/lance/lance.h
+index 986905b..c6c3985 100644
+--- a/include/lance/lance.h
++++ b/include/lance/lance.h
+@@ -863,6 +863,79 @@ int32_t lance_scanner_set_substrait_filter(
+ size_t len
+ );
+
++/** Type of a dynamically named scan metric. */
++typedef enum {
++ LANCE_SCAN_METRIC_COUNT = 0,
++ LANCE_SCAN_METRIC_TIME_NANOSECONDS = 1,
++} LanceScanMetricKind;
++
++/**
++ * Borrowed view of one dynamically named scan metric.
++ *
++ * `name` is not NUL-terminated. `name` and this structure are valid only for
++ * the duration of the LanceScanStatisticsCallback invocation.
++ */
++typedef struct {
++ const char* name;
++ size_t name_len;
++ LanceScanMetricKind kind;
++ uint64_t value;
++} LanceScanMetric;
++
++/**
++ * Borrowed view of the execution statistics for one finalized scan.
++ *
++ * The fixed fields are stable summary metrics. `metrics` contains additional
++ * implementation-specific counters and timings. Those names are not a stable
++ * API and are intended for diagnostics and profiles. Dynamic metrics are
++ * best-effort and may be omitted if they cannot be materialized. `metrics` is
++ * NULL when `metrics_len` is zero.
++ */
++typedef struct {
++ uint64_t iops;
++ uint64_t requests;
++ uint64_t bytes_read;
++ uint64_t indices_loaded;
++ uint64_t index_partitions_loaded;
++ uint64_t index_comparisons;
++ const LanceScanMetric* metrics;
++ size_t metrics_len;
++} LanceScanStatistics;
++
++/**
++ * Receives scan statistics when a stream reaches EOF, fails, or is released.
++ *
++ * The statistics and all nested pointers are borrowed and valid only for the
++ * duration of this call. The callback may run on the thread that consumes or
++ * releases the scan stream and must therefore be thread-safe. It must return
++ * normally without throwing an exception or unwinding, and must not call any
++ * `lance_scanner_*` function with the originating scanner.
++ *
++ * Scan statistics are diagnostic and best-effort. The callback must handle
its
++ * own errors and must not use them to abort or throw across this FFI
boundary.
++ */
++typedef void (*LanceScanStatisticsCallback)(
++ void* callback_ctx,
++ const LanceScanStatistics* statistics
++);
++
++/**
++ * Register the execution-statistics callback for this scanner.
++ *
++ * Must be called before starting the scan; registering after the scan starts
++ * returns an error. `callback` must not be NULL. `callback_ctx` may be NULL.
A
++ * non-NULL `callback_ctx` must remain valid, and `callback` must remain
valid,
++ * until the stream reaches EOF, fails, or is released. Replaces a previously
++ * registered callback.
++ *
++ * @return 0 on success, -1 on error
++ */
++int32_t lance_scanner_set_statistics_callback(
++ LanceScanner* scanner,
++ LanceScanStatisticsCallback callback,
++ void* callback_ctx
++);
++
+ /** Close and free a scanner handle. */
+ void lance_scanner_close(LanceScanner* scanner);
+
+diff --git a/include/lance/lance.hpp b/include/lance/lance.hpp
+index afce358..9440a23 100644
+--- a/include/lance/lance.hpp
++++ b/include/lance/lance.hpp
+@@ -1127,6 +1127,16 @@ class Scanner {
+ return substrait_filter(bytes.data(), bytes.size());
+ }
+
++ /// Register a callback for scan execution statistics before starting the
scan.
++ /// The callback may run on the thread that consumes or releases the
stream. It
++ /// must be thread-safe, must not throw, and must not re-enter the
originating
++ /// scanner. A non-null callback context must outlive the exported stream.
++ Scanner& statistics_callback(LanceScanStatisticsCallback callback, void*
callback_ctx) {
++ if (lance_scanner_set_statistics_callback(handle_.get(), callback,
callback_ctx) != 0)
++ check_error();
++ return *this;
++ }
++
+ /// Restrict the next k-NN query to a subset of vector index segments.
+ /// Pass `len` 16-byte UUIDs concatenated as a single byte buffer
+ /// (total bytes = `len * 16`). Pass len=0 (and any pointer) to clear.
+diff --git a/src/scanner.rs b/src/scanner.rs
+index f44b82f..d95089e 100644
+--- a/src/scanner.rs
++++ b/src/scanner.rs
+@@ -14,7 +14,9 @@ use arrow::ffi_stream::FFI_ArrowArrayStream;
+ use arrow_schema::SchemaRef;
+ use futures::{FutureExt, Stream, StreamExt};
+ use lance::Dataset;
+-use lance::dataset::scanner::DatasetRecordBatchStream;
++use lance::dataset::scanner::{
++ DatasetRecordBatchStream, ExecutionStatsCallback, ExecutionSummaryCounts,
++};
+ use lance_core::Result;
+ use lance_index::scalar::FullTextSearchQuery;
+ use lance_io::stream::RecordBatchStream;
+@@ -69,6 +71,8 @@ pub struct LanceScanner {
+ // the spawned async task can poison the handle from outside this call
+ // frame via `poison_flag()`.
+ poisoned: Arc<AtomicBool>,
++ scan_statistics_callback: Option<ExecutionStatsCallback>,
++ scan_started: AtomicBool,
+ // Materialized on first iteration call
+ stream: Option<Pin<Box<DatasetRecordBatchStream>>>,
+ #[allow(dead_code)]
+@@ -122,6 +126,8 @@ impl LanceScanner {
+ prefilter: false,
+ fts_query: None,
+ poisoned: Arc::new(AtomicBool::new(false)),
++ scan_statistics_callback: None,
++ scan_started: AtomicBool::new(false),
+ stream: None,
+ schema: None,
+ }
+@@ -157,6 +163,7 @@ impl LanceScanner {
+
+ /// Build the underlying Scanner and open a stream.
+ fn materialize_stream(&mut self) -> Result<()> {
++ self.scan_started.store(true, Ordering::Release);
+ let mut scanner = self.dataset.scan();
+ if let Some(cols) = &self.columns {
+ scanner.project(cols)?;
+@@ -212,6 +219,9 @@ impl LanceScanner {
+ if let Some(fts) = &self.fts_query {
+ scanner.full_text_search(fts.clone())?;
+ }
++ if let Some(callback) = &self.scan_statistics_callback {
++ scanner.scan_stats_callback(callback.clone());
++ }
+ let stream = block_on(scanner.try_into_stream())?;
+ self.schema = Some(stream.schema());
+ self.stream = Some(Box::pin(stream));
+@@ -220,6 +230,7 @@ impl LanceScanner {
+
+ /// Build a Scanner (without materializing) and return it.
+ fn build_scanner(&self) -> Result<lance::dataset::scanner::Scanner> {
++ self.scan_started.store(true, Ordering::Release);
+ let mut scanner = self.dataset.scan();
+ if let Some(cols) = &self.columns {
+ scanner.project(cols)?;
+@@ -274,10 +285,122 @@ impl LanceScanner {
+ if let Some(fts) = &self.fts_query {
+ scanner.full_text_search(fts.clone())?;
+ }
++ if let Some(callback) = &self.scan_statistics_callback {
++ scanner.scan_stats_callback(callback.clone());
++ }
+ Ok(scanner)
+ }
+ }
+
++/// Type of a dynamically named scan metric.
++#[repr(i32)]
++#[derive(Clone, Copy, Debug, PartialEq, Eq)]
++pub enum LanceScanMetricKind {
++ /// Monotonically accumulated counter.
++ Count = 0,
++ /// Accumulated duration in nanoseconds.
++ TimeNanoseconds = 1,
++}
++
++/// Borrowed view of one dynamically named scan metric.
++///
++/// `name` is not NUL-terminated. Both `name` and this structure are valid
only
++/// for the duration of the scan statistics callback.
++#[repr(C)]
++#[derive(Clone, Copy, Debug)]
++pub struct LanceScanMetric {
++ pub name: *const c_char,
++ pub name_len: usize,
++ pub kind: LanceScanMetricKind,
++ pub value: u64,
++}
++
++/// Borrowed view of the execution statistics for one completed scan.
++///
++/// The fixed fields are stable summary metrics. `metrics` contains additional
++/// implementation-specific counters and timings and is valid only for the
++/// duration of the callback.
++#[repr(C)]
++#[derive(Clone, Copy, Debug)]
++pub struct LanceScanStatistics {
++ pub iops: u64,
++ pub requests: u64,
++ pub bytes_read: u64,
++ pub indices_loaded: u64,
++ pub index_partitions_loaded: u64,
++ pub index_comparisons: u64,
++ pub metrics: *const LanceScanMetric,
++ pub metrics_len: usize,
++}
++
++/// Callback invoked when a scan stream reaches EOF, fails, or is released.
++///
++/// The callback is an FFI boundary and must return normally without unwinding
++/// or throwing an exception. It must not call back into `lance_scanner_*`
with
++/// the originating scanner.
++pub type LanceScanStatisticsCallback =
++ Option<unsafe extern "C" fn(ctx: *mut c_void, statistics: *const
LanceScanStatistics)>;
++
++struct SendScanStatisticsCallback {
++ callback: unsafe extern "C" fn(*mut c_void, *const LanceScanStatistics),
++ ctx: *mut c_void,
++}
++
++// SAFETY: The C API requires the callback and its context to remain valid and
++// safe to invoke from the thread that consumes or releases the scan stream.
++unsafe impl Send for SendScanStatisticsCallback {}
++unsafe impl Sync for SendScanStatisticsCallback {}
++
++impl SendScanStatisticsCallback {
++ fn invoke(&self, counts: &ExecutionSummaryCounts) {
++ // Dynamic profile metrics are best-effort. Use fallible reservation
so
++ // allocation failure omits them instead of aborting the embedding
process.
++ let mut metrics = Vec::new();
++ if let Some(metrics_len) =
counts.all_counts.len().checked_add(counts.all_times.len())
++ && metrics.try_reserve_exact(metrics_len).is_ok()
++ {
++ metrics.extend(
++ counts
++ .all_counts
++ .iter()
++ .map(|(name, value)| LanceScanMetric {
++ name: name.as_ptr().cast(),
++ name_len: name.len(),
++ kind: LanceScanMetricKind::Count,
++ value: *value as u64,
++ }),
++ );
++ metrics.extend(
++ counts
++ .all_times
++ .iter()
++ .map(|(name, value)| LanceScanMetric {
++ name: name.as_ptr().cast(),
++ name_len: name.len(),
++ kind: LanceScanMetricKind::TimeNanoseconds,
++ value: *value as u64,
++ }),
++ );
++ }
++
++ let statistics = LanceScanStatistics {
++ iops: counts.iops as u64,
++ requests: counts.requests as u64,
++ bytes_read: counts.bytes_read as u64,
++ indices_loaded: counts.indices_loaded as u64,
++ index_partitions_loaded: counts.parts_loaded as u64,
++ index_comparisons: counts.index_comparisons as u64,
++ metrics: if metrics.is_empty() {
++ ptr::null()
++ } else {
++ metrics.as_ptr()
++ },
++ metrics_len: metrics.len(),
++ };
++ unsafe { (self.callback)(self.ctx, &statistics) };
++ }
++}
++
+ // ---------------------------------------------------------------------------
+ // Poison check shared by all `lance_scanner_*` entry points
+ // ---------------------------------------------------------------------------
+@@ -529,6 +652,60 @@ unsafe fn scanner_set_substrait_filter_inner(
+ Ok(0)
+ }
+
++/// Register a callback that receives execution statistics when the scan
stream
++/// reaches EOF, fails, or is released.
++///
++/// The callback and `callback_ctx` must remain valid until the scan stream is
++/// finalized. Metric names and arrays passed to the callback are borrowed and
++/// must be copied if the caller needs to retain them. The callback must be
++/// thread-safe, must return normally without unwinding or throwing an
++/// exception, and must not call `lance_scanner_*` with the originating
scanner.
++#[unsafe(no_mangle)]
++pub unsafe extern "C" fn lance_scanner_set_statistics_callback(
++ scanner: *mut LanceScanner,
++ callback: LanceScanStatisticsCallback,
++ callback_ctx: *mut c_void,
++) -> i32 {
++ scanner_poison_check!(scanner, -1);
++ ffi_try!(
++ unsafe { scanner_set_statistics_callback_inner(scanner, callback,
callback_ctx) },
++ neg
++ )
++}
++
++unsafe fn scanner_set_statistics_callback_inner(
++ scanner: *mut LanceScanner,
++ callback: LanceScanStatisticsCallback,
++ callback_ctx: *mut c_void,
++) -> Result<i32> {
++ if scanner.is_null() {
++ return Err(lance_core::Error::invalid_input_source(
++ "scanner is NULL".into(),
++ ));
++ }
++ let Some(callback) = callback else {
++ return Err(lance_core::Error::invalid_input_source(
++ "statistics callback is NULL".into(),
++ ));
++ };
++
++ let s = unsafe { &mut *scanner };
++ if s.scan_started.load(Ordering::Acquire) {
++ return Err(lance_core::Error::invalid_input_source(
++ "statistics callback must be registered before the scan
starts".into(),
++ ));
++ }
++
++ let callback = SendScanStatisticsCallback {
++ callback,
++ ctx: callback_ctx,
++ };
++ s.scan_statistics_callback = Some(Arc::new(move |counts:
&ExecutionSummaryCounts| {
++ callback.invoke(counts);
++ }));
++ Ok(0)
++}
++
+ /// Close and free a scanner handle.
+ ///
+ /// Best-effort (issue #61): this drops a possibly-live
+diff --git a/tests/c_api_test.rs b/tests/c_api_test.rs
+index d6cd928..db35fea 100644
+--- a/tests/c_api_test.rs
++++ b/tests/c_api_test.rs
+@@ -6,7 +6,7 @@
+ //! These tests call the `extern "C"` functions directly from Rust,
+ //! validating the C API contract without needing a C compiler.
+
+-use std::ffi::{CString, c_char};
++use std::ffi::{CString, c_char, c_void};
+ use std::process::Command;
+ use std::ptr;
+ use std::sync::Arc;
+@@ -99,6 +99,58 @@ fn c_str(s: &str) -> CString {
+ CString::new(s).unwrap()
+ }
+
++#[derive(Default)]
++struct CapturedScanStatistics {
++ calls: usize,
++ iops: u64,
++ requests: u64,
++ bytes_read: u64,
++ indices_loaded: u64,
++ index_partitions_loaded: u64,
++ index_comparisons: u64,
++ metrics: Vec<(String, LanceScanMetricKind, u64)>,
++}
++
++unsafe extern "C" fn capture_scan_statistics(
++ callback_ctx: *mut c_void,
++ statistics: *const LanceScanStatistics,
++) {
++ assert!(!callback_ctx.is_null());
++ assert!(!statistics.is_null());
++ let captured = unsafe { &mut
*callback_ctx.cast::<CapturedScanStatistics>() };
++ let statistics = unsafe { &*statistics };
++ let metrics = if statistics.metrics_len == 0 {
++ &[]
++ } else {
++ assert!(!statistics.metrics.is_null());
++ unsafe { std::slice::from_raw_parts(statistics.metrics,
statistics.metrics_len) }
++ };
++
++ captured.calls += 1;
++ captured.iops = statistics.iops;
++ captured.requests = statistics.requests;
++ captured.bytes_read = statistics.bytes_read;
++ captured.indices_loaded = statistics.indices_loaded;
++ captured.index_partitions_loaded = statistics.index_partitions_loaded;
++ captured.index_comparisons = statistics.index_comparisons;
++ captured.metrics = metrics
++ .iter()
++ .map(|metric| {
++ let name = if metric.name_len == 0 {
++ &[]
++ } else {
++ assert!(!metric.name.is_null());
++ unsafe { std::slice::from_raw_parts(metric.name.cast::<u8>(),
metric.name_len) }
++ };
++ (
++ std::str::from_utf8(name).unwrap().to_owned(),
++ metric.kind,
++ metric.value,
++ )
++ })
++ .collect();
++}
++
+ /// Helper: build a tiny dataset whose `value` column is nullable AND contains
+ /// at least one NULL. Used by tests that need to exercise upstream's
+ /// nullability-tightening pre-scan failure path.
+@@ -284,6 +336,140 @@ fn test_scanner_to_arrow_stream() {
+ unsafe { lance_dataset_close(ds) };
+ }
+
++#[test]
++fn test_scanner_statistics_callback_with_next_multi_fragment() {
++ let (_tmp, uri) = create_multi_fragment_dataset();
++ let c_uri = c_str(&uri);
++ let ds = unsafe { lance_dataset_open(c_uri.as_ptr(), ptr::null(), 0) };
++ assert!(!ds.is_null());
++ assert_eq!(unsafe { lance_dataset_fragment_count(ds) }, 2);
++
++ let scanner = unsafe { lance_scanner_new(ds, ptr::null(), ptr::null()) };
++ assert!(!scanner.is_null());
++ let mut captured = CapturedScanStatistics::default();
++ assert_eq!(
++ unsafe {
++ lance_scanner_set_statistics_callback(
++ scanner,
++ Some(capture_scan_statistics),
++ (&mut captured as *mut CapturedScanStatistics).cast(),
++ )
++ },
++ 0
++ );
++
++ loop {
++ let mut batch = ptr::null_mut();
++ match unsafe { lance_scanner_next(scanner, &mut batch) } {
++ 0 => unsafe { lance_batch_free(batch) },
++ 1 => break,
++ status => panic!("scanner_next returned error: {status}"),
++ }
++ }
++
++ assert_eq!(captured.calls, 1);
++ assert!(captured.bytes_read > 0);
++ assert!(captured.requests > 0);
++ assert!(captured.metrics.iter().all(|(name, _, _)| !name.is_empty()));
++
++ unsafe { lance_scanner_close(scanner) };
++ unsafe { lance_dataset_close(ds) };
++}
++
++#[test]
++fn test_scanner_statistics_callback_with_arrow_stream() {
++ let (_tmp, uri) = create_test_dataset();
++ let c_uri = c_str(&uri);
++ let ds = unsafe { lance_dataset_open(c_uri.as_ptr(), ptr::null(), 0) };
++ assert!(!ds.is_null());
++
++ let scanner = unsafe { lance_scanner_new(ds, ptr::null(), ptr::null()) };
++ assert!(!scanner.is_null());
++ let mut captured = CapturedScanStatistics::default();
++ assert_eq!(
++ unsafe {
++ lance_scanner_set_statistics_callback(
++ scanner,
++ Some(capture_scan_statistics),
++ (&mut captured as *mut CapturedScanStatistics).cast(),
++ )
++ },
++ 0
++ );
++
++ let mut stream = FFI_ArrowArrayStream::empty();
++ assert_eq!(unsafe { lance_scanner_to_arrow_stream(scanner, &mut stream)
}, 0);
++ let reader = unsafe { ArrowArrayStreamReader::from_raw(&mut stream)
}.unwrap();
++ assert_eq!(reader.map(|batch| batch.unwrap().num_rows()).sum::<usize>(),
5);
++ assert_eq!(captured.calls, 1);
++ assert!(captured.bytes_read > 0);
++
++ unsafe { lance_scanner_close(scanner) };
++ unsafe { lance_dataset_close(ds) };
++}
++
++#[test]
++fn test_scanner_statistics_callback_rejects_null_inputs() {
++ assert_eq!(
++ unsafe {
++ lance_scanner_set_statistics_callback(
++ ptr::null_mut(),
++ Some(capture_scan_statistics),
++ ptr::null_mut(),
++ )
++ },
++ -1
++ );
++ assert_eq!(lance_last_error_code(), LanceErrorCode::InvalidArgument);
++
++ let (_tmp, uri) = create_test_dataset();
++ let c_uri = c_str(&uri);
++ let ds = unsafe { lance_dataset_open(c_uri.as_ptr(), ptr::null(), 0) };
++ let scanner = unsafe { lance_scanner_new(ds, ptr::null(), ptr::null()) };
++ assert_eq!(
++ unsafe { lance_scanner_set_statistics_callback(scanner, None,
ptr::null_mut()) },
++ -1
++ );
++ assert_eq!(lance_last_error_code(), LanceErrorCode::InvalidArgument);
++
++ unsafe { lance_scanner_close(scanner) };
++ unsafe { lance_dataset_close(ds) };
++}
++
++#[test]
++fn test_scanner_statistics_callback_rejects_registration_after_scan_started()
{
++ let (_tmp, uri) = create_test_dataset();
++ let c_uri = c_str(&uri);
++ let ds = unsafe { lance_dataset_open(c_uri.as_ptr(), ptr::null(), 0) };
++ assert!(!ds.is_null());
++ let scanner = unsafe { lance_scanner_new(ds, ptr::null(), ptr::null()) };
++ assert!(!scanner.is_null());
++
++ let mut batch = ptr::null_mut();
++ assert_eq!(unsafe { lance_scanner_next(scanner, &mut batch) }, 0);
++ assert!(!batch.is_null());
++ unsafe { lance_batch_free(batch) };
++
++ let mut captured = CapturedScanStatistics::default();
++ assert_eq!(
++ unsafe {
++ lance_scanner_set_statistics_callback(
++ scanner,
++ Some(capture_scan_statistics),
++ (&mut captured as *mut CapturedScanStatistics).cast(),
++ )
++ },
++ -1
++ );
++ assert_eq!(lance_last_error_code(), LanceErrorCode::InvalidArgument);
++ let error = take_last_error_message();
++ assert!(error.contains("before the scan starts"), "{error}");
++
++ unsafe { lance_scanner_close(scanner) };
++ assert_eq!(captured.calls, 0);
++ unsafe { lance_dataset_close(ds) };
++}
++
+ #[test]
+ fn test_scanner_with_filter() {
+ let (_tmp, uri) = create_test_dataset();
+
+From 8e6e92140031c06184da079b8f6c194799c0de88 Mon Sep 17 00:00:00 2001
+From: zhangstar333 <[email protected]>
+Date: Mon, 24 Aug 2026 17:04:51 +0800
+Subject: [PATCH 2/3] formatter
+
+---
+ include/lance/lance.h | 23 ++++---
+ include/lance/lance.hpp | 9 +--
+ src/scanner.rs | 23 ++++---
+ tests/c_api_test.rs | 134 +++++++++++++++++++++++++++++++++++++++-
+ 4 files changed, 165 insertions(+), 24 deletions(-)
+
+diff --git a/include/lance/lance.h b/include/lance/lance.h
+index c6c3985..1a6822b 100644
+--- a/include/lance/lance.h
++++ b/include/lance/lance.h
+@@ -883,7 +883,7 @@ typedef struct {
+ } LanceScanMetric;
+
+ /**
+- * Borrowed view of the execution statistics for one finalized scan.
++ * Borrowed view of the execution statistics for one fully consumed scan.
+ *
+ * The fixed fields are stable summary metrics. `metrics` contains additional
+ * implementation-specific counters and timings. Those names are not a stable
+@@ -903,13 +903,13 @@ typedef struct {
+ } LanceScanStatistics;
+
+ /**
+- * Receives scan statistics when a stream reaches EOF, fails, or is released.
++ * Receives scan statistics after a stream is fully consumed to EOF.
+ *
+ * The statistics and all nested pointers are borrowed and valid only for the
+- * duration of this call. The callback may run on the thread that consumes or
+- * releases the scan stream and must therefore be thread-safe. It must return
+- * normally without throwing an exception or unwinding, and must not call any
+- * `lance_scanner_*` function with the originating scanner.
++ * duration of this call. The callback may run on the thread that observes EOF
++ * and must therefore be thread-safe. It must return normally without throwing
++ * an exception or unwinding, and must not call any `lance_scanner_*` function
++ * with the originating scanner.
+ *
+ * Scan statistics are diagnostic and best-effort. The callback must handle
its
+ * own errors and must not use them to abort or throw across this FFI
boundary.
+@@ -925,8 +925,15 @@ typedef void (*LanceScanStatisticsCallback)(
+ * Must be called before starting the scan; registering after the scan starts
+ * returns an error. `callback` must not be NULL. `callback_ctx` may be NULL.
A
+ * non-NULL `callback_ctx` must remain valid, and `callback` must remain
valid,
+- * until the stream reaches EOF, fails, or is released. Replaces a previously
+- * registered callback.
++ * until the callback returns or, if the callback has not run, until the
owning
++ * scan stream is released. For `lance_scanner_next` and
++ * `lance_scanner_poll_next`, the scanner owns the stream. For an exported
++ * ArrowArrayStream, the Arrow stream owns it independently of the scanner.
++ *
++ * The callback is invoked exactly once when the stream is fully consumed to
++ * EOF. It is not guaranteed to run if execution fails, the scan is cancelled,
++ * or the scanner / ArrowArrayStream is released before EOF. Replaces a
++ * previously registered callback.
+ *
+ * @return 0 on success, -1 on error
+ */
+diff --git a/include/lance/lance.hpp b/include/lance/lance.hpp
+index 9440a23..4268801 100644
+--- a/include/lance/lance.hpp
++++ b/include/lance/lance.hpp
+@@ -1127,10 +1127,11 @@ class Scanner {
+ return substrait_filter(bytes.data(), bytes.size());
+ }
+
+- /// Register a callback for scan execution statistics before starting the
scan.
+- /// The callback may run on the thread that consumes or releases the
stream. It
+- /// must be thread-safe, must not throw, and must not re-enter the
originating
+- /// scanner. A non-null callback context must outlive the exported stream.
++ /// Register a callback for scan statistics after successful full
exhaustion.
++ /// The callback is not guaranteed on error, cancellation, or early
release. It
++ /// may run on the thread that observes EOF, must be thread-safe, must
not throw,
++ /// and must not re-enter the originating scanner. The callback and a
non-null
++ /// context must remain valid until the callback returns or the stream is
released.
+ Scanner& statistics_callback(LanceScanStatisticsCallback callback, void*
callback_ctx) {
+ if (lance_scanner_set_statistics_callback(handle_.get(), callback,
callback_ctx) != 0)
+ check_error();
+diff --git a/src/scanner.rs b/src/scanner.rs
+index d95089e..414c269 100644
+--- a/src/scanner.rs
++++ b/src/scanner.rs
+@@ -315,7 +315,7 @@ pub struct LanceScanMetric {
+ pub value: u64,
+ }
+
+-/// Borrowed view of the execution statistics for one completed scan.
++/// Borrowed view of the execution statistics for one fully consumed scan.
+ ///
+ /// The fixed fields are stable summary metrics. `metrics` contains additional
+ /// implementation-specific counters and timings and is valid only for the
+@@ -333,7 +333,7 @@ pub struct LanceScanStatistics {
+ pub metrics_len: usize,
+ }
+
+-/// Callback invoked when a scan stream reaches EOF, fails, or is released.
++/// Callback invoked after a scan stream is fully consumed to EOF.
+ ///
+ /// The callback is an FFI boundary and must return normally without unwinding
+ /// or throwing an exception. It must not call back into `lance_scanner_*`
with
+@@ -347,7 +347,7 @@ struct SendScanStatisticsCallback {
+ }
+
+ // SAFETY: The C API requires the callback and its context to remain valid and
+-// safe to invoke from the thread that consumes or releases the scan stream.
++// safe to invoke from the thread that observes the scan stream's EOF.
+ unsafe impl Send for SendScanStatisticsCallback {}
+ unsafe impl Sync for SendScanStatisticsCallback {}
+
+@@ -652,14 +652,17 @@ unsafe fn scanner_set_substrait_filter_inner(
+ Ok(0)
+ }
+
+-/// Register a callback that receives execution statistics when the scan
stream
+-/// reaches EOF, fails, or is released.
++/// Register a callback that receives execution statistics after the scan
stream
++/// is fully consumed to EOF.
+ ///
+-/// The callback and `callback_ctx` must remain valid until the scan stream is
+-/// finalized. Metric names and arrays passed to the callback are borrowed and
+-/// must be copied if the caller needs to retain them. The callback must be
+-/// thread-safe, must return normally without unwinding or throwing an
+-/// exception, and must not call `lance_scanner_*` with the originating
scanner.
++/// The callback is not guaranteed to run if execution fails, the scan is
++/// cancelled, or the scanner / exported Arrow stream is released before EOF.
++/// The callback and `callback_ctx` must remain valid until the callback
returns
++/// or, if it has not run, until the owning scan stream is released. Metric
names
++/// and arrays passed to the callback are borrowed and must be copied if the
++/// caller needs to retain them. The callback must be thread-safe, must return
++/// normally without unwinding or throwing an exception, and must not call
++/// `lance_scanner_*` with the originating scanner.
+ #[unsafe(no_mangle)]
+ pub unsafe extern "C" fn lance_scanner_set_statistics_callback(
+ scanner: *mut LanceScanner,
+diff --git a/tests/c_api_test.rs b/tests/c_api_test.rs
+index db35fea..c1bb71d 100644
+--- a/tests/c_api_test.rs
++++ b/tests/c_api_test.rs
+@@ -372,7 +372,43 @@ fn
test_scanner_statistics_callback_with_next_multi_fragment() {
+ assert!(captured.requests > 0);
+ assert!(captured.metrics.iter().all(|(name, _, _)| !name.is_empty()));
+
++ let mut batch = ptr::null_mut();
++ assert_eq!(unsafe { lance_scanner_next(scanner, &mut batch) }, 1);
++ assert!(batch.is_null());
++ assert_eq!(captured.calls, 1, "callback must run exactly once");
++
++ unsafe { lance_scanner_close(scanner) };
++ unsafe { lance_dataset_close(ds) };
++}
++
++#[test]
++fn test_scanner_statistics_callback_not_called_on_early_scanner_close() {
++ let (_tmp, uri) = create_test_dataset();
++ let c_uri = c_str(&uri);
++ let ds = unsafe { lance_dataset_open(c_uri.as_ptr(), ptr::null(), 0) };
++ assert!(!ds.is_null());
++
++ let scanner = unsafe { lance_scanner_new(ds, ptr::null(), ptr::null()) };
++ assert!(!scanner.is_null());
++ let mut captured = CapturedScanStatistics::default();
++ assert_eq!(
++ unsafe {
++ lance_scanner_set_statistics_callback(
++ scanner,
++ Some(capture_scan_statistics),
++ (&mut captured as *mut CapturedScanStatistics).cast(),
++ )
++ },
++ 0
++ );
++
++ let mut batch = ptr::null_mut();
++ assert_eq!(unsafe { lance_scanner_next(scanner, &mut batch) }, 0);
++ assert!(!batch.is_null());
++ unsafe { lance_batch_free(batch) };
++
+ unsafe { lance_scanner_close(scanner) };
++ assert_eq!(captured.calls, 0);
+ unsafe { lance_dataset_close(ds) };
+ }
+
+@@ -398,9 +434,15 @@ fn test_scanner_statistics_callback_with_arrow_stream() {
+ );
+
+ let mut stream = FFI_ArrowArrayStream::empty();
+- assert_eq!(unsafe { lance_scanner_to_arrow_stream(scanner, &mut stream)
}, 0);
++ assert_eq!(
++ unsafe { lance_scanner_to_arrow_stream(scanner, &mut stream) },
++ 0
++ );
+ let reader = unsafe { ArrowArrayStreamReader::from_raw(&mut stream)
}.unwrap();
+- assert_eq!(reader.map(|batch| batch.unwrap().num_rows()).sum::<usize>(),
5);
++ assert_eq!(
++ reader.map(|batch| batch.unwrap().num_rows()).sum::<usize>(),
++ 5
++ );
+ assert_eq!(captured.calls, 1);
+ assert!(captured.bytes_read > 0);
+
+@@ -408,6 +450,74 @@ fn test_scanner_statistics_callback_with_arrow_stream() {
+ unsafe { lance_dataset_close(ds) };
+ }
+
++#[test]
++fn
test_scanner_statistics_callback_not_called_on_early_arrow_stream_release() {
++ let (_tmp, uri) = create_test_dataset();
++ let c_uri = c_str(&uri);
++ let ds = unsafe { lance_dataset_open(c_uri.as_ptr(), ptr::null(), 0) };
++ assert!(!ds.is_null());
++
++ let scanner = unsafe { lance_scanner_new(ds, ptr::null(), ptr::null()) };
++ assert!(!scanner.is_null());
++ let mut captured = CapturedScanStatistics::default();
++ assert_eq!(
++ unsafe {
++ lance_scanner_set_statistics_callback(
++ scanner,
++ Some(capture_scan_statistics),
++ (&mut captured as *mut CapturedScanStatistics).cast(),
++ )
++ },
++ 0
++ );
++
++ let mut stream = FFI_ArrowArrayStream::empty();
++ assert_eq!(
++ unsafe { lance_scanner_to_arrow_stream(scanner, &mut stream) },
++ 0
++ );
++ let mut reader = unsafe { ArrowArrayStreamReader::from_raw(&mut stream)
}.unwrap();
++ assert!(reader.next().unwrap().is_ok());
++ drop(reader);
++
++ assert_eq!(captured.calls, 0);
++ unsafe { lance_scanner_close(scanner) };
++ assert_eq!(captured.calls, 0);
++ unsafe { lance_dataset_close(ds) };
++}
++
++#[test]
++fn test_scanner_statistics_callback_not_called_on_materialization_error() {
++ let (_tmp, uri) = create_test_dataset();
++ let c_uri = c_str(&uri);
++ let ds = unsafe { lance_dataset_open(c_uri.as_ptr(), ptr::null(), 0) };
++ assert!(!ds.is_null());
++
++ let bad_filter = c_str("NOT A VALID >>> FILTER ???");
++ let scanner = unsafe { lance_scanner_new(ds, ptr::null(),
bad_filter.as_ptr()) };
++ assert!(!scanner.is_null());
++ let mut captured = CapturedScanStatistics::default();
++ assert_eq!(
++ unsafe {
++ lance_scanner_set_statistics_callback(
++ scanner,
++ Some(capture_scan_statistics),
++ (&mut captured as *mut CapturedScanStatistics).cast(),
++ )
++ },
++ 0
++ );
++
++ let mut batch = ptr::null_mut();
++ assert_eq!(unsafe { lance_scanner_next(scanner, &mut batch) }, -1);
++ assert!(batch.is_null());
++ assert_eq!(captured.calls, 0);
++
++ unsafe { lance_scanner_close(scanner) };
++ assert_eq!(captured.calls, 0);
++ unsafe { lance_dataset_close(ds) };
++}
++
+ #[test]
+ fn test_scanner_statistics_callback_rejects_null_inputs() {
+ assert_eq!(
+@@ -1367,6 +1477,17 @@ fn test_poll_next_basic() {
+ let c_uri = c_str(&uri_clone);
+ let ds = unsafe { lance_dataset_open(c_uri.as_ptr(), ptr::null(), 0)
};
+ let scanner = unsafe { lance_scanner_new(ds, ptr::null(),
ptr::null()) };
++ let mut captured = CapturedScanStatistics::default();
++ assert_eq!(
++ unsafe {
++ lance_scanner_set_statistics_callback(
++ scanner,
++ Some(capture_scan_statistics),
++ (&mut captured as *mut CapturedScanStatistics).cast(),
++ )
++ },
++ 0
++ );
+
+ use std::sync::atomic::{AtomicBool, Ordering};
+ static WOKE: AtomicBool = AtomicBool::new(false);
+@@ -1401,6 +1522,15 @@ fn test_poll_next_basic() {
+ assert!(iterations < 1000, "poll loop should not spin forever");
+ }
+ assert_eq!(total_rows, 5);
++ assert_eq!(captured.calls, 1);
++
++ let mut batch: *mut LanceBatch = ptr::null_mut();
++ assert_eq!(
++ unsafe { lance_scanner_poll_next(scanner, test_waker,
ptr::null_mut(), &mut batch) },
++ LancePollStatus::Finished
++ );
++ assert!(batch.is_null());
++ assert_eq!(captured.calls, 1, "callback must run exactly once");
+
+ unsafe { lance_scanner_close(scanner) };
+ unsafe { lance_dataset_close(ds) };
+
+From fa168ef99951d0396e50c060e540dee93e14e2be Mon Sep 17 00:00:00 2001
+From: zhangstar333 <[email protected]>
+Date: Mon, 24 Aug 2026 20:47:41 +0800
+Subject: [PATCH 3/3] update
+
+---
+ include/lance/lance.h | 55 ++++++++---
+ include/lance/lance.hpp | 22 ++++-
+ src/scanner.rs | 51 +++++++---
+ tests/c_api_test.rs | 188 +++++++++++++++++++++++++++++++------
+ tests/cpp/test_c_api.c | 39 +++++++-
+ tests/cpp/test_cpp_api.cpp | 28 +++++-
+ 6 files changed, 319 insertions(+), 64 deletions(-)
+
+diff --git a/include/lance/lance.h b/include/lance/lance.h
+index 1a6822b..5b12f3d 100644
+--- a/include/lance/lance.h
++++ b/include/lance/lance.h
+@@ -873,7 +873,8 @@ typedef enum {
+ * Borrowed view of one dynamically named scan metric.
+ *
+ * `name` is not NUL-terminated. `name` and this structure are valid only for
+- * the duration of the LanceScanStatisticsCallback invocation.
++ * the duration of the LanceScanStatisticsCallback invocation. Metric order is
++ * unspecified.
+ */
+ typedef struct {
+ const char* name;
+@@ -905,11 +906,22 @@ typedef struct {
+ /**
+ * Receives scan statistics after a stream is fully consumed to EOF.
+ *
+- * The statistics and all nested pointers are borrowed and valid only for the
+- * duration of this call. The callback may run on the thread that observes EOF
+- * and must therefore be thread-safe. It must return normally without throwing
+- * an exception or unwinding, and must not call any `lance_scanner_*` function
+- * with the originating scanner.
++ * `statistics` is non-NULL. It and all nested pointers are borrowed and valid
++ * only for the duration of this call. The callback may run on the thread that
++ * observes EOF and must therefore be thread-safe. It must return normally
++ * without throwing an exception or unwinding, and must not call any
++ * `lance_scanner_*` function with the originating scanner.
++ *
++ * From callback entry until the enclosing operation that observes EOF has
++ * returned to its caller, the callback must not directly or indirectly cause
++ * `get_schema`, `get_next`, `get_last_error`, or `release` to be called on
any
++ * ArrowArrayStream derived from the originating scanner, nor cause such a
++ * stream to be moved, destroyed, or otherwise accessed. This includes
signaling
++ * or scheduling another thread to act based only on callback completion: the
++ * callback returns before the enclosing stream operation does. Such
interaction
++ * is reentrant and has undefined behavior. Normal access may resume only
after
++ * the enclosing ArrowArrayStream `get_next`, `lance_scanner_next`, or
++ * `lance_scanner_poll_next` call returns to its caller.
+ *
+ * Scan statistics are diagnostic and best-effort. The callback must handle
its
+ * own errors and must not use them to abort or throw across this FFI
boundary.
+@@ -925,15 +937,27 @@ typedef void (*LanceScanStatisticsCallback)(
+ * Must be called before starting the scan; registering after the scan starts
+ * returns an error. `callback` must not be NULL. `callback_ctx` may be NULL.
A
+ * non-NULL `callback_ctx` must remain valid, and `callback` must remain
valid,
+- * until the callback returns or, if the callback has not run, until the
owning
+- * scan stream is released. For `lance_scanner_next` and
+- * `lance_scanner_poll_next`, the scanner owns the stream. For an exported
+- * ArrowArrayStream, the Arrow stream owns it independently of the scanner.
+- *
+- * The callback is invoked exactly once when the stream is fully consumed to
+- * EOF. It is not guaranteed to run if execution fails, the scan is cancelled,
+- * or the scanner / ArrowArrayStream is released before EOF. Replaces a
+- * previously registered callback.
++ * until all of the following are true: the scanner is closed, every in-flight
++ * `lance_scanner_scan_async` call has delivered its completion callback, and
++ * every ArrowArrayStream derived from the scanner has been released. The
++ * registration remains installed after a callback returns and applies to
++ * streams created later from the same scanner. For `lance_scanner_next` and
++ * `lance_scanner_poll_next`, the scanner owns the stream. Exported and
++ * asynchronous ArrowArrayStreams own their registrations independently of the
++ * scanner and may invoke the callback after the scanner is closed. Concurrent
++ * streams may invoke the callback concurrently.
++ *
++ * The callback is invoked exactly once for each derived stream that is fully
++ * consumed to EOF. It is not guaranteed to run for a stream if execution
fails,
++ * the scan is cancelled, or the scanner / ArrowArrayStream is released before
++ * EOF. Before scanning starts, a new registration replaces the previous one;
++ * after a successful replacement, the previous callback and context are no
++ * longer retained and may be retired.
++ *
++ * From callback entry until the enclosing EOF-observing operation returns,
the
++ * callback must not directly or indirectly cause interaction with any
++ * ArrowArrayStream derived from this scanner; see LanceScanStatisticsCallback
++ * for the complete reentrancy restriction.
+ *
+ * @return 0 on success, -1 on error
+ */
+@@ -950,6 +974,7 @@ void lance_scanner_close(LanceScanner* scanner);
+
+ /**
+ * Materialize the scan as an ArrowArrayStream (blocking).
++ * The scanner remains valid, and each call creates an independent stream.
+ *
+ * Reading the exported stream may surface a mid-iteration panic as one
+ * error through the Arrow C stream contract (nonzero get_next plus
+diff --git a/include/lance/lance.hpp b/include/lance/lance.hpp
+index 4268801..8aa97e2 100644
+--- a/include/lance/lance.hpp
++++ b/include/lance/lance.hpp
+@@ -1128,10 +1128,22 @@ class Scanner {
+ }
+
+ /// Register a callback for scan statistics after successful full
exhaustion.
+- /// The callback is not guaranteed on error, cancellation, or early
release. It
+- /// may run on the thread that observes EOF, must be thread-safe, must
not throw,
+- /// and must not re-enter the originating scanner. The callback and a
non-null
+- /// context must remain valid until the callback returns or the stream is
released.
++ /// The registration applies to every stream derived from this scanner,
including
++ /// concurrent streams and streams created after an earlier callback
returns. The
++ /// callback is not guaranteed on error, cancellation, or early release.
It may
++ /// run on the thread that observes EOF, must be thread-safe, must not
throw, and
++ /// must not re-enter the originating scanner. The callback and a
non-null context
++ /// must remain valid until the scanner is closed, all async scan
requests have
++ /// delivered their completion callbacks, and all derived streams are
released.
++ /// From callback entry until the enclosing operation that observes EOF
has
++ /// returned to its caller, the callback must not directly or indirectly
cause
++ /// any ArrowArrayStream derived from this Scanner to be accessed, called,
++ /// released, moved, or destroyed. This includes signaling or scheduling
another
++ /// thread to act based only on callback completion: the callback returns
before
++ /// the enclosing stream operation does. Such interaction is reentrant
and has
++ /// undefined behavior. Normal access may resume only after the enclosing
++ /// ArrowArrayStream `get_next`, `lance_scanner_next`, or
++ /// `lance_scanner_poll_next` call returns.
+ Scanner& statistics_callback(LanceScanStatisticsCallback callback, void*
callback_ctx) {
+ if (lance_scanner_set_statistics_callback(handle_.get(), callback,
callback_ctx) != 0)
+ check_error();
+@@ -1153,7 +1165,7 @@ class Scanner {
+ return index_segments(reinterpret_cast<const uint8_t*>(uuids.data()),
uuids.size());
+ }
+
+- /// Materialize the scan as an ArrowArrayStream (blocking).
++ /// Materialize an independent ArrowArrayStream (blocking). The scanner
remains valid.
+ void to_arrow_stream(ArrowArrayStream* out) {
+ if (lance_scanner_to_arrow_stream(handle_.get(), out) != 0)
+ check_error();
+diff --git a/src/scanner.rs b/src/scanner.rs
+index 414c269..ef9d290 100644
+--- a/src/scanner.rs
++++ b/src/scanner.rs
+@@ -305,7 +305,7 @@ pub enum LanceScanMetricKind {
+ /// Borrowed view of one dynamically named scan metric.
+ ///
+ /// `name` is not NUL-terminated. Both `name` and this structure are valid
only
+-/// for the duration of the scan statistics callback.
++/// for the duration of the scan statistics callback. Metric order is
unspecified.
+ #[repr(C)]
+ #[derive(Clone, Copy, Debug)]
+ pub struct LanceScanMetric {
+@@ -318,8 +318,10 @@ pub struct LanceScanMetric {
+ /// Borrowed view of the execution statistics for one fully consumed scan.
+ ///
+ /// The fixed fields are stable summary metrics. `metrics` contains additional
+-/// implementation-specific counters and timings and is valid only for the
+-/// duration of the callback.
++/// implementation-specific counters and timings whose names are not a stable
API
++/// and are intended only for diagnostics and profiles. Dynamic metrics are
++/// best-effort and may be omitted if they cannot be materialized. `metrics`
is
++/// null when `metrics_len` is zero and is valid only for the callback
duration.
+ #[repr(C)]
+ #[derive(Clone, Copy, Debug)]
+ pub struct LanceScanStatistics {
+@@ -333,13 +335,20 @@ pub struct LanceScanStatistics {
+ pub metrics_len: usize,
+ }
+
+-/// Callback invoked after a scan stream is fully consumed to EOF.
++/// Callback invoked once for each derived scan stream that is fully consumed
to EOF.
+ ///
+ /// The callback is an FFI boundary and must return normally without unwinding
+ /// or throwing an exception. It must not call back into `lance_scanner_*`
with
+-/// the originating scanner.
++/// the originating scanner. From callback entry until the enclosing operation
++/// that observes EOF has returned to its caller, the callback must not
directly
++/// or indirectly cause any Arrow C stream derived from the originating
scanner to
++/// be called, released, moved, destroyed, or otherwise accessed. This
includes
++/// signaling or scheduling another thread to act based only on callback
completion:
++/// the callback returns before the enclosing stream operation does. Such
interaction
++/// is reentrant and has undefined behavior. Normal access may resume only
after the
++/// enclosing `get_next`, `lance_scanner_next`, or `lance_scanner_poll_next`
returns.
+ pub type LanceScanStatisticsCallback =
+- Option<unsafe extern "C" fn(ctx: *mut c_void, statistics: *const
LanceScanStatistics)>;
++ Option<unsafe extern "C" fn(callback_ctx: *mut c_void, statistics: *const
LanceScanStatistics)>;
+
+ struct SendScanStatisticsCallback {
+ callback: unsafe extern "C" fn(*mut c_void, *const LanceScanStatistics),
+@@ -347,7 +356,9 @@ struct SendScanStatisticsCallback {
+ }
+
+ // SAFETY: The C API requires the callback and its context to remain valid and
+-// safe to invoke from the thread that observes the scan stream's EOF.
++// safe to invoke until the scanner is closed, every in-flight asynchronous
scan
++// has delivered its completion callback, and every derived stream has been
++// released. Concurrent derived streams may invoke the callback concurrently.
+ unsafe impl Send for SendScanStatisticsCallback {}
+ unsafe impl Sync for SendScanStatisticsCallback {}
+
+@@ -657,12 +668,24 @@ unsafe fn scanner_set_substrait_filter_inner(
+ ///
+ /// The callback is not guaranteed to run if execution fails, the scan is
+ /// cancelled, or the scanner / exported Arrow stream is released before EOF.
+-/// The callback and `callback_ctx` must remain valid until the callback
returns
+-/// or, if it has not run, until the owning scan stream is released. Metric
names
+-/// and arrays passed to the callback are borrowed and must be copied if the
+-/// caller needs to retain them. The callback must be thread-safe, must return
+-/// normally without unwinding or throwing an exception, and must not call
+-/// `lance_scanner_*` with the originating scanner.
++/// The registration applies to every stream derived from this scanner,
including
++/// streams created after an earlier callback has returned. The callback and
++/// `callback_ctx` must remain valid until the scanner is closed, every
in-flight
++/// asynchronous scan has delivered its completion callback, and every derived
++/// stream has been released.
++/// Metric names and arrays passed to the callback are borrowed and must be
copied
++/// if the caller needs to retain them. The callback must be thread-safe, must
++/// return normally without unwinding or throwing an exception, and must not
call
++/// `lance_scanner_*` with the originating scanner. From callback entry until
the
++/// enclosing operation that observes EOF has returned to its caller, the
callback
++/// must not directly or indirectly cause any Arrow C stream derived from that
++/// scanner to be called, released, moved, destroyed, or otherwise accessed.
This
++/// includes signaling or scheduling another thread to act based only on
callback
++/// completion: the callback returns before the enclosing stream operation
does.
++/// Such interaction is reentrant and has undefined behavior. Normal access
may
++/// resume only after the enclosing `get_next`, `lance_scanner_next`, or
++/// `lance_scanner_poll_next` returns. Replacing the registration before
scanning
++/// starts immediately releases the previous registration.
+ #[unsafe(no_mangle)]
+ pub unsafe extern "C" fn lance_scanner_set_statistics_callback(
+ scanner: *mut LanceScanner,
+@@ -732,7 +755,7 @@ pub unsafe extern "C" fn lance_scanner_close(scanner: *mut
LanceScanner) {
+ /// Materialize the scan as an Arrow C Data Interface `ArrowArrayStream`.
+ ///
+ /// This is the preferred API for simple integrations — blocks the calling
thread.
+-/// The scanner is consumed by this call and should not be used afterward
(close it).
++/// The scanner remains valid and may be used to create additional streams.
+ ///
+ /// The exported stream is panic-guarded (issue #61): a panic during export
+ /// poisons the scanner — this call returns -1 with `LANCE_ERR_PANIC`, and
+diff --git a/tests/c_api_test.rs b/tests/c_api_test.rs
+index c1bb71d..daa7425 100644
+--- a/tests/c_api_test.rs
++++ b/tests/c_api_test.rs
+@@ -10,6 +10,7 @@ use std::ffi::{CString, c_char, c_void};
+ use std::process::Command;
+ use std::ptr;
+ use std::sync::Arc;
++use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering as AtomicOrdering};
+
+ use arrow::ffi::from_ffi;
+ use arrow::ffi::{FFI_ArrowArray, FFI_ArrowSchema};
+@@ -151,6 +152,29 @@ unsafe extern "C" fn capture_scan_statistics(
+ .collect();
+ }
+
++#[derive(Default)]
++struct AtomicScanStatisticsCapture {
++ calls: AtomicUsize,
++ invalid_statistics: AtomicBool,
++}
++
++unsafe extern "C" fn capture_scan_statistics_atomically(
++ callback_ctx: *mut c_void,
++ statistics: *const LanceScanStatistics,
++) {
++ if callback_ctx.is_null() {
++ return;
++ }
++ let captured = unsafe {
&*callback_ctx.cast::<AtomicScanStatisticsCapture>() };
++ if statistics.is_null() {
++ captured
++ .invalid_statistics
++ .store(true, AtomicOrdering::SeqCst);
++ return;
++ }
++ captured.calls.fetch_add(1, AtomicOrdering::SeqCst);
++}
++
+ /// Helper: build a tiny dataset whose `value` column is nullable AND contains
+ /// at least one NULL. Used by tests that need to exercise upstream's
+ /// nullability-tightening pre-scan failure path.
+@@ -413,7 +437,7 @@ fn
test_scanner_statistics_callback_not_called_on_early_scanner_close() {
+ }
+
+ #[test]
+-fn test_scanner_statistics_callback_with_arrow_stream() {
++fn test_scanner_statistics_callback_applies_to_reused_scanner() {
+ let (_tmp, uri) = create_test_dataset();
+ let c_uri = c_str(&uri);
+ let ds = unsafe { lance_dataset_open(c_uri.as_ptr(), ptr::null(), 0) };
+@@ -433,20 +457,87 @@ fn test_scanner_statistics_callback_with_arrow_stream() {
+ 0
+ );
+
+- let mut stream = FFI_ArrowArrayStream::empty();
++ let mut first_stream = FFI_ArrowArrayStream::empty();
+ assert_eq!(
+- unsafe { lance_scanner_to_arrow_stream(scanner, &mut stream) },
++ unsafe { lance_scanner_to_arrow_stream(scanner, &mut first_stream) },
+ 0
+ );
+- let reader = unsafe { ArrowArrayStreamReader::from_raw(&mut stream)
}.unwrap();
++ let first_reader = unsafe { ArrowArrayStreamReader::from_raw(&mut
first_stream) }.unwrap();
+ assert_eq!(
+- reader.map(|batch| batch.unwrap().num_rows()).sum::<usize>(),
++ first_reader
++ .map(|batch| batch.unwrap().num_rows())
++ .sum::<usize>(),
+ 5
+ );
+ assert_eq!(captured.calls, 1);
+- assert!(captured.bytes_read > 0);
+
++ let mut second_stream = FFI_ArrowArrayStream::empty();
++ assert_eq!(
++ unsafe { lance_scanner_to_arrow_stream(scanner, &mut second_stream) },
++ 0
++ );
+ unsafe { lance_scanner_close(scanner) };
++
++ let second_reader = unsafe { ArrowArrayStreamReader::from_raw(&mut
second_stream) }.unwrap();
++ assert_eq!(
++ second_reader
++ .map(|batch| batch.unwrap().num_rows())
++ .sum::<usize>(),
++ 5
++ );
++ assert_eq!(captured.calls, 2);
++
++ unsafe { lance_dataset_close(ds) };
++}
++
++#[test]
++fn test_scanner_statistics_callback_supports_concurrent_exported_streams() {
++ struct SendableArrowStream(FFI_ArrowArrayStream);
++ unsafe impl Send for SendableArrowStream {}
++
++ fn consume_stream(mut stream: SendableArrowStream) -> usize {
++ let reader = unsafe { ArrowArrayStreamReader::from_raw(&mut stream.0)
}.unwrap();
++ reader.map(|batch| batch.unwrap().num_rows()).sum()
++ }
++
++ let (_tmp, uri) = create_test_dataset();
++ let c_uri = c_str(&uri);
++ let ds = unsafe { lance_dataset_open(c_uri.as_ptr(), ptr::null(), 0) };
++ assert!(!ds.is_null());
++
++ let scanner = unsafe { lance_scanner_new(ds, ptr::null(), ptr::null()) };
++ assert!(!scanner.is_null());
++ let captured = Arc::new(AtomicScanStatisticsCapture::default());
++ assert_eq!(
++ unsafe {
++ lance_scanner_set_statistics_callback(
++ scanner,
++ Some(capture_scan_statistics_atomically),
++ Arc::as_ptr(&captured).cast_mut().cast(),
++ )
++ },
++ 0
++ );
++
++ let mut first_stream = FFI_ArrowArrayStream::empty();
++ let mut second_stream = FFI_ArrowArrayStream::empty();
++ assert_eq!(
++ unsafe { lance_scanner_to_arrow_stream(scanner, &mut first_stream) },
++ 0
++ );
++ assert_eq!(
++ unsafe { lance_scanner_to_arrow_stream(scanner, &mut second_stream) },
++ 0
++ );
++ unsafe { lance_scanner_close(scanner) };
++
++ let first = std::thread::spawn(move ||
consume_stream(SendableArrowStream(first_stream)));
++ let second = std::thread::spawn(move ||
consume_stream(SendableArrowStream(second_stream)));
++ assert_eq!(first.join().unwrap(), 5);
++ assert_eq!(second.join().unwrap(), 5);
++ assert_eq!(captured.calls.load(AtomicOrdering::SeqCst), 2);
++ assert!(!captured.invalid_statistics.load(AtomicOrdering::SeqCst));
++
+ unsafe { lance_dataset_close(ds) };
+ }
+
+@@ -546,6 +637,55 @@ fn test_scanner_statistics_callback_rejects_null_inputs()
{
+ unsafe { lance_dataset_close(ds) };
+ }
+
++#[test]
++fn test_scanner_statistics_callback_replaces_registration_before_scan() {
++ let (_tmp, uri) = create_test_dataset();
++ let c_uri = c_str(&uri);
++ let ds = unsafe { lance_dataset_open(c_uri.as_ptr(), ptr::null(), 0) };
++ assert!(!ds.is_null());
++ let scanner = unsafe { lance_scanner_new(ds, ptr::null(), ptr::null()) };
++ assert!(!scanner.is_null());
++
++ let mut replaced = CapturedScanStatistics::default();
++ let mut active = CapturedScanStatistics::default();
++ assert_eq!(
++ unsafe {
++ lance_scanner_set_statistics_callback(
++ scanner,
++ Some(capture_scan_statistics),
++ (&mut replaced as *mut CapturedScanStatistics).cast(),
++ )
++ },
++ 0
++ );
++ assert_eq!(
++ unsafe {
++ lance_scanner_set_statistics_callback(
++ scanner,
++ Some(capture_scan_statistics),
++ (&mut active as *mut CapturedScanStatistics).cast(),
++ )
++ },
++ 0
++ );
++
++ let mut stream = FFI_ArrowArrayStream::empty();
++ assert_eq!(
++ unsafe { lance_scanner_to_arrow_stream(scanner, &mut stream) },
++ 0
++ );
++ let reader = unsafe { ArrowArrayStreamReader::from_raw(&mut stream)
}.unwrap();
++ assert_eq!(
++ reader.map(|batch| batch.unwrap().num_rows()).sum::<usize>(),
++ 5
++ );
++ assert_eq!(replaced.calls, 0);
++ assert_eq!(active.calls, 1);
++
++ unsafe { lance_scanner_close(scanner) };
++ unsafe { lance_dataset_close(ds) };
++}
++
+ #[test]
+ fn test_scanner_statistics_callback_rejects_registration_after_scan_started()
{
+ let (_tmp, uri) = create_test_dataset();
+@@ -814,6 +954,17 @@ fn test_scanner_scan_async() {
+
+ let scanner = unsafe { lance_scanner_new(ds, ptr::null(), ptr::null()) };
+ assert!(!scanner.is_null());
++ let captured = Arc::new(AtomicScanStatisticsCapture::default());
++ assert_eq!(
++ unsafe {
++ lance_scanner_set_statistics_callback(
++ scanner,
++ Some(capture_scan_statistics_atomically),
++ Arc::as_ptr(&captured).cast_mut().cast(),
++ )
++ },
++ 0
++ );
+
+ // Synchronization primitive for the async callback.
+ struct CallbackResult {
+@@ -845,6 +996,7 @@ fn test_scanner_scan_async() {
+ on_complete,
+ Arc::as_ptr(&pair_clone) as *mut std::ffi::c_void,
+ );
++ lance_scanner_close(scanner);
+ }
+
+ // Wait for callback.
+@@ -861,8 +1013,9 @@ fn test_scanner_scan_async() {
+ let reader = unsafe { ArrowArrayStreamReader::from_raw(ffi_stream)
}.unwrap();
+ let total_rows: usize = reader.map(|r| r.unwrap().num_rows()).sum();
+ assert_eq!(total_rows, 5);
++ assert_eq!(captured.calls.load(AtomicOrdering::SeqCst), 1);
++ assert!(!captured.invalid_statistics.load(AtomicOrdering::SeqCst));
+
+- unsafe { lance_scanner_close(scanner) };
+ unsafe { lance_dataset_close(ds) };
+ }
+
+@@ -1477,18 +1630,6 @@ fn test_poll_next_basic() {
+ let c_uri = c_str(&uri_clone);
+ let ds = unsafe { lance_dataset_open(c_uri.as_ptr(), ptr::null(), 0)
};
+ let scanner = unsafe { lance_scanner_new(ds, ptr::null(),
ptr::null()) };
+- let mut captured = CapturedScanStatistics::default();
+- assert_eq!(
+- unsafe {
+- lance_scanner_set_statistics_callback(
+- scanner,
+- Some(capture_scan_statistics),
+- (&mut captured as *mut CapturedScanStatistics).cast(),
+- )
+- },
+- 0
+- );
+-
+ use std::sync::atomic::{AtomicBool, Ordering};
+ static WOKE: AtomicBool = AtomicBool::new(false);
+ unsafe extern "C" fn test_waker(_ctx: *mut std::ffi::c_void) {
+@@ -1522,15 +1663,6 @@ fn test_poll_next_basic() {
+ assert!(iterations < 1000, "poll loop should not spin forever");
+ }
+ assert_eq!(total_rows, 5);
+- assert_eq!(captured.calls, 1);
+-
+- let mut batch: *mut LanceBatch = ptr::null_mut();
+- assert_eq!(
+- unsafe { lance_scanner_poll_next(scanner, test_waker,
ptr::null_mut(), &mut batch) },
+- LancePollStatus::Finished
+- );
+- assert!(batch.is_null());
+- assert_eq!(captured.calls, 1, "callback must run exactly once");
+
+ unsafe { lance_scanner_close(scanner) };
+ unsafe { lance_dataset_close(ds) };
+diff --git a/tests/cpp/test_c_api.c b/tests/cpp/test_c_api.c
+index b3b78f0..4499cb9 100644
+--- a/tests/cpp/test_c_api.c
++++ b/tests/cpp/test_c_api.c
+@@ -39,6 +39,36 @@
+ }
\
+ } while (0)
+
++typedef struct {
++ uint64_t calls;
++ uint64_t bytes_read;
++ int invalid;
++} ScanStatisticsCapture;
++
++static void capture_scan_statistics(
++ void *callback_ctx,
++ const LanceScanStatistics *statistics
++) {
++ if (callback_ctx == NULL) return;
++ ScanStatisticsCapture *captured = (ScanStatisticsCapture *)callback_ctx;
++ if (statistics == NULL ||
++ (statistics->metrics_len > 0 && statistics->metrics == NULL)) {
++ captured->invalid = 1;
++ return;
++ }
++ for (size_t i = 0; i < statistics->metrics_len; ++i) {
++ const LanceScanMetric *metric = &statistics->metrics[i];
++ if ((metric->name_len > 0 && metric->name == NULL) ||
++ (metric->kind != LANCE_SCAN_METRIC_COUNT &&
++ metric->kind != LANCE_SCAN_METRIC_TIME_NANOSECONDS)) {
++ captured->invalid = 1;
++ return;
++ }
++ }
++ captured->calls += 1;
++ captured->bytes_read = statistics->bytes_read;
++}
++
+ static void test_open_and_metadata(const char *uri) {
+ printf(" test_open_and_metadata... ");
+
+@@ -84,10 +114,14 @@ static void test_scan(const char *uri) {
+ /* Full scan via ArrowArrayStream */
+ LanceScanner *scanner = lance_scanner_new(ds, NULL, NULL);
+ ASSERT(scanner != NULL, "scanner creation failed");
++ ScanStatisticsCapture captured = {0};
++ int32_t rc = lance_scanner_set_statistics_callback(
++ scanner, capture_scan_statistics, &captured);
++ ASSERT(rc == 0, "statistics callback registration failed");
+
+ struct ArrowArrayStream stream;
+ memset(&stream, 0, sizeof(stream));
+- int32_t rc = lance_scanner_to_arrow_stream(scanner, &stream);
++ rc = lance_scanner_to_arrow_stream(scanner, &stream);
+ ASSERT(rc == 0, "to_arrow_stream failed");
+
+ /* Read schema from stream */
+@@ -113,6 +147,9 @@ static void test_scan(const char *uri) {
+ }
+
+ ASSERT(total_rows == expected_rows, "row count mismatch");
++ ASSERT(captured.calls == 1, "statistics callback count mismatch");
++ ASSERT(captured.bytes_read > 0, "statistics should report bytes read");
++ ASSERT(captured.invalid == 0, "statistics callback received invalid
data");
+ printf("rows=%llu... ", (unsigned long long)total_rows);
+
+ if (stream.release) stream.release(&stream);
+diff --git a/tests/cpp/test_cpp_api.cpp b/tests/cpp/test_cpp_api.cpp
+index f8ae701..3293bfb 100644
+--- a/tests/cpp/test_cpp_api.cpp
++++ b/tests/cpp/test_cpp_api.cpp
+@@ -25,6 +25,25 @@
+ #define TEST(name) printf(" %s... ", #name)
+ #define PASS() printf("OK\n")
+
++struct ScanStatisticsCapture {
++ uint64_t calls = 0;
++ uint64_t bytes_read = 0;
++ bool invalid = false;
++};
++
++static void capture_scan_statistics(
++ void* callback_ctx,
++ const LanceScanStatistics* statistics) noexcept {
++ if (!callback_ctx) return;
++ auto* captured = static_cast<ScanStatisticsCapture*>(callback_ctx);
++ if (!statistics || (statistics->metrics_len > 0 && !statistics->metrics))
{
++ captured->invalid = true;
++ return;
++ }
++ captured->calls += 1;
++ captured->bytes_read = statistics->bytes_read;
++}
++
+ static void test_dataset_open(const std::string& uri) {
+ TEST(test_dataset_open);
+
+@@ -70,7 +89,11 @@ static void test_scanner_fluent(const std::string& uri) {
+
+ // Fluent builder pattern.
+ auto scanner = ds.scan();
+- scanner.limit(5).offset(0).batch_size(2);
++ ScanStatisticsCapture captured;
++ scanner.limit(5)
++ .offset(0)
++ .batch_size(2)
++ .statistics_callback(capture_scan_statistics, &captured);
+
+ ArrowArrayStream stream;
+ memset(&stream, 0, sizeof(stream));
+@@ -89,6 +112,9 @@ static void test_scanner_fluent(const std::string& uri) {
+ }
+
+ assert(total == 5);
++ assert(captured.calls == 1);
++ assert(captured.bytes_read > 0);
++ assert(!captured.invalid);
+ printf("rows=%llu... ", (unsigned long long)total);
+
+ if (stream.release) stream.release(&stream);
diff --git a/thirdparty/vars.sh b/thirdparty/vars.sh
index 3836ee9e558..219bdef38b0 100644
--- a/thirdparty/vars.sh
+++ b/thirdparty/vars.sh
@@ -552,10 +552,10 @@ PUGIXML_SOURCE=pugixml-1.15
PUGIXML_MD5SUM="3b894c29455eb33a40b165c6e2de5895"
# lance-c
-LANCE_C_DOWNLOAD="https://github.com/lance-format/lance-c/archive/refs/tags/v0.1.6.tar.gz";
-LANCE_C_NAME="lance-c-v0.1.6.tar.gz"
-LANCE_C_SOURCE="lance-c-0.1.6"
-LANCE_C_MD5SUM="1599faa2532d9ce963db1188f7435a56"
+LANCE_C_DOWNLOAD="https://github.com/lance-format/lance-c/archive/refs/tags/v0.1.7.tar.gz";
+LANCE_C_NAME="lance-c-v0.1.7.tar.gz"
+LANCE_C_SOURCE="lance-c-0.1.7"
+LANCE_C_MD5SUM="15ef7cd20a2e1606384251cb2d41d42f"
# all thirdparties which need to be downloaded is set in array TP_ARCHIVES
export TP_ARCHIVES=(
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
