text: /*
   * 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.
   */
  #include "ao_reader.h"
  #include "tuplebatch.h"
  #include "utils/datum.h"
  
  extern  MemTuple
  appendonlygettup(AppendOnlyScanDesc scan,
                   ScanDirection dir __attribute__((unused)),
                   int nkeys,
                   ScanKey key,
                   TupleTableSlot *slot);
  
  void
  BeginVScanAppendOnlyRelation(ScanState *scanState)
  {
      BeginScanAppendOnlyRelation(scanState);
      VectorizedState* vs = (VectorizedState*)scanState->ps.vectorized;
      TupleBatch tb = scanState->ss_ScanTupleSlot->PRIVATE_tb;
      vs->ao = palloc0(sizeof(aoinfo));
      vs->ao->proj = palloc0(sizeof(bool) * tb->ncols);
      GetNeededColumnsForScan((Node* )scanState->ps.plan->targetlist,vs->ao->proj,tb->ncols);
      GetNeededColumnsForScan((Node* )scanState->ps.plan->qual,vs->ao->proj,tb->ncols);
  }
  
  void
  EndVScanAppendOnlyRelation(ScanState *scanState)
  {
      VectorizedState* vs = (VectorizedState*)scanState->ps.vectorized;
      pfree(vs->ao->proj);
      pfree(vs->ao);
      EndScanAppendOnlyRelation(scanState);
  }
  
  static TupleTableSlot *
  AOScanNext(ScanState *scanState)
  {
  	Assert(IsA(scanState, TableScanState) ||
  		   IsA(scanState, DynamicTableScanState));
  	AppendOnlyScanState *node = (AppendOnlyScanState *)scanState;
  	VectorizedState* vs = scanState->ps.vectorized;
  
  	AppendOnlyScanDesc scandesc;
  	EState	   *estate;
  	ScanDirection direction;
  	TupleTableSlot *slot;
  
  	Assert((node->ss.scan_state & SCAN_SCAN) != 0);
  
  	estate = node->ss.ps.state;
  	scandesc = node->aos_ScanDesc;
  	direction = estate->es_direction;
  	slot = node->ss.ss_ScanTupleSlot;
  
  	MemTuple tup = appendonlygettup(scandesc, direction, scandesc->aos_nkeys, scandesc->aos_key, slot);
  
  	if (tup == NULL)
  	{
          vs->ao->isDone = true;
          return NULL;
  	}
  
  	pgstat_count_heap_getnext(scandesc->aos_rd);
  
      return slot;
  }
  
  
  TupleTableSlot *
  AppendOnlyVScanNext(ScanState *scanState)
  {
      TupleTableSlot *slot = scanState->ss_ScanTupleSlot;
      TupleBatch tb = (TupleBatch)slot->PRIVATE_tb;
      VectorizedState* vs = scanState->ps.vectorized;
  
      if(vs->ao->isDone)
      {
          ExecClearTuple(slot);
          return slot;
      }
  
      for(tb->nrows = 0;tb->nrows < tb->batchsize;tb->nrows ++)
      {
          slot = AOScanNext(scanState);
  
          if(TupIsNull(slot))
              break;
  
          for(int i = 0;i < tb->ncols ; i ++)
          {
             if(vs->ao->proj[i])
              {
                  Oid hawqTypeID = slot->tts_tupleDescriptor->attrs[i]->atttypid;
                  Oid hawqVTypeID = GetVtype(hawqTypeID);
                  if(!tb->datagroup[i])
                      tbCreateColumn(tb,i,hawqVTypeID);
  
                  tb->datagroup[i]->values[tb->nrows] = slot_getattr(slot,i + 1, &(tb->datagroup[i]->isnull[tb->nrows]));
  
                  /* if attribute is a reference, deep copy the data out to prevent ao table buffer free before vectorized scan batch done */
                  if(!slot->tts_mt_bind->tupdesc->attrs[i]->attbyval)
                      tb->datagroup[i]->values[tb->nrows]= datumCopy(tb->datagroup[i]->values[tb->nrows],
                                                                     slot->tts_mt_bind->tupdesc->attrs[i]->attbyval,
                                                                     slot->tts_mt_bind->tupdesc->attrs[i]->attlen);
  
              }
          }
      }
  
      if(!slot)
          slot = scanState->ss_ScanTupleSlot;
  
      for(int i = 0;i < tb->ncols ; i ++)
      {
          if(vs->ao->proj[i])
          {
              if(tb->datagroup[i])
                  tb->datagroup[i]->dim = tb->nrows;
          }
      }
  
      if (tb->nrows == 0)
          ExecClearTuple(slot);
      else
          TupSetVirtualTupleNValid(slot, tb->ncols);
      return slot;
  }
  
  