From 6140b77b699ae5e3fec2bb232a9a690608e399c6 Mon Sep 17 00:00:00 2001 From: rohithreddykota Date: Mon, 8 Jun 2026 15:36:30 -0400 Subject: [PATCH 1/2] feat: add partition range filter to `rill project refresh` Adds --partition-key, --partition-start, and --partition-end flags so operators can refresh a contiguous range of incremental-model partitions without having to look up each MD5 partition key manually. Resolution happens client-side and the matched partitions are listed for confirmation before triggering; --yes skips the prompt. --- cli/cmd/project/refresh.go | 114 ++++++++++++++++++++++++++++++++++++- 1 file changed, 111 insertions(+), 3 deletions(-) diff --git a/cli/cmd/project/refresh.go b/cli/cmd/project/refresh.go index 515822ac0fcc..f4e006dec101 100644 --- a/cli/cmd/project/refresh.go +++ b/cli/cmd/project/refresh.go @@ -1,20 +1,24 @@ package project import ( + "context" "fmt" + "strconv" "github.com/rilldata/rill/cli/pkg/cmdutil" runtimev1 "github.com/rilldata/rill/proto/gen/rill/runtime/v1" "github.com/rilldata/rill/runtime" "github.com/spf13/cobra" "github.com/spf13/pflag" + "google.golang.org/protobuf/types/known/structpb" ) func RefreshCmd(ch *cmdutil.Helper) *cobra.Command { var project, path, branch string var local bool var models, modelPartitions, sources, metricViews, alerts, reports, connectors []string - var all, full, erroredPartitions, parser bool + var all, full, erroredPartitions, parser, yes bool + var partitionKey, partitionStart, partitionEnd string refreshCmd := &cobra.Command{ Use: "refresh []", @@ -79,11 +83,26 @@ func RefreshCmd(ch *cmdutil.Helper) *cobra.Command { // Merge sources into models since sources have been deprecated and are no longer created on the backend. models = append(models, sources...) + // Validate partition-range flags. All three must be set together, and the range mode + // is mutually exclusive with --partition and --errored-partitions. + rangeMode := partitionKey != "" || partitionStart != "" || partitionEnd != "" + if rangeMode { + if partitionKey == "" || partitionStart == "" || partitionEnd == "" { + return fmt.Errorf("--partition-key, --partition-start, and --partition-end must all be set together") + } + if len(modelPartitions) > 0 || erroredPartitions { + return fmt.Errorf("--partition-key cannot be combined with --partition or --errored-partitions") + } + if partitionStart > partitionEnd { + return fmt.Errorf("--partition-start (%q) must be <= --partition-end (%q)", partitionStart, partitionEnd) + } + } + // Build model triggers - if len(modelPartitions) > 0 || erroredPartitions { + if len(modelPartitions) > 0 || erroredPartitions || rangeMode { // If partitions are specified, ensure exactly one model is specified. if len(models) != 1 { - return fmt.Errorf("must specify exactly one --model when using --partition or --errored-partitions") + return fmt.Errorf("must specify exactly one --model when using --partition, --errored-partitions, or --partition-key") } // Since it's a common error, do an early check to ensure the model is incremental. @@ -101,6 +120,31 @@ func RefreshCmd(ch *cmdutil.Helper) *cobra.Command { return fmt.Errorf("can't refresh partitions on model %q because it is not incremental", mn) } } + + // Resolve partition range to concrete partition keys. + if rangeMode { + matched, err := resolvePartitionRange(cmd.Context(), rt, instanceID, models[0], partitionKey, partitionStart, partitionEnd) + if err != nil { + return err + } + if len(matched) == 0 { + ch.Printf("No partitions match %s in [%s, %s] on model %q.\n", partitionKey, partitionStart, partitionEnd, models[0]) + return nil + } + + ch.PrintModelPartitions(matched) + + if !yes && ch.Interactive { + if err := cmdutil.ConfirmPrompt(fmt.Sprintf("Refresh %d partition(s)?", len(matched)), true); err != nil { + return err + } + } + + modelPartitions = make([]string, 0, len(matched)) + for _, p := range matched { + modelPartitions = append(modelPartitions, p.Key) + } + } var modelTriggers []*runtimev1.RefreshModelTrigger for _, m := range models { modelTriggers = append(modelTriggers, &runtimev1.RefreshModelTrigger{ @@ -150,6 +194,10 @@ func RefreshCmd(ch *cmdutil.Helper) *cobra.Command { refreshCmd.Flags().StringSliceVar(&models, "model", nil, "Refresh a model") refreshCmd.Flags().StringSliceVar(&modelPartitions, "partition", nil, "Refresh a model partition (must set --model)") refreshCmd.Flags().BoolVar(&erroredPartitions, "errored-partitions", false, "Refresh all model partitions with errors (must set --model)") + refreshCmd.Flags().StringVar(&partitionKey, "partition-key", "", "Name of the field in the partition data to range-filter on (must set --model)") + refreshCmd.Flags().StringVar(&partitionStart, "partition-start", "", "Inclusive lower bound for --partition-key (lexicographic string compare)") + refreshCmd.Flags().StringVar(&partitionEnd, "partition-end", "", "Inclusive upper bound for --partition-key (lexicographic string compare)") + refreshCmd.Flags().BoolVar(&yes, "yes", false, "Skip the partition-range refresh confirmation prompt") refreshCmd.Flags().StringSliceVar(&sources, "source", nil, "Refresh a source") refreshCmd.Flags().StringSliceVar(&metricViews, "metrics-view", nil, "Refresh a metrics view") refreshCmd.Flags().StringSliceVar(&alerts, "alert", nil, "Refresh an alert") @@ -159,3 +207,63 @@ func RefreshCmd(ch *cmdutil.Helper) *cobra.Command { return refreshCmd } + +// resolvePartitionRange lists all partitions of the given model and returns those whose +// data field `key` falls within [start, end] (inclusive, lexicographic string compare). +func resolvePartitionRange(ctx context.Context, rt runtimev1.RuntimeServiceClient, instanceID, model, key, start, end string) ([]*runtimev1.ModelPartition, error) { + var matched []*runtimev1.ModelPartition + var pageToken string + var sawAnyPartition bool + for { + res, err := rt.GetModelPartitions(ctx, &runtimev1.GetModelPartitionsRequest{ + InstanceId: instanceID, + Model: model, + PageSize: 100, + PageToken: pageToken, + }) + if err != nil { + return nil, fmt.Errorf("failed to list partitions for model %q: %w", model, err) + } + + for _, p := range res.Partitions { + sawAnyPartition = true + if p.Data == nil { + continue + } + fields := p.Data.GetFields() + v, ok := fields[key] + if !ok { + available := make([]string, 0, len(fields)) + for f := range fields { + available = append(available, f) + } + return nil, fmt.Errorf("partition field %q not found on partition %q; available fields: %v", key, p.Key, available) + } + + var s string + switch k := v.Kind.(type) { + case *structpb.Value_StringValue: + s = k.StringValue + case *structpb.Value_NumberValue: + s = strconv.FormatFloat(k.NumberValue, 'f', -1, 64) + case *structpb.Value_BoolValue: + s = strconv.FormatBool(k.BoolValue) + default: + return nil, fmt.Errorf("partition %q: unsupported type %T for field %q", p.Key, v.Kind, key) + } + if s >= start && s <= end { + matched = append(matched, p) + } + } + + if res.NextPageToken == "" { + break + } + pageToken = res.NextPageToken + } + + if !sawAnyPartition { + return nil, fmt.Errorf("model %q has no partitions to filter", model) + } + return matched, nil +} From e476250facbdb3b3fcc2a6aad2a2274027ecef36 Mon Sep 17 00:00:00 2001 From: rohithreddykota Date: Tue, 7 Jul 2026 14:19:20 -0400 Subject: [PATCH 2/2] use skip partitions --- cli/cmd/project/refresh.go | 42 +++++++++----- cli/cmd/project/skip_partition.go | 58 ++++++++++++++++++- docs/docs/reference/cli/project/refresh.md | 36 +++++++----- .../reference/cli/project/skip-partition.md | 20 ++++--- 4 files changed, 117 insertions(+), 39 deletions(-) diff --git a/cli/cmd/project/refresh.go b/cli/cmd/project/refresh.go index 80435a66a7ee..a955526ffecd 100644 --- a/cli/cmd/project/refresh.go +++ b/cli/cmd/project/refresh.go @@ -17,7 +17,7 @@ func RefreshCmd(ch *cmdutil.Helper) *cobra.Command { var project, path, branch string var local bool var models, modelPartitions, sources, metricViews, alerts, reports, connectors []string - var all, full, erroredPartitions, skippedPartitions, parser, yes bool + var all, full, erroredPartitions, skippedPartitions, parser, force bool var partitionKey, partitionStart, partitionEnd string refreshCmd := &cobra.Command{ @@ -83,15 +83,19 @@ func RefreshCmd(ch *cmdutil.Helper) *cobra.Command { // Merge sources into models since sources have been deprecated and are no longer created on the backend. models = append(models, sources...) - // Validate partition-range flags. All three must be set together, and the range mode - // is mutually exclusive with --partition and --errored-partitions. + // Validate partition-range flags. All three must be set together. The range can be + // narrowed by --errored-partitions or --skipped-partitions (but not both), and cannot + // be combined with an explicit --partition list. rangeMode := partitionKey != "" || partitionStart != "" || partitionEnd != "" if rangeMode { if partitionKey == "" || partitionStart == "" || partitionEnd == "" { return fmt.Errorf("--partition-key, --partition-start, and --partition-end must all be set together") } - if len(modelPartitions) > 0 || erroredPartitions || skippedPartitions { - return fmt.Errorf("--partition-key cannot be combined with --partition, --errored-partitions, or --skipped-partitions") + if len(modelPartitions) > 0 { + return fmt.Errorf("--partition-key cannot be combined with --partition") + } + if erroredPartitions && skippedPartitions { + return fmt.Errorf("--errored-partitions and --skipped-partitions cannot be combined") } if partitionStart > partitionEnd { return fmt.Errorf("--partition-start (%q) must be <= --partition-end (%q)", partitionStart, partitionEnd) @@ -121,9 +125,10 @@ func RefreshCmd(ch *cmdutil.Helper) *cobra.Command { } } - // Resolve partition range to concrete partition keys. + // Resolve partition range to concrete partition keys. When --errored-partitions or + // --skipped-partitions is also set, the range is narrowed to partitions in that state. if rangeMode { - matched, err := resolvePartitionRange(cmd.Context(), rt, instanceID, models[0], partitionKey, partitionStart, partitionEnd) + matched, err := resolvePartitionRange(cmd.Context(), rt, instanceID, models[0], partitionKey, partitionStart, partitionEnd, false, erroredPartitions, skippedPartitions) if err != nil { return err } @@ -134,7 +139,7 @@ func RefreshCmd(ch *cmdutil.Helper) *cobra.Command { ch.PrintModelPartitions(matched) - if !yes && ch.Interactive { + if !force && ch.Interactive { if err := cmdutil.ConfirmPrompt(fmt.Sprintf("Refresh %d partition(s)?", len(matched)), true); err != nil { return err } @@ -144,6 +149,10 @@ func RefreshCmd(ch *cmdutil.Helper) *cobra.Command { for _, p := range matched { modelPartitions = append(modelPartitions, p.Key) } + // The resolved key list is authoritative; clear the model-wide flags so the trigger + // refreshes only the matched partitions rather than all errored/skipped ones. + erroredPartitions = false + skippedPartitions = false } var modelTriggers []*runtimev1.RefreshModelTrigger for _, m := range models { @@ -199,7 +208,7 @@ func RefreshCmd(ch *cmdutil.Helper) *cobra.Command { refreshCmd.Flags().StringVar(&partitionKey, "partition-key", "", "Name of the field in the partition data to range-filter on (must set --model)") refreshCmd.Flags().StringVar(&partitionStart, "partition-start", "", "Inclusive lower bound for --partition-key (lexicographic string compare)") refreshCmd.Flags().StringVar(&partitionEnd, "partition-end", "", "Inclusive upper bound for --partition-key (lexicographic string compare)") - refreshCmd.Flags().BoolVar(&yes, "yes", false, "Skip the partition-range refresh confirmation prompt") + refreshCmd.Flags().BoolVar(&force, "force", false, "Skip the partition-range refresh confirmation prompt") refreshCmd.Flags().StringSliceVar(&sources, "source", nil, "Refresh a source") refreshCmd.Flags().StringSliceVar(&metricViews, "metrics-view", nil, "Refresh a metrics view") refreshCmd.Flags().StringSliceVar(&alerts, "alert", nil, "Refresh an alert") @@ -210,9 +219,10 @@ func RefreshCmd(ch *cmdutil.Helper) *cobra.Command { return refreshCmd } -// resolvePartitionRange lists all partitions of the given model and returns those whose -// data field `key` falls within [start, end] (inclusive, lexicographic string compare). -func resolvePartitionRange(ctx context.Context, rt runtimev1.RuntimeServiceClient, instanceID, model, key, start, end string) ([]*runtimev1.ModelPartition, error) { +// resolvePartitionRange lists partitions of the given model and returns those whose data field +// `key` falls within [start, end] (inclusive, lexicographic string compare). The pending, errored, +// and skipped flags narrow the listing to partitions in the corresponding state (server-side). +func resolvePartitionRange(ctx context.Context, rt runtimev1.RuntimeServiceClient, instanceID, model, key, start, end string, pending, errored, skipped bool) ([]*runtimev1.ModelPartition, error) { var matched []*runtimev1.ModelPartition var pageToken string var sawAnyPartition bool @@ -220,6 +230,9 @@ func resolvePartitionRange(ctx context.Context, rt runtimev1.RuntimeServiceClien res, err := rt.GetModelPartitions(ctx, &runtimev1.GetModelPartitionsRequest{ InstanceId: instanceID, Model: model, + Pending: pending, + Errored: errored, + Skipped: skipped, PageSize: 100, PageToken: pageToken, }) @@ -264,7 +277,10 @@ func resolvePartitionRange(ctx context.Context, rt runtimev1.RuntimeServiceClien pageToken = res.NextPageToken } - if !sawAnyPartition { + // When no state filter is applied, an empty listing means the model genuinely has no + // partitions, which is a usage error. With a state filter, an empty listing just means no + // partitions are in that state, so let the caller report the friendly "no match" message. + if !sawAnyPartition && !pending && !errored && !skipped { return nil, fmt.Errorf("model %q has no partitions to filter", model) } return matched, nil diff --git a/cli/cmd/project/skip_partition.go b/cli/cmd/project/skip_partition.go index 82adfb3d9261..c85a4858f125 100644 --- a/cli/cmd/project/skip_partition.go +++ b/cli/cmd/project/skip_partition.go @@ -11,7 +11,8 @@ import ( func SkipPartitionCmd(ch *cmdutil.Helper) *cobra.Command { var project, path, branch, model string var partitions []string - var pending, errored, local bool + var pending, errored, local, force bool + var partitionKey, partitionStart, partitionEnd string skipCmd := &cobra.Command{ Use: "skip-partition [] ", @@ -35,11 +36,60 @@ func SkipPartitionCmd(ch *cmdutil.Helper) *cobra.Command { } } + // Validate partition-range flags. All three must be set together. The range can be + // narrowed by --pending or --errored (but not both), and cannot be combined with an + // explicit --partition list. + rangeMode := partitionKey != "" || partitionStart != "" || partitionEnd != "" + if rangeMode { + if partitionKey == "" || partitionStart == "" || partitionEnd == "" { + return fmt.Errorf("--partition-key, --partition-start, and --partition-end must all be set together") + } + if len(partitions) > 0 { + return fmt.Errorf("--partition-key cannot be combined with --partition") + } + if pending && errored { + return fmt.Errorf("--pending and --errored cannot be combined") + } + if partitionStart > partitionEnd { + return fmt.Errorf("--partition-start (%q) must be <= --partition-end (%q)", partitionStart, partitionEnd) + } + } + rt, instanceID, err := ch.OpenRuntimeClient(cmd.Context(), ch.Org, project, branch, local) if err != nil { return err } + // Resolve a partition range to concrete partition keys. When --pending or --errored is + // also set, the range is narrowed to partitions in that state. + if rangeMode { + matched, err := resolvePartitionRange(cmd.Context(), rt, instanceID, model, partitionKey, partitionStart, partitionEnd, pending, errored, false) + if err != nil { + return err + } + if len(matched) == 0 { + ch.Printf("No partitions match %s in [%s, %s] on model %q.\n", partitionKey, partitionStart, partitionEnd, model) + return nil + } + + ch.PrintModelPartitions(matched) + + if !force && ch.Interactive { + if err := cmdutil.ConfirmPrompt(fmt.Sprintf("Skip %d partition(s)?", len(matched)), true); err != nil { + return err + } + } + + partitions = make([]string, 0, len(matched)) + for _, p := range matched { + partitions = append(partitions, p.Key) + } + // The resolved key list is authoritative; clear the state flags so only the matched + // partitions are skipped rather than all pending/errored ones. + pending = false + errored = false + } + _, err = rt.SkipModelPartitions(cmd.Context(), &runtimev1.SkipModelPartitionsRequest{ InstanceId: instanceID, Model: model, @@ -65,7 +115,11 @@ func SkipPartitionCmd(ch *cmdutil.Helper) *cobra.Command { skipCmd.Flags().StringSliceVar(&partitions, "partition", nil, "Skip specific partitions by key") skipCmd.Flags().BoolVar(&pending, "pending", false, "Skip all pending partitions") skipCmd.Flags().BoolVar(&errored, "errored", false, "Skip all errored partitions") - skipCmd.MarkFlagsOneRequired("partition", "pending", "errored") + skipCmd.Flags().StringVar(&partitionKey, "partition-key", "", "Name of the field in the partition data to range-filter on") + skipCmd.Flags().StringVar(&partitionStart, "partition-start", "", "Inclusive lower bound for --partition-key (lexicographic string compare)") + skipCmd.Flags().StringVar(&partitionEnd, "partition-end", "", "Inclusive upper bound for --partition-key (lexicographic string compare)") + skipCmd.Flags().BoolVar(&force, "force", false, "Skip the partition-range confirmation prompt") + skipCmd.MarkFlagsOneRequired("partition", "pending", "errored", "partition-key", "partition-start", "partition-end") skipCmd.Flags().BoolVar(&local, "local", false, "Target locally running Rill") return skipCmd diff --git a/docs/docs/reference/cli/project/refresh.md b/docs/docs/reference/cli/project/refresh.md index 9ca90b01d425..77a7d9227ae7 100644 --- a/docs/docs/reference/cli/project/refresh.md +++ b/docs/docs/reference/cli/project/refresh.md @@ -13,22 +13,26 @@ rill project refresh [] [flags] ### Flags ``` - --project string Project name - --path string Project directory (default ".") - --branch string Target deployment by Git branch (default: primary deployment) - --local Target locally running Rill - --all Refresh all resources except alerts and reports (default) - --full Fully reload the targeted models (use with --all or --model) - --model strings Refresh a model - --partition strings Refresh a model partition (must set --model) - --errored-partitions Refresh all model partitions with errors (must set --model) - --skipped-partitions Refresh all skipped model partitions (must set --model) - --source strings Refresh a source - --metrics-view strings Refresh a metrics view - --alert strings Refresh an alert - --report strings Refresh a report - --connector strings Re-validate a connector - --parser Refresh the parser (forces a pull from Github) + --project string Project name + --path string Project directory (default ".") + --branch string Target deployment by Git branch (default: primary deployment) + --local Target locally running Rill + --all Refresh all resources except alerts and reports (default) + --full Fully reload the targeted models (use with --all or --model) + --model strings Refresh a model + --partition strings Refresh a model partition (must set --model) + --errored-partitions Refresh all model partitions with errors (must set --model) + --skipped-partitions Refresh all skipped model partitions (must set --model) + --partition-key string Name of the field in the partition data to range-filter on (must set --model) + --partition-start string Inclusive lower bound for --partition-key (lexicographic string compare) + --partition-end string Inclusive upper bound for --partition-key (lexicographic string compare) + --force Skip the partition-range refresh confirmation prompt + --source strings Refresh a source + --metrics-view strings Refresh a metrics view + --alert strings Refresh an alert + --report strings Refresh a report + --connector strings Re-validate a connector + --parser Refresh the parser (forces a pull from Github) ``` ### Global flags diff --git a/docs/docs/reference/cli/project/skip-partition.md b/docs/docs/reference/cli/project/skip-partition.md index 62a82bf68497..095be5d624c9 100644 --- a/docs/docs/reference/cli/project/skip-partition.md +++ b/docs/docs/reference/cli/project/skip-partition.md @@ -17,14 +17,18 @@ rill project skip-partition [] [flags] ### Flags ``` - --project string Project Name - --path string Project directory (default ".") - --branch string Target deployment by Git branch (default: primary deployment) - --model string Model Name - --partition strings Skip specific partitions by key - --pending Skip all pending partitions - --errored Skip all errored partitions - --local Target locally running Rill + --project string Project Name + --path string Project directory (default ".") + --branch string Target deployment by Git branch (default: primary deployment) + --model string Model Name + --partition strings Skip specific partitions by key + --pending Skip all pending partitions + --errored Skip all errored partitions + --partition-key string Name of the field in the partition data to range-filter on + --partition-start string Inclusive lower bound for --partition-key (lexicographic string compare) + --partition-end string Inclusive upper bound for --partition-key (lexicographic string compare) + --force Skip the partition-range confirmation prompt + --local Target locally running Rill ``` ### Global flags