@Override
    public void addInput(Page page)
    {
        checkState(!finishing, "Operator is already finishing");
        boolean done = topNBuilder.processPage(requireNonNull(page, "page is null")).process();
        // there is no grouping so work will always be done
        verify(done);
        topNBuilder.updateMemoryReservations();
    }
@Test
    public void testExceedMemoryLimit()
            throws Exception
    {
        List<Page> input = rowPagesBuilder(BIGINT)
                .row(1L)
                .build();

        DriverContext smallDiverContext = createTaskContext(executor, scheduledExecutor, TEST_SESSION, new DataSize(1, BYTE))
                .addPipelineContext(0, true, true, false)
                .addDriverContext();

        TopNOperatorFactory operatorFactory = new TopNOperatorFactory(
                0,
                new PlanNodeId("test"),
                ImmutableList.of(BIGINT),
                100,
                ImmutableList.of(0),
                ImmutableList.of(ASC_NULLS_LAST));
        try (Operator operator = operatorFactory.createOperator(smallDiverContext)) {
            operator.addInput(input.get(0));
            fail("must fail because of exceeding local memory limit");
        }
        catch (ExceededMemoryLimitException ignore) {
        }
    }