/*
    *  Licensed to the Apache Software Foundation (ASF) under one
    *  or more contributor license agreements.  See the NOTICE file
    *  distributed with this work for additional information
    *  regarding copyright ownership.  The ASF licenses this file
    *  to you under the Apache License, Version 2.0 (the
    *  "License"); you may not use this file except in compliance
    *  with the License.  You may obtain a copy of the License at
    *  
   *    http://www.apache.org/licenses/LICENSE-2.0
   *  
   *  Unless required by applicable law or agreed to in writing,
   *  software distributed under the License is distributed on an
   *  "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
   *  KIND, either express or implied.  See the License for the
   *  specific language governing permissions and limitations
   *  under the License. 
   *  
   */
  package org.apache.directory.server.xdbm.search.impl;
  
  
  import java.io.IOException;
  import java.util.HashSet;
  import java.util.List;
  import java.util.Set;
  
  import org.apache.directory.api.ldap.model.constants.SchemaConstants;
  import org.apache.directory.api.ldap.model.cursor.Cursor;
  import org.apache.directory.api.ldap.model.exception.LdapException;
  import org.apache.directory.api.ldap.model.exception.LdapOtherException;
  import org.apache.directory.api.ldap.model.filter.AndNode;
  import org.apache.directory.api.ldap.model.filter.ApproximateNode;
  import org.apache.directory.api.ldap.model.filter.AssertionNode;
  import org.apache.directory.api.ldap.model.filter.BranchNode;
  import org.apache.directory.api.ldap.model.filter.EqualityNode;
  import org.apache.directory.api.ldap.model.filter.ExprNode;
  import org.apache.directory.api.ldap.model.filter.ExtensibleNode;
  import org.apache.directory.api.ldap.model.filter.GreaterEqNode;
  import org.apache.directory.api.ldap.model.filter.LeafNode;
  import org.apache.directory.api.ldap.model.filter.LessEqNode;
  import org.apache.directory.api.ldap.model.filter.NotNode;
  import org.apache.directory.api.ldap.model.filter.OrNode;
  import org.apache.directory.api.ldap.model.filter.PresenceNode;
  import org.apache.directory.api.ldap.model.filter.ScopeNode;
  import org.apache.directory.api.ldap.model.filter.SimpleNode;
  import org.apache.directory.api.ldap.model.filter.SubstringNode;
  import org.apache.directory.api.util.Strings;
  import org.apache.directory.server.core.api.partition.Partition;
  import org.apache.directory.server.core.api.partition.PartitionTxn;
  import org.apache.directory.server.i18n.I18n;
  import org.apache.directory.server.xdbm.Index;
  import org.apache.directory.server.xdbm.IndexNotFoundException;
  import org.apache.directory.server.xdbm.Store;
  import org.apache.directory.server.xdbm.search.Optimizer;
  
  
  /**
   * Optimizer that annotates the filter using scan counts.
   * 
   * @author <a href="mailto:dev@directory.apache.org">Apache Directory Project</a>
   */
  public class DefaultOptimizer implements Optimizer
  {
      /* Package protected*/ static final String CANDIDATES_ANNOTATION_KEY = "candidates";
      
      /* Package protected*/ static final String COUNT_ANNOTATION = "count"; 
  
      /** the database this optimizer operates on */
      private final Store db;
      private String contextEntryId;
  
  
      /**
       * Creates an optimizer on a database.
       *
       * @param db the database this optimizer works for.
       */
      public DefaultOptimizer( Store db )
      {
          this.db = db;
      }
  
  
      // This will suppress PMD.EmptyCatchBlock warnings in this method
      @SuppressWarnings("PMD.EmptyCatchBlock")
      private String getContextEntryId( PartitionTxn partitionTxn ) throws LdapException
      {
          if ( contextEntryId == null )
          {
              try
              {
                  this.contextEntryId = db.getEntryId( partitionTxn, ( ( Partition ) db ).getSuffixDn() );
              }
              catch ( Exception e )
              {
                  // might not have been created
              }
          }
 
         if ( contextEntryId == null )
         {
             return Partition.DEFAULT_ID;
         }
 
         return contextEntryId;
     }
 
 
     /**
      * Annotates the expression tree to determine optimal evaluation order based
      * on the scan count for indices that exist for each expression node.  If an
      * index on the attribute does not exist an IndexNotFoundException will be
      * thrown.
      *
      * {@inheritDoc}
      */
     @Override
     @SuppressWarnings("unchecked")
     public Long annotate( PartitionTxn partitionTxn, ExprNode node ) throws LdapException
     {
         // Start off with the worst case unless scan count says otherwise.
         Long count = Long.MAX_VALUE;
 
         /* --------------------------------------------------------------------
          *                 H A N D L E   L E A F   N O D E S          
          * --------------------------------------------------------------------
          * 
          * Each leaf node is based on an attribute and it represents a condition
          * that needs to be statisfied.  We ask the index (if one exists) for 
          * the attribute to give us a scan count of all the candidates that 
          * would satisfy the attribute assertion represented by the leaf node.
          * 
          * This is conducted differently based on the type of the leaf node.
          * Comments on each node type explain how each scan count is arrived at.
          */
 
         if ( node instanceof ScopeNode )
         {
             count = getScopeScan( partitionTxn, ( ScopeNode ) node );
         }
         else if ( node instanceof AssertionNode )
         {
             /* 
              * Leave it up to the assertion node to determine just how much it
              * will cost us.  Anyway it defaults to a maximum scan count if a
              * scan count is not specified by the implementation.
              */
         }
         else if ( node.isLeaf() )
         {
             LeafNode leaf = ( LeafNode ) node;
 
             try
             {  
                 if ( node instanceof PresenceNode )
                 {
                     count = getPresenceScan( partitionTxn, ( PresenceNode ) leaf );
                 }
                 else if ( node instanceof EqualityNode )
                 {
                     count = getEqualityScan( partitionTxn, ( EqualityNode ) leaf );
                 }
                 else if ( node instanceof GreaterEqNode )
                 {
                     count = getGreaterLessScan( partitionTxn, ( GreaterEqNode ) leaf, SimpleNode.EVAL_GREATER );
                 }
                 else if ( node instanceof LessEqNode )
                 {
                     count = getGreaterLessScan( partitionTxn, ( SimpleNode ) leaf, SimpleNode.EVAL_LESSER );
                 }
                 else if ( node instanceof SubstringNode )
                 {
                     /** Cannot really say so we presume the total index count */
                     count = getSubstringScan( partitionTxn, ( SubstringNode ) leaf );
                 }
                 else if ( node instanceof ExtensibleNode )
                 {
                     /** Cannot really say so we presume the total index count */
                     count = getFullScan( partitionTxn, leaf );
                 }
                 else if ( node instanceof ApproximateNode )
                 {
                     /** Feature not implemented so we just use equality matching */
                     count = getEqualityScan( partitionTxn, ( ApproximateNode ) leaf );
                 }
                 else
                 {
                     throw new IllegalArgumentException( I18n.err( I18n.ERR_711 ) );
                 }
             }
             catch ( IndexNotFoundException | IOException e )
             {
                 throw new LdapOtherException( e.getMessage(), e );
             }
         }
         // --------------------------------------------------------------------
         //                 H A N D L E   B R A N C H   N O D E S       
         // --------------------------------------------------------------------
         else
         {
             if ( node instanceof AndNode )
             {
                 count = getConjunctionScan( partitionTxn, ( AndNode ) node );
             }
             else if ( node instanceof OrNode )
             {
                 count = getDisjunctionScan( partitionTxn, ( OrNode ) node );
             }
             else if ( node instanceof NotNode )
             {
                 annotate( partitionTxn, ( ( NotNode ) node ).getFirstChild() );
 
                 /*
                  * A negation filter is always worst case since we will have
                  * to retrieve all entries from the master table then test
                  * each one against the negated child filter.  There is no way
                  * to use the indices.
                  */
                 count = Long.MAX_VALUE;
             }
             else
             {
                 throw new IllegalArgumentException( I18n.err( I18n.ERR_712 ) );
             }
         }
 
         // Protect against overflow when counting.
         if ( count < 0L )
         {
             count = Long.MAX_VALUE;
         }
 
         node.set( COUNT_ANNOTATION, count );
 
         return count;
     }
 
 
     /**
      * ANDs or Conjunctions take the count of the smallest child as their count.
      * This is the best that a conjunction can do and should be used rather than
      * the worst case. Notice that we annotate the child node with a recursive 
      * call before accessing its count parameter making the chain recursion 
      * depth first.
      *
      * @param node a AND (Conjunction) BranchNode
      * @return the calculated scan count
      * @throws Exception if there is an error
      */
     private long getConjunctionScan( PartitionTxn partitionTxn, BranchNode node ) throws LdapException
     {
         long count = Long.MAX_VALUE;
         List<ExprNode> children = node.getChildren();
 
         for ( ExprNode child : children )
         {
             if ( ( count == 1 ) && ( child instanceof ScopeNode ) )
             {
                 // We can stop here
                 break;
             }
 
             annotate( partitionTxn, child );
             count = Math.min( ( ( Long ) child.get( COUNT_ANNOTATION ) ), count );
 
             if ( count == 0 )
             {
                 // No need to continue
                 break;
             }
         }
 
         return count;
     }
 
 
     /**
      * Disjunctions (OR) are the union of candidates across all subexpressions 
      * so we add all the counts of the child nodes. Notice that we annotate the 
      * child node with a recursive call.
      *
      * @param node the OR branch node
      * @return the scan count on the OR node
      * @throws Exception if there is an error
      */
     private long getDisjunctionScan( PartitionTxn partitionTxn, BranchNode node ) throws LdapException
     {
         List<ExprNode> children = node.getChildren();
         long total = 0L;
 
         for ( ExprNode child : children )
         {
             annotate( partitionTxn, child );
             total += ( Long ) child.get( COUNT_ANNOTATION );
 
             if ( total == Long.MAX_VALUE )
             {
                 // We can stop here withoit evaluating the following filters
                 break;
             }
         }
 
         return total;
     }
 
 
     /**
      * Gets the worst case scan count for all entries that satisfy the equality
      * assertion in the SimpleNode argument.  
      *
      * @param node the node to get a scan count for 
      * @return the worst case
      * @throws Exception if there is an error accessing an index
      */
     @SuppressWarnings("unchecked")
     private <V> long getEqualityScan( PartitionTxn partitionTxn, SimpleNode<V> node ) throws LdapException, IndexNotFoundException, IOException
     {
         if ( db.hasIndexOn( node.getAttributeType() ) )
         {
             Index<V, String> idx = ( Index<V, String> ) db.getIndex( node.getAttributeType() );
 
             String normalizedKey;
             
             if ( node.getValue().isSchemaAware() )
             {
                 normalizedKey = node.getValue().getNormalized();
             }
             else
             {
                 normalizedKey = node.getAttributeType().getEquality().getNormalizer().normalize( node.getValue().getString() );
             }
             
             Cursor<String> result = idx.forwardValueCursor( partitionTxn, ( V ) normalizedKey );
             Set<String> values = new HashSet<>();
             int nbFound = 0;
 
             for ( String value : result )
             {
                 values.add( value );
                 nbFound++;
 
                 // Arbitrary stop gathering the candidates if we have more than 100
                 if ( nbFound == 100 )
                 {
                     break;
                 }
             }
 
             result.close();
 
             if ( nbFound < 100 )
             {
                 // Store the found candidates in the node
                 node.set( CANDIDATES_ANNOTATION_KEY, values );
 
                 return values.size();
             }
             else
             {
                 // Reset the candidates annotation
                 node.set( CANDIDATES_ANNOTATION_KEY, null );
 
                 return idx.count( partitionTxn, ( V ) node.getValue().getNormalized() );
             }
         }
 
         // count for non-indexed attribute is unknown so we presume da worst
         return Long.MAX_VALUE;
     }
 
 
     /**
      * Gets a scan count of the nodes that satisfy the greater or less than test
      * specified by the node.
      *
      * @param node the greater or less than node to get a count for 
      * @param isGreaterThan if true test is for >=, otherwise <=
      * @return the scan count of all nodes satisfying the Ava
      * @throws Exception if there is an error accessing an index
      */
     @SuppressWarnings("unchecked")
     private <V> long getGreaterLessScan( PartitionTxn partitionTxn, SimpleNode<V> node, boolean isGreaterThan ) throws LdapException, IndexNotFoundException
     {
         if ( db.hasIndexOn( node.getAttributeType() ) )
         {
             Index<V, String> idx = ( Index<V, String> ) db.getIndex( node.getAttributeType() );
 
             if ( isGreaterThan )
             {
                 return idx.greaterThanCount( partitionTxn, ( V ) node.getValue().getString() );
             }
             else
             {
                 return idx.lessThanCount( partitionTxn, ( V ) node.getValue().getString() );
             }
         }
 
         // count for non-indexed attribute is unknown so we presume da worst
         return Long.MAX_VALUE;
     }
 
 
     /**
      * Get a scan count based on a Substring node : we will count the entries that are greater
      * than ABC where the filter is (attr=ABC*). Any other filter won't be evaluated (for instance,
      * a filter like (attr=*ABC) will resolve to a full scan atm - we could have created a reverted
      * index for such a case -, and filters like (attr=*ABC*) also esolve to a full scan).
      * 
      * @param node The substring node
      * @return The number of candidates
      * @throws Exception If there is an error accessing an index
      */
     private long getSubstringScan( PartitionTxn partitionTxn, SubstringNode node ) throws LdapException, IndexNotFoundException
     {
         if ( db.hasIndexOn( node.getAttributeType() ) )
         {
             Index<String, String> idx = ( Index<String, String> ) db.getIndex( node.getAttributeType() );
 
             String initial = node.getInitial();
 
             if ( Strings.isEmpty( initial ) )
             {
                 // Not a (attr=ABC*) filter : full index scan
                 return idx.count( partitionTxn );
             }
             else
             {
                 return idx.greaterThanCount( partitionTxn, initial );
             }
         }
         else
         {
             // count for non-indexed attribute is unknown so we presume da worst
             return Long.MAX_VALUE;
         }
     }
 
 
     /**
      * Gets the total number of entries within the database index if one is 
      * available otherwise the count of all the entries within the database is
      * returned.
      *
      * @param node the leaf node to get a full scan count for 
      * @return the worst case full scan count
      * @throws Exception if there is an error access database indices
      */
     private long getFullScan( PartitionTxn partitionTxn, LeafNode node ) throws LdapException, IndexNotFoundException
     {
         if ( db.hasIndexOn( node.getAttributeType() ) )
         {
             Index<?, ?> idx = db.getIndex( node.getAttributeType() );
             return idx.count( partitionTxn );
         }
 
         return Long.MAX_VALUE;
     }
 
 
     /**
      * Gets the number of entries that would be returned by a presence node
      * assertion.  Leverages the presence system index for scan counts.
      *
      * @param node the presence node
      * @return the number of entries matched for the presence of an attribute
      * @throws Exception if errors result
      */
     private long getPresenceScan( PartitionTxn partitionTxn, PresenceNode node ) throws LdapException
     {
         if ( db.hasUserIndexOn( node.getAttributeType() )
              || node.getAttributeType().getOid().equals( SchemaConstants.ADMINISTRATIVE_ROLE_AT_OID ) )
         {
             Index<String, String> presenceIndex = db.getPresenceIndex();
 
             return presenceIndex.count( partitionTxn, node.getAttributeType().getOid() );
         }
         else if ( db.hasSystemIndexOn( node.getAttributeType() )
             || ( node.getAttributeType().getOid() == SchemaConstants.ENTRY_UUID_AT_OID ) )
         {
             // the system indices (objectClass, entryUUID and entryCSN) are maintained for
             // each entry, so we could just return the database count
             return db.count( partitionTxn );
         }
 
         return Long.MAX_VALUE;
     }
 
 
     /**
      * Gets the scan count for the scope node attached to this filter.
      *
      * @param node the ScopeNode
      * @return the scan count for scope
      * @throws Exception if any errors result
      */
     private long getScopeScan( PartitionTxn partitionTxn, ScopeNode node ) throws LdapException
     {
         String id = node.getBaseId();
 
         switch ( node.getScope() )
         {
             case OBJECT:
                 return 1L;
 
             case ONELEVEL:
                 return db.getChildCount( partitionTxn, id );
 
             case SUBTREE:
                 if ( id == getContextEntryId( partitionTxn ) )
                 {
                     return db.count( partitionTxn );
                 }
                 else
                 {
                     return db.getRdnIndex().reverseLookup( partitionTxn, id ).getNbDescendants() + 1L;
                 }
 
             default:
                 throw new IllegalArgumentException( I18n.err( I18n.ERR_713 ) );
         }
     }
 }