Try to merge events from the current batch to the last persisted batch
in order to reduce the number of smaller batches in the retention.

@return <code>true</code> if the merge operation is performed successfully
@throws IOException
357314
protected boolean mergeEventsToLastBatch() throws IOException {
        // Checks if we can merge _batch into _lastBatch
        if(_lastBatch != null && _eventBatchSize >= (_lastBatch.getSize() + _batch.getSize())) {
            _batch.setCompletionTime(System.currentTimeMillis());
            
            if(_flushListener != null) {
                _flushListener.beforeFlush(_batch);
            }
            
            // Creates a copy of the _lastBatch
            SimpleEventBatch<T> copy = ((SimpleEventBatch<T>)_lastBatch).clone();
            
            try {
                // Add events from _batch to the copy
                Iterator<Event<T>> iter = _batch.iterator();
                while(iter.hasNext()) {
                    copy.put(iter.next());
                }
                copy.setCompletionTime(_batch.getCompletionTime());
                
                /* Flush starts automatically upon adding _batch to BytesDB
                 * because the constructor sets update batchSize to 1.
                 */
                byte[] bytes = _eventBatchSerializer.serialize(copy);
                _store.set(_lastBatchCursor.getLookup(), bytes, getOffset());
            } catch (Exception e) {
                _logger.info("events merge aborted", e);
                return false;
            }
            
            if(_flushListener != null) {
                _flushListener.afterFlush(_batch);
            }
            
            _batchLock.lock();
            try {
                // Updated _lastBatch
                _lastBatch = copy;
                _lastBatchCursor.setHeader(copy.getHeader());
                _logger.info(_batch.getSize() + " events merged to EventBatch " + _lastBatchCursor.getLookup());
                
                // Create the next batch
                _batch = nextEventBatch(_batch.getOrigin() + _batch.getSize(), _batch.getMaxClock());
            } finally {
                _batchLock.unlock();
            }
            
            return true;
        }
        
        return false;
    }
1303455
@Override
    public boolean process(ContentEvent event) {
       
        if (event.isLastEvent() || 
                (instancesCount > 0 && instancesCount% this.sampleFrequency == 0)) {
            if (model.implementsMicroClusterer()) {

                Clustering clustering = model.getMicroClusteringResult();

                ClusteringResultContentEvent resultEvent = new ClusteringResultContentEvent(clustering, event.isLastEvent());

                this.outputStream.put(resultEvent);
            }
        }

        updateStats(event);
        return false;
    }
708260
public void flush() {
        List<BatchEntry> temp;

        bufferLock.lock();
        try {
            // Reset the last flush timestamp, even if the batch is empty or flush fails
            lastFlush = System.currentTimeMillis();

            // No-op if batch is empty
            if (batch == batchSize) {
                logger.trace("[{}] Batch empty, not flushing", name);
                return;
            }

            // Declare the batch empty, regardless of flush success/failure
            batch = batchSize;

            // If something goes wrong we still have a copy to recover.
            temp = buffer;
            buffer = new LinkedList<>();

        } finally {
            bufferLock.unlock();
        }

        long start = System.currentTimeMillis();
        try {
            flushTransactionLock.lock();
            start = System.currentTimeMillis();

            processBatch(temp);

            logger.trace("[{}] Batch flushed. Took {} ms, {} rows.", name, (System.currentTimeMillis() - start), temp.size());
        } catch (final Exception e) {
            if (this.maxFlushRetries > 0) {
                logger.warn(dev, "[{}] Error occurred while flushing. Retrying.", name, e);
            }

            boolean success = false;
            int retryCount;

            for (retryCount = 0; retryCount < this.maxFlushRetries && !success; retryCount++) {
                try {
                    Thread.sleep(this.flushRetryDelay);

                    // If the connection was established, we might need a rollback.
                    if (de.checkConnection() && de.isTransactionActive()) {
                        de.rollback();
                    }

                    processBatch(temp);

                    success = true;
                } catch (final InterruptedException ex) {
                    logger.debug("Interrupted while trying to flush batch. Stopping retries.");
                    Thread.currentThread().interrupt();
                    break;
                } catch (final Exception ex) {
                    logger.warn(dev, "[{}] Error occurred while flushing (retry attempt {}).", name, retryCount + 1, ex);
                }
            }

            if (!success) {
                try {
                    if (de.isTransactionActive()) {
                        de.rollback();
                    }
                } catch (final Exception ee) {
                    ee.addSuppressed(e);
                    logger.trace("[{}] Batch failed to check the flush transaction state", name, ee);
                }

                onFlushFailure(temp.toArray(new BatchEntry[temp.size()]));
                logger.error(dev, "[{}] Error occurred while flushing. Aborting batch flush.", name, e);
            } else {
                logger.trace("[{}] Batch flushed. Took {} ms, {} retries, {} rows.", name,
                        (System.currentTimeMillis() - start), retryCount, temp.size());
            }
        } finally {
            try {
                if (de.isTransactionActive()) {
                    de.rollback();
                }
            } catch (final Exception e) {
                logger.trace("[{}] Batch failed to check the flush transaction state", name, e);
            } finally {
                flushTransactionLock.unlock();
            }
        }
    }
479065
@Override
  public Deferred<Object> persist() {
    final byte[] q = Arrays.copyOfRange(batched_qualifier, 0, qualifier_index);
    final byte[] v = Arrays.copyOfRange(batched_value, 0, value_index);
    final byte[] r = Arrays.copyOfRange(row_key, 0, row_key.length);
    final long base_time = this.base_time; // shadow fixes issue #1436
    System.out.println(Arrays.toString(q) + "  " + Arrays.toString(v) + "  " + Arrays.toString(r));
    reset();
    return tsdb.put(r, q, v, base_time);
  }
1264624
@SuppressWarnings("unchecked")
    @Override
    public Future<?> runBatchAsync(Batch<T> batch, AfterExecute<T> afterExecute) {

        for (Map.Entry<Address, T> entry : batch.getMap().entrySet()) {
            Address address = entry.getKey();
            T opFromBatch = entry.getValue();

            BoxedByteArray mapKey;
            try {
                mapKey = new BoxedByteArray(address.toWriteKey(idService));
            } catch (Exception e) {
                throw new RuntimeException(e);
            }

            if (commitType == CommitType.READ_COMBINE_CAS) {
                int casRetriesRemaining = casRetries;
                do {
                    Optional<byte[]> oldBytes;
                    try {
                        oldBytes = getRaw(address);
                    } catch (Exception e) {
                        throw new RuntimeException(e);
                    }

                    T newOp;
                    if (oldBytes.isPresent()) {
                        T oldOp = deserializer.fromBytes(oldBytes.get());
                        newOp = (T) opFromBatch.add(oldOp);
                        if (log.isDebugEnabled()) {
                            log.debug("Combined " + oldOp + " and " + opFromBatch + " into " +
                                    newOp);
                        }
                    } else {
                        newOp = opFromBatch;
                    }

                    if (oldBytes.isPresent()) {
                        // Compare and swap, if the key is still mapped to the same byte array.
                        if (map.replace(mapKey, oldBytes.get(), newOp.serialize())) {
                            if (log.isDebugEnabled()) {
                                log.debug("Successful CAS overwrite at key " +
                                        Hex.encodeHexString(mapKey.bytes) + " with " +
                                        Hex.encodeHexString(newOp.serialize()));
                            }
                            break;
                        }
                    } else {
                        // Compare and swap, if the key is still absent from the map.
                        if (map.putIfAbsent(mapKey, newOp.serialize()) == null) {
                            // null is returned when there was no existing mapping for the 
                            // given key, which is the success case.
                            if (log.isDebugEnabled()) {
                                log.debug("Successful CAS insert without existing key for key " +
                                        Hex.encodeHexString(mapKey.bytes) + " with " +
                                        Hex.encodeHexString(newOp.serialize()));
                            }
                            break;
                        }
                    }
                } while (casRetriesRemaining-- > 0);
                if (casRetriesRemaining == -1) {
                    RuntimeException e = new RuntimeException("CAS retries exhausted");
                    afterExecute.afterExecute(e);
                    throw e;
                }
            } else if (commitType == CommitType.OVERWRITE) {
                map.put(mapKey, opFromBatch.serialize());
                if (log.isDebugEnabled()) {
                    log.debug("Write of key " + Hex.encodeHexString(mapKey.bytes));
                }
            } else {
                throw new AssertionError("Unsupported commit type: " + commitType);
            }
        }
        batch.reset();
        afterExecute.afterExecute(null); // null throwable => success
        return nullFuture;
    }
549094
public synchronized boolean update()
        throws IOException
    {
        if (_resource!=null && !_resource.exists())
        {
            clear();
            return true;
        }
        
        long lm=_resource.lastModified();
        
        if (lm==_lastModified && (_buf!=null || _list!=null))
            return false;
        _lastModified=lm;
        
        if (_resource.isDirectory())
            _list=_resource.list();

        if (_list==null)
        {
            int l=(int)_resource.length();
            if (l<0)
                l=1024;
            ByteArrayOutputStream2 bout = new ByteArrayOutputStream2(l);
            InputStream in = _resource.getInputStream();
            try{IO.copy(in,bout);} finally {in.close();}
            _buf=bout.getBuf();
            if (_buf.length!=l)
                _buf=bout.toByteArray();
        }
        return true;
    }
1298954
protected boolean consumeCompactionBatch() {
        /*
         * Consume one compaction update batch generated by the compactor.
         */
        CompactionUpdateBatch updateBatch = _compactor.pollCompactionBatch();
        if(updateBatch != null) {
            try {
                consumeCompaction(updateBatch);
            } catch (Exception e) {
                _log.error("failed to consume compaction batch " + updateBatch.getDescriptiveId(), e);
            } finally {
                _compactor.recycleCompactionBatch(updateBatch);
            }
            return true;
        }
        
        return false;
    }
116208
public void finish() throws IOException {
		boolean success = false;

		try {
			checkInProduceState();

			for (ResultSubpartition subpartition : subpartitions) {
				subpartition.finish();
			}

			success = true;
		}
		finally {
			if (success) {
				isFinished = true;

				notifyPipelinedConsumers();
			}
		}
	}
848310
function mergeBatchResults(ordered, batch, bulkResult, err, result) {
  // If we have an error set the result to be the err object
  if (err) {
    result = err;
  } else if (result && result.result) {
    result = result.result;
  } else if (result == null) {
    return;
  }

  // Do we have a top level error stop processing and return
  if (result.ok === 0 && bulkResult.ok === 1) {
    bulkResult.ok = 0;

    const writeError = {
      index: 0,
      code: result.code || 0,
      errmsg: result.message,
      op: batch.operations[0]
    };

    bulkResult.writeErrors.push(new WriteError(writeError));
    return;
  } else if (result.ok === 0 && bulkResult.ok === 0) {
    return;
  }

  // Deal with opTime if available
  if (result.opTime || result.lastOp) {
    const opTime = result.lastOp || result.opTime;
    let lastOpTS = null;
    let lastOpT = null;

    // We have a time stamp
    if (opTime && opTime._bsontype === 'Timestamp') {
      if (bulkResult.lastOp == null) {
        bulkResult.lastOp = opTime;
      } else if (opTime.greaterThan(bulkResult.lastOp)) {
        bulkResult.lastOp = opTime;
      }
    } else {
      // Existing TS
      if (bulkResult.lastOp) {
        lastOpTS =
          typeof bulkResult.lastOp.ts === 'number'
            ? Long.fromNumber(bulkResult.lastOp.ts)
            : bulkResult.lastOp.ts;
        lastOpT =
          typeof bulkResult.lastOp.t === 'number'
            ? Long.fromNumber(bulkResult.lastOp.t)
            : bulkResult.lastOp.t;
      }

      // Current OpTime TS
      const opTimeTS = typeof opTime.ts === 'number' ? Long.fromNumber(opTime.ts) : opTime.ts;
      const opTimeT = typeof opTime.t === 'number' ? Long.fromNumber(opTime.t) : opTime.t;

      // Compare the opTime's
      if (bulkResult.lastOp == null) {
        bulkResult.lastOp = opTime;
      } else if (opTimeTS.greaterThan(lastOpTS)) {
        bulkResult.lastOp = opTime;
      } else if (opTimeTS.equals(lastOpTS)) {
        if (opTimeT.greaterThan(lastOpT)) {
          bulkResult.lastOp = opTime;
        }
      }
    }
  }

  // If we have an insert Batch type
  if (batch.batchType === INSERT && result.n) {
    bulkResult.nInserted = bulkResult.nInserted + result.n;
  }

  // If we have an insert Batch type
  if (batch.batchType === REMOVE && result.n) {
    bulkResult.nRemoved = bulkResult.nRemoved + result.n;
  }

  let nUpserted = 0;

  // We have an array of upserted values, we need to rewrite the indexes
  if (Array.isArray(result.upserted)) {
    nUpserted = result.upserted.length;

    for (let i = 0; i < result.upserted.length; i++) {
      bulkResult.upserted.push({
        index: result.upserted[i].index + batch.originalZeroIndex,
        _id: result.upserted[i]._id
      });
    }
  } else if (result.upserted) {
    nUpserted = 1;

    bulkResult.upserted.push({
      index: batch.originalZeroIndex,
      _id: result.upserted
    });
  }

  // If we have an update Batch type
  if (batch.batchType === UPDATE && result.n) {
    const nModified = result.nModified;
    bulkResult.nUpserted = bulkResult.nUpserted + nUpserted;
    bulkResult.nMatched = bulkResult.nMatched + (result.n - nUpserted);

    if (typeof nModified === 'number') {
      bulkResult.nModified = bulkResult.nModified + nModified;
    } else {
      bulkResult.nModified = null;
    }
  }

  if (Array.isArray(result.writeErrors)) {
    for (let i = 0; i < result.writeErrors.length; i++) {
      const writeError = {
        index: batch.originalZeroIndex + result.writeErrors[i].index,
        code: result.writeErrors[i].code,
        errmsg: result.writeErrors[i].errmsg,
        op: batch.operations[result.writeErrors[i].index]
      };

      bulkResult.writeErrors.push(new WriteError(writeError));
    }
  }

  if (result.writeConcernError) {
    bulkResult.writeConcernErrors.push(new WriteConcernError(result.writeConcernError));
  }
}
1257667
void mergeRolledBackTransaction(Object[] list, int start, int limit) {

        for (int i = start; i < limit; i++) {
            RowAction rowact = (RowAction) list[i];

            if (rowact == null || rowact.type == RowActionBase.ACTION_NONE
                    || rowact.type == RowActionBase.ACTION_DELETE_FINAL) {
                continue;
            }

            Row row = rowact.memoryRow;

            if (row == null) {
                PersistentStore store =
                    rowact.session.sessionData.getRowStore(rowact.table);

                row = (Row) store.get(rowact.getPos(), false);
            }

            if (row == null) {
                continue;
            }

            synchronized (row) {
                rowact.mergeRollback(row);
            }
        }

//        } catch (Throwable t) {
//            System.out.println("throw in merge");
//            t.printStackTrace();
//        }
    }
281585
private CompletableFuture<SegmentProperties> mergeInStorage(SegmentMetadata transactionMetadata, MergeSegmentOperation mergeOp, TimeoutTimer timer) {
        return this.storage
                .getStreamSegmentInfo(transactionMetadata.getName(), timer.getRemaining())
                .thenAcceptAsync(transProperties -> {
                    // One last verification before the actual merger:
                    // Check that the Storage agrees with our metadata (if not, we have a problem ...)
                    if (transProperties.getLength() != transactionMetadata.getStorageLength()) {
                        throw new CompletionException(new DataCorruptionException(String.format(
                                "Transaction Segment '%s' cannot be merged into parent '%s' because its metadata disagrees with the Storage. Metadata.StorageLength=%d, Storage.StorageLength=%d",
                                transactionMetadata.getName(),
                                this.metadata.getName(),
                                transactionMetadata.getStorageLength(),
                                transProperties.getLength())));
                    }

                    if (transProperties.getLength() != mergeOp.getLength()) {
                        throw new CompletionException(new DataCorruptionException(String.format(
                                "Transaction Segment '%s' cannot be merged into parent '%s' because the declared length in the operation disagrees with the Storage. Operation.Length=%d, Storage.StorageLength=%d",
                                transactionMetadata.getName(),
                                this.metadata.getName(),
                                mergeOp.getLength(),
                                transProperties.getLength())));
                    }
                }, this.executor)
                .thenComposeAsync(v -> createSegmentIfNecessary(
                        () -> storage.concat(this.handle.get(), mergeOp.getStreamSegmentOffset(), transactionMetadata.getName(), timer.getRemaining()),
                        timer.getRemaining()), this.executor)
                .exceptionally(ex -> {
                    ex = Exceptions.unwrap(ex);
                    if (transactionMetadata.getLength() == 0
                            && ex instanceof StreamSegmentNotExistsException
                            && ((StreamSegmentNotExistsException) ex).getStreamSegmentName().equals(transactionMetadata.getName())) {
                        log.warn("{}: Not applying '{}' because source segment is missing (storage) and had no data.", this.traceObjectId, mergeOp);
                        return null;
                    } else {
                        throw new CompletionException(ex);
                    }
                })
                .thenComposeAsync(v -> storage.getStreamSegmentInfo(this.metadata.getName(), timer.getRemaining()), this.executor);
    }
583321
public boolean next() throws IOException {
		boolean hasNext = true;
		try {
			oldId = stream.readInt();
			oldText = stream.readUTFAsArray();
		} catch (EOFException e) {
			hasNext = false;
		}
		return hasNext;
	}
654099
protected final boolean mergeLists(List<UnclosedBranchDescriptor> child1open,
										List<UnclosedBranchDescriptor> child2open,
										List<UnclosedBranchDescriptor> result,
										boolean markJoinedBranchesAsPipelineBreaking) {

		//remove branches which have already been closed
		removeClosedBranches(child1open);
		removeClosedBranches(child2open);
		
		result.clear();
		
		// check how many open branches we have. the cases:
		// 1) if both are null or empty, the result is null
		// 2) if one side is null (or empty), the result is the other side.
		// 3) both are set, then we need to merge.
		if (child1open == null || child1open.isEmpty()) {
			if(child2open != null && !child2open.isEmpty()) {
				result.addAll(child2open);
			}
			return false;
		}
		
		if (child2open == null || child2open.isEmpty()) {
			result.addAll(child1open);
			return false;
		}

		int index1 = child1open.size() - 1;
		int index2 = child2open.size() - 1;
		
		boolean didCloseABranch = false;

		// as both lists (child1open and child2open) are sorted in ascending ID order
		// we can do a merge-join-like loop which preserved the order in the result list
		// and eliminates duplicates
		while (index1 >= 0 || index2 >= 0) {
			int id1 = -1;
			int id2 = index2 >= 0 ? child2open.get(index2).getBranchingNode().getId() : -1;

			while (index1 >= 0 && (id1 = child1open.get(index1).getBranchingNode().getId()) > id2) {
				result.add(child1open.get(index1));
				index1--;
			}
			while (index2 >= 0 && (id2 = child2open.get(index2).getBranchingNode().getId()) > id1) {
				result.add(child2open.get(index2));
				index2--;
			}

			// match: they share a common branching child
			if (id1 == id2) {
				didCloseABranch = true;
				
				// if this is the latest common child, remember it
				OptimizerNode currBanchingNode = child1open.get(index1).getBranchingNode();
				
				long vector1 = child1open.get(index1).getJoinedPathsVector();
				long vector2 = child2open.get(index2).getJoinedPathsVector();
				
				// check if this is the same descriptor, (meaning that it contains the same paths)
				// if it is the same, add it only once, otherwise process the join of the paths
				if (vector1 == vector2) {
					result.add(child1open.get(index1));
				}
				else {
					// we merge (re-join) a branch

					// mark the branch as a point where we break the pipeline
					if (markJoinedBranchesAsPipelineBreaking) {
						currBanchingNode.markAllOutgoingConnectionsAsPipelineBreaking();
					}

					if (this.hereJoinedBranches == null) {
						this.hereJoinedBranches = new ArrayList<OptimizerNode>(2);
					}
					this.hereJoinedBranches.add(currBanchingNode);

					// see, if this node closes the branch
					long joinedInputs = vector1 | vector2;

					// this is 2^size - 1, which is all bits set at positions 0..size-1
					long allInputs = (0x1L << currBanchingNode.getOutgoingConnections().size()) - 1;

					if (joinedInputs == allInputs) {
						// closed - we can remove it from the stack
						addClosedBranch(currBanchingNode);
					} else {
						// not quite closed
						result.add(new UnclosedBranchDescriptor(currBanchingNode, joinedInputs));
					}
				}

				index1--;
				index2--;
			}
		}

		// merged. now we need to reverse the list, because we added the elements in reverse order
		Collections.reverse(result);
		return didCloseABranch;
	}
731776
public synchronized boolean trim() throws IOException {
        boolean modified = false;

        for (Entry<String,RangeSet> e : new Hashtable<>(usages).entrySet()) {// copy because we mutate
            Job j = Jenkins.getInstance().getItemByFullName(e.getKey(),Job.class);
            if(j==null) {// no such job any more. recycle the record
                modified = true;
                usages.remove(e.getKey());
                continue;
            }

            Run firstBuild = j.getFirstBuild();
            if(firstBuild==null) {// no builds. recycle the whole record
                modified = true;
                usages.remove(e.getKey());
                continue;
            }

            RangeSet cur = e.getValue();

            // builds that are around without the keepLog flag on are normally clustered together (in terms of build #)
            // so our basic strategy is to discard everything up to the first ephemeral build, except those builds
            // that are marked as kept
            RangeSet kept = new RangeSet();
            Run r = firstBuild;
            while (r!=null && r.isKeepLog()) {
                kept.add(r.getNumber());
                r = r.getNextBuild();
            }

            if (r==null) {
                // all the build records are permanently kept ones, so we'll just have to keep 'kept' out of whatever currently in 'cur'
                modified |= cur.retainAll(kept);
            } else {
                // otherwise we are ready to discard [0,r.number) except those marked as 'kept'
                RangeSet discarding =  new RangeSet(new Range(-1,r.getNumber()));
                discarding.removeAll(kept);
                modified |= cur.removeAll(discarding);
            }

            if (cur.isEmpty()) {
                usages.remove(e.getKey());
                modified = true;
            }
        }

        if (modified) {
            if (logger.isLoggable(Level.FINE)) {
                logger.log(Level.FINE, "Saving trimmed {0}", getFingerprintFile(md5sum));
            }
            save();
        }

        return modified;
    }
506520
private boolean deleteBatch(List<String> objIDs) {
        if (objIDs.size() == 0) {
            return false;
        }
        m_logger.debug("Deleting batch of {} objects from {}", objIDs.size(), m_tableDef.getTableName());
        BatchObjectUpdater batchUpdater = new BatchObjectUpdater(m_tableDef);
        BatchResult batchResult = batchUpdater.deleteBatch(objIDs);
        if (batchResult.isFailed()) {
            m_logger.error("Batch query failed: {}", batchResult.getErrorMessage());
            return false;
        }
        return true;
    }
1104868
private boolean performFlushIfRequired() throws IOException {
        if (anyAreSet(state, FLUSHING_BUFFER)) {
            final ByteBuffer[] bufs = new ByteBuffer[additionalBuffer == null ? 1 : 2];
            long totalLength = 0;
            bufs[0] = currentBuffer.getBuffer();
            totalLength += bufs[0].remaining();
            if (additionalBuffer != null) {
                bufs[1] = additionalBuffer;
                totalLength += bufs[1].remaining();
            }
            if (totalLength > 0) {
                long total = 0;
                long res = 0;
                do {
                    res = next.write(bufs, 0, bufs.length);
                    total += res;
                    if (res == 0) {
                        return false;
                    }
                } while (total < totalLength);
            }
            additionalBuffer = null;
            currentBuffer.getBuffer().clear();
            state = state & ~FLUSHING_BUFFER;
        }
        return true;
    }
273711
public SQLMergeClause addBatch() {
        if (!configuration.getTemplates().isNativeMerge()) {
            throw new IllegalStateException("batch only supported for databases that support native merge");
        }

        batches.add(new SQLMergeBatch(keys, columns, values, subQuery));
        columns.clear();
        values.clear();
        keys.clear();
        subQuery = null;
        return this;
    }