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 ce5f5dd4239 [refactor](lance) Group Lance scan parameters in Thrift
(#67110)
ce5f5dd4239 is described below
commit ce5f5dd4239f35b6efba43d2376dffd587ad4d9b
Author: zhangstar333 <[email protected]>
AuthorDate: Wed Aug 26 10:35:56 2026 +0800
[refactor](lance) Group Lance scan parameters in Thrift (#67110)
### What problem does this PR solve?
Problem Summary:
Group Lance-specific scan-node parameters into `TLanceScanParams` and
update the FE/BE access paths.
Keep `TFileScanRangeParams` clean and make Lance scan parameters easier
to extend.
### Release note
None
### Check List (For Author)
- Test <!-- At least one of them must be included. -->
- [ ] Regression test
- [ ] Unit Test
- [ ] Manual test (add detailed scripts or steps below)
- [x] 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:
- [ ] 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/format_v2/table/lance_reader.cpp | 44 ++++++++++++++--------
be/test/format_v2/table/lance_reader_test.cpp | 22 ++++++++---
.../datasource/lance/source/LanceScanNode.java | 16 ++++++--
.../doris/datasource/tvf/source/TVFScanNode.java | 4 +-
.../ExternalFileTableValuedFunction.java | 4 +-
.../doris/datasource/LanceThriftContractTest.java | 12 ++++--
gensrc/thrift/PlanNodes.thrift | 37 ++++++++++--------
7 files changed, 92 insertions(+), 47 deletions(-)
diff --git a/be/src/format_v2/table/lance_reader.cpp
b/be/src/format_v2/table/lance_reader.cpp
index d909aac8e5e..29d334f4b98 100644
--- a/be/src/format_v2/table/lance_reader.cpp
+++ b/be/src/format_v2/table/lance_reader.cpp
@@ -302,10 +302,12 @@ Status LanceTableReader::init(TableReadOptions&& options)
{
DORIS_CHECK(_scan_params != nullptr);
_ctz = _runtime_state->timezone_obj();
- _vector_search = _scan_params->__isset.external_search_request;
+ const auto& lance_scan_params = _scan_params->lance_scan_params;
+ _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 = _scan_params->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 ||
@@ -317,11 +319,12 @@ Status LanceTableReader::init(TableReadOptions&& options)
{
_scanner_profile->add_info_string("LanceUseIndex", use_index ? "true"
: "false");
_fragment_count = ADD_COUNTER(_scanner_profile, "LanceFragmentCount",
TUnit::UNIT);
}
- if (_scan_params->__isset.lance_substrait_filter) {
+ if (_scan_params->__isset.lance_scan_params &&
+ lance_scan_params.__isset.lance_substrait_filter) {
_scanner_profile->add_info_string("LancePushdownFormat", "SUBSTRAIT");
_scanner_profile->add_info_string(
"LanceSubstraitFilterBytes",
- std::to_string(_scan_params->lance_substrait_filter.size()));
+
std::to_string(lance_scan_params.lance_substrait_filter.size()));
}
_output_name_to_idx.clear();
@@ -522,14 +525,16 @@ Status
LanceTableReader::_validate_external_search_request() const {
// Thrift boundary. Recheck structural invariants and values used for
allocation, pointer
// arithmetic, C-string calls, and narrowing conversions before accessing
them below.
DORIS_CHECK(_scan_params != nullptr);
- DORIS_CHECK(_scan_params->__isset.external_search_request);
- if (_scan_params->__isset.lance_substrait_filter) {
+ DORIS_CHECK(_scan_params->__isset.lance_scan_params);
+ const auto& lance_scan_params = _scan_params->lance_scan_params;
+ DORIS_CHECK(lance_scan_params.__isset.external_search_request);
+ if (lance_scan_params.__isset.lance_substrait_filter) {
return Status::InvalidArgument(
"Lance vector search cannot combine its pre-search filter with
"
"lance_substrait_filter");
}
- const auto& request = _scan_params->external_search_request;
+ const auto& request = lance_scan_params.external_search_request;
if (request.schema_version != 1) {
return Status::NotSupported("unsupported external search schema
version: {}",
request.schema_version);
@@ -671,9 +676,10 @@ Status LanceTableReader::_open_scanner(const
TFileRangeDesc& range) {
}
columns.emplace_back(nullptr);
+ const auto& lance_scan_params = _scan_params->lance_scan_params;
const char* sql_filter = nullptr;
if (_vector_search) {
- const auto& request = _scan_params->external_search_request;
+ const auto& request = lance_scan_params.external_search_request;
if (request.__isset.search_filter &&
request.search_filter.format == TSearchFilterFormat::SQL) {
sql_filter = request.search_filter.payload.c_str();
@@ -690,16 +696,17 @@ Status LanceTableReader::_open_scanner(const
TFileRangeDesc& range) {
return _lance_error("enable Lance row id output");
}
- if (_scan_params->__isset.lance_substrait_filter &&
- !_scan_params->lance_substrait_filter.empty()) {
- const auto& filter = _scan_params->lance_substrait_filter;
+ if (_scan_params->__isset.lance_scan_params &&
+ lance_scan_params.__isset.lance_substrait_filter &&
+ !lance_scan_params.lance_substrait_filter.empty()) {
+ const auto& filter = lance_scan_params.lance_substrait_filter;
if (lance_scanner_set_substrait_filter(
scanner, reinterpret_cast<const uint8_t*>(filter.data()),
filter.size()) != 0) {
return _lance_error("set Lance Substrait filter");
}
}
if (_vector_search) {
- const auto& request = _scan_params->external_search_request;
+ const auto& request = lance_scan_params.external_search_request;
if (request.__isset.search_filter &&
request.search_filter.format == TSearchFilterFormat::SUBSTRAIT) {
const auto& filter = request.search_filter.payload;
@@ -782,7 +789,10 @@ Status LanceTableReader::_open_scanner(const
TFileRangeDesc& range) {
Status LanceTableReader::_configure_vector_search(LanceScanner* scanner) const
{
DORIS_CHECK(scanner != nullptr);
DORIS_CHECK(_scan_params != nullptr);
- const auto& request = _scan_params->external_search_request;
+ DORIS_CHECK(_scan_params->__isset.lance_scan_params);
+ const auto& lance_scan_params = _scan_params->lance_scan_params;
+ DORIS_CHECK(lance_scan_params.__isset.external_search_request);
+ const auto& request = lance_scan_params.external_search_request;
const auto& vector = request.search_query.vector_search;
const auto& query = vector.query_vector;
const auto dimension = static_cast<size_t>(query.dimension);
@@ -1033,11 +1043,13 @@ Status LanceTableReader::_fill_block_from_record_batch(
Status LanceTableReader::_storage_options(const TFileScanRangeParams*
scan_params,
std::vector<std::string>* options) {
options->clear();
- if (scan_params == nullptr || !scan_params->__isset.lance_storage_options)
{
+ if (scan_params == nullptr || !scan_params->__isset.lance_scan_params ||
+ !scan_params->lance_scan_params.__isset.lance_storage_options) {
return Status::OK();
}
- options->reserve(scan_params->lance_storage_options.size() * 2);
- for (const auto& [key, value] : scan_params->lance_storage_options) {
+ const auto& storage_options =
scan_params->lance_scan_params.lance_storage_options;
+ options->reserve(storage_options.size() * 2);
+ for (const auto& [key, value] : storage_options) {
// These become C strings below, so a NUL would truncate the option
here while the FE went
// on using the whole thing, and the two halves would open the dataset
with different
// configuration. The FE rejects these on both paths it builds options
from - its own
diff --git a/be/test/format_v2/table/lance_reader_test.cpp
b/be/test/format_v2/table/lance_reader_test.cpp
index 9b2943f0ad6..a576f1f63a2 100644
--- a/be/test/format_v2/table/lance_reader_test.cpp
+++ b/be/test/format_v2/table/lance_reader_test.cpp
@@ -224,8 +224,10 @@ TFileScanRangeParams make_float32_vector_search_params(
request.__set_search_filter(std::move(search_filter));
}
+ TLanceScanParams lance_scan_params;
+ lance_scan_params.__set_external_search_request(std::move(request));
TFileScanRangeParams scan_params;
- scan_params.__set_external_search_request(std::move(request));
+ scan_params.__set_lance_scan_params(std::move(lance_scan_params));
return scan_params;
}
@@ -238,7 +240,8 @@ TEST(LanceTableReaderVectorSearchTest,
RejectsMalformedVectorPayloadBeforeReadin
RuntimeState state(query_globals);
RuntimeProfile profile("lance_vector_search_invalid_request");
auto scan_params = make_float32_vector_search_params({0.0F, 0.0F, 0.0F},
2, 0);
-
scan_params.external_search_request.search_query.vector_search.query_vector.__set_dimension(4);
+
scan_params.lance_scan_params.external_search_request.search_query.vector_search.query_vector
+ .__set_dimension(4);
LanceTableReader reader;
const auto status = init_reader(&reader, columns, &state, &profile,
&scan_params);
@@ -323,7 +326,8 @@ TEST(LanceTableReaderVectorSearchTest,
SearchesWholeSnapshotWithOffsetAndDistanc
state.set_query_options(query_options);
RuntimeProfile profile("lance_vector_search_fixture");
auto scan_params = make_float32_vector_search_params({0.0F, 0.0F, 0.0F},
2, 1);
- auto& search_options =
scan_params.external_search_request.vector_search_options;
+ auto& search_options =
+
scan_params.lance_scan_params.external_search_request.vector_search_options;
search_options.__set_nprobes(4);
search_options.__set_refine_factor(2);
search_options.__set_ef(16);
@@ -611,7 +615,9 @@ TEST(LanceTableReaderFilterTest,
PushesFilterOnNonProjectedColumn) {
"Og8QRioLZG9yaXMtbGFuY2U=";
std::string substrait_filter;
ASSERT_TRUE(base64_decode(substrait_filter_base64, &substrait_filter));
- scan_params.__set_lance_substrait_filter(std::move(substrait_filter));
+ TLanceScanParams lance_scan_params;
+
lance_scan_params.__set_lance_substrait_filter(std::move(substrait_filter));
+ scan_params.__set_lance_scan_params(std::move(lance_scan_params));
LanceTableReader reader;
ASSERT_TRUE(init_reader(&reader, columns, &state, &profile,
&scan_params).ok());
@@ -686,7 +692,9 @@ void expect_filtered_row_ids(const char* profile_name,
const std::string& substr
TFileScanRangeParams scan_params;
std::string substrait_filter;
ASSERT_TRUE(base64_decode(substrait_filter_base64, &substrait_filter));
- scan_params.__set_lance_substrait_filter(std::move(substrait_filter));
+ TLanceScanParams lance_scan_params;
+
lance_scan_params.__set_lance_substrait_filter(std::move(substrait_filter));
+ scan_params.__set_lance_scan_params(std::move(lance_scan_params));
LanceTableReader reader;
ASSERT_TRUE(init_reader(&reader, columns, &state, &profile,
&scan_params).ok());
@@ -960,8 +968,10 @@ TEST(LanceTableReaderScanTest,
RejectsStorageOptionWithEmbeddedNul) {
// using the whole thing, leaving the two halves opening the dataset with
different
// configuration. Dropping it instead of failing would only move that
divergence.
TFileScanRangeParams scan_params;
- scan_params.__set_lance_storage_options(
+ TLanceScanParams lance_scan_params;
+ lance_scan_params.__set_lance_storage_options(
{{std::string("aws_region\0ignored", 18), "us-east-1"}});
+ scan_params.__set_lance_scan_params(std::move(lance_scan_params));
LanceTableReader reader;
ASSERT_TRUE(init_reader(&reader, columns, &state, &profile,
&scan_params).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 ccae0570ec9..38ab936bd74 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
@@ -40,6 +40,7 @@ import org.apache.doris.thrift.TExternalSearchRequest;
import org.apache.doris.thrift.TFileFormatType;
import org.apache.doris.thrift.TFileRangeDesc;
import org.apache.doris.thrift.TLanceFileDesc;
+import org.apache.doris.thrift.TLanceScanParams;
import org.apache.doris.thrift.TTableFormatFileDesc;
import org.apache.doris.thrift.TVectorMetric;
import org.apache.doris.thrift.TVectorSearchParams;
@@ -125,10 +126,18 @@ public class LanceScanNode extends FileQueryScanNode {
if (isExternalSearch()) {
// Search output comes from the FunctionGenTable because it adds
generated columns such
// as _distance. The real Lance table is still retained for
storage and metadata access.
-
params.setExternalSearchRequest(createFragmentSearchRequest(externalSearchRequest));
+ getOrCreateLanceScanParams()
+
.setExternalSearchRequest(createFragmentSearchRequest(externalSearchRequest));
}
}
+ private TLanceScanParams getOrCreateLanceScanParams() {
+ if (!params.isSetLanceScanParams()) {
+ params.setLanceScanParams(new TLanceScanParams());
+ }
+ return params.getLanceScanParams();
+ }
+
// A fragment-level LIMIT can be pushed into an ordinary Lance scan only
when every predicate
// is already pushed into Lance (conjuncts is empty). Otherwise Doris
re-filters the returned
// rows and truncating a fragment early could drop valid results.
@@ -162,12 +171,13 @@ public class LanceScanNode extends FileQueryScanNode {
public void createScanRangeLocations() throws UserException {
super.createScanRangeLocations();
if (lanceSubstraitFilter.length > 0) {
-
params.setLanceSubstraitFilter(ByteBuffer.wrap(lanceSubstraitFilter));
+ getOrCreateLanceScanParams()
+
.setLanceSubstraitFilter(ByteBuffer.wrap(lanceSubstraitFilter));
}
// Set at ScanNode level so credentials are not serialized once per
fragment split.
Map<String, String> lanceStorageOptions =
plannedMetadata.getLanceStorageOptions();
if (!lanceStorageOptions.isEmpty()) {
- params.setLanceStorageOptions(lanceStorageOptions);
+
getOrCreateLanceScanParams().setLanceStorageOptions(lanceStorageOptions);
}
}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/datasource/tvf/source/TVFScanNode.java
b/fe/fe-core/src/main/java/org/apache/doris/datasource/tvf/source/TVFScanNode.java
index ba1a47396d6..f7e0e86c5b0 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/datasource/tvf/source/TVFScanNode.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/datasource/tvf/source/TVFScanNode.java
@@ -49,6 +49,7 @@ import org.apache.doris.thrift.TFileFormatType;
import org.apache.doris.thrift.TFileRangeDesc;
import org.apache.doris.thrift.TFileType;
import org.apache.doris.thrift.TLanceFileDesc;
+import org.apache.doris.thrift.TLanceScanParams;
import org.apache.doris.thrift.TTableFormatFileDesc;
import com.google.common.collect.Lists;
@@ -136,7 +137,8 @@ public class TVFScanNode extends FileQueryScanNode {
tableValuedFunction.getFilePath(),
Collections.singletonList(tableValuedFunction.getStorageProperties()));
if (!lanceStorageOptions.isEmpty()) {
- params.setLanceStorageOptions(lanceStorageOptions);
+ params.setLanceScanParams(
+ new
TLanceScanParams().setLanceStorageOptions(lanceStorageOptions));
}
}
}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/tablefunction/ExternalFileTableValuedFunction.java
b/fe/fe-core/src/main/java/org/apache/doris/tablefunction/ExternalFileTableValuedFunction.java
index 0e80b40eada..8955deba76f 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/tablefunction/ExternalFileTableValuedFunction.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/tablefunction/ExternalFileTableValuedFunction.java
@@ -82,6 +82,7 @@ import org.apache.doris.thrift.TFileScanRangeParams;
import org.apache.doris.thrift.TFileType;
import org.apache.doris.thrift.THdfsParams;
import org.apache.doris.thrift.TLanceFileDesc;
+import org.apache.doris.thrift.TLanceScanParams;
import org.apache.doris.thrift.TNetworkAddress;
import org.apache.doris.thrift.TPrimitiveType;
import org.apache.doris.thrift.TStatusCode;
@@ -541,7 +542,8 @@ public abstract class ExternalFileTableValuedFunction
extends TableValuedFunctio
Map<String, String> lanceStorageOptions =
LanceStorageOptions.forUri(
filePath, Collections.singletonList(storageProperties));
if (!lanceStorageOptions.isEmpty()) {
-
fileScanRangeParams.setLanceStorageOptions(lanceStorageOptions);
+ fileScanRangeParams.setLanceScanParams(
+ new
TLanceScanParams().setLanceStorageOptions(lanceStorageOptions));
}
}
fileScanRangeParams.setFileAttributes(getFileAttributes());
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/datasource/LanceThriftContractTest.java
b/fe/fe-core/src/test/java/org/apache/doris/datasource/LanceThriftContractTest.java
index c541af25049..331872d1552 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/datasource/LanceThriftContractTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/datasource/LanceThriftContractTest.java
@@ -20,6 +20,7 @@ package org.apache.doris.datasource;
import org.apache.doris.thrift.TFileFormatType;
import org.apache.doris.thrift.TFileScanRangeParams;
import org.apache.doris.thrift.TLanceFileDesc;
+import org.apache.doris.thrift.TLanceScanParams;
import org.apache.doris.thrift.TTableFormatFileDesc;
import org.apache.thrift.TDeserializer;
@@ -92,7 +93,8 @@ public class LanceThriftContractTest {
TFileScanRangeParams source = new TFileScanRangeParams()
.setFormatType(TFileFormatType.FORMAT_LANCE)
- .setLanceStorageOptions(storageOptions);
+ .setLanceScanParams(
+ new
TLanceScanParams().setLanceStorageOptions(storageOptions));
TSerializer serializer = new TSerializer(new
TCompactProtocol.Factory());
byte[] bytes = serializer.serialize(source);
@@ -102,8 +104,10 @@ public class LanceThriftContractTest {
// Whatever the namespace vended has to reach lance-c unchanged,
including keys Doris
// itself assigns no meaning to.
- Assert.assertTrue(restored.isSetLanceStorageOptions());
- Assert.assertEquals(storageOptions, restored.getLanceStorageOptions());
+ Assert.assertTrue(restored.isSetLanceScanParams());
+
Assert.assertTrue(restored.getLanceScanParams().isSetLanceStorageOptions());
+ Assert.assertEquals(storageOptions,
+ restored.getLanceScanParams().getLanceStorageOptions());
}
@Test
@@ -118,6 +122,6 @@ public class LanceThriftContractTest {
new TDeserializer(new
TCompactProtocol.Factory()).deserialize(restored, bytes);
// A local dataset needs no storage configuration at all.
- Assert.assertFalse(restored.isSetLanceStorageOptions());
+ Assert.assertFalse(restored.isSetLanceScanParams());
}
}
diff --git a/gensrc/thrift/PlanNodes.thrift b/gensrc/thrift/PlanNodes.thrift
index f0c26fa1175..b543afca315 100644
--- a/gensrc/thrift/PlanNodes.thrift
+++ b/gensrc/thrift/PlanNodes.thrift
@@ -536,6 +536,25 @@ struct TLanceFileDesc {
5: optional list<binary> index_segment_uuids
}
+struct TLanceScanParams {
+ // Serialized Substrait ExtendedExpression executed by the native Lance
scanner. Set at
+ // ScanNode level so it is not serialized once per fragment split.
+ 1: optional binary lance_substrait_filter
+ // Provider-independent search request. Set at ScanNode level so all
ranges use the same logical
+ // query. Lance vector search uses one range per fragment and Doris merges
the split-local
+ // candidates.
+ 2: optional TExternalSearchRequest external_search_request
+ // Lance-native storage options, handed to lance-c untranslated. The
namespace protocol treats
+ // storage_options as opaque configuration passed directly to Lance, so
any key vocabulary the
+ // BE imposed here would drop options it does not happen to know -
including credentials a
+ // namespace spelled with a different accepted alias, and every non-S3
provider's keys.
+ // Set at ScanNode level so credentials are not serialized once per
fragment split.
+ // These are the initial options for the scan and are never refreshed:
lance-c opens datasets
+ // with a static option set, so credentials that expire mid-scan are not
re-vended. Renewal
+ // needs a refresh channel of its own, which this field is not.
+ 3: optional map<string, string> lance_storage_options
+}
+
struct TTableFormatFileDesc {
1: optional string table_format_type
2: optional TIcebergFileDesc iceberg_params
@@ -632,22 +651,8 @@ struct TFileScanRangeParams {
34: optional i32 iceberg_scan_semantics_version
// FE-generated identity for sharing a deserialized table across JNI
scanners in one scan node.
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.
- 37: optional binary lance_substrait_filter
- // Provider-independent search request. Set at ScanNode level so all
ranges use the same logical
- // query. Lance vector search uses one range per fragment and Doris merges
the split-local
- // candidates.
- 38: optional TExternalSearchRequest external_search_request
- // Lance-native storage options, handed to lance-c untranslated. The
namespace protocol treats
- // storage_options as opaque configuration passed directly to Lance, so
any key vocabulary the
- // BE imposed here would drop options it does not happen to know -
including credentials a
- // namespace spelled with a different accepted alias, and every non-S3
provider's keys.
- // Set at ScanNode level so credentials are not serialized once per
fragment split.
- // These are the initial options for the scan and are never refreshed:
lance-c opens datasets
- // with a static option set, so credentials that expire mid-scan are not
re-vended. Renewal
- // needs a refresh channel of its own, which this field is not.
- 39: optional map<string, string> lance_storage_options
+ // 31-33 and 36 are used in master; do not allocate them in branch-4.1.
+ 37: optional TLanceScanParams lance_scan_params
}
struct TFileRangeDesc {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
