Trino UDF for generating row-level checksum

RowChecksumFunction.java

package io.trino.operator.scalar;

import com.google.common.collect.ImmutableList;
import io.airlift.slice.Slice;
import io.airlift.slice.Slices;
import io.trino.annotation.UsedByGeneratedCode;
import io.trino.metadata.SqlScalarFunction;
import io.trino.spi.block.Block;
import io.trino.spi.block.SqlRow;
import io.trino.spi.function.BoundSignature;
import io.trino.spi.function.FunctionMetadata;
import io.trino.spi.function.Signature;
import io.trino.spi.function.TypeVariableConstraint;
import io.trino.spi.type.Type;
import io.trino.spi.type.TypeOperators;
import io.trino.spi.type.TypeSignature;
import io.trino.spi.type.TypeSignatureParameter;
import io.trino.type.BlockTypeOperators;

import java.lang.invoke.MethodHandle;
import java.util.ArrayList;
import java.util.List;

import static io.trino.spi.function.InvocationConvention.InvocationArgumentConvention.NEVER_NULL;
import static io.trino.spi.function.InvocationConvention.InvocationReturnConvention.FAIL_ON_NULL;
import static io.trino.spi.type.VarbinaryType.VARBINARY;
import static io.trino.util.Reflection.methodHandle;

public class RowChecksumFunction extends SqlScalarFunction {

    private static final long PRIME64 = 0x9E3779B185EBCA87L;
    private final BlockTypeOperators blockTypeOperators;
    private static final MethodHandle METHOD_HANDLE = methodHandle(RowChecksumFunction.class, "rowToString", List.class, SqlRow.class);

    public RowChecksumFunction(TypeOperators typeOperators) {
        super(FunctionMetadata.scalarBuilder("row_checksum")
                .signature(Signature.builder()
                        .typeVariableConstraint(
                                TypeVariableConstraint.builder("T")
                                        .variadicBound("row")
                                        .build())
                        .returnType(VARBINARY)
                        .argumentType(new TypeSignature("T"))
                        .build())
                .description("Generate row-level checksum")
                .build());

        this.blockTypeOperators = new BlockTypeOperators(typeOperators);
    }

    @Override
    protected SpecializedSqlScalarFunction specialize(BoundSignature boundSignature)
    {
        Type type = boundSignature.getArgumentType(0);

        List<Type> fieldTypes = type.getTypeParameters();
        List<BlockTypeOperators.BlockPositionXxHash64> fieldHashOperators = new ArrayList<>(fieldTypes.size());
        List<TypeSignatureParameter> typeSignatureParameters = type.getTypeSignature().getParameters();
        for (int i = 0; i < fieldTypes.size(); i++) {
            fieldHashOperators.add(blockTypeOperators.getXxHash64Operator(type));
        }
        MethodHandle methodHandle = METHOD_HANDLE.bindTo(fieldHashOperators);

        return new ChoicesSpecializedSqlScalarFunction(
                boundSignature,
                FAIL_ON_NULL,
                ImmutableList.of(NEVER_NULL),
                methodHandle);
    }

    @UsedByGeneratedCode
    public static Slice rowToString(List<BlockTypeOperators.BlockPositionXxHash64> fieldHashOperators, SqlRow sqlRow)
    {
        try {
            int rawIndex = sqlRow.getRawIndex();
            long hash = 0;
            for (int i = 0; i < sqlRow.getFieldCount(); i++) {
                Block fieldBlock = sqlRow.getRawFieldBlock(i);
                if (fieldBlock.isNull(rawIndex)) {
                    hash = hash + PRIME64;
                } else {
                    hash = hash + fieldHashOperators.get(i).xxHash64(fieldBlock, rawIndex) * PRIME64;
                }
            }
            Slice value = Slices.allocate(Long.BYTES);
            value.setLong(0, hash);
            return value;
        }
        catch (RuntimeException e) {
            throw e;
        }
        catch (Exception e) {
            throw new RuntimeException(e);
        }
    }
}