@Override
    public Page getNextPage()
    {
        if (closed) {
            return null;
        }

        if (serverResponseIterator == null) {
            serverResponseIterator = queryPinot(split);
        }
        ByteBuffer byteBuffer = null;
        try {
            // Pinot gRPC server response iterator returns:
            //   - n data blocks based on inbound message size;
            //   - 1 metadata of the query results.
            // So we need to check ResponseType of each ServerResponse.
            if (serverResponseIterator.hasNext()) {
                long startTimeNanos = System.nanoTime();
                Server.ServerResponse serverResponse = serverResponseIterator.next();
                readTimeNanos += System.nanoTime() - startTimeNanos;
                final String responseType = serverResponse.getMetadataOrThrow("responseType");
                switch (responseType) {
                    case CommonConstants.Query.Response.ResponseType.DATA:
                        estimatedMemoryUsageInBytes = serverResponse.getSerializedSize();
                        // Store each dataTable which will later be constructed into Pages.
                        try {
                            byteBuffer = serverResponse.getPayload().asReadOnlyByteBuffer();
                            DataTable dataTable = DataTableFactory.getDataTable(byteBuffer);
                            checkExceptions(dataTable, split, PinotSessionProperties.isMarkDataFetchExceptionsAsRetriable(session));
                            currentDataTable = new PinotSegmentPageSource.PinotDataTableWithSize(dataTable, serverResponse.getSerializedSize());
                        }
                        catch (IOException e) {
                            throw new PinotException(
                                PINOT_DATA_FETCH_EXCEPTION,
                                split.getSegmentPinotQuery(),
                                String.format("Encountered Pinot exceptions when fetching data table from Split: < %s >", split),
                                e);
                        }
                        break;
                    case CommonConstants.Query.Response.ResponseType.METADATA:
                        // The last part of the response is Metadata
                        currentDataTable = null;
                        serverResponseIterator = null;
                        close();
                        return null;
                    default:
                        throw new PinotException(
                            PINOT_UNEXPECTED_RESPONSE,
                            split.getSegmentPinotQuery(),
                            String.format("Encountered Pinot exceptions, unknown response type - %s", responseType));
                }
            }
            Page page = fillNextPage();
            completedPositions += currentDataTable.getDataTable().getNumberOfRows();
            return page;
        }
        finally {
            if (byteBuffer != null) {
                ((Buffer) byteBuffer).clear();
            }
        }
    }
@Test
    public void testPrunedColumns()
    {
        PinotSessionProperties pinotSessionProperties = new PinotSessionProperties(pinotConfig);
        ConnectorSession session = new TestingConnectorSession(pinotSessionProperties.getSessionProperties());
        List<DataTable> dataTables = IntStream.range(0, 3).mapToObj(i -> createDataTableWithAllTypes()).collect(toImmutableList());
        List<PinotColumnHandle> expectedColumnHandles = createPinotColumnHandlesWithAllTypes();
        PinotSplit mockPinotSplit = new PinotSplit(pinotConnectorId.toString(), PinotSplit.SplitType.SEGMENT, expectedColumnHandles, Optional.empty(), Optional.of("blah"), ImmutableList.of("seg"), Optional.of("host"), getGrpcPort());

        ImmutableList.Builder<Integer> columnsSurvivingBuilder = ImmutableList.builder();
        for (int i = expectedColumnHandles.size() - 1; i >= 0; i--) {
            if (i % 2 == 0) {
                columnsSurvivingBuilder.add(i);
            }
        }
        List<Integer> columnsSurviving = columnsSurvivingBuilder.build();
        List<PinotColumnHandle> handlesSurviving = columnsSurviving.stream().map(expectedColumnHandles::get).collect(toImmutableList());
        PinotSegmentPageSource pinotSegmentPageSource = getPinotSegmentPageSource(session, dataTables, mockPinotSplit, handlesSurviving);

        for (int i = 0; i < dataTables.size(); ++i) {
            Page page = requireNonNull(pinotSegmentPageSource.getNextPage(), "Expected a valid page");
            Assert.assertEquals(page.getChannelCount(), columnsSurviving.size());
            for (int j = 0; j < columnsSurviving.size(); ++j) {
                Block block = page.getBlock(j);
                int originalColumnIndex = columnsSurviving.get(j);
                Type type = PinotColumnUtils.getPrestoTypeFromPinotType(getFieldSpec("dontcare", ALL_TYPES.get(originalColumnIndex)), false, false);
                long maxHashCode = Long.MIN_VALUE;
                for (int k = 0; k < NUM_ROWS; k++) {
                    maxHashCode = Math.max(type.hash(block, k), maxHashCode);
                }
                Assert.assertTrue(maxHashCode != 0, "Not all column values can have hash code 0");
            }
        }
    }