github-actions[bot] commented on code in PR #66841:
URL: https://github.com/apache/doris/pull/66841#discussion_r3796633506
##########
gensrc/thrift/PlanNodes.thrift:
##########
@@ -631,10 +631,11 @@ struct TFileScanRangeParams {
35: optional string serialized_table_cache_key
// Serialized Substrait ExtendedExpression executed by the native Lance
scanner. Set at
// ScanNode level so it is not serialized once per fragment split.
- 36: optional binary lance_substrait_filter
+ 37: optional binary lance_substrait_filter
Review Comment:
[P1] Preserve the deployed field IDs for rolling upgrades. Thrift identifies
fields by number, so moving `lance_substrait_filter` from 36 to 37 (and the
existing search request from 37 to 38) makes mixed-version peers skip them.
This is result-changing for ordinary Lance scans: FE removes a successfully
pushed conjunct from the residual list and sends its only remaining copy in
field 36, while a new BE no longer recognizes field 36 and can return
unfiltered rows. The search request is also dropped across old/new peers
because field 37 changes wire type. Please keep these existing fields at 36 and
37; the member/type names can change without renumbering them.
##########
be/src/format_v2/table/lance_reader.cpp:
##########
@@ -369,33 +368,13 @@ Status LanceTableReader::prepare_split(const
SplitReadOptions& options) {
if (current_split_pruned()) {
return Status::OK();
}
-
- if (_vector_search) {
- const auto& lance_params =
options.current_range.table_format_params.lance_params;
- if (lance_params.version <= 0) {
- return Status::InvalidArgument(
- "Lance vector search requires a fixed positive dataset
version");
- }
- if (lance_params.__isset.fragment_ids) {
- return Status::InvalidArgument("Lance vector search split must not
set fragment ids");
- }
- if (_search_split_prepared) {
- return Status::InvalidArgument(
- "Lance vector search supports exactly one whole-dataset
split");
- }
- }
-
- const auto key = _dataset_key(options.current_range);
- if (_dataset == nullptr) {
- RETURN_IF_ERROR(_open_dataset(key));
- _opened_dataset_key = key;
- } else if (!_opened_dataset_key.has_value() || *_opened_dataset_key !=
key) {
+ if (_global_rowid_output_idx.has_value() &&
!_global_rowid_context.has_value()) {
return Status::InvalidArgument(
- "Lance reader cannot mix dataset snapshots or storage options
in one scan");
+ "Lance global row id requested without global row id context");
}
+ RETURN_IF_ERROR(_ensure_dataset_open(options.current_range));
Review Comment:
[P1] Retain the negative-fragment check before opening this scanner. An old
FE widens Lance Java's signed `int Fragment.getId()` without this PR's new
unsigned conversion, so a high-bit u32 ID arrives negative during a rolling
upgrade. The unchanged scanner path casts it to a huge u64 rather than the
original fragment ID, and [pinned
lance-c](https://github.com/lance-format/lance-c/blob/v0.1.6/src/scanner.rs#L118-L131)
accepts that nonexistent ID by selecting an empty fragment set; the scan then
succeeds while silently dropping the fragment. Keeping the
deserialized-boundary validation preserves an explicit failure for
old-FE/new-BE execution instead of wrong results.
##########
be/src/storage/utils.h:
##########
@@ -249,4 +252,24 @@ struct GlobalRowLoacationV2 {
auto operator<=>(const GlobalRowLoacationV2&) const = default;
};
+// Global row location for data sources whose native row ID is wider than the
+// uint32_t row ordinal carried by GlobalRowLoacationV2. Keep V2 unchanged for
+// existing internal, Parquet, and ORC readers.
+struct GlobalRowLocationV3 {
+ static constexpr uint8_t VERSION = 1;
+
+ GlobalRowLocationV3(int64_t bid, uint32_t fid, uint64_t rid)
+ : version(VERSION), backend_id(bid), file_id(fid), row_id(rid) {}
+ uint8_t version;
+ std::array<uint8_t, 7> reserved_before_backend_id {};
+ int64_t backend_id;
+ uint32_t file_id;
+ std::array<uint8_t, 4> reserved_before_row_id {};
Review Comment:
[P1] Gate this V3 row-location encoding on the materializing BE's
capability. During a rolling upgrade, a new scan BE can send these bytes
through the distributed TopN to an old materialization BE. The old release code
raw-casts every value as the 24-byte V2 layout without checking its version or
length. This 32-byte layout preserves the backend/file offsets but puts zero
padding at V2's row-id offset 20, so the old BE silently decodes every V3 value
as row ID 0 and may splice the first row's deferred columns into a different
selected row. Please negotiate V3 support (or use an old-reader-safe
representation) before emitting it.
##########
be/src/exec/rowid_fetcher.cpp:
##########
@@ -927,13 +1001,26 @@ Status RowIdStorageReader::read_batch_external_row(
RETURN_IF_ERROR(remote_scan_sched->submit_scan_task(
SimplifiedScanTask(
[&, idx, scan_info]() -> Status {
+ Defer complete_task {[&]() {
+ semaphore.release();
+ if (++producer_count ==
scan_rows.size()) {
+ std::lock_guard<std::mutex>
lock(mtx);
+ cv.notify_one();
+ }
+ }};
const auto& [row_ids, file_mapping] =
scan_info;
- return
read_external_row_from_file_mapping(
+ auto status =
read_external_row_from_file_mapping(
idx, row_ids, file_mapping,
slots, query_id,
runtime_state, scan_blocks,
row_id_block_idx,
fetch_statistics,
rpc_scan_params,
- colname_to_slot_id,
producer_count,
- scan_rows.size(), semaphore,
cv, mtx, tuple_desc);
+ colname_to_slot_id,
tuple_desc);
+ if (!status.ok()) {
+ std::lock_guard<std::mutex>
lock(mtx);
+ if (scan_status.ok()) {
+ scan_status = status;
+ }
+ }
+ return status;
Review Comment:
[P1] Mark this one-shot scheduler task complete even when the fetch fails.
`SimplifiedScanTask` stores a `std::function<bool()>`, so returning an error
`Status` here converts to `false`. `ScannerSplitRunner` then never sets its
completion future, it is not auto-rescheduled, and the one-shot wrapper removes
its task handle only for a `true` result. The new defer wakes the waiter, so
the caller returns `scan_status` while the executor still retains an unfinished
split/closure capturing this stack frame by reference; repeated read errors
leak those scheduler tasks for the process lifetime. Record the error in
`scan_status`, but return `true` from the scheduler callback after the single
attempt.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/lance/source/LanceScanNode.java:
##########
@@ -152,42 +153,34 @@ public void createScanRangeLocations() throws
UserException {
@Override
public List<Split> getSplits(int numBackends) throws UserException {
- if (isExternalSearch()) {
- plannedVersion = plannedMetadata.getVersion();
- plannedFragments = plannedMetadata.getFragments().size();
-
- // Do not attach fragment IDs. A vector index is a dataset-wide
structure and one
- // scanner must see every fragment visible in this pinned snapshot
to produce global
- // TopK. Fragment-level parallel search and result merging are
intentionally deferred.
- return Collections.singletonList(LanceSplit.wholeDatasetAtVersion(
- plannedMetadata.getDatasetUri(),
plannedMetadata.getVersion(),
- plannedMetadata.getRowCount()));
- } else {
- LanceTableMetadata metadata = plannedMetadata;
- plannedVersion = metadata.getVersion();
- plannedFragments = metadata.getFragments().size();
- Set<Long> fragmentIds = new HashSet<>();
- long targetRows = 1;
- for (LanceTableMetadata.LanceFragmentInfo fragment :
metadata.getFragments()) {
- if (!fragmentIds.add(fragment.getId())) {
- throw new UserException("Duplicate Lance fragment id " +
fragment.getId()
- + " at dataset version " + metadata.getVersion());
- }
- targetRows = Math.max(targetRows,
Math.max(fragment.getPhysicalRows(), 1));
+ LanceTableMetadata metadata = plannedMetadata;
+ plannedVersion = metadata.getVersion();
+ plannedFragments = metadata.getFragments().size();
+ if (isExternalSearch() && plannedVersion <= 0) {
+ throw new UserException(
+ "Lance vector search requires a fixed positive dataset
version");
+ }
+ Set<Long> fragmentIds = new HashSet<>();
+ long targetRows = 1;
+ for (LanceFragmentInfo fragment : metadata.getFragments()) {
+ if (!fragmentIds.add(fragment.getId())) {
+ throw new UserException("Duplicate Lance fragment id " +
fragment.getId()
+ + " at dataset version " + metadata.getVersion());
}
+ targetRows = Math.max(targetRows,
Math.max(fragment.getPhysicalRows(), 1));
+ }
- // Use the largest fragment as one standard split so smaller
fragments keep
- // their relative row-count weight during backend assignment.
Physical rows drive
- // the weight because the BE legacy reader scans physical batches
before deletions.
- List<Split> splits = new ArrayList<>(plannedFragments);
- for (LanceTableMetadata.LanceFragmentInfo fragment :
metadata.getFragments()) {
- LanceSplit split = new LanceSplit(metadata.getDatasetUri(),
metadata.getVersion(),
- fragment.getId(), fragment.getPhysicalRows());
- split.setTargetSplitSize(targetRows);
- splits.add(split);
- }
- return splits;
+ // Keep one fragment per split. Use the largest fragment's physical
row count as the
+ // normalization baseline for split weights, so backend scheduling
reflects the relative
+ // amount of physical data each fragment scans, including rows covered
by deletion metadata.
+ List<Split> splits = new ArrayList<>(plannedFragments);
+ for (LanceFragmentInfo fragment : metadata.getFragments()) {
+ LanceSplit split = new LanceSplit(metadata.getDatasetUri(),
metadata.getVersion(),
Review Comment:
[P1] Gate fragment-scoped vector search on BE capability during rolling
upgrades. Once the existing search request stays at field 37, these new
one-fragment ranges decode on an old BE, but its [pinned Lance
path](https://github.com/lance-format/lance-c/blob/v0.1.6/src/scanner.rs#L145-L178)
installs `nearest()` before prefilter (and enables prefilter only for an
explicit filter). The lance-c patch in this PR documents that
fragment-restricted nearest is accepted only when prefilter is active first, so
a new-FE/old-BE query now fails instead of using the previously supported
whole-snapshot search. Please retain the old whole-snapshot plan or exclude old
scan BEs until every target supports distributed Lance search.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
