feat: add `catalog_checkpoint_interval` lifecycle config

data_types/src/database_rules.rs

@@ -151,6 +151,9 @@ pub struct LifecycleRules {
     /// If the background worker doesn't find anything to do it
     /// will sleep for this many milliseconds before looking again
     pub worker_backoff_millis: Option<NonZeroU64>,
+
+    /// After how many transactions should IOx write a new checkpoint?
+    pub catalog_checkpoint_interval: Option<NonZeroU64>,
 }
 
 /// This struct specifies the rules for the order to sort partitions

generated_types/protos/influxdata/iox/management/v1/database_rules.proto

@@ -110,6 +110,11 @@ message LifecycleRules {
   // If 0, the default backoff is used
   // See server::db::lifecycle::DEFAULT_LIFECYCLE_BACKOFF
   uint64 worker_backoff_millis = 10;
+
+  // After how many transactions should IOx write a new checkpoint?
+  //
+  // If 0, no checkpoints will be written.
+  uint64 catalog_checkpoint_interval = 11;
 }
 
 message DatabaseRules {

generated_types/src/database_rules/lifecycle.rs

@@ -35,6 +35,9 @@ impl From<LifecycleRules> for management::LifecycleRules {
             persist: config.persist,
             immutable: config.immutable,
             worker_backoff_millis: config.worker_backoff_millis.map_or(0, NonZeroU64::get),
+            catalog_checkpoint_interval: config
+                .catalog_checkpoint_interval
+                .map_or(0, NonZeroU64::get),
         }
     }
 }
@@ -54,6 +57,7 @@ impl TryFrom<management::LifecycleRules> for LifecycleRules {
             persist: proto.persist,
             immutable: proto.immutable,
             worker_backoff_millis: NonZeroU64::new(proto.worker_backoff_millis),
+            catalog_checkpoint_interval: NonZeroU64::new(proto.catalog_checkpoint_interval),
         })
     }
 }
@@ -141,6 +145,7 @@ mod tests {
             persist: true,
             immutable: true,
             worker_backoff_millis: 1000,
+            catalog_checkpoint_interval: 10,
         };
 
         let config: LifecycleRules = protobuf.clone().try_into().unwrap();

src/commands/database.rs

@@ -113,6 +113,12 @@ struct Create {
     /// Do not allow writing new data to this database
     #[structopt(long)]
     immutable: bool,
+
+    /// After how many transactions should IOx write a new checkpoint?
+    ///
+    /// If 0, no checkpoints will be written.
+    #[structopt(long, default_value = "100")]
+    catalog_checkpoint_interval: u64,
 }
 
 /// Get list of databases
@@ -181,6 +187,7 @@ pub async fn command(url: String, config: Config) -> Result<()> {
                     persist: command.persist,
                     immutable: command.immutable,
                     worker_backoff_millis: Default::default(),
+                    catalog_checkpoint_interval: command.catalog_checkpoint_interval,
                 }),
 
                 // Default to hourly partitions