diff --git a/docs/src/operations/ddl/alter-table.md b/docs/src/operations/ddl/alter-table.md index 54412b2f5..7ca9d0593 100644 --- a/docs/src/operations/ddl/alter-table.md +++ b/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: diff --git a/integration-tests/test_lance_spark.py b/integration-tests/test_lance_spark.py index 995760be2..c8712749b 100644 --- a/integration-tests/test_lance_spark.py +++ b/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.""" diff --git a/lance-spark-base_2.12/src/main/java/org/lance/spark/BaseLanceNamespaceSparkCatalog.java b/lance-spark-base_2.12/src/main/java/org/lance/spark/BaseLanceNamespaceSparkCatalog.java index 435e1b6eb..ecd442cfa 100644 --- a/lance-spark-base_2.12/src/main/java/org/lance/spark/BaseLanceNamespaceSparkCatalog.java +++ b/lance-spark-base_2.12/src/main/java/org/lance/spark/BaseLanceNamespaceSparkCatalog.java @@ -916,6 +916,7 @@ private void deregisterQuietly(List tableIdList) { public Table alterTable(Identifier ident, TableChange... changes) throws NoSuchTableException { Map propsToSet = new HashMap<>(); Set keysToRemove = new HashSet<>(); + List 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 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 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); diff --git a/lance-spark-base_2.12/src/main/java/org/lance/spark/LanceSchemaEvolution.java b/lance-spark-base_2.12/src/main/java/org/lance/spark/LanceSchemaEvolution.java new file mode 100644 index 000000000..ad753171e --- /dev/null +++ b/lance-spark-base_2.12/src/main/java/org/lance/spark/LanceSchemaEvolution.java @@ -0,0 +1,304 @@ +/* + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.lance.spark; + +import org.lance.Dataset; +import org.lance.schema.ColumnAlteration; +import org.lance.spark.utils.FieldPathUtils; + +import org.apache.arrow.vector.types.pojo.Field; +import org.apache.arrow.vector.types.pojo.Schema; +import org.apache.spark.sql.connector.catalog.TableChange.AddColumn; +import org.apache.spark.sql.connector.catalog.TableChange.ColumnChange; +import org.apache.spark.sql.connector.catalog.TableChange.DeleteColumn; +import org.apache.spark.sql.connector.catalog.TableChange.RenameColumn; +import org.apache.spark.sql.connector.catalog.TableChange.UpdateColumnNullability; +import org.apache.spark.sql.connector.catalog.TableChange.UpdateColumnType; +import org.apache.spark.sql.types.MetadataBuilder; +import org.apache.spark.sql.types.StructField; +import org.apache.spark.sql.types.StructType; +import org.apache.spark.sql.util.LanceArrowUtils; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashSet; +import java.util.LinkedHashSet; +import java.util.List; +import java.util.Set; + +/** + * Translates a Spark {@link ColumnChange} schema-evolution request (produced by {@code ALTER TABLE + * ADD/DROP/RENAME/ALTER COLUMN}) into a single atomic {@link Dataset} mutation. + * + *

To preserve Spark's ordered, all-or-nothing {@code alterTable} contract, an accepted request + * is committed through exactly one core operation ({@code addColumns}, {@code dropColumns}, or + * {@code alterColumns} — each commits its whole batch atomically against the current + * schema with no ordering between the batched entries). Because the batch cannot express + * changes that depend on an earlier change in the same request, the request is validated against + * the current schema and the following are rejected before anything is written, so a + * rejected request never leaves the table partially mutated: + * + *

    + *
  • requests that would require more than one core mutation (mixing additions, drops, and + * alterations, since heterogeneous batching is not supported by the core yet); + *
  • requests where two changes target the same column (their order would be lost when collapsed + * into one batch); + *
  • nested (multi-part) column paths, since validation is top-level only. + *
+ */ +final class LanceSchemaEvolution { + + /** Lance file format version string for the legacy ("0.1") format. */ + private static final String LEGACY_FILE_FORMAT_VERSION = "0.1"; + + /** The single core mutation an accepted request compiles down to. */ + private enum Kind { + ADD, + DROP, + ALTER + } + + private LanceSchemaEvolution() {} + + static void apply(Dataset dataset, List changes) { + if (changes.isEmpty()) { + return; + } + + boolean legacyFormat = LEGACY_FILE_FORMAT_VERSION.equals(dataset.getLanceFileFormatVersion()); + Set currentFields = topLevelFieldNames(dataset); + + // Validate the whole request against the current schema and confirm it compiles to a single + // core mutation before writing anything. Because the batched core operation applies to the + // current schema with no ordering between entries, a column may be targeted at most once and + // every referenced column is checked against the current schema (not an evolving one). + // `occupied` additionally simulates the request in order (removing rename sources, inserting + // destinations) so ordered rename semantics — e.g. rejecting a rename whose destination name is + // already taken at that step — are preserved even though the core batch itself is unordered. + Kind kind = null; + Set touched = new HashSet<>(); + Set occupied = new LinkedHashSet<>(currentFields); + for (ColumnChange change : changes) { + Kind changeKind = validate(change, currentFields, touched, occupied, legacyFormat); + if (kind == null) { + kind = changeKind; + } else if (kind != changeKind) { + throw new UnsupportedOperationException( + "A single ALTER TABLE cannot mix column additions, drops, and alterations; " + + "issue them as separate statements."); + } + } + + switch (kind) { + case ADD: + applyAdds(dataset, changes); + break; + case DROP: + applyDrops(dataset, changes, currentFields); + break; + case ALTER: + applyAlters(dataset, changes); + break; + default: + throw new IllegalStateException("Unexpected change kind: " + kind); + } + } + + private static Kind validate( + ColumnChange change, + Set currentFields, + Set touched, + Set occupied, + boolean legacyFormat) { + if (change instanceof AddColumn) { + AddColumn add = (AddColumn) change; + requireTopLevel(add.fieldNames(), "Adding nested columns"); + if (add.position() != null) { + throw new UnsupportedOperationException( + "ADD COLUMN with FIRST/AFTER position is not supported; columns are appended."); + } + if (add.defaultValue() != null) { + throw new UnsupportedOperationException( + "ADD COLUMN with a DEFAULT value is not supported; new columns are filled with NULL."); + } + if (legacyFormat) { + throw new UnsupportedOperationException( + "ADD COLUMN is not supported on legacy-format (" + + LEGACY_FILE_FORMAT_VERSION + + ") tables."); + } + String name = add.fieldNames()[0]; + if (currentFields.contains(name)) { + throw new UnsupportedOperationException("Cannot add existing column: " + name); + } + requireDistinct(touched, name); + return Kind.ADD; + } else if (change instanceof DeleteColumn) { + DeleteColumn delete = (DeleteColumn) change; + requireTopLevel(delete.fieldNames(), "Dropping nested columns"); + String name = delete.fieldNames()[0]; + // Resolve absence first: DROP COLUMN IF EXISTS on a missing column is a no-op regardless of + // how the absent name is spelled, so its representability must not be considered. + boolean present = currentFields.contains(name); + if (!present) { + if (!delete.ifExists()) { + throw new UnsupportedOperationException("Cannot drop missing column: " + name); + } + } else if (name.indexOf('`') >= 0) { + // Only an existing column actually reaches dropColumns, and the current core drop API + // cannot resolve a backtick-containing name under either the canonical (escaped) or raw + // representation. Reject it up front instead of failing mid-commit with "field not found". + throw new UnsupportedOperationException( + "Dropping a column whose name contains a backtick is not supported by the current " + + "Lance version: " + + name); + } + requireDistinct(touched, name); + return Kind.DROP; + } else if (change instanceof RenameColumn) { + RenameColumn rename = (RenameColumn) change; + requireTopLevel(rename.fieldNames(), "Renaming nested columns"); + String source = rename.fieldNames()[0]; + requireExists(currentFields, source); + requireDistinct(touched, source); + // Simulate the rename in request order: the destination must not already be occupied at this + // step (by an original column or an earlier rename's destination). This preserves Spark's + // ordered semantics even though the core alterColumns batch applies without ordering. + occupied.remove(source); + if (!occupied.add(rename.newName())) { + throw new UnsupportedOperationException( + "Cannot rename column '" + + source + + "' to '" + + rename.newName() + + "': the target name is already in use."); + } + return Kind.ALTER; + } else if (change instanceof UpdateColumnNullability) { + UpdateColumnNullability updateNull = (UpdateColumnNullability) change; + requireTopLevel(updateNull.fieldNames(), "Altering nested columns"); + requireExists(currentFields, updateNull.fieldNames()[0]); + requireDistinct(touched, updateNull.fieldNames()[0]); + return Kind.ALTER; + } else if (change instanceof UpdateColumnType) { + UpdateColumnType updateType = (UpdateColumnType) change; + // The current lance-core JNI drops the cast target type on the way to Rust, silently + // turning a type change into a no-op. Reject it explicitly rather than lying about it. + throw new UnsupportedOperationException( + "Changing the type of column '" + + String.join(".", updateType.fieldNames()) + + "' is not supported by the current Lance version."); + } else { + throw new UnsupportedOperationException( + "Unsupported column change type: " + change.getClass().getSimpleName()); + } + } + + private static void applyAdds(Dataset dataset, List changes) { + List fields = new ArrayList<>(changes.size()); + for (ColumnChange change : changes) { + AddColumn add = (AddColumn) change; + MetadataBuilder metadataBuilder = new MetadataBuilder(); + if (add.comment() != null) { + metadataBuilder.putString("comment", add.comment()); + } + fields.add( + new StructField( + add.fieldNames()[0], add.dataType(), add.isNullable(), metadataBuilder.build())); + } + Schema arrowSchema = + LanceArrowUtils.toArrowSchema( + new StructType(fields.toArray(new StructField[0])), "UTC", true); + dataset.addColumns(arrowSchema.getFields()); + } + + private static void applyDrops( + Dataset dataset, List changes, Set currentFields) { + List toDrop = new ArrayList<>(changes.size()); + for (ColumnChange change : changes) { + DeleteColumn delete = (DeleteColumn) change; + String name = delete.fieldNames()[0]; + // Validation already guaranteed a non-ifExists drop targets an existing column; skip a + // missing IF EXISTS target so the core is never asked to drop a nonexistent column. + if (delete.ifExists() && !currentFields.contains(name)) { + continue; + } + toDrop.add(canonicalPath(name)); + } + if (!toDrop.isEmpty()) { + dataset.dropColumns(toDrop); + } + } + + private static void applyAlters(Dataset dataset, List changes) { + // Each change targets a distinct existing column (enforced during validation), so one + // ColumnAlteration per change keeps the request a single alterColumns commit. + List alterations = new ArrayList<>(changes.size()); + for (ColumnChange change : changes) { + if (change instanceof RenameColumn) { + RenameColumn rename = (RenameColumn) change; + alterations.add( + new ColumnAlteration.Builder(canonicalPath(rename.fieldNames()[0])) + .rename(rename.newName()) + .build()); + } else if (change instanceof UpdateColumnNullability) { + UpdateColumnNullability updateNull = (UpdateColumnNullability) change; + alterations.add( + new ColumnAlteration.Builder(canonicalPath(updateNull.fieldNames()[0])) + .nullable(updateNull.nullable()) + .build()); + } + } + dataset.alterColumns(alterations); + } + + private static void requireTopLevel(String[] fieldNames, String action) { + if (fieldNames.length != 1) { + throw new UnsupportedOperationException( + action + " is not supported: " + String.join(".", fieldNames)); + } + } + + private static void requireExists(Set currentFields, String name) { + if (!currentFields.contains(name)) { + throw new UnsupportedOperationException("Cannot alter missing column: " + name); + } + } + + /** + * Escapes a top-level column name into a canonical Lance field path so names containing path + * syntax (dots, spaces, etc.) resolve correctly. + */ + private static String canonicalPath(String name) { + return FieldPathUtils.canonicalPath(Collections.singletonList(name)); + } + + private static void requireDistinct(Set touched, String name) { + if (!touched.add(name)) { + throw new UnsupportedOperationException( + "Column '" + + name + + "' is targeted by more than one change in the same ALTER TABLE; " + + "issue the changes as separate statements."); + } + } + + private static Set topLevelFieldNames(Dataset dataset) { + Set names = new LinkedHashSet<>(); + for (Field field : dataset.getSchema().getFields()) { + names.add(field.getName()); + } + return names; + } +} diff --git a/lance-spark-base_2.12/src/test/java/org/lance/spark/SparkLanceNamespaceTestBase.java b/lance-spark-base_2.12/src/test/java/org/lance/spark/SparkLanceNamespaceTestBase.java index 69fc405b7..9c44d8a95 100644 --- a/lance-spark-base_2.12/src/test/java/org/lance/spark/SparkLanceNamespaceTestBase.java +++ b/lance-spark-base_2.12/src/test/java/org/lance/spark/SparkLanceNamespaceTestBase.java @@ -31,6 +31,8 @@ import org.apache.spark.sql.connector.catalog.TableCatalog; import org.apache.spark.sql.connector.catalog.TableChange; import org.apache.spark.sql.connector.catalog.functions.UnboundFunction; +import org.apache.spark.sql.types.DataTypes; +import org.apache.spark.sql.types.StructType; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; @@ -39,6 +41,7 @@ import java.io.IOException; import java.nio.file.Path; import java.time.format.DateTimeFormatter; +import java.util.Arrays; import java.util.Collections; import java.util.HashMap; import java.util.List; @@ -953,16 +956,411 @@ public void testUnsupportedTableChangeThrows() throws Exception { spark.sql("CREATE TABLE " + fullName + " (id BIGINT NOT NULL, name STRING)"); Identifier ident = Identifier.of(new String[] {"default"}, tableName); + // Updating a column comment is a recognized column change that Lance does not support. UnsupportedOperationException ex = assertThrows( UnsupportedOperationException.class, () -> { catalog.alterTable( - ident, - TableChange.addColumn( - new String[] {"new_col"}, org.apache.spark.sql.types.DataTypes.StringType)); + ident, TableChange.updateColumnComment(new String[] {"id"}, "the id")); }); - assertTrue(ex.getMessage().contains("Only SET/UNSET TBLPROPERTIES is supported")); + assertTrue(ex.getMessage().contains("Unsupported column change type: UpdateColumnComment")); + } + + @Test + public void testAddColumn() throws Exception { + String tableName = generateTableName("add_column"); + String fullName = catalogName + ".default." + tableName; + + spark.sql("CREATE TABLE " + fullName + " (id BIGINT NOT NULL, name STRING)"); + spark.sql("INSERT INTO " + fullName + " VALUES (1, 'Alice')"); + spark.sql("ALTER TABLE " + fullName + " ADD COLUMN age INT"); + + StructType schema = spark.table(fullName).schema(); + assertTrue(Arrays.asList(schema.fieldNames()).contains("age")); + + Row row = spark.sql("SELECT id, name, age FROM " + fullName).collectAsList().get(0); + assertEquals(1L, row.getLong(0)); + assertEquals("Alice", row.getString(1)); + assertTrue(row.isNullAt(2)); + } + + @Test + public void testDropColumn() throws Exception { + String tableName = generateTableName("drop_column"); + String fullName = catalogName + ".default." + tableName; + + spark.sql("CREATE TABLE " + fullName + " (id BIGINT NOT NULL, name STRING, age INT)"); + spark.sql("INSERT INTO " + fullName + " VALUES (1, 'Alice', 30)"); + spark.sql("ALTER TABLE " + fullName + " DROP COLUMN age"); + + StructType schema = spark.table(fullName).schema(); + assertFalse(Arrays.asList(schema.fieldNames()).contains("age")); + + Row row = spark.sql("SELECT * FROM " + fullName).collectAsList().get(0); + assertEquals(2, row.length()); + assertEquals(1L, row.getLong(0)); + assertEquals("Alice", row.getString(1)); + } + + @Test + public void testRenameColumn() throws Exception { + String tableName = generateTableName("rename_column"); + String fullName = catalogName + ".default." + tableName; + + spark.sql("CREATE TABLE " + fullName + " (id BIGINT NOT NULL, name STRING)"); + spark.sql("INSERT INTO " + fullName + " VALUES (1, 'Alice')"); + spark.sql("ALTER TABLE " + fullName + " RENAME COLUMN name TO full_name"); + + StructType schema = spark.table(fullName).schema(); + assertTrue(Arrays.asList(schema.fieldNames()).contains("full_name")); + assertFalse(Arrays.asList(schema.fieldNames()).contains("name")); + + Row row = spark.sql("SELECT id, full_name FROM " + fullName).collectAsList().get(0); + assertEquals("Alice", row.getString(1)); + } + + @Test + public void testAlterColumnTypeUnsupported() throws Exception { + String tableName = generateTableName("alter_column_type"); + String fullName = catalogName + ".default." + tableName; + + spark.sql("CREATE TABLE " + fullName + " (id INT NOT NULL, name STRING)"); + + // Changing a column type is not supported by the current Lance version; it must fail loudly + // rather than silently no-op. + Identifier ident = Identifier.of(new String[] {"default"}, tableName); + UnsupportedOperationException ex = + assertThrows( + UnsupportedOperationException.class, + () -> + catalog.alterTable( + ident, TableChange.updateColumnType(new String[] {"id"}, DataTypes.LongType))); + assertTrue(ex.getMessage().contains("Changing the type of column")); + } + + @Test + public void testAlterColumnDropNotNull() throws Exception { + String tableName = generateTableName("alter_column_nullable"); + String fullName = catalogName + ".default." + tableName; + + spark.sql("CREATE TABLE " + fullName + " (id BIGINT NOT NULL, name STRING)"); + assertFalse(spark.table(fullName).schema().apply(0).nullable()); + + spark.sql("ALTER TABLE " + fullName + " ALTER COLUMN id DROP NOT NULL"); + + assertTrue(spark.table(fullName).schema().apply(0).nullable()); + } + + @Test + public void testAlterTableRejectsRequestAtomically() throws Exception { + String tableName = generateTableName("atomic_reject"); + String fullName = catalogName + ".default." + tableName; + + spark.sql("CREATE TABLE " + fullName + " (id BIGINT NOT NULL, name STRING)"); + + // A supported ADD COLUMN batched with an unsupported column-type change must be rejected as a + // whole: neither change may take effect. + Identifier ident = Identifier.of(new String[] {"default"}, tableName); + assertThrows( + UnsupportedOperationException.class, + () -> + catalog.alterTable( + ident, + TableChange.addColumn(new String[] {"added"}, DataTypes.IntegerType), + TableChange.updateColumnType(new String[] {"id"}, DataTypes.LongType))); + + assertFalse(Arrays.asList(catalog.loadTable(ident).schema().fieldNames()).contains("added")); + } + + @Test + public void testAddMultipleColumnsInOneStatement() throws Exception { + String tableName = generateTableName("add_multi"); + String fullName = catalogName + ".default." + tableName; + + spark.sql("CREATE TABLE " + fullName + " (id BIGINT NOT NULL)"); + spark.sql("ALTER TABLE " + fullName + " ADD COLUMNS (age INT, email STRING)"); + + List columns = Arrays.asList(spark.table(fullName).schema().fieldNames()); + assertTrue(columns.contains("age")); + assertTrue(columns.contains("email")); + } + + @Test + public void testAlterTableRejectsMixedColumnChangeKinds() throws Exception { + String tableName = generateTableName("mixed_kinds"); + String fullName = catalogName + ".default." + tableName; + + spark.sql("CREATE TABLE " + fullName + " (id BIGINT NOT NULL, name STRING)"); + + // ADD and DROP compile to different core mutations and cannot be committed atomically together. + Identifier ident = Identifier.of(new String[] {"default"}, tableName); + UnsupportedOperationException ex = + assertThrows( + UnsupportedOperationException.class, + () -> + catalog.alterTable( + ident, + TableChange.addColumn(new String[] {"added"}, DataTypes.IntegerType), + TableChange.deleteColumn(new String[] {"name"}, false))); + assertTrue(ex.getMessage().contains("cannot mix")); + + List columns = Arrays.asList(catalog.loadTable(ident).schema().fieldNames()); + assertFalse(columns.contains("added")); + assertTrue(columns.contains("name")); + } + + @Test + public void testAlterTableRejectsColumnChangeMixedWithProperties() throws Exception { + String tableName = generateTableName("mixed_prop"); + String fullName = catalogName + ".default." + tableName; + + spark.sql("CREATE TABLE " + fullName + " (id BIGINT NOT NULL)"); + + // A column change and a TBLPROPERTIES change are separate commits, so combining them is + // rejected before either is written. + Identifier ident = Identifier.of(new String[] {"default"}, tableName); + UnsupportedOperationException ex = + assertThrows( + UnsupportedOperationException.class, + () -> + catalog.alterTable( + ident, + TableChange.addColumn(new String[] {"added"}, DataTypes.IntegerType), + TableChange.setProperty("k", "v"))); + assertTrue(ex.getMessage().contains("TBLPROPERTIES")); + + assertFalse(Arrays.asList(catalog.loadTable(ident).schema().fieldNames()).contains("added")); + } + + @Test + public void testAlterTableRejectsRenameDependentChange() throws Exception { + String tableName = generateTableName("rename_dependent"); + String fullName = catalogName + ".default." + tableName; + + spark.sql("CREATE TABLE " + fullName + " (id BIGINT NOT NULL, name STRING NOT NULL)"); + + // A change that depends on an earlier rename in the same request cannot be expressed as one + // core batch (which targets the current schema), so it is rejected before any mutation. + Identifier ident = Identifier.of(new String[] {"default"}, tableName); + assertThrows( + UnsupportedOperationException.class, + () -> + catalog.alterTable( + ident, + TableChange.renameColumn(new String[] {"name"}, "full_name"), + TableChange.updateColumnNullability(new String[] {"full_name"}, true))); + + assertArrayEquals(new String[] {"id", "name"}, catalog.loadTable(ident).schema().fieldNames()); + } + + @Test + public void testAlterTableRejectsRepeatedColumnTarget() throws Exception { + String tableName = generateTableName("repeated_target"); + String fullName = catalogName + ".default." + tableName; + + spark.sql("CREATE TABLE " + fullName + " (id BIGINT NOT NULL, name STRING)"); + + // Two changes targeting the same column would lose their order when collapsed into one core + // batch, so the request is rejected. + Identifier ident = Identifier.of(new String[] {"default"}, tableName); + UnsupportedOperationException ex = + assertThrows( + UnsupportedOperationException.class, + () -> + catalog.alterTable( + ident, + TableChange.updateColumnNullability(new String[] {"name"}, false), + TableChange.updateColumnNullability(new String[] {"name"}, true))); + assertTrue(ex.getMessage().contains("more than one change")); + } + + @Test + public void testDropNestedColumnRejected() throws Exception { + String tableName = generateTableName("drop_nested"); + String fullName = catalogName + ".default." + tableName; + + spark.sql("CREATE TABLE " + fullName + " (id BIGINT NOT NULL, s STRUCT)"); + + // Validation is top-level only; a nested path (even with IF EXISTS) is rejected up front + // rather than reaching the core with a path it would error on. + Identifier ident = Identifier.of(new String[] {"default"}, tableName); + assertThrows( + UnsupportedOperationException.class, + () -> + catalog.alterTable( + ident, TableChange.deleteColumn(new String[] {"s", "missing"}, true))); + + assertArrayEquals(new String[] {"id", "s"}, catalog.loadTable(ident).schema().fieldNames()); + } + + @Test + public void testAlterTableRejectsRenameToOccupiedName() throws Exception { + String tableName = generateTableName("rename_occupied"); + String fullName = catalogName + ".default." + tableName; + + spark.sql("CREATE TABLE " + fullName + " (a INT, b INT)"); + + // In Spark's ordered semantics, `a -> b` must fail because `b` already exists at that step, so + // the whole request is rejected before any mutation. + Identifier ident = Identifier.of(new String[] {"default"}, tableName); + assertThrows( + UnsupportedOperationException.class, + () -> + catalog.alterTable( + ident, + TableChange.renameColumn(new String[] {"a"}, "b"), + TableChange.renameColumn(new String[] {"b"}, "c"))); + + assertArrayEquals(new String[] {"a", "b"}, catalog.loadTable(ident).schema().fieldNames()); + } + + @Test + public void testDropColumnWithSpecialCharacterName() throws Exception { + String tableName = generateTableName("drop_special"); + String fullName = catalogName + ".default." + tableName; + + // A top-level column name containing Lance path syntax (here a space) is escaped into a + // canonical field path before it reaches the core. + spark.sql("CREATE TABLE " + fullName + " (`weird name` INT, keep INT)"); + + Identifier ident = Identifier.of(new String[] {"default"}, tableName); + catalog.alterTable(ident, TableChange.deleteColumn(new String[] {"weird name"}, false)); + + assertArrayEquals(new String[] {"keep"}, catalog.loadTable(ident).schema().fieldNames()); + } + + @Test + public void testDropColumnWithBacktickNameRejected() throws Exception { + String tableName = generateTableName("drop_backtick"); + String fullName = catalogName + ".default." + tableName; + + // The current core drop API cannot resolve a backtick-containing name, so the request is + // rejected before any mutation rather than failing mid-commit. + spark.sql("CREATE TABLE " + fullName + " (`a``b` INT, keep INT)"); + + Identifier ident = Identifier.of(new String[] {"default"}, tableName); + UnsupportedOperationException ex = + assertThrows( + UnsupportedOperationException.class, + () -> catalog.alterTable(ident, TableChange.deleteColumn(new String[] {"a`b"}, false))); + assertTrue(ex.getMessage().contains("backtick")); + + assertArrayEquals(new String[] {"a`b", "keep"}, catalog.loadTable(ident).schema().fieldNames()); + } + + @Test + public void testDropColumnIfExistsMissingBacktickNameIsNoOp() throws Exception { + String tableName = generateTableName("drop_missing_backtick"); + String fullName = catalogName + ".default." + tableName; + + spark.sql("CREATE TABLE " + fullName + " (id INT)"); + + // IF EXISTS is an existence contract: a missing name is a no-op regardless of whether the core + // could represent it, so an absent backtick name must not raise. + Identifier ident = Identifier.of(new String[] {"default"}, tableName); + catalog.alterTable(ident, TableChange.deleteColumn(new String[] {"missing`name"}, true)); + + assertArrayEquals(new String[] {"id"}, catalog.loadTable(ident).schema().fieldNames()); + } + + @Test + public void testRenameColumnWithSpecialCharacterName() throws Exception { + String tableName = generateTableName("rename_special"); + String fullName = catalogName + ".default." + tableName; + + spark.sql("CREATE TABLE " + fullName + " (`weird name` INT NOT NULL, keep INT)"); + + // Rename and nullability sources are canonicalized too. + Identifier ident = Identifier.of(new String[] {"default"}, tableName); + catalog.alterTable(ident, TableChange.renameColumn(new String[] {"weird name"}, "renamed")); + + List columns = Arrays.asList(catalog.loadTable(ident).schema().fieldNames()); + assertTrue(columns.contains("renamed")); + assertFalse(columns.contains("weird name")); + } + + @Test + public void testRenameThenNullabilityOnDistinctColumns() throws Exception { + String tableName = generateTableName("alter_distinct"); + String fullName = catalogName + ".default." + tableName; + + spark.sql("CREATE TABLE " + fullName + " (id BIGINT NOT NULL, name STRING)"); + + // Renaming one column and relaxing another column's nullability in one statement targets two + // distinct existing columns, so it commits as a single alterColumns batch. + Identifier ident = Identifier.of(new String[] {"default"}, tableName); + catalog.alterTable( + ident, + TableChange.renameColumn(new String[] {"name"}, "full_name"), + TableChange.updateColumnNullability(new String[] {"id"}, true)); + + StructType schema = catalog.loadTable(ident).schema(); + assertArrayEquals(new String[] {"id", "full_name"}, schema.fieldNames()); + assertTrue(schema.apply(schema.fieldIndex("id")).nullable()); + } + + @Test + public void testDropColumnIfExistsMissingIsNoOp() throws Exception { + String tableName = generateTableName("drop_if_exists"); + String fullName = catalogName + ".default." + tableName; + + spark.sql("CREATE TABLE " + fullName + " (id BIGINT NOT NULL, name STRING)"); + + // DROP COLUMN IF EXISTS on a missing column must not fail. + spark.sql("ALTER TABLE " + fullName + " DROP COLUMN IF EXISTS missing"); + + assertArrayEquals(new String[] {"id", "name"}, spark.table(fullName).schema().fieldNames()); + } + + @Test + public void testAddColumnWithDefaultRejected() throws Exception { + String tableName = generateTableName("add_default"); + String fullName = catalogName + ".default." + tableName; + + spark.sql("CREATE TABLE " + fullName + " (id BIGINT NOT NULL)"); + + // A DEFAULT value cannot be honored, so it must be rejected rather than silently dropped. + Identifier ident = Identifier.of(new String[] {"default"}, tableName); + UnsupportedOperationException ex = + assertThrows( + UnsupportedOperationException.class, + () -> + catalog.alterTable( + ident, + TableChange.addColumn( + new String[] {"with_default"}, + DataTypes.IntegerType, + true, + null, + null, + new org.apache.spark.sql.connector.catalog.ColumnDefaultValue( + "7", + new org.apache.spark.sql.connector.expressions.LiteralValue<>( + 7, DataTypes.IntegerType))))); + assertTrue(ex.getMessage().contains("DEFAULT")); + } + + @Test + public void testAddColumnRejectedOnLegacyFormat() throws Exception { + String tableName = generateTableName("add_legacy"); + String fullName = catalogName + ".default." + tableName; + + spark.sql( + "CREATE TABLE " + + fullName + + " (id INT NOT NULL) TBLPROPERTIES ('file_format_version'='LEGACY')"); + + // ADD COLUMN is unsupported on legacy-format tables and must fail loudly rather than surface a + // raw core error. + Identifier ident = Identifier.of(new String[] {"default"}, tableName); + UnsupportedOperationException ex = + assertThrows( + UnsupportedOperationException.class, + () -> + catalog.alterTable( + ident, TableChange.addColumn(new String[] {"added"}, DataTypes.IntegerType))); + assertTrue(ex.getMessage().contains("legacy")); } @Test