EMR Serverless StarRocks × Lance: Unifying Multimodal Search and Analytics in One SQL

Multimodal Lakehouse: Retrieval and Analysis on Shared Data

As AI applications penetrate business, video and image content are continuously processed into samples, labels, and vectors. The same data batch must serve feature engineering, model training, and online retrieval simultaneously. When these tasks scatter across multiple systems, each keeping its own data copy, storage overhead, sync latency, and version inconsistency emerge. The industry converges on a multimodal Lakehouse: object storage holds data, while processing, training, and analysis/retrieval engines collaborate on the same content via open formats.

For businesses, after finding similar content, they need to join orders, devices, or experiment records. SQL can place range filtering, content retrieval, and subsequent statistics in a single relational plan. Three typical scenarios illustrate this:

Autonomous Driving

Content Retrieval & Sample Organization: Given a missed pedestrian at night, constrain collection batch and annotation version, retrieve similar driving frames or target regions, locate parent video and adjacent frames via frame_id, assemble samples for re-review.

Further Relational Analysis: Join hit frames with annotations and model evaluation records via frame_id, filter by model and evaluation version; further join vehicle configurations, analyze miss types and version differences, generate re-annotation and regression test lists.

Embodied AI

Content Retrieval & Sample Organization: Given object slippage during transport, constrain robot arm model, object category, and policy version, retrieve similar failure frames, locate the Episode (task execution record).

Further Relational Analysis: Join multi-view video, gripper state, and action trajectories before/after failure, aggregate retry counts and duration per Episode, compare policy versions, prepare retraining and regression samples.

E-commerce Logistics

Content Retrieval & Sample Organization: Beyond package images, anomaly ticket text and sorting/transport trajectories generate embeddings; combine warehouse and time conditions, retrieve similar cases by appearance, text semantics, or trajectory patterns, gather multimodal clues.

Further Relational Analysis: Join images, tickets, and trajectory records via package ID, further join orders, packaging batches, and carriers, investigate whether anomalies concentrate on specific routes, operation stages, or packaging batches.

Lance: Data Organization and Access for AI Training

Computer vision and model training require both full scans and random sampling. Columnar formats like Parquet excel at scans but random sampling may pull and decode large pages. Lance, designed for such training needs, offers a columnar format balancing scan and random access, reducing irrelevant reads during sampling.

Lance separates indexes from column files, allowing index rebuilds (e.g., upgrading search structure or quantization) while reusing existing data. During indexed queries, the engine first locates candidate records, then reads column data on demand; column file page organization and encoding directly affect I/O and decode volume at this step.

StarRocks Integration Architecture: From Planning to Distributed Execution

The following Lance integration and execution optimizations are implemented in EMR Serverless StarRocks' Stella kernel. Integrating Lance requires organizing index search and data reading separately. The FE fixes the data version and index coverage, generates search tasks; after hits are produced, read tasks are grouped by fragment, file, and required columns. Both task types share the query version but can use different parallelism.

For queries needing full visible data coverage, indexed ranges use index search; uncovered fragments get supplemental scans. Tasks are partitioned by coverage to avoid misses or duplicates; task-to-BE mapping is decided by the scheduler based on node load and cache distribution.

Execution Node Division of Labor

Lance SDK is written in Rust. Integrating it into the C++ BE requires bringing cross-language data access into a unified execution and resource management system, focusing on four aspects:

Rust and C++ Columnar Execution Bridging

Lance SDK plans intra-task filtering, index search, and column reads; DataFusion provides columnar and streaming execution; C++ BE handles Pipeline, Exchange, and downstream relational operators. The adapter converts SDK-returned columnar batches into Chunks, connecting to Join, aggregation, and sorting.

Unified Memory Management

The adapter must account for Rust temporary buffers, Arrow batches, in-flight reads, and Chunk conversion overhead within a single query budget, clarifying cross-language data ownership and release timing. Compatible layouts reuse Arrow buffers; conversions incur extra accounted cost; cross-query reused index and metadata caches are managed separately at BE-level capacity.

Thread, Coroutine, and I/O Cooperative Scheduling

Index search compute parallelism, Rust async task count, and object storage I/O concurrency must be co-designed, then coupled to BE Pipeline consumption speed via batch buffering and backpressure. Budgets for in-flight requests and prefetch depth reduce thread oversubscription, buffer buildup, and async read pressure on query memory.

Cache Affinity and Load Correction

Index and metadata caches are maintained independently per BE; immutable objects are reused by identity; row location and visibility bind to snapshot reads. The scheduler selects execution locations based on existing cache, task size, and node load, preferring hot data reuse and dispersing hotspots via load correction.

Two Materialization Strategies and Subsequent OLAP Analysis

In distributed TopK retrieval, each BE produces local candidates, then global merge determines the global TopK hit set. If wide business columns (trajectory details, long text) are materialized for all local candidates before global merge, non-selected records waste I/O, decode, and network bandwidth; deferring these columns requires extra row location and refetch requests.

StarRocks incorporates column read position into execution planning, supporting two paths: materialize needed columns at source BE and pass candidates via Exchange, or complete global filtering first, then materialize required columns. The latter is called Global Late Materialization (GLM). The planner chooses based on query structure, deferrable columns, and reader capabilities.

Non-GLM

Source BE materializes columns on demand, forms Chunks, and passes them downstream via cross-node Exchange. When the reader supports multi-pass column reads, local filtering or TopK can narrow candidates before reading output columns, reusing local reads and cache, reducing extra refetch round-trips. This path suits narrow return columns, low local-to-global candidate elimination, or high location/RPC cost.

GLM

First pass row IDs, scores, and necessary columns via Exchange; after global filtering or TopK narrows candidates, parse row locations under a fixed snapshot, batch refetch by fragment and file. Returned columns align with hit records and fill Chunks for downstream computation. When deferred columns are wide and candidate elimination is high, this reduces useless reads, decodes, and transfers; but row location, RPC, and refetch scheduling costs must be accounted.

Both paths are constrained by reader capabilities and query semantics. Columns needed for candidate qualification, ranking, and refine must be read before respective operations; columns used only in later projection or analysis can be deferred. Both paths feed into Join, expression evaluation, and aggregation; materialization position is unified in the plan.

Retrieval and OLAP Analysis in a Single SQL

Take package damage investigation: after retrieving 100 similar images, read package IDs, join orders and logistics records, deduplicate by package, and aggregate carrier and packaging batch distributions. SQL places retrieval scope, join relations, and aggregation granularity in one plan; the hit set flows directly into downstream analysis.

In the illustrated GLM path, after global TopK is determined, StarRocks groups hit records by source, batch-requests required columns; Lance completes location and reads under a fixed snapshot. StarRocks aligns returned columns with hit records by data source and row ID, forming a Chunk (columnar batch). A Chunk stores a batch of records by column; distance, package ID, and business attributes at the same row position correspond to one hit record.

The completed Chunk enters StarRocks' vectorized execution pipeline, sequentially performing Join, deduplication, aggregation, and sorting. When cross-node computation is needed, Exchange redistributes data by join or group keys. The non-GLM path materializes at source and similarly feeds these operators.

The figure only shows hit package distribution, not overall damage rate. Order and logistics joins must select valid records per analysis scope; filters or joins participating in candidate qualification must complete before TopK truncation.

Customer Case Study: 2-Billion-Scale Hybrid Retrieval and Analysis

A large e-commerce trajectory platform needed attribute-filtered similar trajectory retrieval followed by business record analysis, while returning wide columns like trajectory details. The platform manages 2 billion 1024-dim Float32 vectors, organized in 500 fragments and 100 index segments.

Data Scale and Index Organization

Raw Vectors: 8.192 TB (each row 1024 × 4 = 4096 B)

500 Fragments: Avg 4M rows, 16.384 GB/fragment (raw vectors only)

100 Index Segments: Avg 20M rows/index segment (indexes and data fragments organized separately)

Index and Refine: Configured per Retrieval Goal

Index selection balances capacity and recall; whether to enable refine depends on per-query latency and accuracy targets. StarRocks supports query-level settings, allowing the same index to use different refine configs.

High QPS, Low Latency

This case's real-time retrieval uses IVF_SQ with TopK=1000 and refine disabled, avoiding raw vector reads and distance recomputation for reranking.

Higher Recall and Ranking Quality

Enabling refine with refine factor=5 as a tuning starting point, then adjusting based on actual recall, latency, and raw vector read volume. Refine reranks found candidates; recall is also affected by index search scope.

All three configurations include 8 B row ID per record; SQ uses 8 bits/dimension, PQ uses 128 8-bit codes per vector. SQ vector + row ID payload totals 2.064 TB; together with raw vectors 10.256 TB. IVF centroids, quantization parameters, and file metadata are counted separately.

Wide Column Return: Materializing After Global TopK

Each row has 8 KiB of deferrable business column payload; 100 independent search tasks each return local Top1000. Columns used for filtering and ranking are excluded from this comparison.

Materializing at source for 100k candidates equals 819.2 MB; after global TopK, only 1000 selected records need materialization, totaling 8.192 MB — a 99% payload reduction. Therefore, this case chooses GLM: pass row locations and scores via Exchange, then refetch wide columns for selected records, form Chunks for downstream analysis. Payload is counted by record count and encoded vector size or business column width; actual I/O and transfer are also affected by compression, read granularity, and cache, so payload reduction does not equal end-to-end speedup proportionally.

Business Impact: From Offline Batch to Real-Time Ad-hoc Analysis via Single SQL

The original solution used Ray + Lance to find TopK datasets, then handed off to StarRocks for joins and aggregation, with retrieval and analysis depending on batch handoffs. Now, StarRocks organizes hybrid retrieval, TopK, on-demand column reads, and subsequent OLAP in a single SQL, enabling businesses to launch ad-hoc analysis directly.

Under this customer's workload, Alibaba Cloud EMR Serverless StarRocks reduced end-to-end data availability wait to 1/40 of the original, from minutes to seconds; hybrid retrieval performance reached 16x that of Ray + Lance.

Multimodal Lakehouse as Systems Engineering: Product Collaboration and Continuous Optimization

To keep performance, cost, and data freshness jointly optimal long-term, unified system design is needed from ingestion, feature backfill, index building, to query serving. Write batch size affects file count and object storage requests; index update cadence affects when new data becomes efficiently searchable; resource allocation impacts both service latency and running cost. Object storage, processing frameworks, management services, and StarRocks must collaborate around these metrics.

Product Combination and Responsibilities

OSS / S3 + Lance: Store raw content, organize column data, snapshots, and indexes

Ray / Spark: Clean data, compute features, and backfill by column

Alibaba Cloud DLF: Unified table catalog, automatically build and continuously maintain Lance indexes per config

StarRocks: Execute SQL, unify analysis, retrieval, and resource scheduling

After feature backfill, indexes must promptly cover new features; queries use data and indexes valid under the corresponding version. File compaction, index maintenance, and resource provisioning should also be tuned based on StarRocks' analysis and retrieval performance.

StarRocks Continuous Optimizations for Mixed Workloads

Joint management of fragments, index segments, and incremental coverage: Avoid accumulation of small files and excessive incremental segments; monitor search parallelism, maintenance task size, and load distribution. After compaction and index rebuild, update coverage and location info by new version, preserve caches for reusable objects. New data visibility time and index serviceability time must be managed separately.

Separate observation of search, refine, Exchange, and deferred column reads: Distinguish costs in StarRocks query metrics; record candidate scale, search latency, raw vector read volume, Exchange bytes, and Fetch/Lookup latency. Index and data reads must separately track cache hits, evaluated alongside full query latency to assess optimization gains.

Provide optimizer with more reliable decision bases: Accumulate filter selectivity, local and global shrink ratios, column widths, and batch read overhead to refine materialization strategy cost estimation. Validate non-GLM vs GLM with fixed model, distance metric, recall target, and candidate budget, covering varying K, projection widths, cold/warm caches, and concurrency levels.

Manage mixed workloads via compute groups and background scheduling: Interactive retrieval, batch analysis, index maintenance, and feature engineering have different resource goals. StarRocks isolates retrieval and analysis loads via compute groups, configuring concurrency and resource budgets separately; background tasks must coordinate shared object storage bandwidth and request pressure, controlling tail latency jitter.

In this case, the business shifted from stepped retrieval and batch analysis to single-SQL-driven real-time ad-hoc analysis. Supporting this change requires handling index search, distributed execution, on-demand column reads, and subsequent relational computation simultaneously, while managing resource contention across workloads. Stella implements these in the execution kernel; EMR Serverless StarRocks provides fully managed service, elastic resources, and multi-compute-group isolation, letting enterprises organize analysis and retrieval services per response, throughput, and cost requirements.

Going forward, Alibaba Cloud will continue refining Stella with real customer loads, improving hybrid retrieval, materialization path selection, and resource scheduling, while continuously collaborating with Lance's data and indexing capabilities. They will also push the Lance integration core code to the StarRocks community, fostering mutual advancement between cloud production practice and open ecosystem.