github-actions[bot] commented on code in PR #69493:
URL: https://github.com/apache/doris/pull/69493#discussion_r4358943078
##########
regression-test/framework/src/main/groovy/org/apache/doris/regression/util/ResultUtils.groovy:
##########
@@ -45,6 +47,21 @@ class ResultUtils {
private static final Pattern DATETIME_PATTERN = Pattern.compile(
"(\\d{4}-\\d{2}-\\d{2})[T
](\\d{2}:\\d{2})(?::(\\d{2})(?:\\.(\\d{1,9}))?)?")
private static final Pattern UNICODE_ESCAPE_PATTERN =
Pattern.compile("[0-9a-fA-F]{4}")
+ /**
+ * A "0x" token as Doris renders a VARBINARY nested in arrays/maps/structs.
+ * The negative lookbehind keeps the match at a VALUE/ELEMENT boundary: a
+ * token preceded by an alphanumeric ("prefix0x41"), a quote (a nested
+ * STRING the engine renders quoted) or a backslash is a literal-string
+ * fragment, not a binary rendering — substituting it could make unequal
+ * rows compare equal (Spark "prefixA" vs Doris "prefix0x41"). Only a token
+ * that Doris could only have produced as a binary element ("[0x616C, "
+ * / "{0x41: 0x42}" / "ROW(0x61)") is decoded. A literal string whose whole
+ * value coincides with a bare token ("0x41") is indistinguishable from a
+ * binary rendering without column types and remains the one known
+ * ambiguity.
+ */
+ private static final Pattern NESTED_BINARY_TOKEN = Pattern.compile(
+ "(?i)(?<![0-9a-zA-Z'"\\])0x(?:[0-9a-f]{2})+(?![0-9a-fA-F])")
Review Comment:
[P1] Escape the quote and backslash in the Groovy regex literal
The double quote immediately after the apostrophe in the lookbehind is
unescaped inside a double-quoted Groovy string, so it ends the literal and
leaves the following backslashes outside any string. ResultUtils belongs to the
shared regression framework, so the framework cannot compile and suites cannot
start, regardless of whether the Paimon test is enabled. Use valid Groovy
quoting and preserve the regex's literal backslash through both escaping layers.
##########
be/src/format_v2/table/paimon_rust_table_reader.cpp:
##########
@@ -509,6 +626,316 @@ Status PaimonRustTableReader::_open_split_reader(const
TFileRangeDesc& range) {
return Status::OK();
}
+Status PaimonRustTableReader::_open_vector_split_reader(const TFileRangeDesc&
range) {
+ // Validate every __isset / value-range up front (mirrors
+ // LanceTableReader::_validate_external_search_request). The build path
below
+ // then uses the fields unchecked, matching lance's split of validate-once
/
+ // build-without-recheck.
+ RETURN_IF_ERROR(_validate_external_search_request());
+ const auto& request = _scan_params->external_search_request;
+ const auto& vector = request.search_query.vector_search;
+
+ // 0. Distance metric. It decides which score transform the reader has to
undo
+ // (see _convert_score_to_distance), so defaulting would silently return
wrong
+ // distances. __isset was checked by _validate_external_search_request; the
+ // value is rejected here if unsupported.
+ ScoreTransform transform;
+ RETURN_IF_ERROR(parse_score_transform(vector.metric, &transform));
+ _score_transform = transform;
+
+ // 1. The per-split payload is the serialized BucketVectorSearchSplit (raw
+ // thrift binary, NOT base64; magic "PKVSPLIT", v1). Grab it first so we
fail
+ // fast before any filesystem IO.
+ const auto& payload =
_current_range.table_format_params.paimon_params.vector_payload;
+ if (payload.payload_type != TPaimonVectorPayloadType::BUCKET_SPLIT_BYTES) {
+ return Status::InternalError("paimon-rust vector search unsupported
payload type: {}",
+ static_cast<int>(payload.payload_type));
+ }
+ const std::string& split_bytes = payload.bytes;
+ if (split_bytes.empty()) {
+ return Status::InternalError("paimon-rust vector search split payload
is empty");
+ }
+
+ // 2. Resolve and open (or reuse) the table (same as the normal read path);
+ // options carry storage credentials to the FileIO.
+ RETURN_IF_ERROR(_open_paimon_table(range));
+
+ // 3. Build and configure the vector-search builder. No read.batch-size
here,
+ // unlike the batch path (_open_split_reader): paimon-rust's PK vector
readers
+ // construct their DataFileReaders without with_batch_size, so a
+ // read.batch-size option — from the schema, t@options or these search
+ // options — is silently ignored on the vector path, which batches at the
+ // parquet-rust default. Known difference; the vector stages are
+ // single-column streamed into bounded Top-K heaps and the output is
+ // capped by top_k, so no action needed until paimon-rust wires it.
+ paimon_result_vector_search_builder vb_res =
+ paimon_table_new_vector_search_builder(_handles->table.get());
+ if (vb_res.error != nullptr) {
+ return Status::InternalError("paimon-rust new vector search builder
failed: {}",
+ consume_error(vb_res.error));
+ }
+ vector_search_builder_ptr vb(vb_res.builder);
+
+ // 3a. Target vector column.
+ if (paimon_error* err =
+ paimon_vector_search_builder_with_vector_column(vb.get(),
vector.column.c_str())) {
+ return Status::InternalError("paimon-rust set vector column failed:
{}",
+ consume_error(err));
+ }
+ // 3b. Query vector. The wire form is a TSearchVector carrying a FLOAT32,
+ // little-endian binary blob; decode it into the f32 array the C API wants.
+ // v1 only supports FLOAT32 (paimon-rust's index backends operate on f32);
any
+ // other element type was already rejected by validate. Dimension and
+ // byte-size were validated there too, so decode unchecked here.
+ const auto& search_vec = vector.query_vector;
+ const auto dimension = static_cast<size_t>(search_vec.dimension);
+ std::vector<float> query_vector(dimension);
+ const auto* bytes = search_vec.values.data();
+ for (size_t i = 0; i < dimension; ++i) {
+ query_vector[i] = std::bit_cast<float>(LittleEndian::Load32(
+ reinterpret_cast<const uint8_t*>(bytes + i *
sizeof(uint32_t))));
+ }
+ if (paimon_error* err = paimon_vector_search_builder_with_query_vector(
+ vb.get(), query_vector.data(), query_vector.size())) {
+ return Status::InternalError("paimon-rust set query vector failed:
{}", consume_error(err));
+ }
+ // 3c. Retrieval limit (user_limit + user_offset, folded on FE). A negative
+ // limit would wrap when cast to uintptr_t, so guard it.
+ if (vector.top_k < 0) {
+ return Status::InternalError("paimon-rust vector search limit is
negative: {}",
+ vector.top_k);
+ }
+ if (paimon_error* err = paimon_vector_search_builder_with_limit(
+ vb.get(), static_cast<uintptr_t>(vector.top_k))) {
+ return Status::InternalError("paimon-rust set limit failed: {}",
consume_error(err));
+ }
+ // 3d. Index/search options (refine_factor, ivf.nprobe, "<algo>.metric",
...).
+ if (request.__isset.paimon_options &&
request.paimon_options.__isset.options &&
+ !request.paimon_options.options.empty()) {
+ std::vector<paimon_option> c_search_options;
+ c_search_options.reserve(request.paimon_options.options.size());
+ for (const auto& kv : request.paimon_options.options) {
+ c_search_options.emplace_back(kv.first.c_str(), kv.second.c_str());
+ }
+ if (paimon_error* err = paimon_vector_search_builder_with_options(
+ vb.get(), c_search_options.data(),
c_search_options.size())) {
+ return Status::InternalError("paimon-rust set search options
failed: {}",
+ consume_error(err));
+ }
+ }
+ // 3e. Materialized user columns. The score field is always appended by
+ // paimon-rust itself and must NOT be listed here. Empty projection = only
the
+ // score column (index-only scan). Log the user columns: an empty list is
+ // legitimate, so it is logged as such rather than skipped.
+ auto read_columns = _build_read_columns();
+ std::vector<const char*> projection;
+ projection.reserve(read_columns.size() + 1);
+ for (const auto& col : read_columns) {
+ projection.push_back(col.c_str());
+ }
+ projection.push_back(nullptr);
+ LOG(INFO) << "paimon-rust vector search: projection=[" <<
join(read_columns, ", ") << "] ("
+ << read_columns.size() << " column(s), distance column '"
+ << kPaimonSearchDistanceColumn << "' produced by the reader)";
+ if (paimon_error* err =
+ paimon_vector_search_builder_with_projection(vb.get(),
projection.data())) {
+ return Status::InternalError("paimon-rust set vector projection
failed: {}",
+ consume_error(err));
+ }
+
+ // 3f. Optional residual scalar filter from the FE data-half push-down. The
+ // FE-side DV/merge-on-read interception is intentionally not done (see
+ // PushDownVectorTopNIntoPaimonScan): the Rust candidate search stage
applies
+ // the filter together with deletion vectors, and a table shape Rust cannot
+ // handle correctly is Rust's responsibility to reject.
+ if (!_conjuncts.empty()) {
+ std::vector<std::string> names;
+ std::vector<DataTypePtr> types;
+ names.reserve(_projected_columns.size());
+ types.reserve(_projected_columns.size());
+ for (const auto& col : _projected_columns) {
+ if (col.is_partition_key || col.name ==
kPaimonSearchDistanceColumn) {
+ continue;
+ }
+ names.push_back(col.name);
+ types.push_back(col.type);
+ }
+ PaimonRustPredicateConverter converter(names, types,
_handles->table.get());
+ paimon_predicate* predicate = converter.build(_conjuncts);
+ if (predicate != nullptr) {
+ // with_filter consumes the predicate on every path.
+ if (paimon_error* err =
paimon_vector_search_builder_with_filter(vb.get(), predicate)) {
+ return Status::InternalError("paimon-rust set vector filter
failed: {}",
+ consume_error(err));
+ }
+ LOG(INFO) << "paimon-rust vector search: applied residual filter";
+ }
+ }
+
+ // 4. Execute the search over the single FE-planned bucket split, via the
+ // eager Top-K chain (paimon-rust #822). Search, refine, local Top-K and
+ // materialization all happen in paimon_vector_read_read; the returned
arrow
+ // stream is best-first and carries the __paimon_search_score column. The
+ // intermediate handles are locals: the migration contract (see
+ // doris-vector-search-api-migration.md) guarantees the record batch reader
+ // is independent once paimon_vector_read_read returns, so they are
released
+ // when this function exits.
+ //
+ // split: the bytes only need to live until deserialize returns.
+ // scan: only needs to live until the plan is constructed.
+ // builder: only needs to live until scan and read are created.
+ // read/plan: only need to live until paimon_vector_read_read returns.
+ paimon_result_bucket_vector_search_split split_result =
+ paimon_bucket_vector_search_split_deserialize(
+ reinterpret_cast<const uint8_t*>(split_bytes.data()),
split_bytes.size());
+ if (split_result.error != nullptr) {
+ return Status::InternalError("paimon-rust deserialize vector split
failed: {}",
+ consume_error(split_result.error));
+ }
+ bucket_vector_search_split_ptr split(split_result.split);
+
+ paimon_result_vector_scan scan_result =
paimon_vector_search_builder_new_scan(vb.get());
+ if (scan_result.error != nullptr) {
+ return Status::InternalError("paimon-rust new vector scan failed: {}",
+ consume_error(scan_result.error));
+ }
+ vector_scan_ptr scan(scan_result.scan);
+
+ paimon_result_vector_read read_result =
paimon_vector_search_builder_new_read(vb.get());
+ if (read_result.error != nullptr) {
+ return Status::InternalError("paimon-rust new vector read failed: {}",
+ consume_error(read_result.error));
+ }
+ vector_read_ptr read(read_result.read);
+
+ // Must use plan_from_bucket_splits: it plans from the FE-assigned split,
+ // whereas paimon_vector_scan_plan / execute_read replan from the table and
+ // would ignore the split this scanner was given.
+ const paimon_bucket_vector_search_split* splits[] = {split.get()};
+ paimon_result_vector_plan plan_result =
+ paimon_vector_scan_plan_from_bucket_splits(scan.get(), splits, 1);
+ if (plan_result.error != nullptr) {
+ return Status::InternalError("paimon-rust build vector plan failed:
{}",
+ consume_error(plan_result.error));
+ }
+ vector_plan_ptr plan(plan_result.plan);
+
+ paimon_result_record_batch_reader rdr_res =
paimon_vector_read_read(read.get(), plan.get());
Review Comment:
[P1] Bound and account eager vector materialization
This call completes search and materialization before returning. The pinned
Rust
[materialize_positions](https://github.com/apache/paimon-rust/blob/v0.4.0-rc1/crates/paimon/src/table/pk_vector_read.rs#L156-L183)
retains every projected Top-K batch and then interleaves them into one
reordered batch while the inputs remain alive. FE forwards the unrestricted
LIMIT+OFFSET as K: with a million matching rows in one bucket and 16KiB
payloads, `SELECT id,payload FROM t ORDER BY l2_distance_approximate(vec,q)
LIMIT 1 OFFSET 999999` retains over 15GiB of payload before the first block,
plus the reordered copy. These Rust Vec/Arrow allocations bypass Doris's
tracked allocator, and this path neither reserves memory nor honors the
adaptive batch size. A small query mem_limit therefore cannot reject the
allocation before it threatens BE process memory. Add tracked admission/bounded
materialization, or decline such rewrites before invoking this eager API.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/property/metastore/AbstractPaimonProperties.java:
##########
@@ -123,6 +124,34 @@ protected void appendUserHadoopConfig(Configuration conf)
{
normalizeS3Config().forEach(conf::set);
}
+ /**
+ * Mirrors the S3 credentials Doris assembled into the Hadoop configuration
+ * onto the catalog options, so the pinned Paimon 2.1 SDK sees them.
+ *
+ * <p>Doris stores a catalog's storage credentials (fs.s3a.access.key,
fs.s3a.secret.key,
+ * fs.s3a.endpoint, ...) only in the Hadoop Configuration handed to
+ * CatalogContext.create(options, conf) — the paimon 1.4 FileIO consulted
exactly
+ * that conf. The pinned 2.1 S3FileIO instead reads its credentials
exclusively from
+ * CatalogContext.options() (S3FileIO#configure ->
loadHadoopConfigFromContext copies
+ * the s3./s3a./fs.s3a.-prefixed option entries and never looks at the
conf), so
+ * without this mirror every s3-warehouse catalog initializes the SDK with
no
+ * credentials and the first file operation fails on the ambient AWS
provider chain
+ * ("Unable to load credentials from system settings"). Mirroring the
whole fs.s3a.*
+ * set keeps user extras (path style, connection limits) flowing as well;
entries
+ * outside the prefix are not the SDK's concern.
+ */
+ protected void mirrorHadoopS3OptionsToCatalogOptions(Configuration conf) {
+ if (conf == null || catalogOptions == null) {
+ return;
+ }
+ conf.forEach(entry -> {
+ String key = entry.getKey();
+ if (key.startsWith(HADOOP_S3A_PREFIX) &&
StringUtils.isNotBlank(entry.getValue())) {
+ catalogOptions.set(key, entry.getValue());
Review Comment:
[P2] Preserve explicit S3 options when mirroring Hadoop defaults
An existing catalog can specify static `s3.access_key`/`s3.secret_key` plus
`paimon.fs.s3a.aws.credentials.provider=org.apache.hadoop.fs.s3a.auth.AssumedRoleCredentialProvider`
and its role options. `appendCatalogOptions` preserves that explicit provider,
but `userStoragePrefixes` omits `paimon.fs.s3a.`, so the generated Hadoop
configuration still carries SimpleAWSCredentialsProvider from the static S3
keys. This unconditional set replaces the requested role provider with Simple.
Paimon's S3FileIO honors these catalog options in both 1.4.2 and 2.1.0, so a
previously working warehouse restricted to the role now opens as the base
identity and fails AccessDenied. Fill only absent catalog options, or apply
explicit user options after the Hadoop defaults; this helper affects
filesystem, HMS and JDBC catalogs.
--
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]
