Vectorization result cache - Realtime Compute for Apache Flink

This document describes how to configure an external cache table for the AI_EMBED function to reuse vectorization results for identical input text, thereby reducing model calls, lowering latency, and saving costs.

Background information

The AI_EMBED function uses the CACHE_TABLE parameter to specify an external table as a persistent cache for vectorization results. At runtime, it operates as follows:

Cache hit: If the asynchronous lookup finds a match, the function returns the result directly, skipping the model call.
Cache miss: If the asynchronous lookup fails to find a match, the function calls the model to compute the vector, returns the result, and asynchronously writes it back to the cache table.

This mechanism prevents redundant model calls for repeated requests with the same input text.

Limitations

This feature is supported only by Realtime Compute for Apache Flink VVR 11.8.preview1 and later versions.
Enable Stream Storage Apache Fluss Edition. The cache table must be a Fluss primary key table.

Create a cache table

The primary key type of the cache table must be compatible with the type of the CACHE_KEY.

Explicitly specified key mode

The primary key type must be compatible with the type of the column specified by CACHE_KEY. For example, if the cache_key in the source table is an INT:

CREATE TABLE `fluss`.`default`.`ai_cache` (
  cache_key   INT NOT NULL,
  embedding   ARRAY<FLOAT>,PRIMARY KEY (cache_key) NOT ENFORCED
);

Automatically generated key mode

The primary key must be of type STRING to store the SHA256 hash value:

CREATE TABLE `fluss`.`default`.`ai_cache_sha` (
  cache_key   STRING NOT NULL,
  embedding   ARRAY<FLOAT>,PRIMARY KEY (cache_key) NOT ENFORCED
);

Syntax

AI_EMBED(
  MODEL       => MODEL <MODEL NAME>,
  INPUT       => <INPUT COLUMN NAME>
  [, DIMENSION   => <DIMENSION VALUE>]
  [, CACHE_TABLE => TABLE <catalog>.<db>.<table>
     [, CACHE_KEY => DESCRIPTOR(<column>)]]
  [, CONFIG      => MAP[...]]
)

Parameters

Parameter	Required	Description
`MODEL`	Yes	A reference to the model. For more information, see Register model.
`INPUT`	Yes	The input text column.
`CACHE_TABLE`	No	A reference to the cache table. The syntax is `TABLE <catalog>.<db>.<table>`.
`CACHE_KEY`	No	The cache key column. The syntax is `DESCRIPTOR(<column>)`. If not specified, the system automatically generates a composite key using an SHA256 hash.
`CONFIG`	No	The runtime configuration.

CACHE_KEY modes

Explicitly specified (Recommended): Uses the value from a specific column in the input table as the lookup key. This mode is recommended when your data already has a unique key.

CACHE_KEY => DESCRIPTOR(cache_key)

Automatic generation: If you do not pass CACHE_KEY , the system generates a Key based on the following rules:

-- If CACHE_KEY is not provided, the system automatically generates the key:
-- key = LOWER(SHA256(model_id || ':' || dimension || ':' || input_text))
CACHE_TABLE => TABLE `fluss-ql`.`default`.`ai_cache1`

This mode incurs the overhead of computing an SHA256 hash at runtime, but it automatically ensures a unique key for each combination of model, dimension, and input text.

Performance tuning

Tuning parameters

SET 'table.exec.async-ml-predict.max-concurrent-operations' = '100';
SET 'table.exec.async-ml-predict.output-mode' = 'ALLOW_UNORDERED';
SET 'table.exec.async-ml-predict.timeout' = '120s';

Parameters

Parameter	Default	Description
`max-concurrent-operations`	10	The maximum number of records for concurrent cache lookups or model calls. Increasing this value can improve throughput. However, setting it too high might overload the model service or cause excessive memory usage.
`output-mode`	`ORDERED`	The output order mode. `ORDERED`: Guarantees that the output is in the same order as the input. This is suitable for downstream operations that depend on order, such as joins or window functions. `ALLOW_UNORDERED`: Outputs records as soon as they are complete. This is recommended for insert-only streams.
`timeout`	3min	The maximum total time allowed for both the lookup and prediction. If this value is set too low, a normal operation on a cache miss might time out.

Recommendation: Using ALLOW_UNORDERED provides significant performance benefits for the following reasons:

A cache hit typically requires only a single lookup, whereas a cache miss requires an additional model call.
In ORDERED mode, a single cache miss blocks the output of all subsequent cache hits.
In ALLOW_UNORDERED mode, cache hits are output immediately without waiting for preceding misses to complete.

Note

Only insert-only streams use the ALLOW_UNORDERED mode. For streams with retractions, the system falls back to ORDERED mode, even if ALLOW_UNORDERED is specified.

Monitoring metrics

While a job is running, you can use Flink metrics to monitor the effectiveness of the cache.

Metric name	Description	Notes
`ai_function_cache.cache_hit`	cache hit count	Hit rate = hit / (hit + miss).
`ai_function_cache.cache_miss`	cache miss count	A consistently high value suggests that the cache is not warmed up or that the keys are unstable.
`ai_function_cache.writeback_attempt`	write-back attempt count	This value should be close to the miss count.
`ai_function_cache.writeback_skipped_null_output`	write-backs skipped due to null output	A high value indicates a potential issue with the model.

Example

Example of a Fluss primary key table:

-- Create the cache table (Fluss) in advance
CREATE TABLE `fluss`.`default`.`embedding_cache` (
  cache_key   STRING NOT NULL,
  embedding   ARRAY<FLOAT>,
  PRIMARY KEY (cache_key) NOT ENFORCED
);

This example uses a Fluss primary key table as the cache and sets the output mode to ALLOW_UNORDERED to improve throughput.

-- Tuning parameters
SET 'table.exec.async-ml-predict.max-concurrent-operations' = '100';
SET 'table.exec.async-ml-predict.output-mode' = 'ALLOW_UNORDERED';

-- Create the model
CREATE TEMPORARY MODEL embedding_model
  INPUT (text STRING)
  OUTPUT (embedding ARRAY<FLOAT>)
WITH (
  'provider' = 'openai-compat',
  'task' = 'embeddings',
  'model' = 'text-embedding-v4',
  'dimension' = '1024'
);

-- Business query
INSERT INTO result_sink
SELECT id, content, embedding
FROM source_table, LATERAL TABLE(AI_EMBED(
  MODEL       => MODEL embedding_model,
  INPUT       => content,
  CONFIG      => MAP['async', 'true'],
  CACHE_TABLE => TABLE `fluss`.`default`.`embedding_cache`
));