Method join

query/aql_batchexecutor.go:116–166

join

()

// join
func (e *BatchExecutorImpl) join() {
	e.qc.doProfile(func() {
		// join foreign tables
		for joinTableID, foreignTable := range e.qc.OOPK.foreignTables {
			if foreignTable != nil {
				// prepare foreign table recordIDs
				// Note:
				// RecordID {
				//   int32_t batchID
				// 	 uint32_t index
				// }
				// takes up 8 bytes
				e.qc.OOPK.currentBatch.foreignTableRecordIDsD = append(e.qc.OOPK.currentBatch.foreignTableRecordIDsD, deviceAllocate(8*e.qc.OOPK.currentBatch.size, e.qc.Device))
				mainTableJoinColumnIndex := e.qc.TableScanners[0].ColumnsByIDs[foreignTable.remoteJoinColumn.ColumnID]
				// perform hash lookup
				e.qc.OOPK.currentBatch.prepareForeignRecordIDs(mainTableJoinColumnIndex, joinTableID, *foreignTable, e.stream, e.qc.Device)
			}
		}
		e.qc.reportTimingForCurrentBatch(e.stream, &e.start, prepareForeignRecordIDsTiming)
	}, "joins", e.stream)

	e.qc.doProfile(func() {
		// process filters that involves foreign table columns if any
		for _, filter := range e.qc.OOPK.ForeignTableCommonFilters {
			e.qc.OOPK.currentBatch.processExpression(filter, nil,
				e.qc.TableScanners, e.qc.OOPK.foreignTables, e.stream, e.qc.Device, e.qc.OOPK.currentBatch.filterAction)
		}
		e.qc.reportTimingForCurrentBatch(e.stream, &e.start, foreignTableFilterEvalTiming)
	}, "filters", e.stream)

	if e.qc.OOPK.geoIntersection != nil {
		// allocate two predicate vector for geo intersect
		numWords := (e.qc.OOPK.geoIntersection.numShapes + 31) / 32
		e.qc.OOPK.currentBatch.geoPredicateVectorD = deviceAllocate(e.qc.OOPK.currentBatch.size*4*numWords, e.qc.Device)
	}

	e.sizeBeforeGeoFilter = e.qc.OOPK.currentBatch.size
	e.qc.doProfile(func() {
		if e.qc.OOPK.geoIntersection != nil {
			pointColumnIndex := e.qc.TableScanners[e.qc.OOPK.geoIntersection.pointTableID].
				ColumnsByIDs[e.qc.OOPK.geoIntersection.pointColumnID]
			e.qc.OOPK.currentBatch.geoIntersect(
				e.qc.OOPK.geoIntersection,
				pointColumnIndex,
				e.qc.OOPK.foreignTables,
				e.qc.OOPK.currentBatch.geoPredicateVectorD,
				e.stream, e.qc.Device)
		}
		e.qc.reportTimingForCurrentBatch(e.stream, &e.start, geoIntersectEvalTiming)
	}, "geo_intersect", e.stream)
}

// evalMeasures is to fill measure values
func (e *BatchExecutorImpl) evalMeasures() {