Skip to main content

Advanced Querying and Filtering

Flyte uses a structured filtering system in its repository layer to perform complex searches across runs, actions, and events. This system abstracts SQL generation, allowing you to compose queries using logical operators while protecting against SQL injection through field validation.

Building Basic Filters​

The runs/repository/impl package provides several factory functions to create Filter objects for standard field comparisons. These filters map internal field names to SQL expressions.

import (
    "github.com/flyteorg/flyte/v2/runs/repository/impl"
    "github.com/flyteorg/flyte/v2/runs/repository/interfaces"
)

// Create a simple equality filter
projectFilter := impl.NewEqualFilter("project", "flytesnacks")

// Create a filter for multiple values (SQL IN)
statusFilter := impl.NewEqualFilter("phase", []int32{2, 3}) // Uses FilterExpressionValueIn internally

Available factory functions in runs/repository/impl/filters.go include:

  • NewEqualFilter(field, value): Matches exact values or a set of values.
  • NewNotEqualFilter(field, value): Matches values not equal to the input.
  • NewProjectIdFilter(projectId): Matches both project and domain.
  • NewTaskNameFilter(taskName): Matches project, domain, and name for tasks.
  • NewIsRootActionFilter(): Specifically filters for root actions (runs) where parent_action_name is NULL.

Composing Complex Queries​

Filters can be combined using .And() and .Or() methods defined in the interfaces.Filter interface. This creates a compositeFilter that recursively generates the appropriate SQL with parentheses.

// Combine multiple criteria with AND
filter := impl.NewIsRootActionFilter().
    And(impl.NewEqualFilter("project", "proj1")).
    And(impl.NewEqualFilter("domain", "development"))

// Use OR for alternative criteria
filter = impl.NewEqualFilter("phase", 4).
    Or(impl.NewEqualFilter("phase", 5))

Executing Queries with Pagination​

To execute a query, wrap your filter in a interfaces.ListResourceInput struct and pass it to a repository method like ListActions. Flyte supports both offset-based and keyset-based pagination.

Keyset pagination uses a CursorToken (an RFC3339Nano timestamp) to fetch results strictly after a specific creation time.

input := interfaces.ListResourceInput{
    Filter: impl.NewIsRootActionFilter(),
    Limit:  50,
}

// Execute the query
runs, err := actionRepo.ListActions(ctx, input)

// To get the next page, use the CreatedAt of the last item as the cursor
if len(runs) > 0 {
    nextPageInput := interfaces.ListResourceInput{
        Filter:      impl.NewIsRootActionFilter(),
        Limit:       50,
        CursorToken: runs[len(runs)-1].CreatedAt.UTC().Format(time.RFC3339Nano),
    }
}

Sorting Results​

You can specify sort order using NewSortParameter. If no sort parameters are provided, ListActions defaults to phase ASC, created_at DESC.

input := interfaces.ListResourceInput{
    Filter: impl.NewEqualFilter("project", "my-project"),
    SortParameters: []interfaces.SortParameter{
        impl.NewSortParameter("created_at", interfaces.SortOrderDescending),
    },
}

Security and Validation​

When converting external API requests (Protobuf) into repository filters, Flyte uses ConvertProtoFilters to validate field names against a set of allowed columns. This prevents SQL injection by ensuring only known database columns can be used in filter expressions.

// Example of validating and converting proto filters
allowedColumns := sets.NewString("project", "domain", "phase", "created_at")
filter, err := impl.ConvertProtoFilters(request.GetFilters(), allowedColumns)
if err != nil {
    // Handle invalid filter field error
}

The basicFilter.QueryExpression method handles the actual SQL generation, using placeholders (?) for values to ensure they are safely handled by the underlying database driver. For example, FilterExpressionContains automatically wraps the value in % wildcards and uses the LIKE operator.

// Internal implementation snippet from runs/repository/impl/filters.go
case interfaces.FilterExpressionContains:
    query = fmt.Sprintf("%s LIKE ?", column)
    f.value = fmt.Sprintf("%%%v%%", f.value)