Catalog and Caching Integration

Flyte integrates with the Data Catalog to provide memoization, allowing tasks to skip execution if their outputs for a specific set of inputs and task version are already available. This integration is handled through synchronous and asynchronous clients that manage cache lookups, data uploads, and execution reservations.

Synchronous Cache Management

The primary mechanism for task-level memoization in Flyte is the catalog.Client interface. During the task reconciliation process, Flyte uses this client to check for existing artifacts before starting a task and to persist results once a task completes successfully.

Checking for Cache Hits

When a task is marked as discoverable, the TaskActionReconciler in executor/pkg/controller/taskaction_cache.go attempts to retrieve cached outputs using the Get method. If a match is found, the task transitions directly to a success state without executing the underlying plugin.

// From executor/pkg/controller/taskaction_cache.go
entry, err := r.Catalog.Get(ctx, cacheCfg.key)
if err == nil {
    // If found, write the cached outputs to the task's output writer
    if err := tCtx.OutputWriter().Put(ctx, entry.GetOutputs()); err != nil {
        return pluginsCore.UnknownTransition, false, fmt.Errorf("persisting cached outputs: %w", err)
    }

    // Transition the task to Success with a CACHE_HIT status
    info := cacheTaskInfo(corepb.CatalogCacheStatus_CACHE_HIT, "cache hit")
    return pluginsCore.DoTransition(pluginsCore.PhaseInfoSuccess(info)), true, nil
}

Persisting Task Outputs

After a successful task execution, Flyte populates the catalog by calling Put. This stores the task's outputs associated with a unique Key, making them available for future executions.

// From executor/pkg/controller/taskaction_cache.go
func (r *TaskActionReconciler) writeTaskOutputsToCache(ctx context.Context, tCtx pluginsCore.TaskExecutionContext, key catalog.Key) error {
    outputPaths := ioutils.NewReadOnlyOutputFilePaths(ctx, r.DataStore, tCtx.OutputWriter().GetOutputPrefixPath())
    outputReader := ioutils.NewRemoteFileOutputReader(ctx, r.DataStore, outputPaths, 0)
    _, err := r.Catalog.Put(ctx, key, outputReader, cacheMetadataForUpload(tCtx, key.Identifier))
    return err
}

Asynchronous Operations with Workqueues

For plugins that perform background data transfers or need to interact with the catalog without blocking the main execution loop, Flyte provides the AsyncClient. This client queues DownloadRequest and UploadRequest objects into internal workqueues.

The AsyncClientImpl (found in flyteplugins/go/tasks/pluginmachinery/catalog/async_client_impl.go) manages two workqueue.IndexedWorkQueue instances: a Reader for downloads and a Writer for uploads.

Queuing a Download Request

When you call Download, the client generates a unique work item ID based on the request key and target path, then queues it for processing.

// From flyteplugins/go/tasks/pluginmachinery/catalog/async_client_impl.go
func (c AsyncClientImpl) Download(ctx context.Context, requests ...DownloadRequest) (outputFuture DownloadFuture, err error) {
    // ...
    for idx, request := range requests {
        uniqueOutputLoc, _ := consistentHash(request.Target.GetOutputPrefixPath().String())
        workItemID := formatWorkItemID(request.Key, idx, uniqueOutputLoc)
        
        err = c.Reader.Queue(ctx, workItemID, NewReaderWorkItem(request.Key, request.Target))
        // ...
    }
    // Returns a future that can be polled for completion
    return newDownloadFuture(status, respErr, cachedResults, len(requests), cachedCount), nil
}

The AsyncClient uses a specialized base32 encoder for work item IDs to ensure they are safe for internal indexing and consistent hashing.

Cache Key Construction

The catalog.Key is the unique identifier for a cached artifact. It is constructed from the task's identity and its input data. If any component of the key changes, Flyte treats it as a cache miss.

The Key struct in flyteplugins/go/tasks/pluginmachinery/catalog/client.go includes:

• Identifier: The task's project, domain, name, and version.
• CacheVersion: A user-defined string that can be bumped to invalidate existing caches.
• InputReader: Used to retrieve and hash the actual input values.
• CacheIgnoreInputVars: A list of input variables that should be excluded from the hash calculation.

// From executor/pkg/controller/taskaction_cache.go
key := catalog.Key{
    Identifier:           proto.Clone(taskTemplate.GetId()).(*corepb.Identifier),
    CacheVersion:         taskAction.Spec.CacheKey,
    CacheIgnoreInputVars: metadata.GetCacheIgnoreInputVars(),
    TypedInterface:       taskTemplate.GetInterface(),
    InputReader:          tCtx.InputReader(),
}

Serializable Caching and Reservations

To prevent redundant concurrent executions of the same task (e.g., when multiple workflows trigger the same discoverable task simultaneously), Flyte supports Serializable Caching. This uses a reservation system managed by the Client.

When a task is serializable, the reconciler attempts to acquire a reservation using GetOrExtendReservation.

• If the current execution owns the reservation, it proceeds.
• If another execution owns a valid reservation, the current task transitions to a WaitingForCache phase.

// From executor/pkg/controller/taskaction_cache.go
if cacheCfg.serializable {
    reservation, err := r.Catalog.GetOrExtendReservation(ctx, cacheCfg.key, cacheCfg.ownerID, cacheReservationHeartbeatInterval)
    if err != nil {
        return pluginsCore.UnknownTransition, false, fmt.Errorf("acquiring cache reservation: %w", err)
    }
    
    if reservation.GetOwnerId() != cacheCfg.ownerID {
        // Another execution is currently running this task and populating the cache
        info := cacheTaskInfo(corepb.CatalogCacheStatus_CACHE_MISS, "waiting for serialized cache owner")
        return pluginsCore.DoTransition(pluginsCore.PhaseInfoWaitingForCache(taskAction.Status.PluginPhaseVersion, info)), true, nil
    }
}

Reservations require periodic heartbeats. The cacheReservationHeartbeatInterval is typically aligned with the task's requeue duration to ensure the reservation remains valid while the task is running.

Configuration

The catalog integration is configured via the catalogCache section in the Flyte configuration. This defines the behavior of the workqueues and cache expiration policies.

Setting	Description	Default
reader.workers	Number of concurrent workers for catalog downloads.	10
writer.maxRetries	Maximum retries for failed catalog uploads.	3
maxCacheAge	Duration after which a cache entry is considered expired.	0 (Never)
cacheKey.enforceExecutionProjectDomain	If true, uses the execution's project/domain for the key instead of the task's.	false

These settings are defined in the Config struct in flyteplugins/go/tasks/pluginmachinery/catalog/config.go. For example, the ReaderWorkqueueConfig and WriterWorkqueueConfig control the IndexedWorkQueue behavior used by the AsyncClient.