class Limit extends ForwardingLogicalPlan {

    final Symbol limit;
    final Symbol offset;

    static LogicalPlan create(LogicalPlan source, @Nullable Symbol limit, @Nullable Symbol offset) {<FILL_FUNCTION_BODY>}

    public Limit(LogicalPlan source, Symbol limit, Symbol offset) {
        super(source);
        this.limit = limit;
        this.offset = offset;
    }

    public Symbol limit() {
        return limit;
    }

    public Symbol offset() {
        return offset;
    }

    @Override
    public ExecutionPlan build(DependencyCarrier executor,
                               PlannerContext plannerContext,
                               Set<PlanHint> planHints,
                               ProjectionBuilder projectionBuilder,
                               int limitHint,
                               int offsetHint,
                               @Nullable OrderBy order,
                               @Nullable Integer pageSizeHint,
                               Row params,
                               SubQueryResults subQueryResults) {
        int limit = Objects.requireNonNullElse(
            DataTypes.INTEGER.sanitizeValue(evaluate(
                plannerContext.transactionContext(),
                plannerContext.nodeContext(),
                this.limit,
                params,
                subQueryResults)),
            NO_LIMIT);
        int offset = Objects.requireNonNullElse(
            DataTypes.INTEGER.sanitizeValue(evaluate(
                plannerContext.transactionContext(),
                plannerContext.nodeContext(),
                this.offset,
                params,
                subQueryResults)),
            NO_OFFSET);

        ExecutionPlan executionPlan = source.build(
            executor, plannerContext, planHints, projectionBuilder, limit, offset, order, pageSizeHint, params, subQueryResults);
        List<DataType<?>> sourceTypes = Symbols.typeView(source.outputs());
        ResultDescription resultDescription = executionPlan.resultDescription();
        if (limit == NO_LIMIT && offset == NO_OFFSET) {
            return executionPlan;
        }
        if (resultDescription.hasRemainingLimitOrOffset()
            && (resultDescription.limit() != limit || resultDescription.offset() != offset)) {

            executionPlan = Merge.ensureOnHandler(executionPlan, plannerContext);
            resultDescription = executionPlan.resultDescription();
        }
        if (ExecutionPhases.executesOnHandler(plannerContext.handlerNode(), resultDescription.nodeIds())) {
            executionPlan.addProjection(
                new LimitAndOffsetProjection(limit, offset, sourceTypes), LimitAndOffset.NO_LIMIT, 0, resultDescription.orderBy());
        } else if (resultDescription.limit() != limit || resultDescription.offset() != 0) {
            executionPlan.addProjection(
                new LimitAndOffsetProjection(limit + offset, 0, sourceTypes), limit, offset, resultDescription.orderBy());
        }
        return executionPlan;
    }

    @Override
    public LogicalPlan replaceSources(List<LogicalPlan> sources) {
        return new Limit(Lists.getOnlyElement(sources), limit, offset);
    }

    @Override
    public @Nullable FetchRewrite rewriteToFetch(Collection<Symbol> usedColumns) {
        FetchRewrite fetchRewrite = source.rewriteToFetch(usedColumns);
        if (fetchRewrite == null) {
            return null;
        }
        return new FetchRewrite(
            fetchRewrite.replacedOutputs(),
            new Limit(fetchRewrite.newPlan(), this.limit, this.offset)
        );
    }

    @Override
    public Map<LogicalPlan, SelectSymbol> dependencies() {
        return source.dependencies();
    }


    @Override
    public String toString() {
        return "Limit{" +
               "source=" + source +
               ", limit=" + limit +
               ", offset=" + offset +
               '}';
    }

    @Override
    public <C, R> R accept(LogicalPlanVisitor<C, R> visitor, C context) {
        return visitor.visitLimit(this, context);
    }

    @Override
    public void print(PrintContext printContext) {
        printContext
            .text("Limit[")
            .text(limit.toString())
            .text(";")
            .text(offset.toString())
            .text("]");
        printStats(printContext);
        printContext.nest(source::print);
    }

    static int limitAndOffset(int limit, int offset) {
        if (limit == LimitAndOffset.NO_LIMIT) {
            return limit;
        }
        return limit + offset;
    }
}

        if (limit == null && offset == null) {
            return source;
        } else {
            return new Limit(
                source,
                Objects.requireNonNullElse(limit, Literal.of(-1L)),
                Objects.requireNonNullElse(offset, Literal.of(0)));
        }
    
<methods>public void <init>(io.crate.planner.operators.LogicalPlan) ,public Map<io.crate.planner.operators.LogicalPlan,io.crate.expression.symbol.SelectSymbol> dependencies() ,public List<io.crate.expression.symbol.Symbol> outputs() ,public io.crate.planner.operators.LogicalPlan pruneOutputsExcept(SequencedCollection<io.crate.expression.symbol.Symbol>) ,public List<io.crate.metadata.RelationName> relationNames() ,public io.crate.planner.operators.LogicalPlan source() ,public List<io.crate.planner.operators.LogicalPlan> sources() ,public boolean supportsDistributedReads() <variables>final non-sealed io.crate.planner.operators.LogicalPlan source,private final non-sealed List<io.crate.planner.operators.LogicalPlan> sources