You are an expert in writing code in many different languages. Your goal is to perform code completion for the following code, keeping in mind the rest of the code and the file meta data.
    Metadata details:
    filename: hash_table.go
    path of file: ./repos/tidb/pkg/executor
    
    the code of the file until where you have to start completion: // Copyright 2019 PingCAP, Inc.
//
// Licensed 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.

package executor

import (
	"fmt"
	"hash"
	"hash/fnv"
	"sync/atomic"
	"time"
	"unsafe"

	"github.com/pingcap/errors"
	"github.com/pingcap/tidb/pkg/sessionctx"
	"github.com/pingcap/tidb/pkg/sessionctx/stmtctx"
	"github.com/pingcap/tidb/pkg/types"
	"github.com/pingcap/tidb/pkg/util/bitmap"
	"github.com/pingcap/tidb/pkg/util/chunk"
	"github.com/pingcap/tidb/pkg/util/codec"
	"github.com/pingcap/tidb/pkg/util/disk"
	"github.com/pingcap/tidb/pkg/util/execdetails"
	"github.com/pingcap/tidb/pkg/util/hack"
	"github.com/pingcap/tidb/pkg/util/memory"
)

// hashContext keeps the needed hash context of a db table in hash join.
type hashContext struct {
	// allTypes one-to-one correspondence with keyColIdx
	allTypes        []*types.FieldType
	keyColIdx       []int
	naKeyColIdx     []int
	buf             []byte
	hashVals        []hash.Hash64
	hasNull         []bool
	naHasNull       []bool
	naColNullBitMap []*bitmap.ConcurrentBitmap
}

func (hc *hashContext) initHash(rows int) {
	if hc.buf == nil {
		hc.buf = make([]byte, 1)
	}

	if len(hc.hashVals) < rows {
		hc.hasNull = make([]bool, rows)
		hc.hashVals = make([]hash.Hash64, rows)
		for i := 0; i < rows; i++ {
			hc.hashVals[i] = fnv.New64()
		}
	} else {
		for i := 0; i < rows; i++ {
			hc.hasNull[i] = false
			hc.hashVals[i].Reset()
		}
	}
	if len(hc.naKeyColIdx) > 0 {
		// isNAAJ
		if len(hc.naColNullBitMap) < rows {
			hc.naHasNull = make([]bool, rows)
			hc.naColNullBitMap = make([]*bitmap.ConcurrentBitmap, rows)
			for i := 0; i < rows; i++ {
				hc.naColNullBitMap[i] = bitmap.NewConcurrentBitmap(len(hc.naKeyColIdx))
			}
		} else {
			for i := 0; i < rows; i++ {
				hc.naHasNull[
    
func (c *hashRowContainer) PutChunkSelected(chk *chunk.Chunk, selected, ignoreNulls []bool) error {
	start := time.Now()
	defer func() { c.stat.buildTableElapse += time.Since(start) }()

	chkIdx := uint32(c.rowContainer.NumChunks())
	err := c.rowContainer.Add(chk)
	if err != nil {
		return err
	}
	numRows := chk.NumRows()
	c.hCtx.initHash(numRows)

	hCtx := c.hCtx
	// By now, the combination of 1 and 2 can't take a run at same time.
	// 1: write the row data of join key to hashVals. (normal EQ key should ignore the null values.) null-EQ for Except statement is an exception.
	for keyIdx, colIdx := range c.hCtx.keyColIdx {
		ignoreNull := len(ignoreNulls) > keyIdx && ignoreNulls[keyIdx]
		err := codec.HashChunkSelected(c.sc.TypeCtx(), hCtx.hashVals, chk, hCtx.allTypes[keyIdx], colIdx, hCtx.buf, hCtx.hasNull, selected, ignoreNull)
		if err != nil {
			return errors.Trace(err)
		}
	}
	// 2: write the row data of NA join key to hashVals. (NA EQ key should collect all rows including null value as one bucket.)
	isNAAJ := len(c.hCtx.naKeyColIdx) > 0
	hasNullMark := make([]bool, len(hCtx.hasNull))
	for keyIdx, colIdx := range c.hCtx.naKeyColIdx {
		// NAAJ won't ignore any null values, but collect them as one hash bucket.
		err := codec.HashChunkSelected(c.sc.TypeCtx(), hCtx.hashVals, chk, hCtx.allTypes[keyIdx], colIdx, hCtx.buf, hCtx.hasNull, selected, false)
		if err != nil {
			return errors.Trace(err)
		}
		// todo: we can collect the bitmap in codec.HashChunkSelected to avoid loop here, but the params modification is quite big.
		// after fetch one NA column, collect the null value to null bitmap for every row. (use hasNull flag to accelerate)
		// eg: if a NA Join cols is (a, b, c), for every build row here we maintained a 3-bit map to mark which column are null for them.
		for rowIdx := 0; rowIdx < numRows; rowIdx++ {
			if hCtx.hasNull[rowIdx] {
				hCtx.naColNullBitMap[rowIdx].UnsafeSet(keyIdx)
				// clean and try fetch next NA join col.
				hCtx.hasNull[rowIdx] = false
				// just a mark variable for whether there is a null in at least one NA join column.
				hasNullMark[rowIdx] = true
			}
		}
	}
	for i := 0; i < numRows; i++ {
		if isNAAJ {
			if selected != nil && !selected[i] {
				continue
			}
			if hasNullMark[i] {
				// collect the null rows to slice.
				rowPtr := chunk.RowPtr{ChkIdx: chkIdx, RowIdx: uint32(i)}
				// do not directly ref the null bits map here, because the bit map will be reset and reused in next batch of chunk data.
				c.hashNANullBucket.entries = append(c.hashNANullBucket.entries, &naEntry{rowPtr, c.hCtx.naColNullBitMap[i].Clone()})
			} else {
				// insert the not-null rows to hash table.
				key := c.hCtx.hashVals[i].Sum64()
				rowPtr := chunk.RowPtr{ChkIdx: chkIdx, RowIdx: uint32(i)}
				c.hashTable.Put(key, rowPtr)
			}
		} else {
			if (selected != nil && !selected[i]) || c.hCtx.hasNull[i] {
				continue
			}
			key := c.hCtx.hashVals[i].Sum64()
			rowPtr := chunk.RowPtr{ChkIdx: chkIdx, RowIdx: uint32(i)}
			c.hashTable.Put(key, rowPtr)
		}
	}
	c.GetMemTracker().Consume(c.hashTable.GetAndCleanMemoryDelta())
	return nil
}