code: @Override
    public org.apache.spark.unsafe.KVIterator<UnsafeRow, UnsafeRow> rowIterator() {
      return new org.apache.spark.unsafe.KVIterator<UnsafeRow, UnsafeRow>() {
        private final UnsafeRow key = new UnsafeRow(keySchema.length());
        private final UnsafeRow value = new UnsafeRow(valueSchema.length());
  
        private long offsetInPage = 0;
        private int recordsInPage = 0;
  
        private int currentklen;
        private int currentvlen;
        private int totalLength;
  
        private boolean initialized = false;
  
        private void init() {
          if (page != null) {
            offsetInPage = page.getBaseOffset();
            recordsInPage = numRows;
          }
          initialized = true;
        }
  
        @Override
        public boolean next() {
          if (!initialized) init();
          //searching for the next non empty page is records is now zero
          if (recordsInPage == 0) {
            freeCurrentPage();
            return false;
          }
  
          totalLength = Platform.getInt(base, offsetInPage) - 4;
          currentklen = Platform.getInt(base, offsetInPage + 4);
          currentvlen = totalLength - currentklen;
  
          key.pointTo(base, offsetInPage + 8, currentklen);
          value.pointTo(base, offsetInPage + 8 + currentklen, currentvlen);
  
          offsetInPage += 8 + totalLength + 8;
          recordsInPage -= 1;
          return true;
        }
  
        @Override
        public UnsafeRow getKey() {
          return key;
        }
  
        @Override
        public UnsafeRow getValue() {
          return value;
        }
  
        @Override
        public void close() {
          // do nothing
        }
  
        private void freeCurrentPage() {
          if (page != null) {
            freePage(page);
            page = null;
          }
        }
      };
    }
docs: Returns an iterator to go through all rows