class DateBinFunction extends Scalar<Long, Object> {

    public static final String NAME = "date_bin";

    public static void register(Functions.Builder module) {
        module.add(
            Signature.scalar(
                NAME,
                DataTypes.INTERVAL.getTypeSignature(),
                DataTypes.TIMESTAMPZ.getTypeSignature(), // source
                DataTypes.TIMESTAMPZ.getTypeSignature(), // origin
                DataTypes.TIMESTAMPZ.getTypeSignature()
            ).withFeatures(EnumSet.of(Feature.DETERMINISTIC, Feature.COMPARISON_REPLACEMENT, Feature.NULLABLE)),
            DateBinFunction::new);

        module.add(
            Signature.scalar(
                NAME,
                DataTypes.INTERVAL.getTypeSignature(),
                DataTypes.TIMESTAMP.getTypeSignature(), // source
                DataTypes.TIMESTAMP.getTypeSignature(), // origin
                DataTypes.TIMESTAMP.getTypeSignature()
            ).withFeatures(EnumSet.of(Feature.DETERMINISTIC, Feature.COMPARISON_REPLACEMENT, Feature.NULLABLE)),
            DateBinFunction::new);
    }

    private DateBinFunction(Signature signature, BoundSignature boundSignature) {
        super(signature, boundSignature);
    }

    @Override
    public Scalar<Long, Object> compile(List<Symbol> arguments, String currentUser, Roles roles) {
        assert arguments.size() == 3 : "Invalid number of arguments";

        if (arguments.get(0) instanceof Input<?> input) {
            var value = input.value();
            if (value != null) {
                Period p = (Period) value;
                checkMonthsAndYears(p);
                long intervalInMs = p.toStandardDuration().getMillis();
                if (intervalInMs == 0) {
                    throw new IllegalArgumentException("Interval cannot be zero");
                }
                return new CompiledDateBin(signature, boundSignature, intervalInMs);
            }
        }
        return this;
    }

    @Override
    @SafeVarargs
    public final Long evaluate(TransactionContext txnCtx, NodeContext nodeCtx, Input<Object> ... args) {<FILL_FUNCTION_BODY>}

    /**
     * Similar to the check called in {@link Period#toStandardDuration()} anyway,
     * but do it beforehand to provide a better error message.
     */
    private static void checkMonthsAndYears(Period p) {
        if (p.getMonths() != 0 || p.getYears() != 0) {
            throw new IllegalArgumentException("Cannot use intervals containing months or years");
        }
    }

    private static long getBinnedTimestamp(long interval, long timestamp, long origin) {
        if (interval == 0) {
            throw new IllegalArgumentException("Interval cannot be zero");
        }

        /*
         in Java % operator returns negative result only if dividend is negative.
         https://docs.oracle.com/javase/specs/jls/se8/html/jls-15.html#jls-15.17.3

         We need to shift timestamp by the timeline to the "earlier" direction, to the left.
         If diff is negative, remainder will be also negative (see link above), we subtract negative
         to move to right side of the bin and then subtract abs(interval) to move to beginning of the bin.
        */

        long diff = timestamp - origin;
        if (diff >= 0) {
            // diff % interval >= 0 regardless of the interval sign.
            return timestamp - diff % interval;
        } else {
            // diff % interval < 0 regardless of the interval sign.
            if (interval > 0) {
                return timestamp - diff % interval - interval;
            } else {
                return timestamp - diff % interval + interval;
            }
        }
    }

    private static class CompiledDateBin extends Scalar<Long, Object> {

        private final long intervalInMs;

        private CompiledDateBin(Signature signature, BoundSignature boundSignature, long intervalInMs) {
            super(signature, boundSignature);
            this.intervalInMs = intervalInMs;
        }

        @Override
        @SafeVarargs
        public final Long evaluate(TransactionContext txnCtx, NodeContext nodeContext, Input<Object>... args) {
            // Validation for arguments amount is done in compile.

            var timestamp = args[1].value();
            var origin = args[2].value();

            if (timestamp == null || origin == null) {
                return null;
            }

            return getBinnedTimestamp(intervalInMs, (long) timestamp, (long) origin);
        }
    }
}

        assert args.length == 3 : "Invalid number of arguments";

        var interval = args[0].value();
        var timestamp = args[1].value();
        var origin = args[2].value();

        if (interval == null || timestamp == null || origin == null) {
            return null;
        }

        Period p = (Period) interval;
        checkMonthsAndYears(p);
        return getBinnedTimestamp(p.toStandardDuration().getMillis(), (long) timestamp, (long) origin);
    
<methods>public BoundSignature boundSignature() ,public Scalar<java.lang.Long,java.lang.Object> compile(List<io.crate.expression.symbol.Symbol>, java.lang.String, io.crate.role.Roles) ,public transient abstract java.lang.Long evaluate(io.crate.metadata.TransactionContext, io.crate.metadata.NodeContext, Input<java.lang.Object>[]) ,public io.crate.expression.symbol.Symbol normalizeSymbol(io.crate.expression.symbol.Function, io.crate.metadata.TransactionContext, io.crate.metadata.NodeContext) ,public io.crate.metadata.functions.Signature signature() <variables>public static final Set<io.crate.metadata.Scalar.Feature> DETERMINISTIC_AND_COMPARISON_REPLACEMENT,public static final Set<io.crate.metadata.Scalar.Feature> DETERMINISTIC_ONLY,public static final Set<io.crate.metadata.Scalar.Feature> NO_FEATURES,protected final non-sealed BoundSignature boundSignature,protected final non-sealed io.crate.metadata.functions.Signature signature