text: #include <stdio.h>
  #include <stdlib.h>
  #include <string.h>
  
  #include "adc.h"
  #include "macrodef.h"
  #include "protots.h"
  
  #ifdef UNIX
  #include <errno.h>
  #endif
  
  extern int32 computeChecksum(ADC_VIEW_CNTL *avp,treeNode *t,uint64 *ordern);
  extern int32 WriteViewToDiskCS(ADC_VIEW_CNTL *avp,treeNode *t,uint64 *ordern);
  
  int32 ReadWholeInputData(ADC_VIEW_CNTL *avp, FILE *inpf){
    uint32 iRec = 0;
    uint32 inpBufferLineSize, inpBufferPace, inpRecSize, ib = 0;
  
    FSEEK(inpf, 0L, SEEK_SET);
    inpRecSize = 8*avp->nm+4*avp->nTopDims;
    inpBufferLineSize = inpRecSize;
    if (inpBufferLineSize%8) inpBufferLineSize += 4;
    inpBufferPace = inpBufferLineSize/4;
  
    while(fread(&avp->inpDataBuffer[ib], inpRecSize, 1, inpf)){
       iRec++;
       ib += inpBufferPace;      
    }
    avp->nRowsToRead = iRec;
    FSEEK(inpf, 0L, SEEK_SET);
    
    if(avp->nInputRecs != iRec){
       fprintf(stderr, " ReadWholeInputData(): wrong input data reading.\n");
       return ADC_INTERNAL_ERROR;
    }  
    return ADC_OK;
  }
  int32 ComputeMemoryFittedView (ADC_VIEW_CNTL *avp){
    uint32 iRec = 0;
    uint32 viewBuf[MAX_VIEW_ROW_SIZE_IN_INTS];
    uint32 inpBufferLineSize, inpBufferPace, inpRecSize,ib;
    uint64 ordern=0;
  #ifdef VIEW_FILE_OUTPUT
    uint32 retCode;
  #endif
  
    FSEEK(avp->viewFile, 0L, SEEK_END);
    inpRecSize = 8*avp->nm+4*avp->nTopDims;
    inpBufferLineSize = inpRecSize;
    if (inpBufferLineSize%8) inpBufferLineSize += 4;
    inpBufferPace = inpBufferLineSize/4;
  
    InitializeTree(avp->tree, avp->nv, avp->nm);
  
    ib=0;
    for ( iRec = 1; iRec <= avp->nRowsToRead; iRec++ ){ 
        SelectToView( &avp->inpDataBuffer[ib], avp->selection, viewBuf, 
    		             avp->nd, avp->nm, avp->nv );
        ib += inpBufferPace;
        TreeInsert(avp->tree, viewBuf);
        if(avp->tree->memoryIsFull){
    	fprintf(stderr, "ComputeMemoryFittedView(): Not enough memory.\n");
    	return 1; 
        }
    }
  
  #ifdef VIEW_FILE_OUTPUT
    if( retCode = WriteViewToDiskCS(avp, avp->tree->root.left,&ordern) ){ 
      fprintf(stderr, "ComputeMemoryFittedView() Write error is occured.\n");
      return retCode;
    }
  #else
    computeChecksum(avp,avp->tree->root.left,&ordern);
  #endif
   
    avp->nViewRows = avp->tree->count;
    avp->totalOfViewRows += avp->nViewRows; 			      
    InitializeTree(avp->tree, avp->nv, avp->nm);
    return ADC_OK;
  }
  
  int32 SharedSortAggregate(ADC_VIEW_CNTL *avp){
     int32 retCode;
    uint32 iRec = 0;
    uint32   attrs[MAX_VIEW_ROW_SIZE_IN_INTS]; 
    uint32 currBuf[MAX_VIEW_ROW_SIZE_IN_INTS]; 
     int64 chunkOffset = 0;
     int64 inpfOffset;
    uint32 nPart = 0;
    uint32 prevV;
    uint32 currV;
    uint32 total = 0;
    unsigned char *ib;
    uint32 ibsize = SSA_BUFFER_SIZE;
    uint32 nib;
    uint32 iib;
    uint32 nreg;
    uint32 nlst;
    uint32 nsgs;
    uint32 ncur;
    uint32 ibOffset = 0;
    uint64 ordern=0;
     
    ib = (unsigned char*) malloc(ibsize); 
    if (!ib){ 
      fprintf(stderr,"SharedSortAggregate: memory allocation failed\n"); 
      return ADC_MEMORY_ALLOCATION_FAILURE; 
    }
    
    nib = ibsize/avp->inpRecSize;
    nsgs = avp->nRowsToRead/nib;
    
    if (nsgs == 0){
        nreg = avp->nRowsToRead; 
        nlst = nreg; 
        nsgs = 1; 
    }else{
       nreg = nib;
       if (avp->nRowsToRead%nib) {
         nsgs++; 
         nlst = avp->nRowsToRead%nib;
       }else{
         nlst = nreg;			   
       }
    }
    
    avp->nViewRows = 0; 
    for( iib = 1; iib <= nsgs; iib++ ){ 
      if(iib > 1) FSEEK(avp->viewFile, inpfOffset, SEEK_SET);
      if( iib == nsgs ) ncur = nlst; else ncur = nreg;
      	  
      fread(ib, ncur*avp->inpRecSize, 1, avp->viewFile);
      inpfOffset = ftell(avp->viewFile);
  
      for( ibOffset = 0, iRec = 1; iRec <= ncur; iRec++ ){
        memcpy(attrs, &ib[ibOffset], avp->inpRecSize);
        ibOffset += avp->inpRecSize;
        SelectToView(attrs, avp->selection, currBuf, avp->nd, avp->nm, avp->nv); 
        currV = currBuf[2*avp->nm];
  
        if(iib == 1 && iRec == 1){ 
          prevV = currV; 
          nPart = 1;
          InitializeTree(avp->tree, avp->nv, avp->nm);
          TreeInsert(avp->tree, currBuf);
        }else{
           if (currV == prevV){
              nPart++;
  	    TreeInsert (avp->tree, currBuf);
              if (avp->tree->memoryIsFull){
  	      avp->chunksParams[avp->numberOfChunks].curChunkNum =
  	                                             avp->tree->count;
  	      avp->chunksParams[avp->numberOfChunks].chunkOffset = chunkOffset;
                (avp->numberOfChunks)++;
  	      if(avp->numberOfChunks >= MAX_NUM_OF_CHUNKS){
                  fprintf(stderr,"Too many chunks were created.\n"); 
  		exit(1);
                }
                chunkOffset += (uint64)(avp->tree->count*avp->outRecSize);
                retCode=WriteChunkToDisk(avp->outRecSize, avp->fileOfChunks,
  	                               avp->tree->root.left, avp->logf);                                       
                if(retCode!=ADC_OK){
  		fprintf(stderr,"SharedSortAggregate: Write error occured.\n"); 
  		return retCode;
  	      }
                InitializeTree(avp->tree, avp->nv, avp->nm);
  	    } /* memoryIsFull */
           }else{
  	   if(avp->numberOfChunks && avp->tree->count!=0){ 
  	     avp->chunksParams[avp->numberOfChunks].curChunkNum =
  	        				     avp->tree->count;
  	     avp->chunksParams[avp->numberOfChunks].chunkOffset = chunkOffset;
               (avp->numberOfChunks)++;
               chunkOffset += 
  	    	      (uint64)(avp->tree->count*(4*avp->nv + 8*avp->nm));
  	     retCode=WriteChunkToDisk( avp->outRecSize, avp->fileOfChunks,
  	   				 avp->tree->root.left, avp->logf);
               if(retCode!=ADC_OK){
  	       fprintf(stderr,"SharedSortAggregate: Write error occured.\n");
  	       return retCode;    
  	      }
  	    }
              FSEEK(avp->viewFile, 0L, SEEK_END);
              if(!avp->numberOfChunks){
                 avp->nViewRows += avp->tree->count;
  	       retCode = WriteViewToDiskCS(avp, avp->tree->root.left,&ordern);
  	       if(retCode!=ADC_OK){ 
  	          fprintf(stderr, 
  	        	 "SharedSortAggregate: Write error occured.\n");
  	          return retCode;
  	       }
   	     }else{
  	       retCode=MultiWayMerge(avp);
  	       if(retCode!=ADC_OK) {
  	         fprintf(stderr,"SharedSortAggregate.MultiWayMerge: failed.\n");
  	         return retCode;
  	       } 
  	     }
               InitializeTree(avp->tree, avp->nv, avp->nm);
               TreeInsert(avp->tree, currBuf);
               total += nPart;
               nPart = 1;
            }
         }
         prevV = currV;
      } /* iRec */
    } /* iib */
  
    if(avp->numberOfChunks && avp->tree->count!=0) { 
      avp->chunksParams[avp->numberOfChunks].curChunkNum = avp->tree->count;
      avp->chunksParams[avp->numberOfChunks].chunkOffset = chunkOffset;
      (avp->numberOfChunks)++;
      chunkOffset += (uint64)(avp->tree->count*(4*avp->nv + 8*avp->nm));
      retCode=WriteChunkToDisk(avp->outRecSize, avp->fileOfChunks,
      			     avp->tree->root.left, avp->logf);
      if(retCode!=ADC_OK){
        fprintf(stderr,"SharedSortAggregate: Write error occured.\n");
        return retCode;	 
      }
    }
    FSEEK(avp->viewFile, 0L, SEEK_END);
    if(!avp->numberOfChunks){
      avp->nViewRows += avp->tree->count;
      if( retCode = WriteViewToDiskCS(avp, avp->tree->root.left,&ordern)){ 
        fprintf(stderr, "SharedSortAggregate: Write error occured.\n");
        return retCode;
      }	 
    }else{
       retCode=MultiWayMerge(avp);
       if(retCode!=ADC_OK) {
         fprintf(stderr,"SharedSortAggregate.MultiWayMerge failed.\n");
         return retCode;
       } 
    }
    FSEEK(avp->fileOfChunks, 0L, SEEK_SET);
    
    total += nPart;
    avp->totalOfViewRows += avp->nViewRows;
    if(ib) free(ib);
    return  ADC_OK;
  }
  int32 PrefixedAggregate(ADC_VIEW_CNTL *avp, FILE *iof){
     uint32 i;
     uint32 iRec = 0;
     uint32   attrs[MAX_VIEW_ROW_SIZE_IN_INTS]; 
     uint32 aggrBuf[MAX_VIEW_ROW_SIZE_IN_INTS];
     uint32 currBuf[MAX_VIEW_ROW_SIZE_IN_INTS]; 
     uint32 prevBuf[MAX_VIEW_ROW_SIZE_IN_INTS];
      int64 *aggrmp;
      int64 *currmp;
      int32 compRes;
     uint32 nOut = 0; 
     uint32 mpOffset = 0;
     uint32 nOutBufRecs;
     uint32 nViewRows = 0;
      int64 inpfOffset;
  
      aggrmp = (int64*) &aggrBuf[0];
      currmp = (int64*) &currBuf[0];
      
      for(i = 0; i < 2*avp->nm+avp->nv; i++){prevBuf[i] = 0; aggrBuf[i] = 0;}
      nOutBufRecs = avp->memoryLimit/avp->outRecSize;
  
      for(iRec = 1; iRec <= avp->nRowsToRead; iRec++ ){ 
        fread(attrs, avp->inpRecSize, 1, iof);
        SelectToView(attrs, avp->selection, currBuf, avp->nd, avp->nm, avp->nv);
        if (iRec == 1) memcpy(aggrBuf, currBuf, avp->outRecSize);
        else{
         compRes = KeyComp( &currBuf[2*avp->nm], &prevBuf[2*avp->nm], avp->nv);
  
         switch(compRes){
  	  case  1: 
  	    memcpy(&avp->memPool[mpOffset], aggrBuf, avp->outRecSize);
  	    mpOffset += avp->outRecSize;
  	    nOut++;
  	    for ( i = 0; i < avp->nm; i++ ){
  	      avp->mSums[i] += aggrmp[i];
  	      avp->checksums[i] += nOut*aggrmp[i]%measbound;
  	    }    
  	    memcpy(aggrBuf, currBuf, avp->outRecSize);
  	    break;
  	  case  0: 
  	    for ( i = 0; i < avp->nm; i++ ) aggrmp[i] += currmp[i];
  	    break;
  	  case -1: 
  	    fprintf(stderr,"PrefixedAggregate: wrong parent view order.\n"); 
  	    exit(1);
  	    break; 
  	  default: 
  	    fprintf(stderr,"PrefixedAggregate: wrong KeyComp() result.\n"); 
  	    exit(1);
  	    break;
         }     
      
         if (nOut == nOutBufRecs){
  	     inpfOffset = ftell(iof);
  	     FSEEK(iof, 0L, SEEK_END);
  	     WriteToFile(avp->memPool, nOut*avp->outRecSize, 1, iof, stderr);
  	     FSEEK(iof, inpfOffset, SEEK_SET);
  	     mpOffset = 0;
  	     nViewRows += nOut;
  	     nOut = 0; 
         }
       }
       memcpy(prevBuf, currBuf, avp->outRecSize);
     }
     memcpy(&avp->memPool[mpOffset], aggrBuf, avp->outRecSize);
     nOut++;
     for ( i = 0; i < avp->nm; i++ ){
       avp->mSums[i] += aggrmp[i];
       avp->checksums[i] += nOut*aggrmp[i]%measbound;
     }
     FSEEK(iof, 0L, SEEK_END);
     WriteToFile(avp->memPool, nOut*avp->outRecSize, 1, iof, stderr);
     avp->nViewRows	 = nViewRows+nOut;
     avp->totalOfViewRows += avp->nViewRows;
     return ADC_OK;
  }
  int32 RunFormation (ADC_VIEW_CNTL *avp, FILE *inpf){
     uint32 iRec = 0;
     uint32 viewBuf[MAX_VIEW_ROW_SIZE_IN_INTS];
     uint32   attrs[MAX_VIEW_ROW_SIZE_IN_INTS]; 
      int64 chunkOffset = 0;
  
     InitializeTree(avp->tree, avp->nv, avp->nm);
  
     for(iRec = 1; iRec <= avp->nRowsToRead; iRec++ ){ 
       fread(attrs, avp->inpRecSize, 1, inpf);
       SelectToView(attrs, avp->selection, viewBuf, avp->nd, avp->nm, avp->nv); 
       TreeInsert(avp->tree, viewBuf);
  
       if(avp->tree->memoryIsFull) {
          avp->chunksParams[avp->numberOfChunks].curChunkNum = avp->tree->count;
  	    avp->chunksParams[avp->numberOfChunks].chunkOffset  = chunkOffset;		 
          (avp->numberOfChunks)++;
  	    if (avp->numberOfChunks >= MAX_NUM_OF_CHUNKS) {
            fprintf(stderr, "RunFormation: Too many chunks were created.\n"); 
            return ADC_INTERNAL_ERROR;
          }
          chunkOffset += (uint64)(avp->tree->count*avp->outRecSize);
          if(WriteChunkToDisk( avp->outRecSize, avp->fileOfChunks,
  	                         avp->tree->root.left, avp->logf )){
  	       fprintf(stderr, 
  	         "RunFormation.WriteChunkToDisk: Write error is occured.\n");
  	       return ADC_WRITE_FAILED;
  	    }
          InitializeTree(avp->tree, avp->nv, avp->nm);
         }
     } /* Insertion ... */
     if(avp->numberOfChunks && avp->tree->count!=0) { 
       avp->chunksParams[avp->numberOfChunks].curChunkNum = avp->tree->count;
       avp->chunksParams[avp->numberOfChunks].chunkOffset  = chunkOffset;
       (avp->numberOfChunks)++;
       chunkOffset += (uint64)(avp->tree->count*(4*avp->nv + 8*avp->nm));
       if(WriteChunkToDisk(avp->outRecSize, avp->fileOfChunks,
                           avp->tree->root.left, avp->logf)){
         fprintf(stderr, 
              "RunFormation(.WriteChunkToDisk: Write error is occured.\n");
         return ADC_WRITE_FAILED;  
       }
     }
     FSEEK(avp->viewFile, 0L, SEEK_END);
     return ADC_OK;
  }
  void SeekAndReadNextSubChunk( uint32 multiChunkBuffer[], 
                                uint32 k,
                                FILE *inFile,
  		              uint32 chunkRecSize, 
  		              uint64 inFileOffs,
  		              uint32 subChunkNum){
     int64 ret;
    
     ret = FSEEK(inFile, inFileOffs, SEEK_SET);
     if (ret < 0){
        fprintf(stderr,"SeekAndReadNextSubChunk.fseek() < 0 "); 
        exit(1); 
     }
     fread(&multiChunkBuffer[k], chunkRecSize*subChunkNum, 1, inFile);
  }
  void ReadSubChunk(
              uint32 chunkRecSize,
              uint32 *multiChunkBuffer,
              uint32 mwBufRecSizeInInt,
              uint32 iChunk,
              uint32 regSubChunkSize,
              CHUNKS *chunks,  
                FILE *fileOfChunks
              ){
     if (chunks[iChunk].curChunkNum > 0){
        if(chunks[iChunk].curChunkNum < regSubChunkSize){
  	SeekAndReadNextSubChunk(multiChunkBuffer,
  	   			(iChunk*regSubChunkSize +
  	   			(regSubChunkSize-chunks[iChunk].curChunkNum))*
  	   			mwBufRecSizeInInt,
  	   			fileOfChunks,
  	   			chunkRecSize,
  	   			chunks[iChunk].chunkOffset,
  	   			chunks[iChunk].curChunkNum);
  	chunks[iChunk].posSubChunk=regSubChunkSize-chunks[iChunk].curChunkNum;
  	chunks[iChunk].curSubChunk=chunks[iChunk].curChunkNum;
  	chunks[iChunk].curChunkNum=0;
  	chunks[iChunk].chunkOffset=-1;
        }else{
  	SeekAndReadNextSubChunk(multiChunkBuffer,
  	   			iChunk*regSubChunkSize*mwBufRecSizeInInt,
  	   			fileOfChunks,
  	   			chunkRecSize,
  	   			chunks[iChunk].chunkOffset,
  	   			regSubChunkSize);
  	chunks[iChunk].posSubChunk = 0;
  	chunks[iChunk].curSubChunk = regSubChunkSize;
  	chunks[iChunk].curChunkNum -= regSubChunkSize;
  	chunks[iChunk].chunkOffset += regSubChunkSize * chunkRecSize;
        }
     }
  }
  int32 MultiWayMerge(ADC_VIEW_CNTL *avp){
     uint32 outputBuffer[OUTPUT_BUFFER_SIZE];
     uint32 r_buf       [OUTPUT_BUFFER_SIZE];
     uint32 min_r_buf   [OUTPUT_BUFFER_SIZE];
     uint32 first_one;
     uint32 i;
     uint32 iChunk;
     uint32 min_r_chunk;
     uint32 sPos;
     uint32 iPos;
     uint32 numEmptyBufs;
     uint32 numEmptyRuns;
     uint32 mwBufRecSizeInInt;
     uint32 chunkRecSize;
     uint32 *multiChunkBuffer;
     uint32   regSubChunkSize;
      int32 compRes;
      int64 *m_min_r_buf;
      int64 *m_outputBuffer;
  
     FSEEK(avp->fileOfChunks, 0L, SEEK_SET);
  
     multiChunkBuffer = (uint32*) &avp->memPool[0];
     first_one = 1;
     avp->nViewRows  = 0; 
  
     chunkRecSize = avp->outRecSize;
     mwBufRecSizeInInt = chunkRecSize/4;
     m_min_r_buf = (int64*)&min_r_buf[0];
     m_outputBuffer = (int64*)&outputBuffer[0];
  
     mwBufRecSizeInInt = chunkRecSize/4;
     regSubChunkSize = (avp->memoryLimit/avp->numberOfChunks)/chunkRecSize;
  	 
     if (regSubChunkSize==0) {
       fprintf(stderr,
               "MultiWayMerge: Not enough memory to run the external sort\n");
       return ADC_INTERNAL_ERROR;
     }
     multiChunkBuffer = (uint32*) &avp->memPool[0];
  
     for(i = 0; i < avp->numberOfChunks; i++ ){
        ReadSubChunk( 
                     chunkRecSize,
                     multiChunkBuffer,
                     mwBufRecSizeInInt,
                     i,
                     regSubChunkSize,
                     avp->chunksParams,  
                     avp->fileOfChunks
        );
     }
     while(1){
       for(iChunk = 0;iChunk<avp->numberOfChunks;iChunk++){
         if (avp->chunksParams[iChunk].curSubChunk > 0){
       	sPos = iChunk*regSubChunkSize*mwBufRecSizeInInt;
      	iPos = sPos+mwBufRecSizeInInt*avp->chunksParams[iChunk].posSubChunk;
       	memcpy(&min_r_buf[0], &multiChunkBuffer[iPos], avp->outRecSize);
  	    min_r_chunk = iChunk;
       	break;
         }
       }
       for ( iChunk = min_r_chunk; iChunk < avp->numberOfChunks; iChunk++ ){
         uint32 iPos;
  
         if (avp->chunksParams[iChunk].curSubChunk > 0){
            iPos = mwBufRecSizeInInt*(iChunk*regSubChunkSize+
                                     avp->chunksParams[iChunk].posSubChunk);
            memcpy(&r_buf[0],&multiChunkBuffer[iPos],avp->outRecSize);
  
            compRes=KeyComp(&r_buf[2*avp->nm],&min_r_buf[2*avp->nm],avp->nv);	
            if(compRes < 0) {
       	      memcpy(&min_r_buf[0], &r_buf[0], avp->outRecSize);
  	          min_r_chunk = iChunk;
            }
         }
       }
       /* Step forward */
       if(avp->chunksParams[min_r_chunk].curSubChunk != 0){
         avp->chunksParams[min_r_chunk].curSubChunk--;
         avp->chunksParams[min_r_chunk].posSubChunk++;
       }
  
         /* Aggreagation if a duplicate is encountered */
         if(first_one){
           memcpy( &outputBuffer[0], &min_r_buf[0], avp->outRecSize);
           first_one = 0;
         }else{
           compRes = KeyComp( &outputBuffer[2*avp->nm], 
          		    &min_r_buf[2*avp->nm], avp->nv );
           if(!compRes){
             for(i = 0; i < avp->nm; i++ ){ 
               m_outputBuffer[i] += m_min_r_buf[i]; 
             }
           }else{
             WriteToFile(outputBuffer,avp->outRecSize,1,avp->viewFile,stderr);
             avp->nViewRows++;
             for(i=0;i<avp->nm;i++){
  	     avp->mSums[i]+=m_outputBuffer[i];
  	     avp->checksums[i] += avp->nViewRows*m_outputBuffer[i]%measbound;
  	   }
             memcpy( &outputBuffer[0], &min_r_buf[0], avp->outRecSize );
          }
        }
  
        for(numEmptyBufs = 0, 
            numEmptyRuns = 0, i = 0; i < avp->numberOfChunks; i++ ){
  	     if (avp->chunksParams[i].curSubChunk == 0) numEmptyBufs++;
           if (avp->chunksParams[i].curChunkNum == 0) numEmptyRuns++;
        }
        if(   numEmptyBufs == avp->numberOfChunks 
            &&numEmptyRuns == avp->numberOfChunks) break;
  
        if(avp->chunksParams[min_r_chunk].curSubChunk == 0) {
          ReadSubChunk( 
          	 chunkRecSize,
          	 multiChunkBuffer,
          	 mwBufRecSizeInInt,
          	 min_r_chunk,
          	 regSubChunkSize,
          	 avp->chunksParams,
          	 avp->fileOfChunks);
        }
     } /* while(1) */
  
     WriteToFile( outputBuffer, avp->outRecSize, 1, avp->viewFile, stderr);	  
     avp->nViewRows++;
     for(i = 0; i < avp->nm; i++ ){ 
       avp->mSums[i] += m_outputBuffer[i]; 
       avp->checksums[i] += avp->nViewRows*m_outputBuffer[i]%measbound;
     }
  
     avp->totalOfViewRows += avp->nViewRows;
     return ADC_OK;
  }
  void SelectToView( uint32 * ib, uint32 *ix, uint32 *viewBuf, 
                     uint32 nd, uint32 nm, uint32 nv ){
     uint32 i, j;
     for ( j = 0, i = 0; i < nv; i++ ) viewBuf[2*nm+j++] = ib[2*nm+ix[i]-1];
     memcpy(&viewBuf[0], &ib[0], MSR_FSZ*nm);
  }
  FILE * AdcFileOpen(const char *fileName, const char *mode){
     FILE *fr;
     if ((fr = (FILE*) fopen(fileName, mode))==NULL)
        fprintf(stderr, "AdcFileOpen: Cannot open the file %s errno = %d\n",  
                         fileName, errno);
     return fr;
  }
  void AdcFileName(char *adcFileName, const char *adcName, 
  		 const char *fileName, uint32 taskNumber){
    sprintf(adcFileName, "%s.%s.%d",adcName,fileName,taskNumber);
  }
  ADC_VIEW_CNTL * NewAdcViewCntl(ADC_VIEW_PARS *adcpp, uint32 pnum){
     ADC_VIEW_CNTL *adccntl;
     uint32 i, j, k;
  #ifdef IN_CORE
     uint32 ux;
  #endif
     char id[8+1];
     
     adccntl = (ADC_VIEW_CNTL *) malloc(sizeof(ADC_VIEW_CNTL));
     if (adccntl==NULL) return NULL;
     
     adccntl->ndid = adcpp->ndid;
     adccntl->taskNumber = pnum;
     adccntl->retCode = 0;
     adccntl->swapIt = 0;
     strcpy(adccntl->adcName, adcpp->adcName);
     adccntl->nTopDims = adcpp->nd;
     adccntl->nd = adcpp->nd;
     adccntl->nm = adcpp->nm;
     adccntl->nInputRecs = adcpp->nInputRecs;
     adccntl->inpRecSize = GetRecSize(adccntl->nd,adccntl->nm);
     adccntl->outRecSize = GetRecSize(adccntl->nv,adccntl->nm);
     adccntl->accViewFileOffset = 0;
     adccntl->totalViewFileSize = 0;
     adccntl->numberOfMadeViews = 0;
     adccntl->numberOfViewsMadeFromInput = 0;
     adccntl->numberOfPrefixedGroupbys = 0;
     adccntl->numberOfSharedSortGroupbys = 0;
     adccntl->totalOfViewRows = 0;
     adccntl->memoryLimit = adcpp->memoryLimit;
     adccntl->nTasks = adcpp->nTasks;
     strcpy(adccntl->inpFileName, adcpp->adcInpFileName);
     sprintf(id, ".%d", adcpp->ndid);
     
     AdcFileName(adccntl->adcLogFileName, 
                 adccntl->adcName, "logf", adccntl->taskNumber);
     strcat(adccntl->adcLogFileName, id);            
     adccntl->logf = AdcFileOpen(adccntl->adcLogFileName, "w");
  
     AdcFileName(adccntl->inpFileName, adccntl->adcName, "dat", adcpp->ndid);
     adccntl->inpf = AdcFileOpen(adccntl->inpFileName, "rb");
     if(!adccntl->inpf){ 
       adccntl->retCode = ADC_FILE_OPEN_FAILURE; 
       return(adccntl);
     } 
  
     AdcFileName(adccntl->viewFileName, adccntl->adcName, 
                 "view.dat", adccntl->taskNumber);
     strcat(adccntl->viewFileName, id);            
     adccntl->viewFile = AdcFileOpen(adccntl->viewFileName, "wb+");
  
     AdcFileName(adccntl->chunksFileName, adccntl->adcName, 
                 "chunks.dat", adccntl->taskNumber);
     strcat(adccntl->chunksFileName, id);            
     adccntl->fileOfChunks = AdcFileOpen(adccntl->chunksFileName,"wb+");
  
     AdcFileName(adccntl->groupbyFileName, adccntl->adcName, 
                 "groupby.dat", adccntl->taskNumber);
     strcat(adccntl->groupbyFileName, id);
     adccntl->groupbyFile = AdcFileOpen(adccntl->groupbyFileName,"wb+");
  
     AdcFileName(adccntl->adcViewSizesFileName, adccntl->adcName, 
                 "view.sz", adcpp->ndid);
     adccntl->adcViewSizesFile = AdcFileOpen(adccntl->adcViewSizesFileName,"r");
     if(!adccntl->adcViewSizesFile){
       adccntl->retCode = ADC_FILE_OPEN_FAILURE;
       return(adccntl);
     }
  
     AdcFileName(adccntl->viewSizesFileName, adccntl->adcName, 
                 "viewsz.dat", adccntl->taskNumber);
     strcat(adccntl->viewSizesFileName, id);            
     adccntl->viewSizesFile = AdcFileOpen(adccntl->viewSizesFileName, "wb+");
     
     adccntl->chunksParams = (CHUNKS*) malloc(MAX_NUM_OF_CHUNKS*sizeof(CHUNKS));
     if(adccntl->chunksParams==NULL){ 
       fprintf(adccntl->logf,"NewAdcViewCntl: Cannot allocate 'chunksParsms'\n");
       adccntl->retCode = ADC_MEMORY_ALLOCATION_FAILURE;
       return(adccntl);
     }
     adccntl->memPool = (unsigned char*) malloc(adccntl->memoryLimit);
     if(adccntl->memPool == NULL ){
        fprintf(adccntl->logf, 
                "NewAdcViewCntl: Cannot allocate 'main memory pool'\n"); 
        adccntl->retCode = ADC_MEMORY_ALLOCATION_FAILURE;
        return(adccntl);
     }
     
  #ifdef IN_CORE   
     /* add a condition to allocate this memory buffer, THIS is IMPORTANT */
     ux = 4*adccntl->nTopDims + 8*adccntl->nm;
     if (adccntl->nTopDims%8) ux += 4;
     adccntl->inpDataBuffer = (uint32*) malloc(adccntl->nInputRecs*ux);
     if(adccntl->inpDataBuffer == NULL ){
        fprintf(adccntl->logf,
                "NewAdcViewCntl: Cannot allocate 'input data buffer'\n"); 
        adccntl->retCode = ADC_MEMORY_ALLOCATION_FAILURE;
        return(adccntl);
     }
  #endif
     adccntl->numberOfChunks = 0;
  
     for ( i = 0; i < adccntl->nm; i++ ){
       adccntl->mSums[i] = 0;
       adccntl->checksums[i] = 0;
       adccntl->totchs[i] = 0;
    }
     adccntl->tree = CreateEmptyTree(adccntl->nd, adccntl->nm, 
                                     adccntl->memoryLimit, adccntl->memPool);
     if(!adccntl->tree){
        fprintf(adccntl->logf,"\nNewAdcViewCntl.CreateEmptyTree failed.\n");
        adccntl->retCode = ADC_MEMORY_ALLOCATION_FAILURE;
        return(adccntl);
     }
  
     adccntl->nv = adcpp->nd; /* default */
     for ( i = 0; i < adccntl->nv; i++ ) adccntl->selection[i]=i+1;
     
     adccntl->nViewLimit = (1<<adcpp->nd)-1;
     adccntl->jpp=(JOB_POOL *) malloc((adccntl->nViewLimit+1)*sizeof(JOB_POOL));
     if ( adccntl->jpp == NULL){
        fprintf(adccntl->logf,
          "\n Not enough space to allocate %ld byte for a job pool.", 
          (long)(adccntl->nViewLimit+1)*sizeof(JOB_POOL));
        adccntl->retCode = ADC_MEMORY_ALLOCATION_FAILURE; 
        return(adccntl);
     }
     adccntl->lpp = (LAYER * ) malloc( (adcpp->nd+1)*sizeof(LAYER));
     if ( adccntl->lpp == NULL){
        fprintf(adccntl->logf,
          "\n Not enough space to allocate %ld byte for a layer reference array.", 
          (long)(adcpp->nd+1)*sizeof(LAYER));
        adccntl->retCode = ADC_MEMORY_ALLOCATION_FAILURE;
        return(adccntl);
     }
  
     for ( j = 1, i = 1; i <= adcpp->nd; i++ ) {
        k =  NumOfCombsFromNbyK ( adcpp->nd, i );
        adccntl->lpp[i].layerIndex = j;
        j += k;
        adccntl->lpp[i].layerQuantityLimit = k;
        adccntl->lpp[i].layerCurrentPopulation = 0;
     }    
        
     JobPoolInit ( adccntl->jpp, (adccntl->nViewLimit+1), adcpp->nd );
  
     fprintf(adccntl->logf,"\nMeaning of the log file colums is as follows:\n");
     fprintf(adccntl->logf,
       "Row Number | Groupby | View Size | Measure Sums | Number of Chunks\n");
  
     adccntl->verificationFailed = 1;
     return adccntl;
  }
  void InitAdcViewCntl(ADC_VIEW_CNTL *adccntl, 
  		     uint32 nSelectedDims, 
  		     uint32 *selection, 
  		     uint32 fromParent ){
     uint32 i;
     
     adccntl->nv = nSelectedDims;
     
     for (i = 0; i < adccntl->nm; i++ ) adccntl->mSums[i] = 0;
     for (i = 0; i < adccntl->nv; i++ ) adccntl->selection[i] = selection[i];
  
     adccntl->outRecSize = GetRecSize(adccntl->nv,adccntl->nm);
     adccntl->numberOfChunks = 0;
     adccntl->fromParent = fromParent;
     adccntl->nViewRows = 0;
  
     if(fromParent){
       adccntl->nd = adccntl->smallestParentLevel;
       FSEEK(adccntl->viewFile, adccntl->viewOffset, SEEK_SET);
       adccntl->nRowsToRead = adccntl->nParentViewRows;
     }else{
       adccntl->nd = adccntl->nTopDims;
       adccntl->nRowsToRead = adccntl->nInputRecs;
     }
     adccntl->inpRecSize = GetRecSize(adccntl->nd,adccntl->nm);
     adccntl->outRecSize = GetRecSize(adccntl->nv,adccntl->nm);
  }
  int32 CloseAdcView(ADC_VIEW_CNTL *adccntl){
     if (adccntl->inpf) fclose(adccntl->inpf);
     if (adccntl->viewFile) fclose(adccntl->viewFile);
     if (adccntl->fileOfChunks) fclose(adccntl->fileOfChunks);
     if (adccntl->groupbyFile) fclose(adccntl->groupbyFile);
     if (adccntl->adcViewSizesFile) fclose(adccntl->adcViewSizesFile);
     if (adccntl->viewSizesFile) fclose(adccntl->viewSizesFile);
     
     if (DeleteOneFile(adccntl->chunksFileName))       
        return ADC_FILE_DELETE_FAILURE;
     if (DeleteOneFile(adccntl->viewSizesFileName))    
        return ADC_FILE_DELETE_FAILURE;
  
     if (DeleteOneFile(adccntl->groupbyFileName))      
        return ADC_FILE_DELETE_FAILURE;
  
     if (adccntl->chunksParams){ 
       free(adccntl->chunksParams); 
       adccntl->chunksParams=NULL; 
     }  
     if (adccntl->memPool){ free(adccntl->memPool); adccntl->memPool=NULL;} 
     if (adccntl->jpp){ free(adccntl->jpp); adccntl->jpp=NULL; } 
     if (adccntl->lpp){ free(adccntl->lpp); adccntl->lpp=NULL; } 
  
     if (adccntl->logf) fclose(adccntl->logf);
     free(adccntl);
     return ADC_OK;
  }
  void AdcCntlLog(ADC_VIEW_CNTL *adccntlp){
    fprintf(adccntlp->logf,"    memoryLimit = %20d\n",
      adccntlp->memoryLimit);
    fprintf(adccntlp->logf,"    treeNodeSize = %20d\n",
      adccntlp->tree->treeNodeSize);
    fprintf(adccntlp->logf," treeMemoryLimit = %20d\n",
      adccntlp->tree->memoryLimit);
    fprintf(adccntlp->logf,"    nNodesLimit = %20d\n",
      adccntlp->tree->nNodesLimit);
    fprintf(adccntlp->logf,"freeNodeCounter = %20d\n",
      adccntlp->tree->freeNodeCounter);
    fprintf(adccntlp->logf,"	nViewRows = %20d\n",
      adccntlp->nViewRows);
  }
  int32 ViewSizesVerification(ADC_VIEW_CNTL *adccntlp){
       char inps[MAX_PARAM_LINE_SIZE];
       char msg[64];
       uint32 *viewCounts;
       uint32 selection_viewSize[2];
       uint32 sz;
       uint32 sel[64];
       uint32 i;
       uint32 k;
       uint64 tx;
       uint32 iTx; 
     
       viewCounts = (uint32 *) &adccntlp->memPool[0];
       for ( i = 0; i <= adccntlp->nViewLimit; i++) viewCounts[i] = 0;
       
       FSEEK(adccntlp->viewSizesFile, 0L, SEEK_SET);
       FSEEK(adccntlp->adcViewSizesFile, 0L, SEEK_SET);     
  
       while(fread(selection_viewSize, 8, 1, adccntlp->viewSizesFile)){
          viewCounts[selection_viewSize[0]] = selection_viewSize[1];
       }
       k = 0;
       while ( fscanf(adccntlp->adcViewSizesFile, "%s", inps) != EOF ){
          if ( strcmp(inps, "Selection:") == 0 ) {
             while ( fscanf(adccntlp->adcViewSizesFile, "%s", inps)) {
               if ( strcmp(inps, "View") == 0 ) break; 
               sel[k++] = atoi(inps);	  
             }
          }
          
          if ( strcmp(inps, "Size:") == 0 ) {
             fscanf(adccntlp->adcViewSizesFile, "%s", inps);
             sz = atoi(inps);
             CreateBinTuple(&tx, sel, k);
             iTx = (int32)(tx>>(64-adccntlp->nTopDims)); 
             adccntlp->verificationFailed = 0;
             if (!adccntlp->numberOfMadeViews) adccntlp->verificationFailed = 1;
  
             if ( viewCounts[iTx] != 0){
                if (viewCounts[iTx] != sz) {
                   if (viewCounts[iTx] != adccntlp->nInputRecs){
                     fprintf(adccntlp->logf, 
                             "A view size is wrong: genSz=%d calcSz=%d\n",
                     	                               sz, viewCounts[iTx]);
                     adccntlp->verificationFailed = 1;
                     return ADC_VERIFICATION_FAILED;
                   }
                }               
             }
             k = 0;
          }  
       } /* of while() */
  
       fprintf(adccntlp->logf,
         "\n\nMeaning of the log file colums is as follows:\n");
       fprintf(adccntlp->logf, 
         "Row Number | Groupby | View Size | Measure Sums | Number of Chunks\n");
  
       if (!adccntlp->verificationFailed) 
            strcpy(msg, "Verification=passed");
       else strcpy(msg, "Verification=failed");
       FSEEK(adccntlp->logf, 0L, SEEK_SET);
       fprintf(adccntlp->logf, msg);
       FSEEK(adccntlp->logf, 0L, SEEK_END);
       FSEEK(adccntlp->viewSizesFile, 0L, SEEK_SET);
       return ADC_OK;
  }
  int32 ComputeGivenGroupbys(ADC_VIEW_CNTL *adccntlp){
      int32 retCode;
     uint32 i;
     uint64 binRepTuple;
     uint32 ut32;
     uint32 nViews = 0;
     uint32 nSelectedDims;
     uint32 smp;
  #ifdef IN_CORE
     uint32 firstView = 1;
  #endif
     uint32 selection_viewsize[2];
     char ttout[16];
  
     while (fread(&binRepTuple, 8, 1, adccntlp->groupbyFile )){
       for(i = 0; i < adccntlp->nm; i++) adccntlp->checksums[i]=0;
       nViews++;
       swap8(&binRepTuple);
  
       GetRegTupleFromBin64(binRepTuple, adccntlp->selection,
                            adccntlp->nTopDims, &nSelectedDims);
       ut32 = (uint32)(binRepTuple>>(64-adccntlp->nTopDims));
       selection_viewsize[0] = ut32;
       ut32 <<= (32-adccntlp->nTopDims);
       adccntlp->groupby = ut32;
  #ifndef IN_CORE
       smp = GetParent(adccntlp, ut32);
  #endif
  #ifdef IN_CORE
       if (firstView) {
         firstView = 0;
         if(ReadWholeInputData(adccntlp, adccntlp->inpf)) {
            fprintf(stderr, "ReadWholeInputData failed.\n");
            return ADC_INTERNAL_ERROR;   
         }
       }
       smp = noneParent;
  #endif
  
       if (smp != noneParent)
       GetRegTupleFromParent(binRepTuple, 
                             adccntlp->parBinRepTuple, 
                             adccntlp->selection,
                             adccntlp->nTopDims);
       InitAdcViewCntl(adccntlp, nSelectedDims, 
                       adccntlp->selection, (smp == noneParent)?0:1);
  #ifdef IN_CORE
        if(retCode = ComputeMemoryFittedView(adccntlp)) {
           fprintf(stderr, "ComputeMemoryFittedView failed.\n");
           return retCode;
        }
  #else
  #ifdef OPTIMIZATION
       if (smp == prefixedParent){
          if (retCode = PrefixedAggregate(adccntlp, adccntlp->viewFile)) {
             fprintf(stderr, 
  	     "ComputeGivenGroupbys.PrefixedAggregate failed.\n");
             return retCode;
          }
          adccntlp->numberOfPrefixedGroupbys++;
       }else if (smp == sharedSortParent) {
          if (retCode = SharedSortAggregate(adccntlp)) {
             fprintf(stderr, 
  	     "ComputeGivenGroupbys.SharedSortAggregate failed.\n");
             return retCode;
          }
          adccntlp->numberOfSharedSortGroupbys++;
       }else
  #endif /* OPTIMIZATION */     
       { 
          if( smp != noneParent ) {
  	  retCode = RunFormation(adccntlp, adccntlp->viewFile);
            if(retCode!=ADC_OK){
                fprintf(stderr, 
  	  	  "ComputrGivenGroupbys.RunFormation failed.\n");
                return retCode; 
              }
  	  }else{
  	    if ((retCode=RunFormation (adccntlp, adccntlp->inpf)) != ADC_OK){
                fprintf(stderr, 
  	  	  "ComputrGivenGroupbys.RunFormation failed.\n");
                return retCode;
              }
  	    adccntlp->numberOfViewsMadeFromInput++;
  	  }
          if(!adccntlp->numberOfChunks){
            uint64 ordern=0;
            adccntlp->nViewRows        = adccntlp->tree->count;
            adccntlp->totalOfViewRows += adccntlp->nViewRows;
  	  retCode=WriteViewToDiskCS(adccntlp,adccntlp->tree->root.left,&ordern);
  	  if(retCode!=ADC_OK){
              fprintf(stderr,
  	            "ComputeGivenGroupbys.WriteViewToDisk: Write error.\n");
  	    return ADC_WRITE_FAILED;
  	  }
          }else { 
            retCode=MultiWayMerge(adccntlp);
            if(retCode!=ADC_OK) {
  	     fprintf(stderr,"ComputeGivenGroupbys.MultiWayMerge failed.\n");
  	     return retCode;
  	  } 
          } 
        }
       
       JobPoolUpdate(adccntlp);
  
       adccntlp->accViewFileOffset += 
         (int64)(adccntlp->nViewRows*adccntlp->outRecSize);
       FSEEK(adccntlp->fileOfChunks, 0L, SEEK_SET);
       FSEEK(adccntlp->inpf, 0L, SEEK_SET);
  #endif /* IN_CORE */
       for( i = 0; i < adccntlp->nm; i++) 
         adccntlp->totchs[i]+=adccntlp->checksums[i];
       selection_viewsize[1] = adccntlp->nViewRows;
       fwrite(selection_viewsize, 8, 1, adccntlp->viewSizesFile);
       adccntlp->totalViewFileSize += 
                              adccntlp->outRecSize*adccntlp->nViewRows;
       sprintf(ttout, "%7d ", nViews);
       WriteOne32Tuple(ttout, adccntlp->groupby, 
                       adccntlp->nTopDims, adccntlp->logf);
       fprintf(adccntlp->logf, " |  %15d | ", adccntlp->nViewRows); 
       for ( i = 0; i < adccntlp->nm; i++ ){ 
          fprintf(adccntlp->logf, " %20lld", adccntlp->checksums[i]);
       }
       fprintf(adccntlp->logf, " | %5d", adccntlp->numberOfChunks);
     }
     adccntlp->numberOfMadeViews = nViews;  
     if(ViewSizesVerification(adccntlp)) return ADC_VERIFICATION_FAILED;
     return ADC_OK;
  }
  