Method transferLiveBatch

query/aql_processor.go:505–532

transferLiveBatch returns a functor to transfer a live batch to device memory. The size parameter will be either the size of the batch or num records in last batch. hostColumns will always be empty since we should not release a vector party of a live batch. Start row will always be zero as well.

(batch *memstore.LiveBatch, size int)

// size of the batch or num records in last batch. hostColumns will always be empty since we should not release a vector
// party of a live batch. Start row will always be zero as well.
func (qc *AQLQueryContext) transferLiveBatch(batch *memstore.LiveBatch, size int) batchTransferExecutor {
	return func(stream unsafe.Pointer) (deviceColumns []deviceVectorPartySlice, hostVPs []memCom.VectorParty,
		firstColumn, startRow, totalBytes, numTransfers, sizeAfterPrefilter int) {
		// Allocate column inputs.
		firstColumn = -1
		deviceColumns = make([]deviceVectorPartySlice, len(qc.TableScanners[0].Columns))
		for i, columnID := range qc.TableScanners[0].Columns {
			usage := qc.TableScanners[0].ColumnUsages[columnID]
			if usage&(columnUsedByAllBatches|columnUsedByLiveBatches) != 0 {
				if firstColumn < 0 {
					firstColumn = i
				}
				sourceVP := batch.Columns[columnID]
				if sourceVP == nil {
					continue
				}

				hostColumn := sourceVP.(memstore.TransferableVectorParty).GetHostVectorPartySlice(0, size)
				deviceColumns[i] = hostToDeviceColumn(hostColumn, qc.Device)
				b, t := copyHostToDevice(hostColumn, deviceColumns[i], stream, qc.Device)
				totalBytes += b
				numTransfers += t
			}
		}
		sizeAfterPrefilter = size
		return
	}
}

// liveBatchTimeFilterExecutor returns a functor to apply custom time filters to live batch.
func (qc *AQLQueryContext) liveBatchCustomFilterExecutor(cutoff uint32) customFilterExecutor {