Add validation to Insert statement parsing to match the column names with supplied values

docs/ReleaseNotes.md

@@ -22,8 +22,8 @@ Our API stability annotations have been updated to reflect greater API instabili
 * **Bug fix** Fix 1 [(Issue #NNN)](https://github.com/FoundationDB/fdb-record-layer/issues/NNN)
 * **Bug fix** Fix 2 [(Issue #NNN)](https://github.com/FoundationDB/fdb-record-layer/issues/NNN)
 * **Bug fix** Fix 3 [(Issue #NNN)](https://github.com/FoundationDB/fdb-record-layer/issues/NNN)
+* **Bug fix** Insert statement does not fully validate the column names with supplied valuesgo [(Issue #3069)](https://github.com/FoundationDB/fdb-record-layer/issues/3069)
 * **Bug fix** ungrouped GROUP BY queries result in infinite continuations when maxRows is 1 [(Issue #3093)](https://github.com/FoundationDB/fdb-record-layer/issues/3093)
-* **Bug fix** Fix 5 [(Issue #NNN)](https://github.com/FoundationDB/fdb-record-layer/issues/NNN)
 * **Performance** Improvement 1 [(Issue #NNN)](https://github.com/FoundationDB/fdb-record-layer/issues/NNN)
 * **Performance** Improvement 2 [(Issue #NNN)](https://github.com/FoundationDB/fdb-record-layer/issues/NNN)
 * **Performance** Improvement 3 [(Issue #NNN)](https://github.com/FoundationDB/fdb-record-layer/issues/NNN)

fdb-relational-core/src/main/java/com/apple/foundationdb/relational/recordlayer/query/Expression.java

@@ -21,34 +21,27 @@
 package com.apple.foundationdb.relational.recordlayer.query;
 
 import com.apple.foundationdb.annotation.API;
-
 import com.apple.foundationdb.record.query.plan.cascades.AliasMap;
 import com.apple.foundationdb.record.query.plan.cascades.Column;
 import com.apple.foundationdb.record.query.plan.cascades.CorrelationIdentifier;
 import com.apple.foundationdb.record.query.plan.cascades.predicates.QueryPredicate;
-import com.apple.foundationdb.record.query.plan.cascades.typing.Type;
-import com.apple.foundationdb.record.query.plan.cascades.values.AbstractArrayConstructorValue;
 import com.apple.foundationdb.record.query.plan.cascades.values.AggregateValue;
 import com.apple.foundationdb.record.query.plan.cascades.values.AndOrValue;
 import com.apple.foundationdb.record.query.plan.cascades.values.ArithmeticValue;
 import com.apple.foundationdb.record.query.plan.cascades.values.BooleanValue;
 import com.apple.foundationdb.record.query.plan.cascades.values.ConstantObjectValue;
 import com.apple.foundationdb.record.query.plan.cascades.values.LiteralValue;
 import com.apple.foundationdb.record.query.plan.cascades.values.NotValue;
-import com.apple.foundationdb.record.query.plan.cascades.values.NullValue;
 import com.apple.foundationdb.record.query.plan.cascades.values.RecordConstructorValue;
 import com.apple.foundationdb.record.query.plan.cascades.values.RelOpValue;
 import com.apple.foundationdb.record.query.plan.cascades.values.Value;
-import com.apple.foundationdb.relational.api.exceptions.ErrorCode;
 import com.apple.foundationdb.relational.api.metadata.DataType;
 import com.apple.foundationdb.relational.recordlayer.metadata.DataTypeUtils;
 import com.apple.foundationdb.relational.util.Assert;
-
 import com.google.common.base.Suppliers;
 import com.google.common.base.Verify;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.Streams;
-import com.google.protobuf.ByteString;
 
 import javax.annotation.Nonnull;
 import java.util.Collection;
@@ -305,53 +298,5 @@ public static QueryPredicate toUnderlyingPredicate(@Nonnull Expression expressio
                 return result.get();
             }
         }
-
-        @Nonnull
-        public static Expression resolveDefaultValue(@Nonnull final Type type) {
-            // TODO use metadata default values -- for now just do this:
-            //
-            // resolution rules:
-            // - type is nullable ==> null
-            // - type is not nullable
-            //   - type is of an array type ==> empty array
-            //   - type is a numeric type ==> 0 element
-            //   - type is string ==> ''
-            //   - type is a boolean ==> false
-            //   - type is bytes ==> zero length byte array
-            //   - type is record or enum ==> error
-            if (type.isNullable()) {
-                return Expression.fromUnderlying(new NullValue(type));
-            } else {
-                switch (type.getTypeCode()) {
-                    case UNKNOWN:
-                    case ANY:
-                    case NULL:
-                        throw Assert.failUnchecked("internal typing error; target type is not properly resolved");
-                    case BOOLEAN:
-                        return Expression.fromUnderlying(LiteralValue.ofScalar(false));
-                    case BYTES:
-                        return Expression.fromUnderlying(LiteralValue.ofScalar(ByteString.empty()));
-                    case DOUBLE:
-                        return Expression.fromUnderlying(LiteralValue.ofScalar(0.0d));
-                    case FLOAT:
-                        return Expression.fromUnderlying(LiteralValue.ofScalar(0.0f));
-                    case INT:
-                        return Expression.fromUnderlying(LiteralValue.ofScalar(0));
-                    case LONG:
-                        return Expression.fromUnderlying(LiteralValue.ofScalar(0L));
-                    case STRING:
-                        return Expression.fromUnderlying(LiteralValue.ofScalar(""));
-                    case ENUM:
-                        throw Assert.failUnchecked(ErrorCode.CANNOT_CONVERT_TYPE, "non-nullable enums must be specified");
-                    case RECORD:
-                        throw Assert.failUnchecked(ErrorCode.CANNOT_CONVERT_TYPE, "non-nullable records must be specified");
-                    case ARRAY:
-                        final var elementType = Assert.notNullUnchecked(((Type.Array) type).getElementType());
-                        return Expression.fromUnderlying(AbstractArrayConstructorValue.LightArrayConstructorValue.emptyArray(elementType));
-                    default:
-                        throw Assert.failUnchecked("unsupported type");
-                }
-            }
-        }
     }
 }

fdb-relational-core/src/main/java/com/apple/foundationdb/relational/recordlayer/query/visitors/DdlVisitor.java

@@ -118,6 +118,11 @@ public RecordLayerColumn visitColumnDefinition(@Nonnull RelationalParser.ColumnD
         final var columnId = visitUid(ctx.colName);
         final var isRepeated = ctx.ARRAY() != null;
         final var isNullable = ctx.columnConstraint() != null ? (Boolean) ctx.columnConstraint().accept(this) : true;
+        // TODO: We currently do not support NOT NULL for any type other than ARRAY. This is because there is no way to
+        //       specify not "nullability" at the RecordMetaData level. For ARRAY, specifying that is actually possible
+        //       by means of NullableArrayWrapper. In essence, we don't actually need a wrapper per se for non-array types,
+        //       but a way to represent it in RecordMetadata.
+        Assert.thatUnchecked(isRepeated || isNullable, ErrorCode.UNSUPPORTED_OPERATION, "NOT NULL is only allowed for ARRAY column type");
         containsNullableArray = containsNullableArray || (isRepeated && isNullable);
         final var columnTypeId = ctx.columnType().customType != null ? visitUid(ctx.columnType().customType) : Identifier.of(ctx.columnType().getText());
         final var semanticAnalyzer = getDelegate().getSemanticAnalyzer();

fdb-relational-core/src/main/java/com/apple/foundationdb/relational/recordlayer/query/visitors/ExpressionVisitor.java

@@ -764,15 +764,21 @@ private Expressions parseRecordFieldsUnderReorderings(@Nonnull final List<? exte
             final var targetTypeReorderings = ImmutableList.copyOf(Assert.notNullUnchecked(
                     state.getTargetTypeReorderings().get().getChildrenMap()).keySet());
             final var resultColumnsBuilder = ImmutableList.<Expression>builder();
-
+            Assert.thatUnchecked(targetTypeReorderings.size() >= providedColumnContexts.size(), ErrorCode.SYNTAX_ERROR, "Too many parameters");
             for (final var elementField : elementFields) {
                 final int index = targetTypeReorderings.indexOf(elementField.getFieldName());
                 final var fieldType = elementField.getFieldType();
-                final Expression currentFieldColumns;
-                if (index >= 0) {
+                Expression currentFieldColumns = null;
+                if (index >= 0 && index < providedColumnContexts.size()) {
                     currentFieldColumns = parseRecordField(providedColumnContexts.get(index), elementField);
+                } else if (index >= providedColumnContexts.size()) {
+                    // column is declared but the value is not provided
+                    Assert.failUnchecked(ErrorCode.SYNTAX_ERROR, "Value of column \"" + elementField.getFieldName() + "\" is not provided");
                 } else {
-                    currentFieldColumns = Expression.Utils.resolveDefaultValue(fieldType);
+                    // We do not yet support default values for any types, hence it makes to simply fail if the field type
+                    // expects non-null but no value is provided.
+                    Assert.thatUnchecked(fieldType.isNullable(), ErrorCode.NOT_NULL_VIOLATION, "null value in column \"" + elementField.getFieldName() + "\" violates not-null constraint");
+                    currentFieldColumns = Expression.fromUnderlying(new NullValue(fieldType));
                 }
                 resultColumnsBuilder.add(currentFieldColumns);
             }

fdb-relational-core/src/test/java/com/apple/foundationdb/relational/api/ddl/DdlStatementParsingTest.java

@@ -64,6 +64,7 @@
 import javax.annotation.Nullable;
 import java.net.URI;
 import java.sql.SQLException;
+import java.sql.Types;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
@@ -254,6 +255,75 @@ public ConstantAction getCreateSchemaTemplateConstantAction(@Nonnull SchemaTempl
         });
     }
 
+    private static Stream<Arguments> typesMap() {
+        return Stream.of(
+                Arguments.of(Types.INTEGER, "INTEGER"),
+                Arguments.of(Types.BIGINT, "BIGINT"),
+                Arguments.of(Types.FLOAT, "FLOAT"),
+                Arguments.of(Types.DOUBLE, "DOUBLE"),
+                Arguments.of(Types.VARCHAR, "STRING"),
+                Arguments.of(Types.BOOLEAN, "BOOLEAN"),
+                Arguments.of(Types.BINARY, "BYTES"),
+                Arguments.of(Types.STRUCT, "BAZ"),
+                Arguments.of(Types.ARRAY, "STRING ARRAY")
+        );
+    }
+
+    @ParameterizedTest
+    @MethodSource("typesMap")
+    void columnTypeWithNull(int sqlType, @Nonnull String sqlTypeName) throws Exception {
+        final String stmt = "CREATE SCHEMA TEMPLATE test_template " +
+                "CREATE TYPE AS STRUCT baz (a bigint, b bigint) " +
+                "CREATE TABLE bar (id bigint, foo_field " + sqlTypeName + " null, PRIMARY KEY(id))";
+        shouldWorkWithInjectedFactory(stmt, new AbstractMetadataOperationsFactory() {
+            @Nonnull
+            @Override
+            public ConstantAction getCreateSchemaTemplateConstantAction(@Nonnull final SchemaTemplate template, @Nonnull final Options templateProperties) {
+                checkColumnNullability(template, sqlType, true);
+                return txn -> {
+                };
+            }
+        });
+    }
+
+    @ParameterizedTest
+    @MethodSource("typesMap")
+    void columnTypeWithNotNull(int sqlType, @Nonnull String sqlTypeName) throws Exception {
+        final String stmt = "CREATE SCHEMA TEMPLATE test_template " +
+                "CREATE TYPE AS STRUCT baz (a bigint, b bigint) " +
+                "CREATE TABLE bar (id bigint, foo_field " + sqlTypeName + " not null, PRIMARY KEY(id))";
+        if (sqlType == Types.ARRAY) {
+            shouldWorkWithInjectedFactory(stmt, new AbstractMetadataOperationsFactory() {
+                @Nonnull
+                @Override
+                public ConstantAction getCreateSchemaTemplateConstantAction(@Nonnull final SchemaTemplate template, @Nonnull final Options templateProperties) {
+                    checkColumnNullability(template, sqlType, false);
+                    return txn -> {
+                    };
+                }
+            });
+        } else {
+            shouldFailWith(stmt, ErrorCode.UNSUPPORTED_OPERATION);
+        }
+    }
+
+    private static void checkColumnNullability(@Nonnull final SchemaTemplate template, int sqlType, boolean isNullable) {
+        Assertions.assertInstanceOf(RecordLayerSchemaTemplate.class, template);
+        Assertions.assertEquals(1, ((RecordLayerSchemaTemplate) template).getTables().size(), "should have only 1 table");
+        final var table = ((RecordLayerSchemaTemplate) template).findTableByName("BAR");
+        Assertions.assertTrue(table.isPresent());
+        final var columns = table.get().getColumns();
+        Assertions.assertEquals(2, columns.size());
+        final var maybeNullableArrayColumn = columns.stream().filter(c -> c.getName().equals("FOO_FIELD")).findFirst();
+        Assertions.assertTrue(maybeNullableArrayColumn.isPresent());
+        if (isNullable) {
+            Assertions.assertTrue(maybeNullableArrayColumn.get().getDatatype().isNullable());
+        } else {
+            Assertions.assertFalse(maybeNullableArrayColumn.get().getDatatype().isNullable());
+        }
+        Assertions.assertEquals(sqlType, maybeNullableArrayColumn.get().getDatatype().getJdbcSqlCode());
+    }
+
     @Test
     void failsToParseEmptyTemplateStatements() throws Exception {
         //empty template statements are invalid, and can be rejected in the parser

fdb-relational-core/src/test/java/com/apple/foundationdb/relational/recordlayer/InsertTest.java

@@ -59,7 +59,7 @@ public class InsertTest {
     public final SimpleDatabaseRule database = new SimpleDatabaseRule(relationalExtension, InsertTest.class, TestSchemas.restaurant());
 
     @Test
-    void canInsertWithMultipleRecordTypes() throws RelationalException, SQLException {
+    void canInsertWithMultipleRecordTypes() throws SQLException {
         /*
          * We want to make sure that we don't accidentally pick up data from different tables
          */
@@ -284,7 +284,7 @@ void canInsertNullableArray() throws SQLException, RelationalException {
     }
 
     @Test
-    void replaceOnInsert() throws SQLException, RelationalException {
+    void replaceOnInsert() throws SQLException {
         try (RelationalConnection conn = DriverManager.getConnection(database.getConnectionUri().toString()).unwrap(RelationalConnection.class)) {
             conn.setSchema("TEST_SCHEMA");
             try (RelationalStatement s = conn.createStatement()) {