Add support for NOT NULL in DDL statements

Enable NOT NULL in the parser and add a flag in the metadata to check
whether or not a connector supports this new functionality. If a
connector doesn't support NOT NULL semantics in columns, yet this is
supplied, then a semantic exception is thrown indicating to the user
that this functionality is not supported in their catalog.

Extracted-From: https://github.com/prestodb/presto

presto-main/src/main/java/io/prestosql/execution/AddColumnTask.java

@@ -37,10 +37,12 @@
 import static com.google.common.util.concurrent.Futures.immediateFuture;
 import static io.prestosql.metadata.MetadataUtil.createQualifiedObjectName;
 import static io.prestosql.spi.StandardErrorCode.NOT_FOUND;
+import static io.prestosql.spi.connector.ConnectorCapabilities.NOT_NULL_COLUMN_CONSTRAINT;
 import static io.prestosql.spi.type.TypeSignature.parseTypeSignature;
 import static io.prestosql.sql.NodeUtils.mapFromProperties;
 import static io.prestosql.sql.analyzer.SemanticErrorCode.COLUMN_ALREADY_EXISTS;
 import static io.prestosql.sql.analyzer.SemanticErrorCode.MISSING_TABLE;
+import static io.prestosql.sql.analyzer.SemanticErrorCode.NOT_SUPPORTED;
 import static io.prestosql.sql.analyzer.SemanticErrorCode.TYPE_MISMATCH;
 import static io.prestosql.type.UnknownType.UNKNOWN;
 import static java.util.Locale.ENGLISH;
@@ -85,6 +87,9 @@ public ListenableFuture<?> execute(AddColumn statement, TransactionManager trans
         if (columnHandles.containsKey(element.getName().getValue().toLowerCase(ENGLISH))) {
             throw new SemanticException(COLUMN_ALREADY_EXISTS, statement, "Column '%s' already exists", element.getName());
         }
+        if (!element.isNullable() && !metadata.getConnectorCapabilities(session, connectorId).contains(NOT_NULL_COLUMN_CONSTRAINT)) {
+            throw new SemanticException(NOT_SUPPORTED, element, "Catalog '%s' does not support NOT NULL for column '%s'", connectorId.getCatalogName(), element.getName());
+        }
 
         Map<String, Expression> sqlProperties = mapFromProperties(element.getProperties());
         Map<String, Object> columnProperties = metadata.getColumnPropertyManager().getProperties(
@@ -95,7 +100,13 @@ public ListenableFuture<?> execute(AddColumn statement, TransactionManager trans
                 metadata,
                 parameters);
 
-        ColumnMetadata column = new ColumnMetadata(element.getName().getValue(), type, element.getComment().orElse(null), null, false, columnProperties);
+        ColumnMetadata column = new ColumnMetadata(
+                element.getName().getValue(),
+                type,
+                element.isNullable(), element.getComment().orElse(null),
+                null,
+                false,
+                columnProperties);
 
         metadata.addColumn(session, tableHandle.get(), column);
 

presto-main/src/main/java/io/prestosql/execution/CreateTableTask.java

@@ -50,6 +50,7 @@
 import static io.prestosql.spi.StandardErrorCode.ALREADY_EXISTS;
 import static io.prestosql.spi.StandardErrorCode.GENERIC_INTERNAL_ERROR;
 import static io.prestosql.spi.StandardErrorCode.NOT_FOUND;
+import static io.prestosql.spi.connector.ConnectorCapabilities.NOT_NULL_COLUMN_CONSTRAINT;
 import static io.prestosql.spi.type.TypeSignature.parseTypeSignature;
 import static io.prestosql.sql.NodeUtils.mapFromProperties;
 import static io.prestosql.sql.analyzer.SemanticErrorCode.DUPLICATE_COLUMN_NAME;
@@ -118,6 +119,9 @@ public ListenableFuture<?> internalExecute(CreateTable statement, Metadata metad
                 if (columns.containsKey(name)) {
                     throw new SemanticException(DUPLICATE_COLUMN_NAME, column, "Column name '%s' specified more than once", column.getName());
                 }
+                if (!column.isNullable() && !metadata.getConnectorCapabilities(session, connectorId).contains(NOT_NULL_COLUMN_CONSTRAINT)) {
+                    throw new SemanticException(NOT_SUPPORTED, column, "Catalog '%s' does not support non-null column for column name '%s'", connectorId.getCatalogName(), column.getName());
+                }
 
                 Map<String, Expression> sqlProperties = mapFromProperties(column.getProperties());
                 Map<String, Object> columnProperties = metadata.getColumnPropertyManager().getProperties(
@@ -128,7 +132,13 @@ public ListenableFuture<?> internalExecute(CreateTable statement, Metadata metad
                         metadata,
                         parameters);
 
-                columns.put(name, new ColumnMetadata(name, type, column.getComment().orElse(null), null, false, columnProperties));
+                columns.put(name, new ColumnMetadata(
+                        name,
+                        type,
+                        column.isNullable(), column.getComment().orElse(null),
+                        null,
+                        false,
+                        columnProperties));
             }
             else if (element instanceof LikeClause) {
                 LikeClause likeClause = (LikeClause) element;

presto-main/src/main/java/io/prestosql/metadata/CatalogMetadata.java

@@ -16,12 +16,15 @@
 import com.google.common.collect.ImmutableList;
 import io.prestosql.Session;
 import io.prestosql.connector.ConnectorId;
+import io.prestosql.spi.connector.ConnectorCapabilities;
 import io.prestosql.spi.connector.ConnectorMetadata;
 import io.prestosql.spi.connector.ConnectorTransactionHandle;
 
 import java.util.List;
+import java.util.Set;
 
 import static com.google.common.base.MoreObjects.toStringHelper;
+import static com.google.common.collect.Sets.immutableEnumSet;
 import static java.util.Objects.requireNonNull;
 
 public class CatalogMetadata
@@ -39,6 +42,7 @@ public class CatalogMetadata
     private final ConnectorId systemTablesId;
     private final ConnectorMetadata systemTables;
     private final ConnectorTransactionHandle systemTablesTransactionHandle;
+    private final Set<ConnectorCapabilities> connectorCapabilities;
 
     public CatalogMetadata(
             ConnectorId connectorId,
@@ -49,7 +53,8 @@ public CatalogMetadata(
             ConnectorTransactionHandle informationSchemaTransactionHandle,
             ConnectorId systemTablesId,
             ConnectorMetadata systemTables,
-            ConnectorTransactionHandle systemTablesTransactionHandle)
+            ConnectorTransactionHandle systemTablesTransactionHandle,
+            Set<ConnectorCapabilities> connectorCapabilities)
     {
         this.connectorId = requireNonNull(connectorId, "connectorId is null");
         this.metadata = requireNonNull(metadata, "metadata is null");
@@ -60,6 +65,7 @@ public CatalogMetadata(
         this.systemTablesId = requireNonNull(systemTablesId, "systemTablesId is null");
         this.systemTables = requireNonNull(systemTables, "systemTables is null");
         this.systemTablesTransactionHandle = requireNonNull(systemTablesTransactionHandle, "systemTablesTransactionHandle is null");
+        this.connectorCapabilities = immutableEnumSet(requireNonNull(connectorCapabilities, "connectorCapabilities is null"));
     }
 
     public ConnectorId getConnectorId()
@@ -118,6 +124,11 @@ public List<ConnectorId> listConnectorIds()
         return ImmutableList.of(informationSchemaId, systemTablesId, connectorId);
     }
 
+    public Set<ConnectorCapabilities> getConnectorCapabilities()
+    {
+        return connectorCapabilities;
+    }
+
     @Override
     public String toString()
     {

presto-main/src/main/java/io/prestosql/metadata/Metadata.java

@@ -21,6 +21,7 @@
 import io.prestosql.spi.connector.CatalogSchemaName;
 import io.prestosql.spi.connector.ColumnHandle;
 import io.prestosql.spi.connector.ColumnMetadata;
+import io.prestosql.spi.connector.ConnectorCapabilities;
 import io.prestosql.spi.connector.ConnectorOutputMetadata;
 import io.prestosql.spi.connector.ConnectorTableMetadata;
 import io.prestosql.spi.connector.Constraint;
@@ -382,6 +383,8 @@ public interface Metadata
 
     AnalyzePropertyManager getAnalyzePropertyManager();
 
+    Set<ConnectorCapabilities> getConnectorCapabilities(Session session, ConnectorId catalogName);
+
     @Deprecated
     boolean usesLegacyTableLayouts(Session session, TableHandle table);
 }

presto-main/src/main/java/io/prestosql/metadata/MetadataManager.java

@@ -33,6 +33,7 @@
 import io.prestosql.spi.connector.CatalogSchemaName;
 import io.prestosql.spi.connector.ColumnHandle;
 import io.prestosql.spi.connector.ColumnMetadata;
+import io.prestosql.spi.connector.ConnectorCapabilities;
 import io.prestosql.spi.connector.ConnectorInsertTableHandle;
 import io.prestosql.spi.connector.ConnectorMetadata;
 import io.prestosql.spi.connector.ConnectorOutputMetadata;
@@ -1164,6 +1165,12 @@ public AnalyzePropertyManager getAnalyzePropertyManager()
         return analyzePropertyManager;
     }
 
+    @Override
+    public Set<ConnectorCapabilities> getConnectorCapabilities(Session session, ConnectorId connectorId)
+    {
+        return getCatalogMetadata(session, connectorId).getConnectorCapabilities();
+    }
+
     @Override
     public boolean usesLegacyTableLayouts(Session session, TableHandle table)
     {

presto-main/src/main/java/io/prestosql/sql/rewrite/ShowQueriesRewrite.java

@@ -457,7 +457,7 @@ protected Node visitShowCreate(ShowCreate node, Void context)
                         .filter(column -> !column.isHidden())
                         .map(column -> {
                             List<Property> propertyNodes = buildProperties(objectName, Optional.of(column.getName()), INVALID_COLUMN_PROPERTY, column.getProperties(), allColumnProperties);
-                            return new ColumnDefinition(new Identifier(column.getName()), column.getType().getDisplayName(), propertyNodes, Optional.ofNullable(column.getComment()));
+                            return new ColumnDefinition(new Identifier(column.getName()), column.getType().getDisplayName(), column.isNullable(), propertyNodes, Optional.ofNullable(column.getComment()));
                         })
                         .collect(toImmutableList());
 

presto-main/src/main/java/io/prestosql/transaction/InMemoryTransactionManager.java

@@ -412,15 +412,23 @@ private synchronized CatalogMetadata getTransactionCatalogMetadata(ConnectorId c
             if (catalogMetadata == null) {
                 Catalog catalog = catalogsByConnectorId.get(connectorId);
                 verify(catalog != null, "Unknown connectorId: %s", connectorId);
+                Connector connector = catalog.getConnector(connectorId);
 
                 ConnectorTransactionMetadata metadata = createConnectorTransactionMetadata(catalog.getConnectorId(), catalog);
                 ConnectorTransactionMetadata informationSchema = createConnectorTransactionMetadata(catalog.getInformationSchemaId(), catalog);
                 ConnectorTransactionMetadata systemTables = createConnectorTransactionMetadata(catalog.getSystemTablesId(), catalog);
 
                 catalogMetadata = new CatalogMetadata(
-                        metadata.getConnectorId(), metadata.getConnectorMetadata(), metadata.getTransactionHandle(),
-                        informationSchema.getConnectorId(), informationSchema.getConnectorMetadata(), informationSchema.getTransactionHandle(),
-                        systemTables.getConnectorId(), systemTables.getConnectorMetadata(), systemTables.getTransactionHandle());
+                        metadata.getConnectorId(),
+                        metadata.getConnectorMetadata(),
+                        metadata.getTransactionHandle(),
+                        informationSchema.getConnectorId(),
+                        informationSchema.getConnectorMetadata(),
+                        informationSchema.getTransactionHandle(),
+                        systemTables.getConnectorId(),
+                        systemTables.getConnectorMetadata(),
+                        systemTables.getTransactionHandle(),
+                        connector.getCapabilities());
 
                 this.catalogMetadata.put(catalog.getConnectorId(), catalogMetadata);
                 this.catalogMetadata.put(catalog.getInformationSchemaId(), catalogMetadata);

presto-main/src/test/java/io/prestosql/execution/TestCreateTableTask.java

@@ -27,31 +27,42 @@
 import io.prestosql.security.AllowAllAccessControl;
 import io.prestosql.spi.PrestoException;
 import io.prestosql.spi.connector.ColumnHandle;
+import io.prestosql.spi.connector.ColumnMetadata;
+import io.prestosql.spi.connector.ConnectorCapabilities;
 import io.prestosql.spi.connector.ConnectorTableMetadata;
 import io.prestosql.spi.type.Type;
 import io.prestosql.spi.type.TypeManager;
 import io.prestosql.spi.type.TypeSignature;
+import io.prestosql.sql.analyzer.SemanticException;
 import io.prestosql.sql.tree.ColumnDefinition;
 import io.prestosql.sql.tree.CreateTable;
 import io.prestosql.sql.tree.QualifiedName;
+import io.prestosql.sql.tree.TableElement;
 import io.prestosql.transaction.TransactionManager;
 import io.prestosql.type.TypeRegistry;
 import org.testng.annotations.BeforeMethod;
 import org.testng.annotations.Test;
 
+import java.util.List;
 import java.util.Optional;
-import java.util.concurrent.atomic.AtomicInteger;
+import java.util.Set;
+import java.util.concurrent.CopyOnWriteArrayList;
 
+import static com.google.common.collect.Sets.immutableEnumSet;
 import static io.airlift.concurrent.MoreFutures.getFutureValue;
 import static io.prestosql.spi.StandardErrorCode.ALREADY_EXISTS;
+import static io.prestosql.spi.connector.ConnectorCapabilities.NOT_NULL_COLUMN_CONSTRAINT;
 import static io.prestosql.spi.session.PropertyMetadata.stringProperty;
 import static io.prestosql.sql.QueryUtil.identifier;
 import static io.prestosql.testing.TestingSession.createBogusTestingCatalog;
 import static io.prestosql.testing.TestingSession.testSessionBuilder;
 import static io.prestosql.transaction.InMemoryTransactionManager.createTestTransactionManager;
 import static java.util.Collections.emptyList;
+import static java.util.Collections.emptySet;
+import static java.util.Locale.ENGLISH;
 import static java.util.Objects.requireNonNull;
 import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertFalse;
 import static org.testng.Assert.assertTrue;
 import static org.testng.Assert.fail;
 
@@ -87,14 +98,15 @@ public void setUp()
         metadata = new MockMetadata(typeManager,
                 tablePropertyManager,
                 columnPropertyManager,
-                testCatalog.getConnectorId());
+                testCatalog.getConnectorId(),
+                emptySet());
     }
 
     @Test
     public void testCreateTableNotExistsTrue()
     {
         CreateTable statement = new CreateTable(QualifiedName.of("test_table"),
-                ImmutableList.of(new ColumnDefinition(identifier("a"), "BIGINT", emptyList(), Optional.empty())),
+                ImmutableList.of(new ColumnDefinition(identifier("a"), "BIGINT", true, emptyList(), Optional.empty())),
                 true,
                 ImmutableList.of(),
                 Optional.empty());
@@ -107,7 +119,7 @@ public void testCreateTableNotExistsTrue()
     public void testCreateTableNotExistsFalse()
     {
         CreateTable statement = new CreateTable(QualifiedName.of("test_table"),
-                ImmutableList.of(new ColumnDefinition(identifier("a"), "BIGINT", emptyList(), Optional.empty())),
+                ImmutableList.of(new ColumnDefinition(identifier("a"), "BIGINT", true, emptyList(), Optional.empty())),
                 false,
                 ImmutableList.of(),
                 Optional.empty());
@@ -125,31 +137,79 @@ public void testCreateTableNotExistsFalse()
         assertEquals(metadata.getCreateTableCallCount(), 1);
     }
 
+    @Test
+    public void testCreateWithNotNullColumns()
+    {
+        metadata.setConnectorCapabilities(NOT_NULL_COLUMN_CONSTRAINT);
+        List<TableElement> inputColumns = ImmutableList.of(
+                new ColumnDefinition(identifier("a"), "DATE", true, emptyList(), Optional.empty()),
+                new ColumnDefinition(identifier("b"), "VARCHAR", false, emptyList(), Optional.empty()),
+                new ColumnDefinition(identifier("c"), "VARBINARY", false, emptyList(), Optional.empty()));
+        CreateTable statement = new CreateTable(QualifiedName.of("test_table"), inputColumns, true, ImmutableList.of(), Optional.empty());
+
+        getFutureValue(new CreateTableTask().internalExecute(statement, metadata, new AllowAllAccessControl(), testSession, emptyList()));
+        assertEquals(metadata.getCreateTableCallCount(), 1);
+        List<ColumnMetadata> columns = metadata.getReceivedTableMetadata().get(0).getColumns();
+        assertEquals(columns.size(), 3);
+
+        assertEquals(columns.get(0).getName(), "a");
+        assertEquals(columns.get(0).getType().getDisplayName().toUpperCase(ENGLISH), "DATE");
+        assertTrue(columns.get(0).isNullable());
+
+        assertEquals(columns.get(1).getName(), "b");
+        assertEquals(columns.get(1).getType().getDisplayName().toUpperCase(ENGLISH), "VARCHAR");
+        assertFalse(columns.get(1).isNullable());
+
+        assertEquals(columns.get(2).getName(), "c");
+        assertEquals(columns.get(2).getType().getDisplayName().toUpperCase(ENGLISH), "VARBINARY");
+        assertFalse(columns.get(2).isNullable());
+    }
+
+    @Test(expectedExceptions = SemanticException.class, expectedExceptionsMessageRegExp = ".*does not support non-null column for column name 'b'")
+    public void testCreateWithUnsupportedConnectorThrowsWhenNotNull()
+    {
+        List<TableElement> inputColumns = ImmutableList.of(
+                new ColumnDefinition(identifier("a"), "DATE", true, emptyList(), Optional.empty()),
+                new ColumnDefinition(identifier("b"), "VARCHAR", false, emptyList(), Optional.empty()),
+                new ColumnDefinition(identifier("c"), "VARBINARY", false, emptyList(), Optional.empty()));
+        CreateTable statement = new CreateTable(
+                QualifiedName.of("test_table"),
+                inputColumns,
+                true,
+                ImmutableList.of(),
+                Optional.empty());
+
+        getFutureValue(new CreateTableTask().internalExecute(statement, metadata, new AllowAllAccessControl(), testSession, emptyList()));
+    }
+
     private static class MockMetadata
             extends AbstractMockMetadata
     {
         private final TypeManager typeManager;
         private final TablePropertyManager tablePropertyManager;
         private final ColumnPropertyManager columnPropertyManager;
         private final ConnectorId catalogHandle;
-        private AtomicInteger createTableCallCount = new AtomicInteger();
+        private final List<ConnectorTableMetadata> tables = new CopyOnWriteArrayList<>();
+        private Set<ConnectorCapabilities> connectorCapabilities;
 
         public MockMetadata(
                 TypeManager typeManager,
                 TablePropertyManager tablePropertyManager,
                 ColumnPropertyManager columnPropertyManager,
-                ConnectorId catalogHandle)
+                ConnectorId catalogHandle,
+                Set<ConnectorCapabilities> connectorCapabilities)
         {
             this.typeManager = requireNonNull(typeManager, "typeManager is null");
             this.tablePropertyManager = requireNonNull(tablePropertyManager, "tablePropertyManager is null");
             this.columnPropertyManager = requireNonNull(columnPropertyManager, "columnPropertyManager is null");
             this.catalogHandle = requireNonNull(catalogHandle, "catalogHandle is null");
+            this.connectorCapabilities = requireNonNull(immutableEnumSet(connectorCapabilities), "connectorCapabilities is null");
         }
 
         @Override
         public void createTable(Session session, String catalogName, ConnectorTableMetadata tableMetadata, boolean ignoreExisting)
         {
-            createTableCallCount.incrementAndGet();
+            tables.add(tableMetadata);
             if (!ignoreExisting) {
                 throw new PrestoException(ALREADY_EXISTS, "Table already exists");
             }
@@ -190,13 +250,29 @@ public Optional<TableHandle> getTableHandle(Session session, QualifiedObjectName
 
         public int getCreateTableCallCount()
         {
-            return createTableCallCount.get();
+            return tables.size();
+        }
+
+        public List<ConnectorTableMetadata> getReceivedTableMetadata()
+        {
+            return tables;
         }
 
         @Override
         public void dropColumn(Session session, TableHandle tableHandle, ColumnHandle column)
         {
             throw new UnsupportedOperationException();
         }
+
+        @Override
+        public Set<ConnectorCapabilities> getConnectorCapabilities(Session session, ConnectorId catalogName)
+        {
+            return connectorCapabilities;
+        }
+
+        public void setConnectorCapabilities(ConnectorCapabilities... connectorCapabilities)
+        {
+            this.connectorCapabilities = immutableEnumSet(ImmutableList.copyOf(connectorCapabilities));
+        }
     }
 }