feat: support schema evolution DDLs (ADD/DROP/RENAME/ALTER COLUMN)

docs/src/operations/ddl/alter-table.md

@@ -45,6 +45,55 @@ CREATE TABLE users (id BIGINT, name STRING)
 ALTER TABLE users SET TBLPROPERTIES ('enable_stable_row_ids' = 'true');
 ```
 
+## ADD COLUMN
+
+Add one or more top-level columns. New columns are appended and filled with `NULL` for existing rows:
+
+```sql
+ALTER TABLE users ADD COLUMN age INT;
+ALTER TABLE users ADD COLUMNS (age INT, email STRING);
+```
+
+## DROP COLUMN
+
+Remove a column from the table:
+
+```sql
+ALTER TABLE users DROP COLUMN age;
+ALTER TABLE users DROP COLUMN IF EXISTS age;
+```
+
+## RENAME COLUMN
+
+Rename a column:
+
+```sql
+ALTER TABLE users RENAME COLUMN name TO full_name;
+```
+
+## ALTER COLUMN
+
+Change a column's nullability:
+
+```sql
+ALTER TABLE users ALTER COLUMN id DROP NOT NULL;
+```
+
+!!! note
+Column schema evolution operates on top-level columns. Each `ALTER TABLE` is committed as a single
+atomic Lance operation against the current schema, so one statement may batch changes of the same
+kind that target distinct columns (e.g. adding several columns), but it may not mix column
+additions, drops, and alterations, combine a column change with `TBLPROPERTIES` changes, target the
+same column more than once, or apply a change that depends on an earlier change in the same
+statement (such as altering a column by its just-assigned new name) — issue those as separate
+statements. The whole request is validated first, so if any change is unsupported, none is applied.
+The following are **not** currently supported and are rejected before any change is written:
+
+- Adding a column at a specific position (`FIRST`/`AFTER`) — columns are always appended.
+- Adding a column with a `DEFAULT` value — new columns are filled with `NULL`.
+- Adding a column to a legacy-format (`file_format_version='LEGACY'`) table.
+- Changing a column's data type (`ALTER COLUMN ... TYPE`) or updating a column comment.
+
 ## Rename Table
 
 Rename a table within the same namespace:

integration-tests/test_lance_spark.py

@@ -587,6 +587,63 @@ def test_properties_persist_after_insert(self, spark):
         assert props["team"] == "data-eng"
 
 
+class TestDDLAlterTableColumns:
+    """Test ALTER TABLE schema evolution (ADD/DROP/RENAME/ALTER COLUMN)."""
+
+    def test_add_column(self, spark):
+        """ADD COLUMN appends a nullable column filled with NULL for existing rows."""
+        spark.sql("CREATE TABLE default.test_table (id INT, name STRING)")
+        spark.sql("INSERT INTO default.test_table VALUES (1, 'Alice')")
+        spark.sql("ALTER TABLE default.test_table ADD COLUMN age INT")
+
+        columns = spark.table("default.test_table").columns
+        assert "age" in columns
+
+        row = spark.sql("SELECT id, name, age FROM default.test_table").collect()[0]
+        assert row.id == 1
+        assert row.name == "Alice"
+        assert row.age is None
+
+    def test_drop_column(self, spark):
+        """DROP COLUMN removes the column from the schema and results."""
+        spark.sql("CREATE TABLE default.test_table (id INT, name STRING, age INT)")
+        spark.sql("INSERT INTO default.test_table VALUES (1, 'Alice', 30)")
+        spark.sql("ALTER TABLE default.test_table DROP COLUMN age")
+
+        columns = spark.table("default.test_table").columns
+        assert "age" not in columns
+        assert columns == ["id", "name"]
+
+    def test_drop_column_if_exists_missing(self, spark):
+        """DROP COLUMN IF EXISTS on a missing column is a no-op."""
+        spark.sql("CREATE TABLE default.test_table (id INT, name STRING)")
+        spark.sql("ALTER TABLE default.test_table DROP COLUMN IF EXISTS missing")
+
+        assert spark.table("default.test_table").columns == ["id", "name"]
+
+    def test_rename_column(self, spark):
+        """RENAME COLUMN renames the column while preserving data."""
+        spark.sql("CREATE TABLE default.test_table (id INT, name STRING)")
+        spark.sql("INSERT INTO default.test_table VALUES (1, 'Alice')")
+        spark.sql("ALTER TABLE default.test_table RENAME COLUMN name TO full_name")
+
+        columns = spark.table("default.test_table").columns
+        assert "full_name" in columns
+        assert "name" not in columns
+
+        row = spark.sql("SELECT id, full_name FROM default.test_table").collect()[0]
+        assert row.full_name == "Alice"
+
+    def test_alter_column_drop_not_null(self, spark):
+        """ALTER COLUMN DROP NOT NULL relaxes a column's nullability."""
+        spark.sql("CREATE TABLE default.test_table (id INT NOT NULL, name STRING)")
+        assert spark.table("default.test_table").schema["id"].nullable is False
+
+        spark.sql("ALTER TABLE default.test_table ALTER COLUMN id DROP NOT NULL")
+
+        assert spark.table("default.test_table").schema["id"].nullable is True
+
+
 class TestDDLColumnCompression:
     """Test per-column compression TBLPROPERTIES → Arrow field metadata pipeline."""
 

lance-spark-base_2.12/src/main/java/org/lance/spark/BaseLanceNamespaceSparkCatalog.java

@@ -916,6 +916,7 @@ private void deregisterQuietly(List<String> tableIdList) {
   public Table alterTable(Identifier ident, TableChange... changes) throws NoSuchTableException {
     Map<String, String> propsToSet = new HashMap<>();
     Set<String> keysToRemove = new HashSet<>();
+    List<TableChange.ColumnChange> columnChanges = new ArrayList<>();
 
     for (TableChange change : changes) {
       if (change instanceof TableChange.SetProperty) {
@@ -924,11 +925,13 @@ public Table alterTable(Identifier ident, TableChange... changes) throws NoSuchT
       } else if (change instanceof TableChange.RemoveProperty) {
         TableChange.RemoveProperty removeProp = (TableChange.RemoveProperty) change;
         keysToRemove.add(removeProp.property());
+      } else if (change instanceof TableChange.ColumnChange) {
+        columnChanges.add((TableChange.ColumnChange) change);
       } else {
         throw new UnsupportedOperationException(
             "Unsupported table change type: "
                 + change.getClass().getSimpleName()
-                + ". Only SET/UNSET TBLPROPERTIES is supported.");
+                + ". Only SET/UNSET TBLPROPERTIES and column schema evolution are supported.");
       }
     }
 
@@ -939,23 +942,40 @@ public Table alterTable(Identifier ident, TableChange... changes) throws NoSuchT
               + " can only be set at table creation.");
     }
 
-    if (propsToSet.isEmpty() && keysToRemove.isEmpty()) {
+    boolean hasPropertyChange = !propsToSet.isEmpty() || !keysToRemove.isEmpty();
+
+    if (!hasPropertyChange && columnChanges.isEmpty()) {
       // No changes to apply, just return the current table
       return loadTable(ident);
     }
 
+    // Column schema evolution and property updates commit through separate core mutations, so a
+    // request mixing them cannot be applied atomically. Reject it before writing anything rather
+    // than risk a partially-applied ALTER TABLE.
+    if (hasPropertyChange && !columnChanges.isEmpty()) {
+      throw new UnsupportedOperationException(
+          "A single ALTER TABLE cannot combine column schema evolution with TBLPROPERTIES "
+              + "changes; issue them as separate statements.");
+    }
+
     ResolvedTable resolved = resolveIdentifier(ident);
 
     try (Dataset dataset = Utils.openDatasetBuilder(resolved.readOptions).build()) {
-      // Dataset.updateConfig uses replace semantics (overwrites entire config),
-      // so we must read-merge-write to preserve existing properties.
-      Map<String, String> merged = new HashMap<>(dataset.getConfig());
-      merged.putAll(propsToSet);
-      keysToRemove.forEach(merged::remove);
-      boolean managedVersioning =
-          resolved.describeResponse != null
-              && Boolean.TRUE.equals(resolved.describeResponse.getManagedVersioning());
-      updateDatasetConfig(dataset, merged, managedVersioning, resolved.tableIdList);
+      if (!columnChanges.isEmpty()) {
+        // Schema-evolution changes commit through the dataset's own handler, which
+        // openDatasetBuilder wires for managed versioning when applicable.
+        LanceSchemaEvolution.apply(dataset, columnChanges);
+      } else {
+        // Dataset.updateConfig uses replace semantics (overwrites entire config),
+        // so we must read-merge-write to preserve existing properties.
+        Map<String, String> merged = new HashMap<>(dataset.getConfig());
+        merged.putAll(propsToSet);
+        keysToRemove.forEach(merged::remove);
+        boolean managedVersioning =
+            resolved.describeResponse != null
+                && Boolean.TRUE.equals(resolved.describeResponse.getManagedVersioning());
+        updateDatasetConfig(dataset, merged, managedVersioning, resolved.tableIdList);
+      }
     }
 
     return loadTable(ident);