Package org.elasticsearch.xpack.esql.enrich

Class AbstractLookupService<R extends org.elasticsearch.xpack.esql.enrich.AbstractLookupService.Request,T extends org.elasticsearch.xpack.esql.enrich.AbstractLookupService.TransportRequest>

java.lang.Object

org.elasticsearch.xpack.esql.enrich.AbstractLookupService<R,T>

Direct Known Subclasses:

EnrichLookupService, LookupFromIndexService

public abstract class AbstractLookupService<R extends org.elasticsearch.xpack.esql.enrich.AbstractLookupService.Request,T extends org.elasticsearch.xpack.esql.enrich.AbstractLookupService.TransportRequest>
extends Object

AbstractLookupService performs a LEFT JOIN for a given input page against another index that must have only a single shard.

This registers a TransportRequestHandler so we can handle requests to join data that isn't local to the node, but it is much faster if the data is already local.

The join process spawns a Driver per incoming page which runs in two, three or four stages:

Stage 1: Finding matching document IDs for the input page. This stage is done by the EnrichQuerySourceOperator. The output page of this stage is represented as [DocVector, IntBlock: positions of the input terms].

Stage 2: Extracting field values for the matched document IDs. The output page is represented as [DocVector, IntBlock: positions, Block: field1, Block: field2,...].

Stage 3: Optionally the BulkLookupMvFilterOperator removes false-positive multivalue matches when the BulkKeywordLookup optimization is active.

Stage 4: Optionally this combines the extracted values based on positions and filling nulls for positions without matches. This is done by MergePositionsOperator. The output page is represented as [Block: field1, Block: field2,...].

The Page.getPositionCount() of the output Page is equal to the Page.getPositionCount() of the input page. In other words - it returns the same number of rows that it was sent no matter how many documents match.

Nested Class Summary

Nested Classes

Modifier and Type	Class	Description
static final record	AbstractLookupService.LookupShardContext	AbstractLookupService uses this to power the queries and field loading that it needs to perform to actually do the lookup.
static interface	AbstractLookupService.LookupShardContextFactory	Create a AbstractLookupService.LookupShardContext for a locally allocated ShardId.

Field Summary

Fields

Modifier and Type	Field	Description
protected final BigArrays	bigArrays
protected final BlockFactory	blockFactory
protected final ClusterService	clusterService
protected final Executor	executor
protected final IndicesService	indicesService
protected final LocalCircuitBreaker.SizeSettings	localBreakerSettings
static final ReferenceAttribute	LOOKUP_POSITIONS_FIELD	Represents an integer column that corresponds to the positions of the page that we perform the lookup for.
protected final AbstractLookupService.LookupShardContextFactory	lookupShardContextFactory
protected final boolean	mergePages	Should output pages be combined into a single resulting page? If this is true we'll run a MergePositionsOperator to merge all output Pages into a single result, merging each found document into one row per input row, squashing the fields into multivalued fields.
protected final PlannerSettings.Holder	plannerSettings
protected final ProjectResolver	projectResolver
protected final TransportService	transportService

Method Summary

Modifier and Type	Method	Description
protected abstract org.elasticsearch.xpack.esql.enrich.AbstractLookupService.LookupResponse	createLookupResponse(List<Page> resultPages, BlockFactory blockFactory)	Build the response.
protected Page	createNullResponse(int positionCount, List<NamedExpression> extractFields)
protected void	doLookup(T request, CancellableTask task, ActionListener<org.elasticsearch.xpack.esql.enrich.AbstractLookupService.LookupResponse> listener)
static String	extractFieldName(NamedExpression extractField)
CircuitBreaker	getBreaker()
Executor	getExecutor()
ThreadContext	getThreadContext()
final void	lookupAsync(R request, DiscoveryNode targetNode, CancellableTask parentTask, ActionListener<org.elasticsearch.xpack.esql.enrich.AbstractLookupService.LookupResponse> outListener)	Perform the actual lookup using an explicit target node.
final void	lookupAsync(R request, CancellableTask parentTask, ActionListener<org.elasticsearch.xpack.esql.enrich.AbstractLookupService.LookupResponse> outListener)	Perform the actual lookup, returning the full AbstractLookupService.LookupResponse.
protected abstract LookupEnrichQueryGenerator	queryList(T request, SearchExecutionContext context, AliasFilter aliasFilter, Warnings warnings)	Build a list of queries to perform inside the actual lookup.
protected abstract org.elasticsearch.xpack.esql.enrich.AbstractLookupService.LookupResponse	readLookupResponse(StreamInput in, BlockFactory blockFactory)	Read the response from a StreamInput.
protected final void	respondWithPages(ActionListener<org.elasticsearch.xpack.esql.enrich.AbstractLookupService.LookupResponse> listener, List<Page> pages)	Helper to create a LookupResponse from pages and send it to the listener.
protected void	sendChildRequest(CancellableTask parentTask, ActionListener<org.elasticsearch.xpack.esql.enrich.AbstractLookupService.LookupResponse> delegate, DiscoveryNode targetNode, T transportRequest)
protected static QueryList	termQueryList(MappedFieldType field, AliasFilter aliasFilter, int channelOffset, DataType inputDataType)
protected abstract T	transportRequest(R request, ShardId shardId)	Convert a request as sent to lookupAsync(R, org.elasticsearch.tasks.CancellableTask, org.elasticsearch.action.ActionListener<org.elasticsearch.xpack.esql.enrich.AbstractLookupService.LookupResponse>) into a transport request after preflight checks have been performed.

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Field Details

LOOKUP_POSITIONS_FIELD

public static final ReferenceAttribute LOOKUP_POSITIONS_FIELD

Represents an integer column that corresponds to the positions of the page that we perform the lookup for.

clusterService

protected final ClusterService clusterService

indicesService

protected final IndicesService indicesService

lookupShardContextFactory

protected final AbstractLookupService.LookupShardContextFactory lookupShardContextFactory

transportService

protected final TransportService transportService

executor

protected final Executor executor

blockFactory

protected final BlockFactory blockFactory

bigArrays

protected final BigArrays bigArrays

localBreakerSettings

protected final LocalCircuitBreaker.SizeSettings localBreakerSettings

projectResolver

protected final ProjectResolver projectResolver

plannerSettings

protected final PlannerSettings.Holder plannerSettings

mergePages

protected final boolean mergePages

Should output pages be combined into a single resulting page? If this is true we'll run a MergePositionsOperator to merge all output Pages into a single result, merging each found document into one row per input row, squashing the fields into multivalued fields. If this is false then we'll skip this step, and it's up to the caller to figure out what to do with a List of resulting pages.

Method Details

getThreadContext

public ThreadContext getThreadContext()

transportRequest

protected abstract T transportRequest(R request,
 ShardId shardId)

Convert a request as sent to lookupAsync(R, org.elasticsearch.tasks.CancellableTask, org.elasticsearch.action.ActionListener<org.elasticsearch.xpack.esql.enrich.AbstractLookupService.LookupResponse>) into a transport request after preflight checks have been performed.

queryList

protected abstract LookupEnrichQueryGenerator queryList(T request,
 SearchExecutionContext context,
 AliasFilter aliasFilter,
 Warnings warnings)

Build a list of queries to perform inside the actual lookup.

createLookupResponse

protected abstract org.elasticsearch.xpack.esql.enrich.AbstractLookupService.LookupResponse createLookupResponse(List<Page> resultPages,
 BlockFactory blockFactory)

Build the response.

respondWithPages

protected final void respondWithPages(ActionListener<org.elasticsearch.xpack.esql.enrich.AbstractLookupService.LookupResponse> listener,
 List<Page> pages)

Helper to create a LookupResponse from pages and send it to the listener. The response is released after sending via ActionListener.respondAndRelease(org.elasticsearch.action.ActionListener<R>, R).

readLookupResponse

protected abstract org.elasticsearch.xpack.esql.enrich.AbstractLookupService.LookupResponse readLookupResponse(StreamInput in,
 BlockFactory blockFactory)
                                                                                                        throws IOException

Read the response from a StreamInput.

Throws:

IOException

termQueryList

protected static QueryList termQueryList(MappedFieldType field,
 AliasFilter aliasFilter,
 int channelOffset,
 DataType inputDataType)

lookupAsync

public final void lookupAsync(R request,
 CancellableTask parentTask,
 ActionListener<org.elasticsearch.xpack.esql.enrich.AbstractLookupService.LookupResponse> outListener)

Perform the actual lookup, returning the full AbstractLookupService.LookupResponse. Callers can access pages via AbstractLookupService.LookupResponse.takePages() and must call TransportMessage.decRef() when done. This method determines the target node internally via shard routing.

lookupAsync

public final void lookupAsync(R request,
 DiscoveryNode targetNode,
 CancellableTask parentTask,
 ActionListener<org.elasticsearch.xpack.esql.enrich.AbstractLookupService.LookupResponse> outListener)

Perform the actual lookup using an explicit target node. Use this overload when the caller has already determined which node should handle the request, to avoid inconsistent node selection when replicas exist.

sendChildRequest

protected void sendChildRequest(CancellableTask parentTask,
 ActionListener<org.elasticsearch.xpack.esql.enrich.AbstractLookupService.LookupResponse> delegate,
 DiscoveryNode targetNode,
 T transportRequest)

doLookup

protected void doLookup(T request,
 CancellableTask task,
 ActionListener<org.elasticsearch.xpack.esql.enrich.AbstractLookupService.LookupResponse> listener)

getBreaker

public CircuitBreaker getBreaker()

getExecutor

public Executor getExecutor()

extractFieldName

public static String extractFieldName(NamedExpression extractField)

createNullResponse

protected Page createNullResponse(int positionCount,
 List<NamedExpression> extractFields)