NdbQueryOperationImpl.hpp

/*
   Copyright (c) 2011, Oracle and/or its affiliates. All rights reserved.

   This program is free software; you can redistribute it and/or modify
   it under the terms of the GNU General Public License as published by
   the Free Software Foundation; version 2 of the License.

   This program is distributed in the hope that it will be useful,
   but WITHOUT ANY WARRANTY; without even the implied warranty of
   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
   GNU General Public License for more details.

   You should have received a copy of the GNU General Public License
   along with this program; if not, write to the Free Software
   Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA */

#ifndef NdbQueryOperationImpl_H
#define NdbQueryOperationImpl_H

#include "NdbQueryOperation.hpp"
#include "NdbQueryBuilderImpl.hpp"
#include "NdbIndexScanOperation.hpp"
#include <NdbError.hpp>
#include <ndb_limits.h>
#include <Vector.hpp>

// Forward declarations
class NdbTableImpl;
class NdbIndexImpl;
class NdbApiSignal;
class NdbResultStream;
class NdbParamOperand;
class NdbTransaction;
class NdbReceiver;
class NdbOut;
class NdbRootFragment;

class NdbBulkAllocator
{
public:
  explicit NdbBulkAllocator(size_t objSize);
  ~NdbBulkAllocator()
  { reset(); };

  int init(Uint32 maxObjs);

  void reset();

  void* allocObjMem(Uint32 noOfObjs);
private:
  static const char endMarker = -15;

  const size_t m_objSize;

  Uint32 m_maxObjs;

  char* m_buffer;

  Uint32 m_nextObjNo;

  // No copying.
  NdbBulkAllocator(const NdbBulkAllocator&);
  NdbBulkAllocator& operator= (const NdbBulkAllocator&);
};

class NdbQueryImpl {

  /* NdbQueryOperations are allowed to access it containing query */
  friend class NdbQueryOperationImpl;
  
  friend NdbOut& operator<<(NdbOut& out, const class NdbQueryOperationImpl&);

public:
  static NdbQueryImpl* buildQuery(NdbTransaction& trans, 
                                  const NdbQueryDefImpl& queryDef);

  Uint32 getNoOfOperations() const;

  Uint32 getNoOfLeafOperations() const;

  // Get a specific NdbQueryOperation instance by ident specified
  // when the NdbQueryOperationDef was created.
  NdbQueryOperationImpl& getQueryOperation(Uint32 ident) const;
  NdbQueryOperationImpl* getQueryOperation(const char* ident) const;
  // Consider to introduce these as convenient shortcuts
//NdbQueryOperationDefImpl& getQueryOperationDef(Uint32 ident) const;
//NdbQueryOperationDefImpl* getQueryOperationDef(const char* ident) const;

  Uint32 getNoOfParameters() const;
  const NdbParamOperand* getParameter(const char* name) const;
  const NdbParamOperand* getParameter(Uint32 num) const;

  NdbQuery::NextResultOutcome nextResult(bool fetchAllowed, bool forceSend);

  int close(bool forceSend);

  void release();

  NdbTransaction& getNdbTransaction() const
  { return m_transaction; }

  const NdbError& getNdbError() const;

  void setErrorCode(int aErrorCode);

  int assignParameters(const NdbQueryParamValue paramValues[]);

  int setBound(const NdbRecord *keyRecord,
               const NdbIndexScanOperation::IndexBound *bound);

  int prepareSend();

  int doSend(int aNodeId, bool lastFlag);

  NdbQuery& getInterface()
  { return m_interface; }

  NdbQueryImpl* getNext() const
  { return m_next; }

  void setNext(NdbQueryImpl* next)
  { m_next = next; }

  const NdbQueryDefImpl& getQueryDef() const
  {
    assert(m_queryDef);
    return *m_queryDef;
  }

  bool execTCKEYCONF();

  void execCLOSE_SCAN_REP(int errorCode, bool needClose);

  bool hasCompleted() const
  { return (m_state == Closed); 
  }
  
  void setStartIndicator()
  { 
    assert(!getQueryDef().isScanQuery());
    m_startIndicator = true; 
  }

  void setCommitIndicator()
  {  
    assert(!getQueryDef().isScanQuery());
    m_commitIndicator = true; 
  }

  int isPrunable(bool& pruned);

  Uint32 getRootFragCount() const
  { return m_rootFragCount; }

  NdbBulkAllocator& getResultStreamAlloc()
  { return m_resultStreamAlloc; }

  NdbBulkAllocator& getTupleSetAlloc()
  { return m_tupleSetAlloc; }

  NdbBulkAllocator& getRowBufferAlloc()
  { return m_rowBufferAlloc; }

private:
  enum FetchResult{
    FetchResult_gotError = -4,  // There is an error avail in 'm_error.code'
    FetchResult_sendFail = -3,
    FetchResult_nodeFail = -2,
    FetchResult_timeOut = -1,
    FetchResult_ok = 0,
    FetchResult_noMoreData = 1,
    FetchResult_noMoreCache = 2
  };

  class OrderedFragSet{
  public:
    // For calculating need for dynamically allocated memory.
    static const Uint32 pointersPerFragment = 2;

    explicit OrderedFragSet();

    ~OrderedFragSet(); 

    void prepare(NdbBulkAllocator& allocator,
                 NdbQueryOptions::ScanOrdering ordering, 
                 int capacity,  
                 const NdbRecord* keyRecord,
                 const NdbRecord* resultRecord);

    void prepareMoreResults(NdbRootFragment rootFrags[], Uint32 cnt);  // Need mutex lock

    NdbRootFragment* getCurrent() const;

    void reorganize();

    void clear();

    Uint32 getFetchMore(NdbRootFragment** &rootFrags);

  private:

    int m_capacity;
    int m_activeFragCount;
    int m_fetchMoreFragCount;
    int m_finalFragCount;
    NdbQueryOptions::ScanOrdering m_ordering;
    const NdbRecord* m_keyRecord;
    const NdbRecord* m_resultRecord;
    NdbRootFragment** m_activeFrags;
    NdbRootFragment** m_fetchMoreFrags;

    void add(NdbRootFragment& frag);

    int compare(const NdbRootFragment& frag1,
                const NdbRootFragment& frag2) const;

    bool verifySortOrder() const;

    // No copying.
    OrderedFragSet(const OrderedFragSet&);
    OrderedFragSet& operator=(const OrderedFragSet&);
  }; // class OrderedFragSet

  NdbQuery m_interface;

  enum {        // State of NdbQuery in API
    Initial,    // Constructed object, assiciated with a defined query
    Defined,    // Parameter values has been assigned
    Prepared,   // KeyInfo & AttrInfo prepared for execution
    Executing,  // Signal with exec. req. sent to TC
    EndOfData,  // All results rows consumed
    Closed,     // Query has been ::close()'ed 
    Failed,     
    Destructed
  } m_state;

  enum {        // Assumed state of query cursor in TC block
    Inactive,   // Execution not started at TC
    Active
  } m_tcState;

  NdbQueryImpl* m_next;
  const NdbQueryDefImpl* m_queryDef;

  NdbError m_error;

  int m_errorReceived;   // BEWARE: protect with PollGuard mutex

  NdbTransaction& m_transaction;

  NdbTransaction* m_scanTransaction;

  NdbQueryOperationImpl *m_operations;  // 'Array of ' OperationImpls
  Uint32 m_countOperations;             // #elements in above array

  Uint32 m_globalCursor;

  Uint32 m_pendingFrags;  // BEWARE: protect with PollGuard mutex

  Uint32 m_rootFragCount;

  NdbRootFragment* m_rootFrags;

  OrderedFragSet m_applFrags;

  Uint32 m_finalBatchFrags; // BEWARE: protect with PollGuard mutex

  Uint32 m_num_bounds;

  Uint32 m_shortestBound;

  Uint32Buffer m_attrInfo;  // ATTRINFO: QueryTree + serialized parameters
  Uint32Buffer m_keyInfo;   // KEYINFO:  Lookup key or scan bounds

  bool m_startIndicator;

  bool m_commitIndicator;

  enum {
    Prune_Unknown, 
    Prune_Yes, // The root is a prunable range scan.
    Prune_No   // The root is not a prunable range scan.
  } m_prunability;

  Uint32 m_pruneHashVal;

  NdbBulkAllocator m_operationAlloc;

  NdbBulkAllocator m_tupleSetAlloc;

  NdbBulkAllocator m_resultStreamAlloc;

  NdbBulkAllocator m_pointerAlloc;

  NdbBulkAllocator m_rowBufferAlloc;

  // Only constructable from factory ::buildQuery();
  explicit NdbQueryImpl(
             NdbTransaction& trans,
             const NdbQueryDefImpl& queryDef);

  ~NdbQueryImpl();

  void postFetchRelease();

  NdbQuery::NextResultOutcome nextRootResult(bool fetchAllowed, bool forceSend);

  int sendFetchMore(NdbRootFragment* rootFrags[], Uint32 cnt,
                    bool forceSend);

  FetchResult awaitMoreResults(bool forceSend);

  bool hasReceivedError();                                   // Need mutex lock

  void setFetchTerminated(int aErrorCode, bool needClose);   // Need mutex lock

  int closeTcCursor(bool forceSend);

  int sendClose(int nodeId);

  const NdbQuery& getInterface() const
  { return m_interface; }

  NdbQueryOperationImpl& getRoot() const 
  { return getQueryOperation(0U); }

  bool handleBatchComplete(NdbRootFragment& rootFrag);

  NdbBulkAllocator& getPointerAlloc()
  { return m_pointerAlloc; }

}; // class NdbQueryImpl


class NdbQueryOperationImpl {

  friend NdbOut& operator<<(NdbOut& out, const NdbQueryOperationImpl&);

  friend class NdbQueryImpl;

public:
  Uint32 getNoOfParentOperations() const;
  NdbQueryOperationImpl& getParentOperation(Uint32 i) const;
  NdbQueryOperationImpl* getParentOperation() const;

  Uint32 getNoOfChildOperations() const;
  NdbQueryOperationImpl& getChildOperation(Uint32 i) const;

  NdbQueryOperationImpl& getRoot() const
  { return m_queryImpl.getRoot(); }

  const NdbQueryDefImpl& getQueryDef() const
  { return m_queryImpl.getQueryDef(); }

  const NdbQueryOperationDefImpl& getQueryOperationDef() const
  { return m_operationDef; }

  // Get the entire query object which this operation is part of
  NdbQueryImpl& getQuery() const
  { return m_queryImpl; }

  NdbRecAttr* getValue(const char* anAttrName, char* resultBuffer);
  NdbRecAttr* getValue(Uint32 anAttrId, char* resultBuffer);
  NdbRecAttr* getValue(const NdbColumnImpl&, char* resultBuffer);

  int setResultRowBuf (const NdbRecord *rec,
                       char* resBuffer,
                       const unsigned char* result_mask);

  int setResultRowRef (const NdbRecord* rec,
                       const char* & bufRef,
                       const unsigned char* result_mask);

  NdbQuery::NextResultOutcome firstResult();

  NdbQuery::NextResultOutcome nextResult(bool fetchAllowed, bool forceSend);

  bool isRowNULL() const;    // Row associated with Operation is NULL value?

  bool isRowChanged() const; // Prev ::nextResult() on NdbQuery retrieved a new
                             // value for this NdbQueryOperation

  bool execTRANSID_AI(const Uint32* ptr, Uint32 len);
  
  bool execTCKEYREF(const NdbApiSignal* aSignal);

  bool execSCAN_TABCONF(Uint32 tcPtrI, Uint32 rowCount, Uint32 nodeMask,
                        NdbReceiver* receiver); 

  const NdbQueryOperation& getInterface() const
  { return m_interface; }
  NdbQueryOperation& getInterface()
  { return m_interface; }

  int setOrdering(NdbQueryOptions::ScanOrdering ordering);

  NdbQueryOptions::ScanOrdering getOrdering() const
  { return m_ordering; }

  int setParallelism(Uint32 parallelism);

  int setMaxParallelism();

  int setAdaptiveParallelism();

  int setBatchSize(Uint32 batchSize);

  int setInterpretedCode(const NdbInterpretedCode& code);
  bool hasInterpretedCode() const;

  bool checkMagicNumber() const
  { return m_magic == MAGIC; }

  Uint32 getMaxBatchRows() const
  { return m_maxBatchRows; }

  Uint32 getRowSize() const;

  const NdbRecord* getNdbRecord() const
  { return m_ndbRecord; }

private:

  STATIC_CONST (MAGIC = 0xfade1234);

  NdbQueryOperation m_interface;
  const Uint32 m_magic;
  NdbQueryImpl& m_queryImpl;
  const NdbQueryOperationDefImpl& m_operationDef;

  /* MAYBE: replace m_children with navigation via m_operationDef.getChildOperation().*/
  NdbQueryOperationImpl* m_parent;
  Vector<NdbQueryOperationImpl*> m_children;

  Uint32 m_maxBatchRows;

  Uint32Buffer m_params;

  char* m_resultBuffer;
  const char** m_resultRef;
  bool m_isRowNull;

  const NdbRecord* m_ndbRecord;
  const unsigned char* m_read_mask;

  NdbRecAttr* m_firstRecAttr;
  NdbRecAttr* m_lastRecAttr;

  NdbQueryOptions::ScanOrdering m_ordering;

  NdbInterpretedCode* m_interpretedCode;

  bool m_diskInUserProjection;

  Uint32 m_parallelism;
  
  mutable Uint32 m_rowSize;

  explicit NdbQueryOperationImpl(NdbQueryImpl& queryImpl, 
                                 const NdbQueryOperationDefImpl& def);
  ~NdbQueryOperationImpl();

  void fetchRow(NdbResultStream& resultStream);

  void nullifyResult();

  void postFetchRelease();

  Int32 getNoOfDescendantOperations() const;

  Uint32 getNoOfLeafOperations() const;

  int serializeParams(const NdbQueryParamValue* paramValues);

  int serializeProject(Uint32Buffer& attrInfo);

  Uint32 calculateBatchedRows(const NdbQueryOperationImpl* closestScan);
  void setBatchedRows(Uint32 batchedRows);

  int prepareAttrInfo(Uint32Buffer& attrInfo);

  int prepareKeyInfo(Uint32Buffer& keyInfo,
                     const NdbQueryParamValue* actualParam);

  int prepareLookupKeyInfo(
                     Uint32Buffer& keyInfo,
                     const NdbQueryOperandImpl* const keys[],
                     const NdbQueryParamValue* actualParam);

  int prepareIndexKeyInfo(
                     Uint32Buffer& keyInfo,
                     const NdbQueryOperationDefImpl::IndexBound* bounds,
                     const NdbQueryParamValue* actualParam);

  Uint32 getIdOfReceiver() const;
  
  int prepareInterpretedCode(Uint32Buffer& attrInfo) const;

  bool diskInUserProjection() const
  { return m_diskInUserProjection; }

}; // class NdbQueryOperationImpl


#endif