Disallow adding not null columns in Iceberg

plugin/trino-base-jdbc/src/test/java/io/trino/plugin/jdbc/TestJdbcConnectorTest.java

@@ -257,6 +257,12 @@ protected String errorMessageForInsertIntoNotNullColumn(String columnName)
         return format("NULL not allowed for column \"%s\"(?s).*", columnName.toUpperCase(ENGLISH));
     }
 
+    @Override
+    protected void verifyAddNotNullColumnToNonEmptyTableFailurePermissible(Throwable e)
+    {
+        assertThat(e).hasMessageContaining("NULL not allowed for column");
+    }
+
     @Override
     public void testAddColumnConcurrently()
     {

plugin/trino-clickhouse/src/test/java/io/trino/plugin/clickhouse/BaseClickHouseConnectorTest.java

@@ -165,6 +165,25 @@ public void testAddColumn()
         assertFalse(getQueryRunner().tableExists(getSession(), tableName));
     }
 
+    @Override
+    public void testAddNotNullColumnToNonEmptyTable()
+    {
+        // Override because the default storage type doesn't support adding columns
+        try (TestTable table = new TestTable(getQueryRunner()::execute, "test_add_notnull_col", "(a_varchar varchar NOT NULL)  WITH (engine = 'MergeTree', order_by = ARRAY['a_varchar'])")) {
+            String tableName = table.getName();
+
+            assertUpdate("ALTER TABLE " + tableName + " ADD COLUMN b_varchar varchar NOT NULL");
+            assertFalse(columnIsNullable(tableName, "b_varchar"));
+
+            assertUpdate("INSERT INTO " + tableName + " VALUES ('a', 'b')", 1);
+
+            // ClickHouse set an empty character as the default value
+            assertUpdate("ALTER TABLE " + tableName + " ADD COLUMN c_varchar varchar NOT NULL");
+            assertFalse(columnIsNullable(tableName, "c_varchar"));
+            assertQuery("SELECT c_varchar FROM " + tableName, "VALUES ''");
+        }
+    }
+
     @Test
     @Override
     public void testAddColumnWithComment()

plugin/trino-delta-lake/src/test/java/io/trino/plugin/deltalake/BaseDeltaLakeMinioConnectorTest.java

@@ -825,6 +825,14 @@ public void testTableWithNonNullableColumns()
         assertQuery("SELECT * FROM " + tableName, "VALUES(1, null, 100), (2, null, 200)");
     }
 
+    @Override
+    public void testAddNotNullColumnToNonEmptyTable()
+    {
+        // TODO https://github.com/trinodb/trino/issues/13587 Disallow adding a new column when the table isn't empty
+        assertThatThrownBy(super::testAddNotNullColumnToNonEmptyTable)
+                .hasMessageContaining("expected [false] but found [true]");
+    }
+
     @Override
     protected String createSchemaSql(String schemaName)
     {

plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/IcebergMetadata.java

@@ -1352,6 +1352,10 @@ private static Term toIcebergTerm(Schema schema, PartitionField partitionField)
     @Override
     public void addColumn(ConnectorSession session, ConnectorTableHandle tableHandle, ColumnMetadata column)
     {
+        // TODO https://github.com/trinodb/trino/issues/13587 Allow NOT NULL constraint when the table is empty
+        if (!column.isNullable()) {
+            throw new TrinoException(NOT_SUPPORTED, "This connector does not support adding not null columns");
+        }
         Table icebergTable = catalog.loadTable(session, ((IcebergTableHandle) tableHandle).getSchemaTableName());
         icebergTable.updateSchema().addColumn(column.getName(), toIcebergType(column.getType()), column.getComment()).commit();
     }

plugin/trino-iceberg/src/test/java/io/trino/plugin/iceberg/BaseIcebergConnectorTest.java

@@ -1104,6 +1104,19 @@ protected String errorMessageForInsertIntoNotNullColumn(String columnName)
         return "NULL value not allowed for NOT NULL column: " + columnName;
     }
 
+    @Override
+    public void testAddNotNullColumnToNonEmptyTable()
+    {
+        // Override because the connector supports both ADD COLUMN and NOT NULL constraint, but it doesn't support adding NOT NULL columns
+        try (TestTable table = new TestTable(getQueryRunner()::execute, "test_add_notnull_col", "(a_varchar varchar)")) {
+            String tableName = table.getName();
+
+            assertQueryFails(
+                    "ALTER TABLE " + tableName + " ADD COLUMN b_varchar varchar NOT NULL",
+                    "This connector does not support adding not null columns");
+        }
+    }
+
     @Test
     public void testSchemaEvolution()
     {

plugin/trino-oracle/src/test/java/io/trino/plugin/oracle/BaseOracleConnectorTest.java

@@ -469,6 +469,12 @@ protected String errorMessageForInsertIntoNotNullColumn(String columnName)
         return format("ORA-01400: cannot insert NULL into \\(.*\"%s\"\\)\n", columnName.toUpperCase(ENGLISH));
     }
 
+    @Override
+    protected void verifyAddNotNullColumnToNonEmptyTableFailurePermissible(Throwable e)
+    {
+        assertThat(e).hasMessage("ORA-01758: table must be empty to add mandatory (NOT NULL) column\n");
+    }
+
     @Override
     protected void verifyConcurrentAddColumnFailurePermissible(Exception e)
     {

plugin/trino-postgresql/src/test/java/io/trino/plugin/postgresql/TestPostgreSqlConnectorTest.java

@@ -165,6 +165,12 @@ protected TestTable createTableWithUnsupportedColumn()
                 "(one bigint, two decimal(50,0), three varchar(10))");
     }
 
+    @Override
+    protected void verifyAddNotNullColumnToNonEmptyTableFailurePermissible(Throwable e)
+    {
+        assertThat(e).hasMessageMatching("ERROR: column \".*\" contains null values");
+    }
+
     @Test
     public void testViews()
     {

plugin/trino-sqlserver/src/test/java/io/trino/plugin/sqlserver/BaseSqlServerConnectorTest.java

@@ -179,6 +179,16 @@ public void testReadMetadataWithRelationsConcurrentModifications()
         }
     }
 
+    @Override
+    protected void verifyAddNotNullColumnToNonEmptyTableFailurePermissible(Throwable e)
+    {
+        assertThat(e).hasMessageMatching(
+                "ALTER TABLE only allows columns to be added that can contain nulls, " +
+                        "or have a DEFAULT definition specified, or the column being added is an identity or timestamp column, " +
+                        "or alternatively if none of the previous conditions are satisfied the table must be empty to allow addition of this column\\. " +
+                        "Column '.*' cannot be added to non-empty table '.*' because it does not satisfy these conditions\\.");
+    }
+
     @Override
     protected void verifyConcurrentAddColumnFailurePermissible(Exception e)
     {

testing/trino-testing/src/main/java/io/trino/testing/BaseConnectorTest.java

@@ -127,6 +127,7 @@
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
 import static org.testng.Assert.assertFalse;
+import static org.testng.Assert.assertNotNull;
 import static org.testng.Assert.assertNull;
 import static org.testng.Assert.assertTrue;
 import static org.testng.Assert.fail;
@@ -1875,6 +1876,50 @@ public void testAddColumnWithComment()
         }
     }
 
+    @Test
+    public void testAddNotNullColumnToNonEmptyTable()
+    {
+        skipTestUnless(hasBehavior(SUPPORTS_ADD_COLUMN));
+
+        if (!hasBehavior(SUPPORTS_NOT_NULL_CONSTRAINT)) {
+            assertQueryFails(
+                    "ALTER TABLE nation ADD COLUMN test_add_not_null_col bigint NOT NULL",
+                    ".* Catalog '.*' does not support NOT NULL for column '.*'");
+            return;
+        }
+
+        try (TestTable table = new TestTable(getQueryRunner()::execute, "test_add_notnull_col", "(a_varchar varchar)")) {
+            String tableName = table.getName();
+
+            assertUpdate("ALTER TABLE " + tableName + " ADD COLUMN b_varchar varchar NOT NULL");
+            assertFalse(columnIsNullable(tableName, "b_varchar"));
+
+            assertUpdate("INSERT INTO " + tableName + " VALUES ('a', 'b')", 1);
+            try {
+                assertUpdate("ALTER TABLE " + tableName + " ADD COLUMN c_varchar varchar NOT NULL");
+                assertFalse(columnIsNullable(tableName, "c_varchar"));
+                // Remote database might set implicit default values
+                assertNotNull(computeScalar("SELECT c_varchar FROM " + tableName));
+            }
+            catch (Throwable e) {
+                verifyAddNotNullColumnToNonEmptyTableFailurePermissible(e);
+            }
+        }
+    }
+
+    protected boolean columnIsNullable(String tableName, String columnName)
+    {
+        String isNullable = (String) computeScalar(
+                "SELECT is_nullable FROM information_schema.columns WHERE " +
+                "table_schema = '" + getSession().getSchema().orElseThrow() + "' AND table_name = '" + tableName + "' AND column_name = '" + columnName + "'");
+        return "YES".equals(isNullable);
+    }
+
+    protected void verifyAddNotNullColumnToNonEmptyTableFailurePermissible(Throwable e)
+    {
+        throw new AssertionError("Unexpected adding not null columns failure", e);
+    }
+
     @Test
     public void testDropColumn()
     {