fix fmt, clippy, rebase fixes, and small test fixes

misc/python/materialize/parallel_workload/database.py

@@ -9,11 +9,10 @@
 
 import random
 import threading
-from collections.abc import Iterator, Sequence
+from collections.abc import Iterator
 from copy import copy
 from enum import Enum
 
-import psycopg
 from pg8000.native import identifier, literal
 
 from materialize.data_ingest.data_type import (
@@ -196,7 +195,7 @@ def create(self, exe: Executor) -> None:
 
 
 class DBObject:
-    columns: Sequence[Column]
+    columns: list[Column]
     lock: threading.Lock
 
     def __init__(self):

src/adapter/src/catalog.rs

@@ -1447,8 +1447,8 @@ impl Catalog {
                 .iter()
                 .map(|(id, _)| id)
                 .chain(new_global_expressions.iter().map(|(id, _)| id))
-                .filter_map(|id| self.get_entry_by_global_id(id).index())
-                .map(|index| index.on);
+                .map(|id| self.get_entry_by_global_id(id))
+                .filter_map(|entry| entry.index().map(|index| index.on));
             let invalidate_ids = self.invalidate_for_index(ons);
             expr_cache
                 .update(

src/adapter/src/catalog/state.rs

@@ -821,9 +821,9 @@ impl CatalogState {
     /// Returns the [`RelationDesc`] for a [`GlobalId`], if the provided [`GlobalId`] refers to an
     /// object that returns rows.
     pub fn try_get_desc_by_global_id(&self, id: &GlobalId) -> Option<Cow<RelationDesc>> {
-        let entry = self.try_get_entry_by_global_id(&id)?;
+        let entry = self.try_get_entry_by_global_id(id)?;
         let desc = match entry.item() {
-            CatalogItem::Table(table) => Cow::Owned(table.desc_for(&id)),
+            CatalogItem::Table(table) => Cow::Owned(table.desc_for(id)),
             // TODO(alter_table): Support schema evolution on sources.
             other => other.desc_opt(RelationVersionSelector::Latest)?,
         };

src/adapter/src/coord/sequencer/inner.rs

@@ -4775,7 +4775,8 @@ impl Coordinator {
         } = plan;
 
         // TODO(alter_table): Support allocating GlobalIds without a CatalogItemId.
-        let (_, new_global_id) = self.catalog.allocate_user_id().await?;
+        let id_ts = self.get_catalog_write_ts().await;
+        let (_, new_global_id) = self.catalog.allocate_user_id(id_ts).await?;
         let ops = vec![catalog::Op::AlterAddColumn {
             id: relation_id,
             new_global_id,

src/adapter/src/optimize/dataflows.rs

@@ -211,7 +211,7 @@ impl<'a> DataflowBuilder<'a> {
                 let entry = self.catalog.get_entry(id);
                 match entry.item() {
                     CatalogItem::Table(table) => {
-                        dataflow.import_source(*id, table.desc_for(&id).typ().clone(), monotonic);
+                        dataflow.import_source(*id, table.desc_for(id).typ().clone(), monotonic);
                     }
                     CatalogItem::Source(source) => {
                         dataflow.import_source(*id, source.desc.typ().clone(), monotonic);

src/sql/src/catalog.rs

@@ -722,10 +722,7 @@ pub trait CatalogItem {
 
     /// Returns the [`CatalogCollectionItem`] for a specific version of this
     /// [`CatalogItem`].
-    fn at_version(
-        &self,
-        version: RelationVersionSelector,
-    ) -> Box<dyn CatalogCollectionItem>;
+    fn at_version(&self, version: RelationVersionSelector) -> Box<dyn CatalogCollectionItem>;
 
     /// The latest version of this item, if it's version-able.
     fn latest_version(&self) -> Option<RelationVersion>;

src/sql/src/names.rs

@@ -1536,8 +1536,8 @@ impl<'a> NameResolver<'a> {
         };
         let alter_table_enabled = self.catalog.system_vars().enable_alter_table_add_column();
         let version = match version {
-            // If there isn't a version specified, and this item supports
-            // versioning, track the latest.
+            // If there isn't a version specified, and this item supports versioning, track the
+            // latest.
             None => match item.latest_version() {
                 // Only track the version of the referenced object, if the feature is enabled.
                 Some(v) if alter_table_enabled => RelationVersionSelector::Specific(v),

src/sql/src/plan/statement/ddl.rs

@@ -444,7 +444,14 @@ pub fn plan_create_table(
 
     let create_sql = normalize::create_statement(scx, Statement::CreateTable(stmt.clone()))?;
 
-    let options = plan_table_options(scx, &desc, with_options.clone())?;
+    // Table options should only consider the original columns, since those
+    // were the only ones in scope when the table was created.
+    //
+    // TODO(alter_table): Will need to reconsider this when we support ALTERing
+    // the PARTITION BY columns.
+    let original_desc = desc.at_version(RelationVersionSelector::Specific(RelationVersion::root()));
+    let options = plan_table_options(scx, &original_desc, with_options.clone())?;
+
     let compaction_window = options.iter().find_map(|o| {
         #[allow(irrefutable_let_patterns)]
         if let crate::plan::TableOption::RetainHistory(lcw) = o {

src/sql/src/session/vars/definitions.rs

@@ -2150,7 +2150,7 @@ feature_flags!(
         name: enable_alter_table_add_column,
         desc: "Enable ALTER TABLE ... ADD COLUMN ...",
         default: false,
-        enable_for_item_parsing: true,
+        enable_for_item_parsing: false,
     },
     {
         name: enable_graceful_cluster_reconfiguration,

src/storage-controller/src/persist_handles.rs

@@ -289,8 +289,10 @@ impl<T: Timestamp + Lattice + Codec64 + TimestampManipulation> TxnsTableWorker<T
                     tx,
                 } => {
                     async {
-                        self.drop_handles(vec![existing_collection], forget_ts).await;
-                        self.register(register_ts, vec![(new_collection, handle)]).await;
+                        self.drop_handles(vec![existing_collection], forget_ts)
+                            .await;
+                        self.register(register_ts, vec![(new_collection, handle)])
+                            .await;
                     }
                     .instrument(span)
                     .await;

src/testdrive/src/action/consistency.rs

@@ -220,6 +220,7 @@ async fn check_catalog_state(state: &State) -> Result<(), anyhow::Error> {
         // first lines that differs and show context around it.
         let diff = similar::TextDiff::from_lines(&memory_catalog, &disk_catalog)
             .unified_diff()
+            .header("memory", "disk")
             .context_radius(50)
             .to_string()
             .lines()

test/legacy-upgrade/check-from-v0.126.0-alter-table.td

@@ -40,17 +40,17 @@
 > SELECT * FROM alter_table_t1_v2 ORDER BY a DESC;
 <null> <null>
 600    orange
-500    apple 
-400    world 
+500    apple
+400    world
 300    <null>
-200    hello 
+200    hello
 100    <null>
 
 > SELECT * FROM alter_table_t1_mv2 ORDER BY a DESC;
 <null> <null>
 600    orange
-500    apple 
-400    world 
+500    apple
+400    world
 300    <null>
-200    hello 
+200    hello
 100    <null>

test/sqllogictest/alter-table.slt

@@ -489,4 +489,3 @@ DROP TABLE t2 CASCADE;
 query TTIT
 SELECT * FROM mz_internal.mz_comments;
 ----
-