Prevent invalid constraint combinations

patch by Stefan Miklosovic; reviewed by Bernardo Botella for CASSANDRA-20330

CHANGES.txt

@@ -1,4 +1,5 @@
 5.1
+ * Prevent invalid constraint combinations (CASSANDRA-20330)
  * Support CREATE TABLE LIKE WITH INDEXES (CASSANDRA-19965)
  * Invalidate relevant prepared statements on every change to TableMetadata (CASSANDRA-20318)
  * Add per type max size guardrails (CASSANDRA-19677)

src/java/org/apache/cassandra/cql3/constraints/AbstractFunctionConstraint.java

@@ -18,14 +18,52 @@
 
 package org.apache.cassandra.cql3.constraints;
 
+import java.util.Set;
+
 import org.apache.cassandra.cql3.ColumnIdentifier;
+import org.apache.cassandra.cql3.CqlBuilder;
+import org.apache.cassandra.cql3.Operator;
+import org.apache.cassandra.utils.LocalizeString;
 
 public abstract class AbstractFunctionConstraint<T> extends ColumnConstraint<T>
 {
-    public AbstractFunctionConstraint(ColumnIdentifier columnName)
+    protected final Operator relationType;
+    protected final String term;
+
+    public AbstractFunctionConstraint(ColumnIdentifier columnName, Operator relationType, String term)
     {
         super(columnName);
+        this.relationType = relationType;
+        this.term = term;
     }
 
-    public abstract String name();
+    public Operator relationType()
+    {
+        return relationType;
+    }
+
+    public String term()
+    {
+        return term;
+    }
+
+    public abstract Set<Operator> getSupportedOperators();
+
+    @Override
+    public void appendCqlTo(CqlBuilder builder)
+    {
+        builder.append(toString());
+    }
+
+    public static <T extends Enum<T>> T getEnum(Class<T> enumClass, String functionName)
+    {
+        try
+        {
+            return Enum.valueOf(enumClass, LocalizeString.toUpperCaseLocalized(functionName));
+        }
+        catch (IllegalArgumentException e)
+        {
+            throw new InvalidConstraintDefinitionException("Unrecognized constraint function: " + functionName);
+        }
+    }
 }

src/java/org/apache/cassandra/cql3/constraints/AbstractFunctionSatisfiabilityChecker.java

@@ -0,0 +1,218 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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.apache.cassandra.cql3.constraints;
+
+import java.nio.ByteBuffer;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Set;
+import java.util.TreeSet;
+
+import org.apache.cassandra.cql3.Operator;
+import org.apache.cassandra.schema.ColumnMetadata;
+import org.apache.cassandra.utils.Pair;
+
+import static java.lang.String.format;
+import static org.apache.cassandra.cql3.Operator.EQ;
+import static org.apache.cassandra.cql3.Operator.GT;
+import static org.apache.cassandra.cql3.Operator.GTE;
+import static org.apache.cassandra.cql3.Operator.LT;
+import static org.apache.cassandra.cql3.Operator.LTE;
+import static org.apache.cassandra.cql3.Operator.NEQ;
+import static org.apache.cassandra.cql3.constraints.ColumnConstraint.ConstraintType.FUNCTION;
+
+public abstract class AbstractFunctionSatisfiabilityChecker<CONSTRAINT_TYPE extends AbstractFunctionConstraint<CONSTRAINT_TYPE>>
+{
+    /**
+     * Performs check if constraints are satisfiable or not.
+     *
+     * @param functionName   name of function
+     * @param constraints    list of constraints to set
+     * @param columnMetadata metadata of a column.
+     */
+    public void check(String functionName, List<ColumnConstraint<?>> constraints, ColumnMetadata columnMetadata)
+    {
+        Pair<List<CONSTRAINT_TYPE>, List<CONSTRAINT_TYPE>> filteredConstraints = categorizeConstraints(constraints, functionName);
+
+        if (filteredConstraints.left.isEmpty())
+            return;
+
+        checkNumberOfConstraints(columnMetadata, filteredConstraints);
+        checkSupportedOperators(filteredConstraints.left, functionName);
+        ensureSatisfiability(columnMetadata, functionName, filteredConstraints.left);
+    }
+
+    /**
+     * Categorizes given constraints into two lists. The first list, the left one in Pair, contains all
+     * constraints of implementation-specific {@link org.apache.cassandra.cql3.constraints.ColumnConstraint.ConstraintType}.
+     * The second list, the right one in Pair, contains all constraints of such constraint type which do have "not equal" operator.
+     *
+     * @param constraints  constraints to categorize
+     * @param functionName name of function
+     * @return pair of categorized constraints
+     */
+    abstract Pair<List<CONSTRAINT_TYPE>, List<CONSTRAINT_TYPE>> categorizeConstraints(List<ColumnConstraint<?>> constraints, String functionName);
+
+    private void checkSupportedOperators(List<CONSTRAINT_TYPE> allConstraints, String functionName)
+    {
+        for (CONSTRAINT_TYPE constraint : allConstraints)
+        {
+            if (!constraint.getSupportedOperators().contains(constraint.relationType()))
+                throw new InvalidConstraintDefinitionException(format("%s constraint of relation '%s' is not supported. Only these are: %s",
+                                                                      functionName,
+                                                                      constraint.relationType(),
+                                                                      constraint.getSupportedOperators()));
+        }
+    }
+
+    /**
+     * Checks if there are no duplicate constraints having same operator.
+     *
+     * @param columnMetadata      medata of a column
+     * @param filteredConstraints pair of all constraints and all constraints having not-equal operator
+     */
+    private void checkNumberOfConstraints(ColumnMetadata columnMetadata, Pair<List<CONSTRAINT_TYPE>, List<CONSTRAINT_TYPE>> filteredConstraints)
+    {
+        List<? extends AbstractFunctionConstraint<CONSTRAINT_TYPE>> allConstraints = filteredConstraints.left;
+        List<? extends AbstractFunctionConstraint<CONSTRAINT_TYPE>> notEqualConstraints = filteredConstraints.right;
+
+        if ((allConstraints.size() - notEqualConstraints.size() > 2))
+        {
+            throw new InvalidConstraintDefinitionException(format("There can not be more than 2 constraints (not including non-equal relations) on a column '%s' but you have specified %s",
+                                                                  columnMetadata.name,
+                                                                  allConstraints.size()));
+        }
+
+        if (notEqualConstraints.size() > 1)
+        {
+            Set<String> uniqueTerms = new TreeSet<>();
+            for (AbstractFunctionConstraint<CONSTRAINT_TYPE> notEqual : notEqualConstraints)
+            {
+                if (!uniqueTerms.add(notEqual.term()))
+                    throw new InvalidConstraintDefinitionException(format("There are duplicate constraint definitions on column '%s': %s",
+                                                                          columnMetadata.name,
+                                                                          notEqual));
+            }
+        }
+    }
+
+    private void ensureSatisfiability(ColumnMetadata columnMetadata,
+                                      String constraintName,
+                                      List<CONSTRAINT_TYPE> allConstraints)
+    {
+        if (allConstraints.size() != 2)
+            return;
+
+        Operator firstRelation = allConstraints.get(0).relationType();
+        String firstTerm = allConstraints.get(0).term();
+        Operator secondRelation = allConstraints.get(1).relationType();
+        String secondTerm = allConstraints.get(1).term();
+
+        if ((firstRelation == GT && secondRelation == GTE) ||
+            (firstRelation == GTE && secondRelation == GT) ||
+            (firstRelation == LT && secondRelation == LTE) ||
+            (firstRelation == LTE && secondRelation == LT) ||
+            (firstRelation == EQ || secondRelation == EQ))
+        {
+            throw new InvalidConstraintDefinitionException(format("Constraints combination of %s is not supported: %s %s %s, %s %s %s",
+                                                                  constraintName,
+                                                                  columnMetadata.name,
+                                                                  firstRelation,
+                                                                  firstTerm,
+                                                                  columnMetadata.name,
+                                                                  secondRelation,
+                                                                  secondTerm));
+        }
+        else if (firstRelation == NEQ && secondRelation == NEQ)
+        {
+            if (firstTerm.equals(secondTerm))
+                throw new InvalidConstraintDefinitionException(format("There are duplicate constraint definitions on column '%s'.", columnMetadata.name));
+        }
+        else
+        {
+            ByteBuffer firstTermBuffer = columnMetadata.type.fromString(firstTerm);
+            ByteBuffer secondTermBuffer = columnMetadata.type.fromString(secondTerm);
+
+            boolean firstSatisfaction = firstRelation.isSatisfiedBy(columnMetadata.type, secondTermBuffer, firstTermBuffer);
+            boolean secondSatisfaction = secondRelation.isSatisfiedBy(columnMetadata.type, firstTermBuffer, secondTermBuffer);
+
+            if (!firstSatisfaction || !secondSatisfaction)
+                throw new InvalidConstraintDefinitionException(format("Constraints of %s are not satisfiable: %s %s %s, %s %s %s",
+                                                                      constraintName,
+                                                                      columnMetadata.name,
+                                                                      firstRelation,
+                                                                      firstTerm,
+                                                                      columnMetadata.name,
+                                                                      secondRelation,
+                                                                      secondTerm));
+        }
+    }
+
+    public static final AbstractFunctionSatisfiabilityChecker<ScalarColumnConstraint> SCALAR_SATISFIABILITY_CHECKER = new AbstractFunctionSatisfiabilityChecker<>()
+    {
+        @Override
+        public Pair<List<ScalarColumnConstraint>, List<ScalarColumnConstraint>> categorizeConstraints(List<ColumnConstraint<?>> constraints, String functionName)
+        {
+            List<ScalarColumnConstraint> scalars = new LinkedList<>();
+            List<ScalarColumnConstraint> notEqualScalars = new LinkedList<>();
+
+            for (ColumnConstraint<?> columnConstraint : constraints)
+            {
+                if (columnConstraint.getConstraintType() == ColumnConstraint.ConstraintType.SCALAR)
+                {
+                    ScalarColumnConstraint scalarColumnConstraint = (ScalarColumnConstraint) columnConstraint;
+                    scalars.add(scalarColumnConstraint);
+                    if (scalarColumnConstraint.relationType() == NEQ)
+                        notEqualScalars.add(scalarColumnConstraint);
+                }
+            }
+
+            return Pair.create(scalars, notEqualScalars);
+        }
+    };
+
+    public static final AbstractFunctionSatisfiabilityChecker<FunctionColumnConstraint> FUNCTION_SATISFIABILITY_CHECKER = new AbstractFunctionSatisfiabilityChecker<>()
+    {
+        @Override
+        public Pair<List<FunctionColumnConstraint>, List<FunctionColumnConstraint>> categorizeConstraints(List<ColumnConstraint<?>> constraints, String functionName)
+        {
+            List<FunctionColumnConstraint> funnctionColumnConstraints = new LinkedList<>();
+            List<FunctionColumnConstraint> notEqualConstraints = new LinkedList<>();
+
+            for (ColumnConstraint<?> columnConstraint : constraints)
+            {
+                if (columnConstraint.getConstraintType() != FUNCTION)
+                    continue;
+
+                FunctionColumnConstraint functionColumnConstraint = (FunctionColumnConstraint) columnConstraint;
+
+                ConstraintFunction function = functionColumnConstraint.function();
+
+                if (!function.name.equals(functionName))
+                    continue;
+
+                funnctionColumnConstraints.add(functionColumnConstraint);
+                if (functionColumnConstraint.relationType() == NEQ)
+                    notEqualConstraints.add(functionColumnConstraint);
+            }
+
+            return Pair.create(funnctionColumnConstraints, notEqualConstraints);
+        }
+    };
+}

src/java/org/apache/cassandra/cql3/constraints/ColumnConstraint.java

@@ -19,12 +19,17 @@
 package org.apache.cassandra.cql3.constraints;
 
 import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
 
 import org.apache.cassandra.cql3.ColumnIdentifier;
 import org.apache.cassandra.cql3.CqlBuilder;
+import org.apache.cassandra.cql3.constraints.ColumnConstraints.DuplicatesChecker;
+import org.apache.cassandra.cql3.constraints.ScalarColumnConstraint.ScalarColumnConstraintSatisfiabilityChecker;
 import org.apache.cassandra.db.marshal.AbstractType;
-import org.apache.cassandra.tcm.serialization.MetadataSerializer;
 import org.apache.cassandra.schema.ColumnMetadata;
+import org.apache.cassandra.tcm.serialization.MetadataSerializer;
 
 /**
  * Common class for the conditions that a CQL Constraint needs to implement to be integrated in the
@@ -46,32 +51,65 @@ public abstract class ColumnConstraint<T>
         // The order of that enum matters!!
         // We are serializing its enum position instead of its name.
         // Changing this enum would affect how that int is interpreted when deserializing.
-        COMPOSED(ColumnConstraints.serializer),
-        FUNCTION(FunctionColumnConstraint.serializer),
-        SCALAR(ScalarColumnConstraint.serializer),
-        UNARY_FUNCTION(UnaryFunctionColumnConstraint.serializer);
+        COMPOSED(ColumnConstraints.serializer, new DuplicatesChecker()),
+        FUNCTION(FunctionColumnConstraint.serializer, FunctionColumnConstraint.Functions.values()),
+        SCALAR(ScalarColumnConstraint.serializer, new ScalarColumnConstraintSatisfiabilityChecker()),
+        UNARY_FUNCTION(UnaryFunctionColumnConstraint.serializer, UnaryFunctionColumnConstraint.Functions.values());
 
         private final MetadataSerializer<?> serializer;
+        private final SatisfiabilityChecker[] satisfiabilityCheckers;
 
-        ConstraintType(MetadataSerializer<?> serializer)
+        ConstraintType(MetadataSerializer<?> serializer, SatisfiabilityChecker satisfiabilityChecker)
+        {
+            this(serializer, new SatisfiabilityChecker[]{ satisfiabilityChecker });
+        }
+
+        ConstraintType(MetadataSerializer<?> serializer, SatisfiabilityChecker[] satisfiabilityCheckers)
         {
             this.serializer = serializer;
+            this.satisfiabilityCheckers = satisfiabilityCheckers;
         }
 
         public static MetadataSerializer<?> getSerializer(int i)
         {
             return ConstraintType.values()[i].serializer;
         }
+
+        public static SatisfiabilityChecker[] getSatisfiabilityCheckers()
+        {
+            List<SatisfiabilityChecker> result = new ArrayList<>();
+            for (ConstraintType constraintType : ConstraintType.values())
+                result.addAll(Arrays.asList(constraintType.satisfiabilityCheckers));
+
+            return result.toArray(new SatisfiabilityChecker[0]);
+        }
+    }
+
+    public abstract String name();
+
+    /**
+     * Typically includes name of a constraint as in {@link #name()},
+     * plus an operator of a function, if constraint is a function.
+     * Full name serves as String which uniquely distinguishes two constraints even of same names for the purpose
+     * of checking if there is a specific constraint used twice. A duplicit usage of a constraint is illegal.
+     *
+     * @return full name of a constraint, with an operator.
+     */
+    public String fullName()
+    {
+        return name();
     }
 
     public abstract MetadataSerializer<T> serializer();
 
     public abstract void appendCqlTo(CqlBuilder builder);
 
+    public abstract boolean enablesDuplicateDefinitions(String name);
+
     /**
      * Method that evaluates the condition. It can either succeed or throw a {@link ConstraintViolationException}.
      *
-     * @param valueType value type of the column value under test
+     * @param valueType   value type of the column value under test
      * @param columnValue Column value to be evaluated at write time
      */
     public void evaluate(AbstractType<?> valueType, ByteBuffer columnValue) throws ConstraintViolationException

src/java/org/apache/cassandra/cql3/constraints/ColumnConstraints.java

@@ -24,8 +24,9 @@ import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
 import java.util.Objects;
+import java.util.Set;
+import java.util.TreeSet;
 
-import org.apache.cassandra.cql3.ColumnIdentifier;
 import org.apache.cassandra.cql3.CqlBuilder;
 import org.apache.cassandra.db.TypeSizes;
 import org.apache.cassandra.db.marshal.AbstractType;
@@ -35,6 +36,8 @@ import org.apache.cassandra.schema.ColumnMetadata;
 import org.apache.cassandra.tcm.serialization.MetadataSerializer;
 import org.apache.cassandra.tcm.serialization.Version;
 
+import static java.lang.String.format;
+
 // group of constraints for the column
 public class ColumnConstraints extends ColumnConstraint<ColumnConstraints>
 {
@@ -49,6 +52,12 @@ public class ColumnConstraints extends ColumnConstraint<ColumnConstraints>
         this.constraints = constraints;
     }
 
+    @Override
+    public String name()
+    {
+        return getConstraintType().name();
+    }
+
     @Override
     public MetadataSerializer<ColumnConstraints> serializer()
     {
@@ -62,6 +71,12 @@ public class ColumnConstraints extends ColumnConstraint<ColumnConstraints>
             constraint.appendCqlTo(builder);
     }
 
+    @Override
+    public boolean enablesDuplicateDefinitions(String name)
+    {
+        return false;
+    }
+
     @Override
     public void evaluate(AbstractType<?> valueType, ByteBuffer columnValue) throws ConstraintViolationException
     {
@@ -101,11 +116,6 @@ public class ColumnConstraints extends ColumnConstraint<ColumnConstraints>
         return false;
     }
 
-    public void checkInvalidConstraintsCombinations(ColumnIdentifier columnName)
-    {
-        // TODO check duplicities etc CASSANDRA-20330
-    }
-
     @Override
     public void validate(ColumnMetadata columnMetadata) throws InvalidConstraintDefinitionException
     {
@@ -114,6 +124,12 @@ public class ColumnConstraints extends ColumnConstraint<ColumnConstraints>
                                                            + columnMetadata.name + " of type " + columnMetadata.type.asCQL3Type()
                                                            + " for the table " + columnMetadata.ksName + "." + columnMetadata.cfName);
 
+        // this will look at constraints as a whole,
+        // checking if combinations of a particular constraint make sense (duplicities, satisfiability etc.).
+        for (SatisfiabilityChecker satisfiabilityChecker : ConstraintType.getSatisfiabilityCheckers())
+            satisfiabilityChecker.checkSatisfiability(constraints, columnMetadata);
+
+        // this validation will check whether it makes sense to execute such constraint on a given column
         for (ColumnConstraint<?> constraint : constraints)
             constraint.validate(columnMetadata);
     }
@@ -124,6 +140,33 @@ public class ColumnConstraints extends ColumnConstraint<ColumnConstraints>
         return ConstraintType.COMPOSED;
     }
 
+    public static class DuplicatesChecker implements SatisfiabilityChecker
+    {
+        @Override
+        public void checkSatisfiability(List<ColumnConstraint<?>> constraints, ColumnMetadata columnMetadata)
+        {
+            Set<String> constraintNames = new TreeSet<>();
+            List<String> duplicateConstraints = new ArrayList<>();
+
+            for (ColumnConstraint<?> constraint : constraints)
+            {
+                String constraintFullName = constraint.fullName();
+                String constraintName = constraint.name();
+
+                if (!constraintNames.add(constraintFullName))
+                {
+                    if (!constraint.enablesDuplicateDefinitions(constraintName))
+                        duplicateConstraints.add(constraintFullName);
+                }
+            }
+
+            if (!duplicateConstraints.isEmpty())
+                throw new InvalidConstraintDefinitionException(format("There are duplicate constraint definitions on column '%s': %s",
+                                                                      columnMetadata.name,
+                                                                      duplicateConstraints));
+        }
+    }
+
     private static class Noop extends ColumnConstraints
     {
         private Noop()
@@ -136,6 +179,12 @@ public class ColumnConstraints extends ColumnConstraint<ColumnConstraints>
         {
             // Do nothing. It is always valid
         }
+
+        @Override
+        public String name()
+        {
+            return "NO_OP";
+        }
     }
 
     public final static class Raw

src/java/org/apache/cassandra/cql3/constraints/ConstraintFunction.java

@@ -19,17 +19,27 @@
 package org.apache.cassandra.cql3.constraints;
 
 import java.nio.ByteBuffer;
+import java.util.Set;
 
 import org.apache.cassandra.cql3.ColumnIdentifier;
 import org.apache.cassandra.cql3.Operator;
 import org.apache.cassandra.db.marshal.AbstractType;
 import org.apache.cassandra.schema.ColumnMetadata;
 
+import static org.apache.cassandra.cql3.Operator.EQ;
+import static org.apache.cassandra.cql3.Operator.GT;
+import static org.apache.cassandra.cql3.Operator.GTE;
+import static org.apache.cassandra.cql3.Operator.LT;
+import static org.apache.cassandra.cql3.Operator.LTE;
+import static org.apache.cassandra.cql3.Operator.NEQ;
+
 /**
  * Interface to be implemented by functions that are executed as part of CQL constraints.
  */
 public abstract class ConstraintFunction
 {
+    public static final Set<Operator> DEFAULT_FUNCTION_OPERATORS = Set.of(EQ, NEQ, GTE, GT, LTE, LT);
+
     protected final ColumnIdentifier columnName;
     protected final String name;
 
@@ -70,4 +80,14 @@ public abstract class ConstraintFunction
      * if the CQL statement is valid or needs to be rejected as invalid throwing a {@link InvalidConstraintDefinitionException}
      */
     public abstract void validate(ColumnMetadata columnMetadata) throws InvalidConstraintDefinitionException;
+
+    /**
+     * Return operators this function supports. By default, it returns an empty set, modelling unary function.
+     *
+     * @return set of operators this function is allowed to have.
+     */
+    public Set<Operator> getSupportedOperators()
+    {
+        return Set.of();
+    }
 }

src/java/org/apache/cassandra/cql3/constraints/FunctionColumnConstraint.java

@@ -20,10 +20,11 @@ package org.apache.cassandra.cql3.constraints;
 
 import java.io.IOException;
 import java.nio.ByteBuffer;
+import java.util.List;
+import java.util.Set;
 import java.util.function.Function;
 
 import org.apache.cassandra.cql3.ColumnIdentifier;
-import org.apache.cassandra.cql3.CqlBuilder;
 import org.apache.cassandra.cql3.Operator;
 import org.apache.cassandra.db.TypeSizes;
 import org.apache.cassandra.db.marshal.AbstractType;
@@ -32,15 +33,15 @@ import org.apache.cassandra.io.util.DataOutputPlus;
 import org.apache.cassandra.schema.ColumnMetadata;
 import org.apache.cassandra.tcm.serialization.MetadataSerializer;
 import org.apache.cassandra.tcm.serialization.Version;
-import org.apache.cassandra.utils.LocalizeString;
+
+import static org.apache.cassandra.cql3.Operator.NEQ;
+import static org.apache.cassandra.cql3.constraints.AbstractFunctionSatisfiabilityChecker.FUNCTION_SATISFIABILITY_CHECKER;
 
 public class FunctionColumnConstraint extends AbstractFunctionConstraint<FunctionColumnConstraint>
 {
     public static final Serializer serializer = new Serializer();
 
     private final ConstraintFunction function;
-    private final Operator relationType;
-    private final String term;
 
     public final static class Raw
     {
@@ -63,9 +64,16 @@ public class FunctionColumnConstraint extends AbstractFunctionConstraint<Functio
         }
     }
 
-    private enum Functions
+    public enum Functions implements SatisfiabilityChecker
     {
-        LENGTH(LengthConstraint::new);
+        LENGTH(LengthConstraint::new)
+        {
+            @Override
+            public void checkSatisfiability(List<ColumnConstraint<?>> constraints, ColumnMetadata columnMetadata)
+            {
+                FUNCTION_SATISFIABILITY_CHECKER.check(name(), constraints, columnMetadata);
+            }
+        };
 
         private final Function<ColumnIdentifier, ConstraintFunction> functionCreator;
 
@@ -77,33 +85,42 @@ public class FunctionColumnConstraint extends AbstractFunctionConstraint<Functio
 
     private static ConstraintFunction createConstraintFunction(String functionName, ColumnIdentifier columnName)
     {
-        try
-        {
-            return Functions.valueOf(LocalizeString.toUpperCaseLocalized(functionName)).functionCreator.apply(columnName);
-        }
-        catch (IllegalArgumentException ex)
-        {
-            throw new InvalidConstraintDefinitionException("Unrecognized constraint function: " + functionName);
-        }
+        return getEnum(Functions.class, functionName).functionCreator.apply(columnName);
     }
 
     private FunctionColumnConstraint(ConstraintFunction function, ColumnIdentifier columnName, Operator relationType, String term)
     {
-        super(columnName);
+        super(columnName, relationType, term);
         this.function = function;
-        this.relationType = relationType;
-        this.term = term;
     }
 
+    public ConstraintFunction function()
+    {
+        return function;
+    }
+
+    @Override
+    public Set<Operator> getSupportedOperators()
+    {
+        return function.getSupportedOperators();
+    }
+
+    @Override
     public String name()
     {
         return function.name;
     }
 
     @Override
-    public void appendCqlTo(CqlBuilder builder)
+    public String fullName()
     {
-        builder.append(toString());
+        return function.name + ' ' + relationType;
+    }
+
+    @Override
+    public boolean enablesDuplicateDefinitions(String name)
+    {
+        return relationType == NEQ;
     }
 
     @Override

src/java/org/apache/cassandra/cql3/constraints/LengthConstraint.java

@@ -19,6 +19,7 @@
 package org.apache.cassandra.cql3.constraints;
 
 import java.nio.ByteBuffer;
+import java.util.Set;
 
 import org.apache.cassandra.cql3.ColumnIdentifier;
 import org.apache.cassandra.cql3.Operator;
@@ -32,11 +33,12 @@ import org.apache.cassandra.utils.ByteBufferUtil;
 
 public class LengthConstraint extends ConstraintFunction
 {
-    private static final AbstractType<?>[] SUPPORTED_TYPES = new AbstractType[] { BytesType.instance, UTF8Type.instance, AsciiType.instance };
+    private static final String NAME = "LENGTH";
+    private static final AbstractType<?>[] SUPPORTED_TYPES = new AbstractType[]{ BytesType.instance, UTF8Type.instance, AsciiType.instance };
 
     public LengthConstraint(ColumnIdentifier columnName)
     {
-        super(columnName, "LENGTH");
+        super(columnName, NAME);
     }
 
     @Override
@@ -50,7 +52,7 @@ public class LengthConstraint extends ConstraintFunction
 
         if (!relationType.isSatisfiedBy(Int32Type.instance, leftOperand, rightOperand))
             throw new ConstraintViolationException("Column value does not satisfy value constraint for column '" + columnName + "'. "
-                                                   + "It has a length of " + valueLength + " and it should be should be "
+                                                   + "It has a length of " + valueLength + " and it should be "
                                                    + relationType + ' ' + term);
     }
 
@@ -72,6 +74,12 @@ public class LengthConstraint extends ConstraintFunction
             throw invalidConstraintDefinitionException(columnMetadata.type);
     }
 
+    @Override
+    public Set<Operator> getSupportedOperators()
+    {
+        return DEFAULT_FUNCTION_OPERATORS;
+    }
+
     private int getValueLength(ByteBuffer value, AbstractType<?> valueType)
     {
         if (valueType.getClass() == BytesType.class)

src/java/org/apache/cassandra/cql3/constraints/SatisfiabilityChecker.java

@@ -0,0 +1,45 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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.apache.cassandra.cql3.constraints;
+
+import java.util.List;
+
+import org.apache.cassandra.schema.ColumnMetadata;
+
+public interface SatisfiabilityChecker
+{
+    void checkSatisfiability(List<ColumnConstraint<?>> constraints, ColumnMetadata columnMetadata);
+
+    default boolean enableDuplicateDefinitions()
+    {
+        return true;
+    }
+
+    interface UnaryFunctionSatisfiabilityChecker extends SatisfiabilityChecker
+    {
+        default void checkSatisfiability(List<ColumnConstraint<?>> constraints, ColumnMetadata columnMetadata)
+        {
+        }
+
+        default boolean enableDuplicateDefinitions()
+        {
+            return false;
+        }
+    }
+}

src/java/org/apache/cassandra/cql3/constraints/ScalarColumnConstraint.java

@@ -20,9 +20,10 @@ package org.apache.cassandra.cql3.constraints;
 
 import java.io.IOException;
 import java.nio.ByteBuffer;
+import java.util.List;
+import java.util.Set;
 
 import org.apache.cassandra.cql3.ColumnIdentifier;
-import org.apache.cassandra.cql3.CqlBuilder;
 import org.apache.cassandra.cql3.Operator;
 import org.apache.cassandra.db.TypeSizes;
 import org.apache.cassandra.db.marshal.AbstractType;
@@ -32,12 +33,19 @@ import org.apache.cassandra.schema.ColumnMetadata;
 import org.apache.cassandra.tcm.serialization.MetadataSerializer;
 import org.apache.cassandra.tcm.serialization.Version;
 
-public class ScalarColumnConstraint extends ColumnConstraint<ScalarColumnConstraint>
-{
-    public final static Serializer serializer = new Serializer();
+import static org.apache.cassandra.cql3.Operator.EQ;
+import static org.apache.cassandra.cql3.Operator.GT;
+import static org.apache.cassandra.cql3.Operator.GTE;
+import static org.apache.cassandra.cql3.Operator.LT;
+import static org.apache.cassandra.cql3.Operator.LTE;
+import static org.apache.cassandra.cql3.Operator.NEQ;
+import static org.apache.cassandra.cql3.constraints.AbstractFunctionSatisfiabilityChecker.SCALAR_SATISFIABILITY_CHECKER;
 
-    private final Operator relationType;
-    private final String term;
+public class ScalarColumnConstraint extends AbstractFunctionConstraint<ScalarColumnConstraint>
+{
+    public static final Set<Operator> SUPPORTED_OPERATORS = Set.of(EQ, NEQ, GTE, GT, LTE, LT);
+
+    public static final Serializer serializer = new Serializer();
 
     public final static class Raw
     {
@@ -58,13 +66,25 @@ public class ScalarColumnConstraint extends ColumnConstraint<ScalarColumnConstra
         }
     }
 
-    private ScalarColumnConstraint(ColumnIdentifier param, Operator relationType, String term)
+    public static class ScalarColumnConstraintSatisfiabilityChecker implements SatisfiabilityChecker
     {
-        super(param);
-        this.relationType = relationType;
-        this.term = term;
+        @Override
+        public void checkSatisfiability(List<ColumnConstraint<?>> constraints, ColumnMetadata columnMetadata)
+        {
+            SCALAR_SATISFIABILITY_CHECKER.check("scalar", constraints, columnMetadata);
+        }
     }
 
+    private ScalarColumnConstraint(ColumnIdentifier param, Operator relationType, String term)
+    {
+        super(param, relationType, term);
+    }
+
+    @Override
+    public Set<Operator> getSupportedOperators()
+    {
+        return SUPPORTED_OPERATORS;
+    }
 
     @Override
     protected void internalEvaluate(AbstractType<?> valueType, ByteBuffer columnValue)
@@ -103,6 +123,12 @@ public class ScalarColumnConstraint extends ColumnConstraint<ScalarColumnConstra
         return columnName + " " + relationType + " " + term;
     }
 
+    @Override
+    public String name()
+    {
+        return columnName + " " + relationType;
+    }
+
     @Override
     public MetadataSerializer<ScalarColumnConstraint> serializer()
     {
@@ -110,9 +136,9 @@ public class ScalarColumnConstraint extends ColumnConstraint<ScalarColumnConstra
     }
 
     @Override
-    public void appendCqlTo(CqlBuilder builder)
+    public boolean enablesDuplicateDefinitions(String name)
     {
-        builder.append(toString());
+        return relationType == NEQ;
     }
 
     private static class Serializer implements MetadataSerializer<ScalarColumnConstraint>

src/java/org/apache/cassandra/cql3/constraints/UnaryFunctionColumnConstraint.java

@@ -20,10 +20,12 @@ package org.apache.cassandra.cql3.constraints;
 
 import java.io.IOException;
 import java.nio.ByteBuffer;
+import java.util.Set;
 import java.util.function.Function;
 
 import org.apache.cassandra.cql3.ColumnIdentifier;
-import org.apache.cassandra.cql3.CqlBuilder;
+import org.apache.cassandra.cql3.Operator;
+import org.apache.cassandra.cql3.constraints.SatisfiabilityChecker.UnaryFunctionSatisfiabilityChecker;
 import org.apache.cassandra.db.TypeSizes;
 import org.apache.cassandra.db.marshal.AbstractType;
 import org.apache.cassandra.io.util.DataInputPlus;
@@ -31,7 +33,6 @@ import org.apache.cassandra.io.util.DataOutputPlus;
 import org.apache.cassandra.schema.ColumnMetadata;
 import org.apache.cassandra.tcm.serialization.MetadataSerializer;
 import org.apache.cassandra.tcm.serialization.Version;
-import org.apache.cassandra.utils.LocalizeString;
 
 import static org.apache.cassandra.cql3.constraints.ColumnConstraint.ConstraintType.UNARY_FUNCTION;
 
@@ -58,7 +59,7 @@ public class UnaryFunctionColumnConstraint extends AbstractFunctionConstraint<Un
         }
     }
 
-    private enum Functions
+    public enum Functions implements UnaryFunctionSatisfiabilityChecker
     {
         NOT_NULL(NotNullConstraint::new);
 
@@ -72,22 +73,16 @@ public class UnaryFunctionColumnConstraint extends AbstractFunctionConstraint<Un
 
     private static ConstraintFunction createConstraintFunction(String functionName, ColumnIdentifier columnName)
     {
-        try
-        {
-            return Functions.valueOf(LocalizeString.toUpperCaseLocalized(functionName)).functionCreator.apply(columnName);
-        }
-        catch (IllegalArgumentException ex)
-        {
-            throw new InvalidConstraintDefinitionException("Unrecognized constraint function: " + functionName);
-        }
+        return getEnum(Functions.class, functionName).functionCreator.apply(columnName);
     }
 
     private UnaryFunctionColumnConstraint(ConstraintFunction function, ColumnIdentifier columnName)
     {
-        super(columnName);
+        super(columnName, null, null);
         this.function = function;
     }
 
+    @Override
     public String name()
     {
         return function.name;
@@ -100,9 +95,15 @@ public class UnaryFunctionColumnConstraint extends AbstractFunctionConstraint<Un
     }
 
     @Override
-    public void appendCqlTo(CqlBuilder builder)
+    public Set<Operator> getSupportedOperators()
     {
-        builder.append(toString());
+        return Set.of();
+    }
+
+    @Override
+    public boolean enablesDuplicateDefinitions(String name)
+    {
+        return Functions.valueOf(name).enableDuplicateDefinitions();
     }
 
     @Override

src/java/org/apache/cassandra/cql3/statements/schema/CreateTableStatement.java

@@ -346,16 +346,18 @@ public final class CreateTableStatement extends AlterSchemaStatement
 
         for (int i = 0; i < partitionKeyColumns.size(); i++)
         {
-            ColumnConstraints constraints = validateConstraints(partitionKeyColumns.get(i));
             ColumnProperties properties = partitionKeyColumnProperties.get(i);
-            builder.addPartitionKeyColumn(partitionKeyColumns.get(i), properties.type, properties.mask, constraints);
+            ColumnIdentifier columnIdentifier = partitionKeyColumns.get(i);
+            builder.addPartitionKeyColumn(columnIdentifier, properties.type, properties.mask);
+            builder.getColumn(columnIdentifier).setColumnConstraints(columnConstraints.get(columnIdentifier));
         }
 
         for (int i = 0; i < clusteringColumns.size(); i++)
         {
-            ColumnConstraints constraints = validateConstraints(clusteringColumns.get(i));
             ColumnProperties properties = clusteringColumnProperties.get(i);
-            builder.addClusteringColumn(clusteringColumns.get(i), properties.type, properties.mask, constraints);
+            ColumnIdentifier columnIdentifier = clusteringColumns.get(i);
+            builder.addClusteringColumn(columnIdentifier, properties.type, properties.mask);
+            builder.getColumn(columnIdentifier).setColumnConstraints(columnConstraints.get(columnIdentifier));
         }
 
         if (useCompactStorage)
@@ -365,22 +367,16 @@ public final class CreateTableStatement extends AlterSchemaStatement
         else
         {
             columns.forEach((column, properties) -> {
-                ColumnConstraints constraints = validateConstraints(column);
                 if (staticColumns.contains(column))
-                    builder.addStaticColumn(column, properties.type, properties.mask, constraints);
+                    builder.addStaticColumn(column, properties.type, properties.mask);
                 else
-                    builder.addRegularColumn(column, properties.type, properties.mask, constraints);
+                    builder.addRegularColumn(column, properties.type, properties.mask);
             });
         }
 
-        return builder;
-    }
+        columns.keySet().forEach(id -> builder.getColumn(id).setColumnConstraints(columnConstraints.get(id)));
 
-    private ColumnConstraints validateConstraints(ColumnIdentifier columnIdentifier)
-    {
-        ColumnConstraints constraints = columnConstraints.get(columnIdentifier);
-        constraints.checkInvalidConstraintsCombinations(columnIdentifier);
-        return constraints;
+        return builder;
     }
 
     private void validateCompactTable(List<ColumnProperties> clusteringColumnProperties,

src/java/org/apache/cassandra/schema/ColumnMetadata.java

@@ -31,7 +31,6 @@ import com.google.common.collect.Collections2;
 import com.google.common.collect.Lists;
 
 import org.apache.cassandra.cql3.*;
-import org.apache.cassandra.cql3.constraints.AbstractFunctionConstraint;
 import org.apache.cassandra.cql3.constraints.ColumnConstraint;
 import org.apache.cassandra.cql3.constraints.ColumnConstraints;
 import org.apache.cassandra.cql3.functions.masking.ColumnMask;
@@ -327,7 +326,7 @@ public final class ColumnMetadata extends ColumnSpecification implements Selecta
             if (constraint.getConstraintType() == ColumnConstraint.ConstraintType.UNARY_FUNCTION ||
                 constraint.getConstraintType() == ColumnConstraint.ConstraintType.FUNCTION)
             {
-                if (((AbstractFunctionConstraint<?>) constraint).name().equals(name))
+                if (constraint.name().equals(name))
                     return true;
             }
         }
@@ -366,7 +365,7 @@ public final class ColumnMetadata extends ColumnSpecification implements Selecta
 
     public void setColumnConstraints(ColumnConstraints constraints)
     {
-        constraints.checkInvalidConstraintsCombinations(name);
+        constraints.validate(this);
         this.columnConstraints = constraints;
     }
 

test/distributed/org/apache/cassandra/distributed/test/ColumnConstraintsTest.java

@@ -23,6 +23,7 @@ import java.util.Map;
 import java.util.Set;
 
 import org.apache.cassandra.cql3.constraints.ConstraintViolationException;
+import org.apache.cassandra.cql3.constraints.InvalidConstraintDefinitionException;
 import org.apache.cassandra.exceptions.InvalidRequestException;
 import org.junit.Test;
 
@@ -51,7 +52,7 @@ public class ColumnConstraintsTest extends TestBaseImpl
         {
             assertThrowsInvalidConstraintException(cluster, String.format("CREATE TABLE %s (pk int, ck1 text CHECK ck1 < 100, ck2 int, v int, " +
                                                                           "PRIMARY KEY ((pk), ck1, ck2));", tableName),
-                                                   "ck1 is not a number");
+                                                   "Column 'ck1' is not a number type.");
 
             assertThrowsInvalidConstraintException(cluster, String.format("CREATE TABLE %s (pk int, ck1 int CHECK LENGTH(ck1) < 100, ck2 int, v int, " +
                                                                           "PRIMARY KEY ((pk), ck1, ck2));", tableName),
@@ -319,6 +320,6 @@ public class ColumnConstraintsTest extends TestBaseImpl
         assertThatThrownBy(() -> cluster.schemaChange(statement))
                   .describedAs(description)
                   .has(new Condition<Throwable>(t -> t.getClass().getCanonicalName()
-                                                      .equals(InvalidRequestException.class.getCanonicalName()), description));
+                                                      .equals(InvalidConstraintDefinitionException.class.getCanonicalName()), description));
     }
 }

test/unit/org/apache/cassandra/contraints/ConstraintsSatisfiabilityTest.java

@@ -0,0 +1,180 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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.apache.cassandra.contraints;
+
+import org.junit.Test;
+
+import org.apache.cassandra.cql3.ColumnIdentifier;
+import org.apache.cassandra.cql3.Operator;
+import org.apache.cassandra.cql3.constraints.ColumnConstraints;
+import org.apache.cassandra.cql3.constraints.FunctionColumnConstraint;
+import org.apache.cassandra.cql3.constraints.ScalarColumnConstraint;
+import org.apache.cassandra.cql3.constraints.ScalarColumnConstraint.Raw;
+import org.apache.cassandra.db.marshal.IntegerType;
+import org.apache.cassandra.db.marshal.UTF8Type;
+import org.apache.cassandra.schema.ColumnMetadata;
+import org.assertj.core.api.ThrowableAssert;
+
+import static java.util.List.of;
+import static org.apache.cassandra.cql3.Operator.EQ;
+import static org.apache.cassandra.cql3.Operator.GT;
+import static org.apache.cassandra.cql3.Operator.GTE;
+import static org.apache.cassandra.cql3.Operator.LT;
+import static org.apache.cassandra.cql3.Operator.LTE;
+import static org.apache.cassandra.cql3.Operator.NEQ;
+import static org.apache.cassandra.cql3.constraints.ScalarColumnConstraint.SUPPORTED_OPERATORS;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+public class ConstraintsSatisfiabilityTest
+{
+    private static final ColumnIdentifier columnIdentifier = new ColumnIdentifier("a_column", false);
+    private static final ColumnIdentifier lengthFunctionIdentifier = new ColumnIdentifier("LENGTH", false);
+    private static final ColumnMetadata regularIntColumn = new ColumnMetadata("a", "b", columnIdentifier, IntegerType.instance, -1, ColumnMetadata.Kind.REGULAR, null);
+    private static final ColumnMetadata regularStringColumn = new ColumnMetadata("a", "b", columnIdentifier, UTF8Type.instance, -1, ColumnMetadata.Kind.REGULAR, null);
+
+    @Test
+    public void testScalarSatisfiability() throws Throwable
+    {
+        run(this::scalar, regularIntColumn);
+    }
+
+    @Test
+    public void testLengthSatisfiability() throws Throwable
+    {
+        run(this::length, regularStringColumn);
+    }
+
+    private <T> void run(QuadFunction<T> quadFunction, ColumnMetadata columnMetadata) throws Throwable
+    {
+        for (Operator op1 : SUPPORTED_OPERATORS)
+        {
+            for (Operator op2 : SUPPORTED_OPERATORS)
+            {
+                if (op1 == op2)
+                {
+                    if (op1 == NEQ)
+                    {
+                        // a_column != 0 and a_column != 10 -> valid
+                        check(op1, 0, op2, 100, quadFunction, null, columnMetadata);
+                        // does not make sense to check twice
+                        // check a_column != 0 and a_column != 0
+                        check(op1, 0, op2, 0, quadFunction, "There are duplicate constraint definitions on column", columnMetadata);
+                    }
+                    else
+                        check(op1, 0, op2, 100, quadFunction, "There are duplicate constraint definitions on column", columnMetadata);
+                }
+                else if ((op1 == GT && op2 == GTE) ||
+                         (op1 == GTE && op2 == GT) ||
+                         (op1 == LT && op2 == LTE) ||
+                         (op1 == LTE && op2 == LT) ||
+                         (op1 == EQ || op2 == EQ))
+                {
+                    check(op1, 0, op2, 100, quadFunction, "not supported", columnMetadata);
+                }
+                else if ((op1 == LTE && op2 == GT) ||
+                         (op1 == LT && op2 == GT) ||
+                         (op1 == LTE && op2 == GTE) ||
+                         (op1 == LT && op2 == GTE))
+                {
+                    check(op1, 0, op2, 100, quadFunction, "are not satisfiable", columnMetadata);
+                }
+                else if (!(op1 == NEQ || op2 == NEQ))
+                {
+                    check(op1, 0, op2, 100, quadFunction, null, columnMetadata);
+                }
+                else
+                {
+                    // this is valid
+                    // a_column < 0, a_column != 10
+                }
+            }
+        }
+    }
+
+    @Test
+    public void testNumberOfScalarConstraints()
+    {
+        // one
+        new ColumnConstraints(of(scalar(LT, 5))).validate(regularIntColumn);
+
+        // two
+        new ColumnConstraints(of(scalar(LT, 5), scalar(GT, 0))).validate(regularIntColumn);
+
+        // three - invalid
+
+        assertThatThrownBy(() -> new ColumnConstraints(of(scalar(LT, 5), scalar(GT, 0), scalar(GTE, 0))).validate(regularIntColumn))
+        .hasMessage("There can not be more than 2 constraints (not including non-equal relations) on a column 'a_column' but you have specified 3");
+
+        // valid
+        new ColumnConstraints(of(scalar(LT, 5), scalar(GT, 0), scalar(NEQ, 3))).validate(regularIntColumn);
+
+        // valid, because NEQs have different terms
+        new ColumnConstraints(of(scalar(LT, 5), scalar(GT, 0), scalar(NEQ, 3), scalar(NEQ, 4))).validate(regularIntColumn);
+
+        // this has duplicate a_column != 3
+        assertThatThrownBy(() -> new ColumnConstraints(of(scalar(LT, 5), scalar(GT, 0), scalar(NEQ, 3), scalar(NEQ, 3))).validate(regularIntColumn))
+        .hasMessage("There are duplicate constraint definitions on column 'a_column': a_column != 3");
+    }
+
+    private interface QuadFunction<T>
+    {
+        ColumnConstraints f(Operator op1, Integer term1, Operator o2, Integer term2);
+    }
+
+    private <T> void check(Operator operator,
+                           Integer term,
+                           Operator operator2,
+                           Integer term2,
+                           QuadFunction<T> quadFunction,
+                           String exceptionMessage,
+                           ColumnMetadata columnMetadata) throws Throwable
+    {
+        ThrowableAssert.ThrowingCallable callable = () -> quadFunction.f(operator, term, operator2, term2).validate(columnMetadata);
+
+        if (exceptionMessage != null)
+            assertThatThrownBy(callable).hasMessageContaining(exceptionMessage);
+        else
+            callable.call();
+    }
+
+    private ColumnConstraints scalar(Operator operator, Integer term, Operator operator2, Integer term2)
+    {
+        return new ColumnConstraints(of(scalar(operator, term),
+                                        scalar(operator2, term2)));
+    }
+
+    private ScalarColumnConstraint scalar(Operator operator, Integer term)
+    {
+        return new Raw(columnIdentifier, operator, term.toString()).prepare();
+    }
+
+    private FunctionColumnConstraint length(Operator operator, Integer term)
+    {
+        return new FunctionColumnConstraint.Raw(lengthFunctionIdentifier,
+                                                columnIdentifier,
+                                                operator,
+                                                term.toString()).prepare();
+    }
+
+    private ColumnConstraints length(Operator operator, Integer term, Operator operator2, Integer term2)
+    {
+        return new ColumnConstraints(of(length(operator, term),
+                                        length(operator2, term2)));
+    }
+}