Add CQL functions for dynamic data masking

patch by Andrés de la Peña; reviewed by Benjamin Lerer for CASSANDRA-17941

CHANGES.txt

@@ -1,4 +1,5 @@
 4.2
+ * Add CQL functions for dynamic data masking (CASSANDRA-17941)
  * Print friendly error when nodetool attempts to connect to uninitialized server (CASSANDRA-11537)
  * Use G1GC by default, and update default G1GC settings (CASSANDRA-18027)
  * SimpleSeedProvider can resolve multiple IP addresses per DNS record (CASSANDRA-14361)

NEWS.txt

@@ -115,6 +115,13 @@ New features
       the paramater called `resolve_multiple_ip_addresses_per_dns_record` which value is meant to be boolean and by
       default it is set to false. When set to true, SimpleSeedProvider will resolve all IP addresses per DNS record,
       based on the configured name service on the system.
+    - Added new native CQL functions for data masking, allowing to replace or obscure sensitive data. The functions are:
+      - `mask_null` replaces the column value by null.
+      - `mask_default` replaces the data by a fixed default value of the same type.
+      - `mask_replace` replaces the data by a custom value.
+      - `mask_inner` replaces every character but the first and last ones by a fixed character.
+      - `mask_outer` replaces the first and last characters by a fixed character.
+      - `mask_hash` replaces the data by its hash, according to the specified algorithm.
 
 Upgrading
 ---------

doc/modules/cassandra/pages/cql/functions.adoc

@@ -281,6 +281,72 @@ A number of functions are provided to operate on collection columns.
 | `collection_avg` | numeric `set` or `list` | Computes the average of the elements in the collection argument. The average of an empty collection returns zero. The returned value is of the same type as the input collection elements, which might include rounding and truncations. For example `collection_avg([1, 2])` returns `1` instead of `1.5`.
 |===
 
+===== Data masking functions
+
+A number of functions allow to obscure the real contents of a column containing sensitive data.
+
+[cols=",",options="header",]
+|===
+|Function | Description
+
+| `mask_null(value)` | Replaces the first argument by a `null` column. The returned value is always an absent column, as it didn't exist, and not a not-null column representing a `null` value.
+
+Examples:
+
+`mask_null('Alice')` -> `null`
+
+`mask_null(123)` -> `null`
+
+| `mask_default(value)` | Replaces its argument by an arbitrary, fixed default value of the same type. This will be `\***\***` for text values, zero for numeric values, `false` for booleans, etc.
+
+Examples:
+
+`mask_default('Alice')` -> `'\****'`
+
+`mask_default(123)` -> `0`
+
+| `mask_replace(value, replacement])` | Replaces the first argument by the replacement value on the second argument. The replacement value needs to have the same type as the replaced value.
+
+Examples:
+
+`mask_replace('Alice', 'REDACTED')` -> `'REDACTED'`
+
+`mask_replace(123, -1)` -> `-1`
+
+| `mask_inner(value, begin, end, [padding])` | Returns a copy of the first `text`, `varchar` or `ascii` argument, replacing each character except the first and last ones by a padding character. The 2nd and 3rd arguments are the size of the exposed prefix and suffix. The optional 4th argument is the padding character, `\*` by default.
+
+Examples:
+
+`mask_inner('Alice', 1, 2)` -> `'A**ce'`
+
+`mask_inner('Alice', 1, null)` -> `'A****'`
+
+`mask_inner('Alice', null, 2)` -> `'***ce'`
+
+`mask_inner('Alice', 2, 1, '\#')` -> `'Al##e'`
+
+| `mask_outer(value, begin, end, [padding])` | Returns a copy of the first `text`, `varchar` or `ascii` argument, replacing the first and last character by a padding character. The 2nd and 3rd arguments are the size of the exposed prefix and suffix. The optional 4th argument is the padding character, `\*` by default.
+
+Examples:
+
+`mask_outer('Alice', 1, 2)` -> `'*li**'`
+
+`mask_outer('Alice', 1, null)` -> `'*lice'`
+
+`mask_outer('Alice', null, 2)` -> `'Ali**'`
+
+`mask_outer('Alice', 2, 1, '\#')` -> `'##ic#'`
+
+| `mask_hash(value, [algorithm])` | Returns a `blob` containing the hash of the first argument. The optional 2nd argument is the hashing algorithm to be used, according the available Java security provider. The default hashing algorithm is `SHA-256`.
+
+Examples:
+
+`mask_hash('Alice')`
+
+`mask_hash('Alice', 'SHA-512')`
+
+|===
+
 [[user-defined-scalar-functions]]
 ==== User-defined functions
 

src/java/org/apache/cassandra/cql3/CQL3Type.java

@@ -177,7 +177,7 @@ public interface CQL3Type
             this.type = type;
         }
 
-        public AbstractType<?> getType()
+        public CollectionType<?> getType()
         {
             return type;
         }
@@ -410,7 +410,7 @@ public interface CQL3Type
             return new Tuple(type);
         }
 
-        public AbstractType<?> getType()
+        public TupleType getType()
         {
             return type;
         }

src/java/org/apache/cassandra/cql3/functions/FromJsonFct.java

@@ -22,6 +22,7 @@ import java.nio.ByteBuffer;
 import java.util.*;
 import java.util.concurrent.ConcurrentHashMap;
 
+import org.apache.cassandra.cql3.CQL3Type;
 import org.apache.cassandra.cql3.Json;
 
 import org.apache.cassandra.cql3.QueryOptions;
@@ -81,7 +82,7 @@ public class FromJsonFct extends NativeScalarFunction
 
     public static void addFunctionsTo(NativeFunctions functions)
     {
-        functions.add(new FunctionFactory(NAME.name, FunctionParameter.fixed(UTF8Type.instance))
+        functions.add(new FunctionFactory(NAME.name, FunctionParameter.fixed(CQL3Type.Native.TEXT))
         {
             @Override
             protected NativeFunction doGetOrCreateFunction(List<AbstractType<?>> argTypes, AbstractType<?> receiverType)

src/java/org/apache/cassandra/cql3/functions/FunctionFactory.java

@@ -23,6 +23,8 @@ import java.util.Arrays;
 import java.util.List;
 import java.util.stream.Collectors;
 
+import javax.annotation.Nullable;
+
 import org.apache.cassandra.cql3.AssignmentTestable;
 import org.apache.cassandra.db.marshal.AbstractType;
 import org.apache.cassandra.exceptions.InvalidRequestException;
@@ -46,6 +48,9 @@ public abstract class FunctionFactory
     /** The accepted parameters. */
     protected final List<FunctionParameter> parameters;
 
+    private final int numParameters;
+    private final int numMandatoryParameters;
+
     /**
      * @param name the name of the built functions
      * @param parameters the accepted parameters
@@ -54,6 +59,8 @@ public abstract class FunctionFactory
     {
         this.name = FunctionName.nativeFunction(name);
         this.parameters = Arrays.asList(parameters);
+        this.numParameters = parameters.length;
+        this.numMandatoryParameters = (int) this.parameters.stream().filter(p -> !p.isOptional()).count();
     }
 
     public FunctionName name()
@@ -66,18 +73,22 @@ public abstract class FunctionFactory
      *
      * @param args the arguments in the function call for which the function is going to be built
      * @param receiverType the expected return type of the function call for which the function is going to be built
-     * @param receiverKs the name of the recevier keyspace
-     * @param receiverCf the name of the recevier table
-     * @return a function with a signature compatible with the specified function call
+     * @param receiverKeyspace the name of the recevier keyspace
+     * @param receiverTable the name of the recevier table
+     * @return a function with a signature compatible with the specified function call, or {@code null} if the factory
+     * cannot create a function for the supplied arguments but there might be another factory with the same
+     * {@link #name()} able to do it.
      */
+    @Nullable
     public NativeFunction getOrCreateFunction(List<? extends AssignmentTestable> args,
                                               AbstractType<?> receiverType,
-                                              String receiverKs,
-                                              String receiverCf)
+                                              String receiverKeyspace,
+                                              String receiverTable)
     {
         // validate the number of arguments
-        if (args.size() != parameters.size())
-            throw new InvalidRequestException("Invalid number of arguments for function " + this);
+        int numArgs = args.size();
+        if (numArgs < numMandatoryParameters || numArgs > numParameters)
+            throw invalidNumberOfArgumentsException();
 
         // try to infer the types of the arguments
         List<AbstractType<?>> types = new ArrayList<>(args.size());
@@ -85,7 +96,7 @@ public abstract class FunctionFactory
         {
             AssignmentTestable arg = args.get(i);
             FunctionParameter parameter = parameters.get(i);
-            AbstractType<?> type = parameter.inferType(SchemaConstants.SYSTEM_KEYSPACE_NAME, arg, receiverType);
+            AbstractType<?> type = parameter.inferType(SchemaConstants.SYSTEM_KEYSPACE_NAME, arg, receiverType, types);
             if (type == null)
                 throw new InvalidRequestException(String.format("Cannot infer type of argument %s in call to " +
                                                                 "function %s: use type casts to disambiguate",
@@ -98,13 +109,20 @@ public abstract class FunctionFactory
         return doGetOrCreateFunction(types, receiverType);
     }
 
+    public InvalidRequestException invalidNumberOfArgumentsException()
+    {
+        return new InvalidRequestException("Invalid number of arguments for function " + this);
+    }
+
     /**
      * Returns a function compatible with the specified signature.
      *
      * @param argTypes the types of the function arguments
      * @param receiverType the expected return type of the function
-     * @return a function compatible with the specified signature
+     * @return a function compatible with the specified signature, or {@code null} if this cannot create a function for
+     * the supplied arguments but there might be another factory with the same {@link #name()} able to do it.
      */
+    @Nullable
     protected abstract NativeFunction doGetOrCreateFunction(List<AbstractType<?>> argTypes, AbstractType<?> receiverType);
 
     @Override

src/java/org/apache/cassandra/cql3/functions/FunctionParameter.java

@@ -18,9 +18,13 @@
 
 package org.apache.cassandra.cql3.functions;
 
+import java.util.Arrays;
+import java.util.List;
+import java.util.stream.Collectors;
 import javax.annotation.Nullable;
 
 import org.apache.cassandra.cql3.AssignmentTestable;
+import org.apache.cassandra.cql3.CQL3Type;
 import org.apache.cassandra.db.marshal.AbstractType;
 import org.apache.cassandra.db.marshal.CollectionType;
 import org.apache.cassandra.db.marshal.ListType;
@@ -47,7 +51,10 @@ public interface FunctionParameter
      * @return the inferred data type of the parameter, or {@link null} it isn't possible to infer it
      */
     @Nullable
-    default AbstractType<?> inferType(String keyspace, AssignmentTestable arg, @Nullable AbstractType<?> receiverType)
+    default AbstractType<?> inferType(String keyspace,
+                                      AssignmentTestable arg,
+                                      @Nullable AbstractType<?> receiverType,
+                                      List<AbstractType<?>> previousTypes)
     {
         return arg.getCompatibleTypeIfKnown(keyspace);
     }
@@ -55,33 +62,96 @@ public interface FunctionParameter
     void validateType(FunctionName name, AssignmentTestable arg, AbstractType<?> argType);
 
     /**
-     * @param type the accepted data type
-     * @return a function parameter definition that accepts values of a specific data type
+     * @return whether this parameter is optional
      */
-    public static FunctionParameter fixed(AbstractType<?> type)
+    default boolean isOptional()
+    {
+        return false;
+    }
+
+    /**
+     * @param wrapped the wrapped parameter
+     * @return a function parameter definition that accepts the specified wrapped parameter, considering it optional as
+     * defined by {@link #isOptional()}.
+     */
+    static FunctionParameter optional(FunctionParameter wrapped)
     {
         return new FunctionParameter()
         {
+            @Nullable
             @Override
-            public AbstractType<?> inferType(String keyspace, AssignmentTestable arg, AbstractType<?> receiverType)
+            public AbstractType<?> inferType(String keyspace,
+                                             AssignmentTestable arg,
+                                             @Nullable AbstractType<?> receiverType,
+                                             List<AbstractType<?>> previousTypes)
             {
-                AbstractType<?> inferred = arg.getCompatibleTypeIfKnown(keyspace);
-                return inferred != null ? inferred : type;
+                return wrapped.inferType(keyspace, arg, receiverType, previousTypes);
             }
 
             @Override
             public void validateType(FunctionName name, AssignmentTestable arg, AbstractType<?> argType)
             {
-                if (argType.testAssignment(type) == NOT_ASSIGNABLE)
-                    throw new InvalidRequestException(format("Function %s requires an argument of type %s, " +
-                                                             "but found argument %s of type %s",
-                                                             name, type, arg, argType.asCQL3Type()));
+                wrapped.validateType(name, arg, argType);
+            }
+
+            @Override
+            public boolean isOptional()
+            {
+                return true;
             }
 
             @Override
             public String toString()
             {
-                return type.toString();
+                return '[' + wrapped.toString() + ']';
+            }
+        };
+    }
+
+    /**
+     * @return a function parameter definition that accepts values of string-based data types (text, varchar and ascii)
+     */
+    static FunctionParameter string()
+    {
+        return fixed(CQL3Type.Native.TEXT, CQL3Type.Native.VARCHAR, CQL3Type.Native.ASCII);
+    }
+
+    /**
+     * @param types the accepted data types
+     * @return a function parameter definition that accepts values of a specific data type
+     */
+    static FunctionParameter fixed(CQL3Type... types)
+    {
+        assert types.length > 0;
+
+        return new FunctionParameter()
+        {
+            @Override
+            public AbstractType<?> inferType(String keyspace,
+                                             AssignmentTestable arg,
+                                             @Nullable AbstractType<?> receiverType,
+                                             List<AbstractType<?>> previousTypes)
+            {
+                AbstractType<?> inferred = arg.getCompatibleTypeIfKnown(keyspace);
+                return inferred != null ? inferred : types[0].getType();
+            }
+
+            @Override
+            public void validateType(FunctionName name, AssignmentTestable arg, AbstractType<?> argType)
+            {
+                if (Arrays.stream(types).allMatch(t -> argType.testAssignment(t.getType()) == NOT_ASSIGNABLE))
+                    throw new InvalidRequestException(format("Function %s requires an argument of type %s, " +
+                                                             "but found argument %s of type %s",
+                                                             name, this, arg, argType.asCQL3Type()));
+            }
+
+            @Override
+            public String toString()
+            {
+                if (types.length == 1)
+                    return types[0].toString();
+
+                return '[' + Arrays.stream(types).map(Object::toString).collect(Collectors.joining("|")) + ']';
             }
         };
     }
@@ -90,12 +160,15 @@ public interface FunctionParameter
      * @param inferFromReceiver whether the parameter should try to use the function receiver to infer its data type
      * @return a function parameter definition that accepts columns of any data type
      */
-    public static FunctionParameter anyType(boolean inferFromReceiver)
+    static FunctionParameter anyType(boolean inferFromReceiver)
     {
         return new FunctionParameter()
         {
             @Override
-            public AbstractType<?> inferType(String keyspace, AssignmentTestable arg, AbstractType<?> receiverType)
+            public AbstractType<?> inferType(String keyspace,
+                                             AssignmentTestable arg,
+                                             @Nullable AbstractType<?> receiverType,
+                                             List<AbstractType<?>> previousTypes)
             {
                 AbstractType<?> type = arg.getCompatibleTypeIfKnown(keyspace);
                 return type == null && inferFromReceiver ? receiverType : type;
@@ -115,11 +188,41 @@ public interface FunctionParameter
         };
     }
 
+    /**
+     * @return a function parameter definition that accepts values with the same type as the first parameter
+     */
+    static FunctionParameter sameAsFirst()
+    {
+        return new FunctionParameter()
+        {
+            @Override
+            public AbstractType<?> inferType(String keyspace,
+                                             AssignmentTestable arg,
+                                             @Nullable AbstractType<?> receiverType,
+                                             List<AbstractType<?>> previousTypes)
+            {
+                return previousTypes.get(0);
+            }
+
+            @Override
+            public void validateType(FunctionName name, AssignmentTestable arg, AbstractType<?> argType)
+            {
+                // nothing to do here, all types are accepted
+            }
+
+            @Override
+            public String toString()
+            {
+                return "same";
+            }
+        };
+    }
+
     /**
      * @return a function parameter definition that accepts values of type {@link CollectionType}, independently of the
      * types of its elements.
      */
-    public static FunctionParameter anyCollection()
+    static FunctionParameter anyCollection()
     {
         return new FunctionParameter()
         {
@@ -143,7 +246,7 @@ public interface FunctionParameter
     /**
      * @return a function parameter definition that accepts values of type {@link SetType} or {@link ListType}.
      */
-    public static FunctionParameter setOrList()
+    static FunctionParameter setOrList()
     {
         return new FunctionParameter()
         {
@@ -174,7 +277,7 @@ public interface FunctionParameter
      * @return a function parameter definition that accepts values of type {@link SetType} or {@link ListType},
      * provided that its elements are numeric.
      */
-    public static FunctionParameter numericSetOrList()
+    static FunctionParameter numericSetOrList()
     {
         return new FunctionParameter()
         {
@@ -213,7 +316,7 @@ public interface FunctionParameter
      * @return a function parameter definition that accepts values of type {@link MapType}, independently of the types
      * of the map keys and values.
      */
-    public static FunctionParameter anyMap()
+    static FunctionParameter anyMap()
     {
         return new FunctionParameter()
         {

src/java/org/apache/cassandra/cql3/functions/FunctionResolver.java

@@ -20,6 +20,7 @@ package org.apache.cassandra.cql3.functions;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.List;
+import java.util.Objects;
 import java.util.stream.Collectors;
 
 import javax.annotation.Nullable;
@@ -41,10 +42,10 @@ public final class FunctionResolver
     {
     }
 
-    public static ColumnSpecification makeArgSpec(String receiverKs, String receiverCf, Function fun, int i)
+    public static ColumnSpecification makeArgSpec(String receiverKeyspace, String receiverTable, Function fun, int i)
     {
-        return new ColumnSpecification(receiverKs,
-                                       receiverCf,
+        return new ColumnSpecification(receiverKeyspace,
+                                       receiverTable,
                                        new ColumnIdentifier("arg" + i + '(' + fun.name().toString().toLowerCase() + ')', true),
                                        fun.argTypes().get(i));
     }
@@ -53,8 +54,8 @@ public final class FunctionResolver
      * @param keyspace the current keyspace
      * @param name the name of the function
      * @param providedArgs the arguments provided for the function call
-     * @param receiverKs the receiver's keyspace
-     * @param receiverCf the receiver's table
+     * @param receiverKeyspace the receiver's keyspace
+     * @param receiverTable the receiver's table
      * @param receiverType if the receiver type is known (during inserts, for example), this should be the type of
      *                     the receiver
      */
@@ -62,12 +63,12 @@ public final class FunctionResolver
     public static Function get(String keyspace,
                                FunctionName name,
                                List<? extends AssignmentTestable> providedArgs,
-                               String receiverKs,
-                               String receiverCf,
+                               String receiverKeyspace,
+                               String receiverTable,
                                AbstractType<?> receiverType)
     throws InvalidRequestException
     {
-        Collection<Function> candidates = collectCandidates(keyspace, name, receiverKs, receiverCf, providedArgs, receiverType);
+        Collection<Function> candidates = collectCandidates(keyspace, name, receiverKeyspace, receiverTable, providedArgs, receiverType);
 
         if (candidates.isEmpty())
             return null;
@@ -76,17 +77,17 @@ public final class FunctionResolver
         if (candidates.size() == 1)
         {
             Function fun = candidates.iterator().next();
-            validateTypes(keyspace, fun, providedArgs, receiverKs, receiverCf);
+            validateTypes(keyspace, fun, providedArgs, receiverKeyspace, receiverTable);
             return fun;
         }
 
-        return pickBestMatch(keyspace, name, providedArgs, receiverKs, receiverCf, receiverType, candidates);
+        return pickBestMatch(keyspace, name, providedArgs, receiverKeyspace, receiverTable, receiverType, candidates);
     }
 
     private static Collection<Function> collectCandidates(String keyspace,
                                                           FunctionName name,
-                                                          String receiverKs,
-                                                          String receiverCf,
+                                                          String receiverKeyspace,
+                                                          String receiverTable,
                                                           List<? extends AssignmentTestable> providedArgs,
                                                           AbstractType<?> receiverType)
     {
@@ -98,7 +99,8 @@ public final class FunctionResolver
             candidates.addAll(Schema.instance.getUserFunctions(name));
             candidates.addAll(NativeFunctions.instance.getFunctions(name));
             candidates.addAll(NativeFunctions.instance.getFactories(name).stream()
-                                            .map(f -> f.getOrCreateFunction(providedArgs, receiverType, receiverKs, receiverCf))
+                                            .map(f -> f.getOrCreateFunction(providedArgs, receiverType, receiverKeyspace, receiverTable))
+                                            .filter(Objects::nonNull)
                                             .collect(Collectors.toList()));
         }
         else
@@ -110,7 +112,8 @@ public final class FunctionResolver
             FunctionName nativeName = name.asNativeFunction();
             candidates.addAll(NativeFunctions.instance.getFunctions(nativeName));
             candidates.addAll(NativeFunctions.instance.getFactories(nativeName).stream()
-                                            .map(f -> f.getOrCreateFunction(providedArgs, receiverType, receiverKs, receiverCf))
+                                            .map(f -> f.getOrCreateFunction(providedArgs, receiverType, receiverKeyspace, receiverTable))
+                                            .filter(Objects::nonNull)
                                             .collect(Collectors.toList()));
         }
 
@@ -120,8 +123,8 @@ public final class FunctionResolver
     private static Function pickBestMatch(String keyspace,
                                           FunctionName name,
                                           List<? extends AssignmentTestable> providedArgs,
-                                          String receiverKs,
-                                          String receiverCf,
+                                          String receiverKeyspace,
+                                          String receiverTable,
                                           AbstractType<?> receiverType,
                                           Collection<Function> candidates)
     {
@@ -130,7 +133,7 @@ public final class FunctionResolver
         {
             if (matchReturnType(toTest, receiverType))
             {
-                AssignmentTestable.TestResult r = matchAguments(keyspace, toTest, providedArgs, receiverKs, receiverCf);
+                AssignmentTestable.TestResult r = matchAguments(keyspace, toTest, providedArgs, receiverKeyspace, receiverTable);
                 switch (r)
                 {
                     case EXACT_MATCH:
@@ -211,8 +214,8 @@ public final class FunctionResolver
     private static void validateTypes(String keyspace,
                                       Function fun,
                                       List<? extends AssignmentTestable> providedArgs,
-                                      String receiverKs,
-                                      String receiverCf)
+                                      String receiverKeyspace,
+                                      String receiverTable)
     {
         if (providedArgs.size() != fun.argTypes().size())
             throw invalidRequest("Invalid number of arguments in call to function %s: %d required but %d provided",
@@ -227,7 +230,7 @@ public final class FunctionResolver
             if (provided == null)
                 continue;
 
-            ColumnSpecification expected = makeArgSpec(receiverKs, receiverCf, fun, i);
+            ColumnSpecification expected = makeArgSpec(receiverKeyspace, receiverTable, fun, i);
             if (!provided.testAssignment(keyspace, expected).isAssignable())
                 throw invalidRequest("Type error: %s cannot be passed as argument %d of function %s of type %s",
                                      provided, i, fun.name(), expected.type.asCQL3Type());
@@ -237,8 +240,8 @@ public final class FunctionResolver
     private static AssignmentTestable.TestResult matchAguments(String keyspace,
                                                                Function fun,
                                                                List<? extends AssignmentTestable> providedArgs,
-                                                               String receiverKs,
-                                                               String receiverCf)
+                                                               String receiverKeyspace,
+                                                               String receiverTable)
     {
         if (providedArgs.size() != fun.argTypes().size())
             return AssignmentTestable.TestResult.NOT_ASSIGNABLE;
@@ -254,7 +257,7 @@ public final class FunctionResolver
                 continue;
             }
 
-            ColumnSpecification expected = makeArgSpec(receiverKs, receiverCf, fun, i);
+            ColumnSpecification expected = makeArgSpec(receiverKeyspace, receiverTable, fun, i);
             AssignmentTestable.TestResult argRes = provided.testAssignment(keyspace, expected);
             if (argRes == AssignmentTestable.TestResult.NOT_ASSIGNABLE)
                 return AssignmentTestable.TestResult.NOT_ASSIGNABLE;

src/java/org/apache/cassandra/cql3/functions/NativeFunctions.java

@@ -23,6 +23,8 @@ import java.util.Collection;
 import com.google.common.collect.HashMultimap;
 import com.google.common.collect.Multimap;
 
+import org.apache.cassandra.cql3.functions.masking.MaskingFcts;
+
 /**
  * A container of native functions. It stores both pre-built function overloads ({@link NativeFunction}) and
  * dynamic generators of functions ({@link FunctionFactory}).
@@ -43,6 +45,7 @@ public class NativeFunctions
             CollectionFcts.addFunctionsTo(this);
             BytesConversionFcts.addFunctionsTo(this);
             MathFcts.addFunctionsTo(this);
+            MaskingFcts.addFunctionsTo(this);
         }
     };
 

src/java/org/apache/cassandra/cql3/functions/TokenFct.java

@@ -68,20 +68,20 @@ public class TokenFct extends NativeScalarFunction
             @Override
             public NativeFunction getOrCreateFunction(List<? extends AssignmentTestable> args,
                                                       AbstractType<?> receiverType,
-                                                      String receiverKs,
-                                                      String receiverCf)
+                                                      String receiverKeyspace,
+                                                      String receiverTable)
             {
-                if (receiverKs == null)
+                if (receiverKeyspace == null)
                     throw new InvalidRequestException("No receiver keyspace has been specified for function " + name);
 
-                if (receiverCf == null)
+                if (receiverTable == null)
                     throw new InvalidRequestException("No receiver table has been specified for function " + name);
 
-                TableMetadata metadata = Schema.instance.getTableMetadata(receiverKs, receiverCf);
+                TableMetadata metadata = Schema.instance.getTableMetadata(receiverKeyspace, receiverTable);
                 if (metadata == null)
                     throw new InvalidRequestException(String.format("The receiver table %s.%s specified by call to " +
                                                                     "function %s hasn't been found",
-                                                                    receiverKs, receiverCf, name));
+                                                                    receiverKeyspace, receiverTable, name));
 
                 return new TokenFct(metadata);
             }

src/java/org/apache/cassandra/cql3/functions/masking/DefaultMaskingFunction.java

@@ -0,0 +1,70 @@
+/*
+ * 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.functions.masking;
+
+import java.nio.ByteBuffer;
+import java.util.List;
+
+import org.apache.cassandra.cql3.functions.FunctionFactory;
+import org.apache.cassandra.cql3.functions.FunctionName;
+import org.apache.cassandra.cql3.functions.FunctionParameter;
+import org.apache.cassandra.cql3.functions.NativeFunction;
+import org.apache.cassandra.db.marshal.AbstractType;
+import org.apache.cassandra.transport.ProtocolVersion;
+
+/**
+ * A {@link MaskingFunction} that returns a fixed replacement value for the data type of its single argument.
+ * <p>
+ * The default values are defined by {@link AbstractType#getMaskedValue()}, being {@code ****} for text fields,
+ * {@code false} for booleans, zero for numeric types, {@code 1970-01-01} for dates, etc.
+ * <p>
+ * For example, given a text column named "username", {@code mask_default(username)} will always return {@code ****},
+ * independently of the actual value of that column.
+ */
+public class DefaultMaskingFunction extends MaskingFunction
+{
+    public static final String NAME = "default";
+
+    private final ByteBuffer defaultValue;
+
+    private <T> DefaultMaskingFunction(FunctionName name, AbstractType<T> inputType)
+    {
+        super(name, inputType, inputType);
+        this.defaultValue = inputType.getMaskedValue();
+    }
+
+    @Override
+    public final ByteBuffer execute(ProtocolVersion protocolVersion, List<ByteBuffer> parameters)
+    {
+        return defaultValue;
+    }
+
+    /** @return a {@link FunctionFactory} to build new {@link DefaultMaskingFunction}s. */
+    public static FunctionFactory factory()
+    {
+        return new MaskingFunction.Factory(NAME, FunctionParameter.anyType(false))
+        {
+            @Override
+            protected NativeFunction doGetOrCreateFunction(List<AbstractType<?>> argTypes, AbstractType<?> receiverType)
+            {
+                return new DefaultMaskingFunction(name, argTypes.get(0));
+            }
+        };
+    }
+}

src/java/org/apache/cassandra/cql3/functions/masking/HashMaskingFunction.java

@@ -0,0 +1,143 @@
+/*
+ * 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.functions.masking;
+
+import java.nio.ByteBuffer;
+import java.security.MessageDigest;
+import java.security.NoSuchAlgorithmException;
+import java.util.List;
+import java.util.function.Supplier;
+import javax.annotation.Nullable;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Suppliers;
+
+import org.apache.cassandra.cql3.CQL3Type;
+import org.apache.cassandra.cql3.functions.FunctionFactory;
+import org.apache.cassandra.cql3.functions.FunctionName;
+import org.apache.cassandra.cql3.functions.FunctionParameter;
+import org.apache.cassandra.cql3.functions.NativeFunction;
+import org.apache.cassandra.db.marshal.AbstractType;
+import org.apache.cassandra.db.marshal.BytesType;
+import org.apache.cassandra.db.marshal.StringType;
+import org.apache.cassandra.db.marshal.UTF8Type;
+import org.apache.cassandra.exceptions.InvalidRequestException;
+import org.apache.cassandra.transport.ProtocolVersion;
+import org.apache.cassandra.utils.ByteBufferUtil;
+
+/**
+ * A {@link MaskingFunction} that replaces the specified column value by its hash according to the specified
+ * algorithm. The available algorithms are those defined by the registered security {@link java.security.Provider}s.
+ * If no algorithm is passed to the function, the {@link #DEFAULT_ALGORITHM} will be used.
+ */
+public class HashMaskingFunction extends MaskingFunction
+{
+    public static final String NAME = "hash";
+
+    /** The default hashing algorithm to be used if no other algorithm is specified in the call to the function. */
+    public static final String DEFAULT_ALGORITHM = "SHA-256";
+
+    // The default message digest is lazily built to prevent a failure during server startup if the algorithm is not
+    // available. That way, if the algorithm is not found only the calls to the function will fail.
+    private static final Supplier<MessageDigest> DEFAULT_DIGEST = Suppliers.memoize(() -> messageDigest(DEFAULT_ALGORITHM));
+    private static final AbstractType<?>[] DEFAULT_ARGUMENTS = {};
+
+    /** The type of the supplied algorithm argument, {@code null} if that argument isn't supplied. */
+    @Nullable
+    private final StringType algorithmArgumentType;
+
+    private HashMaskingFunction(FunctionName name, AbstractType<?> inputType, @Nullable StringType algorithmArgumentType)
+    {
+        super(name, BytesType.instance, inputType, argumentsType(algorithmArgumentType));
+        this.algorithmArgumentType = algorithmArgumentType;
+    }
+
+    private static AbstractType<?>[] argumentsType(@Nullable StringType algorithmArgumentType)
+    {
+        // the algorithm argument is optional, so we will have different signatures depending on whether that argument
+        // is supplied or not
+        return algorithmArgumentType == null
+               ? DEFAULT_ARGUMENTS
+               : new AbstractType<?>[]{ algorithmArgumentType };
+    }
+
+    @Override
+    public final ByteBuffer execute(ProtocolVersion protocolVersion, List<ByteBuffer> parameters)
+    {
+        MessageDigest digest;
+        if (algorithmArgumentType == null || parameters.get(1) == null)
+        {
+            digest = DEFAULT_DIGEST.get();
+        }
+        else
+        {
+            String algorithm = algorithmArgumentType.compose(parameters.get(1));
+            digest = messageDigest(algorithm);
+        }
+
+        return hash(digest, parameters.get(0));
+    }
+
+    @VisibleForTesting
+    @Nullable
+    static ByteBuffer hash(MessageDigest digest, ByteBuffer value)
+    {
+        if (value == null)
+            return null;
+
+        byte[] hash = digest.digest(ByteBufferUtil.getArray(value));
+        return BytesType.instance.compose(ByteBuffer.wrap(hash));
+    }
+
+    @VisibleForTesting
+    static MessageDigest messageDigest(String algorithm)
+    {
+        try
+        {
+            return MessageDigest.getInstance(algorithm);
+        }
+        catch (NoSuchAlgorithmException e)
+        {
+            throw new InvalidRequestException("Hash algorithm not found: " + algorithm);
+        }
+    }
+
+    /** @return a {@link FunctionFactory} to build new {@link HashMaskingFunction}s. */
+    public static FunctionFactory factory()
+    {
+        return new MaskingFunction.Factory(NAME,
+                                           FunctionParameter.anyType(false),
+                                           FunctionParameter.optional(FunctionParameter.fixed(CQL3Type.Native.TEXT)))
+        {
+            @Override
+            protected NativeFunction doGetOrCreateFunction(List<AbstractType<?>> argTypes, AbstractType<?> receiverType)
+            {
+                switch (argTypes.size())
+                {
+                    case 1:
+                        return new HashMaskingFunction(name, argTypes.get(0), null);
+                    case 2:
+                        return new HashMaskingFunction(name, argTypes.get(0), UTF8Type.instance);
+                    default:
+                        throw invalidNumberOfArgumentsException();
+                }
+            }
+        };
+    }
+}

src/java/org/apache/cassandra/cql3/functions/masking/MaskingFcts.java

@@ -0,0 +1,37 @@
+/*
+ * 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.functions.masking;
+
+import org.apache.cassandra.cql3.functions.NativeFunctions;
+
+/**
+ * A collection of {@link MaskingFunction}s for dynamic data masking, meant to obscure the real value of a column.
+ */
+public class MaskingFcts
+{
+    /** Adds all the available native data masking functions to the specified native functions. */
+    public static void addFunctionsTo(NativeFunctions functions)
+    {
+        functions.add(NullMaskingFunction.factory());
+        functions.add(ReplaceMaskingFunction.factory());
+        functions.add(DefaultMaskingFunction.factory());
+        functions.add(HashMaskingFunction.factory());
+        PartialMaskingFunction.factories().forEach(functions::add);
+    }
+}

src/java/org/apache/cassandra/cql3/functions/masking/MaskingFunction.java

@@ -0,0 +1,62 @@
+/*
+ * 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.functions.masking;
+
+import com.google.common.collect.ObjectArrays;
+
+import org.apache.cassandra.cql3.functions.FunctionFactory;
+import org.apache.cassandra.cql3.functions.FunctionName;
+import org.apache.cassandra.cql3.functions.FunctionParameter;
+import org.apache.cassandra.cql3.functions.NativeScalarFunction;
+import org.apache.cassandra.db.marshal.AbstractType;
+
+/**
+ * A {@link NativeScalarFunction} that totally or partially replaces the original value of a column value,
+ * meant to obscure the real value of the column.
+ * <p>
+ * The names of all masking functions share a common prefix, {@link MaskingFunction#NAME_PREFIX}, to easily identify
+ * their purpose.
+ */
+public abstract class MaskingFunction extends NativeScalarFunction
+{
+    /** The common prefix for the names of all the native data masking functions. */
+    public static final String NAME_PREFIX = "mask_";
+
+    /**
+     * @param name the name of the function
+     * @param outputType the type of the values returned by the function
+     * @param inputType the type of the values accepted by the function, always be the first argument of the function
+     * @param argsType the type of the additional arguments of the function
+     */
+    protected MaskingFunction(FunctionName name,
+                              AbstractType<?> outputType,
+                              AbstractType<?> inputType,
+                              AbstractType<?>... argsType)
+    {
+        super(name.name, outputType, ObjectArrays.concat(inputType, argsType));
+    }
+
+    protected static abstract class Factory extends FunctionFactory
+    {
+        public Factory(String name, FunctionParameter... parameters)
+        {
+            super(NAME_PREFIX + name.toLowerCase(), parameters);
+        }
+    }
+}

src/java/org/apache/cassandra/cql3/functions/masking/NullMaskingFunction.java

@@ -0,0 +1,65 @@
+/*
+ * 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.functions.masking;
+
+import java.nio.ByteBuffer;
+import java.util.List;
+
+import org.apache.cassandra.cql3.functions.FunctionFactory;
+import org.apache.cassandra.cql3.functions.FunctionName;
+import org.apache.cassandra.cql3.functions.FunctionParameter;
+import org.apache.cassandra.cql3.functions.NativeFunction;
+import org.apache.cassandra.db.marshal.AbstractType;
+import org.apache.cassandra.transport.ProtocolVersion;
+
+/**
+ * A {@link MaskingFunction} that always returns a {@code null} column. The returned value is always an absent column,
+ * as it didn't exist, and not a not-null column representing a {@code null} value.
+ * <p>
+ * For example, given a text column named "username", {@code mask_null(username)} will always return {@code null},
+ * independently of the actual value of that column.
+ */
+public class NullMaskingFunction extends MaskingFunction
+{
+    public static final String NAME = "null";
+
+    private NullMaskingFunction(FunctionName name, AbstractType<?> inputType)
+    {
+        super(name, inputType, inputType);
+    }
+
+    @Override
+    public final ByteBuffer execute(ProtocolVersion protocolVersion, List<ByteBuffer> parameters)
+    {
+        return null;
+    }
+
+    /** @return a {@link FunctionFactory} to build new {@link NullMaskingFunction}s. */
+    public static FunctionFactory factory()
+    {
+        return new MaskingFunction.Factory(NAME, FunctionParameter.anyType(false))
+        {
+            @Override
+            protected NativeFunction doGetOrCreateFunction(List<AbstractType<?>> argTypes, AbstractType<?> receiverType)
+            {
+                return new NullMaskingFunction(name, argTypes.get(0));
+            }
+        };
+    }
+}

src/java/org/apache/cassandra/cql3/functions/masking/PartialMaskingFunction.java

@@ -0,0 +1,191 @@
+/*
+ * 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.functions.masking;
+
+import java.nio.ByteBuffer;
+import java.util.Collection;
+import java.util.List;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+import javax.annotation.Nullable;
+
+import com.google.common.annotations.VisibleForTesting;
+import org.apache.commons.lang3.StringUtils;
+
+import org.apache.cassandra.cql3.CQL3Type;
+import org.apache.cassandra.cql3.functions.FunctionFactory;
+import org.apache.cassandra.cql3.functions.FunctionName;
+import org.apache.cassandra.cql3.functions.FunctionParameter;
+import org.apache.cassandra.cql3.functions.NativeFunction;
+import org.apache.cassandra.db.marshal.AbstractType;
+import org.apache.cassandra.db.marshal.Int32Type;
+import org.apache.cassandra.db.marshal.UTF8Type;
+import org.apache.cassandra.exceptions.InvalidRequestException;
+import org.apache.cassandra.transport.ProtocolVersion;
+
+/**
+ * A {@link MaskingFunction} applied to a {@link org.apache.cassandra.db.marshal.StringType} value that,
+ * depending on {@link Type}:
+ * <ul>
+ * <li>Replaces each character between the supplied positions by the supplied padding character. In other words,
+ * it will mask all the characters except the first m and last n.</li>
+ * <li>Replaces each character before and after the supplied positions by the supplied padding character. In other
+ * words, it will only mask all the first m and last n characters.</li>
+ * </ul>
+ * The returned value will allways be of the same type as the first string-based argument.
+ */
+public class PartialMaskingFunction extends MaskingFunction
+{
+    /** The character to be used as padding if no other character is supplied when calling the function. */
+    public static final char DEFAULT_PADDING_CHAR = '*';
+
+    /** The type of partial masking to perform, inner or outer. */
+    private final Type type;
+
+    /** The original type of the masked value. */
+    private final AbstractType<String> inputType;
+
+    /** Whether a padding argument hab been supplied. */
+    @Nullable
+    private final boolean hasPaddingArgument;
+
+    private PartialMaskingFunction(FunctionName name,
+                                   Type type,
+                                   AbstractType<String> inputType,
+                                   boolean hasPaddingArgument)
+    {
+        super(name, inputType, inputType, argumentsType(hasPaddingArgument));
+
+        this.type = type;
+        this.inputType = inputType;
+        this.hasPaddingArgument = hasPaddingArgument;
+    }
+
+    private static AbstractType<?>[] argumentsType(boolean hasPaddingArgument)
+    {
+        // The padding argument is optional, so we provide different signatures depending on whether it's present or not.
+        // Also, the padding argument should be a single character, but we don't have a data type for that, so we use
+        // a string-based argument. We will later validate on execution that the string argument is single-character.
+        return hasPaddingArgument
+               ? new AbstractType<?>[]{ Int32Type.instance, Int32Type.instance, UTF8Type.instance }
+               : new AbstractType<?>[]{ Int32Type.instance, Int32Type.instance };
+    }
+
+    @Override
+    public final ByteBuffer execute(ProtocolVersion protocolVersion, List<ByteBuffer> parameters)
+    {
+        // Parse the beginning and end positions. No validation is needed since the masker accepts negatives,
+        // but we should consider that the arguments migh be null.
+        int begin = parameters.get(1) == null ? 0 : Int32Type.instance.compose(parameters.get(1));
+        int end = parameters.get(2) == null ? 0 : Int32Type.instance.compose(parameters.get(2));
+
+        // Parse the padding character. The type of the argument is a string of any length because we don't have a
+        // character type in CQL, so we should verify that the passed string argument is single-character.
+        char padding = DEFAULT_PADDING_CHAR;
+        if (hasPaddingArgument && parameters.get(3) != null)
+        {
+            String parameter = UTF8Type.instance.compose(parameters.get(3));
+            if (parameter.length() != 1)
+            {
+                throw new InvalidRequestException(String.format("The padding argument for function %s should " +
+                                                                "be single-character, but '%s' has %d characters.",
+                                                                name(), parameter, parameter.length()));
+            }
+            padding = parameter.charAt(0);
+        }
+
+        // Null column values aren't masked
+        ByteBuffer value = parameters.get(0);
+        if (value == null)
+            return null;
+
+        // We mask the string representation of the column value, even if the type of the column is not a string.
+        String stringValue = inputType.compose(value);
+        String maskedValue = type.mask(stringValue, begin, end, padding);
+        return inputType.decompose(maskedValue);
+    }
+
+    public enum Type
+    {
+        /** Masks everything except the first {@code begin} and last {@code end} characters. */
+        INNER
+        {
+            @Override
+            protected boolean shouldMask(int pos, int begin, int end)
+            {
+                return pos >= begin && pos <= end;
+            }
+        },
+        /** Masks only the first {@code begin} and last {@code end} characters. */
+        OUTER
+        {
+            @Override
+            protected boolean shouldMask(int pos, int begin, int end)
+            {
+                return pos < begin || pos > end;
+            }
+        };
+
+        protected abstract boolean shouldMask(int pos, int begin, int end);
+
+        @VisibleForTesting
+        public String mask(String value, int begin, int end, char padding)
+        {
+            if (StringUtils.isEmpty(value))
+                return value;
+
+            int size = value.length();
+            int endIndex = size - 1 - end;
+            char[] chars = new char[size];
+
+            for (int i = 0; i < size; i++)
+            {
+                chars[i] = shouldMask(i, begin, endIndex) ? padding : value.charAt(i);
+            }
+
+            return new String(chars);
+        }
+    }
+
+    /** @return a collection of function factories to build new {@code PartialMaskingFunction} functions. */
+    public static Collection<FunctionFactory> factories()
+    {
+        return Stream.of(Type.values())
+                     .map(PartialMaskingFunction::factory)
+                     .collect(Collectors.toSet());
+    }
+
+    private static FunctionFactory factory(Type type)
+    {
+        return new MaskingFunction.Factory(type.name(),
+                                           FunctionParameter.string(),
+                                           FunctionParameter.fixed(CQL3Type.Native.INT),
+                                           FunctionParameter.fixed(CQL3Type.Native.INT),
+                                           FunctionParameter.optional(FunctionParameter.fixed(CQL3Type.Native.TEXT)))
+        {
+            @Override
+            @SuppressWarnings("unchecked")
+            protected NativeFunction doGetOrCreateFunction(List<AbstractType<?>> argTypes, AbstractType<?> receiverType)
+            {
+                AbstractType<String> inputType = (AbstractType<String>) argTypes.get(0);
+                return new PartialMaskingFunction(name, type, inputType, argTypes.size() == 4);
+            }
+        };
+    }
+}

src/java/org/apache/cassandra/cql3/functions/masking/ReplaceMaskingFunction.java

@@ -0,0 +1,72 @@
+/*
+ * 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.functions.masking;
+
+import java.nio.ByteBuffer;
+import java.util.List;
+
+import org.apache.cassandra.cql3.functions.FunctionFactory;
+import org.apache.cassandra.cql3.functions.FunctionName;
+import org.apache.cassandra.cql3.functions.FunctionParameter;
+import org.apache.cassandra.cql3.functions.NativeFunction;
+import org.apache.cassandra.db.marshal.AbstractType;
+import org.apache.cassandra.transport.ProtocolVersion;
+
+/**
+ * A {@link MaskingFunction} that replaces the specified column value by a certain replacement value.
+ * <p>
+ * The returned replacement value needs to have the same type as the replaced value.
+ * <p>
+ * For example, given a text column named "username", {@code mask_replace(username, '****')} will return {@code ****}.
+ */
+public class ReplaceMaskingFunction extends MaskingFunction
+{
+    public static final String NAME = "replace";
+
+    private ReplaceMaskingFunction(FunctionName name, AbstractType<?> replacedType, AbstractType<?> replacementType)
+    {
+        super(name, replacementType, replacedType, replacementType);
+    }
+
+    @Override
+    public final ByteBuffer execute(ProtocolVersion protocolVersion, List<ByteBuffer> parameters)
+    {
+        return parameters.get(1);
+    }
+
+    /** @return a {@link FunctionFactory} to build new {@link ReplaceMaskingFunction}s. */
+    public static FunctionFactory factory()
+    {
+        return new MaskingFunction.Factory(NAME, FunctionParameter.anyType(true), FunctionParameter.sameAsFirst())
+        {
+            @Override
+            protected NativeFunction doGetOrCreateFunction(List<AbstractType<?>> argTypes, AbstractType<?> receiverType)
+            {
+                AbstractType<?> replacedType = argTypes.get(0);
+                AbstractType<?> replacementType = argTypes.get(1);
+
+                assert replacedType == replacementType
+                : String.format("Both arguments should have the same type, but found %s(%s, %s)",
+                                name, replacedType.asCQL3Type(), replacementType.asCQL3Type());
+
+                return new ReplaceMaskingFunction(name, replacedType, replacementType);
+            }
+        };
+    }
+}

src/java/org/apache/cassandra/db/marshal/AbstractType.java

@@ -715,4 +715,12 @@ public abstract class AbstractType<T> implements Comparator<ByteBuffer>, Assignm
     {
         return this;
     }
+
+    /**
+     * @return A fixed, serialized value to be used when the column is masked, to be returned instead of the real value.
+     */
+    public ByteBuffer getMaskedValue()
+    {
+        throw new UnsupportedOperationException("There isn't a defined masked value for type " + asCQL3Type());
+    }
 }

src/java/org/apache/cassandra/db/marshal/AsciiType.java

@@ -38,6 +38,7 @@ import org.apache.cassandra.utils.ByteBufferUtil;
 public class AsciiType extends StringType
 {
     public static final AsciiType instance = new AsciiType();
+    private static final ByteBuffer MASKED_VALUE = instance.decompose("****");
 
     AsciiType() {super(ComparisonType.BYTE_ORDER);} // singleton
 
@@ -102,4 +103,10 @@ public class AsciiType extends StringType
     {
         return AsciiSerializer.instance;
     }
+
+    @Override
+    public ByteBuffer getMaskedValue()
+    {
+        return MASKED_VALUE;
+    }
 }

src/java/org/apache/cassandra/db/marshal/BooleanType.java

@@ -33,6 +33,8 @@ public class BooleanType extends AbstractType<Boolean>
 {
     public static final BooleanType instance = new BooleanType();
 
+    private static final ByteBuffer MASKED_VALUE = instance.decompose(false);
+
     BooleanType() {super(ComparisonType.CUSTOM);} // singleton
 
     public boolean isEmptyValueMeaningless()
@@ -116,4 +118,10 @@ public class BooleanType extends AbstractType<Boolean>
     {
         return 1;
     }
+
+    @Override
+    public ByteBuffer getMaskedValue()
+    {
+        return MASKED_VALUE;
+    }
 }

src/java/org/apache/cassandra/db/marshal/ByteType.java

@@ -36,6 +36,8 @@ public class ByteType extends NumberType<Byte>
 {
     public static final ByteType instance = new ByteType();
 
+    private static final ByteBuffer MASKED_VALUE = instance.decompose((byte) 0);
+
     ByteType()
     {
         super(ComparisonType.CUSTOM);
@@ -183,4 +185,10 @@ public class ByteType extends NumberType<Byte>
     {
         return ByteBufferUtil.clone(input);
     }
+
+    @Override
+    public ByteBuffer getMaskedValue()
+    {
+        return MASKED_VALUE;
+    }
 }

src/java/org/apache/cassandra/db/marshal/BytesType.java

@@ -33,6 +33,8 @@ public class BytesType extends AbstractType<ByteBuffer>
 {
     public static final BytesType instance = new BytesType();
 
+    private static final ByteBuffer MASKED_VALUE = ByteBufferUtil.EMPTY_BYTE_BUFFER;
+
     BytesType() {super(ComparisonType.BYTE_ORDER);} // singleton
 
     public ByteBuffer fromString(String source)
@@ -94,4 +96,10 @@ public class BytesType extends AbstractType<ByteBuffer>
     {
         return BytesSerializer.instance;
     }
+
+    @Override
+    public ByteBuffer getMaskedValue()
+    {
+        return MASKED_VALUE;
+    }
 }

src/java/org/apache/cassandra/db/marshal/CounterColumnType.java

@@ -32,6 +32,8 @@ public class CounterColumnType extends NumberType<Long>
 {
     public static final CounterColumnType instance = new CounterColumnType();
 
+    private static final ByteBuffer MASKED_VALUE = instance.decompose(0L);
+
     CounterColumnType() {super(ComparisonType.NOT_COMPARABLE);} // singleton
 
     public boolean isEmptyValueMeaningless()
@@ -158,4 +160,10 @@ public class CounterColumnType extends NumberType<Long>
     {
         return ByteBufferUtil.clone(input);
     }
+
+    @Override
+    public ByteBuffer getMaskedValue()
+    {
+        return MASKED_VALUE;
+    }
 }

src/java/org/apache/cassandra/db/marshal/DateType.java

@@ -46,6 +46,8 @@ public class DateType extends AbstractType<Date>
 
     public static final DateType instance = new DateType();
 
+    private static final ByteBuffer MASKED_VALUE = instance.decompose(new Date(0));
+
     DateType() {super(ComparisonType.BYTE_ORDER);} // singleton
 
     public boolean isEmptyValueMeaningless()
@@ -139,4 +141,10 @@ public class DateType extends AbstractType<Date>
     {
         return 8;
     }
+
+    @Override
+    public ByteBuffer getMaskedValue()
+    {
+        return MASKED_VALUE;
+    }
 }

src/java/org/apache/cassandra/db/marshal/DecimalType.java

@@ -42,6 +42,8 @@ import ch.obermuhlner.math.big.BigDecimalMath;
 public class DecimalType extends NumberType<BigDecimal>
 {
     public static final DecimalType instance = new DecimalType();
+
+    private static final ByteBuffer MASKED_VALUE = instance.decompose(BigDecimal.ZERO);
     private static final int MIN_SCALE = 32;
     private static final int MIN_SIGNIFICANT_DIGITS = MIN_SCALE;
     private static final int MAX_SCALE = 1000;
@@ -449,4 +451,10 @@ public class DecimalType extends NumberType<BigDecimal>
         return DecimalType.instance.decompose(
         toBigDecimal(input).setScale(0, RoundingMode.HALF_UP));
     }
+
+    @Override
+    public ByteBuffer getMaskedValue()
+    {
+        return MASKED_VALUE;
+    }
 }

src/java/org/apache/cassandra/db/marshal/DoubleType.java

@@ -35,6 +35,8 @@ public class DoubleType extends NumberType<Double>
 {
     public static final DoubleType instance = new DoubleType();
 
+    private static final ByteBuffer MASKED_VALUE = instance.decompose(0d);
+
     DoubleType() {super(ComparisonType.CUSTOM);} // singleton
 
     public boolean isEmptyValueMeaningless()
@@ -209,4 +211,10 @@ public class DoubleType extends NumberType<Double>
     {
         return ByteBufferUtil.bytes((double) Math.round(toDouble(input)));
     }
+
+    @Override
+    public ByteBuffer getMaskedValue()
+    {
+        return MASKED_VALUE;
+    }
 }

src/java/org/apache/cassandra/db/marshal/DurationType.java

@@ -36,6 +36,8 @@ public class DurationType extends AbstractType<Duration>
 {
     public static final DurationType instance = new DurationType();
 
+    private static final ByteBuffer MASKED_VALUE = instance.decompose(Duration.newInstance(0, 0, 0));
+
     DurationType()
     {
         super(ComparisonType.BYTE_ORDER);
@@ -86,4 +88,10 @@ public class DurationType extends AbstractType<Duration>
     {
         return true;
     }
+
+    @Override
+    public ByteBuffer getMaskedValue()
+    {
+        return MASKED_VALUE;
+    }
 }

src/java/org/apache/cassandra/db/marshal/EmptyType.java

@@ -182,4 +182,10 @@ public class EmptyType extends AbstractType<Void>
             super(message);
         }
     }
+
+    @Override
+    public ByteBuffer getMaskedValue()
+    {
+        return ByteBufferUtil.EMPTY_BYTE_BUFFER;
+    }
 }

src/java/org/apache/cassandra/db/marshal/FloatType.java

@@ -36,6 +36,8 @@ public class FloatType extends NumberType<Float>
 {
     public static final FloatType instance = new FloatType();
 
+    private static final ByteBuffer MASKED_VALUE = instance.decompose(0f);
+
     FloatType() {super(ComparisonType.CUSTOM);} // singleton
 
     public boolean isEmptyValueMeaningless()
@@ -204,4 +206,10 @@ public class FloatType extends NumberType<Float>
     {
         return ByteBufferUtil.bytes((float) Math.round(toFloat(input)));
     }
+
+    @Override
+    public ByteBuffer getMaskedValue()
+    {
+        return MASKED_VALUE;
+    }
 }

src/java/org/apache/cassandra/db/marshal/InetAddressType.java

@@ -18,6 +18,7 @@
 package org.apache.cassandra.db.marshal;
 
 import java.net.InetAddress;
+import java.net.InetSocketAddress;
 import java.nio.ByteBuffer;
 
 import org.apache.cassandra.cql3.CQL3Type;
@@ -33,6 +34,8 @@ public class InetAddressType extends AbstractType<InetAddress>
 {
     public static final InetAddressType instance = new InetAddressType();
 
+    private static final ByteBuffer MASKED_VALUE = instance.decompose(new InetSocketAddress(0).getAddress());
+
     InetAddressType() {super(ComparisonType.BYTE_ORDER);} // singleton
 
     public boolean isEmptyValueMeaningless()
@@ -94,4 +97,10 @@ public class InetAddressType extends AbstractType<InetAddress>
     {
         return InetAddressSerializer.instance;
     }
+
+    @Override
+    public ByteBuffer getMaskedValue()
+    {
+        return MASKED_VALUE;
+    }
 }

src/java/org/apache/cassandra/db/marshal/Int32Type.java

@@ -36,6 +36,8 @@ public class Int32Type extends NumberType<Integer>
 {
     public static final Int32Type instance = new Int32Type();
 
+    private static final ByteBuffer MASKED_VALUE = instance.decompose(0);
+
     Int32Type()
     {
         super(ComparisonType.CUSTOM);
@@ -204,4 +206,10 @@ public class Int32Type extends NumberType<Integer>
     {
         return ByteBufferUtil.clone(input);
     }
+
+    @Override
+    public ByteBuffer getMaskedValue()
+    {
+        return MASKED_VALUE;
+    }
 }

src/java/org/apache/cassandra/db/marshal/IntegerType.java

@@ -38,6 +38,8 @@ public final class IntegerType extends NumberType<BigInteger>
 {
     public static final IntegerType instance = new IntegerType();
 
+    private static final ByteBuffer MASKED_VALUE = instance.decompose(BigInteger.ZERO);
+
     // Constants or escaping values needed to encode/decode variable-length integers in our custom byte-ordered
     // encoding scheme.
     private static final int POSITIVE_VARINT_HEADER = 0x80;
@@ -603,4 +605,10 @@ public final class IntegerType extends NumberType<BigInteger>
     {
         return ByteBufferUtil.clone(input);
     }
+
+    @Override
+    public ByteBuffer getMaskedValue()
+    {
+        return MASKED_VALUE;
+    }
 }

src/java/org/apache/cassandra/db/marshal/LegacyTimeUUIDType.java

@@ -17,18 +17,28 @@
  */
 package org.apache.cassandra.db.marshal;
 
+import java.nio.ByteBuffer;
 import java.util.UUID;
 
 import org.apache.cassandra.serializers.TypeSerializer;
 import org.apache.cassandra.serializers.UUIDSerializer;
+import org.apache.cassandra.utils.TimeUUID;
 
 // Fully compatible with UUID, and indeed is interpreted as UUID for UDF
 public class LegacyTimeUUIDType extends AbstractTimeUUIDType<UUID>
 {
     public static final LegacyTimeUUIDType instance = new LegacyTimeUUIDType();
 
+    private static final ByteBuffer MASKED_VALUE = instance.decompose(TimeUUID.minAtUnixMillis(0).asUUID());
+
     public TypeSerializer<UUID> getSerializer()
     {
         return UUIDSerializer.instance;
     }
+
+    @Override
+    public ByteBuffer getMaskedValue()
+    {
+        return MASKED_VALUE;
+    }
 }

src/java/org/apache/cassandra/db/marshal/LexicalUUIDType.java

@@ -34,6 +34,8 @@ public class LexicalUUIDType extends AbstractType<UUID>
 {
     public static final LexicalUUIDType instance = new LexicalUUIDType();
 
+    private static final ByteBuffer MASKED_VALUE = instance.decompose(UUID.fromString("00000000-0000-0000-0000-000000000000"));
+
     LexicalUUIDType()
     {
         super(ComparisonType.CUSTOM);
@@ -131,4 +133,10 @@ public class LexicalUUIDType extends AbstractType<UUID>
     {
         return 16;
     }
+
+    @Override
+    public ByteBuffer getMaskedValue()
+    {
+        return MASKED_VALUE;
+    }
 }

src/java/org/apache/cassandra/db/marshal/ListType.java

@@ -250,4 +250,10 @@ public class ListType<T> extends CollectionType<List<T>>
     {
         serializer.forEach(input, action);
     }
+
+    @Override
+    public ByteBuffer getMaskedValue()
+    {
+        return decompose(Collections.emptyList());
+    }
 }

src/java/org/apache/cassandra/db/marshal/LongType.java

@@ -36,6 +36,8 @@ public class LongType extends NumberType<Long>
 {
     public static final LongType instance = new LongType();
 
+    private static final ByteBuffer MASKED_VALUE = instance.decompose(0L);
+
     LongType() {super(ComparisonType.CUSTOM);} // singleton
 
     public boolean isEmptyValueMeaningless()
@@ -228,4 +230,10 @@ public class LongType extends NumberType<Long>
     {
         return ByteBufferUtil.clone(input);
     }
+
+    @Override
+    public ByteBuffer getMaskedValue()
+    {
+        return MASKED_VALUE;
+    }
 }

src/java/org/apache/cassandra/db/marshal/MapType.java

@@ -20,6 +20,7 @@ package org.apache.cassandra.db.marshal;
 import java.nio.ByteBuffer;
 import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.Iterator;
 import java.util.List;
@@ -387,4 +388,10 @@ public class MapType<K, V> extends CollectionType<Map<K, V>>
     {
         throw new UnsupportedOperationException();
     }
+
+    @Override
+    public ByteBuffer getMaskedValue()
+    {
+        return decompose(Collections.emptyMap());
+    }
 }

src/java/org/apache/cassandra/db/marshal/ReversedType.java

@@ -218,4 +218,10 @@ public class ReversedType<T> extends AbstractType<T>
             return END_OF_STREAM;
         }
     }
+
+    @Override
+    public ByteBuffer getMaskedValue()
+    {
+        return baseType.getMaskedValue();
+    }
 }

src/java/org/apache/cassandra/db/marshal/SetType.java

@@ -233,4 +233,10 @@ public class SetType<T> extends CollectionType<Set<T>>
     {
         serializer.forEach(input, action);
     }
+
+    @Override
+    public ByteBuffer getMaskedValue()
+    {
+        return decompose(Collections.emptySet());
+    }
 }

src/java/org/apache/cassandra/db/marshal/ShortType.java

@@ -36,6 +36,8 @@ public class ShortType extends NumberType<Short>
 {
     public static final ShortType instance = new ShortType();
 
+    private static final ByteBuffer MASKED_VALUE = instance.decompose((short) 0);
+
     ShortType()
     {
         super(ComparisonType.CUSTOM);
@@ -180,4 +182,10 @@ public class ShortType extends NumberType<Short>
     {
         return ByteBufferUtil.clone(input);
     }
+
+    @Override
+    public ByteBuffer getMaskedValue()
+    {
+        return MASKED_VALUE;
+    }
 }

src/java/org/apache/cassandra/db/marshal/SimpleDateType.java

@@ -39,6 +39,8 @@ public class SimpleDateType extends TemporalType<Integer>
 {
     public static final SimpleDateType instance = new SimpleDateType();
 
+    private static final ByteBuffer MASKED_VALUE = instance.decompose(SimpleDateSerializer.timeInMillisToDay(0));
+
     SimpleDateType() {super(ComparisonType.BYTE_ORDER);} // singleton
 
     @Override
@@ -116,4 +118,10 @@ public class SimpleDateType extends TemporalType<Integer>
         if (!duration.hasDayPrecision())
             throw invalidRequest("The duration must have a day precision. Was: %s", duration);
     }
+
+    @Override
+    public ByteBuffer getMaskedValue()
+    {
+        return MASKED_VALUE;
+    }
 }

src/java/org/apache/cassandra/db/marshal/TimeType.java

@@ -39,6 +39,8 @@ import org.apache.cassandra.utils.bytecomparable.ByteSourceInverse;
 public class TimeType extends TemporalType<Long>
 {
     public static final TimeType instance = new TimeType();
+
+    private static final ByteBuffer DEFAULT_MASKED_VALUE = instance.decompose(0L);
     private TimeType() {super(ComparisonType.BYTE_ORDER);} // singleton
 
     public ByteBuffer fromString(String source) throws MarshalException
@@ -101,4 +103,10 @@ public class TimeType extends TemporalType<Long>
     {
         return decompose(LocalTime.now(ZoneOffset.UTC).toNanoOfDay());
     }
+
+    @Override
+    public ByteBuffer getMaskedValue()
+    {
+        return DEFAULT_MASKED_VALUE;
+    }
 }

src/java/org/apache/cassandra/db/marshal/TimeUUIDType.java

@@ -17,6 +17,8 @@
  */
 package org.apache.cassandra.db.marshal;
 
+import java.nio.ByteBuffer;
+
 import org.apache.cassandra.serializers.TypeSerializer;
 import org.apache.cassandra.utils.TimeUUID;
 
@@ -25,6 +27,8 @@ public class TimeUUIDType extends AbstractTimeUUIDType<TimeUUID>
 {
     public static final TimeUUIDType instance = new TimeUUIDType();
 
+    private static final ByteBuffer MASKED_VALUE = instance.decompose(TimeUUID.minAtUnixMillis(0));
+
     public TypeSerializer<TimeUUID> getSerializer()
     {
         return TimeUUID.Serializer.instance;
@@ -35,4 +39,10 @@ public class TimeUUIDType extends AbstractTimeUUIDType<TimeUUID>
     {
         return LegacyTimeUUIDType.instance;
     }
+
+    @Override
+    public ByteBuffer getMaskedValue()
+    {
+        return MASKED_VALUE;
+    }
 }

src/java/org/apache/cassandra/db/marshal/TimestampType.java

@@ -51,6 +51,8 @@ public class TimestampType extends TemporalType<Date>
 
     public static final TimestampType instance = new TimestampType();
 
+    private static final ByteBuffer MASKED_VALUE = instance.decompose(new Date(0));
+
     private TimestampType() {super(ComparisonType.CUSTOM);} // singleton
 
     public boolean isEmptyValueMeaningless()
@@ -170,4 +172,10 @@ public class TimestampType extends TemporalType<Date>
         if (!duration.hasMillisecondPrecision())
             throw invalidRequest("The duration must have a millisecond precision. Was: %s", duration);
     }
+
+    @Override
+    public ByteBuffer getMaskedValue()
+    {
+        return MASKED_VALUE;
+    }
 }

src/java/org/apache/cassandra/db/marshal/TupleType.java

@@ -549,4 +549,17 @@ public class TupleType extends AbstractType<ByteBuffer>
     {
         return getClass().getName() + TypeParser.stringifyTypeParameters(types, true);
     }
+
+    @Override
+    public ByteBuffer getMaskedValue()
+    {
+        ByteBuffer[] buffers = new ByteBuffer[types.size()];
+        for (int i = 0; i < types.size(); i++)
+        {
+            AbstractType<?> type = types.get(i);
+            buffers[i] = type.getMaskedValue();
+        }
+
+        return serializer.serialize(buildValue(buffers));
+    }
 }

src/java/org/apache/cassandra/db/marshal/UTF8Type.java

@@ -36,6 +36,8 @@ public class UTF8Type extends StringType
 {
     public static final UTF8Type instance = new UTF8Type();
 
+    private static final ByteBuffer MASKED_VALUE = instance.decompose("****");
+
     UTF8Type() {super(ComparisonType.BYTE_ORDER);} // singleton
 
     public ByteBuffer fromString(String source)
@@ -87,4 +89,10 @@ public class UTF8Type extends StringType
     {
         return UTF8Serializer.instance;
     }
+
+    @Override
+    public ByteBuffer getMaskedValue()
+    {
+        return MASKED_VALUE;
+    }
 }

src/java/org/apache/cassandra/db/marshal/UUIDType.java

@@ -49,6 +49,8 @@ public class UUIDType extends AbstractType<UUID>
 {
     public static final UUIDType instance = new UUIDType();
 
+    private static final ByteBuffer MASKED_VALUE = instance.decompose(UUID.fromString("00000000-0000-0000-0000-000000000000"));
+
     UUIDType()
     {
         super(ComparisonType.CUSTOM);
@@ -238,4 +240,10 @@ public class UUIDType extends AbstractType<UUID>
     {
         return 16;
     }
+
+    @Override
+    public ByteBuffer getMaskedValue()
+    {
+        return MASKED_VALUE;
+    }
 }

test/unit/org/apache/cassandra/cql3/CQLTester.java

@@ -2136,12 +2136,12 @@ public abstract class CQLTester
         return new UserTypeValue(fieldNames, fieldValues);
     }
 
-    protected Object list(Object...values)
+    protected List<Object> list(Object...values)
     {
         return Arrays.asList(values);
     }
 
-    protected Object set(Object...values)
+    protected Set<Object> set(Object...values)
     {
         return ImmutableSet.copyOf(values);
     }

test/unit/org/apache/cassandra/cql3/functions/masking/DefaultMaskingFunctionTest.java

@@ -0,0 +1,36 @@
+/*
+ * 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.functions.masking;
+
+import org.apache.cassandra.cql3.CQL3Type;
+
+import static java.lang.String.format;
+
+/**
+ * Tests for {@link DefaultMaskingFunction}.
+ */
+public class DefaultMaskingFunctionTest extends MaskingFunctionTester
+{
+    @Override
+    protected void testMaskingOnColumn(String name, CQL3Type type, Object value) throws Throwable
+    {
+        assertRows(execute(format("SELECT mask_default(%s) FROM %%s", name)),
+                   row(type.getType().getMaskedValue()));
+    }
+}

test/unit/org/apache/cassandra/cql3/functions/masking/HashMaskingFunctionTest.java

@@ -0,0 +1,91 @@
+/*
+ * 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.functions.masking;
+
+import java.nio.ByteBuffer;
+
+import org.junit.Assert;
+import org.junit.BeforeClass;
+
+import com.datastax.driver.core.ColumnDefinitions;
+import com.datastax.driver.core.DataType;
+import com.datastax.driver.core.ResultSet;
+import org.apache.cassandra.cql3.CQL3Type;
+import org.apache.cassandra.exceptions.InvalidRequestException;
+import org.apache.cassandra.schema.SchemaConstants;
+
+import static java.lang.String.format;
+
+/**
+ * Tests for {@link HashMaskingFunction}.
+ */
+public class HashMaskingFunctionTest extends MaskingFunctionTester
+{
+    @BeforeClass
+    public static void beforeClass()
+    {
+        requireNetwork();
+    }
+
+    @Override
+    protected void testMaskingOnColumn(String name, CQL3Type type, Object value) throws Throwable
+    {
+        ByteBuffer serializedValue = serializedValue(type, value);
+
+        // with default algorithm
+        assertRows(execute(format("SELECT mask_hash(%s) FROM %%s", name)),
+                   row(HashMaskingFunction.hash(HashMaskingFunction.messageDigest(HashMaskingFunction.DEFAULT_ALGORITHM),
+                                                serializedValue)));
+
+        // with null algorithm
+        assertRows(execute(format("SELECT mask_hash(%s, null) FROM %%s", name)),
+                   row(HashMaskingFunction.hash(HashMaskingFunction.messageDigest(HashMaskingFunction.DEFAULT_ALGORITHM),
+                                                serializedValue)));
+
+        // with manually specified algorithm
+        assertRows(execute(format("SELECT mask_hash(%s, 'SHA-512') FROM %%s", name)),
+                   row(HashMaskingFunction.hash(HashMaskingFunction.messageDigest("SHA-512"), serializedValue)));
+
+        // with not found ASCII algorithm
+        assertInvalidThrowMessage("Hash algorithm not found",
+                                  InvalidRequestException.class,
+                                  format("SELECT mask_hash(%s, 'unknown-algorithm') FROM %%s", name));
+
+        // with not found UTF-8 algorithm
+        assertInvalidThrowMessage("Hash algorithm not found",
+                                  InvalidRequestException.class,
+                                  format("SELECT mask_hash(%s, 'áéíóú') FROM %%s", name));
+
+        // test result set metadata, it should always be of type blob
+        ResultSet rs = executeNet(format("SELECT mask_hash(%s) FROM %%s", name));
+        ColumnDefinitions definitions = rs.getColumnDefinitions();
+        Assert.assertEquals(1, definitions.size());
+        Assert.assertEquals(DataType.blob(), definitions.getType(0));
+        Assert.assertEquals(format("%s.mask_hash(%s)", SchemaConstants.SYSTEM_KEYSPACE_NAME, name),
+                            definitions.getName(0));
+    }
+
+    private ByteBuffer serializedValue(CQL3Type type, Object value)
+    {
+        if (isNullOrEmptyMultiCell(type, value))
+            return null;
+
+        return makeByteBuffer(value, type.getType());
+    }
+}

test/unit/org/apache/cassandra/cql3/functions/masking/MaskingFunctionTester.java

@@ -0,0 +1,289 @@
+/*
+ * 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.functions.masking;
+
+import java.math.BigDecimal;
+import java.math.BigInteger;
+import java.net.InetSocketAddress;
+import java.util.Collection;
+import java.util.Date;
+import java.util.Map;
+import java.util.UUID;
+
+import com.google.common.collect.ImmutableList;
+import org.junit.Assert;
+import org.junit.Test;
+
+import org.apache.cassandra.cql3.CQL3Type;
+import org.apache.cassandra.cql3.CQLTester;
+import org.apache.cassandra.cql3.Duration;
+import org.apache.cassandra.db.marshal.AbstractType;
+import org.apache.cassandra.db.marshal.CollectionType;
+import org.apache.cassandra.db.marshal.Int32Type;
+import org.apache.cassandra.db.marshal.ListType;
+import org.apache.cassandra.db.marshal.MapType;
+import org.apache.cassandra.db.marshal.SetType;
+import org.apache.cassandra.db.marshal.TupleType;
+import org.apache.cassandra.db.marshal.UTF8Type;
+import org.apache.cassandra.db.marshal.UserType;
+import org.apache.cassandra.schema.KeyspaceMetadata;
+import org.apache.cassandra.schema.Schema;
+import org.apache.cassandra.serializers.SimpleDateSerializer;
+import org.apache.cassandra.utils.ByteBufferUtil;
+import org.apache.cassandra.utils.TimeUUID;
+
+import static java.lang.String.format;
+
+/**
+ * Abstract class for testing a specific implementation of {@link MaskingFunction}.
+ * <p>
+ * It tests the application of the function as defined by {@link #testMaskingOnColumn(String, CQL3Type, Object)}
+ * on all CQL data types on all the possible positions allowed for that type (primary key, regular and static columns).
+ */
+public abstract class MaskingFunctionTester extends CQLTester
+{
+    /**
+     * Tests the native masking function for all CQL native data types.
+     */
+    @Test
+    public void testMaskingOnNative() throws Throwable
+    {
+        for (CQL3Type.Native type : CQL3Type.Native.values())
+        {
+            switch (type)
+            {
+                case EMPTY:
+                    break;
+                case COUNTER:
+                    testMaskingOnCounterColumn(0L, -1L, 1L);
+                    break;
+                case TEXT:
+                case ASCII:
+                case VARCHAR:
+                    testMaskingOnAllColumns(type, "confidential");
+                    break;
+                case BOOLEAN:
+                    testMaskingOnAllColumns(type, true, false);
+                    break;
+                case TINYINT:
+                    testMaskingOnAllColumns(type, (byte) 0, (byte) 2);
+                    break;
+                case SMALLINT:
+                    testMaskingOnAllColumns(type, (short) 0, (short) 2);
+                    break;
+                case INT:
+                    testMaskingOnAllColumns(type, 2, Integer.MIN_VALUE, Integer.MAX_VALUE);
+                    break;
+                case BIGINT:
+                    testMaskingOnAllColumns(type, 2L, Long.MIN_VALUE, Long.MAX_VALUE);
+                    break;
+                case FLOAT:
+                    testMaskingOnAllColumns(type, 2.3f, Float.MIN_VALUE, Float.MAX_VALUE);
+                    break;
+                case DOUBLE:
+                    testMaskingOnAllColumns(type, 2.3d, Double.MIN_VALUE, Double.MAX_VALUE);
+                    break;
+                case VARINT:
+                    testMaskingOnAllColumns(type, BigInteger.valueOf(-1), BigInteger.valueOf(0), BigInteger.valueOf(1));
+                    break;
+                case DECIMAL:
+                    testMaskingOnAllColumns(type, BigDecimal.valueOf(2.3), BigDecimal.valueOf(0), BigDecimal.valueOf(-2.3));
+                    break;
+                case DATE:
+                    testMaskingOnAllColumns(type,
+                                            SimpleDateSerializer.timeInMillisToDay(2),
+                                            SimpleDateSerializer.timeInMillisToDay(Long.MAX_VALUE));
+                    break;
+                case DURATION:
+                    testMaskingOnNotKeyColumns(type, Duration.newInstance(1, 2, 3), Duration.newInstance(3, 2, 1));
+                    break;
+                case TIME:
+                    testMaskingOnAllColumns(CQL3Type.Native.TIME, 2L, (long) Integer.MAX_VALUE);
+                    break;
+                case TIMESTAMP:
+                    testMaskingOnAllColumns(CQL3Type.Native.TIMESTAMP, new Date(2), new Date(Integer.MAX_VALUE));
+                    break;
+                case UUID:
+                    testMaskingOnAllColumns(type, UUID.randomUUID());
+                    break;
+                case TIMEUUID:
+                    testMaskingOnAllColumns(type, TimeUUID.minAtUnixMillis(2), TimeUUID.minAtUnixMillis(Long.MAX_VALUE));
+                    break;
+                case INET:
+                    testMaskingOnAllColumns(type, new InetSocketAddress(0).getAddress());
+                    break;
+                case BLOB:
+                    testMaskingOnAllColumns(type, UTF8Type.instance.decompose("confidential"));
+                    break;
+                default:
+                    throw new AssertionError("Type " + type + " should be tested for masking functions");
+            }
+        }
+    }
+
+    /**
+     * Tests the native masking function for collections.
+     */
+    @Test
+    public void testMaskingOnCollection() throws Throwable
+    {
+        // set
+        Object[] values = new Object[]{ set(), set(1, 2, 3) };
+        testMaskingOnAllColumns(SetType.getInstance(Int32Type.instance, false).asCQL3Type(), values);
+        testMaskingOnNotKeyColumns(SetType.getInstance(Int32Type.instance, true).asCQL3Type(), values);
+
+        // list
+        values = new Object[]{ list(), list(1, 2, 3) };
+        testMaskingOnAllColumns(ListType.getInstance(Int32Type.instance, false).asCQL3Type(), values);
+        testMaskingOnNotKeyColumns(ListType.getInstance(Int32Type.instance, true).asCQL3Type(), values);
+
+        // map
+        values = new Object[]{ map(), map(1, 10, 2, 20, 3, 30) };
+        testMaskingOnAllColumns(MapType.getInstance(Int32Type.instance, Int32Type.instance, false).asCQL3Type(), values);
+        testMaskingOnNotKeyColumns(MapType.getInstance(Int32Type.instance, Int32Type.instance, true).asCQL3Type(), values);
+    }
+
+    /**
+     * Tests the native masking function for tuples.
+     */
+    @Test
+    public void testMaskingOnTuple() throws Throwable
+    {
+        testMaskingOnAllColumns(new TupleType(ImmutableList.of(Int32Type.instance, Int32Type.instance)).asCQL3Type(),
+                                tuple(1, 10), tuple(2, 20));
+    }
+
+    /**
+     * Tests the native masking function for UDTs.
+     */
+    @Test
+    public void testMaskingOnUDT() throws Throwable
+    {
+        String name = createType("CREATE TYPE %s (a int, b text)");
+
+        KeyspaceMetadata ks = Schema.instance.getKeyspaceMetadata(keyspace());
+        Assert.assertNotNull(ks);
+
+        UserType udt = ks.types.get(ByteBufferUtil.bytes(name)).orElseThrow(AssertionError::new);
+        Assert.assertNotNull(udt);
+
+        Object[] values = new Object[]{ userType("a", 1, "b", "Alice"), userType("a", 2, "b", "Bob") };
+        testMaskingOnNotKeyColumns(udt.asCQL3Type(), values);
+        testMaskingOnAllColumns(udt.freeze().asCQL3Type(), values);
+    }
+
+    /**
+     * Tests the native masking function for the specified column type and values on all possible types of column.
+     * That is, when the column is part of the primary key, or a regular column, or a static column.
+     *
+     * @param type  the type of the tested column
+     * @param values the values of the tested column
+     */
+    private void testMaskingOnAllColumns(CQL3Type type, Object... values) throws Throwable
+    {
+        createTable(format("CREATE TABLE %%s (pk %s, ck %<s, s %<s static, v %<s, PRIMARY KEY (pk, ck))", type));
+
+        for (Object value : values)
+        {
+            // Test null values
+            execute("INSERT INTO %s(pk, ck) VALUES (?, ?)", value, value);
+            testMaskingOnColumn("s", type, null);
+            testMaskingOnColumn("v", type, null);
+
+            // Test not-null values
+            execute("INSERT INTO %s(pk, ck, s, v) VALUES (?, ?, ?, ?)", value, value, value, value);
+            testMaskingOnColumn("pk", type, value);
+            testMaskingOnColumn("ck", type, value);
+            testMaskingOnColumn("s", type, value);
+            testMaskingOnColumn("v", type, value);
+
+            // Cleanup
+            execute("DELETE FROM %s WHERE pk=?", value);
+        }
+    }
+
+    /**
+     * Tests the native masking function for the specified column type and values when the column isn't part of the
+     * primary key. That is, when the column is either a regular column or a static column.
+     *
+     * @param type  the type of the tested column
+     * @param values the values of the tested column
+     */
+    private void testMaskingOnNotKeyColumns(CQL3Type type, Object... values) throws Throwable
+    {
+        createTable(format("CREATE TABLE %%s (pk int, ck int, s %s static, v %<s, PRIMARY KEY (pk, ck))", type));
+
+        // Test null values
+        execute("INSERT INTO %s(pk, ck) VALUES (0, 0)");
+        testMaskingOnColumn("s", type, null);
+        testMaskingOnColumn("v", type, null);
+
+        // Test not-null values
+        for (Object value : values)
+        {
+            execute("INSERT INTO %s(pk, ck, s, v) VALUES (0, 0, ?, ?)", value, value);
+            testMaskingOnColumn("s", type, value);
+            testMaskingOnColumn("v", type, value);
+        }
+    }
+
+    /**
+     * Tests the native masking function on the specified counter column values in all the positions where it can appear.
+     * That is, when the counter column is either a regular column or a static column.
+     *
+     * @param values the values of the tested counter column
+     */
+    private void testMaskingOnCounterColumn(Object... values) throws Throwable
+    {
+        createTable("CREATE TABLE %s (pk int, ck int, s counter static, v counter, PRIMARY KEY (pk, ck))");
+        for (Object value : values)
+        {
+            execute("UPDATE %s SET v = v + ?, s = s + ? WHERE pk = 0 AND ck = 0", value, value);
+            testMaskingOnColumn("s", CQL3Type.Native.COUNTER, value);
+            testMaskingOnColumn("v", CQL3Type.Native.COUNTER, value);
+            execute("TRUNCATE %s");
+        }
+    }
+
+    /**
+     * Tests the native masking function for the specified column type and value.
+     * This assumes that the table is already created.
+     *
+     * @param name  the name of the tested column
+     * @param type  the type of the tested column
+     * @param value the value of the tested column
+     */
+    protected abstract void testMaskingOnColumn(String name, CQL3Type type, Object value) throws Throwable;
+
+    protected boolean isNullOrEmptyMultiCell(CQL3Type type, Object value)
+    {
+        if (value == null)
+            return true;
+
+        AbstractType<?> dataType = type.getType();
+        if (dataType.isMultiCell() && dataType.isCollection())
+        {
+            return (((CollectionType<?>) dataType).kind == CollectionType.Kind.MAP)
+                   ? ((Map<?, ?>) value).isEmpty()
+                   : ((Collection<?>) value).isEmpty();
+        }
+
+        return false;
+    }
+}

test/unit/org/apache/cassandra/cql3/functions/masking/NullMaskingFunctionTest.java

@@ -0,0 +1,36 @@
+/*
+ * 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.functions.masking;
+
+import org.apache.cassandra.cql3.CQL3Type;
+
+import static java.lang.String.format;
+
+/**
+ * Tests for {@link NullMaskingFunction}.
+ */
+public class NullMaskingFunctionTest extends MaskingFunctionTester
+{
+    @Override
+    protected void testMaskingOnColumn(String name, CQL3Type type, Object value) throws Throwable
+    {
+        assertRows(execute(format("SELECT mask_null(%s) FROM %%s", name)),
+                   row((Object) null));
+    }
+}

test/unit/org/apache/cassandra/cql3/functions/masking/PartialMaskingFunctionTest.java

@@ -0,0 +1,177 @@
+/*
+ * 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.functions.masking;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+import com.datastax.driver.core.ColumnDefinitions;
+import com.datastax.driver.core.DataType;
+import com.datastax.driver.core.ResultSet;
+import org.apache.cassandra.cql3.CQL3Type;
+import org.apache.cassandra.db.marshal.StringType;
+import org.apache.cassandra.exceptions.InvalidRequestException;
+import org.apache.cassandra.schema.SchemaConstants;
+import org.assertj.core.api.Assertions;
+
+import static java.lang.String.format;
+
+/**
+ * Tests for {@link PartialMaskingFunction}.
+ */
+public class PartialMaskingFunctionTest extends MaskingFunctionTester
+{
+    @Override
+    protected void testMaskingOnColumn(String name, CQL3Type type, Object value) throws Throwable
+    {
+        testMaskingOnColumn(PartialMaskingFunction.Type.INNER, name, type, value);
+        testMaskingOnColumn(PartialMaskingFunction.Type.OUTER, name, type, value);
+    }
+
+    protected void testMaskingOnColumn(PartialMaskingFunction.Type masker, String name, CQL3Type type, Object value) throws Throwable
+    {
+        String functionName = SchemaConstants.SYSTEM_KEYSPACE_NAME + ".mask_" + masker.name().toLowerCase();
+
+        if (type.getType() instanceof StringType)
+        {
+            StringType stringType = (StringType) type.getType();
+            String stringValue = (String) value;
+
+            // ... with default padding
+            assertRows(execute(format("SELECT %s(%s, 1, 2) FROM %%s", functionName, name)),
+                       row(masker.mask(stringValue, 1, 2, PartialMaskingFunction.DEFAULT_PADDING_CHAR)));
+
+            // ... with manually specified ASCII padding
+            assertRows(execute(format("SELECT %s(%s, 1, 2, '#') FROM %%s", functionName, name)),
+                       row(masker.mask(stringValue, 1, 2, '#')));
+
+            // ... with manually specified UTF-8 padding
+            assertRows(execute(format("SELECT %s((text) %s, 1, 2, 'é') FROM %%s", functionName, name)),
+                       row(masker.mask(stringValue, 1, 2, 'é')));
+
+            // ... with not single-character padding
+            assertInvalidThrowMessage("should be single-character",
+                                      InvalidRequestException.class,
+                                      format("SELECT %s(%s, 1, 2, 'too_long') FROM %%s", functionName, name));
+
+            // ... with null padding
+            assertRows(execute(format("SELECT %s(%s, 1, 2, null) FROM %%s", functionName, name)),
+                       row(masker.mask(stringValue, 1, 2, PartialMaskingFunction.DEFAULT_PADDING_CHAR)));
+
+            // ... with null begin
+            assertRows(execute(format("SELECT %s(%s, null, 2) FROM %%s", functionName, name)),
+                       row(masker.mask(stringValue, 0, 2, PartialMaskingFunction.DEFAULT_PADDING_CHAR)));
+
+            // ... with null end
+            assertRows(execute(format("SELECT %s(%s, 1, null) FROM %%s", functionName, name)),
+                       row(masker.mask(stringValue, 1, 0, PartialMaskingFunction.DEFAULT_PADDING_CHAR)));
+
+            // test result set metadata, it should always be of type text, regardless of the type of the column
+            ResultSet rs = executeNet(format("SELECT %s(%s, 1, 2) FROM %%s", functionName, name));
+            ColumnDefinitions definitions = rs.getColumnDefinitions();
+            Assert.assertEquals(1, definitions.size());
+            Assert.assertEquals(driverDataType(stringType), definitions.getType(0));
+            Assert.assertEquals(format("%s(%s, 1, 2)", functionName, name), definitions.getName(0));
+        }
+        else
+        {
+            assertInvalidThrowMessage(format("Function %s requires an argument of type [text|varchar|ascii], " +
+                                             "but found argument %s of type %s",
+                                             functionName, name, type),
+                                      InvalidRequestException.class,
+                                      format("SELECT %s(%s, 1, 2) FROM %%s", functionName, name));
+        }
+    }
+
+    private static DataType driverDataType(StringType type)
+    {
+        switch ((CQL3Type.Native) type.asCQL3Type())
+        {
+            case ASCII:
+                return DataType.ascii();
+            case VARCHAR:
+                return DataType.varchar();
+            case TEXT:
+                return DataType.text();
+            default:
+                throw new AssertionError();
+        }
+    }
+
+    @Test
+    public void testMasking()
+    {
+        // null value
+        testMasking(null, 0, 0, '*', null, null);
+        testMasking(null, 9, 9, '*', null, null);
+        testMasking(null, 0, 0, '#', null, null);
+        testMasking(null, 9, 9, '#', null, null);
+
+        // empty value
+        testMasking("", 0, 0, '*', "", "");
+        testMasking("", 9, 9, '*', "", "");
+        testMasking("", 0, 0, '#', "", "");
+        testMasking("", 9, 9, '#', "", "");
+
+        // single-char value
+        testMasking("a", 0, 0, '*', "*", "a");
+        testMasking("a", 1, 1, '*', "a", "*");
+        testMasking("a", 10, 10, '*', "a", "*");
+        testMasking("a", 0, 0, '#', "#", "a");
+        testMasking("a", 1, 1, '#', "a", "#");
+        testMasking("a", 10, 10, '#', "a", "#");
+
+        // regular value
+        testMasking("abcde", 0, 0, '*', "*****", "abcde");
+        testMasking("abcde", 0, 1, '*', "****e", "abcd*");
+        testMasking("abcde", 0, 2, '*', "***de", "abc**");
+        testMasking("abcde", 0, 3, '*', "**cde", "ab***");
+        testMasking("abcde", 0, 4, '*', "*bcde", "a****");
+        testMasking("abcde", 0, 5, '*', "abcde", "*****");
+        testMasking("abcde", 0, 6, '*', "abcde", "*****");
+        testMasking("abcde", 1, 0, '*', "a****", "*bcde");
+        testMasking("abcde", 2, 0, '*', "ab***", "**cde");
+        testMasking("abcde", 3, 0, '*', "abc**", "***de");
+        testMasking("abcde", 4, 0, '*', "abcd*", "****e");
+        testMasking("abcde", 5, 0, '*', "abcde", "*****");
+        testMasking("abcde", 6, 0, '*', "abcde", "*****");
+        testMasking("abcde", 1, 1, '*', "a***e", "*bcd*");
+        testMasking("abcde", 2, 2, '*', "ab*de", "**c**");
+        testMasking("abcde", 3, 3, '*', "abcde", "*****");
+        testMasking("abcde", 4, 4, '*', "abcde", "*****");
+
+        // special characters
+        testMasking("á#íòü", 0, 0, '*', "*****", "á#íòü");
+        testMasking("á#íòü", 1, 1, '*', "á***ü", "*#íò*");
+        testMasking("á#íòü", 5, 5, '*', "á#íòü", "*****");
+    }
+
+    private static void testMasking(String unmaskedValue,
+                                    int begin,
+                                    int end,
+                                    char padding,
+                                    String innerMaskedValue,
+                                    String outerMaskedValue)
+    {
+        Assertions.assertThat(PartialMaskingFunction.Type.INNER.mask(unmaskedValue, begin, end, padding))
+                  .isIn(innerMaskedValue);
+        Assertions.assertThat(PartialMaskingFunction.Type.OUTER.mask(unmaskedValue, begin, end, padding))
+                  .isIn(outerMaskedValue);
+    }
+}

test/unit/org/apache/cassandra/cql3/functions/masking/ReplaceMaskingFunctionTest.java

@@ -0,0 +1,76 @@
+/*
+ * 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.functions.masking;
+
+import java.math.BigInteger;
+
+import org.junit.Test;
+
+import org.apache.cassandra.cql3.CQL3Type;
+import org.apache.cassandra.db.marshal.AbstractType;
+import org.apache.cassandra.exceptions.InvalidRequestException;
+
+import static java.lang.String.format;
+
+/**
+ * Tests for {@link ReplaceMaskingFunction}.
+ */
+public class ReplaceMaskingFunctionTest extends MaskingFunctionTester
+{
+    @Override
+    protected void testMaskingOnColumn(String name, CQL3Type type, Object value) throws Throwable
+    {
+        // null replacement argument
+        assertRows(execute(format("SELECT mask_replace(%s, ?) FROM %%s", name), (Object) null),
+                   row((Object) null));
+
+        // not-null replacement argument
+        AbstractType<?> t = type.getType();
+        Object replacementValue = t.compose(t.getMaskedValue());
+        String query = format("SELECT mask_replace(%s, ?) FROM %%s", name);
+        assertRows(execute(query, replacementValue), row(replacementValue));
+    }
+
+    @Test
+    public void testReplaceWithDifferentType() throws Throwable
+    {
+        createTable("CREATE TABLE %s (k int PRIMARY KEY, v varint)");
+        execute("INSERT INTO %s(k) VALUES (0)");
+        assertRows(execute("SELECT mask_replace(v, 1) FROM %s"), row(BigInteger.ONE));
+        assertRows(execute("SELECT mask_replace(v, (int) 1) FROM %s"), row(BigInteger.ONE));
+        assertRows(execute("SELECT mask_replace(v, (bigint) 1) FROM %s"), row(BigInteger.ONE));
+        assertRows(execute("SELECT mask_replace(v, (varint) 1) FROM %s"), row(BigInteger.ONE));
+        assertRows(execute("SELECT mask_replace(v, ?) FROM %s", 1), row(BigInteger.ONE));
+        assertRows(execute("SELECT mask_replace(v, ?) FROM %s", 1L), row(BigInteger.ONE));
+        assertRows(execute("SELECT mask_replace(v, ?) FROM %s", BigInteger.ONE), row(BigInteger.ONE));
+        assertInvalidThrowMessage("Type error: 1.2 cannot be passed as argument 1 of function system.mask_replace of type varint",
+                                  InvalidRequestException.class, "SELECT mask_replace(v, 1.2) FROM %s");
+        assertInvalidThrowMessage("Type error: 'secret' cannot be passed as argument 1 of function system.mask_replace of type varint",
+                                  InvalidRequestException.class, "SELECT mask_replace(v, 'secret') FROM %s");
+
+        createTable("CREATE TABLE %s (k int PRIMARY KEY, v varchar)");
+        execute("INSERT INTO %s(k) VALUES (0)");
+        assertRows(execute("SELECT mask_replace(v, 'secret') FROM %s"), row("secret"));
+        assertRows(execute("SELECT mask_replace(v, (ascii) 'secret') FROM %s"), row("secret"));
+        assertRows(execute("SELECT mask_replace(v, (text) 'secret') FROM %s"), row("secret"));
+        assertRows(execute("SELECT mask_replace(v, (varchar) 'secret') FROM %s"), row("secret"));
+        assertInvalidThrowMessage("Type error: 1 cannot be passed as argument 1 of function system.mask_replace of type text",
+                                  InvalidRequestException.class, "SELECT mask_replace(v, 1) FROM %s");
+    }
+}