Persistence Component: Fix alter table command for DuckDB

legend-engine-xts-persistence/legend-engine-xt-persistence-component/legend-engine-xt-persistence-component-relational-core/src/main/java/org/finos/legend/engine/persistence/components/relational/api/IngestModeOptimizer.java

@@ -87,7 +87,9 @@ public IngestMode visitUnitemporalSnapshot(UnitemporalSnapshotAbstract unitempor
                             .addAllPartitionFields(partition.partitionFields())
                             .addAllPartitionSpecList(derivePartitionSpecList(partition.partitionFields(), partition.maxPartitionSpecFilters()))
                             .derivePartitionSpec(partition.derivePartitionSpec())
-                            .maxPartitionSpecFilters(partition.maxPartitionSpecFilters()).build())
+                            .maxPartitionSpecFilters(partition.maxPartitionSpecFilters())
+                            .deleteStrategy(partition.deleteStrategy())
+                            .build())
                         .build();
             }
         }

legend-engine-xts-persistence/legend-engine-xt-persistence-component/legend-engine-xt-persistence-component-relational-duckdb/src/main/java/org/finos/legend/engine/persistence/components/relational/duckdb/DuckDBSink.java

@@ -34,6 +34,7 @@
 import org.finos.legend.engine.persistence.components.logicalplan.datasets.StagedFilesDataset;
 import org.finos.legend.engine.persistence.components.logicalplan.datasets.StagedFilesDatasetReference;
 import org.finos.legend.engine.persistence.components.logicalplan.datasets.StagedFilesSelection;
+import org.finos.legend.engine.persistence.components.logicalplan.operations.Alter;
 import org.finos.legend.engine.persistence.components.logicalplan.operations.Copy;
 import org.finos.legend.engine.persistence.components.logicalplan.operations.Update;
 import org.finos.legend.engine.persistence.components.logicalplan.values.CastFunction;
@@ -50,6 +51,7 @@
 import org.finos.legend.engine.persistence.components.relational.duckdb.jdbc.DuckDBJdbcHelper;
 import org.finos.legend.engine.persistence.components.relational.duckdb.sql.DuckDBDataTypeMapping;
 import org.finos.legend.engine.persistence.components.relational.duckdb.sql.DuckDBJdbcPropertiesToLogicalDataTypeMapping;
+import org.finos.legend.engine.persistence.components.relational.duckdb.sql.visitor.AlterVisitor;
 import org.finos.legend.engine.persistence.components.relational.duckdb.sql.visitor.CastFunctionVisitor;
 import org.finos.legend.engine.persistence.components.relational.duckdb.sql.visitor.ConcatFunctionVisitor;
 import org.finos.legend.engine.persistence.components.relational.duckdb.sql.visitor.CopyVisitor;
@@ -113,6 +115,7 @@ public class DuckDBSink extends AnsiSqlSink
         logicalPlanVisitorByClass.put(SchemaDefinition.class, new SchemaDefinitionVisitor());
         logicalPlanVisitorByClass.put(Field.class, new FieldVisitor());
         logicalPlanVisitorByClass.put(Update.class, new SQLUpdateVisitor());
+        logicalPlanVisitorByClass.put(Alter.class, new AlterVisitor());
         logicalPlanVisitorByClass.put(ParseJsonFunction.class, new ParseJsonFunctionVisitor());
         logicalPlanVisitorByClass.put(Copy.class, new CopyVisitor());
         logicalPlanVisitorByClass.put(StagedFilesDatasetReference.class, new StagedFilesDatasetReferenceVisitor());

legend-engine-xts-persistence/legend-engine-xt-persistence-component/legend-engine-xt-persistence-component-relational-duckdb/src/main/java/org/finos/legend/engine/persistence/components/relational/duckdb/sql/visitor/AlterVisitor.java

@@ -0,0 +1,43 @@
+// Copyright 2025 Goldman Sachs
+//
+// 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.finos.legend.engine.persistence.components.relational.duckdb.sql.visitor;
+
+import org.finos.legend.engine.persistence.components.logicalplan.LogicalPlanNode;
+import org.finos.legend.engine.persistence.components.logicalplan.operations.Alter;
+import org.finos.legend.engine.persistence.components.physicalplan.PhysicalPlanNode;
+import org.finos.legend.engine.persistence.components.relational.duckdb.sqldom.schemaops.statements.AlterTable;
+import org.finos.legend.engine.persistence.components.relational.sqldom.common.AlterOperation;
+import org.finos.legend.engine.persistence.components.transformer.LogicalPlanVisitor;
+import org.finos.legend.engine.persistence.components.transformer.VisitorContext;
+
+import java.util.ArrayList;
+import java.util.List;
+
+public class AlterVisitor implements LogicalPlanVisitor<Alter>
+{
+
+    @Override
+    public VisitorResult visit(PhysicalPlanNode prev, Alter current, VisitorContext context)
+    {
+        AlterTable alterTable = new AlterTable(AlterOperation.valueOf(current.operation().name()));
+        prev.push(alterTable);
+
+        List<LogicalPlanNode> logicalPlanNodeList = new ArrayList<>();
+        logicalPlanNodeList.add(current.dataset());
+        logicalPlanNodeList.add(current.columnDetails());
+
+        return new VisitorResult(alterTable, logicalPlanNodeList);
+    }
+}

legend-engine-xts-persistence/legend-engine-xt-persistence-component/legend-engine-xt-persistence-component-relational-duckdb/src/main/java/org/finos/legend/engine/persistence/components/relational/duckdb/sqldom/schemaops/statements/AlterTable.java

@@ -0,0 +1,126 @@
+// Copyright 2025 Goldman Sachs
+//
+// 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.finos.legend.engine.persistence.components.relational.duckdb.sqldom.schemaops.statements;
+
+import org.finos.legend.engine.persistence.components.relational.sqldom.SqlDomException;
+import org.finos.legend.engine.persistence.components.relational.sqldom.common.AlterOperation;
+import org.finos.legend.engine.persistence.components.relational.sqldom.common.Clause;
+import org.finos.legend.engine.persistence.components.relational.sqldom.constraints.column.NotNullColumnConstraint;
+import org.finos.legend.engine.persistence.components.relational.sqldom.schemaops.Column;
+import org.finos.legend.engine.persistence.components.relational.sqldom.schemaops.expresssions.table.Table;
+import org.finos.legend.engine.persistence.components.relational.sqldom.schemaops.statements.DDLStatement;
+
+import static org.finos.legend.engine.persistence.components.relational.sqldom.common.Clause.COLUMN;
+import static org.finos.legend.engine.persistence.components.relational.sqldom.common.Clause.DATA_TYPE;
+import static org.finos.legend.engine.persistence.components.relational.sqldom.common.Clause.DROP;
+import static org.finos.legend.engine.persistence.components.relational.sqldom.common.Clause.SET;
+import static org.finos.legend.engine.persistence.components.relational.sqldom.utils.SqlGenUtils.WHITE_SPACE;
+
+public class AlterTable implements DDLStatement
+{
+    private final AlterOperation operation;
+    private Table table;
+    private Column columnToAlter;
+
+    public AlterTable(AlterOperation operation)
+    {
+        this.operation = operation;
+    }
+
+    /*
+    ALTER TABLE [ IF EXISTS ] tableName ADD COLUMN columnName { columnDefinition }
+
+     ALTER TABLE [ IF EXISTS ] tableName ALTER COLUMN columnName
+     { columnDefinition }
+     | { SET NULL } }
+     */
+    @Override
+    public void genSql(StringBuilder builder) throws SqlDomException
+    {
+        validate();
+        builder.append(Clause.ALTER.get());
+
+        builder.append(WHITE_SPACE + Clause.TABLE.get());
+
+        // Table name
+        builder.append(WHITE_SPACE);
+        table.genSqlWithoutAlias(builder);
+
+        // Operation
+        builder.append(WHITE_SPACE);
+        if (operation.getParent() == null)
+        {
+            builder.append(operation.name());
+        }
+        else
+        {
+            builder.append(operation.getParent().name());
+        }
+        builder.append(WHITE_SPACE);
+        builder.append(COLUMN);
+        builder.append(WHITE_SPACE);
+
+        switch (operation)
+        {
+            case ADD:
+                columnToAlter.genSql(builder);
+                break;
+            case CHANGE_DATATYPE:
+                columnToAlter.genSqlWithNameOnly(builder);
+                builder.append(WHITE_SPACE);
+                builder.append(SET.get());
+                builder.append(WHITE_SPACE);
+                builder.append(DATA_TYPE.get());
+                builder.append(WHITE_SPACE);
+                columnToAlter.genSqlWithTypeOnly(builder);
+                break;
+            case NULLABLE_COLUMN:
+                columnToAlter.genSqlWithNameOnly(builder);
+                builder.append(WHITE_SPACE);
+                builder.append(DROP);
+                builder.append(WHITE_SPACE);
+                NotNullColumnConstraint notNullColumnConstraint = new NotNullColumnConstraint();
+                notNullColumnConstraint.genSql(builder);
+                break;
+            default:
+                throw new SqlDomException("Alter operation " + operation.name() + " not supported");
+        }
+    }
+
+    @Override
+    public void push(Object node)
+    {
+        if (node instanceof Table)
+        {
+            table = (Table) node;
+        }
+        else if (node instanceof Column)
+        {
+            columnToAlter = (Column) node;
+        }
+    }
+
+    void validate() throws SqlDomException
+    {
+        if (table == null)
+        {
+            throw new SqlDomException("Table is mandatory for Alter Table Command");
+        }
+        if (columnToAlter == null)
+        {
+            throw new SqlDomException("Columns details is mandatory for Alter Table Command");
+        }
+    }
+}

legend-engine-xts-persistence/legend-engine-xt-persistence-component/legend-engine-xt-persistence-component-relational-duckdb/src/test/java/org/finos/legend/engine/persistence/components/e2e/AlterTest.java

@@ -0,0 +1,155 @@
+// Copyright 2025 Goldman Sachs
+//
+// 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.finos.legend.engine.persistence.components.e2e;
+
+import org.finos.legend.engine.persistence.components.logicalplan.LogicalPlan;
+import org.finos.legend.engine.persistence.components.logicalplan.LogicalPlanFactory;
+import org.finos.legend.engine.persistence.components.logicalplan.datasets.DatasetDefinition;
+import org.finos.legend.engine.persistence.components.logicalplan.operations.Alter;
+import org.finos.legend.engine.persistence.components.logicalplan.operations.Operation;
+import org.finos.legend.engine.persistence.components.relational.SqlPlan;
+import org.finos.legend.engine.persistence.components.relational.duckdb.DuckDBSink;
+import org.finos.legend.engine.persistence.components.relational.transformer.RelationalTransformer;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+
+import static org.finos.legend.engine.persistence.components.e2e.TestUtils.alterColumn;
+import static org.finos.legend.engine.persistence.components.e2e.TestUtils.alterColumnName;
+import static org.finos.legend.engine.persistence.components.e2e.TestUtils.batchIdName;
+import static org.finos.legend.engine.persistence.components.e2e.TestUtils.digestName;
+import static org.finos.legend.engine.persistence.components.e2e.TestUtils.expiryDateName;
+import static org.finos.legend.engine.persistence.components.e2e.TestUtils.idName;
+import static org.finos.legend.engine.persistence.components.e2e.TestUtils.incomeChanged;
+import static org.finos.legend.engine.persistence.components.e2e.TestUtils.incomeName;
+import static org.finos.legend.engine.persistence.components.e2e.TestUtils.mainTableName;
+import static org.finos.legend.engine.persistence.components.e2e.TestUtils.name;
+import static org.finos.legend.engine.persistence.components.e2e.TestUtils.nameName;
+import static org.finos.legend.engine.persistence.components.e2e.TestUtils.startTimeName;
+import static org.finos.legend.engine.persistence.components.e2e.TestUtils.testDatabaseName;
+import static org.finos.legend.engine.persistence.components.e2e.TestUtils.testSchemaName;
+
+class AlterTest extends BaseTest
+{
+    private String basePath = "src/test/resources/data/alter-table/";
+
+    @Test
+    void testAlterTableAddColumn() throws Exception
+    {
+        // Prepare main table
+        DatasetDefinition mainTable = TestUtils.getBasicMainTable();
+
+        RelationalTransformer transformer = new RelationalTransformer(DuckDBSink.get());
+        LogicalPlan tableCreationPlan = LogicalPlanFactory.getDatasetCreationPlan(mainTable, true);
+        SqlPlan tableCreationPhysicalPlan = transformer.generatePhysicalPlan(tableCreationPlan);
+        executor.executePhysicalPlan(tableCreationPhysicalPlan);
+        String inputPath = basePath + "input/add_data_pass.csv";
+        insertMainData(inputPath);
+
+        // Generate and execute schema evolution plan
+        Operation add = Alter.of(mainTable, Alter.AlterOperation.ADD, alterColumn, Optional.empty());
+        LogicalPlan logicalPlan = LogicalPlan.builder().addOps(add).build();
+        SqlPlan schemaEvolutionPhysicalPlan = transformer.generatePhysicalPlan(logicalPlan);
+        executor.executePhysicalPlan(schemaEvolutionPhysicalPlan);
+
+        // Verify the new schema
+        List<Map<String, Object>> actualTableData = duckDBSink.executeQuery("select * from \"TEST\".\"main\"");
+        List<String> expectedNewSchema = Arrays.asList(idName, nameName, incomeName, startTimeName, expiryDateName, digestName, batchIdName, alterColumnName);
+        String expectedPath = basePath + "expected/add_expected_pass.csv";
+        TestUtils.assertTableColumnsEquals(expectedNewSchema, actualTableData);
+        TestUtils.assertFileAndTableDataEquals(expectedNewSchema.toArray(new String[]{}), expectedPath, actualTableData);
+    }
+
+    @Test
+    void testAlterTableChangeDataType() throws Exception
+    {
+        // Prepare main table
+        DatasetDefinition mainTable = TestUtils.getBasicMainTable();
+
+        RelationalTransformer transformer = new RelationalTransformer(DuckDBSink.get());
+        LogicalPlan tableCreationPlan = LogicalPlanFactory.getDatasetCreationPlan(mainTable, true);
+        SqlPlan tableCreationPhysicalPlan = transformer.generatePhysicalPlan(tableCreationPlan);
+        executor.executePhysicalPlan(tableCreationPhysicalPlan);
+        String inputPath = basePath + "input/change_type_data_pass.csv";
+        insertMainData(inputPath);
+
+        // Assert column is of type BIGINT before operation
+        String dataType = TestUtils.getColumnDataTypeFromTable(duckDBSink.connection(), testDatabaseName, testSchemaName, mainTableName, incomeName);
+        Assertions.assertEquals("BIGINT", dataType);
+
+        // Generate and execute schema evolution plan
+        Operation changeDataType = Alter.of(mainTable, Alter.AlterOperation.CHANGE_DATATYPE, incomeChanged, Optional.empty());
+        LogicalPlan logicalPlan = LogicalPlan.builder().addOps(changeDataType).build();
+        SqlPlan schemaEvolutionPhysicalPlan = transformer.generatePhysicalPlan(logicalPlan);
+        executor.executePhysicalPlan(schemaEvolutionPhysicalPlan);
+
+        // Verify the new schema
+        List<Map<String, Object>> actualTableData = duckDBSink.executeQuery("select * from \"TEST\".\"main\"");
+        List<String> expectedNewSchema = Arrays.asList(idName, nameName, incomeName, startTimeName, expiryDateName, digestName, batchIdName);
+        String expectedPath = basePath + "expected/change_type_expected_pass.csv";
+        TestUtils.assertTableColumnsEquals(expectedNewSchema, actualTableData);
+        TestUtils.assertFileAndTableDataEquals(expectedNewSchema.toArray(new String[]{}), expectedPath, actualTableData);
+
+        // Assert column is of type INTEGER after operation
+        dataType = TestUtils.getColumnDataTypeFromTable(duckDBSink.connection(), testDatabaseName, testSchemaName, mainTableName, incomeName);
+        Assertions.assertEquals("INTEGER", dataType);
+    }
+
+    @Test
+    void testAlterTableNullableColumn() throws Exception
+    {
+        // Prepare main table
+        DatasetDefinition mainTable = TestUtils.getBasicMainTable();
+        LogicalPlan tableCreationPlan = LogicalPlanFactory.getDatasetCreationPlan(mainTable, true);
+
+        RelationalTransformer transformer = new RelationalTransformer(DuckDBSink.get());
+        SqlPlan tableCreationPhysicalPlan = transformer.generatePhysicalPlan(tableCreationPlan);
+        executor.executePhysicalPlan(tableCreationPhysicalPlan);
+        String inputPath = basePath + "input/nullable_column_data_pass.csv";
+        insertMainData(inputPath);
+
+        // Assert column is not nullable before operation
+        Assertions.assertEquals("NO", TestUtils.getIsColumnNullableFromTable(duckDBSink, mainTableName, nameName));
+
+        // Generate and execute schema evolution plan
+        Operation nullableColumn = Alter.of(mainTable, Alter.AlterOperation.NULLABLE_COLUMN, name, Optional.empty());
+        LogicalPlan logicalPlan = LogicalPlan.builder().addOps(nullableColumn).build();
+        SqlPlan schemaEvolutionPhysicalPlan = transformer.generatePhysicalPlan(logicalPlan);
+        executor.executePhysicalPlan(schemaEvolutionPhysicalPlan);
+
+        // Verify the new schema
+        List<Map<String, Object>> actualTableData = duckDBSink.executeQuery("select * from \"TEST\".\"main\"");
+        List<String> expectedNewSchema = Arrays.asList(idName, nameName, incomeName, startTimeName, expiryDateName, digestName, batchIdName);
+        String expectedPath = basePath + "expected/nullable_column_expected_pass.csv";
+        TestUtils.assertTableColumnsEquals(expectedNewSchema, actualTableData);
+        TestUtils.assertFileAndTableDataEquals(expectedNewSchema.toArray(new String[]{}), expectedPath, actualTableData);
+
+        // Assert column is nullable after operation
+        Assertions.assertEquals("YES", TestUtils.getIsColumnNullableFromTable(duckDBSink, mainTableName, nameName));
+    }
+
+    private void insertMainData(String path) throws Exception
+    {
+        String loadSql = "TRUNCATE TABLE \"TEST\".\"main\";" +
+            "COPY \"TEST\".\"main\"" +
+            "(\"id\", \"name\", \"income\", \"start_time\", \"expiry_date\", \"digest\", \"batch_id\")" +
+            " FROM '" + path + "' CSV";
+        duckDBSink.executeStatement(loadSql);
+    }
+}

legend-engine-xts-persistence/legend-engine-xt-persistence-component/legend-engine-xt-persistence-component-relational-duckdb/src/test/java/org/finos/legend/engine/persistence/components/e2e/TestUtils.java

@@ -1512,8 +1512,8 @@ public static List<String> getColumnsFromTable(Connection connection, String dat
     // This is to check the actual database table - whether columns have the right nullability
     public static String getIsColumnNullableFromTable(RelationalExecutionHelper sink, String tableName, String columnName)
     {
-        List<Map<String, Object>> result = sink.executeQuery("SELECT IS_NULLABLE FROM INFORMATION_SCHEMA.COLUMNS WHERE TABLE_NAME='" + tableName + "' and COLUMN_NAME ='" + columnName + "'");
-        return result.get(0).get("IS_NULLABLE").toString();
+        List<Map<String, Object>> result = sink.executeQuery("SELECT is_nullable FROM information_schema.columns WHERE table_name ='" + tableName + "' and column_name  ='" + columnName + "'");
+        return result.get(0).get("is_nullable").toString();
     }
 
     // This is to check the actual database table - the length (precision) of the column data type