Compare commits

..

3 Commits

Author SHA1 Message Date
Gatefixer 4cb678c746 docs(node): regenerate RRF reranker reference 2026-08-06 00:24:06 +00:00
Gatefixer 3ee81fa556 fix(node): preserve vector rerank query contracts 2026-08-06 00:16:39 +00:00
Gatefixer 48ca05c7d5 fix(node): rerank vector search results 2026-08-05 23:34:58 +00:00
13 changed files with 414 additions and 150 deletions
@@ -10,6 +10,24 @@ Reranks the results using the Reciprocal Rank Fusion (RRF) algorithm.
## Methods
### outputSchema()
```ts
outputSchema(inputSchema): Promise<Schema<any>>
```
Declare the RRF output schema for vector-only query execution.
#### Parameters
* **inputSchema**: `Schema`&lt;`any`&gt;
#### Returns
`Promise`&lt;`Schema`&lt;`any`&gt;&gt;
***
### rerankHybrid()
```ts
@@ -8,6 +8,27 @@
## Methods
### outputSchema()?
```ts
optional outputSchema(inputSchema): Promise<Schema<any>>
```
Declare the schema returned when reranking a vector-only query.
This is required for vector-only reranking so query schema introspection
and execution agree. Hybrid-only rerankers may omit it.
#### Parameters
* **inputSchema**: `Schema`&lt;`any`&gt;
#### Returns
`Promise`&lt;`Schema`&lt;`any`&gt;&gt;
***
### rerankHybrid()
```ts
+16
View File
@@ -79,6 +79,22 @@ describe("rerankers", function () {
expect(result).toHaveLength(2);
});
it("returns relevance scores when reranking a vector search", async function () {
const query = table
.vectorSearch([0.1, 0.1])
.limit(2)
.rerank(await RRFReranker.create());
const schema = await query.outputSchema();
const result = await query.toArray();
expect(schema.fields.map((field) => field.name)).toContain(
"_relevance_score",
);
expect(result).toHaveLength(2);
expect(result[0]._relevance_score).toBeCloseTo(1 / 60);
expect(result[1]._relevance_score).toBeCloseTo(1 / 61);
});
it("does not keep process alive after rerank query", async function () {
const script = `
import * as lancedb from "./dist/index.js";
+25 -13
View File
@@ -5,9 +5,11 @@ import {
Table as ArrowTable,
type IntoVector,
RecordBatch,
createEmptyTable,
extractVectorBuffer,
fromBufferToRecordBatch,
fromRecordBatchToBuffer,
fromTableToBuffer,
tableFromIPC,
} from "./arrow";
import { type IvfPqOptions } from "./indices";
@@ -744,20 +746,30 @@ export class VectorQuery extends StandardQueryBase<NativeVectorQuery> {
}
rerank(reranker: Reranker): VectorQuery {
super.doCall((inner) =>
inner.rerank(async (args) => {
const vecResults = await fromBufferToRecordBatch(args.vecResults);
const ftsResults = await fromBufferToRecordBatch(args.ftsResults);
const result = await reranker.rerankHybrid(
args.query,
vecResults as RecordBatch,
ftsResults as RecordBatch,
);
super.doCall((inner) => {
const outputSchema = reranker.outputSchema?.bind(reranker);
inner.rerank(
async (args) => {
const vecResults = await fromBufferToRecordBatch(args.vecResults);
const ftsResults = await fromBufferToRecordBatch(args.ftsResults);
const result = await reranker.rerankHybrid(
args.query,
vecResults as RecordBatch,
ftsResults as RecordBatch,
);
const buffer = fromRecordBatchToBuffer(result);
return buffer;
}),
);
const buffer = fromRecordBatchToBuffer(result);
return buffer;
},
outputSchema
? async (args) => {
const inputSchema = tableFromIPC(args.inputSchema).schema;
const result = await outputSchema(inputSchema);
return fromTableToBuffer(createEmptyTable(result));
}
: undefined,
);
});
return this;
}
+12 -4
View File
@@ -1,14 +1,22 @@
// SPDX-License-Identifier: Apache-2.0
// SPDX-FileCopyrightText: Copyright The LanceDB Authors
import { RecordBatch } from "apache-arrow";
import { RecordBatch, Schema } from "apache-arrow";
export * from "./rrf";
// Interface for a reranker. A reranker is used to rerank the results from a
// vector and FTS search. This is useful for combining the results from both
// search methods.
// Interface for a reranker. A reranker is used to rerank vector and hybrid
// search results. For vector-only searches, query is empty and ftsResults is an
// empty batch with the same schema as vecResults.
export interface Reranker {
/**
* Declare the schema returned when reranking a vector-only query.
*
* This is required for vector-only reranking so query schema introspection
* and execution agree. Hybrid-only rerankers may omit it.
*/
outputSchema?(inputSchema: Schema): Promise<Schema>;
rerankHybrid(
query: string,
vecResults: RecordBatch,
+12 -1
View File
@@ -1,7 +1,7 @@
// SPDX-License-Identifier: Apache-2.0
// SPDX-FileCopyrightText: Copyright The LanceDB Authors
import { RecordBatch } from "apache-arrow";
import { Field, Float32, RecordBatch, Schema } from "apache-arrow";
import { fromBufferToRecordBatch, fromRecordBatchToBuffer } from "../arrow";
import { RrfReranker as NativeRRFReranker } from "../native";
@@ -24,6 +24,17 @@ export class RRFReranker {
);
}
/** Declare the RRF output schema for vector-only query execution. */
async outputSchema(inputSchema: Schema): Promise<Schema> {
return new Schema(
[
...inputSchema.fields,
new Field("_relevance_score", new Float32(), false),
],
inputSchema.metadata,
);
}
async rerankHybrid(
query: string,
vecResults: RecordBatch,
+3 -2
View File
@@ -6,8 +6,8 @@ use std::sync::Arc;
use crate::error::NapiErrorExt;
use crate::error::convert_error;
use crate::iterator::RecordBatchIterator;
use crate::rerankers::RerankHybridCallbackArgs;
use crate::rerankers::Reranker;
use crate::rerankers::{RerankHybridCallbackArgs, RerankOutputSchemaCallbackArgs};
use crate::util::{parse_distance_type, schema_to_buffer};
use arrow_array::{
Array, Float16Array as ArrowFloat16Array, Float32Array as ArrowFloat32Array,
@@ -388,8 +388,9 @@ impl VectorQuery {
pub fn rerank(
&mut self,
rerank_hybrid: Function<RerankHybridCallbackArgs, Promise<Buffer>>,
output_schema: Option<Function<RerankOutputSchemaCallbackArgs, Promise<Buffer>>>,
) -> napi::Result<()> {
let reranker = Reranker::new(rerank_hybrid)?;
let reranker = Reranker::new(rerank_hybrid, output_schema)?;
self.inner = self.inner.clone().rerank(Arc::new(reranker));
Ok(())
}
+51 -2
View File
@@ -6,7 +6,7 @@ use async_trait::async_trait;
use napi::{bindgen_prelude::*, threadsafe_function::ThreadsafeFunction};
use napi_derive::napi;
use lancedb::ipc::batches_to_ipc_file;
use lancedb::ipc::{batches_to_ipc_file, ipc_file_to_schema, schema_to_ipc_file};
use lancedb::rerankers::Reranker as LanceDBReranker;
use lancedb::{error::Error, ipc::ipc_file_to_batches};
@@ -21,28 +21,72 @@ type RerankHybridFn = ThreadsafeFunction<
true,
>;
type RerankOutputSchemaFn = ThreadsafeFunction<
RerankOutputSchemaCallbackArgs,
Promise<Buffer>,
RerankOutputSchemaCallbackArgs,
Status,
false,
true,
>;
/// Reranker implementation that "wraps" a NodeJS Reranker implementation.
/// This contains references to the callbacks that can be used to invoke the
/// reranking methods on the NodeJS implementation and handles serializing the
/// record batches to Arrow IPC buffers.
pub struct Reranker {
rerank_hybrid: RerankHybridFn,
output_schema: Option<RerankOutputSchemaFn>,
}
impl Reranker {
pub fn new(
rerank_hybrid: Function<RerankHybridCallbackArgs, Promise<Buffer>>,
output_schema: Option<Function<RerankOutputSchemaCallbackArgs, Promise<Buffer>>>,
) -> napi::Result<Self> {
let rerank_hybrid = rerank_hybrid
.build_threadsafe_function()
.weak::<true>()
.build()?;
Ok(Self { rerank_hybrid })
let output_schema = output_schema
.map(|output_schema| {
output_schema
.build_threadsafe_function()
.weak::<true>()
.build()
})
.transpose()?;
Ok(Self {
rerank_hybrid,
output_schema,
})
}
}
#[async_trait]
impl lancedb::rerankers::Reranker for Reranker {
async fn output_schema(
&self,
input: &arrow_schema::SchemaRef,
) -> lancedb::error::Result<arrow_schema::SchemaRef> {
let output_schema = self.output_schema.as_ref().ok_or(Error::NotSupported {
message: "vector rerankers must declare their output schema".to_string(),
})?;
let callback_args = RerankOutputSchemaCallbackArgs {
input_schema: Buffer::from(schema_to_ipc_file(input.as_ref())?),
};
let promised_buffer: Promise<Buffer> = output_schema
.call_async(callback_args)
.await
.map_err(|e| Error::Runtime {
message: format!("napi error status={}, reason={}", e.status, e.reason),
})?;
let buffer = promised_buffer.await.map_err(|e| Error::Runtime {
message: format!("napi error status={}, reason={}", e.status, e.reason),
})?;
ipc_file_to_schema(buffer.to_vec())
}
async fn rerank_hybrid(
&self,
query: &str,
@@ -86,6 +130,11 @@ pub struct RerankHybridCallbackArgs {
pub fts_results: Buffer,
}
#[napi(object)]
pub struct RerankOutputSchemaCallbackArgs {
pub input_schema: Buffer,
}
fn buffer_to_record_batch(buffer: Buffer) -> Result<RecordBatch> {
let mut reader = ipc_file_to_batches(buffer.to_vec()).default_error()?;
reader
+3 -56
View File
@@ -3,7 +3,6 @@
from typing import List
from urllib.parse import unquote, urlparse
import numpy as np
@@ -126,20 +125,9 @@ class InstructorEmbeddingFunction(TextEmbeddingFunction):
@weak_lru(maxsize=1)
def get_model(self):
huggingface_hub = attempt_import_or_raise("huggingface_hub", "huggingface-hub")
missing = object()
original_cached_download = getattr(huggingface_hub, "cached_download", missing)
if original_cached_download is missing:
huggingface_hub.cached_download = _cached_download(huggingface_hub)
try:
instructor_embedding = attempt_import_or_raise(
"InstructorEmbedding", "InstructorEmbedding"
)
finally:
if original_cached_download is missing:
del huggingface_hub.cached_download
instructor_embedding = attempt_import_or_raise(
"InstructorEmbedding", "InstructorEmbedding"
)
torch = attempt_import_or_raise("torch", "torch")
model = instructor_embedding.INSTRUCTOR(self.name)
@@ -152,44 +140,3 @@ class InstructorEmbeddingFunction(TextEmbeddingFunction):
model, {torch.nn.Linear}, dtype=torch.qint8
)
return model
def _cached_download(huggingface_hub):
"""Provide the legacy download API used by sentence-transformers 2.2.x."""
def cached_download(
*,
url,
cache_dir=None,
force_filename=None,
library_name=None,
library_version=None,
user_agent=None,
use_auth_token=None,
**_,
):
path = urlparse(url).path.lstrip("/")
try:
repo_id, resolved_path = path.split("/resolve/", maxsplit=1)
revision, filename = resolved_path.split("/", maxsplit=1)
except ValueError as err:
raise ValueError(f"Unsupported Hugging Face Hub URL: {url}") from err
repo_id = unquote(repo_id)
revision = unquote(revision)
filename = unquote(filename)
# sentence-transformers derives force_filename from this Hub path with
# os.path.join. Using the URL path beneath local_dir produces the same
# local destination without sending Windows separators to the Hub.
return huggingface_hub.hf_hub_download(
repo_id=repo_id,
filename=filename,
revision=revision,
local_dir=cache_dir,
library_name=library_name,
library_version=library_version,
user_agent=user_agent,
token=use_auth_token,
)
return cached_download
-56
View File
@@ -1,11 +1,8 @@
# SPDX-License-Identifier: Apache-2.0
# SPDX-FileCopyrightText: Copyright The LanceDB Authors
import ntpath
import os
import pickle
import sys
from types import ModuleType
from typing import List, Optional, Union
from unittest.mock import MagicMock, patch
@@ -525,59 +522,6 @@ def test_embedding_function_safe_model_dump(embedding_type):
)
def test_instructor_embedding_supports_huggingface_hub_without_cached_download(
tmp_path, monkeypatch
):
from lancedb.embeddings.instructor import InstructorEmbeddingFunction
hub_download = MagicMock(return_value="/cache/1_Pooling/config.json")
huggingface_hub = ModuleType("huggingface_hub")
huggingface_hub.hf_hub_download = hub_download
torch = ModuleType("torch")
monkeypatch.setitem(sys.modules, "huggingface_hub", huggingface_hub)
monkeypatch.setitem(sys.modules, "torch", torch)
monkeypatch.delitem(sys.modules, "InstructorEmbedding", raising=False)
monkeypatch.syspath_prepend(str(tmp_path))
(tmp_path / "InstructorEmbedding.py").write_text(
"from huggingface_hub import cached_download\n\n"
"class INSTRUCTOR:\n"
" def __init__(self, name):\n"
" self.name = name\n"
)
embedding = InstructorEmbeddingFunction.create(show_progress_bar=False)
instructor_model = embedding.get_model()
assert instructor_model.name == "hkunlp/instructor-base"
assert not hasattr(huggingface_hub, "cached_download")
instructor_embedding = sys.modules["InstructorEmbedding"]
path = instructor_embedding.cached_download(
url=(
"https://huggingface.co/hkunlp/instructor-base/resolve/abc123/"
"1_Pooling/config.json"
),
cache_dir="/cache",
force_filename=ntpath.join("1_Pooling", "config.json"),
library_name="sentence-transformers",
library_version="2.2.2",
use_auth_token="token",
)
assert path == "/cache/1_Pooling/config.json"
hub_download.assert_called_once_with(
repo_id="hkunlp/instructor-base",
filename="1_Pooling/config.json",
revision="abc123",
local_dir="/cache",
library_name="sentence-transformers",
library_version="2.2.2",
user_agent=None,
token="token",
)
@patch("time.sleep")
def test_retry(mock_sleep):
test_function = MagicMock(side_effect=[Exception] * 9 + ["result"])
+219 -2
View File
@@ -511,7 +511,9 @@ pub trait QueryBase {
/// Rerank the results using the specified reranker.
///
/// This is currently only supported for Hybrid Search.
/// For vector-only searches, the reranker receives the vector results and an
/// empty full-text result set and must declare its output schema. Reranking
/// multiple query vectors in one query is not supported.
fn rerank(self, reranker: Arc<dyn Reranker>) -> Self;
/// The method to normalize the scores. Can be "rank" or "Score". If "Rank",
@@ -1138,6 +1140,44 @@ pub struct VectorQuery {
}
impl VectorQuery {
fn check_vector_rerank_supported(&self) -> Result<()> {
if self.request.query_vector.len() > 1 {
return Err(Error::NotSupported {
message: "reranking multiple query vectors is not supported; execute one query per vector"
.to_string(),
});
}
Ok(())
}
async fn vector_rerank_output_schema(&self) -> Result<SchemaRef> {
self.check_vector_rerank_supported()?;
// Rerankers receive row IDs internally. Apply their schema transform to
// that exact input and then hide the row ID from the declared public
// schema unless it was explicitly requested.
let vector_query = self.clone().with_row_id();
let plan = vector_query
.create_plan(QueryExecutionOptions::default())
.await?;
let reranker = self
.request
.base
.reranker
.as_ref()
.expect("vector_rerank_output_schema requires a reranker");
let input_schema = plan.schema();
let output_schema = reranker.output_schema(&input_schema).await?;
if self.request.base.with_row_id {
Ok(output_schema)
} else {
Ok(RecordBatch::new_empty(output_schema)
.drop_column(ROW_ID)?
.schema())
}
}
fn new(base: Query) -> Self {
Self {
parent: base.parent,
@@ -1443,6 +1483,61 @@ impl VectorQuery {
Ok(single_batch_stream(results, max_batch_length))
}
async fn execute_vector_rerank(
&self,
options: QueryExecutionOptions,
) -> Result<SendableRecordBatchStream> {
self.check_vector_rerank_supported()?;
let max_batch_length = options.max_batch_length as usize;
let internal_options = options.without_output_batch_length_limit();
// RRF needs row IDs to assign and preserve scores. Keep them internal unless
// the caller explicitly requested them.
let vector_query = self.clone().with_row_id();
let vector_results = vector_query
.inner_execute_with_options(internal_options)
.await?;
let schema = vector_results.schema();
let vector_results = vector_results.try_collect::<Vec<_>>().await?;
let vector_results = concat_batches(&schema, vector_results.iter())?;
let vector_schema = vector_results.schema();
let fts_results = RecordBatch::new_empty(vector_schema.clone());
let reranker = self
.request
.base
.reranker
.as_ref()
.expect("execute_vector_rerank requires a reranker");
let expected_schema = reranker.output_schema(&vector_schema).await?;
let mut results = reranker
.rerank_hybrid("", vector_results, fts_results)
.await?;
check_reranker_result(&results)?;
if results.schema() != expected_schema {
return Err(Error::Schema {
message: format!(
"reranker returned schema {:?}, but declared {:?}",
results.schema(),
expected_schema
),
});
}
let limit = self.request.base.limit.unwrap_or(DEFAULT_TOP_K);
if results.num_rows() > limit {
results = results.slice(0, limit);
}
if !self.request.base.with_row_id {
results = results.drop_column(ROW_ID)?;
}
Ok(single_batch_stream(results, max_batch_length))
}
async fn inner_execute_with_options(
&self,
options: QueryExecutionOptions,
@@ -1495,6 +1590,23 @@ impl ExecutableQuery for VectorQuery {
return Ok(hybrid_result);
}
if self.request.base.reranker.is_some() {
let timeout = options.timeout;
let mut rerank_options = options;
// A single outer deadline covers planning, candidate collection,
// schema declaration, and the complete reranker callback.
rerank_options.timeout = None;
let execution = self.execute_vector_rerank(rerank_options);
return match timeout {
Some(timeout) => tokio::time::timeout(timeout, execution)
.await
.map_err(|_| Error::Timeout {
message: format!("Query timeout after {} ms", timeout.as_millis()),
})?,
None => execution.await,
};
}
self.inner_execute_with_options(options).await
}
@@ -1507,6 +1619,15 @@ impl ExecutableQuery for VectorQuery {
let query = AnyQuery::VectorQuery(self.request.clone());
self.parent.analyze_plan(&query, options).await
}
async fn output_schema(&self) -> Result<SchemaRef> {
if self.request.base.full_text_search.is_none() && self.request.base.reranker.is_some() {
self.vector_rerank_output_schema().await
} else {
let plan = self.create_plan(QueryExecutionOptions::default()).await?;
Ok(plan.schema())
}
}
}
impl HasQuery for VectorQuery {
@@ -1643,7 +1764,13 @@ impl ExecutableQuery for TakeQuery {
#[cfg(test)]
mod tests {
use std::{collections::HashSet, sync::Arc};
use std::{
collections::HashSet,
sync::{
Arc,
atomic::{AtomicBool, Ordering},
},
};
use super::*;
use arrow::{array::downcast_array, compute::concat_batches, datatypes::Int32Type};
@@ -1659,6 +1786,31 @@ mod tests {
use crate::{Table, connect, database::CreateTableMode, index::Index};
#[derive(Debug)]
struct SlowReranker {
invoked: Arc<AtomicBool>,
}
#[async_trait::async_trait]
impl Reranker for SlowReranker {
async fn output_schema(&self, input: &SchemaRef) -> Result<SchemaRef> {
RRFReranker::default().output_schema(input).await
}
async fn rerank_hybrid(
&self,
query: &str,
vector_results: RecordBatch,
fts_results: RecordBatch,
) -> Result<RecordBatch> {
self.invoked.store(true, Ordering::SeqCst);
tokio::time::sleep(Duration::from_secs(2)).await;
RRFReranker::default()
.rerank_hybrid(query, vector_results, fts_results)
.await
}
}
#[tokio::test]
async fn test_setters_getters() {
// TODO: Switch back to memory://foo after https://github.com/lancedb/lancedb/issues/1051
@@ -2370,6 +2522,71 @@ mod tests {
// We don't guarantee order.
assert!(query_index.values().contains(&0));
assert!(query_index.values().contains(&1));
let reranked = query.rerank(Arc::new(RRFReranker::default()));
let Err(execute_error) = reranked.execute().await else {
panic!("multi-vector reranking should be rejected");
};
assert!(
execute_error
.to_string()
.contains("reranking multiple query vectors is not supported")
);
let schema_error = reranked.output_schema().await.unwrap_err();
assert!(
schema_error
.to_string()
.contains("reranking multiple query vectors is not supported")
);
}
#[tokio::test]
async fn test_vector_rerank_timeout_covers_reranker() {
let tmp_dir = tempdir().unwrap();
let table = make_test_table(&tmp_dir).await;
let invoked = Arc::new(AtomicBool::new(false));
let reranker = SlowReranker {
invoked: invoked.clone(),
};
let result = table
.vector_search(&[0.1, 0.2, 0.3, 0.4])
.unwrap()
.limit(1)
.rerank(Arc::new(reranker))
.execute_with_options(QueryExecutionOptions {
timeout: Some(Duration::from_secs(1)),
..Default::default()
})
.await;
assert!(invoked.load(Ordering::SeqCst));
assert!(matches!(result, Err(Error::Timeout { .. })));
}
#[tokio::test]
async fn test_vector_rerank_output_schema_matches_execution() {
let tmp_dir = tempdir().unwrap();
let table = make_test_table(&tmp_dir).await;
let query = table
.vector_search(&[0.1, 0.2, 0.3, 0.4])
.unwrap()
.limit(1)
.rerank(Arc::new(RRFReranker::default()));
let promised = query.output_schema().await.unwrap();
let actual = query
.execute()
.await
.unwrap()
.next()
.await
.unwrap()
.unwrap()
.schema();
assert_eq!(promised, actual);
assert!(promised.column_with_name("_relevance_score").is_some());
}
#[tokio::test]
+18 -5
View File
@@ -8,6 +8,7 @@ use arrow::{
compute::{concat_batches, filter_record_batch},
};
use arrow_array::{BooleanArray, RecordBatch, UInt64Array};
use arrow_schema::SchemaRef;
use async_trait::async_trait;
use lance::dataset::ROW_ID;
@@ -47,16 +48,28 @@ impl std::fmt::Display for NormalizeMethod {
}
}
/// Interface for a reranker. A reranker is used to rerank the results from a
/// vector and FTS search. This is useful for combining the results from both
/// search methods.
/// Interface for a reranker. A reranker is used to rerank vector and hybrid
/// search results. This is useful for combining results from multiple search
/// methods or assigning a relevance score to vector search results.
#[async_trait]
pub trait Reranker: std::fmt::Debug + Sync + Send {
// TODO support vector reranking and FTS reranking. Currently only hybrid reranking is supported.
/// Declare the schema returned by [`Self::rerank_hybrid`] for a vector-only
/// query.
///
/// Vector reranking validates the returned batch against this schema so
/// [`crate::query::ExecutableQuery::output_schema`] and execution cannot
/// disagree. Rerankers that only support hybrid search do not need to
/// implement this method.
async fn output_schema(&self, _input: &SchemaRef) -> Result<SchemaRef> {
Err(Error::NotSupported {
message: "vector rerankers must declare their output schema".to_string(),
})
}
/// Rerank function receives the individual results from the vector and FTS search
/// results. You can choose to use any of the results to generate the final results,
/// allowing maximum flexibility.
/// allowing maximum flexibility. For a vector-only search, `query` is empty and
/// `fts_results` is an empty batch with the same schema as `vector_results`.
async fn rerank_hybrid(
&self,
query: &str,
+16 -9
View File
@@ -9,7 +9,7 @@ use arrow::{
compute::{sort_to_indices, take},
};
use arrow_array::{Float32Array, RecordBatch, UInt64Array};
use arrow_schema::{DataType, Field, Schema, SortOptions};
use arrow_schema::{DataType, Field, Schema, SchemaRef, SortOptions};
use async_trait::async_trait;
use lance::dataset::ROW_ID;
@@ -44,6 +44,19 @@ impl Default for RRFReranker {
#[async_trait]
impl Reranker for RRFReranker {
async fn output_schema(&self, input: &SchemaRef) -> Result<SchemaRef> {
let mut fields = input.fields().to_vec();
fields.push(Arc::new(Field::new(
RELEVANCE_SCORE,
DataType::Float32,
false,
)));
Ok(Arc::new(Schema::new_with_metadata(
fields,
input.metadata().clone(),
)))
}
async fn rerank_hybrid(
&self,
_query: &str,
@@ -135,15 +148,9 @@ impl Reranker for RRFReranker {
.collect();
// add relevance score to schema
let mut fields = combined_results.schema().fields().to_vec();
fields.push(Arc::new(Field::new(
RELEVANCE_SCORE,
DataType::Float32,
false,
)));
let schema = Schema::new(fields);
let schema = self.output_schema(&combined_results.schema()).await?;
let combined_results = RecordBatch::try_new(Arc::new(schema), columns)?;
let combined_results = RecordBatch::try_new(schema, columns)?;
Ok(combined_results)
}