persist: Allow appending Datum::Null to Codec Row data

src/persist-client/src/cfg.rs

@@ -370,6 +370,7 @@ pub fn all_dyncfgs(configs: ConfigSet) -> ConfigSet {
         .add(&crate::stats::STATS_UNTRIMMABLE_COLUMNS_PREFIX)
         .add(&crate::stats::STATS_UNTRIMMABLE_COLUMNS_SUFFIX)
         .add(&crate::fetch::PART_DECODE_FORMAT)
+        .add(&crate::DANGEROUS_ENABLE_SCHEMA_EVOLUTION)
 }
 
 impl PersistConfig {

src/persist-client/src/lib.rs

@@ -23,7 +23,7 @@ use std::sync::Arc;
 use differential_dataflow::difference::Semigroup;
 use differential_dataflow::lattice::Lattice;
 use mz_build_info::{build_info, BuildInfo};
-use mz_dyncfg::ConfigSet;
+use mz_dyncfg::{Config, ConfigSet};
 use mz_ore::{instrument, soft_assert_or_log};
 use mz_persist::location::{Blob, Consensus, ExternalError};
 use mz_persist_types::{Codec, Codec64, Opaque};
@@ -141,6 +141,16 @@ mod internal {
 /// Persist build information.
 pub const BUILD_INFO: BuildInfo = build_info!();
 
+pub(crate) const DANGEROUS_ENABLE_SCHEMA_EVOLUTION: Config<bool> = Config::new(
+    "persist_dangerous_enable_schema_evolution",
+    false,
+    "\
+DANGEROUS DO NOT ENABLE IN PRODUCTION OR STAGING ENVIRONMENTS!
+
+Enable evolving the schema of a Persist shard. Currently dangerous because \
+compaction does not yet handle batches of data with different schemas.",
+);
+
 // Re-export for convenience.
 pub use mz_persist_types::{PersistLocation, ShardId};
 
@@ -639,6 +649,10 @@ impl PersistClient {
         T: Timestamp + Lattice + Codec64,
         D: Semigroup + Codec64 + Send + Sync,
     {
+        if !DANGEROUS_ENABLE_SCHEMA_EVOLUTION.get(&self.cfg.configs) {
+            panic!("tried to evolve the schema of a Persist shard without the feature enabled");
+        }
+
         let mut machine = self
             .make_machine::<K, V, T, D>(shard_id, diagnostics)
             .await?;

src/repr/src/row.rs

@@ -130,15 +130,20 @@ impl Row {
             col_idx += 1;
         }
 
-        // TODO(parkmycar): This is where we'll add new default values to Codec
-        // data whose upstream relation has had columns added.
+        let num_columns = desc.typ().column_types.len();
+        if col_idx < num_columns {
+            let missing_columns = col_idx..num_columns;
+            for _ in missing_columns {
+                packer.push(Datum::Null);
+                col_idx += 1;
+            }
+        }
 
         // HACK(parkmycar): Only validate that the decoded Row matches the RelationDesc if it was
         // non-empty. We have an optimization for queries like COUNT(*) that returns a fake empty
         // Part instead of decoding the data, which this assertion will fail on.
         //
         // TODO(#28146): Remove the check for if the num_columns is 0.
-        let num_columns = desc.typ().column_types.len();
         if num_columns != 0 && col_idx != 0 {
             mz_ore::soft_assert_eq_or_log!(
                 col_idx,