From 14b4a4db34bed1b8cc07edb1224eee07f6a14844 Mon Sep 17 00:00:00 2001 From: Pucheng Yang Date: Fri, 7 Aug 2026 17:57:52 +0000 Subject: [PATCH 1/8] feat: support schema evolution DDLs (ADD/DROP/RENAME/ALTER COLUMN) Closes #62. Extend `BaseLanceNamespaceSparkCatalog.alterTable` to handle Spark `TableChange.ColumnChange` requests, translating them into the corresponding Lance dataset operations via a new `LanceSchemaEvolution` helper: - `ALTER TABLE ADD COLUMN` -> `Dataset.addColumns` - `ALTER TABLE DROP COLUMN` -> `Dataset.dropColumns` - `ALTER TABLE RENAME COLUMN`-> `Dataset.alterColumns` (rename) - `ALTER TABLE ALTER COLUMN ... DROP/SET NOT NULL` -> `Dataset.alterColumns` (nullability) `ALTER COLUMN ... TYPE` is rejected with a clear `UnsupportedOperationException`: the current lance-core JNI drops the cast target type on the way to Rust (it parses `ArrowType.toString()` with `DataType::from_str` and swallows the failure), which would otherwise turn a type change into a silent no-op. Column-comment updates and positional (`FIRST`/`AFTER`) adds are likewise unsupported. Adds Java unit tests (directory + REST namespaces) and PySpark integration tests covering the happy path and the unsupported-type-change error, and documents the new operations in the ALTER TABLE docs. Co-Authored-By: Claude Opus 4.8 (1M context) --- docs/src/operations/ddl/alter-table.md | 38 +++++++ integration-tests/test_lance_spark.py | 50 +++++++++ .../spark/BaseLanceNamespaceSparkCatalog.java | 33 ++++-- .../org/lance/spark/LanceSchemaEvolution.java | 106 ++++++++++++++++++ .../spark/SparkLanceNamespaceTestBase.java | 95 +++++++++++++++- 5 files changed, 307 insertions(+), 15 deletions(-) create mode 100644 lance-spark-base_2.12/src/main/java/org/lance/spark/LanceSchemaEvolution.java diff --git a/docs/src/operations/ddl/alter-table.md b/docs/src/operations/ddl/alter-table.md index 54412b2f5..994fb9f2d 100644 --- a/docs/src/operations/ddl/alter-table.md +++ b/docs/src/operations/ddl/alter-table.md @@ -45,6 +45,44 @@ 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; +``` + +## 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. Adding a column at a specific position +(`FIRST`/`AFTER`) is not supported — columns are always appended. Changing a column's data type +(`ALTER COLUMN ... TYPE`) and updating a column comment are not currently supported. + ## 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..f4c573e18 100644 --- a/integration-tests/test_lance_spark.py +++ b/integration-tests/test_lance_spark.py @@ -587,6 +587,56 @@ 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_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..be922ab15 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,7 +942,7 @@ public Table alterTable(Identifier ident, TableChange... changes) throws NoSuchT + " can only be set at table creation."); } - if (propsToSet.isEmpty() && keysToRemove.isEmpty()) { + if (propsToSet.isEmpty() && keysToRemove.isEmpty() && columnChanges.isEmpty()) { // No changes to apply, just return the current table return loadTable(ident); } @@ -947,15 +950,23 @@ public Table alterTable(Identifier ident, TableChange... changes) throws NoSuchT 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); + } + + if (!propsToSet.isEmpty() || !keysToRemove.isEmpty()) { + // 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..71c100af8 --- /dev/null +++ b/lance-spark-base_2.12/src/main/java/org/lance/spark/LanceSchemaEvolution.java @@ -0,0 +1,106 @@ +/* + * 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.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.Arrays; +import java.util.Collections; +import java.util.List; + +/** + * Translates Spark {@link ColumnChange} schema evolution requests (produced by {@code ALTER TABLE + * ADD/DROP/RENAME/ALTER COLUMN}) into the corresponding {@link Dataset} operations. Changes are + * applied to the open dataset in the order Spark supplies them. + */ +final class LanceSchemaEvolution { + + private LanceSchemaEvolution() {} + + static void apply(Dataset dataset, List changes) { + for (ColumnChange change : changes) { + if (change instanceof AddColumn) { + addColumn(dataset, (AddColumn) change); + } else if (change instanceof DeleteColumn) { + DeleteColumn delete = (DeleteColumn) change; + dataset.dropColumns(Collections.singletonList(path(delete.fieldNames()))); + } else if (change instanceof RenameColumn) { + RenameColumn rename = (RenameColumn) change; + dataset.alterColumns( + Collections.singletonList( + new ColumnAlteration.Builder(path(rename.fieldNames())) + .rename(rename.newName()) + .build())); + } 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 '" + + path(updateType.fieldNames()) + + "' is not supported by the current Lance version."); + } else if (change instanceof UpdateColumnNullability) { + UpdateColumnNullability updateNull = (UpdateColumnNullability) change; + dataset.alterColumns( + Collections.singletonList( + new ColumnAlteration.Builder(path(updateNull.fieldNames())) + .nullable(updateNull.nullable()) + .build())); + } else { + throw new UnsupportedOperationException( + "Unsupported column change type: " + change.getClass().getSimpleName()); + } + } + } + + private static void addColumn(Dataset dataset, AddColumn add) { + String[] fieldNames = add.fieldNames(); + if (fieldNames.length != 1) { + throw new UnsupportedOperationException( + "Adding nested columns is not supported: " + path(fieldNames)); + } + if (add.position() != null) { + throw new UnsupportedOperationException( + "ADD COLUMN with FIRST/AFTER position is not supported; columns are appended."); + } + + MetadataBuilder metadataBuilder = new MetadataBuilder(); + if (add.comment() != null) { + metadataBuilder.putString("comment", add.comment()); + } + StructField field = + new StructField(fieldNames[0], add.dataType(), add.isNullable(), metadataBuilder.build()); + Schema arrowSchema = + LanceArrowUtils.toArrowSchema(new StructType(new StructField[] {field}), "UTC", true); + dataset.addColumns(arrowSchema.getFields()); + } + + private static String path(String[] fieldNames) { + return FieldPathUtils.canonicalPath(Arrays.asList(fieldNames)); + } +} 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..c5ac14e67 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,100 @@ 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 From c6ab38e51273935dc123b442ed54dc5ce324b3fb Mon Sep 17 00:00:00 2001 From: Pucheng Yang Date: Fri, 7 Aug 2026 22:04:25 +0000 Subject: [PATCH 2/8] fix: validate schema-evolution requests atomically and handle edge cases Address review feedback on the schema-evolution DDL support: - Validate the entire ordered ALTER TABLE request against the current schema before mutating anything, so a rejected change never leaves the table partially mutated (Spark's alterTable "all-or-nothing" contract). - Honor DROP COLUMN IF EXISTS: a missing column is skipped instead of raising, while a plain DROP COLUMN on a missing column still fails. - Reject ADD COLUMN with a DEFAULT value rather than silently filling NULL (Lance's all-null add cannot backfill a default). - Reject ADD COLUMN on legacy-format (file_format_version='LEGACY') tables up front instead of surfacing a raw core error. Adds unit tests for atomic rejection, DROP COLUMN IF EXISTS, default rejection, and legacy-format rejection; a DROP COLUMN IF EXISTS integration test; and documents the new limitations. Co-Authored-By: Claude Opus 4.8 (1M context) --- docs/src/operations/ddl/alter-table.md | 13 +- integration-tests/test_lance_spark.py | 7 + .../org/lance/spark/LanceSchemaEvolution.java | 161 +++++++++++++----- .../spark/SparkLanceNamespaceTestBase.java | 84 +++++++++ 4 files changed, 223 insertions(+), 42 deletions(-) diff --git a/docs/src/operations/ddl/alter-table.md b/docs/src/operations/ddl/alter-table.md index 994fb9f2d..09a606eb8 100644 --- a/docs/src/operations/ddl/alter-table.md +++ b/docs/src/operations/ddl/alter-table.md @@ -60,6 +60,7 @@ Remove a column from the table: ```sql ALTER TABLE users DROP COLUMN age; +ALTER TABLE users DROP COLUMN IF EXISTS age; ``` ## RENAME COLUMN @@ -79,9 +80,15 @@ ALTER TABLE users ALTER COLUMN id DROP NOT NULL; ``` !!! note -Column schema evolution operates on top-level columns. Adding a column at a specific position -(`FIRST`/`AFTER`) is not supported — columns are always appended. Changing a column's data type -(`ALTER COLUMN ... TYPE`) and updating a column comment are not currently supported. +Column schema evolution operates on top-level columns. When several column changes are given in +one statement, the whole request is validated first — 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 diff --git a/integration-tests/test_lance_spark.py b/integration-tests/test_lance_spark.py index f4c573e18..c8712749b 100644 --- a/integration-tests/test_lance_spark.py +++ b/integration-tests/test_lance_spark.py @@ -614,6 +614,13 @@ def test_drop_column(self, spark): 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)") 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 index 71c100af8..d5b006e47 100644 --- 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 @@ -31,75 +31,158 @@ import java.util.Arrays; import java.util.Collections; +import java.util.LinkedHashSet; import java.util.List; +import java.util.Set; /** * Translates Spark {@link ColumnChange} schema evolution requests (produced by {@code ALTER TABLE - * ADD/DROP/RENAME/ALTER COLUMN}) into the corresponding {@link Dataset} operations. Changes are - * applied to the open dataset in the order Spark supplies them. + * ADD/DROP/RENAME/ALTER COLUMN}) into the corresponding {@link Dataset} operations. + * + *

The whole ordered request is validated against the current schema first and any unsupported + * option is rejected before any change is written, so a rejected request never leaves the + * table partially mutated. Validated changes are then applied in the order Spark supplies them. */ final class LanceSchemaEvolution { + /** Lance file format version string for the legacy ("0.1") format. */ + private static final String LEGACY_FILE_FORMAT_VERSION = "0.1"; + private LanceSchemaEvolution() {} static void apply(Dataset dataset, List changes) { + boolean legacyFormat = LEGACY_FILE_FORMAT_VERSION.equals(dataset.getLanceFileFormatVersion()); + + // Validate the entire request before mutating anything, so a rejected change never leaves the + // table partially mutated. Validation runs against a simulated copy of the field set that + // tracks the names each change introduces or removes, so ordered requests are checked against + // the evolving schema. + Set simulated = topLevelFieldNames(dataset); + for (ColumnChange change : changes) { + validate(change, simulated, legacyFormat); + } + + // Apply against a fresh live copy so per-change decisions (e.g. DROP COLUMN IF EXISTS) reflect + // the schema state at the point each change is applied. + Set current = topLevelFieldNames(dataset); for (ColumnChange change : changes) { - if (change instanceof AddColumn) { - addColumn(dataset, (AddColumn) change); - } else if (change instanceof DeleteColumn) { - DeleteColumn delete = (DeleteColumn) change; - dataset.dropColumns(Collections.singletonList(path(delete.fieldNames()))); - } else if (change instanceof RenameColumn) { - RenameColumn rename = (RenameColumn) change; - dataset.alterColumns( - Collections.singletonList( - new ColumnAlteration.Builder(path(rename.fieldNames())) - .rename(rename.newName()) - .build())); - } 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. + applyOne(dataset, change, current); + } + } + + private static void validate( + ColumnChange change, Set topLevelFields, boolean legacyFormat) { + if (change instanceof AddColumn) { + AddColumn add = (AddColumn) change; + if (add.fieldNames().length != 1) { + throw new UnsupportedOperationException( + "Adding nested columns is not supported: " + path(add.fieldNames())); + } + 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( - "Changing the type of column '" - + path(updateType.fieldNames()) - + "' is not supported by the current Lance version."); - } else if (change instanceof UpdateColumnNullability) { - UpdateColumnNullability updateNull = (UpdateColumnNullability) change; - dataset.alterColumns( - Collections.singletonList( - new ColumnAlteration.Builder(path(updateNull.fieldNames())) - .nullable(updateNull.nullable()) - .build())); + "ADD COLUMN is not supported on legacy-format (" + + LEGACY_FILE_FORMAT_VERSION + + ") tables."); + } + topLevelFields.add(add.fieldNames()[0]); + } else if (change instanceof DeleteColumn) { + DeleteColumn delete = (DeleteColumn) change; + if (topLevelFields.contains(topLevelName(delete.fieldNames())) || delete.ifExists()) { + topLevelFields.remove(topLevelName(delete.fieldNames())); } else { throw new UnsupportedOperationException( - "Unsupported column change type: " + change.getClass().getSimpleName()); + "Cannot drop missing column: " + path(delete.fieldNames())); } + } else if (change instanceof RenameColumn) { + RenameColumn rename = (RenameColumn) change; + requireExists(topLevelFields, rename.fieldNames()); + topLevelFields.remove(topLevelName(rename.fieldNames())); + topLevelFields.add(rename.newName()); + } else if (change instanceof UpdateColumnNullability) { + UpdateColumnNullability updateNull = (UpdateColumnNullability) change; + requireExists(topLevelFields, updateNull.fieldNames()); + } 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 '" + + path(updateType.fieldNames()) + + "' is not supported by the current Lance version."); + } else { + throw new UnsupportedOperationException( + "Unsupported column change type: " + change.getClass().getSimpleName()); } } - private static void addColumn(Dataset dataset, AddColumn add) { - String[] fieldNames = add.fieldNames(); - if (fieldNames.length != 1) { - throw new UnsupportedOperationException( - "Adding nested columns is not supported: " + path(fieldNames)); - } - if (add.position() != null) { - throw new UnsupportedOperationException( - "ADD COLUMN with FIRST/AFTER position is not supported; columns are appended."); + private static void applyOne(Dataset dataset, ColumnChange change, Set current) { + if (change instanceof AddColumn) { + AddColumn add = (AddColumn) change; + addColumn(dataset, add); + current.add(add.fieldNames()[0]); + } else if (change instanceof DeleteColumn) { + DeleteColumn delete = (DeleteColumn) change; + if (delete.ifExists() && !current.contains(topLevelName(delete.fieldNames()))) { + return; + } + current.remove(topLevelName(delete.fieldNames())); + dataset.dropColumns(Collections.singletonList(path(delete.fieldNames()))); + } else if (change instanceof RenameColumn) { + RenameColumn rename = (RenameColumn) change; + dataset.alterColumns( + Collections.singletonList( + new ColumnAlteration.Builder(path(rename.fieldNames())) + .rename(rename.newName()) + .build())); + current.remove(topLevelName(rename.fieldNames())); + current.add(rename.newName()); + } else if (change instanceof UpdateColumnNullability) { + UpdateColumnNullability updateNull = (UpdateColumnNullability) change; + dataset.alterColumns( + Collections.singletonList( + new ColumnAlteration.Builder(path(updateNull.fieldNames())) + .nullable(updateNull.nullable()) + .build())); } + } + private static void addColumn(Dataset dataset, AddColumn add) { MetadataBuilder metadataBuilder = new MetadataBuilder(); if (add.comment() != null) { metadataBuilder.putString("comment", add.comment()); } StructField field = - new StructField(fieldNames[0], add.dataType(), add.isNullable(), metadataBuilder.build()); + new StructField( + add.fieldNames()[0], add.dataType(), add.isNullable(), metadataBuilder.build()); Schema arrowSchema = LanceArrowUtils.toArrowSchema(new StructType(new StructField[] {field}), "UTC", true); dataset.addColumns(arrowSchema.getFields()); } + private static void requireExists(Set topLevelFields, String[] fieldNames) { + if (!topLevelFields.contains(topLevelName(fieldNames))) { + throw new UnsupportedOperationException("Cannot alter missing column: " + path(fieldNames)); + } + } + + private static Set topLevelFieldNames(Dataset dataset) { + Set names = new LinkedHashSet<>(); + dataset.getSchema().getFields().forEach(f -> names.add(f.getName())); + return names; + } + + private static String topLevelName(String[] fieldNames) { + return fieldNames[0]; + } + private static String path(String[] fieldNames) { return FieldPathUtils.canonicalPath(Arrays.asList(fieldNames)); } 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 c5ac14e67..05be59b77 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 @@ -1052,6 +1052,90 @@ public void testAlterColumnDropNotNull() throws Exception { 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 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 public void testShowTablePropertiesEmpty() throws Exception { String tableName = generateTableName("show_props_empty"); From 02e641a82aec6bd40347045d7953ea9adf37fa7c Mon Sep 17 00:00:00 2001 From: Pucheng Yang Date: Fri, 7 Aug 2026 22:46:38 +0000 Subject: [PATCH 3/8] fix: commit each ALTER TABLE request as one atomic core mutation Address remaining review feedback: preserve Spark's all-or-nothing alterTable contract by committing an accepted request through exactly one Lance core operation instead of a sequence of per-change commits. - Same-kind column changes are batched into a single core call: ADD -> addColumns, DROP -> dropColumns, RENAME/nullability -> alterColumns (rename + nullability edits to the same column are merged into one ColumnAlteration). - Requests that would require more than one core mutation are rejected before the first write: mixing column additions, drops, and alterations in one statement, and combining any column change with TBLPROPERTIES changes (which commit separately). Adds tests for batched multi-column ADD, mixed-kind rejection, and column-change-plus-TBLPROPERTIES rejection; documents the single-mutation restriction. Co-Authored-By: Claude Opus 4.8 (1M context) --- docs/src/operations/ddl/alter-table.md | 10 +- .../spark/BaseLanceNamespaceSparkCatalog.java | 17 +- .../org/lance/spark/LanceSchemaEvolution.java | 161 ++++++++++++------ .../spark/SparkLanceNamespaceTestBase.java | 60 +++++++ 4 files changed, 185 insertions(+), 63 deletions(-) diff --git a/docs/src/operations/ddl/alter-table.md b/docs/src/operations/ddl/alter-table.md index 09a606eb8..ed39fb6f4 100644 --- a/docs/src/operations/ddl/alter-table.md +++ b/docs/src/operations/ddl/alter-table.md @@ -80,10 +80,12 @@ ALTER TABLE users ALTER COLUMN id DROP NOT NULL; ``` !!! note -Column schema evolution operates on top-level columns. When several column changes are given in -one statement, the whole request is validated first — if any change is unsupported, none is -applied. The following are **not** currently supported and are rejected before any change is -written: +Column schema evolution operates on top-level columns. Each `ALTER TABLE` is committed as a single +atomic Lance operation, so one statement may batch changes of the same kind (e.g. adding several +columns), but it may not mix column additions, drops, and alterations, nor combine a column change +with `TBLPROPERTIES` changes — 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`. 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 be922ab15..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 @@ -942,11 +942,22 @@ public Table alterTable(Identifier ident, TableChange... changes) throws NoSuchT + " can only be set at table creation."); } - if (propsToSet.isEmpty() && keysToRemove.isEmpty() && columnChanges.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()) { @@ -954,9 +965,7 @@ public Table alterTable(Identifier ident, TableChange... changes) throws NoSuchT // Schema-evolution changes commit through the dataset's own handler, which // openDatasetBuilder wires for managed versioning when applicable. LanceSchemaEvolution.apply(dataset, columnChanges); - } - - if (!propsToSet.isEmpty() || !keysToRemove.isEmpty()) { + } 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()); 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 index d5b006e47..fe5298692 100644 --- 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 @@ -17,6 +17,7 @@ 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; @@ -29,48 +30,79 @@ import org.apache.spark.sql.types.StructType; import org.apache.spark.sql.util.LanceArrowUtils; +import java.util.ArrayList; import java.util.Arrays; -import java.util.Collections; +import java.util.LinkedHashMap; import java.util.LinkedHashSet; import java.util.List; +import java.util.Map; import java.util.Set; /** - * Translates Spark {@link ColumnChange} schema evolution requests (produced by {@code ALTER TABLE - * ADD/DROP/RENAME/ALTER COLUMN}) into the corresponding {@link Dataset} operations. + * 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. * - *

The whole ordered request is validated against the current schema first and any unsupported - * option is rejected before any change is written, so a rejected request never leaves the - * table partially mutated. Validated changes are then applied in the order Spark supplies them. + *

To preserve Spark's all-or-nothing {@code alterTable} contract, an accepted request is + * committed through exactly one core operation: {@code addColumns}, {@code dropColumns}, or {@code + * alterColumns} (each of which commits its whole batch atomically). The request is fully validated + * against the current schema, and requests that would require more than one core mutation are + * rejected, before anything is written — so a rejected request never leaves the table + * partially mutated. Heterogeneous batching (mixing additions, drops, and alterations in one + * commit) is not supported by the core yet, so such requests must be issued as separate statements. */ 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 fields = topLevelFieldNames(dataset); - // Validate the entire request before mutating anything, so a rejected change never leaves the - // table partially mutated. Validation runs against a simulated copy of the field set that - // tracks the names each change introduces or removes, so ordered requests are checked against - // the evolving schema. - Set simulated = topLevelFieldNames(dataset); + // Validate the whole request and confirm it compiles to a single core mutation before writing + // anything. `fields` tracks the names each change introduces or removes so an ordered request + // is checked against the evolving schema. + Kind kind = null; for (ColumnChange change : changes) { - validate(change, simulated, legacyFormat); + Kind changeKind = validate(change, fields, 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."); + } } - // Apply against a fresh live copy so per-change decisions (e.g. DROP COLUMN IF EXISTS) reflect - // the schema state at the point each change is applied. - Set current = topLevelFieldNames(dataset); - for (ColumnChange change : changes) { - applyOne(dataset, change, current); + switch (kind) { + case ADD: + applyAdds(dataset, changes); + break; + case DROP: + applyDrops(dataset, changes); + break; + case ALTER: + applyAlters(dataset, changes); + break; + default: + throw new IllegalStateException("Unexpected change kind: " + kind); } } - private static void validate( + private static Kind validate( ColumnChange change, Set topLevelFields, boolean legacyFormat) { if (change instanceof AddColumn) { AddColumn add = (AddColumn) change; @@ -93,22 +125,26 @@ private static void validate( + ") tables."); } topLevelFields.add(add.fieldNames()[0]); + return Kind.ADD; } else if (change instanceof DeleteColumn) { DeleteColumn delete = (DeleteColumn) change; - if (topLevelFields.contains(topLevelName(delete.fieldNames())) || delete.ifExists()) { + if (topLevelFields.contains(topLevelName(delete.fieldNames()))) { topLevelFields.remove(topLevelName(delete.fieldNames())); - } else { + } else if (!delete.ifExists()) { throw new UnsupportedOperationException( "Cannot drop missing column: " + path(delete.fieldNames())); } + return Kind.DROP; } else if (change instanceof RenameColumn) { RenameColumn rename = (RenameColumn) change; requireExists(topLevelFields, rename.fieldNames()); topLevelFields.remove(topLevelName(rename.fieldNames())); topLevelFields.add(rename.newName()); + return Kind.ALTER; } else if (change instanceof UpdateColumnNullability) { UpdateColumnNullability updateNull = (UpdateColumnNullability) change; requireExists(topLevelFields, updateNull.fieldNames()); + 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 @@ -123,48 +159,61 @@ private static void validate( } } - private static void applyOne(Dataset dataset, ColumnChange change, Set current) { - if (change instanceof AddColumn) { + private static void applyAdds(Dataset dataset, List changes) { + List fields = new ArrayList<>(changes.size()); + for (ColumnChange change : changes) { AddColumn add = (AddColumn) change; - addColumn(dataset, add); - current.add(add.fieldNames()[0]); - } else if (change instanceof DeleteColumn) { + 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 current = topLevelFieldNames(dataset); + List toDrop = new ArrayList<>(changes.size()); + for (ColumnChange change : changes) { DeleteColumn delete = (DeleteColumn) change; if (delete.ifExists() && !current.contains(topLevelName(delete.fieldNames()))) { - return; + continue; } - current.remove(topLevelName(delete.fieldNames())); - dataset.dropColumns(Collections.singletonList(path(delete.fieldNames()))); - } else if (change instanceof RenameColumn) { - RenameColumn rename = (RenameColumn) change; - dataset.alterColumns( - Collections.singletonList( - new ColumnAlteration.Builder(path(rename.fieldNames())) - .rename(rename.newName()) - .build())); - current.remove(topLevelName(rename.fieldNames())); - current.add(rename.newName()); - } else if (change instanceof UpdateColumnNullability) { - UpdateColumnNullability updateNull = (UpdateColumnNullability) change; - dataset.alterColumns( - Collections.singletonList( - new ColumnAlteration.Builder(path(updateNull.fieldNames())) - .nullable(updateNull.nullable()) - .build())); + toDrop.add(path(delete.fieldNames())); + } + if (!toDrop.isEmpty()) { + dataset.dropColumns(toDrop); } } - private static void addColumn(Dataset dataset, AddColumn add) { - MetadataBuilder metadataBuilder = new MetadataBuilder(); - if (add.comment() != null) { - metadataBuilder.putString("comment", add.comment()); + private static void applyAlters(Dataset dataset, List changes) { + // Merge rename and nullability edits that target the same column into one alteration so the + // whole request stays a single alterColumns commit. + Map builders = new LinkedHashMap<>(); + for (ColumnChange change : changes) { + if (change instanceof RenameColumn) { + RenameColumn rename = (RenameColumn) change; + builders + .computeIfAbsent(path(rename.fieldNames()), ColumnAlteration.Builder::new) + .rename(rename.newName()); + } else if (change instanceof UpdateColumnNullability) { + UpdateColumnNullability updateNull = (UpdateColumnNullability) change; + builders + .computeIfAbsent(path(updateNull.fieldNames()), ColumnAlteration.Builder::new) + .nullable(updateNull.nullable()); + } } - StructField field = - new StructField( - add.fieldNames()[0], add.dataType(), add.isNullable(), metadataBuilder.build()); - Schema arrowSchema = - LanceArrowUtils.toArrowSchema(new StructType(new StructField[] {field}), "UTC", true); - dataset.addColumns(arrowSchema.getFields()); + List alterations = new ArrayList<>(builders.size()); + for (ColumnAlteration.Builder builder : builders.values()) { + alterations.add(builder.build()); + } + dataset.alterColumns(alterations); } private static void requireExists(Set topLevelFields, String[] fieldNames) { @@ -175,7 +224,9 @@ private static void requireExists(Set topLevelFields, String[] fieldName private static Set topLevelFieldNames(Dataset dataset) { Set names = new LinkedHashSet<>(); - dataset.getSchema().getFields().forEach(f -> names.add(f.getName())); + 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 05be59b77..80be61bde 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 @@ -1073,6 +1073,66 @@ public void testAlterTableRejectsRequestAtomically() throws Exception { 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 testDropColumnIfExistsMissingIsNoOp() throws Exception { String tableName = generateTableName("drop_if_exists"); From cadb61685ebaa4a469506ae482cc54aaa23a0ce4 Mon Sep 17 00:00:00 2001 From: Pucheng Yang Date: Fri, 7 Aug 2026 23:05:04 +0000 Subject: [PATCH 4/8] fix: validate schema-evolution requests against the current schema MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Address review feedback on the single-mutation compiler: the batched core operation applies to the current schema with no ordering between its entries, so validating against an evolving schema (or coalescing by path) accepted requests the core cannot represent. - Validate every referenced column against the current schema, not an evolving one, so a change that depends on an earlier change in the same request (e.g. altering a column by its just-assigned new name) is rejected before any write instead of failing mid-commit. - Reject a request that targets the same column more than once: order is lost when changes collapse into one batch, and intermediate validation would be skipped. - Reject nested (multi-part) column paths up front (top-level only), instead of truncating to the leading path element and sending a path the core rejects — this keeps DROP COLUMN IF EXISTS's no-error contract for nested paths. Adds tests for rename-dependent rejection, repeated-target rejection, nested-path rejection, and a distinct-column rename+nullability batch; documents the restrictions. Co-Authored-By: Claude Opus 4.8 (1M context) --- docs/src/operations/ddl/alter-table.md | 12 +- .../org/lance/spark/LanceSchemaEvolution.java | 140 ++++++++++-------- .../spark/SparkLanceNamespaceTestBase.java | 81 ++++++++++ 3 files changed, 167 insertions(+), 66 deletions(-) diff --git a/docs/src/operations/ddl/alter-table.md b/docs/src/operations/ddl/alter-table.md index ed39fb6f4..7ca9d0593 100644 --- a/docs/src/operations/ddl/alter-table.md +++ b/docs/src/operations/ddl/alter-table.md @@ -81,11 +81,13 @@ 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, so one statement may batch changes of the same kind (e.g. adding several -columns), but it may not mix column additions, drops, and alterations, nor combine a column change -with `TBLPROPERTIES` changes — 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: +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`. 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 index fe5298692..f1d283644 100644 --- 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 @@ -15,7 +15,6 @@ 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; @@ -31,24 +30,30 @@ import org.apache.spark.sql.util.LanceArrowUtils; import java.util.ArrayList; -import java.util.Arrays; -import java.util.LinkedHashMap; +import java.util.HashSet; import java.util.LinkedHashSet; import java.util.List; -import java.util.Map; 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 all-or-nothing {@code alterTable} contract, an accepted request is - * committed through exactly one core operation: {@code addColumns}, {@code dropColumns}, or {@code - * alterColumns} (each of which commits its whole batch atomically). The request is fully validated - * against the current schema, and requests that would require more than one core mutation are - * rejected, before anything is written — so a rejected request never leaves the table - * partially mutated. Heterogeneous batching (mixing additions, drops, and alterations in one - * commit) is not supported by the core yet, so such requests must be issued as separate statements. + *

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 { @@ -70,14 +75,16 @@ static void apply(Dataset dataset, List changes) { } boolean legacyFormat = LEGACY_FILE_FORMAT_VERSION.equals(dataset.getLanceFileFormatVersion()); - Set fields = topLevelFieldNames(dataset); + Set currentFields = topLevelFieldNames(dataset); - // Validate the whole request and confirm it compiles to a single core mutation before writing - // anything. `fields` tracks the names each change introduces or removes so an ordered request - // is checked against the evolving schema. + // 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). Kind kind = null; + Set touched = new HashSet<>(); for (ColumnChange change : changes) { - Kind changeKind = validate(change, fields, legacyFormat); + Kind changeKind = validate(change, currentFields, touched, legacyFormat); if (kind == null) { kind = changeKind; } else if (kind != changeKind) { @@ -92,7 +99,7 @@ static void apply(Dataset dataset, List changes) { applyAdds(dataset, changes); break; case DROP: - applyDrops(dataset, changes); + applyDrops(dataset, changes, currentFields); break; case ALTER: applyAlters(dataset, changes); @@ -103,13 +110,10 @@ static void apply(Dataset dataset, List changes) { } private static Kind validate( - ColumnChange change, Set topLevelFields, boolean legacyFormat) { + ColumnChange change, Set currentFields, Set touched, boolean legacyFormat) { if (change instanceof AddColumn) { AddColumn add = (AddColumn) change; - if (add.fieldNames().length != 1) { - throw new UnsupportedOperationException( - "Adding nested columns is not supported: " + path(add.fieldNames())); - } + 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."); @@ -124,26 +128,32 @@ private static Kind validate( + LEGACY_FILE_FORMAT_VERSION + ") tables."); } - topLevelFields.add(add.fieldNames()[0]); + 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; - if (topLevelFields.contains(topLevelName(delete.fieldNames()))) { - topLevelFields.remove(topLevelName(delete.fieldNames())); - } else if (!delete.ifExists()) { - throw new UnsupportedOperationException( - "Cannot drop missing column: " + path(delete.fieldNames())); + requireTopLevel(delete.fieldNames(), "Dropping nested columns"); + String name = delete.fieldNames()[0]; + if (!currentFields.contains(name) && !delete.ifExists()) { + throw new UnsupportedOperationException("Cannot drop missing column: " + name); } + requireDistinct(touched, name); return Kind.DROP; } else if (change instanceof RenameColumn) { RenameColumn rename = (RenameColumn) change; - requireExists(topLevelFields, rename.fieldNames()); - topLevelFields.remove(topLevelName(rename.fieldNames())); - topLevelFields.add(rename.newName()); + requireTopLevel(rename.fieldNames(), "Renaming nested columns"); + requireExists(currentFields, rename.fieldNames()[0]); + requireDistinct(touched, rename.fieldNames()[0]); return Kind.ALTER; } else if (change instanceof UpdateColumnNullability) { UpdateColumnNullability updateNull = (UpdateColumnNullability) change; - requireExists(topLevelFields, updateNull.fieldNames()); + 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; @@ -151,7 +161,7 @@ private static Kind validate( // turning a type change into a no-op. Reject it explicitly rather than lying about it. throw new UnsupportedOperationException( "Changing the type of column '" - + path(updateType.fieldNames()) + + String.join(".", updateType.fieldNames()) + "' is not supported by the current Lance version."); } else { throw new UnsupportedOperationException( @@ -177,15 +187,18 @@ private static void applyAdds(Dataset dataset, List changes) { dataset.addColumns(arrowSchema.getFields()); } - private static void applyDrops(Dataset dataset, List changes) { - Set current = topLevelFieldNames(dataset); + private static void applyDrops( + Dataset dataset, List changes, Set currentFields) { List toDrop = new ArrayList<>(changes.size()); for (ColumnChange change : changes) { DeleteColumn delete = (DeleteColumn) change; - if (delete.ifExists() && !current.contains(topLevelName(delete.fieldNames()))) { + 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(path(delete.fieldNames())); + toDrop.add(name); } if (!toDrop.isEmpty()) { dataset.dropColumns(toDrop); @@ -193,32 +206,45 @@ private static void applyDrops(Dataset dataset, List changes) { } private static void applyAlters(Dataset dataset, List changes) { - // Merge rename and nullability edits that target the same column into one alteration so the - // whole request stays a single alterColumns commit. - Map builders = new LinkedHashMap<>(); + // 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; - builders - .computeIfAbsent(path(rename.fieldNames()), ColumnAlteration.Builder::new) - .rename(rename.newName()); + alterations.add( + new ColumnAlteration.Builder(rename.fieldNames()[0]).rename(rename.newName()).build()); } else if (change instanceof UpdateColumnNullability) { UpdateColumnNullability updateNull = (UpdateColumnNullability) change; - builders - .computeIfAbsent(path(updateNull.fieldNames()), ColumnAlteration.Builder::new) - .nullable(updateNull.nullable()); + alterations.add( + new ColumnAlteration.Builder(updateNull.fieldNames()[0]) + .nullable(updateNull.nullable()) + .build()); } } - List alterations = new ArrayList<>(builders.size()); - for (ColumnAlteration.Builder builder : builders.values()) { - alterations.add(builder.build()); - } dataset.alterColumns(alterations); } - private static void requireExists(Set topLevelFields, String[] fieldNames) { - if (!topLevelFields.contains(topLevelName(fieldNames))) { - throw new UnsupportedOperationException("Cannot alter missing column: " + path(fieldNames)); + 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); + } + } + + 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."); } } @@ -229,12 +255,4 @@ private static Set topLevelFieldNames(Dataset dataset) { } return names; } - - private static String topLevelName(String[] fieldNames) { - return fieldNames[0]; - } - - private static String path(String[] fieldNames) { - return FieldPathUtils.canonicalPath(Arrays.asList(fieldNames)); - } } 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 80be61bde..2903bfc75 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 @@ -1133,6 +1133,87 @@ public void testAlterTableRejectsColumnChangeMixedWithProperties() throws Except 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 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"); From d1d724bcdbaa10a79c94e4e98cf88dbefceba3c7 Mon Sep 17 00:00:00 2001 From: Pucheng Yang Date: Fri, 7 Aug 2026 23:31:45 +0000 Subject: [PATCH 5/8] fix: preserve ordered rename semantics and top-level path identity Address review feedback on the single-mutation compiler: - Simulate rename source-removal and destination-insertion in request order during validation, rejecting a rename whose destination name is already occupied at that step. The core alterColumns batch is unordered, so without this an ordered request like RENAME a->b, b->c would be wrongly accepted as simultaneous renames instead of failing on the a->b collision. - Pass top-level column names to dropColumns/alterColumns verbatim (the Lance path for a top-level field is the name itself), with a test for a special-character column name. Adds tests for ordered rename-to-occupied rejection and a special-character column drop. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../org/lance/spark/LanceSchemaEvolution.java | 30 +++++++++++++--- .../spark/SparkLanceNamespaceTestBase.java | 35 +++++++++++++++++++ 2 files changed, 61 insertions(+), 4 deletions(-) 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 index f1d283644..3ab9e917b 100644 --- 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 @@ -81,10 +81,14 @@ static void apply(Dataset dataset, List changes) { // 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, legacyFormat); + Kind changeKind = validate(change, currentFields, touched, occupied, legacyFormat); if (kind == null) { kind = changeKind; } else if (kind != changeKind) { @@ -110,7 +114,11 @@ static void apply(Dataset dataset, List changes) { } private static Kind validate( - ColumnChange change, Set currentFields, Set touched, boolean legacyFormat) { + ColumnChange change, + Set currentFields, + Set touched, + Set occupied, + boolean legacyFormat) { if (change instanceof AddColumn) { AddColumn add = (AddColumn) change; requireTopLevel(add.fieldNames(), "Adding nested columns"); @@ -146,8 +154,21 @@ private static Kind validate( } else if (change instanceof RenameColumn) { RenameColumn rename = (RenameColumn) change; requireTopLevel(rename.fieldNames(), "Renaming nested columns"); - requireExists(currentFields, rename.fieldNames()[0]); - requireDistinct(touched, rename.fieldNames()[0]); + 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; @@ -198,6 +219,7 @@ private static void applyDrops( if (delete.ifExists() && !currentFields.contains(name)) { continue; } + // Only top-level columns are supported, so the Lance path is the field name verbatim. toDrop.add(name); } if (!toDrop.isEmpty()) { 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 2903bfc75..0409ec358 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 @@ -1194,6 +1194,41 @@ public void testDropNestedColumnRejected() throws Exception { 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 with special characters is passed to the core verbatim. + 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 testRenameThenNullabilityOnDistinctColumns() throws Exception { String tableName = generateTableName("alter_distinct"); From 85c0e0cc1b6238534c473d3defe362a9ad304674 Mon Sep 17 00:00:00 2001 From: Pucheng Yang Date: Fri, 7 Aug 2026 23:53:02 +0000 Subject: [PATCH 6/8] fix: canonicalize column source paths for schema evolution Escape every DROP, RENAME, and nullability source column name into a canonical Lance field path (via FieldPathUtils.canonicalPath) before passing it to the core, so top-level names containing path syntax (dots, spaces, etc.) resolve to the intended field instead of being rejected or misparsed. Adds special-character-name drop and rename tests. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../org/lance/spark/LanceSchemaEvolution.java | 19 +++++++++++++++---- .../spark/SparkLanceNamespaceTestBase.java | 19 ++++++++++++++++++- 2 files changed, 33 insertions(+), 5 deletions(-) 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 index 3ab9e917b..6a1342a89 100644 --- 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 @@ -15,6 +15,7 @@ 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; @@ -30,6 +31,7 @@ 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; @@ -219,8 +221,7 @@ private static void applyDrops( if (delete.ifExists() && !currentFields.contains(name)) { continue; } - // Only top-level columns are supported, so the Lance path is the field name verbatim. - toDrop.add(name); + toDrop.add(canonicalPath(name)); } if (!toDrop.isEmpty()) { dataset.dropColumns(toDrop); @@ -235,11 +236,13 @@ private static void applyAlters(Dataset dataset, List changes) { if (change instanceof RenameColumn) { RenameColumn rename = (RenameColumn) change; alterations.add( - new ColumnAlteration.Builder(rename.fieldNames()[0]).rename(rename.newName()).build()); + 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(updateNull.fieldNames()[0]) + new ColumnAlteration.Builder(canonicalPath(updateNull.fieldNames()[0])) .nullable(updateNull.nullable()) .build()); } @@ -260,6 +263,14 @@ private static void requireExists(Set currentFields, String 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( 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 0409ec358..841bd685d 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 @@ -1220,7 +1220,8 @@ public void testDropColumnWithSpecialCharacterName() throws Exception { String tableName = generateTableName("drop_special"); String fullName = catalogName + ".default." + tableName; - // A top-level column name with special characters is passed to the core verbatim. + // 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); @@ -1229,6 +1230,22 @@ public void testDropColumnWithSpecialCharacterName() throws Exception { assertArrayEquals(new String[] {"keep"}, 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"); From 77afedd8440ad14974e6a5b2b3395683b77e7ad1 Mon Sep 17 00:00:00 2001 From: Pucheng Yang Date: Sat, 8 Aug 2026 00:06:38 +0000 Subject: [PATCH 7/8] fix: reject DROP COLUMN for backtick names unsupported by the core The current core drop API cannot resolve a column whose name contains a backtick under either the canonical (escaped) or raw representation. Detect such names during validation and reject the DROP up front with a clear message, instead of letting it fail mid-commit with a confusing "field not found". RENAME and nullability keep using canonical paths, which the core resolves correctly. Adds a test for the backtick-name DROP rejection. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../org/lance/spark/LanceSchemaEvolution.java | 9 +++++++++ .../spark/SparkLanceNamespaceTestBase.java | 19 +++++++++++++++++++ 2 files changed, 28 insertions(+) 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 index 6a1342a89..8b5f47c52 100644 --- 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 @@ -148,6 +148,15 @@ private static Kind validate( DeleteColumn delete = (DeleteColumn) change; requireTopLevel(delete.fieldNames(), "Dropping nested columns"); String name = delete.fieldNames()[0]; + // The current core drop API cannot resolve a column whose name contains a backtick under + // either the canonical (escaped) or raw representation, so reject it up front instead of + // failing mid-commit with a confusing "field not found". + if (name.indexOf('`') >= 0) { + throw new UnsupportedOperationException( + "Dropping a column whose name contains a backtick is not supported by the current " + + "Lance version: " + + name); + } if (!currentFields.contains(name) && !delete.ifExists()) { throw new UnsupportedOperationException("Cannot drop missing column: " + name); } 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 841bd685d..318801469 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 @@ -1230,6 +1230,25 @@ public void testDropColumnWithSpecialCharacterName() throws Exception { 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 testRenameColumnWithSpecialCharacterName() throws Exception { String tableName = generateTableName("rename_special"); From 7919f98dddf7aa5aba1b5b1ff408434d754da018 Mon Sep 17 00:00:00 2001 From: Pucheng Yang Date: Sat, 8 Aug 2026 00:26:18 +0000 Subject: [PATCH 8/8] fix: DROP COLUMN IF EXISTS ignores a missing backtick name Resolve column absence before applying the backtick-representability guard, so DROP COLUMN IF EXISTS on a missing column is a no-op regardless of how the absent name is spelled. The unsupported-backtick rejection now applies only to a column that actually exists and would reach dropColumns; a plain DROP of a missing column still errors. Adds a test for DROP COLUMN IF EXISTS on a missing backtick name. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../org/lance/spark/LanceSchemaEvolution.java | 18 +++++++++++------- .../spark/SparkLanceNamespaceTestBase.java | 15 +++++++++++++++ 2 files changed, 26 insertions(+), 7 deletions(-) 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 index 8b5f47c52..ad753171e 100644 --- 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 @@ -148,18 +148,22 @@ private static Kind validate( DeleteColumn delete = (DeleteColumn) change; requireTopLevel(delete.fieldNames(), "Dropping nested columns"); String name = delete.fieldNames()[0]; - // The current core drop API cannot resolve a column whose name contains a backtick under - // either the canonical (escaped) or raw representation, so reject it up front instead of - // failing mid-commit with a confusing "field not found". - if (name.indexOf('`') >= 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); } - if (!currentFields.contains(name) && !delete.ifExists()) { - throw new UnsupportedOperationException("Cannot drop missing column: " + name); - } requireDistinct(touched, name); return Kind.DROP; } else if (change instanceof RenameColumn) { 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 318801469..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 @@ -1249,6 +1249,21 @@ public void testDropColumnWithBacktickNameRejected() throws Exception { 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");