core/src/main/java/com/orientechnologies/orient/core/storage/impl/local/OAbstractPaginatedStorage.java
public OBackgroundDelta recordsChangedAfterLSN(final OLogSequenceNumber lsn, final OCommandOutputListener outputListener) {
    final OLogSequenceNumber endLsn;
    // container of rids of changed records
    final SortedSet<ORID> sortedRids = new TreeSet<>();
    try {
      if (!configuration.getContextConfiguration().getValueAsBoolean(OGlobalConfiguration.STORAGE_TRACK_CHANGED_RECORDS_IN_WAL)) {
        throw new IllegalStateException(
            "Cannot find records which were changed starting from provided LSN because tracking of rids of changed records in WAL is switched off, "
                + "to switch it on please set property " + OGlobalConfiguration.STORAGE_TRACK_CHANGED_RECORDS_IN_WAL.getKey()
                + " to the true value, please note that only records"
                + " which are stored after this property was set will be retrieved");
      }

      stateLock.acquireReadLock();
      try {
        if (writeAheadLog == null) {
          return null;
        }

        // we iterate till the last record is contained in wal at the moment when we call this method
        endLsn = writeAheadLog.end();

        if (endLsn == null || lsn.compareTo(endLsn) > 0) {
          OLogManager.instance()
              .warn(this, "Cannot find requested LSN=%s for database sync operation. Last available LSN is %s", lsn, endLsn);
          return null;
        }

        if (lsn.equals(endLsn)) {
          // nothing has changed
          return new OBackgroundDelta(endLsn);
        }

        List<OWriteableWALRecord> records = writeAheadLog.next(lsn, 1);
        if (records.isEmpty()) {
          OLogManager.instance()
              .info(this, "Cannot find requested LSN=%s for database sync operation (last available LSN is %s)", lsn, endLsn);
          return null;
        }

        final OLogSequenceNumber freezeLsn = records.get(0).getLsn();

        writeAheadLog.addCutTillLimit(freezeLsn);
        try {
          records = writeAheadLog.next(lsn, 1_000);
          if (records.isEmpty()) {
            OLogManager.instance()
                .info(this, "Cannot find requested LSN=%s for database sync operation (last available LSN is %s)", lsn, endLsn);
            return null;
          }

          // all information about changed records is contained in atomic operation metadata
          long read = 0;

          readLoop:
          while (!records.isEmpty()) {
            for (final OWALRecord record : records) {
              final OLogSequenceNumber recordLSN = record.getLsn();

              if (endLsn.compareTo(recordLSN) >= 0) {
                if (record instanceof OFileCreatedWALRecord) {
                  throw new ODatabaseException(
                      "Cannot execute delta-sync because a new file has been added. Filename: '" + ((OFileCreatedWALRecord) record)
                          .getFileName() + "' (id=" + ((OFileCreatedWALRecord) record).getFileId() + ")");
                }

                if (record instanceof OFileDeletedWALRecord) {
                  throw new ODatabaseException(
                      "Cannot execute delta-sync because a file has been deleted. File id: " + ((OFileDeletedWALRecord) record)
                          .getFileId());
                }

                if (record instanceof OAtomicUnitEndRecord) {
                  final OAtomicUnitEndRecord atomicUnitEndRecord = (OAtomicUnitEndRecord) record;
                  if (atomicUnitEndRecord.getAtomicOperationMetadata().containsKey(ORecordOperationMetadata.RID_METADATA_KEY)) {
                    final ORecordOperationMetadata recordOperationMetadata = (ORecordOperationMetadata) atomicUnitEndRecord
                        .getAtomicOperationMetadata().get(ORecordOperationMetadata.RID_METADATA_KEY);
                    final Set<ORID> rids = recordOperationMetadata.getValue();
                    sortedRids.addAll(rids);
                  }
                }

                read++;

                if (outputListener != null) {
                  outputListener.onMessage("read " + read + " records from WAL and collected " + sortedRids.size() + " records");
                }
              } else {
                break readLoop;
              }
            }

            records = writeAheadLog.next(records.get(records.size() - 1).getLsn(), 1_000);
          }
        } finally {
          writeAheadLog.removeCutTillLimit(freezeLsn);
        }

      } catch (final IOException e) {
        throw OException.wrapException(new OStorageException("Error of reading of records changed after LSN " + lsn), e);
      } finally {
        stateLock.releaseReadLock();
      }
      OBackgroundDelta b = new OBackgroundDelta(this, outputListener, sortedRids, lsn, endLsn);

      return b;
    } catch (final RuntimeException e) {
      throw logAndPrepareForRethrow(e);
    } catch (final Error e) {
      throw logAndPrepareForRethrow(e);
    } catch (final Throwable t) {
      throw logAndPrepareForRethrow(t);
    }
  }
public OBackgroundDelta recordsChangedAfterLSN(final OLogSequenceNumber lsn, final OCommandOutputListener outputListener) {
    final OLogSequenceNumber endLsn;
    // container of rids of changed records
    final SortedSet<ORID> sortedRids = new TreeSet<>();
    try {
      if (!configuration.getContextConfiguration().getValueAsBoolean(OGlobalConfiguration.STORAGE_TRACK_CHANGED_RECORDS_IN_WAL)) {
        throw new IllegalStateException(
            "Cannot find records which were changed starting from provided LSN because tracking of rids of changed records in WAL is switched off, "
                + "to switch it on please set property " + OGlobalConfiguration.STORAGE_TRACK_CHANGED_RECORDS_IN_WAL.getKey()
                + " to the true value, please note that only records"
                + " which are stored after this property was set will be retrieved");
      }

      stateLock.acquireReadLock();
      try {
        if (writeAheadLog == null) {
          return null;
        }

        // we iterate till the last record is contained in wal at the moment when we call this method
        endLsn = writeAheadLog.end();

        if (endLsn == null || lsn.compareTo(endLsn) > 0) {
          OLogManager.instance()
              .warn(this, "Cannot find requested LSN=%s for database sync operation. Last available LSN is %s", lsn, endLsn);
          return null;
        }

        if (lsn.equals(endLsn)) {
          // nothing has changed
          return new OBackgroundDelta(endLsn);
        }

        List<OWriteableWALRecord> records = writeAheadLog.next(lsn, 1);
        if (records.isEmpty()) {
          OLogManager.instance()
              .info(this, "Cannot find requested LSN=%s for database sync operation (last available LSN is %s)", lsn, endLsn);
          return null;
        }

        final OLogSequenceNumber freezeLsn = records.get(0).getLsn();

        writeAheadLog.addCutTillLimit(freezeLsn);
        try {
          records = writeAheadLog.next(lsn, 1_000);
          if (records.isEmpty()) {
            OLogManager.instance()
                .info(this, "Cannot find requested LSN=%s for database sync operation (last available LSN is %s)", lsn, endLsn);
            return null;
          }

          // all information about changed records is contained in atomic operation metadata
          long read = 0;

          readLoop:
          while (!records.isEmpty()) {
            for (final OWALRecord record : records) {
              final OLogSequenceNumber recordLSN = record.getLsn();

              if (endLsn.compareTo(recordLSN) >= 0) {
                if (record instanceof OFileCreatedWALRecord) {
                  throw new ODatabaseException(
                      "Cannot execute delta-sync because a new file has been added. Filename: '" + ((OFileCreatedWALRecord) record)
                          .getFileName() + "' (id=" + ((OFileCreatedWALRecord) record).getFileId() + ")");
                }

                if (record instanceof OFileDeletedWALRecord) {
                  throw new ODatabaseException(
                      "Cannot execute delta-sync because a file has been deleted. File id: " + ((OFileDeletedWALRecord) record)
                          .getFileId());
                }

                if (record instanceof OAtomicUnitEndRecord) {
                  final OAtomicUnitEndRecord atomicUnitEndRecord = (OAtomicUnitEndRecord) record;
                  if (atomicUnitEndRecord.getAtomicOperationMetadata().containsKey(ORecordOperationMetadata.RID_METADATA_KEY)) {
                    final ORecordOperationMetadata recordOperationMetadata = (ORecordOperationMetadata) atomicUnitEndRecord
                        .getAtomicOperationMetadata().get(ORecordOperationMetadata.RID_METADATA_KEY);
                    final Set<ORID> rids = recordOperationMetadata.getValue();
                    sortedRids.addAll(rids);
                  }
                }

                read++;

                if (outputListener != null) {
                  outputListener.onMessage("read " + read + " records from WAL and collected " + sortedRids.size() + " records");
                }
              } else {
                break readLoop;
              }
            }

            records = writeAheadLog.next(records.get(records.size() - 1).getLsn(), 1_000);
          }
        } finally {
          writeAheadLog.removeCutTillLimit(freezeLsn);
        }

      } catch (final IOException e) {
        throw OException.wrapException(new OStorageException("Error of reading of records changed after LSN " + lsn), e);
      } finally {
        stateLock.releaseReadLock();
      }
      OBackgroundDelta b = new OBackgroundDelta(this, outputListener, sortedRids, lsn, endLsn);

      return b;
    } catch (final RuntimeException e) {
      throw logAndPrepareForRethrow(e);
    } catch (final Error e) {
      throw logAndPrepareForRethrow(e);
    } catch (final Throwable t) {
      throw logAndPrepareForRethrow(t);
    }
  }
public OBackgroundDelta recordsChangedAfterLSN ( final OLogSequenceNumber lsn , final OCommandOutputListener outputListener ) { final OLogSequenceNumber endLsn ; // container of rids of changed records final SortedSet < ORID > sortedRids = new TreeSet <> ( ) ; try { if ( ! configuration . getContextConfiguration ( ) . getValueAsBoolean ( OGlobalConfiguration . STORAGE_TRACK_CHANGED_RECORDS_IN_WAL ) ) { throw new IllegalStateException ( "Cannot find records which were changed starting from provided LSN because tracking of rids of changed records in WAL is switched off, " + "to switch it on please set property " + OGlobalConfiguration . STORAGE_TRACK_CHANGED_RECORDS_IN_WAL . getKey ( ) + " to the true value, please note that only records" + " which are stored after this property was set will be retrieved" ) ; } stateLock . acquireReadLock ( ) ; try { if ( writeAheadLog == null ) { return null ; } // we iterate till the last record is contained in wal at the moment when we call this method endLsn = writeAheadLog . end ( ) ; if ( endLsn == null || lsn . compareTo ( endLsn ) > 0 ) { OLogManager . instance ( ) . warn ( this , "Cannot find requested LSN=%s for database sync operation. Last available LSN is %s" , lsn , endLsn ) ; return null ; } if ( lsn . equals ( endLsn ) ) { // nothing has changed return new OBackgroundDelta ( endLsn ) ; } List < OWriteableWALRecord > records = writeAheadLog . next ( lsn , 1 ) ; if ( records . isEmpty ( ) ) { OLogManager . instance ( ) . info ( this , "Cannot find requested LSN=%s for database sync operation (last available LSN is %s)" , lsn , endLsn ) ; return null ; } final OLogSequenceNumber freezeLsn = records . get ( 0 ) . getLsn ( ) ; writeAheadLog . addCutTillLimit ( freezeLsn ) ; try { records = writeAheadLog . next ( lsn , 1_000 ) ; if ( records . isEmpty ( ) ) { OLogManager . instance ( ) . info ( this , "Cannot find requested LSN=%s for database sync operation (last available LSN is %s)" , lsn , endLsn ) ; return null ; } // all information about changed records is contained in atomic operation metadata long read = 0 ; readLoop : while ( ! records . isEmpty ( ) ) { for ( final OWALRecord record : records ) { final OLogSequenceNumber recordLSN = record . getLsn ( ) ; if ( endLsn . compareTo ( recordLSN ) >= 0 ) { if ( record instanceof OFileCreatedWALRecord ) { throw new ODatabaseException ( "Cannot execute delta-sync because a new file has been added. Filename: '" + ( ( OFileCreatedWALRecord ) record ) . getFileName ( ) + "' (id=" + ( ( OFileCreatedWALRecord ) record ) . getFileId ( ) + ")" ) ; } if ( record instanceof OFileDeletedWALRecord ) { throw new ODatabaseException ( "Cannot execute delta-sync because a file has been deleted. File id: " + ( ( OFileDeletedWALRecord ) record ) . getFileId ( ) ) ; } if ( record instanceof OAtomicUnitEndRecord ) { final OAtomicUnitEndRecord atomicUnitEndRecord = ( OAtomicUnitEndRecord ) record ; if ( atomicUnitEndRecord . getAtomicOperationMetadata ( ) . containsKey ( ORecordOperationMetadata . RID_METADATA_KEY ) ) { final ORecordOperationMetadata recordOperationMetadata = ( ORecordOperationMetadata ) atomicUnitEndRecord . getAtomicOperationMetadata ( ) . get ( ORecordOperationMetadata . RID_METADATA_KEY ) ; final Set < ORID > rids = recordOperationMetadata . getValue ( ) ; sortedRids . addAll ( rids ) ; } } read ++ ; if ( outputListener != null ) { outputListener . onMessage ( "read " + read + " records from WAL and collected " + sortedRids . size ( ) + " records" ) ; } } else { break readLoop ; } } records = writeAheadLog . next ( records . get ( records . size ( ) - 1 ) . getLsn ( ) , 1_000 ) ; } } finally { writeAheadLog . removeCutTillLimit ( freezeLsn ) ; } } catch ( final IOException e ) { throw OException . wrapException ( new OStorageException ( "Error of reading of records changed after LSN " + lsn ) , e ) ; } finally { stateLock . releaseReadLock ( ) ; } OBackgroundDelta b = new OBackgroundDelta ( this , outputListener , sortedRids , lsn , endLsn ) ; return b ; } catch ( final RuntimeException e ) { throw logAndPrepareForRethrow ( e ) ; } catch ( final Error e ) { throw logAndPrepareForRethrow ( e ) ; } catch ( final Throwable t ) { throw logAndPrepareForRethrow ( t ) ; } }
This method finds all the records which were updated starting from (but not including) current LSN and write result in provided
output stream. In output stream will be included all thw records which were updated/deleted/created since passed in LSN till
the current moment. Deleted records are written in output stream first, then created/updated records. All records are sorted by
record id. Data format: <ol> <li>Amount of records (single entry) - 8 bytes</li> <li>Record's cluster id - 4 bytes</li>
<li>Record's cluster position - 8 bytes</li> <li>Delete flag, 1 if record is deleted - 1 byte</li> <li>Record version , only
if record is not deleted - 4 bytes</li> <li>Record type, only if record is not deleted - 1 byte</li> <li>Length of binary
presentation of record, only if record is not deleted - 4 bytes</li> <li>Binary presentation of the record, only if record is
not deleted - length of content is provided in above entity</li> </ol>

@param lsn LSN from which we should find changed records

@see OGlobalConfiguration#STORAGE_TRACK_CHANGED_RECORDS_IN_WAL
This method finds all the records which were updated starting from ( but not including ) current LSN and write result in provided output stream . In output stream will be included all thw records which were updated / deleted / created since passed in LSN till the current moment . Deleted records are written in output stream first then created / updated records . All records are sorted by record id . Data format : <ol > <li > Amount of records ( single entry ) - 8 bytes< / li > <li > Record s cluster id - 4 bytes< / li > <li > Record s cluster position - 8 bytes< / li > <li > Delete flag 1 if record is deleted - 1 byte< / li > <li > Record version only if record is not deleted - 4 bytes< / li > <li > Record type only if record is not deleted - 1 byte< / li > <li > Length of binary presentation of record only if record is not deleted - 4 bytes< / li > <li > Binary presentation of the record only if record is not deleted - length of content is provided in above entity< / li > < / ol >