Add partition range filter to rill project refresh

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, skippedPartitions, parser bool
+	var all, full, erroredPartitions, skippedPartitions, parser, force bool
+	var partitionKey, partitionStart, partitionEnd string
 
 	refreshCmd := &cobra.Command{
 		Use:               "refresh [<project-name>]",
@@ -79,11 +83,30 @@ 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. 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 {
+					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)
+				}
+			}
+
 			// Build model triggers
-			if len(modelPartitions) > 0 || erroredPartitions || skippedPartitions {
+			if len(modelPartitions) > 0 || erroredPartitions || skippedPartitions || 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, --errored-partitions, or --skipped-partitions")
+					return fmt.Errorf("must specify exactly one --model when using --partition, --errored-partitions, --skipped-partitions, or --partition-key")
 				}
 
 				// Since it's a common error, do an early check to ensure the model is incremental.
@@ -101,6 +124,36 @@ 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. 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, false, erroredPartitions, skippedPartitions)
+				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 !force && 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)
+				}
+				// 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 {
 				modelTriggers = append(modelTriggers, &runtimev1.RefreshModelTrigger{
@@ -152,6 +205,10 @@ func RefreshCmd(ch *cmdutil.Helper) *cobra.Command {
 	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().BoolVar(&skippedPartitions, "skipped-partitions", false, "Refresh all skipped model partitions (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(&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")
@@ -161,3 +218,70 @@ func RefreshCmd(ch *cmdutil.Helper) *cobra.Command {
 
 	return refreshCmd
 }
+
+// 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
+	for {
+		res, err := rt.GetModelPartitions(ctx, &runtimev1.GetModelPartitionsRequest{
+			InstanceId: instanceID,
+			Model:      model,
+			Pending:    pending,
+			Errored:    errored,
+			Skipped:    skipped,
+			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
+	}
+
+	// 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
+}

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 [<project>] <model>",
@@ -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

docs/docs/reference/cli/project/refresh.md

@@ -13,22 +13,26 @@ rill project refresh [<project-name>] [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

docs/docs/reference/cli/project/skip-partition.md

@@ -17,14 +17,18 @@ rill project skip-partition [<project>] <model> [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