package com.orientechnologies.orient.core.sql.executor;

import com.orientechnologies.common.concur.OTimeoutException;
import com.orientechnologies.orient.core.command.OCommandContext;

import java.util.Map;
import java.util.Optional;

/**
 * Deletes records coming from upstream steps
 *
 * @author Luigi Dell'Aquila (l.dellaquila-(at)-orientdb.com)
 */
public class DeleteStep extends AbstractExecutionStep {
  private long cost = 0;

  public DeleteStep(OCommandContext ctx, boolean profilingEnabled) {
    super(ctx, profilingEnabled);
  }

  @Override
  public OResultSet syncPull(OCommandContext ctx, int nRecords) throws OTimeoutException {
    OResultSet upstream = getPrev().get().syncPull(ctx, nRecords);
    return new OResultSet() {
      @Override
      public boolean hasNext() {
        return upstream.hasNext();
      }

      @Override
      public OResult next() {
        OResult result = upstream.next();
        long begin = profilingEnabled ? System.nanoTime() : 0;
        try {
          boolean newTx = false;
          if(result.isEdge() && !ctx.getDatabase().getTransaction().isActive()){
            newTx = true;
            ctx.getDatabase().begin();
          }
          try {
            if (result.isElement()) {
              result.getElement().get().delete();
            }
          }finally{
            if(newTx){
              ctx.getDatabase().commit();
            }
          }
          return result;
        } finally {
          if (profilingEnabled) {
            cost += (System.nanoTime() - begin);
          }
        }
      }

      @Override
      public void close() {
        upstream.close();
      }

      @Override
      public Optional<OExecutionPlan> getExecutionPlan() {
        return Optional.empty();
      }

      @Override
      public Map<String, Long> getQueryStats() {
        return null;
      }
    };
  }

  @Override
  public String prettyPrint(int depth, int indent) {
    String spaces = OExecutionStepInternal.getIndent(depth, indent);
    StringBuilder result = new StringBuilder();
    result.append(spaces);
    result.append("+ DELETE");
    if (profilingEnabled) {
      result.append(" (" + getCostFormatted() + ")");
    }
    return result.toString();
  }

  @Override
  public long getCost() {
    return cost;
  }

  @Override
  public OExecutionStep copy(OCommandContext ctx) {
    return new DeleteStep(ctx, this.profilingEnabled);
  }

  @Override
  public boolean canBeCached() {
    return true;
  }
}

@@ -1000,217 +1000,8 @@
 y %7B%0A
-          boolean newTx = false;%0A          if(result.isEdge() && !ctx.getDatabase().getTransaction().isActive())%7B%0A            newTx = true;%0A            ctx.getDatabase().begin();%0A          %7D%0A          try %7B%0A  
     
@@ -1028,26 +1028,24 @@
 lement()) %7B%0A
-  
             
@@ -1084,121 +1084,8 @@
 ();%0A
-            %7D%0A          %7Dfinally%7B%0A            if(newTx)%7B%0A              ctx.getDatabase().commit();%0A            %7D%0A
     

remove transaction from sql delete step, transaction is handled in the implementation